深入 vLLM EPD：Disaggregated Encoder / Encoder-Prefill/Decode 源码拆解

vllm/
├── config/
│   └── ec_transfer.py                          # ★ ECTransferConfig 配置类
├── distributed/
│   └── ec_transfer/                            # ★ EC Transfer 核心模块
│       ├── __init__.py                         #   公开 API
│       ├── ec_transfer_state.py                #   全局单例状态管理
│       └── ec_connector/
│           ├── __init__.py
│           ├── base.py                         # ★ ECConnectorBase 抽象基类
│           ├── factory.py                      # ★ ECConnectorFactory 工厂
│           └── example_connector.py            # ★ ECExampleConnector 参考实现
├── v1/
│   ├── worker/
│   │   ├── gpu_model_runner.py                 #   集成 EC Mixin
│   │   ├── ec_connector_model_runner_mixin.py  # ★ EC 生命周期管理 Mixin
│   │   └── gpu_worker.py                       #   ensure_ec_transfer_initialized()
│   ├── core/
│   │   ├── sched/scheduler.py                  #   调度器 EC 集成
│   │   ├── encoder_cache_manager.py            #   Encoder Cache 管理器
│   │   └── sched/output.py                     #   SchedulerOutput 含 EC 元数据
│   ├── engine/core.py                          #   is_ec_consumer 判断（是否使用 sampler）
│   ├── executor/ray_executor.py                #   uses_sampler 门控
│   └── outputs.py                              #   ECConnectorOutput + make_empty_encoder_model_runner_output
├── config/multimodal.py                        #   mm_encoder_only 配置
└── engine/arg_utils.py                         #   CLI 参数

examples/online_serving/disaggregated_encoder/
├── disagg_epd_proxy.py                         # ★ EPD Proxy 路由代理
├── disagg_1e1pd_example.sh                     #   E+PD 部署脚本
├── disagg_1e1p1d_example.sh                    #   E+P+D 部署脚本
└── README.md

tests/v1/ec_connector/
├── unit/test_ec_example_connector.py           #   单元测试
└── integration/
    ├── test_epd_correctness.py                 #   正确性验证
    └── run_epd_correctness_test.sh             #   集成测试脚本

vllm/
├── config/
│   └── ec_transfer.py                          # ★ ECTransferConfig 配置类
├── distributed/
│   └── ec_transfer/                            # ★ EC Transfer 核心模块
│       ├── __init__.py                         #   公开 API
│       ├── ec_transfer_state.py                #   全局单例状态管理
│       └── ec_connector/
│           ├── __init__.py
│           ├── base.py                         # ★ ECConnectorBase 抽象基类
│           ├── factory.py                      # ★ ECConnectorFactory 工厂
│           └── example_connector.py            # ★ ECExampleConnector 参考实现
├── v1/
│   ├── worker/
│   │   ├── gpu_model_runner.py                 #   集成 EC Mixin
│   │   ├── ec_connector_model_runner_mixin.py  # ★ EC 生命周期管理 Mixin
│   │   └── gpu_worker.py                       #   ensure_ec_transfer_initialized()
│   ├── core/
│   │   ├── sched/scheduler.py                  #   调度器 EC 集成
│   │   ├── encoder_cache_manager.py            #   Encoder Cache 管理器
│   │   └── sched/output.py                     #   SchedulerOutput 含 EC 元数据
│   ├── engine/core.py                          #   is_ec_consumer 判断（是否使用 sampler）
│   ├── executor/ray_executor.py                #   uses_sampler 门控
│   └── outputs.py                              #   ECConnectorOutput + make_empty_encoder_model_runner_output
├── config/multimodal.py                        #   mm_encoder_only 配置
└── engine/arg_utils.py                         #   CLI 参数

examples/online_serving/disaggregated_encoder/
├── disagg_epd_proxy.py                         # ★ EPD Proxy 路由代理
├── disagg_1e1pd_example.sh                     #   E+PD 部署脚本
├── disagg_1e1p1d_example.sh                    #   E+P+D 部署脚本
└── README.md

tests/v1/ec_connector/
├── unit/test_ec_example_connector.py           #   单元测试
└── integration/
    ├── test_epd_correctness.py                 #   正确性验证
    └── run_epd_correctness_test.sh             #   集成测试脚本

ECProducer = Literal["ec_producer", "ec_both"]
ECConsumer = Literal["ec_consumer", "ec_both"]
ECRole = Literal[ECProducer, ECConsumer]
 
@config
class ECTransferConfig:
    """分布式 Encoder Cache 传输配置"""
 
    ec_connector: str | None = None
    """EC 连接器实现名称，如 "ECExampleConnector" """
 
    engine_id: str | None = None
    """引擎 ID，用于 EC 传输标识。自动生成 UUID。"""
 
    ec_buffer_device: str | None = "cuda"
    """EC 连接器缓冲区设备。目前仅支持 'cuda'。"""
 
    ec_buffer_size: float = 1e9
    """缓冲区大小（字节），推荐 1e9（约 1GB）。"""
 
    ec_role: ECRole | None = None
    """当前实例角色：'ec_producer' | 'ec_consumer' | 'ec_both'"""
 
    ec_rank: int | None = None
    """传输集群中的 rank。典型值：0=Encoder，1=PD。"""
 
    ec_parallel_size: int = 1
    """并行实例数。"""
 
    ec_ip: str = "127.0.0.1"
    """连接器 IP 地址。"""
 
    ec_port: int = 14579
    """连接器端口。"""
 
    ec_connector_extra_config: dict[str, Any] = field(default_factory=dict)
    """连接器特定的额外配置。"""
 
    ec_connector_module_path: str | None = None
    """动态加载自定义连接器的 Python 模块路径。"""

ECProducer = Literal["ec_producer", "ec_both"]
ECConsumer = Literal["ec_consumer", "ec_both"]
ECRole = Literal[ECProducer, ECConsumer]
 
@config
class ECTransferConfig:
    """分布式 Encoder Cache 传输配置"""
 
    ec_connector: str | None = None
    """EC 连接器实现名称，如 "ECExampleConnector" """
 
    engine_id: str | None = None
    """引擎 ID，用于 EC 传输标识。自动生成 UUID。"""
 
    ec_buffer_device: str | None = "cuda"
    """EC 连接器缓冲区设备。目前仅支持 'cuda'。"""
 
    ec_buffer_size: float = 1e9
    """缓冲区大小（字节），推荐 1e9（约 1GB）。"""
 
    ec_role: ECRole | None = None
    """当前实例角色：'ec_producer' | 'ec_consumer' | 'ec_both'"""
 
    ec_rank: int | None = None
    """传输集群中的 rank。典型值：0=Encoder，1=PD。"""
 
    ec_parallel_size: int = 1
    """并行实例数。"""
 
    ec_ip: str = "127.0.0.1"
    """连接器 IP 地址。"""
 
    ec_port: int = 14579
    """连接器端口。"""
 
    ec_connector_extra_config: dict[str, Any] = field(default_factory=dict)
    """连接器特定的额外配置。"""
 
    ec_connector_module_path: str | None = None
    """动态加载自定义连接器的 Python 模块路径。"""

@property
def is_ec_transfer_instance(self) -> bool:
    """是否为 EC 传输实例（connector 和 role 均已配置）"""
    return self.ec_connector is not None and self.ec_role in get_args(ECRole)
 
@property
def is_ec_producer(self) -> bool:
    """是否为生产者（Encoder 端）"""
    return self.ec_connector is not None and self.ec_role in get_args(ECProducer)
 
@property
def is_ec_consumer(self) -> bool:
    """是否为消费者（PD 端）"""
    return self.ec_connector is not None and self.ec_role in get_args(ECConsumer)
 
def get_from_extra_config(self, key, default) -> Any:
    """从 extra_config 中获取配置"""
    return self.ec_connector_extra_config.get(key, default)

@property
def is_ec_transfer_instance(self) -> bool:
    """是否为 EC 传输实例（connector 和 role 均已配置）"""
    return self.ec_connector is not None and self.ec_role in get_args(ECRole)
 
@property
def is_ec_producer(self) -> bool:
    """是否为生产者（Encoder 端）"""
    return self.ec_connector is not None and self.ec_role in get_args(ECProducer)
 
@property
def is_ec_consumer(self) -> bool:
    """是否为消费者（PD 端）"""
    return self.ec_connector is not None and self.ec_role in get_args(ECConsumer)
 
def get_from_extra_config(self, key, default) -> Any:
    """从 extra_config 中获取配置"""
    return self.ec_connector_extra_config.get(key, default)

vllm serve $MODEL \
    --enforce-eager              # 必须：禁用 CUDA Graph
    --no-enable-prefix-caching   # 必须：禁用前缀缓存
    --max-num-batched-tokens 114688  # 推荐：大 batch 预算
    --mm-encoder-only            # 可选：跳过语言模型，节省显存
    --gpu-memory-utilization 0.01  # Encoder 很轻量，极低显存占用

