flowchart TB
    subgraph ExecutorModule["Executor 模块"]
        ExecutorBase["ExecutorBase<br/>抽象基类"]
        
        subgraph Implementations["具体实现"]
            UniProc["UniProcExecutor<br/>单进程执行器"]
            MultiProc["MultiProcExecutor<br/>多进程执行器"] 
            Ray["RayDistributedExecutor<br/>Ray分布式执行器"]
            GPU["GPUExecutor<br/>GPU执行器"]
        end
        
        subgraph DistributedBase["分布式基类"]
            DistributedExec["DistributedExecutorBase<br/>分布式执行器基类"]
        end
        
        subgraph WorkerManagement["Worker 管理"]
            WorkerPool["Worker Pool<br/>工作节点池"]
            RPC["RPC Interface<br/>远程过程调用"]
            HealthCheck["Health Check<br/>健康检查"]
        end
    end
    
    subgraph External["外部组件"]
        Engine["LLM Engine"]
        Worker["Worker Base"]
        VllmConfig["VllmConfig"]
        Ray["Ray Framework"]
    end
    
    Engine --> ExecutorBase
    ExecutorBase --> Implementations
    DistributedExec --> MultiProc
    DistributedExec --> Ray
    Implementations --> WorkerManagement
    WorkerManagement --> Worker
    VllmConfig --> ExecutorBase
    Ray --> WorkerPool
    
    classDef executor fill:#e1f5fe
    classDef impl fill:#f3e5f5
    classDef worker fill:#e8f5e8
    classDef external fill:#fff3e0
    
    class ExecutorBase,DistributedExec executor
    class UniProc,MultiProc,Ray,GPU impl
    class WorkerPool,RPC,HealthCheck worker
    class Engine,Worker,VllmConfig,Ray external

def execute_model(
    self, 
    execute_model_req: ExecuteModelRequest
) -> List[SamplerOutput]:
    """
    执行模型推理的核心方法
    协调所有 Worker 节点完成推理任务
    """
    # 1) 启动并行工作循环（如果未启动）
    if self.parallel_worker_tasks is None:
        self.parallel_worker_tasks = self._run_workers(
            "start_worker_execution_loop",
            async_run_tensor_parallel_workers_only=True
        )
    
    # 2) 在驱动节点执行模型推理
    driver_outputs = self._driver_execute_model(execute_model_req)
    
    # 3) 返回采样结果（仅驱动节点返回结果）
    return driver_outputs

def collective_rpc(
    self,
    method: Union[str, Callable],
    timeout: Optional[float] = None,
    args: tuple = (),
    kwargs: Optional[dict[str, Any]] = None
) -> list[Any]:
    """
    在所有 Worker 节点上执行集合远程过程调用
    """
    # 1) 参数验证和预处理
    if kwargs is None:
        kwargs = {}
    
    # 2) 构建RPC调用参数
    rpc_args = (method, args, kwargs)
    
    # 3) 并行调用所有Worker节点
    futures = []
    for worker in self.workers:
        future = worker.execute_method_async(*rpc_args, timeout=timeout)
        futures.append(future)
    
    # 4) 等待所有调用完成并收集结果
    results = []
    for future in futures:
        result = future.get(timeout=timeout)
        results.append(result)
    
    return results

def determine_num_available_blocks(self) -> Tuple[int, int]:
    """
    估算可用的 GPU 和 CPU 内存块数量
    用于优化批处理大小和缓存分配
    """
    # 1) 获取设备内存信息
    gpu_memory = torch.cuda.get_device_properties(0).total_memory
    available_gpu_memory = gpu_memory - self._get_memory_usage()
    
    # 2) 计算单个KV缓存块的内存大小
    block_size = self.cache_config.block_size
    kv_cache_block_size = self._calculate_kv_cache_block_size()
    
    # 3) 估算可分配的块数量
    num_gpu_blocks = available_gpu_memory // kv_cache_block_size
    num_cpu_blocks = self._estimate_cpu_blocks()
    
    # 4) 应用安全边界和预留内存
    num_gpu_blocks = int(num_gpu_blocks * 0.95)  # 预留5%内存
    
    return num_gpu_blocks, num_cpu_blocks