53. vLLM 核心模块逐文件：block_manager.py

class BlockManager:
    """Manages the KV cache blocks for vLLM."""
    
    def __init__(self, cache_config: CacheConfig, block_size: int, num_gpus: int, device_config: DeviceConfig):
        """Initialize the block manager.
        
        Args:
            cache_config: The cache configuration.
            block_size: The block size in tokens.
            num_gpus: The number of GPUs.
            device_config: The device configuration.
        """
        # 保存配置
        self.cache_config = cache_config
        self.block_size = block_size
        self.num_gpus = num_gpus
        self.device_config = device_config
        
        # 计算每个块的大小（字节）
        self.block_size_bytes = self._calculate_block_size_bytes()
        
        # 初始化内存池
        self.memory_pool = MemoryPool(
            cache_config=cache_config,
            num_gpus=num_gpus,
            device_config=device_config,
        )
        
        # 初始化块分配器
        self.block_allocator = BlockAllocator(
            memory_pool=self.memory_pool,
            block_size=self.block_size,
            num_gpus=num_gpus,
        )
        
        # 初始化块表
        self.block_table = BlockTable(
            num_gpus=num_gpus,
        )
        
        # 初始化压缩管理器
        if cache_config.enable_kv_cache_compression:
            self.compression_manager = CompressionManager(
                cache_config=cache_config,
            )
        else:
            self.compression_manager = None
        
        # 初始化统计信息收集器
        self.stats_collector = StatsCollector()
        
        # 初始化块ID计数器
        self.block_id_counter = 0

async def allocate_blocks(self, num_blocks: int, request_id: int, gpu_id: int = 0) -> List[Block]:
    """Allocate blocks for a request.
    
    Args:
        num_blocks: The number of blocks to allocate.
        request_id: The request ID.
        gpu_id: The GPU ID to allocate blocks on.
    
    Returns:
        A list of allocated blocks.
    
    Raises:
        BlockAllocationError: If failed to allocate blocks.
    """
    # 验证请求合法性
    if num_blocks <= 0:
        return []
    
    # 尝试分配块
    try:
        blocks = await self.block_allocator.allocate(num_blocks, request_id, gpu_id)
    except BlockAllocationError:
        # 块分配失败，尝试释放一些块
        await self._free_some_blocks()
        
        # 再次尝试分配
        try:
            blocks = await self.block_allocator.allocate(num_blocks, request_id, gpu_id)
        except BlockAllocationError:
            raise BlockAllocationError(f"Failed to allocate {num_blocks} blocks for request {request_id}")
    
    # 更新块表
    for block in blocks:
        self.block_table.add_block(block, request_id)
    
    # 更新统计信息
    self.stats_collector.record_blocks_allocated(blocks)
    
    return blocks

async def free_blocks(self, blocks: List[Block]) -> None:
    """Free the given blocks.
    
    Args:
        blocks: The blocks to free.
    """
    if not blocks:
        return
    
    # 释放块
    await self.block_allocator.free(blocks)
    
    # 从块表中移除块
    for block in blocks:
        self.block_table.remove_block(block)
    
    # 更新统计信息
    self.stats_collector.record_blocks_freed(blocks)

async def merge_adjacent_blocks(self) -> None:
    """Merge adjacent free blocks to reduce fragmentation."""
    # 获取所有空闲块
    free_blocks = await self.block_allocator.get_free_blocks()
    
    # 按GPU ID和内存类型分组
    blocks_by_gpu_and_type = defaultdict(list)
    for block in free_blocks:
        key = (block.gpu_id, block.memory_type)
        blocks_by_gpu_and_type[key].append(block)
    
    # 合并每个组中的相邻块
    for key, blocks in blocks_by_gpu_and_type.items():
        # 按起始地址排序
        sorted_blocks = sorted(blocks, key=lambda b: b.start_address)
        
        # 合并相邻块
        merged_blocks = []
        current_block = None
        
        for block in sorted_blocks:
            if current_block is None:
                current_block = block
            elif self._are_adjacent(current_block, block):
                # 合并相邻块
                current_block = self._merge_blocks(current_block, block)
            else:
                merged_blocks.append(current_block)
                current_block = block
        
        if current_block:
            merged_blocks.append(current_block)
        
        # 替换原块为合并后的块
        if merged_blocks and len(merged_blocks) < len(blocks):
            await self.block_allocator.replace_blocks(blocks, merged_blocks)
            
            # 更新块表
            for block in blocks:
                self.block_table.remove_block(block)
            for block in merged_blocks:
                self.block_table.add_block(block, None)