vllm serve $MODEL \
    --enforce-eager              # 必须：禁用 CUDA Graph
    --no-enable-prefix-caching   # 必须：禁用前缀缓存
    --max-num-batched-tokens 114688  # 推荐：大 batch 预算
    --mm-encoder-only            # 可选：跳过语言模型，节省显存
    --gpu-memory-utilization 0.01  # Encoder 很轻量，极低显存占用

class MultiModalConfig:
    mm_encoder_only: bool = False
    """
    When enabled, skips the language component of the model.
    This is usually only valid in disaggregated Encoder process.
    """

class MultiModalConfig:
    mm_encoder_only: bool = False
    """
    When enabled, skips the language component of the model.
    This is usually only valid in disaggregated Encoder process.
    """

class ECConnectorBase(ABC):
    def __init__(self, vllm_config: "VllmConfig", role: ECConnectorRole):
        self._connector_metadata: ECConnectorMetadata | None = None
        self._vllm_config = vllm_config
        self._role = role
        # 从 ec_transfer_config 获取角色
        self._is_producer = vllm_config.ec_transfer_config.is_ec_producer
        self._is_consumer = vllm_config.ec_transfer_config.is_ec_consumer

class ECConnectorBase(ABC):
    def __init__(self, vllm_config: "VllmConfig", role: ECConnectorRole):
        self._connector_metadata: ECConnectorMetadata | None = None
        self._vllm_config = vllm_config
        self._role = role
        # 从 ec_transfer_config 获取角色
        self._is_producer = vllm_config.ec_transfer_config.is_ec_producer
        self._is_consumer = vllm_config.ec_transfer_config.is_ec_consumer

    # ---------- 元数据生命周期 ----------
 
    def bind_connector_metadata(self, connector_metadata: ECConnectorMetadata) -> None:
        """
        在每次模型执行前由 Model Runner 调用。
        绑定调度器传来的元数据，用于运行时的 EC 加载决策。
        """
        self._connector_metadata = connector_metadata
 
    def clear_connector_metadata(self) -> None:
        """模型执行后清理元数据。"""
        self._connector_metadata = None
 
    # ---------- 缓存操作 ----------
 
    @abstractmethod
    def start_load_caches(
        self, encoder_cache: dict[str, torch.Tensor], **kwargs
    ) -> None:
        """
        从连接器加载 Encoder Cache 到 vLLM 的 encoder_cache 字典。
 
        在 _gather_mm_embeddings 之前调用。
        encoder_cache 的 key 是 mm_hash（多模态数据散列），value 是编码结果张量。
 
        Args:
            encoder_cache: mm_hash → encoder output tensor 的字典
        """
 
    @abstractmethod
    def save_caches(
        self, encoder_cache: dict[str, torch.Tensor], mm_hash: str, **kwargs
    ) -> None:
        """
        将 Encoder Cache 保存到连接器（共享存储/远程服务）。
 
        在 Encoder 执行完毕后调用。
 
        Args:
            encoder_cache: mm_hash → encoder output tensor 的字典
            mm_hash: 当前要保存的多模态数据的散列值
        """
 
    def get_finished(
        self, finished_req_ids: set[str]
    ) -> tuple[set[str] | None, set[str] | None]:
        """
        返回异步传输完成的请求 ID。
        Returns: (已完成发送的ID集合, 已完成接收的ID集合)
        """
        return None, None
 
    def register_caches(self, ec_caches: dict[str, torch.Tensor]):
        """注册 EC 缓存（P2P 特性预留）。"""
        return

    # ---------- 元数据生命周期 ----------
 
    def bind_connector_metadata(self, connector_metadata: ECConnectorMetadata) -> None:
        """
        在每次模型执行前由 Model Runner 调用。
        绑定调度器传来的元数据，用于运行时的 EC 加载决策。
        """
        self._connector_metadata = connector_metadata
 
    def clear_connector_metadata(self) -> None:
        """模型执行后清理元数据。"""
        self._connector_metadata = None
 
    # ---------- 缓存操作 ----------
 
    @abstractmethod
    def start_load_caches(
        self, encoder_cache: dict[str, torch.Tensor], **kwargs
    ) -> None:
        """
        从连接器加载 Encoder Cache 到 vLLM 的 encoder_cache 字典。
 
        在 _gather_mm_embeddings 之前调用。
        encoder_cache 的 key 是 mm_hash（多模态数据散列），value 是编码结果张量。
 
        Args:
            encoder_cache: mm_hash → encoder output tensor 的字典
        """
 
    @abstractmethod
    def save_caches(
        self, encoder_cache: dict[str, torch.Tensor], mm_hash: str, **kwargs
    ) -> None:
        """
        将 Encoder Cache 保存到连接器（共享存储/远程服务）。
 
        在 Encoder 执行完毕后调用。
 
        Args:
            encoder_cache: mm_hash → encoder output tensor 的字典
            mm_hash: 当前要保存的多模态数据的散列值
        """
 
    def get_finished(
        self, finished_req_ids: set[str]
    ) -> tuple[set[str] | None, set[str] | None]:
        """
        返回异步传输完成的请求 ID。
        Returns: (已完成发送的ID集合, 已完成接收的ID集合)
        """
        return None, None
 
    def register_caches(self, ec_caches: dict[str, torch.Tensor]):
        """注册 EC 缓存（P2P 特性预留）。"""
        return

    @abstractmethod
    def has_cache_item(self, identifier: str) -> bool:
        """
        检查指定多模态数据的 Encoder Cache 是否存在于外部存储。
 
        Args:
            identifier: 多模态数据标识（通常是 mm_hash）
        Returns:
            True 如果缓存存在
        """
 
    @abstractmethod
    def update_state_after_alloc(self, request: "Request", index: int):
        """
        Encoder Cache 分配后更新连接器状态。
 
        调度器分配了缓存位置后调用，连接器根据角色决定是否需要加载缓存。
 
        Args:
            request: 请求对象（含 mm_features）
            index: 当前多模态输入在请求中的索引
        """
 
    @abstractmethod
    def build_connector_meta(
        self, scheduler_output: SchedulerOutput
    ) -> ECConnectorMetadata:
        """
        为当前调度步骤构建 Worker 所需的元数据。
 
        注意：不应修改 scheduler_output 的字段。
        调用后重置连接器内部状态。
        """
 
    def update_connector_output(self, connector_output: ECConnectorOutput):
        """处理从 Worker 返回的连接器输出。"""
        return
 
    def request_finished(
        self, request: "Request"
    ) -> tuple[bool, dict[str, Any] | None]:
        """
        请求完成回调，在 Encoder Cache 被释放前调用。
 
        Returns:
            (是否延迟释放, 额外参数)
            True 表示 cache 正在异步发送，应等 get_finished 确认后再释放。
        """
        return False, None

    @abstractmethod
    def has_cache_item(self, identifier: str) -> bool:
        """
        检查指定多模态数据的 Encoder Cache 是否存在于外部存储。
 
        Args:
            identifier: 多模态数据标识（通常是 mm_hash）
        Returns:
            True 如果缓存存在
        """
 
    @abstractmethod
    def update_state_after_alloc(self, request: "Request", index: int):
        """
        Encoder Cache 分配后更新连接器状态。
 
        调度器分配了缓存位置后调用，连接器根据角色决定是否需要加载缓存。
 
        Args:
            request: 请求对象（含 mm_features）
            index: 当前多模态输入在请求中的索引
        """
 
    @abstractmethod
    def build_connector_meta(
        self, scheduler_output: SchedulerOutput
    ) -> ECConnectorMetadata:
        """
        为当前调度步骤构建 Worker 所需的元数据。
 
        注意：不应修改 scheduler_output 的字段。
        调用后重置连接器内部状态。
        """
 
    def update_connector_output(self, connector_output: ECConnectorOutput):
        """处理从 Worker 返回的连接器输出。"""
        return
 
    def request_finished(
        self, request: "Request"
    ) -> tuple[bool, dict[str, Any] | None]:
        """
        请求完成回调，在 Encoder Cache 被释放前调用。
 
        Returns:
            (是否延迟释放, 额外参数)
            True 表示 cache 正在异步发送，应等 get_finished 确认后再释放。
        """
        return False, None

class ECConnectorRole(enum.Enum):
    SCHEDULER = 0  # 运行于调度器进程，负责决策
    WORKER = 1     # 运行于 Worker 进程，负责执行

class ECConnectorRole(enum.Enum):
    SCHEDULER = 0  # 运行于调度器进程，负责决策
    WORKER = 1     # 运行于 Worker 进程，负责执行

class ECConnectorMetadata(ABC):
    """
    抽象元数据，用于 Scheduler ECConnector → Worker ECConnector 的通信。
    由 build_connector_meta() 构建，通过 SchedulerOutput 传递到 Worker。
    """
    pass

