执行摘要

• 一句话：将 Scheduler IPC 通道封装为专用容器
• 推荐动作：值得精读，尤其学习如何使用 frozen dataclass + classmethod 工厂来封装资源生命周期的模式。该 PR 是调度器重构系列的一部分，体现了逐步提升代码内聚性的思路。

功能与动机

Scheduler 类直接持有多个 IPC 通道属性，且初始化代码散落在 init_ipc_channels 方法中，导致职责不清晰、难以维护。通过将它们集中到 SchedulerIpcChannels 数据类，使得 IPC 通信作为一个独立关注点得到明确类型，同时减少 Scheduler 的字段数量，提高内聚性。

实现拆解

1. 创建 IPC 通道容器：在 scheduler_components/ipc_channels.py 中定义 SchedulerIpcChannels 数据类，使用 @dataclass(frozen=True, slots=True, kw_only=True) 确保不可变性和内存效率。属性包括所有套接字和 SenderWrapper。
2. 提取工厂方法：实现 create 类方法，接收端口参数、rank 零标志、skip_tokenizer_init 和 metrics_enabled 参数，内部完成 zmq.Context 创建和套接字初始化逻辑。
3. 更新 Scheduler 初始化：修改 Scheduler.init_ipc_channels，将分散的 zmq 代码替换为单一的 SchedulerIpcChannels.create(...) 调用，并赋值给 self.ipc_channels。同时移除对 zmq、get_zmq_socket、SenderWrapper 的直接导入。
4. 调整内部引用：Scheduler 中所有对 self.send_to_tokenizer、self.recv_from_tokenizer 等属性的访问改为 self.ipc_channels.send_to_tokenizer 等格式，确保协作类构造参数也相应更新。
5. 适配测试：两个测试文件中的 mock 设置从 scheduler.send_to_tokenizer = MagicMock() 改为 scheduler.ipc_channels = MagicMock()，验证点同步更新。

关键文件：

• python/sglang/srt/managers/scheduler_components/ipc_channels.py（模块 IPC通道；类别 source；类型 core-logic；符号 SchedulerIpcChannels, create）: 新增文件，定义了核心数据类 SchedulerIpcChannels 及其工厂方法 create，是本次重构的核心。
• python/sglang/srt/managers/scheduler.py（模块 调度器核心；类别 source；类型 dependency-wiring）: 主要修改文件，移除了分散的 IPC 通道属性，改为使用 ipc_channels 容器，调整了导入和构造逻辑。
• test/registered/unit/managers/test_scheduler_flush_cache.py（模块 调度器测试；类别 test；类型 test-coverage）: 更新 mock 设置和断言以匹配新结构，确保 flush cache 逻辑正确性。
• test/registered/unit/managers/test_priority_scheduling_disaggregation.py（模块 优先级调度；类别 test；类型 test-coverage）: 更新 mock 设置和断言以匹配新结构，确保优先级调度逻辑正确性。

关键符号：SchedulerIpcChannels, SchedulerIpcChannels.create, Scheduler.init_ipc_channels

关键源码片段

python/sglang/srt/managers/scheduler_components/ipc_channels.py

新增文件，定义了核心数据类 SchedulerIpcChannels 及其工厂方法 create，是本次重构的核心。

from dataclasses import dataclass
from typing import Optional
​
import zmq
​
from sglang.srt.managers.scheduler_components.output_sender import SenderWrapper
from sglang.srt.server_args import PortArgs
from sglang.srt.utils.network import get_zmq_socket
​
​
@dataclass(frozen=True, slots=True, kw_only=True)
class SchedulerIpcChannels:
    """封装 Scheduler 所有 ZeroMQ IPC 通道的不可变数据类。
​
    使用 frozen=True 确保实例创建后不可修改，slots=True 节省内存。
    """
    # 接收通道：rank 0 才有的 PULL 和 DEALER 套接字
    recv_from_tokenizer: Optional[zmq.Socket]
    recv_from_rpc: Optional[zmq.Socket]
    # 发送通道：包装为 SenderWrapper 的 PUSH 套接字
    send_to_tokenizer: SenderWrapper
    send_to_detokenizer: SenderWrapper
    # 可选指标通道
    send_metrics_from_scheduler: Optional[zmq.Socket]
