TensorRT-LLM 源码完全拆解（二十五）：GenerationExecutor 的三根队列——请求如何跨进 worker 世界，结果又如何回到调用者

3.1 单进程 Worker：不是默认，而是两个特例

GenerationExecutorWorker 同时继承 RpcWorkerMixin 和 BaseWorker，而 BaseWorker 又继承 GenerationExecutor。也就是说，worker 自己也可以扮演 executor：请求不跨进程，submit() 直接把请求塞进本地 PyExecutor，GenerationResult 直接由 worker 的响应线程投递。

这条路径只在两种情况下默认启用：

• return_logits=True：需要把 context logits 整体搬回 Python 侧时，跨进程 IPC 的序列化成本过高；
• 环境变量 enable_worker_single_process_for_tp1() 为真：开发者调试 TP1 时希望所有东西在一个进程里，方便打断点。

源码对这条路径的注释写得很直白：TP1 的单进程 worker 会损害流式生成性能。原因并不玄妙：流式返回要求每个 step 的结果尽快离开 GPU 进程；单进程模式下，Python 主线程既要负责 HTTP/生成循环，又要等待引擎，CPU 上的调度与解码会和 GPU 前向争抢同一个 GIL。拆出 worker 进程后，rank0 的响应线程可以持续把结果推给 proxy，主进程只负责消费队列。

所以「单卡就应该单进程」是一个直觉陷阱。TensorRT-LLM 在 Linux 上即使只有一张 GPU，默认也会拆出 proxy 与 worker 两个进程，正是为了让流式输出不被主进程的 Python 工作拖慢。

3.2 IPC Proxy：请求队列 + 结果队列 + 启动状态队列

默认路径创建 GenerationExecutorProxy。它的构造顺序值得按源码走一遍，因为其中包含了几处必须保持顺序的初始化：

1
# 1. 先选 MPI 会话：自己拉起 worker，还是绑定外部已存在的 worker 世界
2
if mpi_session is None:
3
    if mpi_process_pre_spawned:
4
        self.mpi_session = create_mpi_comm_session(model_world_size)
5
    else:
6
        self.mpi_session = MpiPoolSession(n_workers=model_world_size)
7
else:
8
    validate_session_world_size(mpi_session, model_world_size)
9
    self.mpi_session = mpi_session
10

11
# 2. 建三根队列
12
self.request_queue = IpcQueue(is_server=True, name="proxy_request_queue")
13
self.result_queue = FusedIpcQueue(is_server=True, socket_type=zmq.PULL,
14
                                  name="proxy_result_queue")
15
self.worker_init_status_queue = IpcQueue(is_server=True,
16
                                         socket_type=zmq.ROUTER,
17
                                         name="worker_init_status_queue")
18

19
# 3. 通过 MpiPoolSession.submit(worker_main, ...) 拉起 worker 世界
20
# 4. 等 worker 的 READY，再建 RPC client 与错误监控线程
21
# 5. 启动结果分发线程

三根队列的分工必须分清：

result_queue 在普通模式下是 PAIR，因为只有 rank0 worker 一个生产者；一旦启用 num_postprocess_workers > 0，结果会由多个后处理进程分别写回，因此改成 PULL，让多个 PUSH 端连接。这个 socket 类型的变化不是风格问题：PAIR 只允许一对一，多个生产者同时写会直接破坏消息边界。

request_queue 在单前端模式下也是 PAIR，但在多前端模式下变成 PUSH/PULL：rank0 worker 绑定 PULL 端，所有前端进程都作为 PUSH 端连接。这样才能让多个 trtllm-serve 前端共享同一个引擎。

3.3 启动握手：为什么 READY 之前还要先发进程身份

_start_executor_workers() 通过 mpi_session.submit(worker_main, ...) 把 worker 世界拉起来，然后进入 _wait_for_executor_workers_ready()。这个等待循环同时做三件事：

1. 轮询 worker_init_status_queue，收到消息后立刻回 "ACK"；
2. 识别 WORKER_PROCESS_IDENTITIES_SIGNAL，把每个 rank 的 pid 登记到 WorkerProcessMonitor；
3. 在等待期间持续检测 worker 是否已经死亡。

worker 侧的顺序是刻意安排的：

1
mpi_comm().barrier()
2
worker_process_identities = mpi_comm().allgather(
3
    capture_worker_process_identity(mpi_rank()))
4

5
# 先发身份，再构造 backend。模型加载可能持续几分钟，
6
# 外部 kill 掉的 worker 未必能走完 MPI future。
7
if is_leader and worker_process_identities_signal is not None:
8
    identities_msg = (worker_process_identities_signal, None,
9
                      worker_process_identities)
10
    worker_init_status_queue.notify_with_retry(identities_msg)
11

12
worker = worker_cls(...)   # 这里才真正 create_py_executor、加载权重
13

14
ready_msg = (ready_signal, None, worker_process_identities)
15
worker_init_status_queue.notify_with_retry(ready_msg)