class ECConnectorMetadata(ABC):
    """
    抽象元数据，用于 Scheduler ECConnector → Worker ECConnector 的通信。
    由 build_connector_meta() 构建，通过 SchedulerOutput 传递到 Worker。
    """
    pass

class ECConnectorFactory:
    _registry: dict[str, Callable[[], type[ECConnectorBase]]] = {}
 
    @classmethod
    def register_connector(cls, name: str, module_path: str, class_name: str) -> None:
        """注册懒加载的 EC 连接器。"""
        def loader() -> type[ECConnectorBase]:
            module = importlib.import_module(module_path)
            return getattr(module, class_name)
        cls._registry[name] = loader
 
    @classmethod
    def create_connector(
        cls, config: "VllmConfig", role: ECConnectorRole,
    ) -> ECConnectorBase:
        """根据配置创建连接器实例。"""
        ec_transfer_config = config.ec_transfer_config
        connector_cls = cls.get_connector_class(ec_transfer_config)
        return connector_cls(config, role)
 
    @classmethod
    def get_connector_class(
        cls, ec_transfer_config: "ECTransferConfig"
    ) -> type[ECConnectorBase]:
        """按名称或动态路径获取连接器类。"""
        connector_name = ec_transfer_config.ec_connector
        if connector_name in cls._registry:
            return cls._registry[connector_name]()
        # 支持动态模块加载
        connector_module_path = ec_transfer_config.ec_connector_module_path
        if connector_module_path is not None:
            module = importlib.import_module(connector_module_path)
            return getattr(module, connector_name)
        raise ValueError(f"Unsupported connector type: {connector_name}")
 
# 注册内置连接器
ECConnectorFactory.register_connector(
    "ECExampleConnector",
    "vllm.distributed.ec_transfer.ec_connector.example_connector",
    "ECExampleConnector",
)

class ECConnectorFactory:
    _registry: dict[str, Callable[[], type[ECConnectorBase]]] = {}
 
    @classmethod
    def register_connector(cls, name: str, module_path: str, class_name: str) -> None:
        """注册懒加载的 EC 连接器。"""
        def loader() -> type[ECConnectorBase]:
            module = importlib.import_module(module_path)
            return getattr(module, class_name)
        cls._registry[name] = loader
 
    @classmethod
    def create_connector(
        cls, config: "VllmConfig", role: ECConnectorRole,
    ) -> ECConnectorBase:
        """根据配置创建连接器实例。"""
        ec_transfer_config = config.ec_transfer_config
        connector_cls = cls.get_connector_class(ec_transfer_config)
        return connector_cls(config, role)
 
    @classmethod
    def get_connector_class(
        cls, ec_transfer_config: "ECTransferConfig"
    ) -> type[ECConnectorBase]:
        """按名称或动态路径获取连接器类。"""
        connector_name = ec_transfer_config.ec_connector
        if connector_name in cls._registry:
            return cls._registry[connector_name]()
        # 支持动态模块加载
        connector_module_path = ec_transfer_config.ec_connector_module_path
        if connector_module_path is not None:
            module = importlib.import_module(connector_module_path)
            return getattr(module, connector_name)
        raise ValueError(f"Unsupported connector type: {connector_name}")
 
# 注册内置连接器
ECConnectorFactory.register_connector(
    "ECExampleConnector",
    "vllm.distributed.ec_transfer.ec_connector.example_connector",
    "ECExampleConnector",
)

_EC_CONNECTOR_AGENT: ECConnectorBase | None = None
 
def get_ec_transfer() -> ECConnectorBase:
    """获取全局 EC Connector 实例（仅在 Worker 进程中可用）"""
    assert _EC_CONNECTOR_AGENT is not None, \
        "disaggregated EC cache is not initialized"
    return _EC_CONNECTOR_AGENT
 
def has_ec_transfer() -> bool:
    """检查 EC 传输是否已初始化"""
    return _EC_CONNECTOR_AGENT is not None
 
def ensure_ec_transfer_initialized(vllm_config: "VllmConfig") -> None:
    """
    在 Worker 进程启动时调用，初始化全局 EC Connector。
    仅在 ec_transfer_config 已配置且 _EC_CONNECTOR_AGENT 未初始化时执行。
    """
    global _EC_CONNECTOR_AGENT
    if vllm_config.ec_transfer_config is None:
        return
    if (vllm_config.ec_transfer_config.is_ec_transfer_instance
            and _EC_CONNECTOR_AGENT is None):
        _EC_CONNECTOR_AGENT = ECConnectorFactory.create_connector(
            config=vllm_config, role=ECConnectorRole.WORKER
        )

_EC_CONNECTOR_AGENT: ECConnectorBase | None = None
 
def get_ec_transfer() -> ECConnectorBase:
    """获取全局 EC Connector 实例（仅在 Worker 进程中可用）"""
    assert _EC_CONNECTOR_AGENT is not None, \
        "disaggregated EC cache is not initialized"
    return _EC_CONNECTOR_AGENT
 
def has_ec_transfer() -> bool:
    """检查 EC 传输是否已初始化"""
    return _EC_CONNECTOR_AGENT is not None
 
def ensure_ec_transfer_initialized(vllm_config: "VllmConfig") -> None:
    """
    在 Worker 进程启动时调用，初始化全局 EC Connector。
    仅在 ec_transfer_config 已配置且 _EC_CONNECTOR_AGENT 未初始化时执行。
    """
    global _EC_CONNECTOR_AGENT
    if vllm_config.ec_transfer_config is None:
        return
    if (vllm_config.ec_transfer_config.is_ec_transfer_instance
            and _EC_CONNECTOR_AGENT is None):
        _EC_CONNECTOR_AGENT = ECConnectorFactory.create_connector(
            config=vllm_config, role=ECConnectorRole.WORKER
        )

from vllm.distributed.ec_transfer.ec_transfer_state import (
    ensure_ec_transfer_initialized,
    get_ec_transfer,
    has_ec_transfer,
)

from vllm.distributed.ec_transfer.ec_transfer_state import (
    ensure_ec_transfer_initialized,
    get_ec_transfer,
    has_ec_transfer,
)

{shared_storage_path}/
├── {mm_hash_1}/
│   └── encoder_cache.safetensors    # 保存的 encoder 输出张量
├── {mm_hash_2}/
│   └── encoder_cache.safetensors
└── ...

{shared_storage_path}/
├── {mm_hash_1}/
│   └── encoder_cache.safetensors    # 保存的 encoder 输出张量
├── {mm_hash_2}/
│   └── encoder_cache.safetensors
└── ...

@dataclass
class MMMeta:
    """单个多模态项的元数据"""
    mm_hash: str       # 多模态数据的哈希值
    num_token: int     # encoder 输出的 token 数
 
    @staticmethod
    def make_meta(mm_hash, num_token) -> "MMMeta":
        return MMMeta(mm_hash=mm_hash, num_token=num_token)
 
@dataclass
class ECExampleConnectorMetadata(ECConnectorMetadata):
    """Scheduler → Worker 的元数据：需要加载的多模态项列表"""
    mm_datas: list[MMMeta]
 
    def __init__(self):
        self.mm_datas = []
 
    def add_mm_data(self, mm_data: MMMeta):
        self.mm_datas.append(mm_data)

@dataclass
class MMMeta:
    """单个多模态项的元数据"""
    mm_hash: str       # 多模态数据的哈希值
    num_token: int     # encoder 输出的 token 数
 
    @staticmethod
    def make_meta(mm_hash, num_token) -> "MMMeta":
        return MMMeta(mm_hash=mm_hash, num_token=num_token)
 
@dataclass
class ECExampleConnectorMetadata(ECConnectorMetadata):
    """Scheduler → Worker 的元数据：需要加载的多模态项列表"""
    mm_datas: list[MMMeta]
 
    def __init__(self):
        self.mm_datas = []
 
    def add_mm_data(self, mm_data: MMMeta):
        self.mm_datas.append(mm_data)

def save_caches(self, encoder_cache, mm_hash, **kwargs) -> None:
    """将 Encoder 输出保存到磁盘（仅限 Producer 角色）"""
    # 非 Producer 直接返回
    if not self.is_producer:
        return
    filename = self._generate_filename_debug(mm_hash)
    ec_cache = encoder_cache[mm_hash]
    # 保存为 safetensors 格式（先移到 CPU）
    tensors = {"ec_cache": ec_cache.detach().cpu()}
    safetensors.torch.save_file(tensors, filename)

def save_caches(self, encoder_cache, mm_hash, **kwargs) -> None:
    """将 Encoder 输出保存到磁盘（仅限 Producer 角色）"""
    # 非 Producer 直接返回
    if not self.is_producer:
        return
    filename = self._generate_filename_debug(mm_hash)
    ec_cache = encoder_cache[mm_hash]
    # 保存为 safetensors 格式（先移到 CPU）
    tensors = {"ec_cache": ec_cache.detach().cpu()}
    safetensors.torch.save_file(tensors, filename)

