allocator

AllocationStrategy

Bases: Enum

Enumeration of allocation strategies

AllocationRequest dataclass

AllocationRequest(size: int, alignment: int = 1, stream: Optional[int] = None, priority: int = 0, metadata: Optional[Dict[str, Any]] = None)

Represents a memory allocation request

AllocationResult dataclass

AllocationResult(success: bool, block: Optional[MemoryBlock] = None, address: Optional[int] = None, actual_size: Optional[int] = None, allocation_time_ns: Optional[int] = None, error_message: Optional[str] = None, strategy_info: Optional[Dict[str, Any]] = None)

Represents the result of an allocation attempt

FreeRequest dataclass

FreeRequest(address: int, expected_size: Optional[int] = None, metadata: Optional[Dict[str, Any]] = None)

Represents a memory deallocation request

FreeResult dataclass

FreeResult(success: bool, address: Optional[int] = None, freed_size: Optional[int] = None, free_time_ns: Optional[int] = None, coalesced: bool = False, coalesced_size: Optional[int] = None, error_message: Optional[str] = None, strategy_info: Optional[Dict[str, Any]] = None)

Represents the result of a deallocation attempt

MemoryAllocator

MemoryAllocator(name: str)

Bases: ABC

Abstract base class for memory allocation algorithms

Source code in ures/memory/allocator.py

def __init__(self, name: str):
    self.name = name
    self.allocation_count = 0
    self.free_count = 0
    self.total_allocated = 0
    self.total_freed = 0
    self.allocation_times = []
    self.free_times = []
    self.allocated_blocks: Dict[int, MemoryBlock] = (
        {}
    )  # Track allocated blocks by address

allocate abstractmethod

allocate(pool: BlockPool, request: AllocationRequest) -> AllocationResult

Allocate memory according to the algorithm's strategy

Source code in ures/memory/allocator.py

@abstractmethod
def allocate(self, pool: BlockPool, request: AllocationRequest) -> AllocationResult:
    """Allocate memory according to the algorithm's strategy"""
    pass

free abstractmethod

free(pool: BlockPool, request: FreeRequest) -> FreeResult

Free memory according to the algorithm's strategy

Source code in ures/memory/allocator.py

@abstractmethod
def free(self, pool: BlockPool, request: FreeRequest) -> FreeResult:
    """Free memory according to the algorithm's strategy"""
    pass

can_allocate abstractmethod

can_allocate(pool: BlockPool, request: AllocationRequest) -> bool

Check if allocation is possible without actually allocating

Source code in ures/memory/allocator.py

@abstractmethod
def can_allocate(self, pool: BlockPool, request: AllocationRequest) -> bool:
    """Check if allocation is possible without actually allocating"""
    pass

can_free

can_free(pool: BlockPool, request: FreeRequest) -> bool

Check if deallocation is possible

Source code in ures/memory/allocator.py

def can_free(self, pool: BlockPool, request: FreeRequest) -> bool:
    """Check if deallocation is possible"""
    return request.address in self.allocated_blocks

get_statistics

get_statistics() -> Dict[str, Any]

Get allocator statistics

Source code in ures/memory/allocator.py

def get_statistics(self) -> Dict[str, Any]:
    """Get allocator statistics"""
    avg_alloc_time = (
        sum(self.allocation_times) / len(self.allocation_times)
        if self.allocation_times
        else 0
    )
    avg_free_time = (
        sum(self.free_times) / len(self.free_times) if self.free_times else 0
    )

    return {
        "name": self.name,
        "allocation_count": self.allocation_count,
        "free_count": self.free_count,
        "total_allocated": self.total_allocated,
        "total_freed": self.total_freed,
        "currently_allocated": self.total_allocated - self.total_freed,
        "active_blocks": len(self.allocated_blocks),
        "average_allocation_time_ns": avg_alloc_time,
        "average_free_time_ns": avg_free_time,
        "allocation_times": self.allocation_times.copy(),
        "free_times": self.free_times.copy(),
    }

FirstFitAllocator

FirstFitAllocator()

Bases: MemoryAllocator

First-fit allocation algorithm

Source code in ures/memory/allocator.py

def __init__(self):
    super().__init__("First Fit")

BestFitAllocator

BestFitAllocator()

Bases: MemoryAllocator

Best-fit allocation algorithm

Source code in ures/memory/allocator.py

def __init__(self):
    super().__init__("Best Fit")

WorstFitAllocator

WorstFitAllocator()

