Memory Management

/**
 * Interface for memory allocation strategies
 */
interface MemoryAllocator {
    /** 
     * Allocate a memory block of the specified size
     * @param size Size in bytes to allocate
     * @return MemoryBlock representing the allocated memory
     * @throws OutOfMemoryError if allocation fails
     */
    MemoryBlock allocate(long size) throws OutOfMemoryError;
    
    /**
     * Free a previously allocated memory block
     * @param memory MemoryBlock to free
     */
    void free(MemoryBlock memory);
    
    // Standard allocator instances
    static final MemoryAllocator HEAP = new HeapMemoryAllocator();
    static final MemoryAllocator UNSAFE = new UnsafeMemoryAllocator();
    
    // Debug configuration
    static final boolean MEMORY_DEBUG_FILL_ENABLED;
    static final byte MEMORY_DEBUG_FILL_CLEAN_VALUE = (byte)0xa5;
    static final byte MEMORY_DEBUG_FILL_FREED_VALUE = (byte)0x5a;
}

import org.apache.spark.unsafe.memory.*;

// Using heap allocator
MemoryAllocator heapAllocator = MemoryAllocator.HEAP;
MemoryBlock heapBlock = heapAllocator.allocate(1024);
// ... use block ...
heapAllocator.free(heapBlock);

// Using off-heap allocator
MemoryAllocator unsafeAllocator = MemoryAllocator.UNSAFE;  
MemoryBlock offHeapBlock = unsafeAllocator.allocate(2048);
// ... use block ...
unsafeAllocator.free(offHeapBlock);

/**
 * Memory allocator using JVM heap with pooling for large allocations
 */
class HeapMemoryAllocator implements MemoryAllocator {
    /**
     * Allocate heap memory, using pooling for large allocations
     * @param size Size in bytes to allocate
     * @return MemoryBlock backed by heap memory
     * @throws OutOfMemoryError if allocation fails
     */
    public MemoryBlock allocate(long size) throws OutOfMemoryError;
    
    /**
     * Free heap memory, returning large allocations to pool
     * @param memory MemoryBlock to free
     */
    public void free(MemoryBlock memory);
}

/**
 * Memory allocator using off-heap memory via Unsafe
 */
class UnsafeMemoryAllocator implements MemoryAllocator {
    /**
     * Allocate off-heap memory using Unsafe
     * @param size Size in bytes to allocate
     * @return MemoryBlock backed by off-heap memory
     * @throws OutOfMemoryError if allocation fails
     */
    public MemoryBlock allocate(long size) throws OutOfMemoryError;
    
    /**
     * Free off-heap memory using Unsafe
     * @param memory MemoryBlock to free
     */
    public void free(MemoryBlock memory);
}

/**
 * Memory location tracked by address or object reference
 */
class MemoryLocation {
    /**
     * Create memory location with object and offset
     * @param obj Base object (null for off-heap)
     * @param offset Offset within object or direct address
     */
    public MemoryLocation(Object obj, long offset);
    
    /**
     * Default constructor for uninitialized location
     */
    public MemoryLocation();
    
    /**
     * Update the memory location
     * @param newObj New base object
     * @param newOffset New offset
     */
    public void setObjAndOffset(Object newObj, long newOffset);
    
    /**
     * Get the base object (null for off-heap memory)
     * @return Base object or null
     */
    public final Object getBaseObject();
    
    /**
     * Get the offset within object or direct address
     * @return Offset or address
     */
    public final long getBaseOffset();
}

/**
 * Consecutive block of memory with fixed size
 */
class MemoryBlock extends MemoryLocation {
    /**
     * Create memory block with object, offset, and length
     * @param obj Base object (null for off-heap)
     * @param offset Offset within object or direct address
     * @param length Size of the block in bytes
     */
    public MemoryBlock(Object obj, long offset, long length);
    
    /**
     * Get the size of this memory block
     * @return Size in bytes
     */
    public long size();
    
    /**
     * Fill the entire block with a specific byte value
     * @param value Byte value to fill with
     */
    public void fill(byte value);
    
    /**
     * Create a memory block backed by a long array
     * @param array Long array to wrap
     * @return MemoryBlock backed by the array
     */
    public static MemoryBlock fromLongArray(long[] array);
    
    // Page number for TaskMemoryManager integration
    public int pageNumber;
    
    // Special page number constants
    public static final int NO_PAGE_NUMBER = -1;
    public static final int FREED_IN_TMM_PAGE_NUMBER = -2;
    public static final int FREED_IN_ALLOCATOR_PAGE_NUMBER = -3;
}

import org.apache.spark.unsafe.memory.*;

// Create and use memory blocks
MemoryAllocator allocator = MemoryAllocator.HEAP;
MemoryBlock block = allocator.allocate(1024);

// Fill block with zeros
block.fill((byte) 0);

// Get memory location info
Object baseObj = block.getBaseObject();
long baseOffset = block.getBaseOffset();
long size = block.size();

// Create block from long array
long[] data = {1L, 2L, 3L, 4L};
MemoryBlock arrayBlock = MemoryBlock.fromLongArray(data);

// Clean up
allocator.free(block);
// Note: arrayBlock doesn't need explicit freeing as it's backed by a Java array

// For temporary data that should be GC managed
MemoryAllocator heapAllocator = MemoryAllocator.HEAP;

// For large data that should not contribute to GC pressure  
MemoryAllocator offHeapAllocator = MemoryAllocator.UNSAFE;

// The choice depends on:
// - Data lifetime (short-lived: heap, long-lived: off-heap)
// - Size (small: heap, large: off-heap)
// - GC pressure considerations

import org.apache.spark.unsafe.memory.*;

public void processData() {
    MemoryAllocator allocator = MemoryAllocator.HEAP;
    MemoryBlock block = null;
    
    try {
        block = allocator.allocate(1024);
        
        // Use the memory block
        block.fill((byte) 0);
        
        // ... processing logic ...
        
    } catch (OutOfMemoryError e) {
        // Handle allocation failure
        System.err.println("Failed to allocate memory: " + e.getMessage());
    } finally {
        // Always free allocated memory
        if (block != null) {
            allocator.free(block);
        }
    }
}