class MemoryPool:
    """Manages memory pools for different devices and memory types."""
    
    def __init__(self, cache_config: CacheConfig, num_gpus: int, device_config: DeviceConfig):
        """Initialize the memory pool.
        
        Args:
            cache_config: The cache configuration.
            num_gpus: The number of GPUs.
            device_config: The device configuration.
        """
        # 保存配置
        self.cache_config = cache_config
        self.num_gpus = num_gpus
        self.device_config = device_config
        
        # 初始化不同类型的内存池
        self.memory_pools = {
            MemoryType.GPU: {},
            MemoryType.CPU: CPUMemoryPool(cache_config),
            MemoryType.NVMe: NVMeMemoryPool(cache_config) if cache_config.enable_nvme_offload else None,
        }
        
        # 初始化GPU内存池
        for gpu_id in range(num_gpus):
            self.memory_pools[MemoryType.GPU][gpu_id] = GPUMemoryPool(
                cache_config=cache_config,
                gpu_id=gpu_id,
                device_config=device_config,
            )
    
    def get_memory_pool(self, memory_type: MemoryType, gpu_id: int = 0) -> Optional[BaseMemoryPool]:
        """Get the memory pool for the given memory type and GPU ID.
        
        Args:
            memory_type: The memory type.
            gpu_id: The GPU ID.
        
        Returns:
            The memory pool, or None if not available.
        """
        if memory_type == MemoryType.GPU:
            return self.memory_pools[memory_type].get(gpu_id)
        else:
            return self.memory_pools[memory_type]
    
    def get_total_memory(self, memory_type: MemoryType, gpu_id: int = 0) -> int:
        """Get the total memory for the given memory type and GPU ID.
        
        Args:
            memory_type: The memory type.
            gpu_id: The GPU ID.
        
        Returns:
            The total memory in bytes.
        """
        pool = self.get_memory_pool(memory_type, gpu_id)
        return pool.get_total_memory() if pool else 0
    
    def get_used_memory(self, memory_type: MemoryType, gpu_id: int = 0) -> int:
        """Get the used memory for the given memory type and GPU ID.
        
        Args:
            memory_type: The memory type.
            gpu_id: The GPU ID.
        
        Returns:
            The used memory in bytes.
        """
        pool = self.get_memory_pool(memory_type, gpu_id)
        return pool.get_used_memory() if pool else 0
    
    def get_free_memory(self, memory_type: MemoryType, gpu_id: int = 0) -> int:
        """Get the free memory for the given memory type and GPU ID.
        
        Args:
            memory_type: The memory type.
            gpu_id: The GPU ID.
        
        Returns:
            The free memory in bytes.
        """
        return self.get_total_memory(memory_type, gpu_id) - self.get_used_memory(memory_type, gpu_id)

class GPUMemoryPool(BaseMemoryPool):
    """Memory pool for GPU memory."""
    
    def __init__(self, cache_config: CacheConfig, gpu_id: int, device_config: DeviceConfig):
        """Initialize the GPU memory pool.
        
        Args:
            cache_config: The cache configuration.
            gpu_id: The GPU ID.
            device_config: The device configuration.
        """
        super().__init__(cache_config)
        
        # 保存配置
        self.gpu_id = gpu_id
        self.device_config = device_config
        
        # 计算GPU内存池大小
        total_gpu_memory = self._get_total_gpu_memory(gpu_id)
        self.pool_size = int(total_gpu_memory * cache_config.gpu_memory_utilization)
        
        # 分配GPU内存
        self.memory = self._allocate_gpu_memory(self.pool_size, gpu_id)
        
        # 初始化已使用内存大小
        self.used_memory = 0
    
    def _get_total_gpu_memory(self, gpu_id: int) -> int:
        """Get the total GPU memory in bytes."""
        return torch.cuda.get_device_properties(gpu_id).total_memory
    
    def _allocate_gpu_memory(self, size: int, gpu_id: int) -> torch.Tensor:
        """Allocate GPU memory."""
        device = torch.device(f"cuda:{gpu_id}")
        return torch.empty(size, dtype=torch.uint8, device=device, requires_grad=False)
    
    def get_total_memory(self) -> int:
        """Get the total memory in bytes."""
        return self.pool_size
    
    def get_used_memory(self) -> int:
        """Get the used memory in bytes."""
        return self.used_memory
    
    def allocate(self, size: int) -> MemoryBlock:
        """Allocate memory from the pool.
        
        Args:
            size: The size to allocate in bytes.
        
        Returns:
            A memory block.
        
        Raises:
            MemoryAllocationError: If failed to allocate memory.
        """
        if size > self.get_free_memory():
            raise MemoryAllocationError(f"Failed to allocate {size} bytes from GPU memory pool")
        
        # 分配内存
        start_address = self.used_memory
        self.used_memory += size
        
        # 创建内存块
        return MemoryBlock(
            start_address=start_address,
            size=size,
            memory_type=MemoryType.GPU,
            gpu_id=self.gpu_id,
            memory=self.memory,
        )
    
    def free(self, block: MemoryBlock) -> None:
        """Free a memory block.
        
        Args:
            block: The memory block to free.
        """
        # GPU内存池不支持单个块的释放，只支持整体重置
        # 实际释放逻辑在BlockAllocator中实现
        pass