为什么身份要先于 READY 发送？因为模型构造可能非常慢，而 proxy 需要一个独立于 MPI future 的存活信号。如果 worker 在加载权重时被 OOM killer 杀掉，MPI future 可能永远不会完成；此时 proxy 只能靠 pid 级别的 WorkerProcessMonitor 发现「这个进程已经没了」，从而快速失败，而不是一直等到 READY 超时。这是「初始化阶段」与「运行阶段」两类故障检测必须分开的典型例子。

还有一个容易被忽略的细节：add_done_callback(mpi_done_callback) 必须在 READY 之后才注册。如果 future 已经完成，add_done_callback 会同步执行回调，把引擎标记为 dead；而 _fail_initialization() 需要先 shutdown_abort() 再标记 dead，顺序反了会让 MPI_Abort 的升级路径失效，导致持有权重的 rank 卡死。源码注释把这一点写得很清楚，说明这是踩过坑之后固定下来的顺序。

3.4 RPC Proxy、Ray 与多前端挂载

orchestrator_type="rpc" 时走 GenerationExecutorRpcProxy：进程模型与 IPC proxy 类似，但请求/结果不再走本地 ZMQ，而是通过 RPC 通道；适合 worker world 已经在别处启动、前端只想远程调用的场景。

orchestrator_type="ray" 时走 RayExecutor。Ray 自己负责 actor 的放置、重启与资源调度，TensorRT-LLM 只保留「请求如何送到 rank0、结果如何聚合」的逻辑。官方文档给出的 Ray 编排结构如下：

图中可以看到，Ray 路径下 driver、worker actor 与引擎进程是三层分离的；TensorRT-LLM 不再自己维护 MPI pool，而是把 worker 的生命周期交给 Ray。这也是为什么 collective_rpc 的多 rank 能力只在 RPC/Ray 路径完整支持，而 IPC proxy 对多 rank 只放行 sleep、wakeup 这类在 rank 内部自行协调的方法。

多前端挂载（num_serve_frontends > 1）是另一条独立分支。启动器 proxy 先创建一个共享 IPC 目录与 32 字节 HMAC key，rank0 worker 绑定唯一的请求入口，每个前端绑定自己的结果 lane；TLLM_EXECUTOR_ATTACH_INFO 通过环境变量把地址和 key 交给被 trtllm-serve 拉起的子进程。挂载进程走 GenerationExecutorFrontendProxy，它刻意跳过 GenerationExecutorProxy.__init__，因此不会创建 MPI session、不会拉起 worker、也不会注册发送 shutdown 哨兵的 atexit 钩子。引擎的所有权只属于启动器前端，挂载前端只拥有自己的请求通道和结果通道；关闭时它只 abort 自己的请求，不发送 None 哨兵。

为了让 worker 知道把结果送回哪个前端，client_id 的高位被用来编码 frontend id：

1
def _get_next_client_id(self) -> int:
2
    return namespace_client_id(self._frontend_id,
3
                               super()._get_next_client_id())

worker 侧再用 frontend_lane_index(response.client_id, len(lanes)) 选出对应的 PUSH lane。把路由信息塞进 ID 而不是单独维护映射表，好处是无状态：worker 只要看到 client_id 就能决定去向，不需要和 proxy 同步一张会过期的表。

到此，五种形态可以放在一张表里：

4.1 先登记结果，再发送请求

GenerationResult 必须在 request_queue.put() 之前登记进 self._results。原因在于结果分发线程的动作：它从 result_queue 收到一条响应后，用 client_id 去 self._results 里找对应的结果对象的队列。如果请求先被发送、worker 的响应先回来，而结果对象还没登记，分发线程就会因找不到 client_id 而丢弃响应——请求从此永远等不到结果。分布式系统中「先登记、后发送」是消费端可见性的必要条件。

登记之后又立刻再检查一次引擎是否死亡，是为了关闭另一个窗口：错误监控线程的 _mark_engine_dead() 会对当时 _results 的快照做一次性遍历，把 EngineDeadError 塞进每个待决结果。如果请求恰好在「登记完成」与「快照发生」之间插入，它会漏掉这次广播，永远阻塞在一个已经没有生产者的队列上。所以 submit() 在登记之后、发送之前再检查一遍，并在命中时把刚登记的条目删掉再抛错。

GenerationResult 的 client_id 其实就是 request.id，而 worker 内部还有另一个 id——PyExecutor 的 request_id。worker 维护映射 _client_id_to_request_id，用于取消请求。两个 id 的分工很清楚：client_id 属于边界层，用来配对请求与结果；request_id 属于引擎层，用来在调度器里标识一条序列。DeepInfer 系列里也能看到相似的双 id 设计，这是所有带前端的推理引擎绕不开的一层。

4.2 参数只传一次

GenerationResult.__init__ 中有一个很容易被忽略的字段：

