vLLM 源码完全拆解（二十九）：异步调度——调度器为什么可以在采样结果落回 CPU 之前，就决定下一步跑谁

同步解码的关键路径上，CPU 在等什么

先看清楚要省掉的是什么。一次 decode step 在同步调度下的时间线大致是：

1
CPU: schedule(S_t) ──启动前向──┐                        ┌── update_from_output(S_t) ── schedule(S_{t+1})
2
GPU:                 │ execute(S_t) → sample → D2H ────┘

CPU 把 S_t 交给执行器后，前向启动即返回（execute_model(..., non_block=True) 得到一个 future），然后在 future.result() 上等待采样完成与结果回传。GPU 真正的工作量很小——每请求一个 token 的 decode 本来就是访存密集型，算力利用率低——但 CPU 在等的这段时间里什么都没做：它既不能提前调用 schedule，也不能提前准备下一批输入。整个系统在 decode 阶段被压成「GPU 跑一下、CPU 等一下」的锯齿，而这个锯齿里 CPU 侧的等待往往与 GPU 的实际执行时间同量级，因为中间还夹着采样结果从 GPU 拷回主机、多进程间传递 SamplerOutput、以及 Python 侧的逐请求记账。

异步调度的目标很明确：让 CPU 在 GPU 执行第 t 步的同时，就把第 t+1 步的调度做完。时间线变成：

1
CPU: schedule(S_t) → schedule(S_{t+1}) → update(S_t) → schedule(S_{t+2}) → update(S_{t+1}) ...
2
GPU:        execute(S_t) → sample →          execute(S_{t+1}) → sample → ...

但问题立刻出现：第 t+1 步的调度依赖于第 t 步的采样结果。如果 token 还没采出来，调度器怎么知道第 t+1 步该给这个请求分配多少 token、它是否已经达到 max_tokens、KV 需要预留几个块？

答案是先按最坏情况记账，等结果回来再对账。这正是异步调度全部复杂度的来源。

在架构上，这套「同时维持两批在途」的能力来自 EngineCore 的批量队列：异步调度把 max_concurrent_batches 从 PP 专属的 1 提升到 2（V1 模型执行器）或 pp_size + 1（V2 模型执行器），于是 EngineCore.step_fn 从单步的 step 切换为 step_with_batch_queue。下文的占位符账本，全部是在这个双缓冲结构里成立的。

异步调度是建立在 V1 的多进程拓扑之上的：每个 API Server 对应一个 EngineCore，再向一组 GPU Worker 分发前向。异步调度省掉的那次「等待」发生在图中的 API Server ↔ Engine Core ↔ GPU Worker 这条链路上——CPU 侧的三段各自可以并行推进。

图 1：vLLM V1 进程架构（TP=2、DP=4，共 17 个进程：4 个 API Server + 4 个 Engine Core + 8 个 GPU Worker + 1 个 DP Coordinator）。来源：vLLM 官方设计文档 arch_overview.md。

开关、默认值与兼容性边界

入口在 SchedulerConfig 的 async_scheduling 字段（vllm/config/scheduler.py 中声明为 async_scheduling: bool | None = None），对应 CLI 的 --async-scheduling / --no-async-scheduling。

缺省值是 None 不是小事：VllmConfig 的初始化阶段会把它解析成真正的布尔值。解析逻辑很值得读——它决定了在什么组合下异步调度「想开但开不了」：

1
executor_supports_async_sched = executor_class.supports_async_scheduling()
2
...
3
# Async scheduling incompatibilities specific to the V1 model runner.
4
v1_async_sched_incompat: str | None = None
5
if not self.use_v2_model_runner:
6
    spec_config = self.speculative_config
7
    if spec_config is not None and (
8
        spec_config.method not in get_args(EagleModelTypes)
9
        and spec_config.method not in get_args(NgramGPUTypes)
10
        and spec_config.method not in ("draft_model", "dspark", "dflash")
11
    ):
12
        v1_async_sched_incompat = f"{spec_config.method}-based speculative decoding"
13
    elif spec_config is not None and spec_config.disable_padded_drafter_batch:
14
        v1_async_sched_incompat = "disable_padded_drafter_batch=True"
15
    elif self.parallel_config.pipeline_parallel_size > 1:
16
        v1_async_sched_incompat = "pipeline parallelism"
17

18
if self.scheduler_config.async_scheduling:
19
    # Async scheduling explicitly enabled, hard fail any incompatibilities.
20
    if v1_async_sched_incompat is not None:
21
        raise ValueError(
22
            "Async scheduling is not supported by the V1 model runner "
23
            f"with {v1_async_sched_incompat}."
24
        )
25
    ...
26
elif self.scheduler_config.async_scheduling is None:
27
    # Enable async scheduling unless there is an incompatible option.
28
    if self.model_config.runner_type == "pooling":
29
        self.scheduler_config.async_scheduling = False
30
    elif not executor_supports_async_sched:
31
        ...
32
        self.scheduler_config.async_scheduling = False
33
    elif v1_async_sched_incompat is not None:
34
        ...
35
        self.scheduler_config.async_scheduling = False
36
    else:
37
        self.scheduler_config.async_scheduling = True

三条信息：其一，在 V1 模型执行器上，异步调度目前是默认开启的（用户没显式关、执行器支持、没有不兼容项）；其二，V1 上它不支持流水并行，也不支持不经过草稿头对齐的投机方法（除 EAGLE 系、N-gram 系与 draft_model、dspark、dflash 之外，例如 Medusa 式的多头草稿在 V1 上与之不兼容）；其三，用户显式打开时遇到不兼容组合不是「静默降级」，而是直接抛错。这条「显式开=硬失败，默认开=软降级」的区分，是为了防止用户在以为开了异步调度的情况下实际跑在同步路径上。

开关打开之后，这个标志在四个地方被观察：

• SchedulerConfig.get_scheduler_cls() 据此返回 AsyncScheduler 而不是 Scheduler；
• VllmConfig.max_concurrent_batches 据此把在途批次数从 PP 专用的 pp_size 提升到至少 2（V1 是 2，V2 是 pp_size + 1），EngineCore 再据此启用 batch_queue 并把 step_fn 切到 step_with_batch_queue；
• Scheduler 基类根据它决定 update_from_output 是否要扣减占位符、是否延迟释放 KV 块；
• GPUModelRunner 读取 self.use_async_scheduling，几乎每一处需要「CPU 与 GPU 同步」的地方都会分叉。

批次数这条是容易被忽略的根：没有第二批在途，占位符就无从预支。VllmConfig.max_concurrent_batches 的注释写得很直白：

1
@property
2
def max_concurrent_batches(self) -> int:
3
    # PP requires PP-size concurrent batches to fill the pipeline.
4
    # Async scheduling requires 2 concurrent batches to overlap.
5
    pp_size = self.parallel_config.pipeline_parallel_size
6
    if self.scheduler_config.async_scheduling:
7
        if self.use_v2_model_runner:
8
            return pp_size + 1
9
        # V1 Model Runner does not fully support async scheduling with PP.