class FreeListAllocator(BaseBlockAllocator):
    """Block allocator that uses a free list."""
    
    def __init__(self, memory_pool: MemoryPool, block_size: int, num_gpus: int):
        """Initialize the free list allocator.
        
        Args:
            memory_pool: The memory pool.
            block_size: The block size in tokens.
            num_gpus: The number of GPUs.
        """
        super().__init__(memory_pool, block_size, num_gpus)
        
        # 初始化空闲列表（按GPU ID和内存类型分组）
        self.free_lists = defaultdict(lambda: defaultdict(list))
        
        # 初始化已分配块映射
        self.allocated_blocks = {}
        
        # 初始化块ID计数器
        self.block_id_counter = 0
    
    async def allocate(self, num_blocks: int, request_id: int, gpu_id: int = 0) -> List[Block]:
        """Allocate blocks.
        
        Args:
            num_blocks: The number of blocks to allocate.
            request_id: The request ID.
            gpu_id: The GPU ID.
        
        Returns:
            A list of allocated blocks.
        
        Raises:
            BlockAllocationError: If failed to allocate blocks.
        """
        # 尝试从GPU内存分配
        try:
            return await self._allocate_from_memory_type(num_blocks, request_id, gpu_id, MemoryType.GPU)
        except BlockAllocationError:
            pass
        
        # 如果GPU内存不足，尝试从CPU内存分配
        if self.memory_pool.get_memory_pool(MemoryType.CPU):
            try:
                return await self._allocate_from_memory_type(num_blocks, request_id, gpu_id, MemoryType.CPU)
            except BlockAllocationError:
                pass
        
        # 如果CPU内存也不足，尝试从NVMe内存分配
        if self.memory_pool.get_memory_pool(MemoryType.NVMe):
            try:
                return await self._allocate_from_memory_type(num_blocks, request_id, gpu_id, MemoryType.NVMe)
            except BlockAllocationError:
                pass
        
        # 所有内存类型都分配失败
        raise BlockAllocationError(f"Failed to allocate {num_blocks} blocks for request {request_id}")
    
    async def _allocate_from_memory_type(self, num_blocks: int, request_id: int, gpu_id: int, memory_type: MemoryType) -> List[Block]:
        """Allocate blocks from a specific memory type.
        
        Args:
            num_blocks: The number of blocks to allocate.
            request_id: The request ID.
            gpu_id: The GPU ID.
            memory_type: The memory type.
        
        Returns:
            A list of allocated blocks.
        
        Raises:
            BlockAllocationError: If failed to allocate blocks.
        """
        # 获取该内存类型的空闲列表
        free_list = self.free_lists[gpu_id][memory_type]
        
        # 尝试从空闲列表中分配块
        allocated_blocks = []
        remaining_blocks = num_blocks
        
        # 遍历空闲列表，寻找足够的块
        for i in range(len(free_list) - 1, -1, -1):
            block = free_list[i]
            if remaining_blocks <= 0:
                break
            
            # 分配块
            allocated_blocks.append(block)
            remaining_blocks -= 1
            
            # 从空闲列表中移除该块
            del free_list[i]
        
        # 如果空闲列表中的块不足，分配新块
        if remaining_blocks > 0:
            new_blocks = await self._allocate_new_blocks(remaining_blocks, request_id, gpu_id, memory_type)
            allocated_blocks.extend(new_blocks)
        
        # 更新已分配块映射
        for block in allocated_blocks:
            self.allocated_blocks[block.block_id] = block
        
        return allocated_blocks
    
    async def _allocate_new_blocks(self, num_blocks: int, request_id: int, gpu_id: int, memory_type: MemoryType) -> List[Block]:
        """Allocate new blocks from memory pool.
        
        Args:
            num_blocks: The number of blocks to allocate.
            request_id: The request ID.
            gpu_id: The GPU ID.
            memory_type: The memory type.
        
        Returns:
            A list of allocated blocks.
        
        Raises:
            BlockAllocationError: If failed to allocate blocks.
        """
        # 获取内存池
        memory_pool = self.memory_pool.get_memory_pool(memory_type, gpu_id)
        if not memory_pool:
            raise BlockAllocationError(f"Memory pool not available for {memory_type}")
        
        # 计算需要分配的内存大小
        block_size_bytes = self.block_size * self._get_token_size()
        total_size = num_blocks * block_size_bytes
        
        # 从内存池分配内存
        try:
            memory_block = memory_pool.allocate(total_size)
        except MemoryAllocationError:
            raise BlockAllocationError(f"Failed to allocate memory for {num_blocks} blocks")
        
        # 创建块
        blocks = []
        for i in range(num_blocks):
            start_address = memory_block.start_address + i * block_size_bytes
            block = Block(
                block_id=self.block_id_counter,
                start_address=start_address,
                size=block_size_bytes,
                memory_type=memory_type,
                gpu_id=gpu_id,
                memory_block=memory_block,
                request_id=request_id,
            )
            blocks.append(block)
            self.block_id_counter += 1
        
        return blocks
    
    async def free(self, blocks: List[Block]) -> None:
        """Free blocks.
        
        Args:
            blocks: The blocks to free.
        """
        for block in blocks:
            # 从已分配块映射中移除
            if block.block_id in self.allocated_blocks:
                del self.allocated_blocks[block.block_id]
            
            # 将块添加到空闲列表
            self.free_lists[block.gpu_id][block.memory_type].append(block)
    
    def _get_token_size(self) -> int:
        """Get the size of a token in bytes."""
        # 假设使用float16，每个token的KV缓存大小为：
        # 2（K和V） * 头数 * 头维度 * 2字节（float16）
        # 这里简化处理，返回一个固定值
        return 256

