源码精读：TokenizerManager —— 请求入口进程

谁该读这一篇？ 想搞清"客户端 HTTP 请求是怎么被 tokenize 并发送给 Scheduler 的"、"为什么 TM 是单点风险"的工程师。 前置阅读： 01-server-entry.md、01-overview/02-architecture.md。 耗时： 35 分钟 学完能： 1. 画出 generate_request 完整调用链； 2. 说出 ReqState 这个数据结构在 in-flight 请求里承担什么； 3. 解释 TM 为何用 asyncio 而 Scheduler 用 while True； 4. 在源码里定位"tokenize"、"send to scheduler"、"receive from detokenizer"三个关键点； 5. 调试 TM 卡死时的步骤。

1. TM 的职责

HTTP request ─▶ TokenizerManager ─▶ ZMQ ─▶ Scheduler
                       ▲                       │
                       │                       ▼
                       └── ZMQ ◀── Detokenizer ◀── (token id stream)

具体做：

Tokenize：HF tokenizer 把 string prompt 编码成 token id 列表。
构建请求：组装 TokenizedGenerateReqInput，附带采样参数、stop、约束等。
维护 ReqState：为每个 in-flight 请求建一个状态对象，挂 asyncio.Queue。
发送给 Scheduler：ZMQ PUSH。
接收 Detokenizer 输出：ZMQ PULL，把 chunk 塞回对应 ReqState.text_queue。
流式 yield 给 HTTP：HTTP handler await 队列拿 chunk，组装 SSE。

整个进程是 asyncio event loop，所有 I/O 是异步的。

2. `TokenizerManager.init`（236+）

源码：tokenizer_manager.py:236。

def __init__(self, server_args, port_args):
    self.server_args = server_args
    self.enable_metrics = server_args.enable_metrics
    ...
    self.init_model_config()                # ModelConfig
    self.init_tokenizer_and_processor()     # HF tokenizer / multimodal processor
    self.init_ipc_channels(port_args)       # ZMQ socket
    self.init_running_status()              # in-flight 字典
    self.init_request_logging_and_dumping() # 日志 / dump 控制
    self.init_weight_update()               # 权重热更新（RL 训练用）
    self.init_lora()                        # LoRA 状态
    self.init_disaggregation()              # PD 分离 / encoder 分离
    self.init_metric_collector_watchdog()   # metrics + 看门狗
    self.init_request_dispatcher()          # 消息分派器

10 个 init_* 调用，每个职责清晰。第一次读时只关心 model_config、tokenizer、ipc_channels、running_status、request_dispatcher 这五个。

3. `init_ipc_channels`：ZMQ socket 绑定

def init_ipc_channels(self, port_args):
    context = zmq.Context(2)
    # 收 Detokenizer 输出
    self.recv_from_detokenizer = get_zmq_socket(
        context, zmq.PULL, port_args.tokenizer_ipc_name, True
    )
    # 发 Scheduler
    self.send_to_scheduler = get_zmq_socket(
        context, zmq.PUSH, port_args.scheduler_input_ipc_name, False
    )

PUSH/PULL：单向、无应答。 PUSH 端发不出去也不阻塞（高水位时丢弃，可配置）；PULL 端拉不到就 wait。这是为了避免任何一边的延迟传染另一边。

4. `ReqState`：in-flight 状态（142+）

源码：tokenizer_manager.py:142。

@dataclass
class ReqState:
    out_list: List[Any]                # 累计输出（用于非流式）
    finished: bool
    event: asyncio.Event               # finish 信号
    text_queue: Optional[asyncio.Queue] = None  # 流式 chunk 队列
    # ... 其它字段：stream / output_strs / metric 计时器

关键不变量：

每个 in-flight 请求一份 ReqState，存在 self.rid_to_state[rid]。
HTTP handler 通过 await reqstate.text_queue.get() 拿增量。
Detokenizer 输出回来后 TM 把 chunk put() 到对应队列。
请求 finish 后 event.set()，HTTP handler 退出 SSE 循环。

5. `generate_request`：主路径（539+）

源码：tokenizer_manager.py:539。

简化版伪代码：