def start_load_caches(self, encoder_cache, **kwargs) -> None:
    """从磁盘加载 Encoder Cache 到内存（Consumer 调用）"""
    from vllm.platforms import current_platform
 
    metadata = self._get_connector_metadata()
    assert isinstance(metadata, ECExampleConnectorMetadata)
    assert encoder_cache is not None
 
    # 遍历需要加载的多模态项
    for mm_data in metadata.mm_datas:
        # 已在内存中则跳过
        if mm_data.mm_hash in encoder_cache:
            continue
        filename = self._generate_filename_debug(mm_data.mm_hash)
        # 加载到当前设备（cuda/xpu）
        ec_cache = safetensors.torch.load_file(
            filename, device=current_platform.device_type
        )["ec_cache"]
        encoder_cache[mm_data.mm_hash] = ec_cache

def start_load_caches(self, encoder_cache, **kwargs) -> None:
    """从磁盘加载 Encoder Cache 到内存（Consumer 调用）"""
    from vllm.platforms import current_platform
 
    metadata = self._get_connector_metadata()
    assert isinstance(metadata, ECExampleConnectorMetadata)
    assert encoder_cache is not None
 
    # 遍历需要加载的多模态项
    for mm_data in metadata.mm_datas:
        # 已在内存中则跳过
        if mm_data.mm_hash in encoder_cache:
            continue
        filename = self._generate_filename_debug(mm_data.mm_hash)
        # 加载到当前设备（cuda/xpu）
        ec_cache = safetensors.torch.load_file(
            filename, device=current_platform.device_type
        )["ec_cache"]
        encoder_cache[mm_data.mm_hash] = ec_cache

def has_cache_item(self, identifier: str) -> bool:
    """调度器侧检查：缓存文件是否存在"""
    return self._found_match_for_mm_data(identifier)
 
def _found_match_for_mm_data(self, mm_hash) -> bool:
    filename = self._generate_filename_debug(mm_hash)
    return os.path.exists(filename)
 
def update_state_after_alloc(self, request: "Request", index: int) -> None:
    """分配后更新：标记需要加载的多模态项"""
    mm_hash = request.mm_features[index].identifier
    # 仅 Consumer 且缓存存在时才标记
    if not self.is_consumer or not self.has_cache_item(mm_hash):
        return
    num_encoder_token = request.get_num_encoder_embeds(index)
    self._mm_datas_need_loads[mm_hash] = num_encoder_token
 
def build_connector_meta(self, scheduler_output) -> ECConnectorMetadata:
    """构建元数据并重置内部状态"""
    meta = ECExampleConnectorMetadata()
    for mm_hash, num_encoder_token in self._mm_datas_need_loads.items():
        meta.add_mm_data(MMMeta.make_meta(mm_hash, num_encoder_token))
    self._mm_datas_need_loads.clear()  # 重置
    return meta

def has_cache_item(self, identifier: str) -> bool:
    """调度器侧检查：缓存文件是否存在"""
    return self._found_match_for_mm_data(identifier)
 
def _found_match_for_mm_data(self, mm_hash) -> bool:
    filename = self._generate_filename_debug(mm_hash)
    return os.path.exists(filename)
 
def update_state_after_alloc(self, request: "Request", index: int) -> None:
    """分配后更新：标记需要加载的多模态项"""
    mm_hash = request.mm_features[index].identifier
    # 仅 Consumer 且缓存存在时才标记
    if not self.is_consumer or not self.has_cache_item(mm_hash):
        return
    num_encoder_token = request.get_num_encoder_embeds(index)
    self._mm_datas_need_loads[mm_hash] = num_encoder_token
 
def build_connector_meta(self, scheduler_output) -> ECConnectorMetadata:
    """构建元数据并重置内部状态"""
    meta = ECExampleConnectorMetadata()
    for mm_hash, num_encoder_token in self._mm_datas_need_loads.items():
        meta.add_mm_data(MMMeta.make_meta(mm_hash, num_encoder_token))
    self._mm_datas_need_loads.clear()  # 重置
    return meta

class Scheduler:
    def __init__(self, ...):
        # 创建 EC Connector（调度器侧）
        self.ec_connector = None
        if self.vllm_config.ec_transfer_config is not None:
            self.ec_connector = ECConnectorFactory.create_connector(
                config=self.vllm_config, role=ECConnectorRole.SCHEDULER
            )
 
        # Encoder Cache 管理器（管理缓存容量和引用计数）
        self.encoder_cache_manager = (
            EncoderDecoderCacheManager(cache_size=encoder_cache_size)
            if self.is_encoder_decoder
            else EncoderCacheManager(cache_size=encoder_cache_size)
        )

class Scheduler:
    def __init__(self, ...):
        # 创建 EC Connector（调度器侧）
        self.ec_connector = None
        if self.vllm_config.ec_transfer_config is not None:
            self.ec_connector = ECConnectorFactory.create_connector(
                config=self.vllm_config, role=ECConnectorRole.SCHEDULER
            )
 
        # Encoder Cache 管理器（管理缓存容量和引用计数）
        self.encoder_cache_manager = (
            EncoderDecoderCacheManager(cache_size=encoder_cache_size)
            if self.is_encoder_decoder
            else EncoderCacheManager(cache_size=encoder_cache_size)
        )

def _try_schedule_encoder_inputs(
    self, request, num_computed_tokens, num_new_tokens,
    encoder_compute_budget, shift_computed_tokens=0,
) -> tuple[list[int], int, int, list[int]]:
    """
    返回:
        encoder_inputs_to_schedule: 需要本地编码的输入索引
        num_new_tokens: 调整后的 decoder token 数
        encoder_compute_budget: 剩余编码预算
        external_load_encoder_input: 需要从外部加载的输入索引  ← EC 核心
    """
    encoder_inputs_to_schedule = []
    external_load_encoder_input = []
 
    for i, mm_feature in enumerate(request.mm_features):
        # ... 判断 encoder 输出是否在当前调度范围内 ...
 
        # 检查是否已在本地缓存
        if self.encoder_cache_manager.check_and_update_cache(request, i):
            continue  # 本地命中，跳过
 
        # ★ EC 核心判断：检查是否存在于远端 Encoder Cache
        if self.ec_connector is not None and self.ec_connector.has_cache_item(
            item_identifier
        ):
            # 远端有缓存 → 走外部加载路径，不消耗本地编码预算
            mm_hashes_to_schedule.add(item_identifier)
            external_load_encoder_input.append(i)
            num_embeds_to_schedule += num_encoder_embeds
            continue
 
        # 远端也没有 → 需要本地编码
        num_embeds_to_schedule += num_encoder_embeds
        encoder_compute_budget -= num_encoder_embeds
        mm_hashes_to_schedule.add(item_identifier)
        encoder_inputs_to_schedule.append(i)
 
    return (
        encoder_inputs_to_schedule,      # 需要本地编码的
        num_new_tokens,
        encoder_compute_budget,
        external_load_encoder_input,     # 需要从外部加载的（EC Transfer）
    )

def _try_schedule_encoder_inputs(
    self, request, num_computed_tokens, num_new_tokens,
    encoder_compute_budget, shift_computed_tokens=0,
) -> tuple[list[int], int, int, list[int]]:
    """
    返回:
        encoder_inputs_to_schedule: 需要本地编码的输入索引
        num_new_tokens: 调整后的 decoder token 数
        encoder_compute_budget: 剩余编码预算
        external_load_encoder_input: 需要从外部加载的输入索引  ← EC 核心
    """
    encoder_inputs_to_schedule = []
    external_load_encoder_input = []
 
    for i, mm_feature in enumerate(request.mm_features):
        # ... 判断 encoder 输出是否在当前调度范围内 ...
 
        # 检查是否已在本地缓存
        if self.encoder_cache_manager.check_and_update_cache(request, i):
            continue  # 本地命中，跳过
 
        # ★ EC 核心判断：检查是否存在于远端 Encoder Cache
        if self.ec_connector is not None and self.ec_connector.has_cache_item(
            item_identifier
        ):
            # 远端有缓存 → 走外部加载路径，不消耗本地编码预算
            mm_hashes_to_schedule.add(item_identifier)
            external_load_encoder_input.append(i)
            num_embeds_to_schedule += num_encoder_embeds
            continue
 
        # 远端也没有 → 需要本地编码
        num_embeds_to_schedule += num_encoder_embeds
        encoder_compute_budget -= num_encoder_embeds
        mm_hashes_to_schedule.add(item_identifier)
        encoder_inputs_to_schedule.append(i)
 
    return (
        encoder_inputs_to_schedule,      # 需要本地编码的
        num_new_tokens,
        encoder_compute_budget,
        external_load_encoder_input,     # 需要从外部加载的（EC Transfer）
    )

