"""Token blocks.""" from typing import List, Optional from aphrodite.common.utils import Device DEFAULT_LAST_ACCESSED_TIME = -1 class PhysicalTokenBlock: """Represents the state of a block in the KV cache.""" def __init__( self, device: Device, block_number: int, block_size: int, block_hash: int, num_hashed_tokens: int, ) -> None: self.device = device self.block_number = block_number self.block_size = block_size self.block_hash = block_hash self.num_hashed_tokens = num_hashed_tokens self.ref_count = 0 self.last_accessed = DEFAULT_LAST_ACCESSED_TIME self.computed = False def __repr__(self) -> str: return (f'PhysicalTokenBlock(device={self.device}, ' f'block_number={self.block_number}, ' f'num_hashed_tokens={self.num_hashed_tokens}, ' f'ref_count={self.ref_count}, ' f'last_accessed={self.last_accessed}, ' f'computed={self.computed})') class BlockTable: """Holds a list of blocks with caching of their associated block_ids """ def __init__(self, blocks: Optional[List[PhysicalTokenBlock]] = None): self._blocks: List[PhysicalTokenBlock] = [] self._block_ids: List[int] = [] if blocks is not None: for block in blocks: self.append(block) def append(self, block: PhysicalTokenBlock): self._blocks.append(block) self._block_ids.append(block.block_number) def __len__(self) -> int: return len(self._blocks) def __getitem__(self, key): return self._blocks[key] def __setitem__(self, key, value): if isinstance(key, slice): blocks = value self._blocks[key] = blocks self._block_ids[key] = [b.block_number for b in blocks] else: block = value self._blocks[key] = block self._block_ids[key] = block.block_number def reset(self): self._blocks = [] self._block_ids = [] def copy(self) -> "BlockTable": return BlockTable(self._blocks) def list(self) -> List[PhysicalTokenBlock]: return self._blocks def ids(self) -> List[int]: return self._block_ids