1
# Avoid duplicate transmission of sampling_params for a request.
2
# SamplingParams is needed to create dummy results on postprocess workers.
3
self._params_transmitted = False

worker 侧在第一次发送某个 client_id 的响应时，会调用 _get_params_for_first_rsp() 把 sampling_params、postproc_params、disaggregated_params 一并带上；之后把 _params_transmitted 置位，后续响应不再携带这三份参数。如果没有这个标志，一条生成长 4096 token 的请求，每一步都要序列化一遍完整的采样参数、后处理参数与分离式参数，这些字段在 100 个并发请求、每步几十毫秒的节奏下就是一笔持续不断的固定开销。边界层的优化往往不是算子层面的，而是「哪些东西不需要重复过线」。

_get_logprob_params() 只在 prompt_logprobs 不为空时才构造参数，并且会把 drop_context_logits 提前算好：只要用户没有显式请求 context logits，或者启用了后处理并行（此时 logits 会被后处理进程接管），就不会把 logits 放进响应。这是一种「默认不传大对象」的策略，与许多框架把 logits 一律回传的做法不同。

4.3 可选的请求批量发送

默认情况下，_send_request() 直接 request_queue.put(request)。但如果设置了环境变量 TRTLLM_PROXY_BATCH_REQUESTS=1，则会走 _RequestSender：

1
class _RequestSender:
2
    MAX_BATCH_SIZE = 1024
3

4
    def send(self, payload: bytes) -> None:
5
        with self._cond:
6
            if self._closed:
7
                return
8
            if self._thread is None:
9
                self._thread = threading.Thread(target=self._run,
10
                                                daemon=True,
11
                                                name="proxy_request_sender")
12
                self._thread.start()
13
            self._pending.append(payload)
14
            self._cond.notify()
15

16
    def _run(self) -> None:
17
        while True:
18
            with self._cond:
19
                while not self._pending and not self._closed:
20
                    self._cond.wait()
21
                batch = [self._pending.popleft()
22
                         for _ in range(min(len(self._pending), self.MAX_BATCH_SIZE))]
23
            self._request_queue.put(batch)

它的设计取舍写在源码注释里，值得完整理解。在批量模式下，每条请求由调用线程先 pickle 成 bytes，再交给后台发送线程成批发送。这样做的收益是：一个突发请求（例如前端一次收到几百条 prompt）只需要一次 receive、一次 HMAC 校验，worker 再逐条 unpickle；代价是发送线程必须先拿到 GIL 才能发送，在 trtllm-serve 的 asyncio 事件循环繁忙时，单条请求最多要等到解释器的切换间隔（默认 5 ms）才被送出。

把三个参数放在一起看，就能判断什么时候该打开它：

• 请求突发强、单条延迟要求宽松的服务：开启收益明显；
• 延迟敏感、请求稀疏的交互式服务：不建议开启，5 ms 的 GIL 等待对首 token 延迟是实打实的。

另外，一旦开启，request_queue 的写入者就变成了发送线程；shutdown 的 None 哨兵也必须由同一个线程发送，否则 ZMQ socket 会被两个线程同时写，属于未定义行为。源码里 _close_request_sender(send_sentinel=...) 正是为了保证这一点。

4.4 worker 侧如何接住请求

rank0 worker 在 worker_main() 里完成 READY 之后进入主循环：

1
for req in _incoming_requests(request_queue):
2
    if isinstance(req, CancellingRequest):
3
        worker.abort_request(req.id)
4
    elif isinstance(req, StartProfileRequest):
5
        ...
6
        send_profile_ack("start", ack_err, req)
7
    elif isinstance(req, StopProfileRequest):
8
        ...
9
    elif isinstance(req, GenerationRequest):
10
        try:
11
            worker.submit(req)
12
        except RequestError as e:
13
            worker._await_response_helper.temp_error_responses.put(
14
                ErrorResponse(req.id, e, req.id))

_incoming_requests() 是哨兵协议的入口：

1
while (msg := request_queue.get()) is not None:
2
    if isinstance(msg, list):
3
        for payload in msg:
4
            yield pickle.loads(payload)
5
    else:
6
        yield msg

None 就是「关闭引擎」的哨兵。取消请求（CancellingRequest）与 profile 控制请求（StartProfileRequest、StopProfileRequest）跟生成请求共用同一根队列，因此它们的顺序被天然保证：取消必然在该请求之后被处理。这比另外维护一根取消通道要简单，也避免了「取消先于请求到达」的乱序问题。

worker.submit() 内部抛出的 RequestError 不会让 worker 退出，而是转成 ErrorResponse 放进 temp_error_responses。下一次响应分发时，process_responses() 会把这些错误响应拼进结果批次，由 proxy 分发回对应 client_id。这里体现了一个重要边界：请求级错误在 worker 内部就被转译成结果，不经过致命错误通道。