# 对于需要本地编码的输入
if encoder_inputs_to_schedule:
    scheduled_encoder_inputs[request_id] = encoder_inputs_to_schedule
    for i in encoder_inputs_to_schedule:
        self.encoder_cache_manager.allocate(request, i)
        if self.ec_connector is not None:
            self.ec_connector.update_state_after_alloc(request, i)
 
# 对于需要从外部加载的输入（EC Transfer）
if external_load_encoder_input:
    for i in external_load_encoder_input:
        self.encoder_cache_manager.allocate(request, i)
        if self.ec_connector is not None:
            self.ec_connector.update_state_after_alloc(request, i)

# 对于需要本地编码的输入
if encoder_inputs_to_schedule:
    scheduled_encoder_inputs[request_id] = encoder_inputs_to_schedule
    for i in encoder_inputs_to_schedule:
        self.encoder_cache_manager.allocate(request, i)
        if self.ec_connector is not None:
            self.ec_connector.update_state_after_alloc(request, i)
 
# 对于需要从外部加载的输入（EC Transfer）
if external_load_encoder_input:
    for i in external_load_encoder_input:
        self.encoder_cache_manager.allocate(request, i)
        if self.ec_connector is not None:
            self.ec_connector.update_state_after_alloc(request, i)

# schedule() 方法中
if self.ec_connector is not None:
    ec_meta: ECConnectorMetadata = self.ec_connector.build_connector_meta(
        scheduler_output
    )
    scheduler_output.ec_connector_metadata = ec_meta

# schedule() 方法中
if self.ec_connector is not None:
    ec_meta: ECConnectorMetadata = self.ec_connector.build_connector_meta(
        scheduler_output
    )
    scheduler_output.ec_connector_metadata = ec_meta

class EncoderCacheManager:
    cache_size: int                          # 总容量
    num_free_slots: int                      # 可用容量
    cached: dict[str, set[str]]              # mm_hash → 引用该 hash 的 request_id 集合
    freeable: OrderedDict[str, int]          # 可回收的 mm_hash → num_embeds
    freed: list[str]                         # 已释放的 mm_hash
 
    def check_and_update_cache(self, request, input_id) -> bool:
        """检查本地缓存是否命中"""
 
    def can_allocate(self, request, input_id, budget, num_scheduled) -> bool:
        """检查是否有足够空间分配"""
 
    def allocate(self, request, input_id) -> None:
        """分配缓存位置"""
 
    def free(self, request) -> None:
        """释放请求关联的缓存"""

class EncoderCacheManager:
    cache_size: int                          # 总容量
    num_free_slots: int                      # 可用容量
    cached: dict[str, set[str]]              # mm_hash → 引用该 hash 的 request_id 集合
    freeable: OrderedDict[str, int]          # 可回收的 mm_hash → num_embeds
    freed: list[str]                         # 已释放的 mm_hash
 
    def check_and_update_cache(self, request, input_id) -> bool:
        """检查本地缓存是否命中"""
 
    def can_allocate(self, request, input_id, budget, num_scheduled) -> bool:
        """检查是否有足够空间分配"""
 
    def allocate(self, request, input_id) -> None:
        """分配缓存位置"""
 
    def free(self, request) -> None:
        """释放请求关联的缓存"""

class ECConnectorModelRunnerMixin:
 
    @staticmethod
    def maybe_save_ec_to_connector(
        encoder_cache: dict[str, torch.Tensor],
        mm_hash: str,
    ):
        """如果是 Producer，保存 Encoder Cache 到连接器"""
        if not has_ec_transfer():
            return
        connector = get_ec_transfer()
        connector.save_caches(encoder_cache=encoder_cache, mm_hash=mm_hash)
 
    @staticmethod
    def maybe_get_ec_connector_output(
        scheduler_output: "SchedulerOutput",
        encoder_cache: dict[str, torch.Tensor],
        **kwargs,
    ) -> AbstractContextManager[ECConnectorOutput | None]:
        """返回 EC 操作的 Context Manager（如果未配置 EC 则返回 nullcontext）"""
        return (
            ECConnectorModelRunnerMixin._get_ec_connector_output(
                scheduler_output, encoder_cache, **kwargs
            )
            if has_ec_transfer()
            else nullcontext()
        )
 
    @staticmethod
    @contextmanager
    def _get_ec_connector_output(
        scheduler_output, encoder_cache, **kwargs,
    ) -> Generator[ECConnectorOutput, None, None]:
        """
        EC 连接器完整生命周期的 Context Manager：
 
        进入时:
          1. bind_connector_metadata()   — 绑定调度器元数据
          2. start_load_caches()         — 如果是 Consumer，加载外部缓存
 
        [yield → 执行 Encoder + 嵌入收集]
 
        退出时:
          3. get_finished()              — 获取异步传输完成状态
          4. clear_connector_metadata()  — 清理元数据
        """
        output = ECConnectorOutput()
        ec_connector = get_ec_transfer()
 
        assert isinstance(ec_connector, ECConnectorBase)
        assert scheduler_output.ec_connector_metadata is not None
 
        ec_connector.bind_connector_metadata(
            scheduler_output.ec_connector_metadata
        )
 
        # Consumer 角色：在 yield 之前加载缓存
        if ec_connector.is_consumer:
            ec_connector.start_load_caches(encoder_cache, **kwargs)
 
        try:
            yield output
        finally:
            output.finished_sending, output.finished_recving = (
                ec_connector.get_finished(scheduler_output.finished_req_ids)
            )
            ec_connector.clear_connector_metadata()

class ECConnectorModelRunnerMixin:
 
    @staticmethod
    def maybe_save_ec_to_connector(
        encoder_cache: dict[str, torch.Tensor],
        mm_hash: str,
    ):
        """如果是 Producer，保存 Encoder Cache 到连接器"""
        if not has_ec_transfer():
            return
        connector = get_ec_transfer()
        connector.save_caches(encoder_cache=encoder_cache, mm_hash=mm_hash)
 
    @staticmethod
    def maybe_get_ec_connector_output(
        scheduler_output: "SchedulerOutput",
        encoder_cache: dict[str, torch.Tensor],
        **kwargs,
    ) -> AbstractContextManager[ECConnectorOutput | None]:
        """返回 EC 操作的 Context Manager（如果未配置 EC 则返回 nullcontext）"""
        return (
            ECConnectorModelRunnerMixin._get_ec_connector_output(
                scheduler_output, encoder_cache, **kwargs
            )
            if has_ec_transfer()
            else nullcontext()
        )
 
    @staticmethod
    @contextmanager
    def _get_ec_connector_output(
        scheduler_output, encoder_cache, **kwargs,
    ) -> Generator[ECConnectorOutput, None, None]:
        """
        EC 连接器完整生命周期的 Context Manager：
 
        进入时:
          1. bind_connector_metadata()   — 绑定调度器元数据
          2. start_load_caches()         — 如果是 Consumer，加载外部缓存
 
        [yield → 执行 Encoder + 嵌入收集]
 
        退出时:
          3. get_finished()              — 获取异步传输完成状态
          4. clear_connector_metadata()  — 清理元数据
        """
        output = ECConnectorOutput()
        ec_connector = get_ec_transfer()
 
        assert isinstance(ec_connector, ECConnectorBase)
        assert scheduler_output.ec_connector_metadata is not None
 
        ec_connector.bind_connector_metadata(
            scheduler_output.ec_connector_metadata
        )
 
        # Consumer 角色：在 yield 之前加载缓存
        if ec_connector.is_consumer:
            ec_connector.start_load_caches(encoder_cache, **kwargs)
 
        try:
            yield output
        finally:
            output.finished_sending, output.finished_recving = (
                ec_connector.get_finished(scheduler_output.finished_req_ids)
            )
            ec_connector.clear_connector_metadata()

class GPUModelRunner(
    LoRAModelRunnerMixin,
    KVConnectorModelRunnerMixin,
    ECConnectorModelRunnerMixin,    # ← EC 能力
):
    self.encoder_cache: dict[str, torch.Tensor] = {}

class GPUModelRunner(
    LoRAModelRunnerMixin,
    KVConnectorModelRunnerMixin,
    ECConnectorModelRunnerMixin,    # ← EC 能力
):
    self.encoder_cache: dict[str, torch.Tensor] = {}

# _execute_mm_encoder() 执行视觉编码
# 编码完成后，缓存结果并通知 EC Connector
for mm_hash, output in zip(mm_hashes, encoder_outputs):
    self.encoder_cache[mm_hash] = output
    logger.debug("Finish execute for mm hash %s", mm_hash)
    # ★ 调用 Mixin 方法保存到外部存储
    self.maybe_save_ec_to_connector(self.encoder_cache, mm_hash)

# _execute_mm_encoder() 执行视觉编码
# 编码完成后，缓存结果并通知 EC Connector
for mm_hash, output in zip(mm_hashes, encoder_outputs):
    self.encoder_cache[mm_hash] = output
    logger.debug("Finish execute for mm hash %s", mm_hash)
    # ★ 调用 Mixin 方法保存到外部存储
    self.maybe_save_ec_to_connector(self.encoder_cache, mm_hash)