10
        if pp_size <= 1:
11
            return 2
12
    return pp_size

此外 V1 里还有一个与之正交的 VLLM_USE_V2_MODEL_RUNNER（use_v2_model_runner）标记，异步调度与它组合时多出流水并行下的「错拍解码」（decode_stagger），本篇后面单独讲。只开异步调度、不开投机解码，是最干净的形态，我们先把它讲透。

1
class AsyncScheduler(Scheduler):
2
    def __init__(self, *args, **kwargs) -> None:
3
        super().__init__(*args, **kwargs)
4
        # reusable read-only placeholder list for speculative decoding.
5
        self._spec_token_placeholders: list[int] = [-1] * self.num_spec_tokens
6
        self.pp_size = self.parallel_config.pipeline_parallel_size
7
        # Only XPU can disable PP decode microbatching via VLLM_XPU_PP_MICROBATCH.
8
        self.decode_stagger = self.pp_size
9
        if current_platform.is_xpu() and not envs.VLLM_XPU_PP_MICROBATCH:
10
            self.decode_stagger = 1

AsyncScheduler 本身只是一个很薄的子类，它没有重写 schedule，只重写了 _update_after_schedule——在每次调度完成后，立刻为下一步预支账本。真正的语义分叉散落在 Scheduler、Request、EngineCore、GPUModelRunner 与 InputBatch 里。

占位符：num_output_placeholders 到底是什么

异步调度全部机制的枢纽，是 Request 上的一个计数器 num_output_placeholders。它长在 V1 的统一调度模型上——V1 不再区分 prefill 与 decode 阶段，每个请求每一步拿到的就是「本步允许计算的 token 数」，调度结果是一个 {request_id: num_tokens} 的字典：

图 2：V1 调度器的统一预算视图。三个请求在每一步各拿到多少 token，由同一个 token budget 决定，prefill 与 decode 不再分两套调度器。来源：vLLM V1 发布博客。

理解占位符需要先理解 Request 上的三个 token 相关计数：

• num_prompt_tokens：提示词长度，请求创建后不变；
• num_output_tokens：已经确认的输出 token 数，即这些 token 已经在 CPU 侧写进请求对象、可以被客户端看到、可以被停止条件检查；
• num_computed_tokens：已经被 GPU 计算过 KV 的 token 总数（提示词 + 输出）。

第四个量是 num_tokens_with_spec = num_tokens + num_spec_tokens，即在开启投机解码时，「如果草稿全部被接受，这个请求一共会有多少 token」的乐观估计。

num_output_placeholders 是第五个量，含义是：调度器已经替这个请求预支了、但还没有被真实采样结果确认的输出 token 数。它同时被计入 num_computed_tokens 的推算，也就是说调度器在规划时「假装」这些 token 已经被算过 KV。

这三个量之间的不变量可以写成：

在 decode 稳态下，若上一步调度的 token 已经全部落在 KV 里，num_tokens_not_yet_computed 为 0，于是 num_computed_tokens = num_tokens + num_output_placeholders。换句话说，调度器把「下一步将要采出来的那个 token」也算进了已计算。

为什么要这样算？因为异步调度的下一步调度发生在第 t 步前向执行期间，调度器不知道第 t 步会采出什么、更不知道它会被接受还是不接受（投机场景）。它唯一能做的，是假设「第 t 步的采样一定会成功、一定会产生一个 token」——这是 decode 的正常情况——然后据此推进账本。如果假设错了（请求在 max_tokens 边界停住、投机草稿全部被拒、KV 装载失败导致 token 被丢弃），update_from_output 就必须把这笔预支还回来。

1
if (
2
    request.num_output_placeholders > 0
3
    # This is (num_computed_tokens + 1) - (num_output_placeholders - 1).
4
    # Since output placeholders are also included in the computed tokens
5
    # count, we subtract (num_output_placeholders - 1) to remove any draft
6
    # tokens, so that we can be sure no further steps are needed even if
7
    # they are all rejected.
8
    and request.num_computed_tokens + 2 - request.num_output_placeholders
9
    >= request.num_prompt_tokens + request.max_tokens
10
):
11
    # Async scheduling: Avoid scheduling an extra step when we are sure that
12
    # the previous step has reached request.max_tokens.
13
    req_index += 1
14
    continue

1
# Case 2: this step samples the request's last output token. Drafts may
2
# all be rejected, so count only the output tokens the request is sure
3
# to have once this step finishes: the verified ones, plus one from each
4
# step that samples. This step samples one.
5
min_num_output_tokens = request.num_output_tokens + 1
6
if request.num_output_placeholders > 0:
7
    # Async scheduling: a step that samples is still in flight, and it
8
    # adds one more.
9
    min_num_output_tokens += 1
10
# Reaching max_tokens finishes the request, so no step verifies the drafts.
11
return min_num_output_tokens >= request.max_tokens

1
num_scheduled_spec_tokens = (
2
    num_new_tokens
3
    + request.num_computed_tokens
4
    - request.num_tokens
5
    - request.num_output_placeholders
6
)

1
if num_scheduled_spec_tokens > 0:
2
    spec_token_ids = request.spec_token_ids
3
    if len(spec_token_ids) > num_scheduled_spec_tokens:
4
        spec_token_ids = spec_token_ids[:num_scheduled_spec_tokens]
5
    scheduled_spec_decode_tokens[request.request_id] = spec_token_ids

_update_after_schedule：调度完立刻预支下一步

AsyncScheduler._update_after_schedule 是每次调度完成后、执行前向之前被调用的钩子。它做三件事：为输出与草稿各预支占位符、为结构化输出标记 pending、为 V2 模型执行器标记下一步的 decode 资格。

1
def _update_after_schedule(self, scheduler_output: SchedulerOutput) -> None:
2
    super()._update_after_schedule(scheduler_output)
3
    spec_decode_tokens = scheduler_output.scheduled_spec_decode_tokens
4
    # Use the latest num of scheduled draft tokens in next step as placeholder.
5
    self._spec_token_placeholders = [-1] * scheduler_output.num_spec_tokens_to_schedule
6
    for req_id in scheduler_output.num_scheduled_tokens:
7
        request = self.requests[req_id]
8
        if request.is_prefill_chunk:
9
            continue
10

11
        scheduler_output.pending_structured_output_tokens |= (
12
            request.use_structured_output and request.num_output_placeholders > 0
13
        )
14
        # The request will generate num_sampled_tokens_per_step new tokens
15
        # plus num_spec_tokens in this scheduling step. Diffusion has no AR
16
        # bonus token (num_sampled_tokens_per_step == 0) — only the canvas
17
        # (spec) tokens.
18
        cur_num_spec_tokens = len(spec_decode_tokens.get(req_id, ()))
19
        request.num_output_placeholders += (
20
            self.num_sampled_tokens_per_step + cur_num_spec_tokens
21
        )
22
        # Add placeholders for the new draft/spec tokens.
23
        # We will update the actual spec token ids in the worker process.
24
        request.spec_token_ids = self._spec_token_placeholders