async def generate_request(self, obj: GenerateReqInput, request=None):
    # 1. 校验 + 注册请求
    rid = obj.rid or str(uuid.uuid4())
    state = ReqState(...)
    self.rid_to_state[rid] = state

    # 2. tokenize prompt
    token_ids = self.tokenizer.encode(obj.text) if obj.text else obj.input_ids
    # 多模态：还要算 image / video embedding placeholder

    # 3. 构建 TokenizedGenerateReqInput
    tokenized = TokenizedGenerateReqInput(
        rid=rid,
        input_ids=token_ids,
        sampling_params=obj.sampling_params,
        ...
    )

    # 4. 发给 Scheduler
    await self._send_one_request_async(tokenized)

    # 5. 流式回写
    if obj.stream:
        async for chunk in self._handle_streaming_request(rid, state):
            yield chunk
    else:
        await state.event.wait()
        return state.out_list[-1]      # 完整结果

async def _handle_streaming_request(self, rid, state):
    while not state.finished:
        chunk = await state.text_queue.get()
        yield chunk

关键细节：

第 4 步的"发"是同步 ZMQ send（实际是 send_pyobj，pickle 序列化）；不需要异步。
第 5 步循环 await text_queue.get()，没数据时让出 event loop，让别的协程跑。

6. `init_request_dispatcher`：分派 Detokenizer 回来的消息

def init_request_dispatcher(self):
    self._request_dispatcher = TypeBasedDispatcher([
        (BatchStrOutput, self.handle_batch_output),
        (BatchEmbeddingOutput, self.handle_batch_embedding_output),
        (BatchTokenIDOutput, self.handle_batch_token_id_output),
        ...
    ])

async def event_loop(self):
    while True:
        recv_obj = await self.recv_from_detokenizer.recv_pyobj()
        self._request_dispatcher(recv_obj)

收到 BatchStrOutput（一批请求的 detokenize 后字符串）后：

def handle_batch_output(self, batch):
    for rid, text, finished in zip(batch.rids, batch.output_strs, batch.finished):
        state = self.rid_to_state.get(rid)
        if state is None:
            continue                              # 请求已被 abort
        state.text_queue.put_nowait(text)         # 塞队列
        if finished:
            state.finished = True
            state.event.set()
            del self.rid_to_state[rid]            # 清理

7. asyncio 和 multiprocessing 怎么配合

TM 在主进程里跑，FastAPI 路由 handler 是 async def，TM 的 generate_request 也是 async def。所有这些跑在同一个 event loop 里。

ZMQ 用 zmq.asyncio.Context，所以 recv_pyobj() / send_pyobj() 都可以 await。 这一点很关键：阻塞的 ZMQ 会卡死 event loop，整个 TM 没响应。

子进程（Scheduler、Detokenizer）是普通 multiprocessing.Process，不属于 asyncio 范围；它们自己跑同步 while True。跨进程通信靠 ZMQ socket，刚好把"主进程 asyncio + 子进程同步"两边粘起来。

8. 错误处理与超时

如果 Scheduler 子进程崩了：

TM 的 recv_from_detokenizer 收不到回包。
所有 in-flight 请求的 text_queue 一直空。
客户端 HTTP 超时。

兜底：

TM 有 Watchdog 检测 Scheduler 子进程存活；不存活就 fail 当前所有请求。
每个请求有 soft timeout（server_args.req_max_wait_time），超时 abort 并发回错误。
HTTP server 自带超时（uvicorn / granian 配置）。

9. 多 worker 模式

--tokenizer-worker-num N > 1 时启用 MultiTokenizerWorker：

每个 HTTP worker 进程都有自己一份 TM。
Detokenizer 输出需要发回正确的 worker，用 SocketMapping 路由。
高 QPS 场景能解 GIL 问题。

复杂度上去了，新手先用单 worker，瓶颈出现再开。

10. 性能要点

操作	单次耗时（粗略）
tokenize（HF tokenizer，short prompt）	0.5 - 2 ms
pickle 序列化 + ZMQ send	0.1 - 0.5 ms
ZMQ recv + pickle 反序列化	0.1 - 0.5 ms
asyncio yield + reschedule	~10 μs

在 1000+ QPS 时，tokenize 会成为单 TM 进程的 CPU 瓶颈——开 multi worker。

11. 排障 cheatsheet

现象	排查
客户端 timeout、TM 日志无新请求	HTTP 没传到 TM；查 nginx / 网络
TM 有日志收请求但 Scheduler 没收到	ZMQ socket 名字不对 / Scheduler crash
Detokenizer 输出回来但客户端没看到	`rid_to_state` 里没找到 rid；可能 abort 了
TM 进程 CPU 100%	tokenize 瓶颈 → 开 multi worker
TM 进程 RSS 飙升	in-flight 请求堆积 → 检查 max_running_requests
HTTP 健康检查偶现 503	TM 卡死或排队过长