# GPU Model Runner 的 execute_model 流程
ec_connector_output = None
 
if self.supports_mm_inputs and is_first_rank and not is_encoder_decoder:
    # ★ Context Manager 包装 EC 加载 + Encoder 执行 + 嵌入收集
    with self.maybe_get_ec_connector_output(
        scheduler_output,
        encoder_cache=self.encoder_cache,   # 传入内存缓存字典
    ) as ec_connector_output:
        # 1. EC Consumer: start_load_caches() 已在 Context Manager 入口执行
        #    → 外部 Encoder Cache 已加载到 self.encoder_cache
 
        # 2. 如果请求需要本地编码（非 EC 加载的项），执行 Encoder
        self._execute_mm_encoder(scheduler_output)
 
        # 3. 从 encoder_cache 收集嵌入向量
        #    此时 cache 中同时包含本地编码的和外部加载的
        mm_embeds, is_mm_embed = self._gather_mm_embeddings(scheduler_output)
 
    # 4. 将嵌入向量注入到模型输入
    inputs_embeds_scheduled = self.model.embed_input_ids(
        self.input_ids.gpu[:num_scheduled_tokens],
        multimodal_embeddings=mm_embeds,
        is_multimodal=is_mm_embed,
    )

# GPU Model Runner 的 execute_model 流程
ec_connector_output = None
 
if self.supports_mm_inputs and is_first_rank and not is_encoder_decoder:
    # ★ Context Manager 包装 EC 加载 + Encoder 执行 + 嵌入收集
    with self.maybe_get_ec_connector_output(
        scheduler_output,
        encoder_cache=self.encoder_cache,   # 传入内存缓存字典
    ) as ec_connector_output:
        # 1. EC Consumer: start_load_caches() 已在 Context Manager 入口执行
        #    → 外部 Encoder Cache 已加载到 self.encoder_cache
 
        # 2. 如果请求需要本地编码（非 EC 加载的项），执行 Encoder
        self._execute_mm_encoder(scheduler_output)
 
        # 3. 从 encoder_cache 收集嵌入向量
        #    此时 cache 中同时包含本地编码的和外部加载的
        mm_embeds, is_mm_embed = self._gather_mm_embeddings(scheduler_output)
 
    # 4. 将嵌入向量注入到模型输入
    inputs_embeds_scheduled = self.model.embed_input_ids(
        self.input_ids.gpu[:num_scheduled_tokens],
        multimodal_embeddings=mm_embeds,
        is_multimodal=is_mm_embed,
    )

mm_encoder_only: bool = False
"""
When enabled, skips the language component of the model.
This is usually only valid in disaggregated Encoder process.
"""

mm_encoder_only: bool = False
"""
When enabled, skips the language component of the model.
This is usually only valid in disaggregated Encoder process.
"""

if mm_config and mm_config.mm_encoder_only:
    # 跳过语言模型权重加载
    # 跳过 KV Cache 分配
    # 仅执行 Vision Encoder

if mm_config and mm_config.mm_encoder_only:
    # 跳过语言模型权重加载
    # 跳过 KV Cache 分配
    # 仅执行 Vision Encoder

# execute_model 入口
if has_ec_transfer() and not get_ec_transfer().is_consumer:
    # ★ Encoder-only 实例：只运行 Encoder，不运行 LM
    with self.maybe_get_ec_connector_output(
        scheduler_output, encoder_cache=self.encoder_cache
    ) as ec_connector_output:
        self._execute_mm_encoder(scheduler_output)
        return make_empty_encoder_model_runner_output(scheduler_output)
    # 直接返回空输出，不执行 LM forward / sampling

# execute_model 入口
if has_ec_transfer() and not get_ec_transfer().is_consumer:
    # ★ Encoder-only 实例：只运行 Encoder，不运行 LM
    with self.maybe_get_ec_connector_output(
        scheduler_output, encoder_cache=self.encoder_cache
    ) as ec_connector_output:
        self._execute_mm_encoder(scheduler_output)
        return make_empty_encoder_model_runner_output(scheduler_output)
    # 直接返回空输出，不执行 LM forward / sampling

def make_empty_encoder_model_runner_output(scheduler_output):
    """
    创建包含正确 per-request 索引但不含生成数据的 ModelRunnerOutput。
    每个请求返回 [0] 作为占位 token。
    """
    req_ids = list(scheduler_output.num_scheduled_tokens.keys())
    req_id_to_index = {rid: idx for idx, rid in enumerate(req_ids)}
    sampled_token_ids = [[0] for _ in req_ids]   # 占位
    pooler_output = [None for _ in req_ids]
    return ModelRunnerOutput(
        req_ids=req_ids,
        req_id_to_index=req_id_to_index,
        sampled_token_ids=sampled_token_ids,
        pooler_output=pooler_output,
    )

def make_empty_encoder_model_runner_output(scheduler_output):
    """
    创建包含正确 per-request 索引但不含生成数据的 ModelRunnerOutput。
    每个请求返回 [0] 作为占位 token。
    """
    req_ids = list(scheduler_output.num_scheduled_tokens.keys())
    req_id_to_index = {rid: idx for idx, rid in enumerate(req_ids)}
    sampled_token_ids = [[0] for _ in req_ids]   # 占位
    pooler_output = [None for _ in req_ids]
    return ModelRunnerOutput(
        req_ids=req_ids,
        req_id_to_index=req_id_to_index,
        sampled_token_ids=sampled_token_ids,
        pooler_output=pooler_output,
    )

@dataclass
class ECConnectorOutput:
    """EC Connector 的 Worker → Scheduler 输出"""
    finished_sending: set[str] | None = None    # 异步发送完成的请求 ID
    finished_recving: set[str] | None = None    # 异步接收完成的请求 ID
 
@dataclass
class ModelRunnerOutput:
    """模型运行器输出（包含 EC 结果）"""
    req_ids: list[str]
    sampled_token_ids: list[list[int]]
    ec_connector_output: ECConnectorOutput | None = None  # ← EC 输出
    kv_connector_output: KVConnectorOutput | None = None
    # ... 其他字段

@dataclass
class ECConnectorOutput:
    """EC Connector 的 Worker → Scheduler 输出"""
    finished_sending: set[str] | None = None    # 异步发送完成的请求 ID
    finished_recving: set[str] | None = None    # 异步接收完成的请求 ID
 
@dataclass
class ModelRunnerOutput:
    """模型运行器输出（包含 EC 结果）"""
    req_ids: list[str]
    sampled_token_ids: list[list[int]]
    ec_connector_output: ECConnectorOutput | None = None  # ← EC 输出
    kv_connector_output: KVConnectorOutput | None = None
    # ... 其他字段

@dataclass
class SchedulerOutput:
    scheduled_encoder_inputs: dict[str, list[int]]       # 需要编码的输入
    free_encoder_mm_hashes: list[str]                    # 需要释放的 EC
    ec_connector_metadata: ECConnectorMetadata | None = None  # EC 元数据
    kv_connector_metadata: KVConnectorMetadata | None = None  # KV 元数据

@dataclass
class SchedulerOutput:
    scheduled_encoder_inputs: dict[str, list[int]]       # 需要编码的输入
    free_encoder_mm_hashes: list[str]                    # 需要释放的 EC
    ec_connector_metadata: ECConnectorMetadata | None = None  # EC 元数据
    kv_connector_metadata: KVConnectorMetadata | None = None  # KV 元数据

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    req_data = await request.json()
 
    # 随机选择 Encoder 集群（扇出）/ Prefill / Decode 实例
    e_urls = app.state.e_urls    # Encoder 集群 URL 列表
    p_url = random.choice(app.state.p_urls) if app.state.p_urls else None
    d_url = random.choice(app.state.d_urls)
 
    is_streaming = req_data.get("stream", False)
 
    if is_streaming:
        return StreamingResponse(
            forward_stream(req_data, req_id, e_urls, p_url, d_url),
            media_type="text/event-stream",
        )
    result = await forward_non_stream(req_data, req_id, e_urls, p_url, d_url)
    return JSONResponse(content=result)

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    req_data = await request.json()
 
    # 随机选择 Encoder 集群（扇出）/ Prefill / Decode 实例
    e_urls = app.state.e_urls    # Encoder 集群 URL 列表
    p_url = random.choice(app.state.p_urls) if app.state.p_urls else None
    d_url = random.choice(app.state.d_urls)
 
    is_streaming = req_data.get("stream", False)
 
    if is_streaming:
        return StreamingResponse(
            forward_stream(req_data, req_id, e_urls, p_url, d_url),
            media_type="text/event-stream",
        )
    result = await forward_non_stream(req_data, req_id, e_urls, p_url, d_url)
    return JSONResponse(content=result)