25

26
        if self.use_v2_model_runner:
27
            # Set the next step index in which this request is eligible to be
28
            # scheduled for decode (for PP microbatching).
29
            request.next_decode_eligible_step = (
30
                self.current_step + self.decode_stagger
31
            )

逐条拆开：

它只对 decode 段生效。is_prefill_chunk 的请求直接跳过——分块预填充的中间块不采样任何 token，没有可预支的东西。

预支的额度是「每步采样数 + 草稿数」。在 AR（自回归）模型里，num_sampled_tokens_per_step 是 1：每个 decode 步骤采一个 token。若开了投机解码，还要加上这一步调度的草稿数 cur_num_spec_tokens，因为下一步的验证会把这几个草稿位置当作「已计算的候选 token」，若草稿被接受它们就会成为真实输出。对扩散语言模型（diffusion LLM）来说，num_sampled_tokens_per_step 为 0——没有 AR 的 bonus token，只有 canvas 上的 spec token——这一行注释说明框架已经为「非自回归采样」做了区分。

spec_token_ids 被替换成占位列表。这是异步调度里最反直觉的一笔：调度器在调度完下一步时，直接把这个请求的 spec_token_ids 填成 [-1] * num_spec_tokens，而不是等真正的草稿生成出来再回填。

1
# reusable read-only placeholder list for speculative decoding.
2
self._spec_token_placeholders: list[int] = [-1] * self.num_spec_tokens

注意「read-only」这个注释——这个列表是多个请求共享的同一个对象，所有请求的 spec_token_ids 都指向它。之所以能共享，是因为在 worker 进程里它会被整体替换：真正的草稿 token 在上一步 GPU 采样完成之后、下一步输入准备之前，由模型执行器自己取出来并写入 InputBatch，根本不需要经过调度器。调度器这边只是在「下一步要几个验证位置」的记账里占个位。

pending_structured_output_tokens 的标记也建立在占位符上：只要这个请求使用结构化输出并且还有未结算的占位符，这一步就存在「语法要依赖上一步才能算出的 token」的可能性，因此要告知 EngineCore 这一拍的采样需要推迟（机制见后文「结构化输出」一节）。

流水并行的错拍标记只在 V2 模型执行器上启用：next_decode_eligible_step = current_step + decode_stagger。同步流水并行里，一条请求的 decode 结果要从最后一个 rank 回传到第一个 rank 才能进入下一步输入，这本身就有 pp_size 拍的固有延迟。异步调度在 V2 上把这个延迟显式写进请求状态，让调度器不要让这条请求在下一个 rank 准备好之前就进入下一拍。

1
for req_id, num_scheduled_token in num_scheduled_tokens.items():
2
    request = self.requests[req_id]
3
    request.num_computed_tokens += num_scheduled_token
4
    request.num_in_flight_tokens += num_scheduled_token
5
    ...
6
    request.is_prefill_chunk = request.num_computed_tokens < (
7
        request.num_tokens + request.num_output_placeholders
8
    )

1
num_new_tokens = (
2
    request.num_tokens_with_spec
3
    + request.num_output_placeholders
4
    - request.num_computed_tokens
5
)

1
request.num_computed_tokens = 0
2
if request.spec_token_ids:
3
    request.spec_token_ids = []
4
request.num_stale_output_tokens = request.num_in_flight_tokens
5
request.num_output_placeholders = 0

边界条件与已知限制

异步调度的正确性建立在一份账、两处写之上：调度器写一次（预支），模型执行器写一次（回填与对账）。以下边界条件在这套结构下被显式处理，任何改动都需一并审视：

其一，请求在途被中止。EngineCore 在每个 worker 响应返回、update_from_output 执行之前，会先 _process_aborts_queue() 处理这一拍期间收到的中止请求。update_from_output 的循环里对 request is None or request.is_finished() 的请求直接跳过——注释点明这种情况在流水并行与异步调度里都会出现。被跳过的请求不再扣占位符，其 KV 块通过「块回收围栏」延迟归还。

其二，KV 装载失败（前缀缓存/P/D 分离）。kv_connector_output.failed_recving 涉及的请求会被移出本轮更新，num_computed_tokens 回退到「最后一批成功装载的块」的末尾，并要求重新计算失效块。异步模式下这个回退必须与占位符同步，否则下一步的 num_new_tokens 会基于错误的 num_computed_tokens 计算。

其三，扩散语言模型（diffusion LLM）。num_sampled_tokens_per_step 在扩散模型上是 0——没有 AR 的 bonus token，只有 canvas 上的 spec token 需要被验证。_update_after_schedule 的注释专门说明了这一点，意味着 diffusion 的异步调度在「预支多少」上与 AR 不同：只预支草稿位置，不预支额外采样位置。EngineCore 在 diffusion 配置下还会把 scheduler_cls 显式设为 vllm.v1.core.sched.diffusion_scheduler.DiffusionAsyncScheduler 或 DiffusionScheduler，说明扩散路径有自己一套占位符语义。

其四，多引擎与 DP。异步调度默认把 disable_nccl_for_dp_synchronization 置为 True，DP 同步改走 Gloo。注释给出的理由是 MoE 模型在 DP>1 时 NCCL 同步与异步流水冲突；这个默认值可以被用户覆盖，但覆盖后的行为需要单独验证。

其五，混合 KV Cache（Mamba、滑窗）。GPUModelRunner 在准备输入时有一段专门的跳过逻辑：

1
# Skipped under async scheduling (non-align): the CPU copy races with
2
# the in-flight D2H copy and with input-batch row moves.
3
needs_cpu_accepted_counts = self.num_accepted_tokens_event is not None and not (
4
    self.use_async_scheduling and self.cache_config.mamba_cache_mode != "align"
5
)

num_accepted_tokens 是投机解码（EAGLE 式动态树）里每个请求已接受 token 的计数，混合模型需要它来决定状态槽的回滚位置。异步调度下这份 CPU 副本与在途的 D2H 拷贝、以及 batch 行的重排（condense）会竞争，所以框架主动跳过读 CPU 副本，改在 GPU 上直接按 prev_positions 映射重排。这是「异步模式下所有 CPU 副本都要重新审视」的一个典型例子：一个在同步模式下理所当然的读取，在异步模式下变成了竞态源。站内 vLLM 混合 KV Cache 那篇（KVCacheCoordinator 如何让全注意力、滑窗与 Mamba 层共用一个块池）讲过这类模型的块池组织，异步调度在它之上再叠一层「迟一拍」的输入语义。

update_from_output：结果回来时怎么对账

update_from_output 在异步模式下做两件事：把在途的采样 token 写回请求对象（并扣减占位符），以及处理比预期少的 token（停止、KV 装载失败、投机被拒）。

先说返回 token 的写入：

1
def _update_request_with_output(
2
    self, request: Request, new_token_ids: list[int], is_stale: bool = False
3
) -> tuple[list[int], bool]:
4
    num_new_tokens = len(new_token_ids)
5
    stopped = False