Bases: MemoryAllocator

Worst-fit allocation algorithm

Source code in ures/memory/allocator.py

def __init__(self):
    super().__init__("Worst Fit")

NextFitAllocator

NextFitAllocator()

Bases: MemoryAllocator

Next-fit allocation algorithm

Source code in ures/memory/allocator.py

def __init__(self):
    super().__init__("Next Fit")
    self.last_allocated_block = None  # Remember where we last allocated

BuddySystemAllocator

BuddySystemAllocator()

Bases: MemoryAllocator

Buddy system allocation algorithm

Source code in ures/memory/allocator.py

def __init__(self):
    super().__init__("Buddy System")
    self.min_block_size = 64  # Minimum block size (can be configured)

DeviceMemorySimulator

DeviceMemorySimulator(device_id: int, total_memory: int, base_address: int = 268435456)

Simulate a device heap and swap among allocation algorithms.

Examples:

>>> from ures.memory import DeviceMemorySimulator
>>> sim = DeviceMemorySimulator(device_id=0, total_memory=1024)
>>> result = sim.allocate(size=64)
>>> result.success
True
>>> sim.free(result.address).success
True

Source code in ures/memory/allocator.py

def __init__(
    self, device_id: int, total_memory: int, base_address: int = 0x10000000
):
    self.device_id = device_id
    self.total_memory = total_memory
    self.base_address = base_address
    self.pool = BlockPool()
    self.allocators: Dict[str, MemoryAllocator] = {}
    self.current_allocator: Optional[MemoryAllocator] = None
    self.allocation_history: List[Tuple[AllocationRequest, AllocationResult]] = []
    self.free_history: List[Tuple[int, int]] = []  # (address, time_ns)

    # Create initial memory segment
    self.main_segment = self.pool.create_segment(
        start_addr=base_address, size=total_memory, device=device_id
    )

    # Register default allocators
    self.register_allocator(FirstFitAllocator())
    self.register_allocator(BestFitAllocator())
    self.register_allocator(WorstFitAllocator())
    self.register_allocator(NextFitAllocator())
    self.register_allocator(BuddySystemAllocator())

    # Set default allocator
    self.set_allocator("First Fit")

register_allocator

register_allocator(allocator: MemoryAllocator)

Register a new allocation algorithm

Source code in ures/memory/allocator.py

def register_allocator(self, allocator: MemoryAllocator):
    """Register a new allocation algorithm"""
    self.allocators[allocator.name] = allocator

set_allocator

set_allocator(allocator_name: str) -> bool

Set the active allocation algorithm

Source code in ures/memory/allocator.py

def set_allocator(self, allocator_name: str) -> bool:
    """Set the active allocation algorithm"""
    if allocator_name in self.allocators:
        self.current_allocator = self.allocators[allocator_name]
        return True
    return False

get_available_allocators

get_available_allocators() -> List[str]

Get list of available allocator names

Source code in ures/memory/allocator.py

def get_available_allocators(self) -> List[str]:
    """Get list of available allocator names"""
    return list(self.allocators.keys())

allocate

allocate(size: int, alignment: int = 1, stream: Optional[int] = None, priority: int = 0, metadata: Optional[Dict[str, Any]] = None) -> AllocationResult

Allocate memory using the current algorithm

Source code in ures/memory/allocator.py

def allocate(
    self,
    size: int,
    alignment: int = 1,
    stream: Optional[int] = None,
    priority: int = 0,
    metadata: Optional[Dict[str, Any]] = None,
) -> AllocationResult:
    """Allocate memory using the current algorithm"""
    if not self.current_allocator:
        return AllocationResult(success=False, error_message="No allocator set")

    request = AllocationRequest(
        size=size,
        alignment=alignment,
        stream=stream,
        priority=priority,
        metadata=metadata,
    )

    result = self.current_allocator.allocate(self.pool, request)
    self.allocation_history.append((request, result))

    return result

free

free(address: int, expected_size: Optional[int] = None) -> FreeResult

Free memory at the given address using the current allocator

Source code in ures/memory/allocator.py

def free(self, address: int, expected_size: Optional[int] = None) -> FreeResult:
    """Free memory at the given address using the current allocator"""
    if not self.current_allocator:
        return FreeResult(
            success=False, address=address, error_message="No allocator set"
        )

    request = FreeRequest(address=address, expected_size=expected_size)

    result = self.current_allocator.free(self.pool, request)
    self.free_history.append((address, time.time_ns()))

    return result