12. 关键源码索引

内容	文件:行
类定义	`tokenizer_manager.py:223`
`__init__`	`tokenizer_manager.py:236`
`ReqState`	`tokenizer_manager.py:142`
`generate_request`	`tokenizer_manager.py:539`
`init_request_dispatcher`	`tokenizer_manager.py`（找 `init_request_dispatcher`）
Score / 其它任务	`tokenizer_manager_score_mixin.py`
控制 API（pause/resume）	`tokenizer_control_mixin.py`
多 worker	`multi_tokenizer_mixin.py`

13. 小结

TM 是单点（除非开 multi worker），生命周期管 in-flight 请求的 ReqState。
asyncio 主循环 + ZMQ 异步 socket 是 TM 的灵魂。
主路径：tokenize → build TokenizedGenerateReqInput → send → recv chunks → yield SSE。
Watchdog + soft timeout 兜底子进程异常。
高 QPS 用 multi worker；Tokenize 是单线程瓶颈。

14. 自检

TM 用 asyncio，Scheduler 用 while True，为什么不统一？

答案

两边的瓶颈和并发模型不同： TM 是 I/O bound（HTTP 收发、ZMQ recv、SSE 流式回写）—— asyncio 协程让一个进程同时维护几百路 in-flight 请求。 Scheduler 是 compute bound（每步都阻塞等 GPU forward 返回）—— asyncio 在这里没用，反而 async/await 上下文切换开销纯亏。同步 while True 最贴合 GPU 调度节奏。统一就要在两个范式之间做妥协，性能 + 代码复杂度都受损。

ZMQ 选 PUSH/PULL 而非 REQ/REP 的理由？

答案

PUSH/PULL 是单向的，任一端慢不会传染另一端：Detokenizer 慢一拍，Scheduler 仍能继续生产 token id 推下游，背压靠 ZMQ 高水位 + 上层 abort 兜底。 REQ/REP 严格请求-应答配对：Scheduler 每次发完必须等 Detokenizer ACK 才能继续，下游一卡上游就停摆。代价：PUSH/PULL 没应答，错误检测靠 watchdog + 心跳，不能像 REQ/REP 那样立刻知道对端处理状态。在 SGLang 这种"快速失败 + 重启"语义下，这个代价可接受。

rid_to_state 里没找到 rid 时为什么不抛错？

答案

常见情况是 **请求已 abort**（客户端断开 / 超时）：TM 收到 abort 后立刻从字典里 del 掉 ReqState，但此时 Scheduler / Detokenizer 已经在 in-flight 处理这个 rid，结果会回流到 TM。抛错的话每个 abort 都会污染日志、可能让上游崩。SGLang 选择**静默忽略 stale 消息**（continue 即可），既不浪费资源也不刷错误。这是分布式系统里"消息可能 late"的标准处理。

multi worker 模式下 Detokenizer 怎么知道给哪个 TM 发？

答案

`MultiHttpWorkerDetokenizerMixin` 维护一个 `SocketMapping`：每个 TM worker 启动时把自己的 ZMQ pull socket 名注册到 Detokenizer；TM 发请求给 Scheduler 时在消息里带上 `worker_id`，Scheduler 转发给 Detokenizer 时 worker_id 一起带过去，Detokenizer 根据 worker_id 查 SocketMapping 选对应 socket 发回。关键不变量：**同 rid 的所有 token 一定回同一个 TM worker**——否则 ReqState 不在那个 worker，没人能 yield SSE。

tokenize 耗时长（多模态 / 长 prompt）会怎么影响 TPOT？

答案

严格说 tokenize 影响 **TTFT** 而不是 TPOT —— tokenize 只在请求进入时跑一次，TPOT 是后续 token 间隔。但**间接影响**：TM 是单线程（除非 multi worker），tokenize 长 prompt 时 asyncio event loop 被卡住，其它请求的 SSE chunk 也回不出去，表现为"全员 TPOT 抖动"。解决：multi worker 把 tokenize 分到多进程；多模态用 `async_dynamic_batch_tokenizer.py` 异步化；或者前置一个专门的 tokenize 服务。

15. 下一步

03-scheduler.md — Scheduler 主循环。
07-detokenizer.md — 出口进程。
08-io-struct.md — 这些消息体的字段。

上游源码：sglang/python/sglang/srt/managers/tokenizer_manager.py。