6
    for num_new, output_token_id in enumerate(new_token_ids, 1):
7
        request.append_output_token_ids(output_token_id)
8
        # Check for stop and update request state.
9
        stopped = check_stop(request, self.max_model_len)
10
        if stopped:
11
            # num_computed_tokens covers all new tokens but the last one.
12
            # Keep that true after trimming: resumable requests resume from it.
13
            if not is_stale:
14
                request.num_computed_tokens -= len(new_token_ids) - num_new
15
            del new_token_ids[num_new:]  # Trim new tokens if needed.
16
            break
17

18
    if not is_stale and request.num_output_placeholders > 0:
19
        # Resolve all returned tokens, including those trimmed at a stop.
20
        request.num_output_placeholders -= num_new_tokens
21
        assert request.num_output_placeholders >= 0
22
    return new_token_ids, stopped

几个关键点：

减的是「返回的 token 数」，不是「被采纳的 token 数」。注释里说得很清楚——「including those trimmed at a stop」。如果一个请求在停止符处被截断，append_output_token_ids 之后的循环提前 break，后面的 token 被删掉，但它们对应的占位符仍然在这笔扣减里结算。这是因为从调度器的视角，这一步已经预支了这么多 token，不管它们最终是否被保留，KV 都已经算过了，账本必须平衡。

num_computed_tokens 的回退。当请求被截断时，最后一个 token 的 KV 虽然被算了却没有进入输出列表，num_computed_tokens 要扣回去，否则可恢复请求（resumable）的恢复点会错位。可恢复请求这一族在取消与抢占那篇里讲过——流式会话结束一条、下一条从「已计算但未输出的最后一个 token 之前」继续。

is_stale 的语义。在 KV 装载失败或者请求被中途打断时，模型执行器可能返回了一批属于旧步骤的 token。这些 token 不能进入请求对象（否则客户端会看到重复或错序的输出），但占位符已经在调度时预支了，所以 is_stale 分支扣的是 request.num_in_flight_tokens。safe_frontier 的概念也来自这里：

1
safe_frontier = request.num_computed_tokens - request.num_output_placeholders
2
assert safe_frontier >= 0

safe_frontier 是「不依赖任何在途结果的最小安全进度」——num_computed_tokens 里包含了预支的占位符，减去它们之后剩下的才是真正确定的 KV 长度。当一条请求被停掉需要恢复时，它必须从这个安全边界开始，而不是从乐观的 num_computed_tokens。

KV 块回收的延后一拍在异步模式下也有专门处理。注释说得很直接：

1
# Every GPU write enqueued by this and earlier steps has completed, so it is
2
# safe to return deferred-free blocks to the pool.
3
if self.defer_block_free and scheduler_output.total_num_scheduled_tokens > 0:
4
    self.processed_step_seq += 1
5
    self._drain_deferred_frees()
6
    self._drain_deferred_loads()

defer_block_free 即「延后释放」模式：一个块在被请求释放之后，不一定立即回到空闲池，因为可能还有一次已经入队但还没有真正执行完的 GPU 写（例如上一步调度里引用它写 KV）。异步模式下块释放一律走延迟队列——这正是取消与抢占那篇里说的「块回收的围栏」，在异步模式下变成了常态。

模型执行器：采样结果不落 CPU，直接当下一步的输入

光有调度器预支账本还不够。如果每步仍然要把 SamplerOutput.sampled_token_ids 拷回 CPU、传给 EngineCore、再由它传回给 GPUModelRunner 作为下一步的 input_ids，那条往返本身就吞掉了大半收益。异步调度的另一半在模型执行器里：采样结果留在 GPU 上，下一步的输入准备直接从 GPU 张量里填。

1
class AsyncGPUModelRunnerOutput(AsyncModelRunnerOutput):
2
    def __init__(self, model_runner_output, sampled_token_ids, logprobs_tensors,
3
                 invalid_req_indices, async_output_copy_stream, vocab_size, ...):
4
        self._model_runner_output = model_runner_output
5
        self._invalid_req_indices = invalid_req_indices
6
        # Event on the copy stream so we can synchronize the non-blocking copy.
7
        # Blocking (sleep) event to avoid busy-polling the CUDA driver lock.
8
        self.async_copy_ready_event = torch.cuda.Event(blocking=True)
9
        # Keep a reference to the device tensor to avoid it being
10
        # deallocated until we finish copying it to the host.
11
        self._sampled_token_ids = sampled_token_ids
12

13
        # Initiate the copy on a separate stream, but do not synchronize it.
14
        default_stream = torch.cuda.current_stream()
15
        with torch.cuda.stream(async_output_copy_stream):
16
            async_output_copy_stream.wait_stream(default_stream)
17
            self.sampled_token_ids_cpu = self._sampled_token_ids.to(
18
                "cpu", non_blocking=True
19
            )
20
            ...
21
            self.async_copy_ready_event.record()

1
def enqueue_output(self, output: Any):
2
    if isinstance(output, AsyncModelRunnerOutput):
3
        try:
4
            output = output.get_output()
5
        except Exception as e:
6
            ...

1
if not self.use_async_scheduling:
2
    with gpu_sync_allowed():
3
        # Get the valid generated tokens.
4
        max_gen_len = sampled_token_ids.shape[-1]
5
        ...
6
else:
7
    # Cache the sampled tokens on the GPU and avoid CPU sync.
8
    # These will be copied into input_ids in the next step
9
    # when preparing inputs.
10
    # With spec decoding, this is done in propose_draft_token_ids().
11
    if self.input_batch.prev_sampled_token_ids is None:
12
        assert sampled_token_ids.shape[-1] == 1
13
        self.input_batch.prev_sampled_token_ids = sampled_token_ids
14
    self.input_batch.prev_req_id_to_index = {
15
        req_id: i
16
        for i, req_id in enumerate(self.input_batch.req_ids)
17
        if i not in invalid_req_indices_set
18
    }

1
for req_idx in range(num_sampled_tokens):
2
    if self.use_async_scheduling:
3
        sampled_ids = [-1] if req_idx not in invalid_req_indices_set else None
4
    else:
5
        sampled_ids = valid_sampled_token_ids[req_idx]

下一步输入：GPU 上的 scatter_

下一步输入准备的关键在 GPUModelRunner._prepare_inputs（对应函数 _prepare_input_ids）里的 prev_sampled_token_ids 分支——这是异步调度「省掉往返」落地的最后一公里：

1
if self.input_batch.prev_sampled_token_ids is None:
2
    # Normal scheduling case
3
    self.input_ids.copy_to_gpu(total_num_scheduled_tokens)
4
    ...
5
    return
6

7
# Async scheduling case, where some decode requests from the previous
8
# iteration won't have entries in input_ids_cpu and need to be copied
9
# on the GPU from prev_sampled_token_ids.
10
prev_positions = self.prev_positions.np[:num_reqs]
11
for cur_index in range(num_reqs):
12
    prev_index = prev_positions[cur_index]
13
    if prev_index < 0:
14
        continue
15
    ...