BaseWorker.submit() 还做了一件事：只允许 rank0 提交请求。其他 rank 调用会直接抛出 RuntimeError，并提示用户用 if __name__ == "__main__": 保护入口。这是多进程 + spawn 场景下的经典问题：每个子进程都会重新执行模块顶层，如果顶层建 LLM，就会同时出现多个 rank0。把这条检查写进 submit()，相当于给出了一个明确的报错，而不是让用户面对莫名其妙的 MPI 死锁。

6.1 先过滤空响应，再拼上延迟到达的提交错误

1
def process_responses(self, responses):
2
    responses = [r for r in responses if r is not None]
3

4
    while True:
5
        try:
6
            responses.append(self.temp_error_responses.get_nowait())
7
        except Empty:
8
            break
9

10
    return responses

await_responses() 可能返回 None 占位，必须先过滤。temp_error_responses 是 worker 在 submit() 抛 RequestError 时写入的错误队列；把这些错误拼进同一批响应，意味着「提交失败」和「执行失败」对调用者呈现为同一种结果：GenerationResult 的队列里收到一条错误，result() 抛出 RequestError。调用者不需要知道错误发生在边界内还是引擎内。

6.2 单进程路径：直接写进 GenerationResult 的队列

1
def handle_for_worker(self, responses):
2
    event_loop = None
3
    async_queues = []
4
    for response in responses:
5
        queue = self.worker.return_queue(response.client_id)
6
        if not response.has_error():
7
            response = _maybe_wrap_response(self.worker, response)
8

9
        if isinstance(queue, _SyncQueue):
10
            queue.put_nowait(response)
11
            async_queues.append(queue)
12
            event_loop = event_loop or queue.loop
13
        else:
14
            queue.put(response)
15

16
        if response.has_error() or response.result.is_final:
17
            self.worker._pop_result(response.client_id)
18

19
    if async_queues:
20
        _SyncQueue.notify_many(event_loop, async_queues)

return_queue() 的逻辑是：如果 worker 注册了中心化的 result_queue，就写到那里；否则直接写进该 client_id 对应的 GenerationResult.queue。单进程路径下没有中心化队列，所以每个请求各自一个队列。

这里的关键是 _SyncQueue 的批量唤醒。GenerationResult 在构造时根据当前线程有没有事件循环选择队列类型：

1
if has_event_loop():
2
    self.aqueue = AsyncQueue()
3
    self.queue = self.aqueue.sync_q
4
else:
5
    self.queue = Queue()
6
    self.aqueue = None

同步消费者拿到的是标准 queue.Queue；异步消费者拿到的是 AsyncQueue.sync_q，它底层是一个 asyncio 队列的同步视图。如果每条响应都调用一次 notify，一个 batch 里有 64 条响应就会唤醒事件循环 64 次；_SyncQueue.notify_many() 把这些通知合并成一次，事件循环被唤醒后一次性取走所有可用结果。这是流式服务里非常典型的「批量通知」优化，和前面请求批量发送是同一个思路的两端。

_pop_result() 只在 is_final 或 has_error 时调用，避免每个增量 token 都做一次字典删除。_maybe_wrap_response() 则在需要时把 LogProbsResult 与请求级性能指标包进 ResponseWrapper，不改变底层响应的类型。

6.3 IPC 路径：先分桶，再批量过线

IPC 路径的处理函数是 handle_for_ipc_batched()。它比单进程路径多做了三件事：错误类型转换、后处理分桶、以及按前端 lane 分拣。

1
def handle_for_ipc_batched(self, responses):
2
    postproc_batches = ([[] for _ in range(num_postprocess_workers)]
3
                        if self.enable_postprocprocess_parallel else None)
4
    rsp_batch = []
5

6
    for response in responses:
7
        if isinstance(response, ErrorResponse):
8
            pass
9
        elif self.worker._has_background_error():
10
            response = self.worker._create_error_response(response)
11
        elif response.has_error():
12
            response = ErrorResponse(response.client_id, response.error_msg,
13
                                     response.request_id)
14
        else:
15
            response = _maybe_wrap_response(self.worker, response)
16

17
        _send_rsp(self.worker, response,
18
                  postproc_batches=postproc_batches,
19
                  rsp_batch=rsp_batch)
20

21
    if postproc_batches:
22
        for wid, batch in enumerate(postproc_batches):
23
            if batch:
24
                self.worker.postproc_queues[wid].put(batch)
25

26
    if rsp_batch:
27
        if (lanes := self.worker.frontend_result_queues) is not None:
28
            for frontend_id, sub_batch in enumerate(
29
                    bucket_responses_by_frontend(rsp_batch, len(lanes))):
30
                if sub_batch:
31
                    lanes[frontend_id].put(sub_batch)
32
        else:
33
            self.worker.result_queue.put(rsp_batch)

为什么必须把 tllm.Response 转成 ErrorResponse？源码注释给出的原因是：带错误的 tllm.Response 无法被序列化。这是 C++ 绑定对象跨进程序列化的限制；错误信息必须降级成一个纯 Python 的 NamedTuple，才能通过 ZMQ 的 pickle 通道。