​
    @classmethod
    def create(
        cls,
        *,
        port_args: PortArgs,
        is_rank_zero: bool,
        skip_tokenizer_init: bool,
        metrics_enabled: bool,
    ) -> "SchedulerIpcChannels":
        """工厂方法，集中创建所有 IPC 连接，执行所有与 zmq 相关的样板代码。"""
        context = zmq.Context(2)
​
        if is_rank_zero:
            # rank 0 才创建接收和发送套接字
            recv_from_tokenizer = get_zmq_socket(
                context, zmq.PULL, port_args.scheduler_input_ipc_name, False
            )
            recv_from_rpc = get_zmq_socket(
                context, zmq.DEALER, port_args.rpc_ipc_name, False
            )
​
            send_to_tokenizer_raw = get_zmq_socket(
                context, zmq.PUSH, port_args.tokenizer_ipc_name, False
            )
            if skip_tokenizer_init:
                # 直接发送到 TokenizerManager（无需 detokenizer）
                send_to_detokenizer_raw = get_zmq_socket(
                    context, zmq.PUSH, port_args.tokenizer_ipc_name, False
                )
            else:
                # 发送到 DetokenizerManager
                send_to_detokenizer_raw = get_zmq_socket(
                    context, zmq.PUSH, port_args.detokenizer_ipc_name, False
                )
​
            send_to_tokenizer = SenderWrapper(send_to_tokenizer_raw)
            send_to_detokenizer = SenderWrapper(send_to_detokenizer_raw)
        else:
            recv_from_tokenizer = None
            recv_from_rpc = None
            send_to_tokenizer = SenderWrapper(None)
            send_to_detokenizer = SenderWrapper(None)
​
        if metrics_enabled:
            send_metrics_from_scheduler = get_zmq_socket(
                context, zmq.PUSH, port_args.metrics_ipc_name, False
            )
        else:
            send_metrics_from_scheduler = None
​
        return cls(
            recv_from_tokenizer=recv_from_tokenizer,
            recv_from_rpc=recv_from_rpc,
            send_to_tokenizer=send_to_tokenizer,
            send_to_detokenizer=send_to_detokenizer,
            send_metrics_from_scheduler=send_metrics_from_scheduler,
        )

评论区精华

无实质性 review 讨论。PR 只有一个来自 gemini-code-assist 的配额警告评论，与代码变更无关。

• 暂无高价值评论线程

风险与影响

• 风险：本次重构为机械性重命名和移动，逻辑未变，风险较低。但需注意：如果 Scheduler 子类或外部模块通过属性直接访问 send_to_tokenizer 等，会因属性消失而报错。代码库中除测试外没有发现此类用例。测试覆盖了 flush cache 和优先级调度路径，回归风险较小。潜在风险：ipc_channels 字段在 Scheduler 构造过程中可能被其他方法提前引用，但通过检查 init 顺序可避免。
• 影响：对用户无影响：接口不变，功能不变。对系统：少量性能提升（slots 数据类降低内存占用）。对团队：提高了代码可读性和可维护性，为后续 IPC 通道相关功能扩展提供了更清晰的出发点。
• 风险标记：核心路径变更（Scheduler 初始化）, 缺少回归测试覆盖部分路径（非 rank 0 分支）

关联脉络

• PR #25727 Encapsulate the pending-flush bookkeeping in a small wrapper: 同属调度器重构系列，将分散的状态封装到专用包装器，提升内聚性。
• PR #25716 Pack scattered new-token-ratio state into a dedicated tracker: 类似模式，将 new-token-ratio 状态封装为专用跟踪器，属于同一重构方向。