16
    flattened_index = cu_num_tokens[cur_index].item() - 1
17
    sample_flattened_indices.append(flattened_index - draft_len)
18
    spec_flattened_indices.extend(
19
        range(flattened_index - draft_len + 1, flattened_index + 1)
20
    )

展开收起

逐条拆：

prev_positions 建立了「当前 batch 下标 → 上一拍下标」的映射，新请求映射为 -1，直接跳过——它们的输入 token 全在 CPU 侧的 input_ids_cpu 里，照常上传。

flattened_index 是「这一步扁平化后的最后一个 token 的位置」。vLLM 把这一拍的输入拼成一条一维序列，每个请求占 num_scheduled_tokens 个位置，cu_num_tokens 是前缀和。最后一个位置前面的 draft_len 个位置放的是草稿 token，再前面一个位置才是「上一拍采样出来的 token」——也就是 flattened_index - draft_len。

草稿 token 也被单独记下标（spec_flattened_indices），因为它们同样不在 CPU 侧的 input_ids_cpu 里：调度器在 _update_after_schedule 把 spec_token_ids 填成了 -1，worker 侧的 _draft_token_ids 才是真值。

真正执行回填的两段代码，分别对应「batch 完全没变」与「batch 有增删/重排」：

1
if common_indices_match and max_flattened_index == (num_common_tokens - 1):
2
    # Common-case optimization: the batch is unchanged
3
    # and no reordering happened.
4
    # The indices are both the same permutation of 0..N-1 so
5
    # we can copy directly using a single slice.
6
    self.input_ids.gpu[:num_common_tokens].copy_(
7
        self.input_batch.prev_sampled_token_ids[:num_common_tokens, 0],
8
        non_blocking=True,
9
    )
10
    return
11
# Upload the index tensors asynchronously so the scatter can be non-blocking.
12
sampled_tokens_index_tensor = torch.tensor(
13
    sample_flattened_indices, dtype=torch.int64, pin_memory=PIN_MEMORY
14
).to(self.device, non_blocking=True)
15
prev_common_req_indices_tensor = torch.tensor(
16
    prev_indices, dtype=torch.int64, pin_memory=PIN_MEMORY
17
).to(self.device, non_blocking=True)
18
self.input_ids.gpu.scatter_(
19
    dim=0,
20
    index=sampled_tokens_index_tensor,
21
    src=self.input_batch.prev_sampled_token_ids[prev_common_req_indices_tensor, 0],
22
)

展开收起

稳态路径是一次 GPU 到 GPU 的切片拷贝。当这一拍的 batch 与上一拍完全相同、顺序也相同时，prev_sampled_token_ids[:, 0] 本身就按顺序对应每个请求最后一个位置，一句 copy_ 搞定——CPU 一行都不碰。

非稳态路径是 scatter_。下标张量先在 CPU 侧构造（pin_memory）再异步传到 GPU，然后用 scatter_ 把「上一拍的采样列」按 sample_flattened_indices 散布到这一拍的 input_ids 上。整个过程没有 CPU 读取 token 值，CPU 只知道位置。

上一步的采样张量还受输入准备事件的保护：

1
@contextmanager
2
def synchronize_input_prep(self):
3
    if self.prepare_inputs_event is None:
4
        yield
5
        return
6
    # Ensure prior step has finished with reused CPU tensors.
7
    # This is required in the async scheduling case because
8
    # the CPU->GPU transfer happens async.
9
    self.prepare_inputs_event.synchronize()
10
    try:
11
        yield
12
    finally:
13
        self.prepare_inputs_event.record()

CPU 侧的 input_ids_cpu 这类张量在两步之间是复用的，异步上传是 non_blocking=True 的，所以第 t+2 拍准备输入前必须确保第 t+1 拍已经用完这些主机缓冲——prepare_inputs_event 就是这把锁。注释里说得直白：「prior step has finished with reused CPU tensors」。

这一步「只改位置、不读值」的做法，与 V1 从一开始追求的方向一致：调度输出是一个 {request_id: num_tokens} 的字典，下一拍的输入尽量复用上一拍已经落盘的持久化 buffer。异步调度把这个思路再推进一层——连「上一拍最后一个 token」也只在 GPU 上原地搬移，CPU 全程不读它的值。

图 3：V1 的持久化批处理概念图——请求进出时只做局部压缩，不重建整套输入张量。异步调度的 prev_positions 到 prev_req_id_to_index 这套映射，正是建在这套「位置稳定」的持久化 batch 上的。来源：vLLM V1 发布博客。

1
# Update output token ids with tokens sampled in last step
2
# if async scheduling and required by current sampling params.
3
self.input_batch.update_async_output_token_ids()

1
def update_async_output_token_ids(self) -> None:
2
    """In async scheduling case, update output_token_ids in sampling metadata
3
    from prior steps sampled token ids once they've finished copying to CPU.
4
    This is called right before they are needed by the logits processors.
5
    """
6
    output_token_ids = self.sampling_metadata.output_token_ids
7
    if self.sampled_token_ids_cpu is None or not output_token_ids:
8
        # Output token ids not needed or not async scheduling.
9
        return
10
    assert self.prev_req_id_to_index is not None
11
    ...
12
    if sampled_token_ids is None:
13
        assert self.async_copy_ready_event is not None
14
        self.async_copy_ready_event.synchronize()
15
        sampled_token_ids = self.sampled_token_ids_cpu.tolist()

1
first_placeholder = len(req_output_token_ids)
2
while (
3
    first_placeholder > 0
4
    and req_output_token_ids[first_placeholder - 1] == -1
5
):
6
    first_placeholder -= 1
7
num_placeholders = len(req_output_token_ids) - first_placeholder
8
num_to_replace = min(num_sampled_ids, num_placeholders)
9
del new_ids[num_to_replace:]
10
req_output_token_ids[first_placeholder:] = new_ids

1
if self.sampling_metadata.output_token_ids:
2
    self.sampled_token_ids_cpu = sampled_token_ids_cpu
3
    self.async_copy_ready_event = async_copy_ready_event
4
else:
5
    self.sampled_token_ids_cpu = None
6
    self.async_copy_ready_event = None

1
if self.scheduler.has_requests():
2
    scheduler_output = self.scheduler.schedule(self._should_throttle_prefills())
3
    with self.log_error_detail(scheduler_output):
4
        exec_future = self.model_executor.execute_model(
5
            scheduler_output, non_block=True
6
        )
7
    ...
8
    if not scheduler_output.pending_structured_output_tokens:
9
        # We aren't waiting for any tokens, get any grammar output
10
        # and sample immediately.
11
        grammar_output = self.scheduler.get_grammar_bitmask(scheduler_output)
12
        future = self.model_executor.sample_tokens(grammar_output, non_block=True)
13
    else:
14
        # We need to defer sampling until we have processed the model output
15
        # from the prior step.
16
        deferred_scheduler_output = scheduler_output
17

18
    if not deferred_scheduler_output:
19
        # Add this step's future to the queue.
20
        batch_queue.appendleft((future, scheduler_output, exec_future))