后处理分桶用的是 response.client_id % num_postprocess_workers，而不是随机分配。原因是增量 detokenize 依赖同一个请求之前的 CompletionOutput；只有把同一个 client_id 固定映射到同一个后处理进程，后者才能维护该请求的解码状态。这是一个「哈希分片 + 有状态 worker」的经典约束：分片函数必须稳定，扩容时不能改变已有请求的归属，否则状态会断裂。TensorRT-LLM 选择按 client_id 取模，简单且稳定；代价是后处理进程数量变化时，存量请求无法迁移。

_send_rsp() 的路由优先级也值得记住：

1. 如果启用了后处理并行，且响应不是带直连路由的 ErrorResponse，优先进 postproc 队列；
2. 否则如果有 frontend_result_queues，按 frontend_lane_index(client_id) 送到对应前端 lane；
3. 否则如果有 result_queue，追加到 rsp_batch，最后一次性 put；
4. 如果都没有，直接写进该请求的 GenerationResult.queue。

ErrorResponse 在「有直连路由」时会绕过 postproc，直接走结果队列。原因是 PostprocWorker 会读取 input.rsp.result，而错误响应没有这个字段；让它走直连队列，proxy 的 demux 会把它当成终止信号处理。代价是它可能超过同一请求更早的、仍在 postproc 队列里的响应；proxy 在收到错误时会 pop 掉该 client_id，随后迟到的正常响应会因找不到结果而被丢弃，只留下一条 benign 的 warning。这种「错误快于正常响应」的乱序是被接受的，因为错误本身已经终止了请求。

6.4 proxy 侧：分发线程是结果的总出口

proxy 的 dispatch_result_task() 是一个由 ManagedThread 驱动的循环：

1
def dispatch_result_task(self) -> bool:
2
    if (res := self.result_queue.get()) is None:
3
        return False
4

5
    async_queues = []
6
    event_loop = None
7

8
    def process_res(res):
9
        result = self._results.get(res.client_id)
10
        if result is None:
11
            return
12
        queue = result.queue
13
        if isinstance(queue, _SyncQueue):
14
            queue.put_nowait(res)
15
            async_queues.append(queue)
16
            event_loop = event_loop or queue.loop
17
        else:
18
            queue.put(res)
19

20
        if (is_llm_response(res) and res.result.is_final) or isinstance(res, ErrorResponse):
21
            self._results.pop(res.client_id, None)
22

23
    res = res if isinstance(res, list) else [res]
24
    for i in res:
25
        if i is None:
26
            return False
27
        process_res(i)
28

29
    if async_queues:
30
        _SyncQueue.notify_many(event_loop, async_queues)
31
    return True

它与 worker 的 handle_for_ipc_batched() 构成一对镜像：worker 负责把结果按 client_id 打包成 batch，proxy 负责拆包并按 client_id 找到调用者的队列。两边都用 notify_many 做一次批量唤醒，避免每个 token 一次事件循环调度。

res is None 是 worker 发出的关闭哨兵。pre_shutdown() 在关闭请求发送端时会发送一个 None，分发线程读到它就直接返回 False，ManagedThread 随之退出。这里也有一个容易忽略的边界：如果 worker 已经崩溃，哨兵永远不会到达，分发线程会永远阻塞在 result_queue.get()。所以 shutdown() 在 _engine_dead 为真时只给分发线程一个 5 秒的 join 上限，超时后直接泄漏这个 daemon 线程，而不是让整个进程卡死。这是「正确性优先于资源整洁」的典型取舍：进程即将退出，泄漏一个线程比永久挂起可接受得多。

result is None 的分支同样重要。请求可能已经因为错误被 pop，或者已经被用户 abort；此时 worker 的迟到响应会找不到结果。源码选择静默丢弃，而不是抛 KeyError。如果把这里改成断言，abort 场景下几乎必然崩溃，因为 abort 是异步的，worker 在收到取消前还会产出若干条响应。

7.1 两类错误

_handle_background_error() 根据错误对象的类型分流：

1
def _handle_background_error(self, error=None):
2
    if error is not None:
3
        if isinstance(error, RequestError):
4
            # 单请求错误，可以被捕获后忽略
5
            ...
6
        elif isinstance(error, str):
7
            # 单请求错误，字符串形式
8
            error = RequestError(error)
9
        else:
10
            # 后台线程或进程的严重错误
11
            self._set_fatal_error(error)
12
            self.shutdown()
13
        raise error
14

15
    if self._error_queue.empty():
16
        return
17

18
    try:
19
        e = self._error_queue.get_nowait()
20
        ...
21
        if isinstance(e, str):
22
            e = RequestError(e)
23
        elif not isinstance(e, BaseException):
24
            e = RuntimeError(repr(e))
25
        if not isinstance(e, RequestError):
26
            self._set_fatal_error(e)
27
            self.shutdown()