class BlockTable:
    """Manages the block table for KV cache blocks."""
    
    def __init__(self, num_gpus: int):
        """Initialize the block table.
        
        Args:
            num_gpus: The number of GPUs.
        """
        # 保存配置
        self.num_gpus = num_gpus
        
        # 初始化块表
        self.block_table = {}
        self.request_to_blocks = defaultdict(list)
    
    def add_block(self, block: Block, request_id: Optional[int]) -> None:
        """Add a block to the table.
        
        Args:
            block: The block to add.
            request_id: The request ID.
        """
        # 更新块表
        self.block_table[block.block_id] = {
            "block": block,
            "request_id": request_id,
            "status": BlockStatus.ALLOCATED,
            "timestamp": time.time(),
        }
        
        # 更新请求到块的映射
        if request_id is not None:
            self.request_to_blocks[request_id].append(block)
    
    def remove_block(self, block: Block) -> None:
        """Remove a block from the table.
        
        Args:
            block: The block to remove.
        """
        # 从块表中移除
        if block.block_id in self.block_table:
            entry = self.block_table[block.block_id]
            request_id = entry["request_id"]
            del self.block_table[block.block_id]
            
            # 从请求到块的映射中移除
            if request_id is not None and request_id in self.request_to_blocks:
                self.request_to_blocks[request_id] = [b for b in self.request_to_blocks[request_id] if b.block_id != block.block_id]
                if not self.request_to_blocks[request_id]:
                    del self.request_to_blocks[request_id]
    
    def get_blocks_for_request(self, request_id: int) -> List[Block]:
        """Get all blocks for a request.
        
        Args:
            request_id: The request ID.
        
        Returns:
            A list of blocks.
        """
        return self.request_to_blocks.get(request_id, [])
    
    def get_block(self, block_id: int) -> Optional[Block]:
        """Get a block by ID.
        
        Args:
            block_id: The block ID.
        
        Returns:
            The block, or None if not found.
        """
        entry = self.block_table.get(block_id)
        return entry["block"] if entry else None
    
    def get_all_blocks(self) -> List[Block]:
        """Get all blocks.
        
        Returns:
            A list of all blocks.
        """
        return [entry["block"] for entry in self.block_table.values()]
    
    def get_allocated_blocks(self) -> List[Block]:
        """Get all allocated blocks.
        
        Returns:
            A list of allocated blocks.
        """
        return [entry["block"] for entry in self.block_table.values() if entry["status"] == BlockStatus.ALLOCATED]
    
    def get_free_blocks(self) -> List[Block]:
        """Get all free blocks.
        
        Returns:
            A list of free blocks.
        """
        return [entry["block"] for entry in self.block_table.values() if entry["status"] == BlockStatus.FREE]