21
        if len(batch_queue) < self.batch_queue_size and (
22
            model_executed or self.scheduler.has_requests()
23
        ):
24
            # Don't block on next worker response unless the queue is full
25
            # or there are no more requests to schedule.
26
            return None, model_executed
27
...
28
# Block until the next result is available.
29
future, scheduler_output, exec_model_fut = batch_queue.pop()
30
model_output = future.result()
31
self._process_aborts_queue()
32
engine_core_outputs = self.scheduler.update_from_output(
33
    scheduler_output, model_output
34
)

1
if deferred_scheduler_output:
2
    ...
3
    draft_token_ids = self.model_executor.take_draft_token_ids()
4
    ...
5
    grammar_output = self.scheduler.get_grammar_bitmask(deferred_scheduler_output)
6
    future = self.model_executor.sample_tokens(grammar_output, non_block=True)
7
    batch_queue.appendleft((future, deferred_scheduler_output, exec_future))

投机解码与异步调度：占位草稿与真实草稿的两段写入

投机解码在异步模式下需要额外处理，因为存在两类 token 都需要「迟一拍」：

1. 采样出的真实 token：这是上一步的 SamplerOutput，异步调度让它留在 GPU；
2. 草稿模型/草稿头生成的下一个 token：这是上一步末尾跑草稿产生的候选位置，下一步验证要用到。

同步模式下这两类都走 CPU：采样 token 拷回来，草稿 token 拷回来，回填到 spec_token_ids 里。异步模式下它们必须分开处理。

先看草稿侧。Scheduler.post_step 在同步模式下从执行器取草稿 token，异步模式下这一步被跳过：

1
def post_step(self, model_executed: bool) -> None:
2
    # When using async scheduling we can't get draft token ids in advance,
3
    # so we update draft token ids in the worker process and don't
4
    # need to update draft token ids here.
5
    if self.check_for_draft_tokens and not self.async_scheduling and model_executed:
6
        draft_token_ids = self.model_executor.take_draft_token_ids()
7
        if draft_token_ids is not None:
8
            self.scheduler.update_draft_token_ids(draft_token_ids)

草稿的「真实值」被推给了 worker：在 GPUModelRunner 里，下一步输入准备前，执行器自己从 _draft_token_req_ids 对应的草稿输出里取出 token，写进 InputBatch：

1
if self.use_async_scheduling and self._draft_token_req_ids is not None:
2
    draft_token_ids_cpu, _ = self._get_draft_token_ids_cpu()
3
    self.input_batch.update_async_spec_token_ids(draft_token_ids_cpu)

这一步只在「需要历史输出」时才做：注释写着「only when output_token_ids is needed (penalties or bad_words are in use)」。不需要时，草稿 token 根本不需要让 CPU 看到值——它们只在下一次前向里作为「待验证位置」被 GPU 读到。

另外，占位符还是调度器与 worker 之间的一份显式契约。调度器在构造 CachedRequestData 发给 worker 时，输出长度也带着占位符一起发过去：

1
num_output_tokens.append(
2
    req.num_output_tokens + req.num_output_placeholders
3
)

worker 因此被要求把「已确认的输出 token + 预支的占位 token」当作这条请求的当前输出长度来准备输入，而不是等下一步从调度器再拿一次真实值。

模型执行器内部同样保留着占位符语义——_bookkeeping_sync 在异步模式下写进 req_state.output_token_ids 的是 -1：

1
if self.use_async_scheduling:
2
    sampled_ids = [-1] if req_idx not in invalid_req_indices_set else None
3
else:
4
    sampled_ids = valid_sampled_token_ids[req_idx]

这一组 [-1] 写进 req_state.output_token_ids 后，就成了下一步采样时 sampling_metadata.output_token_ids 末尾的占位位——也就是 update_async_output_token_ids 要回填的位置。调度器与输出处理器之间走的仍然是真实 token（由 get_output() 在响应线程里回填），-1 只存在于模型执行器内部、并跨过一拍。

草稿的「验证」与「拒绝」也在调度器里用占位符对齐。update_from_output 对每个调度过草稿的请求算出一笔拒绝账，并在异步模式下从两个计数里同时扣掉：

1
num_draft_tokens = len(scheduled_spec_token_ids)
2
num_sampled = self.num_sampled_tokens_per_step
3
num_accepted = max(len(generated_token_ids) - num_sampled, 0)
4
num_rejected = num_draft_tokens - num_accepted
5
# Rejections roll back num_computed_tokens (and, under async
6
# scheduling, num_output_placeholders, which covers the spec
7
# tokens). A stale rejection count predates the preemption
8
# rollback and must not apply.
9
if not output_is_stale:
10
    if request.num_computed_tokens > 0:
11
        request.num_computed_tokens -= num_rejected
12
    if request.num_output_placeholders > 0:
13
        request.num_output_placeholders -= num_rejected

这里的 num_sampled 是每步必采的 token 数——在投机验证里它对应「接受若干个草稿之后额外补采的那个 bonus token」。len(generated_token_ids) - num_sampled 即被验证接受的草稿数。被拒的草稿 KV 白算了，所以要回滚 num_computed_tokens；但它们在调度时已经被计入了占位符，所以占位符也要同额回滚。output_is_stale 的检查同样必要——如果这一步的输出本来就属于更早一次被抢占回滚过的旧账，再扣一次就会把占位符扣成负数。

补充一个容易被忽略的点：KV 块的回收在异步模式下默认走延迟队列（defer_block_free），被拒草稿占过的块不一定当拍归还。这与 请求的取消与抢占：五个推理框架的五条回收路径 里讨论的块回收围栏是同一个机制，只是异步调度把它变成了常态路径。

1
def update_async_spec_token_ids(self, draft_token_ids: list[list[int]]) -> None:
2
    """In async scheduling case, update spec_token_ids in sampling metadata with
3
    real draft token ids from prior step. This is called right before they are
4
    needed by the rejection sampler for penalty/bad_words computation.
5
    """
6
    if not draft_token_ids or not self.prev_req_id_to_index:
7
        return
8
    if (spec_token_ids := self.sampling_metadata.spec_token_ids) is not None:
9
        for req_id, spec_ids in zip(self.req_ids, spec_token_ids):
10
            if spec_ids:
11
                prev_index = self.prev_req_id_to_index.get(req_id)
12
                if prev_index is not None:
13
                    draft_ids = draft_token_ids[prev_index]
14
                    if draft_ids:
15
                        del draft_ids[len(spec_ids) :]
16
                        spec_ids.clear()
17
                        spec_ids.extend(draft_ids)

结构化输出：语法机不能在占位符上工作

结构化输出（JSON Schema / 正则 / EBNF）的限制采样依赖两件事：一个用「已生成 token 前缀」驱动的 FSM 状态，以及每步根据 logits 计算的 bitmask。异步调度下，「已生成前缀」的末尾是 -1 占位符，语法机没法在它上面推进。

调度器在 _update_after_schedule 里把这类请求标记出来：