28
        raise e
29
    except Empty:
30
        pass

判断标准不是「错误来自哪里」，而是「错误对象是什么类型」。RequestError 表示「这条请求不合法或执行失败」，例如 max_tokens 超出上下文、adapter 加载失败、请求被引擎拒绝；字符串会被包装成 RequestError，因为跨进程通道里的错误信息往往已经退化成字符串。其他任何异常都被视为系统级错误：CUDA OOM、worker 进程崩溃、事件循环异常、MPI 通信失败。系统级错误会先记录为 _fatal_error，再触发 shutdown()，最后重新抛出。

这里有一个重要的顺序：先记录 fatal error，再 shutdown。因为 shutdown() 内部会调用 _handle_background_error() 做收尾；如果不先记录，收尾阶段的二次错误会覆盖原始错误，用户看到的就不是根因。_set_fatal_error() 本身也只保留第一条错误：

1
def _set_fatal_error(self, error):
2
    if self._fatal_error is None:
3
        self._fatal_error = error
4
        logger.error(f"Fatal engine error recorded: {repr(error)}")

注释承认存在一个很窄的 TOCTOU 竞争：两个线程可能同时通过 is None 检查，最终记录的是其中一条错误。但因为两条都是 fatal、都会触发 shutdown，这个竞争只影响日志里显示哪条错误，不影响行为。这种「明确写出可接受竞争」的注释，比假装没有竞争更有价值。

7.2 健康检查不能顺手触发阻塞式关闭

基类的 check_health() 与 _handle_background_error() 不同：它不抛异常，只返回布尔值，并且会排空整个错误队列，确保排在 RequestError 后面的 fatal error 不会被延迟到下一次检查才被发现。proxy 重写了它，原因很实际：基类版本在发现 fatal error 时调用 shutdown()，而 shutdown() 会阻塞等待 MPI future；如果 health check 是从事件循环线程调用的，整个 HTTP 服务会被卡住。proxy 版本因此只调用 pre_shutdown()（非阻塞），把阻塞式清理留给真正的 shutdown 路径。

1
def check_health(self):
2
    if self.doing_shutdown or self._fatal_error is not None:
3
        return False
4
    if self._drain_error_queue():
5
        return self._fatal_error is None and not self.doing_shutdown
6
    if self._check_mpi_workers():
7
        return False
8
    return True

_drain_error_queue() 只处理系统级错误；RequestError 被跳过，因为它不影响引擎健康。_check_mpi_workers() 同时检查本地 pid 监控和 MPI future，覆盖「进程还活着但 future 已完成」与「future 还没完成但进程已经消失」两种死法。

7.3 引擎死亡时，所有待决请求必须立刻醒过来

worker 崩溃最可怕的后果不是崩溃本身，而是调用者阻塞在一个永远不会再有生产者的队列上。TensorRT-LLM 用 EngineDeadError 广播来解决：

1
def _mark_engine_dead(self, error=None):
2
    if self._engine_dead:
3
        return
4
    self._engine_dead = True
5
    dead_error = EngineDeadError(error)
6
    for result in list(self._results.values()):
7
        try:
8
            result.queue.put(dead_error)
9
        except Exception:
10
            pass
11
    release = getattr(getattr(self, 'mpi_session', None),
12
                      'release_exit_joins', None)
13
    if release is not None:
14
        try:
15
            release()
16
        except Exception:
17
            pass

GenerationResult._result_step() 收到 EngineDeadError 时，会把它记为 _terminal_error、置 _done = True，然后抛出。之后再调用 result() 或 aresult()，会直接重新抛出同一条错误，而不会重新阻塞。这个「粘性终止错误」的设计非常重要：很多框架的 future 在失败后再次 get() 会再次等待；TensorRT-LLM 选择让失败永久可见。

_mark_engine_dead() 用 queue.put() 而不是 put_nowait()，注释解释得很清楚：异步路径的 _SyncQueue 需要 put() 才能正确唤醒等待中的事件循环，put_nowait() 只对同步队列有效。它还顺手调用 release_exit_joins()，因为解释器退出时会先 join 非 daemon 线程，再执行 atexit；如果 pool manager 线程卡住，必须在用户代码还活着的时候解除登记，否则进程会挂死在退出阶段。

worker 死亡的通知有两条路径：MPI future 的 add_done_callback 是事件驱动的，进程一退出就触发；_error_monitor_loop 每 5 秒轮询一次，是兜底。源码明确把事件驱动称为「延迟敏感路径」，把轮询称为「backstop」。远程 MPI 会话（RemoteMpiCommSessionClient）没有本地 future，只能靠轮询，因此它的故障发现延迟最多 5 秒。这个差异在多前端或跨机部署时会影响故障恢复时间，值得在运维层面知道。

7.4 shutdown 的顺序为什么必须写成这样

pre_shutdown() 是「通知阶段」，shutdown() 是「等待与清理阶段」。前者可以在任意线程调用，后者是阻塞的。