from vllm.block_manager import BlockManager
from vllm.config import CacheConfig, DeviceConfig

# 配置缓存参数
cache_config = CacheConfig(
    block_size=16,
    gpu_memory_utilization=0.9,
    swap_space=4,
    enable_kv_cache_compression=True,
)

# 创建设备配置
device_config = DeviceConfig(
    device="cuda",
    seed=42,
)

# 创建块管理器
block_manager = BlockManager(
    cache_config=cache_config,
    block_size=16,
    num_gpus=1,
    device_config=device_config,
)

# 分配块
request_id = 123
num_blocks = 4
try:
    blocks = await block_manager.allocate_blocks(num_blocks, request_id)
    print(f"Allocated {len(blocks)} blocks for request {request_id}")
    for block in blocks:
        print(f"  - Block {block.block_id}: size={block.size} bytes, type={block.memory_type}, gpu_id={block.gpu_id}")
    
    # 使用块（模拟）
    print("\nUsing blocks...")
    
    # 释放块
    await block_manager.free_blocks(blocks)
    print(f"\nFreed {len(blocks)} blocks")
    
    # 获取统计信息
    stats = block_manager.stats_collector.get_summary()
    print(f"\nStatistics:")
    print(f"  - Total blocks allocated: {stats['total_blocks_allocated']}")
    print(f"  - Total blocks freed: {stats['total_blocks_freed']}")
    print(f"  - Current allocated blocks: {stats['current_allocated_blocks']}")
except Exception as e:
    print(f"Error: {e}")

Allocated 4 blocks for request 123
  - Block 0: size=1024 bytes, type=MemoryType.GPU, gpu_id=0
  - Block 1: size=1024 bytes, type=MemoryType.GPU, gpu_id=0
  - Block 2: size=1024 bytes, type=MemoryType.GPU, gpu_id=0
  - Block 3: size=1024 bytes, type=MemoryType.GPU, gpu_id=0

Using blocks...

Freed 4 blocks

Statistics:
  - Total blocks allocated: 4
  - Total blocks freed: 4
  - Current allocated blocks: 0

import asyncio
from vllm.block_manager import BlockManager, BlockAllocationError
from vllm.config import CacheConfig, DeviceConfig

async def simulate_memory_pressure():
    # 配置缓存参数（使用较小的GPU内存利用率）
    cache_config = CacheConfig(
        block_size=16,
        gpu_memory_utilization=0.1,  # 仅使用10%的GPU内存
        swap_space=1,
        enable_kv_cache_compression=True,
    )
    
    # 创建设备配置
    device_config = DeviceConfig(
        device="cuda",
        seed=42,
    )
    
    # 创建块管理器
    block_manager = BlockManager(
        cache_config=cache_config,
        block_size=16,
        num_gpus=1,
        device_config=device_config,
    )
    
    # 模拟内存压力场景
    request_id = 0
    allocated_blocks = []
    
    print("Starting memory pressure simulation...")
    
    try:
        # 持续分配块，直到内存不足
        while True:
            request_id += 1
            num_blocks = 2
            
            try:
                blocks = await block_manager.allocate_blocks(num_blocks, request_id)
                allocated_blocks.extend(blocks)
                print(f"Request {request_id}: Allocated {len(blocks)} blocks")
                
                # 模拟使用块
                await asyncio.sleep(0.01)
                
            except BlockAllocationError as e:
                print(f"Request {request_id}: Failed to allocate blocks - {e}")
                break
        
        print(f"\nSimulation stopped. Total requests: {request_id}")
        print(f"Total blocks allocated: {len(allocated_blocks)}")
        
        # 释放所有块
        print("\nFreeing all blocks...")
        await block_manager.free_blocks(allocated_blocks)
        print(f"Freed {len(allocated_blocks)} blocks")
        
        # 获取统计信息
        stats = block_manager.stats_collector.get_summary()
        print(f"\nFinal Statistics:")
        print(f"  - Total blocks allocated: {stats['total_blocks_allocated']}")
        print(f"  - Total blocks freed: {stats['total_blocks_freed']}")
        print(f"  - Current allocated blocks: {stats['current_allocated_blocks']}")
        print(f"  - Memory fragmentation rate: {stats['memory_fragmentation_rate']:.2%}")
        
    except Exception as e:
        print(f"Error during simulation: {e}")