async def forward_non_stream(req_data, req_id, e_urls, p_url, d_url):
    # Step 1: 将多模态项发送到 Encoder 集群
    await fanout_encoder_primer(req_data, e_urls, req_id)
 
    # Step 2: 通过 Prefill 实例（如果存在）
    req_data = await maybe_prefill(req_data, p_url, req_id)
 
    # Step 3: 转发到 Decode/PD 实例
    async with decode_session.post(
        f"{d_url}/v1/chat/completions", json=req_data
    ) as resp:
        return await resp.json()

async def forward_non_stream(req_data, req_id, e_urls, p_url, d_url):
    # Step 1: 将多模态项发送到 Encoder 集群
    await fanout_encoder_primer(req_data, e_urls, req_id)
 
    # Step 2: 通过 Prefill 实例（如果存在）
    req_data = await maybe_prefill(req_data, p_url, req_id)
 
    # Step 3: 转发到 Decode/PD 实例
    async with decode_session.post(
        f"{d_url}/v1/chat/completions", json=req_data
    ) as resp:
        return await resp.json()

MM_TYPES = {"image_url", "audio_url", "input_audio"}
 
def extract_mm_items(request_data: dict) -> list[dict]:
    """从消息中提取所有多模态项（图片/音频）"""
    items = []
    for msg in request_data.get("messages", []):
        content = msg.get("content")
        if not isinstance(content, list):
            continue
        for item in content:
            if item.get("type") in MM_TYPES:
                items.append(item)
    return items
 
async def fanout_encoder_primer(orig_request, e_urls, req_id):
    """
    为每个多模态项创建独立请求，并发发送到 Encoder 集群。
 
    关键设计：
    1. 每个 MM 项 → 1 个独立请求（仅含该 MM 项，无文本）
    2. max_tokens=1（只做编码，不生成文本）
    3. Round-robin 分发到 Encoder 集群
    4. asyncio.gather 并发执行所有请求
    """
    mm_items = extract_mm_items(orig_request)
    if not mm_items:
        return  # 无多模态项，跳过
 
    tasks = []
    url_cycle = (e_urls[i % len(e_urls)] for i in range(len(mm_items)))
 
    for idx, (item, target_url) in enumerate(zip(mm_items, url_cycle)):
        child_req_id = f"{req_id}:{idx}:{uuid.uuid4().hex[:6]}"
        encoder_req = {
            "model": orig_request.get("model"),
            "messages": [
                {"role": "user", "content": [item]},  # 仅含 MM 项
            ],
            "max_tokens": 1,      # 只编码，不生成
            "stream": False,
        }
        tasks.append(
            encode_session.post(
                f"{target_url}/v1/chat/completions",
                json=encoder_req,
                headers={"x-request-id": child_req_id},
            )
        )
 
    results = await asyncio.gather(*tasks, return_exceptions=True)
 
    # 检查所有子请求是否成功
    for idx, r in enumerate(results):
        if isinstance(r, Exception) or r.status != 200:
            raise HTTPException(status_code=502, detail="Encoder request failed")

MM_TYPES = {"image_url", "audio_url", "input_audio"}
 
def extract_mm_items(request_data: dict) -> list[dict]:
    """从消息中提取所有多模态项（图片/音频）"""
    items = []
    for msg in request_data.get("messages", []):
        content = msg.get("content")
        if not isinstance(content, list):
            continue
        for item in content:
            if item.get("type") in MM_TYPES:
                items.append(item)
    return items
 
async def fanout_encoder_primer(orig_request, e_urls, req_id):
    """
    为每个多模态项创建独立请求，并发发送到 Encoder 集群。
 
    关键设计：
    1. 每个 MM 项 → 1 个独立请求（仅含该 MM 项，无文本）
    2. max_tokens=1（只做编码，不生成文本）
    3. Round-robin 分发到 Encoder 集群
    4. asyncio.gather 并发执行所有请求
    """
    mm_items = extract_mm_items(orig_request)
    if not mm_items:
        return  # 无多模态项，跳过
 
    tasks = []
    url_cycle = (e_urls[i % len(e_urls)] for i in range(len(mm_items)))
 
    for idx, (item, target_url) in enumerate(zip(mm_items, url_cycle)):
        child_req_id = f"{req_id}:{idx}:{uuid.uuid4().hex[:6]}"
        encoder_req = {
            "model": orig_request.get("model"),
            "messages": [
                {"role": "user", "content": [item]},  # 仅含 MM 项
            ],
            "max_tokens": 1,      # 只编码，不生成
            "stream": False,
        }
        tasks.append(
            encode_session.post(
                f"{target_url}/v1/chat/completions",
                json=encoder_req,
                headers={"x-request-id": child_req_id},
            )
        )
 
    results = await asyncio.gather(*tasks, return_exceptions=True)
 
    # 检查所有子请求是否成功
    for idx, r in enumerate(results):
        if isinstance(r, Exception) or r.status != 200:
            raise HTTPException(status_code=502, detail="Encoder request failed")

async def maybe_prefill(req_data, p_url, req_id):
    if p_url:
        prefill_response = await process_prefill_stage(req_data, p_url, req_id)
        prefill_response_json = await prefill_response.json()
        # 获取 KV 传输参数（用于 Decode 端通过 NixlConnector 加载 KV）
        kv_transfer_params = prefill_response_json.get("kv_transfer_params", {})
        if kv_transfer_params:
            req_data["kv_transfer_params"] = kv_transfer_params
        return req_data
    else:
        return req_data  # E+PD 模式：直接跳过
 
async def process_prefill_stage(req_data, p_url, req_id):
    """Prefill 阶段：只做 1 token 的 prefill"""
    prefill_request = req_data.copy()
    prefill_request["kv_transfer_params"] = {
        "do_remote_decode": True,
        "do_remote_prefill": False,
        "remote_engine_id": None,
        "remote_block_ids": None,
        "remote_host": None,
        "remote_port": None,
    }
    prefill_request["stream"] = False
    prefill_request["max_tokens"] = 1         # 仅 prefill
    # 发送到 Prefill 实例
    return await prefill_session.post(
        f"{p_url}/v1/chat/completions", json=prefill_request
    )

async def maybe_prefill(req_data, p_url, req_id):
    if p_url:
        prefill_response = await process_prefill_stage(req_data, p_url, req_id)
        prefill_response_json = await prefill_response.json()
        # 获取 KV 传输参数（用于 Decode 端通过 NixlConnector 加载 KV）
        kv_transfer_params = prefill_response_json.get("kv_transfer_params", {})
        if kv_transfer_params:
            req_data["kv_transfer_params"] = kv_transfer_params
        return req_data
    else:
        return req_data  # E+PD 模式：直接跳过
 
async def process_prefill_stage(req_data, p_url, req_id):
    """Prefill 阶段：只做 1 token 的 prefill"""
    prefill_request = req_data.copy()
    prefill_request["kv_transfer_params"] = {
        "do_remote_decode": True,
        "do_remote_prefill": False,
        "remote_engine_id": None,
        "remote_block_ids": None,
        "remote_host": None,
        "remote_port": None,
    }
    prefill_request["stream"] = False
    prefill_request["max_tokens"] = 1         # 仅 prefill
    # 发送到 Prefill 实例
    return await prefill_session.post(
        f"{p_url}/v1/chat/completions", json=prefill_request
    )

# 非流式
async with decode_session.post(
    f"{d_url}/v1/chat/completions", json=req_data
) as resp:
    return await resp.json()
 
# 流式
async with decode_session.post(
    f"{d_url}/v1/chat/completions", json=req_data
) as resp:
    async for chunk in resp.content.iter_chunked(1024):
        yield chunk.decode("utf-8", errors="ignore")

# 非流式
async with decode_session.post(
    f"{d_url}/v1/chat/completions", json=req_data
) as resp:
    return await resp.json()
 
# 流式
async with decode_session.post(
    f"{d_url}/v1/chat/completions", json=req_data
) as resp:
    async for chunk in resp.content.iter_chunked(1024):
        yield chunk.decode("utf-8", errors="ignore")

python disagg_epd_proxy.py \
    --host 0.0.0.0 \
    --port 8000 \
    --encode-servers-urls "http://e1:8001,http://e2:8001"  # Encoder 集群
    --prefill-servers-urls "disable"                        # E+PD 模式
    --decode-servers-urls "http://pd1:8002"                 # PD 实例

python disagg_epd_proxy.py \
    --host 0.0.0.0 \
    --port 8000 \
    --encode-servers-urls "http://e1:8001,http://e2:8001"  # Encoder 集群
    --prefill-servers-urls "disable"                        # E+PD 模式
    --decode-servers-urls "http://pd1:8002"                 # PD 实例

# 共享存储路径
EC_SHARED_STORAGE_PATH="/tmp/ec_cache"
mkdir -p "$EC_SHARED_STORAGE_PATH"
 
