class WorkerProc:
    """Wrapper that runs one Worker in a separate process."""

    READY_STR = "READY"
    rpc_broadcast_mq: MessageQueue | None
    worker_response_mq: MessageQueue | None
	@instrument(span_name="Worker init")
    def __init__(...):
	    self.rank = rank
        wrapper = WorkerWrapperBase(rpc_rank=local_rank, global_rank=rank)
        ...
        wrapper.init_worker(all_kwargs)
        self.worker = wrapper
        ...
        self.worker.init_device()
	    if envs.VLLM_ELASTIC_EP_SCALE_UP_LAUNCH:
            self.worker.elastic_ep_execute("load_model")
        else:
            self.worker.load_model()
        。。

@instrument(span_name="Init device")
def init_device(self):
	if self.device_config.device_type == "cuda":
		...
		# Ray 会设置 NCCL_ASYNC_ERROR_HANDLING，但这个环境变量会导致 CUDA graph
		# 构建时出现异常。CUDA graph 需要同步执行，而该变量会引入异步错误处理，
		# 两者可能引起冲突。
		os.environ.pop("NCCL_ASYNC_ERROR_HANDLING", None)
		...
		# - Ray/external_launcher 场景：这些分布式执行器自己管理GPU映射
        # - 多节点场景(nnodes_within_dp > 1)：每个节点有独立的GPU集合，映射逻辑不同
        # - Ray作为DP后端：Ray的resource pool处理GPU分配
		if (
                parallel_config.distributed_executor_backend
                not in ("ray", "external_launcher")
                and parallel_config.data_parallel_backend != "ray"
                and parallel_config.nnodes_within_dp == 1 # 单节点场景
            ):
            # local DP rank 表示在当前节点内的数据并行编号，而 global rank 可能
            # 跨节点。在单节点场景下，GPU映射只看节点内的local rank。
            dp_local_rank = self.parallel_config.data_parallel_rank_local
            if dp_local_rank is None:
                dp_local_rank = self.parallel_config.data_parallel_index
            # DP副本 0 (dp_local_rank=0)
            # original_local_rank=0 → GPU 0 + 0×2 = GPU 0
            # original_local_rank=0 → GPU 0 + 1×2 = GPU 2 ← 偏移
            # 偏移是为了计算出实际的rank信息，后续用来初始化device
            self.local_rank += dp_local_rank * tp_pp_world_size
            。。
            
		self.device = torch.device(f"cuda:{self.local_rank}")
		# PyTorch的设备API演进：从 torch.cuda.set_device() 到
        # torch.accelerator.set_device_index()。 后续可以不用再手动指定，
		torch.accelerator.set_device_index(self.device)
		...
		# 初始化分布式推理所需的所有环境，包括通信组信息
		# 优先于内存快照处理逻辑，NCCL在初始化会分配内部缓存区，提前初始化用于保证显存计算的准确性
		init_worker_distributed_environment(
			self.vllm_config,
			self.rank,
			self.distributed_init_method,
			self.local_rank,
			current_platform.dist_backend,
		)
		# 1. gc.collect() 回收Python层的垃圾对象，释放可能的GPU引用
        # 2. empty_cache() 释放PyTorch缓存的显存（包括NCCL缓冲区）
        # 得到一个"干净"的显存状态，作为基准线
        gc.collect()
        torch.accelerator.empty_cache()
        # 用于后续计算 KV cache 可用显存。
        # init_snapshot 记录当前可用显存，request_memory 根据模型配置
        # 计算需要预留的 KV cache 大小。
        self.init_snapshot = init_snapshot = MemorySnapshot(device=self.device)
        self.requested_memory = request_memory(init_snapshot, self.cache_config)
    
	# 最后初始化modelrunner，需要依赖设备、分布式通信环境，同时通过快照可以计算出可以分配的kv cache显存大小
	if self.use_v2_model_runner:
		...
		self.model_runner: GPUModelRunner = GPUModelRunnerV2(  # type: ignore
			self.vllm_config, self.device
		)
	else:
		...
		self.model_runner = GPUModelRunnerV1(self.vllm_config, self.device)
	...

import torch


# Simple module for demonstration
class MyModule(torch.nn.Module):
    def __init__(self) -> None:
        super().__init__()
        self.param = torch.nn.Parameter(torch.rand(3, 4))
        self.linear = torch.nn.Linear(4, 5)

    def forward(self, x):
        return self.linear(x + self.param).clamp(min=0.0, max=1.0)


module = MyModule()

from torch.fx import symbolic_trace

# Symbolic tracing frontend - captures the semantics of the module
symbolic_traced: torch.fx.GraphModule = symbolic_trace(module)

# High-level intermediate representation (IR) - Graph representation
# 由一个列表组成 代表函数输入、调用点（函数、方法、 或 torch.nn.Module 实例），以及返回值。
print(symbolic_traced.graph)
"""
graph():
    %x : [num_users=1] = placeholder[target=x]
    %param : [num_users=1] = get_attr[target=param]
    %add : [num_users=1] = call_function[target=operator.add](args = (%x, %param), kwargs = {})
    %linear : [num_users=1] = call_module[target=linear](args = (%add,), kwargs = {})
    %clamp : [num_users=1] = call_method[target=clamp](args = (%linear,), kwargs = {min: 0.0, max: 1.0})
    return clamp
"""

# Code generation - valid Python code
# 使 FX 成为 Python 到 Python（或 模块到模块）转换工具包。对于每个 Graph IR，我们可以 创建与图语义匹配的有效 Python 代码。
print(symbolic_traced.code)
"""
def forward(self, x):
    param = self.param
    add = x + param;  x = param = None
    linear = self.linear(add);  add = None
    clamp = linear.clamp(min = 0.0, max = 1.0);  linear = None
    return clamp
"""

def _process_engine_step(self) -> bool:
    """Called only when there are unfinished local requests."""

    # Step the engine core.
    outputs, model_executed = self.step_fn()
    # Put EngineCoreOutputs into the output queue.
    for output in outputs.items() if outputs else ():
        self.output_queue.put_nowait(output)
    # Post-step hook.
    self.post_step(model_executed)

    return model_executed

def step(self) -> tuple[dict[int, EngineCoreOutputs], bool]:
  if not self.scheduler.has_requests():
      return {}, False
  
  scheduler_output = self.scheduler.schedule()
  # 通过FutureWrapper进行异步包装（复用异步调度的部分逻辑， 在同步调度逻辑里面会等待结果返回）
  # 
  future = self.model_executor.execute_model(scheduler_output, non_block=True)
  # 用于支持结构化输出等
  grammar_output = self.scheduler.get_grammar_bitmask(scheduler_output)
  with self.log_error_detail(scheduler_output):
	  # 同步
      model_output = future.result()
      if model_output is None:
          model_output = self.model_executor.sample_tokens(grammar_output)

  # 处理整个过程中abort的请求
  self._process_aborts_queue()
  engine_core_outputs = self.scheduler.update_from_output(
      scheduler_output, model_output
  )
  return engine_core_outputs, scheduler_output.total_num_scheduled_tokens > 0