1
scheduler_output.pending_structured_output_tokens |= (
2
    request.use_structured_output and request.num_output_placeholders > 0
3
)

pending_structured_output_tokens 是 SchedulerOutput 上的一个布尔字段。它的作用相当于告诉 EngineCore：「这一步里至少有一个请求的语法状态依赖上一步的采样结果」。EngineCore.step_with_batch_queue 据此把采样推迟一拍：只要该标记为真，这一拍只提交 execute_model（前向与 logits 计算不受影响），采样 future 留到下一拍、等上一拍的 token 真正回来后，用真实前缀算语法 bitmask 再提交：

1
if not scheduler_output.pending_structured_output_tokens:
2
    grammar_output = self.scheduler.get_grammar_bitmask(scheduler_output)
3
    future = self.model_executor.sample_tokens(grammar_output, non_block=True)
4
else:
5
    # We need to defer sampling until we have processed the model output
6
    # from the prior step.
7
    deferred_scheduler_output = scheduler_output

在延迟采样那一拍里，模型执行器的状态（logits、隐状态）被保留在 worker 侧，等 deferred_scheduler_output 被处理时再调 sample_tokens。换句话说，语法约束的正确性没有被牺牲，代价是「语法请求」在双缓冲里多占半个槽位、采样被后移一拍。注意这个标记是整批的，不是逐请求的：只要 batch 里有一个结构化输出请求带着未结算占位符，整批的采样都会被推迟。所以在结构化输出-heavy 的负载里，异步调度可能让所有请求的 TPOT 都多一拍——这是引擎层面的 amortization 取舍，而非逐请求优化。

站内结构化输出那篇讲过「语法掩码的增量计算」——异步调度把「更新语法状态」与「使用掩码采样」这两个原本同步的动作拆到了两拍，工程实现必须保证 worker 侧保留的 logits 与延迟后的 bitmask 是对应同一批请求的。

流水并行与 V2 模型执行器：错拍解码

V1 模型执行器的异步调度不支持流水并行——这是 VllmConfig 初始化里那条硬性不兼容判断，显式开启会抛错、默认开启会降级。原因在于 V1 的请求状态是「调度器一份、worker 一份」各自记账的：PP 下采样只发生在最后一个 rank，而第一步的输入需要每个 rank 都预支同一个占位符，V1 没有为这条链路设计统一的采样广播与回填，所以直接判定为不兼容。

V2 模型执行器上这条路走通了，但要额外记一笔账。采样只发生在最后一个 rank，第一步的输入在各个 rank 之间有序传播；同步模式下最后一个 rank 采出 token 后回传第一个 rank，第一个 rank 才能在下一拍准备输入。异步调度把部分回传留在 GPU 上：sample_tokens 里若 use_async_scheduling、PP 世界大小大于 1、且当前是最后一个 rank，就调用 _pp_broadcast_prev_sampled_token_ids(sampled_token_ids) 把采样张量广播给其他 rank；其他 rank 在下一步的 _bookkeeping_sync 路径被跳过（execute_model_state is None）时，用 _pp_receive_prev_sampled_token_ids_to_input_batch() 收下：

1
if self.execute_model_state is None:
2
    kv_connector_output = self.kv_connector_output
3
    self.kv_connector_output = None
4
    # receive sampled token ids from the last PP rank.
5
    if self.use_async_scheduling and not get_pp_group().is_last_rank:
6
        self._pp_receive_prev_sampled_token_ids_to_input_batch()
7
    return ModelRunnerOutput.with_kv_conn_output_only(kv_connector_output)

decode_stagger 就是为这个节拍对齐而生：当 V2 模型执行器启用时，next_decode_eligible_step = current_step + decode_stagger，一条请求至少隔 decode_stagger 拍才能再次解码。默认值是 pp_size，也就是说每一条请求在 pp_size 拍之内只能出现一次 decode。注释里说这个值与 worker 侧采样 token 广播的 slot ring 节拍匹配——把「错拍」写进调度器的判断，而不是靠同步去保证。若平台上禁用 PP 解码微批（XPU 的 VLLM_XPU_PP_MICROBATCH 未开），decode_stagger 退化成 1，也就是每一拍都可以解码。

调度循环里对应的判断：

1
if self.current_step < request.next_decode_eligible_step:
2
    # V2+PP+async: enforce `pp_size` steps between same-req decodes
3
    # to match worker-side sampled-tokens broadcast slot ring cadence.
4
    req_index += 1
5
    continue

一张图看懂：双缓冲里两拍之间发生了什么

把上面的机制拼成一条时间线。同步模式的 step 一次只处理一拍，调度、执行、取结果、对账严格串行：

1
同步（单缓冲）:
2
时间 →
3
CPU   schedule(S_t) ──────────────┐ 等待 future.result()  ┌ update(S_t) ─ schedule(S_{t+1})
4
GPU               execute(S_t) → sample → D2H ───────────┘
5
                  ↑ GPU 跑完后，CPU 才能继续

异步模式（step_with_batch_queue，队列容量 2）的稳态是「排队—取结果」交替：

1
异步（双缓冲）:
2
时间 →
3
CPU   schedule(S_t) → execute(S_t) 入队 → 立刻返回
4
                       schedule(S_{t+1}) → execute(S_{t+1}) 入队 → 队列满
5
                                             pop(S_t) → future.result() 等待
6
                                             update(S_t) → schedule(S_{t+2}) ...
7
GPU            execute(S_t) → sample      execute(S_{t+1}) → sample

对应到账本上：

1
S_t 调度时:
2
  num_output_placeholders += 1 (AR 采样) + k (草稿)
3
  spec_token_ids = [-1] * k
4

5
S_t 结果回来时 (update_from_output):
6
  真实 token 写回 request
7
  num_output_placeholders -= 返回的 token 数
8
  被拒草稿: num_computed_tokens -= num_rejected
9
            num_output_placeholders -= num_rejected

异步模式换来的是：同一时刻，CPU 在准备第 t+2 拍的计划，而 GPU 还在跑第 t+1 拍；采样结果的 D2H 拷贝被启动在第 t 拍，消费在第 t+1 拍的对账里。代价是调度器必须一直带着至少一批「在途账」运行，且模型执行器内部任何需要精确输出 token 的子系统（惩罚、语法、logprobs）都必须知道 -1 是占位符。

收益与代价：什么时候该开，什么时候不该开

vLLM 引擎参数文档对 --async-scheduling 的描述只有一句话：「helps to avoid gaps in GPU utilization, leading to better latency and throughput」。这句话里的关键词是「避免 GPU 空转间隙」——它要省的不是 GPU 的计算量，而是 GPU 等 CPU 的那段时间。在 decode 阶段，GPU 单步的计算本身就很短；真正决定每 token 延迟的往往是 CPU 侧的调度、记账与同步往返。异步调度把这些 CPU 工作从「GPU 执行期间」挪到「GPU 执行期间照常进行」，让 GPU 的利用率曲线更平。

但「默认开」不等于「任何场景都更好」。结合源码里的几条硬性判断，可以归纳出异步调度的适用边界：

