A single TaskMemoryManager manages the memory of a single task (by the task’s taskAttemptId).

When called, the constructor uses the input MemoryManager to know whether it is in Tungsten memory mode (disabled by default) and saves the MemoryManager and taskAttemptId for later use.

It also initializes the internal consumers to be empty.

acquireExecutionMemory allocates up to required size of memory for consumer. When no memory could be allocated, it calls spill on every consumer, itself including. Finally, it returns the allocated memory.

It first calls memoryManager.acquireExecutionMemory(required, taskAttemptId, mode).

When the memory obtained is less than requested (by required), it requests all consumers to spill the remaining required memory.

You may see the following DEBUG message when spill released some memory:

acquireExecutionMemory calls memoryManager.acquireExecutionMemory(required, taskAttemptId, mode) again (it called it at the beginning).

It does the memory acquisition until it gets enough memory or there are no more consumers to request spill from.

You may also see the following ERROR message in the logs when there is an error while requesting spill with OutOfMemoryError followed.

If the earlier spill on the consumers did not work out and there is still not enough memory acquired, acquireExecutionMemory calls spill on the input consumer (that requested more memory!)

If the consumer releases some memory, you should see the following DEBUG message in the logs:

acquireExecutionMemory calls memoryManager.acquireExecutionMemory(required, taskAttemptId, mode) once more.

It records the consumer in consumers registry.

You should see the following DEBUG message in the logs:

It clears page table.

All recorded consumers are queried for the size of used memory. If the memory used is greater than 0, the following WARN message is printed out to the logs:

The consumers collection is then cleared.

MemoryManager.releaseExecutionMemory is executed to release the memory that is not used by any consumer.

Before cleanUpAllAllocatedMemory returns, it calls MemoryManager.releaseAllExecutionMemoryForTask that in turn becomes the return value.

allocatePage allocates a block of memory (aka page) smaller than MAXIMUM_PAGE_SIZE_BYTES maximum size.

It checks size against the internal MAXIMUM_PAGE_SIZE_BYTES maximum size. If it is greater than the maximum size, the following IllegalArgumentException is thrown:

It then acquires execution memory (for the input size and consumer).

It finishes by returning null when no execution memory could be acquired.

With the execution memory acquired, it finds the smallest unallocated page index and records the page number (using allocatedPages registry).

If the index is PAGE_TABLE_SIZE or higher, releaseExecutionMemory(acquired, consumer) is called and then the following IllegalStateException is thrown:

It then attempts to allocate a MemoryBlock from Tungsten MemoryAllocator (calling memoryManager.tungstenMemoryAllocator().allocate(acquired)).

When successful, MemoryBlock gets assigned pageNumber and it gets added to the internal pageTable registry.

You should see the following TRACE message in the logs:

The page is returned.

If a OutOfMemoryError is thrown when allocating a MemoryBlock page, the following WARN message is printed out to the logs:

And acquiredButNotUsed gets acquired memory space with the pageNumber cleared in allocatedPages (i.e. the index for pageNumber gets false).

Another allocatePage attempt is recursively tried.