can_allocate

can_allocate(size: int, stream: Optional[int] = None) -> bool

Check if allocation is possible without actually allocating

Source code in ures/memory/allocator.py

def can_allocate(self, size: int, stream: Optional[int] = None) -> bool:
    """Check if allocation is possible without actually allocating"""
    if not self.current_allocator:
        return False

    request = AllocationRequest(size=size, stream=stream)
    return self.current_allocator.can_allocate(self.pool, request)

get_memory_info

get_memory_info() -> Dict[str, Any]

Get current memory status information

Source code in ures/memory/allocator.py

def get_memory_info(self) -> Dict[str, Any]:
    """Get current memory status information"""
    summary = self.pool.get_memory_summary()

    return {
        "device_id": self.device_id,
        "total_memory": self.total_memory,
        "base_address": self.base_address,
        "base_address_hex": hex(self.base_address),
        "current_allocator": (
            self.current_allocator.name if self.current_allocator else None
        ),
        "memory_summary": summary,
        "allocation_count": len(self.allocation_history),
        "free_count": len(self.free_history),
        "segments": self.pool.list_all_segments(),
    }

get_allocator_statistics

get_allocator_statistics() -> Dict[str, Dict[str, Any]]

Get statistics for all allocators

Source code in ures/memory/allocator.py

def get_allocator_statistics(self) -> Dict[str, Dict[str, Any]]:
    """Get statistics for all allocators"""
    stats = {}
    for name, allocator in self.allocators.items():
        stats[name] = allocator.get_statistics()
    return stats

reset_device

reset_device()

Reset device to initial state

Source code in ures/memory/allocator.py

def reset_device(self):
    """Reset device to initial state"""
    # Clear all data
    self.pool = BlockPool()
    self.allocation_history.clear()
    self.free_history.clear()

    # Reset allocator statistics
    for allocator in self.allocators.values():
        allocator.allocation_count = 0
        allocator.free_count = 0
        allocator.total_allocated = 0
        allocator.total_freed = 0
        allocator.allocation_times.clear()
        allocator.free_times.clear()
        allocator.allocated_blocks.clear()

    # Recreate main segment
    self.main_segment = self.pool.create_segment(
        start_addr=self.base_address, size=self.total_memory, device=self.device_id
    )

simulate_workload

simulate_workload(num_operations: int = 100, size_range: Tuple[int, int] = (64, 4096), free_probability: float = 0.3) -> Dict[str, Any]

Simulate a random workload

Source code in ures/memory/allocator.py

def simulate_workload(
    self,
    num_operations: int = 100,
    size_range: Tuple[int, int] = (64, 4096),
    free_probability: float = 0.3,
) -> Dict[str, Any]:
    """Simulate a random workload"""
    operations = []
    allocated_blocks = []

    for i in range(num_operations):
        if allocated_blocks and random.random() < free_probability:
            # Free a random block
            block_to_free = random.choice(allocated_blocks)
            result = self.free(block_to_free)
            operations.append(
                {
                    "operation": "free",
                    "address": block_to_free,
                    "success": result.success,
                }
            )
            if result.success:
                allocated_blocks.remove(block_to_free)
        else:
            # Allocate a new block
            size = random.randint(*size_range)
            result = self.allocate(size)
            operations.append(
                {
                    "operation": "allocate",
                    "size": size,
                    "success": result.success,
                    "address": result.address,
                }
            )
            if result.success:
                allocated_blocks.append(result.address)

    return {
        "operations": operations,
        "final_memory_info": self.get_memory_info(),
        "allocator_stats": self.get_allocator_statistics(),
    }

print_status

print_status()

Print current device status

Source code in ures/memory/allocator.py

def print_status(self):
    """Print current device status"""
    print(f"=== Device {self.device_id} Memory Status ===")
    info = self.get_memory_info()
    summary = info["memory_summary"]

    print(f"Total Memory: {self.total_memory} bytes")
    print(f"Base Address: {info['base_address_hex']}")
    print(f"Current Allocator: {info['current_allocator']}")
    print(
        f"Allocated: {summary['total_allocated_bytes']} bytes ({summary['overall_utilization']:.1%})"
    )
    print(f"Free: {summary['total_free_bytes']} bytes")
    print(f"Fragmentation: {summary['average_fragmentation']:.1%}")
    print(f"Total Allocations: {info['allocation_count']}")
    print(f"Total Frees: {info['free_count']}")

    self.pool.print_memory_status()