# ==================== Encoder 实例 (GPU 0) ====================
CUDA_VISIBLE_DEVICES=0 vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19534 \
    --gpu-memory-utilization 0.01 \
    --enforce-eager \
    --no-enable-prefix-caching \
    --max-num-batched-tokens 114688 \
    --max-num-seqs 128 \
    --enable-request-id-headers \
    --ec-transfer-config '{
        "ec_connector": "ECExampleConnector",
        "ec_role": "ec_producer",
        "ec_connector_extra_config": {
            "shared_storage_path": "/tmp/ec_cache"
        }
    }'
 
# ==================== PD 实例 (GPU 1) ====================
CUDA_VISIBLE_DEVICES=1 vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19535 \
    --gpu-memory-utilization 0.7 \
    --enforce-eager \
    --max-num-seqs 128 \
    --max-model-len 32768 \
    --enable-request-id-headers \
    --ec-transfer-config '{
        "ec_connector": "ECExampleConnector",
        "ec_role": "ec_consumer",
        "ec_connector_extra_config": {
            "shared_storage_path": "/tmp/ec_cache"
        }
    }'
 
# ==================== Proxy ====================
python disagg_epd_proxy.py \
    --port 10001 \
    --encode-servers-urls "http://localhost:19534" \
    --prefill-servers-urls "disable" \
    --decode-servers-urls "http://localhost:19535"

# 共享存储路径
EC_SHARED_STORAGE_PATH="/tmp/ec_cache"
mkdir -p "$EC_SHARED_STORAGE_PATH"
 
# ==================== Encoder 实例 (GPU 0) ====================
CUDA_VISIBLE_DEVICES=0 vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19534 \
    --gpu-memory-utilization 0.01 \
    --enforce-eager \
    --no-enable-prefix-caching \
    --max-num-batched-tokens 114688 \
    --max-num-seqs 128 \
    --enable-request-id-headers \
    --ec-transfer-config '{
        "ec_connector": "ECExampleConnector",
        "ec_role": "ec_producer",
        "ec_connector_extra_config": {
            "shared_storage_path": "/tmp/ec_cache"
        }
    }'
 
# ==================== PD 实例 (GPU 1) ====================
CUDA_VISIBLE_DEVICES=1 vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19535 \
    --gpu-memory-utilization 0.7 \
    --enforce-eager \
    --max-num-seqs 128 \
    --max-model-len 32768 \
    --enable-request-id-headers \
    --ec-transfer-config '{
        "ec_connector": "ECExampleConnector",
        "ec_role": "ec_consumer",
        "ec_connector_extra_config": {
            "shared_storage_path": "/tmp/ec_cache"
        }
    }'
 
# ==================== Proxy ====================
python disagg_epd_proxy.py \
    --port 10001 \
    --encode-servers-urls "http://localhost:19534" \
    --prefill-servers-urls "disable" \
    --decode-servers-urls "http://localhost:19535"

EC_SHARED_STORAGE_PATH="/tmp/ec_cache"
export UCX_TLS=all
export UCX_NET_DEVICES=all
 
# ==================== Encoder 实例 (GPU 2) ====================
CUDA_VISIBLE_DEVICES=2 vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19534 \
    --gpu-memory-utilization 0.01 \
    --enforce-eager \
    --no-enable-prefix-caching \
    --max-num-batched-tokens 114688 \
    --ec-transfer-config '{
        "ec_connector": "ECExampleConnector",
        "ec_role": "ec_producer",
        "ec_connector_extra_config": {
            "shared_storage_path": "/tmp/ec_cache"
        }
    }'
 
# ==================== Prefill 实例 (GPU 2) ====================
# 同时是 EC Consumer + KV Producer
CUDA_VISIBLE_DEVICES=2 \
VLLM_NIXL_SIDE_CHANNEL_PORT=5559 \
vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19535 \
    --gpu-memory-utilization 0.7 \
    --enforce-eager \
    --ec-transfer-config '{
        "ec_connector": "ECExampleConnector",
        "ec_role": "ec_consumer",
        "ec_connector_extra_config": {
            "shared_storage_path": "/tmp/ec_cache"
        }
    }' \
    --kv-transfer-config '{
        "kv_connector": "NixlConnector",
        "kv_role": "kv_producer"
    }'
 
# ==================== Decode 实例 (GPU 3) ====================
# 仅 KV Consumer，无需 EC 配置
CUDA_VISIBLE_DEVICES=3 \
VLLM_NIXL_SIDE_CHANNEL_PORT=6000 \
vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19536 \
    --gpu-memory-utilization 0.7 \
    --enforce-eager \
    --kv-transfer-config '{
        "kv_connector": "NixlConnector",
        "kv_role": "kv_consumer"
    }'
 
# ==================== Proxy ====================
python disagg_epd_proxy.py \
    --port 10001 \
    --encode-servers-urls "http://localhost:19534" \
    --prefill-servers-urls "http://localhost:19535" \
    --decode-servers-urls "http://localhost:19536"

EC_SHARED_STORAGE_PATH="/tmp/ec_cache"
export UCX_TLS=all
export UCX_NET_DEVICES=all
 
# ==================== Encoder 实例 (GPU 2) ====================
CUDA_VISIBLE_DEVICES=2 vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19534 \
    --gpu-memory-utilization 0.01 \
    --enforce-eager \
    --no-enable-prefix-caching \
    --max-num-batched-tokens 114688 \
    --ec-transfer-config '{
        "ec_connector": "ECExampleConnector",
        "ec_role": "ec_producer",
        "ec_connector_extra_config": {
            "shared_storage_path": "/tmp/ec_cache"
        }
    }'
 
# ==================== Prefill 实例 (GPU 2) ====================
# 同时是 EC Consumer + KV Producer
CUDA_VISIBLE_DEVICES=2 \
VLLM_NIXL_SIDE_CHANNEL_PORT=5559 \
vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19535 \
    --gpu-memory-utilization 0.7 \
    --enforce-eager \
    --ec-transfer-config '{
        "ec_connector": "ECExampleConnector",
        "ec_role": "ec_consumer",
        "ec_connector_extra_config": {
            "shared_storage_path": "/tmp/ec_cache"
        }
    }' \
    --kv-transfer-config '{
        "kv_connector": "NixlConnector",
        "kv_role": "kv_producer"
    }'
 
# ==================== Decode 实例 (GPU 3) ====================
# 仅 KV Consumer，无需 EC 配置
CUDA_VISIBLE_DEVICES=3 \
VLLM_NIXL_SIDE_CHANNEL_PORT=6000 \
vllm serve Qwen/Qwen2.5-VL-3B-Instruct \
    --port 19536 \
    --gpu-memory-utilization 0.7 \
    --enforce-eager \
    --kv-transfer-config '{
        "kv_connector": "NixlConnector",
        "kv_role": "kv_consumer"
    }'
 
# ==================== Proxy ====================
python disagg_epd_proxy.py \
    --port 10001 \
    --encode-servers-urls "http://localhost:19534" \
    --prefill-servers-urls "http://localhost:19535" \
    --decode-servers-urls "http://localhost:19536"

tests/v1/ec_connector/unit/test_ec_example_connector.py

tests/v1/ec_connector/unit/test_ec_example_connector.py

tests/v1/ec_connector/integration/
├── test_epd_correctness.py          # 正确性验证脚本
├── run_epd_correctness_test.sh      # 集成测试入口
├── hato.jpg                         # 测试图片
└── README.md                        # 测试文档

tests/v1/ec_connector/integration/
├── test_epd_correctness.py          # 正确性验证脚本
├── run_epd_correctness_test.sh      # 集成测试入口
├── hato.jpg                         # 测试图片
└── README.md                        # 测试文档

python test_epd_correctness.py \
    --service_url "http://localhost:8000" \
    --model_name "Qwen/Qwen2.5-VL-3B-Instruct" \
    --mode baseline|disagg \
    --baseline_file ".vllm_epd_baseline.txt" \
    --use_mm_prompts

python test_epd_correctness.py \
    --service_url "http://localhost:8000" \
    --model_name "Qwen/Qwen2.5-VL-3B-Instruct" \
    --mode baseline|disagg \
    --baseline_file ".vllm_epd_baseline.txt" \
    --use_mm_prompts

# Step 1: 单实例基线，保存参考输出到 .vllm_baseline.txt
# Step 2: 1E + 1PD → 与基线对比，验证 EC Transfer 正确性
# Step 3: 1P + 1D 基线（NixlConnector 场景，P/D 分离本身有微小差异）
# Step 4: 1E + 1P + 1D → 与 PD 基线对比，验证三级分离正确性

# Step 1: 单实例基线，保存参考输出到 .vllm_baseline.txt
# Step 2: 1E + 1PD → 与基线对比，验证 EC Transfer 正确性
# Step 3: 1P + 1D 基线（NixlConnector 场景，P/D 分离本身有微小差异）
# Step 4: 1E + 1P + 1D → 与 PD 基线对比，验证三级分离正确性