1
def pre_shutdown(self):
2
    if not self.workers_started:
3
        return
4
    if self.doing_shutdown:
5
        return
6
    self.doing_shutdown = True
7

8
    self._worker_process_monitor.close()
9
    self._shutdown_event.set()          # 叫醒 error monitor
10
    self._abort_all_requests()          # 给所有待决请求发 abort
11

12
    send_sentinel = (not self.mpi_futures or
13
                     any(not f.done() for f in self.mpi_futures))
14
    self._close_request_sender(send_sentinel=send_sentinel)

send_sentinel 的条件值得细读。它要覆盖两种情况：

1. 没有任何 future：RemoteMpiCommSessionClient 的 worker 在另一个进程里，submit() 返回空列表；如果不发哨兵，worker 永远收不到关闭信号，proxy 的分发线程会一直阻塞在 result_queue.get()。
2. 至少有一个 future 还活着：正常本地 worker，需要通知其退出。

如果所有 future 都已完成（worker 已经全死了），就不需要发哨兵；发了也没人收，反而可能在已关闭的 socket 上抛错。

shutdown() 的顺序是：

1
1. 若 workers 未启动：关闭请求发送端，清理多前端 IPC 目录，返回
2
2. 若尚未 pre_shutdown：先调用它
3
3. 等待 MPI futures：
4
   - engine_dead：最多等 5 秒，跳过仍未完成的 future
5
   - 正常：所有 future 共用一个 180 秒 deadline
6
   - 超时则把引擎标记为 dead，走 dead-engine 路径
7
4. join error monitor（最多 5 秒）
8
5. 停止并 join dispatch_result_thread（dead 时最多 5 秒）
9
6. 关闭 profile control 单线程执行器（wait=True）
10
7. 关闭 RPC client
11
8. 关闭 request/init/result/resource-governor/profile-ack 队列
12
9. 清理多前端 IPC 目录
13
10. 若拥有 MPI session：
14
    - engine_dead：abandon()，不 join 死掉的世界
15
    - 正常：shutdown()
16
11. 最后再调用一次 _handle_background_error()，处理收尾阶段的新错误

第 3 步的两个 timeout 是整段代码里最有信息量的地方。正常关闭给 180 秒，是因为大引擎的权重释放、进程组销毁、NCCL 通信器拆除可能确实很慢；引擎已死时只给 5 秒，是因为任何 join 死 worker 的操作都可能永久阻塞。源码没有试图「优雅地」等一个已经崩溃的世界，而是明确区分「可等待」与「不可等待」。

第 6 步之所以要 wait=True，是因为 start_profile / stop_profile 的 ZMQ 操作被固定在一个单线程执行器上。如果直接关闭 socket，正在等待 worker ACK 的 profile 调用会踩到已关闭的队列。这个执行器的存在本身就是为了满足 ZMQ 的线程安全约束：socket 不能被多个线程同时访问，而 FastAPI 的线程池不保证同一个请求落在同一个线程上。把「所有 profile 控制流量」钉到一个专用线程，是用一个线程换取正确性。

第 10 步的 abandon() 与 shutdown() 的区别同样重要。shutdown() 会 join pool manager 与 worker 进程；如果 worker 已经崩溃，join 会永远阻塞。abandon() 只放弃会话，不等待。源码在 _mark_engine_dead() 里还会调用 release_exit_joins()，进一步确保解释器退出时不会卡在非 daemon 线程上。这些看起来琐碎的清理，实际上是「推理服务能正常重启」与「进程必须被 kill -9」之间的差别。

7.5 初始化失败为什么不能走普通 shutdown

_fail_initialization() 是专门为启动阶段准备的：

1
def _fail_initialization(self, error, cause=None):
2
    root_cause = cause if cause is not None else error
3
    try:
4
        self._abort_owned_session(root_cause)
5
    except Exception as abort_error:
6
        logger.error(...)
7
    finally:
8
        self._set_fatal_error(root_cause)
9
        if not self.doing_shutdown:
10
            self.pre_shutdown()
11
    raise error from cause

顺序是「先 abort，再 mark」。如果反过来，_mark_engine_dead() 会调用 release_exit_joins()，把 pool 标记为 dead，从而让 shutdown_abort() 内部的阻塞式 shutdown 变成非阻塞，MPI_Abort 的升级路径就失效了。结果是非 leader rank 卡在初始化集合通信里，继续持有权重，直到作业被外部杀掉。源码把这一点写在注释里，并用 finally 保证即使 abort 抛错也会完成 mark，避免清理失败掩盖原始初始化错误。

启动路径还有一个专门的检测函数 _detect_worker_death_during_init()，它只检测、不记录、不清理。原因与上面相同：检测函数如果顺手调用 _check_mpi_workers()，就会触发 mark，破坏 abort 的顺序。因此初始化等待循环使用一个「纯检测」版本，把「发现」与「处理」分开。这种拆分在并发代码里很常见，但 TensorRT-LLM 把它写成了两个独立函数，而不是靠一个布尔参数控制副作用，可读性更好。