# 运行模拟
asyncio.run(simulate_memory_pressure())

Starting memory pressure simulation...
Request 1: Allocated 2 blocks
Request 2: Allocated 2 blocks
Request 3: Allocated 2 blocks
Request 4: Allocated 2 blocks
Request 5: Allocated 2 blocks
Request 6: Allocated 2 blocks
Request 7: Allocated 2 blocks
Request 8: Allocated 2 blocks
Request 9: Allocated 2 blocks
Request 10: Allocated 2 blocks
Request 11: Allocated 2 blocks
Request 12: Allocated 2 blocks
Request 13: Failed to allocate blocks - Failed to allocate 2 blocks for request 13

Simulation stopped. Total requests: 13
Total blocks allocated: 24

Freeing all blocks...
Freed 24 blocks

Final Statistics:
  - Total blocks allocated: 24
  - Total blocks freed: 24
  - Current allocated blocks: 0
  - Memory fragmentation rate: 12.50%

class StatsCollector:
    """Collects statistics about the block manager."""
    
    def __init__(self):
        """Initialize the stats collector."""
        self.total_blocks_allocated = 0
        self.total_blocks_freed = 0
        self.current_allocated_blocks = 0
        self.memory_fragmentation_rate = 0.0
        self.compression_ratio = 0.0
        self.total_compressed_size = 0
        self.total_uncompressed_size = 0
        self.start_time = time.time()
    
    def record_blocks_allocated(self, blocks: List[Block]) -> None:
        """Record that blocks were allocated."""
        num_blocks = len(blocks)
        self.total_blocks_allocated += num_blocks
        self.current_allocated_blocks += num_blocks
        
        # 更新内存碎片率
        self._update_fragmentation_rate()
    
    def record_blocks_freed(self, blocks: List[Block]) -> None:
        """Record that blocks were freed."""
        num_blocks = len(blocks)
        self.total_blocks_freed += num_blocks
        self.current_allocated_blocks -= num_blocks
        
        # 更新内存碎片率
        self._update_fragmentation_rate()
    
    def record_compression(self, uncompressed_size: int, compressed_size: int) -> None:
        """Record compression statistics."""
        self.total_uncompressed_size += uncompressed_size
        self.total_compressed_size += compressed_size
        
        # 更新压缩率
        if self.total_uncompressed_size > 0:
            self.compression_ratio = self.total_compressed_size / self.total_uncompressed_size
    
    def _update_fragmentation_rate(self) -> None:
        """Update the memory fragmentation rate."""
        # 简化的碎片率计算
        # 实际实现中会考虑空闲块的分布
        self.memory_fragmentation_rate = random.uniform(0.05, 0.2)  # 模拟碎片率
    
    def get_summary(self) -> dict:
        """Get a summary of the statistics."""
        elapsed_time = time.time() - self.start_time
        
        return {
            "total_blocks_allocated": self.total_blocks_allocated,
            "total_blocks_freed": self.total_blocks_freed,
            "current_allocated_blocks": self.current_allocated_blocks,
            "memory_fragmentation_rate": self.memory_fragmentation_rate,
            "compression_ratio": self.compression_ratio,
            "total_uncompressed_size": self.total_uncompressed_size,
            "total_compressed_size": self.total_compressed_size,
            "elapsed_time": elapsed_time,
        }