适合的场景：纯 AR 解码为主、无结构化输出或结构化输出比例很低、无投机解码或使用 V1 兼容的 EAGLE/N-gram 系投机、单机多卡（无流水并行）、CPU 主频较低但核心充足的部署（调度开销占比高时收益明显）。vLLM 在 VllmConfig 里对池化模型（embedding/rerank）专门关闭默认异步调度，注释写明「negatively impacts performance of pooling models」——池化请求一次前向就完成，没有「连续 decode 步」可重叠，反而被双缓冲拖累。

需要显式关闭的场景：ROCm + DeepEP 高吞吐 DBO（与异步调度同时开启会破坏 DP+EP 的生成精度，代码里直接禁止）；V1 模型执行器 + 流水并行；V1 + Medusa 式多头草稿或其他未经草稿头对齐的投机方法；使用不支持异步调度的分布式执行器后端。

组合时需要实测的场景：结构化输出（整批采样被推迟一拍）、惩罚/bad_words/logprobs（每步都要 CPU 侧真实 token 副本）、高比例短请求（调度工作量本身很小，收益低而状态复杂度不变）。

它的代价可以归纳成四条：

1. 增加状态复杂度。num_output_placeholders、prev_sampled_token_ids 与 -1 哨兵在三层代码里（调度器、模型执行器、引擎核心的批队列）都要保持一致；任何一处漏减/漏替换，结果都是「输出错位」或「请求卡住不退」。新写调度相关代码时，最常犯的错就是在 update_from_output 的某个早退分支里忘了扣占位符。
2. 对依赖精确前缀的子系统加了一拍延迟。结构化输出整批推迟采样，惩罚与 bad_words 需要跨流同步等 CPU 副本，停止条件的判定要在「最坏情况」下保守推算。
3. 投机解码的收益被削薄。异步调度的成本大头在「CPU 与 GPU 的同步往返」，投机解码用一次前向产出多 token，同样降低了每一步中同步的占比——两者在「每 token CPU 开销」这个维度上部分替代，组合时的收益不一定线性叠加，需要实测。
4. 与 DP/EP、故障恢复的耦合更紧密。异步调度默认会把 DP 同步从 NCCL 切到 Gloo（disable_nccl_for_dp_synchronization），因为异步模式下 NCCL 的 DP 同步会与调度流水冲突；开启故障恢复时这个设置又必须关掉。这类组合规则散落在配置校验里，部署时值得逐条核对。

在 AsyncScheduler 之外，vLLM 这套机制的抽象其实很值得借鉴：「以最坏情况预支账本、在结果回来时对账」 是任何想把 CPU 从 GPU 关键路径上摘掉的系统都可以用的模式。DeepInfer 的 ZOS 用「延后一拍」做同样的事：调度结果不是预支给执行器，而是延后一层落到自己的状态机上，并在延后期间用估计器修正 seqlens——细节可以参考 DeepInfer 源码完全拆解（二）：异步调度与零开销调度 ZOS。SGLang 的 Overlap Scheduler 用的是「两条流」的思路，让 CPU 调度和 GPU 执行在两套独立流上各跑各的、只在关键同步点汇合（详见 SGLang 源码完全拆解（六）：Overlap Scheduler）。三者在同一个问题上走的是三条路：vLLM 预支账本、DeepInfer 延后一拍、SGLang 双流并行。哪条路在什么硬件与场景下最合适，取决于系统的同步粒度与 batch 形态。

小结

异步调度是 vLLM V1 里「调度器与执行器解耦」的极端形态。它把 EngineCore 的两批在途、调度器的 num_output_placeholders 预支、模型执行器的 GPU 常驻采样、输入准备的 scatter_ 回填与输出对象的异步流拷贝串在一起，把 CPU 从「等 GPU 采样」的关键路径上移了出去。代价是整套引擎必须把「在途结果」当作一等状态来管理：调度器按最坏情况预支算账、执行器按真实结果对账、输入准备按占位符回填、语法机推迟采样、投机解码分两段写入草稿。

读懂它之后，再回看 step_with_batch_queue 里的「一批进队立刻返回、队满才取结果」，它的含义已经与同步模式完全不同：那不再是一次等待，而是一次「把这份计划交给 GPU、我自己去准备下一份」的交接；而 num_output_placeholders 就是这次交接的欠条。

下一篇将讨论 vLLM 的可视化调度诊断：当一条请求的 TTFT 异常升高、或者 update_from_output 里的对账数字与客户端看到的 token 数对不上时，如何从 EngineCore 的指标、batch_queue 的长度与 num_output_placeholders 的演变里定位问题；以及异步调度开启后，Metrics 中哪些时间戳的语义发生了变化。这条诊断链路与本文的账本机制直接衔接：不理解「两批在途」，就看不懂为什么一条请求的 IterationDetails 会跨两个 step。

延伸阅读：

• vLLM 源码完全拆解（二）：从 add_request 到 EngineCore.step，一次请求如何穿过 V1 引擎——本文反复引用的 EngineCore.step 在这里被逐行拆过；
• vLLM 源码完全拆解（三）：Scheduler.schedule 如何分配 Token 预算与 KV Block——占位符是如何进入 num_new_tokens 预算的，在这篇里能看到完整上下文；
• vLLM 源码完全拆解（五）：GPUModelRunner 如何把动态请求拼成一次 GPU 前向——InputBatch 的字段与 condense/swap_states 在这里；
• vLLM 源码完全拆解（九）：Sampler——从 Logits 到 Token 的采样管线与输出回传——SamplerOutput 的结构与 logprobs 的计算；
• vLLM 源码完全拆解（十）：投机解码——草稿模型、批量验证与 token 接受流水线——num_sampled_tokens_per_step 与接受/拒绝账的原始出处；
• vLLM 源码完全拆解（十五）：停止条件与增量解码——stop string、detokenizer 流水线与输出回传——真实 token 回到客户端的那一段；
• DeepInfer 源码完全拆解（二）：异步调度与零开销调度 ZOS——把 CPU 后处理移出关键路径——同一问题的另一种解法；
• SGLang 源码完全拆解（六）：Overlap Scheduler 如何重叠 CPU 调度与 GPU 执行——双流重叠的第三种解法。

参考资料

1. vLLM V1: A Major Upgrade to vLLM’s Core Architecture
2. vLLM 设计文档
3. vLLM 引擎参数（Engine Arguments）
4. vLLM 源码：vllm/v1/core/sched/async_scheduler.py
5. vLLM 源码：vllm/v1/core/sched/scheduler.py
6. vLLM 源码：vllm/v1/worker/gpu_input_batch.py
7. vLLM 源码：vllm/v1/worker/gpu_model_runner.py
8. vLLM 源码：vllm/v1/engine/core.py
9. vLLM 源码：vllm/v1/outputs.py
10. vLLM 源码：vllm/v1/executor/multiproc_executor.py
11. vLLM 源码：vllm/config/scheduler.py
12. vLLM 源码：vllm/config/vllm.py