8.1 后处理进程是怎么被拉起的

worker_main() 在 rank0 上启动后处理池：

1
postproc_worker_pool = ProcessPoolExecutor(
2
    max_workers=postproc_worker_config.num_postprocess_workers)
3
for i in range(num_postprocess_workers):
4
    fut = postproc_worker_pool.submit(
5
        postproc_worker_main,
6
        result_queues[i].address,
7
        proxy_result_addrs,
8
        postproc_worker_config.postprocess_tokenizer_dir,
9
        PostprocWorker.default_record_creator,
10
        postproc_worker_config.post_processor_hook,
11
    )
12
    postprocess_worker_futures.append(fut)

每个后处理进程接收三样东西：属于自己的输入队列地址、结果 lane 的地址列表、以及可选的独立 tokenizer 目录。源码在 fork 之前显式调用了一次 gc.collect()，注释写得很具体：

A reused worker process may still hold cyclic garbage from the LLM it hosted before, including CUDA objects. Collect it here: a forked postprocess worker that finalizes them aborts on CUDA.

这是一个非常值得记住的工程细节。在外部 mpirun 启动的长期 worker 进程里，同一个 OS 进程可能先后承接过多个 LLM 实例；如果上一个实例留下的循环垃圾里包含 CUDA 对象，fork 出来的后处理子进程在解释这些对象析构时会触发 CUDA abort。显式 GC 一次，把风险留在父进程里，是成本极低的预防措施。这类问题不会在代码层面暴露，只会在特定部署形态下偶发，因此必须靠注释和纪律保留。

8.2 为什么每个后处理进程有自己的 tokenizer

postprocess_tokenizer_dir 是独立的配置项，而不是复用主 worker 的 tokenizer。原因是主 worker 的 tokenizer 可能来自 trust_remote_code=True 的自定义类，无法跨进程 pickle；后处理进程干脆自己从目录再加一份。源码在 _start_executor_workers() 里也做了对应处理：

1
# Strip the tokenizer from worker_kwargs to avoid MPI pickle failures.
2
# Tokenizers using trust_remote_code=True reference classes in the dynamic
3
# 'transformers_modules' namespace that cannot be deserialized on remote nodes.
4
# The tokenizer is not needed by workers; it is only consumed by the proxy / rank-0 leader.
5
mpi_worker_kwargs = {k: v for k, v in worker_kwargs.items() if k != 'tokenizer'}

这里有两个取舍叠在一起：一方面，tokenizer 只被 rank0 与后处理进程使用，其他 rank 根本不需要；另一方面，带 trust_remote_code 的 tokenizer 无法被 MPI 反序列化。与其试图让所有 rank 都能加载 tokenizer，不如直接不传，让 rank0 从本地目录自行构造。这是「减少跨进程依赖」的又一个实例，和前面参数只传一次、多模态输入只传一份是同一类思路。

8.3 后处理进程的输入需要一份「假的 GenerationResult」

后处理进程需要 sampling_params 来构造一个用于增量解码的临时结果对象。但完整地跨进程传一遍 GenerationResult 代价太高，所以 _send_rsp_to_postproc() 只传一个 PostprocWorker.Input，其中采样参数由 _get_params_for_first_rsp() 决定是否附带：

1
def _get_params_for_first_rsp(worker, client_id):
2
    res = worker._results.get(client_id, None)
3
    assert res is not None
4
    if not res._params_transmitted:
5
        res._params_transmitted = True
6
        return res.sampling_params, res.postproc_params, res.disaggregated_params
7
    return None, None, None

这就是 4.2 节提到的「参数只传一次」的具体应用。后处理进程在收到第一次输入时用这些参数建立本地状态，后续输入不再携带。需要注意，_params_transmitted 是在 worker 侧设置的，而不是后处理进程侧；worker 是唯一的决策者。

8.4 分桶与跨请求顺序

pid = response.client_id % num_postprocess_workers 把同一个请求固定到一个进程。这一约束保证了增量解码状态的一致性，但也带来一个副作用：一个慢请求会拖慢同桶里所有请求的后处理进度。对延迟敏感的服务，后处理进程数量应该足够多，让桶内请求数保持在较低水平；或者干脆不启用后处理并行，把 tokenizer 工作留在 proxy 主进程。

顺带说明一个容易混淆的点：后处理并行时的 result_queue 在 proxy 侧是 PULL，而 worker 侧每个后处理进程各自 PUSH 到同一个地址。如果某个后处理进程崩溃，它负责的桶里的请求会永久得不到最终结果。TensorRT-LLM 并没有为这种情况单独设计恢复路径，故障最终会通过 worker 世界的 future 死亡或前端超时暴露出来。这说明后处理并行是一个性能选项，不是可靠性选项；需要更强容错的部署不应依赖它。