vLLM 源码完全拆解（十七）：KV Connector——P/D 分离中 KV 缓存如何跨实例传输与接管调度

传输这件事，三套不同的问题

把 KV 从 A 搬到 B，至少涉及三类互相纠缠的问题。

第一类是物理传输：用什么链路（NVLink、IB/RDMA、TCP、共享文件系统）、用什么协议（NCCL、NIXL 的 RDMA 读写、ZMQ 控制通道）、怎么注册显存地址、怎么保证搬完。

第二类是语义拼接：源端 KV 是按源端分页池的 block 编号摆放的，目标端有自己的 block 表；MLA 模型只有一份 latent KV，GQA 模型是 K/V 两份；FlashInfer 后端把 [2, num_blocks, ...] 和 [num_blocks, 2, ...] 的维度顺序交换过；源端 TP=4、目标端 TP=8 时，每个目标 rank 要读的部分不一样。搬运必须懂「分页布局 + 并行拓扑」。

第三类是调度语义：decode 端调度器要知道「哪些 token 不必本地算、正在从远端拉、拉完没有」，并且要处理拉取期间的请求状态、失败时的降级、请求结束时远端块何时可以释放。

如果让每个传输后端自己把三类问题都做一遍，结果就是每个后端各写一套调度钩子、各自改 KV 管理器，互相不兼容。vLLM 的选择是把第三类问题收进统一的接口，第二类问题收进连接器实现内部的「描述符/元数据」逻辑，第一类问题交给具体传输库。

两个进程、两个角色

V1 引擎的进程结构决定了这套接口的形状。回顾 vLLM 源码完全拆解（一）：整体架构与模块全景 中的进程拓扑：调度器运行在 EngineCore 进程里，模型执行发生在 Worker 进程里（TP>1 时每个 rank 一个 worker，或按 executor 组织）。

KV Connector 因此被显式拆成两个角色：

• KVConnectorRole.SCHEDULER：与调度器同进程，只做「查询与记账」——判断某请求能从外部拿多少 token、记录哪些请求要拉 KV、在每步结束时把「这一步要搬什么」整理成一份元数据；
• KVConnectorRole.WORKER：与模型执行同进程，只做「执行搬运」——接收元数据、发起/等待异步传输、回答「哪些请求搬完了」。

两侧之间只有一种通信方式：元数据对象。它在每步调度结束时由调度器侧构造，随 SchedulerOutput.kv_connector_metadata 一路下发到 worker，worker 在执行前绑定它：

1
def maybe_setup_kv_connector(scheduler_output: "SchedulerOutput"):
2
    if has_kv_transfer_group():
3
        kv_connector = get_kv_transfer_group()
4
        assert isinstance(kv_connector, KVConnectorBase_V1)
5
        assert scheduler_output.kv_connector_metadata is not None
6
        kv_connector.bind_connector_metadata(
7
            scheduler_output.kv_connector_metadata)
8
        # Background KV cache transfers happen here.
9
        kv_connector.start_load_kv(get_forward_context())

这段代码来自 vllm/v1/worker/gpu_model_runner.py。注意注释里的 Background KV cache transfers：拉取是「后台」的，它可以在这一步的模型执行期间并发进行；而且元数据里描述的请求，不一定是本步正在计算的那些请求——异步加载的请求此时还在排队。

工厂的注释把这种分工说得更直接：

1
# NOTE(Kuntai): v1 connector is explicitly separated into two roles.
2
# Scheduler connector:
3
# - Co-locate with scheduler process
4
# - Should only be used inside the Scheduler class
5
# Worker connector:
6
# - Co-locate with worker process
7
# - Should only be used inside the forward context & attention layer
8
# We build separately to enforce strict separation
9
return connector_cls(config, role)

「严格分离」是有意为之。现实中很多 P/D 分离的 bug 都来自两侧对同一件事的理解不一致：调度器以为「已经分配好了」，worker 以为「还没到」，结果读到未初始化的 KV。角色拆分 + 元数据契约，把这类不一致压缩成一个可以审计的接口。

调度器侧：五个方法

第一个方法的签名值得逐字看：

1
@abstractmethod
2
def get_num_new_matched_tokens(
3
    self,
4
    request: "Request",
5
    num_computed_tokens: int,
6
) -> tuple[int, bool]:
7
    """...
8
    Returns:
9
        - The number of tokens that can be loaded from the
10
          external KV cache beyond what is already computed.
11
        - `True` if external KV cache tokens will be loaded
12
          asynchronously (between scheduler steps). Must be
13
          'False' if the first element is 0.
14
    """

两个返回值共同决定调度器怎么处理这个请求。num_computed_tokens 参数名虽然叫「已计算」，实际传入的是调度器刚刚从本地前缀缓存里 get_computed_blocks 拿到的命中长度：

1
# Get locally-cached tokens.
2
new_computed_blocks, num_new_local_computed_tokens = \
3
    self.kv_cache_manager.get_computed_blocks(request)
4

5
# Get externally-cached tokens if using a KVConnector.
6
if self.connector is not None:
7
    num_external_computed_tokens, load_kv_async = (
8
        self.connector.get_num_new_matched_tokens(
9
            request, num_new_local_computed_tokens))
10

11
# Total computed tokens (local + external).
12
num_computed_tokens = (num_new_local_computed_tokens +
13
                       num_external_computed_tokens)

这是 v1/core/sched/scheduler.py 中 schedule() 里处理新请求的片段。外部命中是叠加在本地命中之上的：本地前缀缓存先尽量命中，剩下的部分再由连接器声明「远端有」。这避免了两个来源重复计算同一段 token。

异步加载如何改变调度行为

load_kv_async=True 时，调度器不会给这个请求派新的计算任务，只给它分配 KV 块：

1
# KVTransfer: loading remote KV, do not allocate for new work.
2
if load_kv_async:
3
    assert num_external_computed_tokens > 0
4
    num_new_tokens = 0

随后分配阶段会调用 allocate_slots 时传入 num_new_tokens + num_external_computed_tokens，并用 delay_cache_blocks=load_kv_async 告诉 KV 管理器：这些块的内容还没到位，先别当成可缓存前缀。分配完成后：

1
if load_kv_async:
2
    # If loading async, allocate memory and put request
3
    # into the WAITING_FOR_REMOTE_KV state.
4
    skipped_waiting_requests.prepend_request(request)
5
    request.status = RequestStatus.WAITING_FOR_REMOTE_KVS
6
    continue

请求被放回等待队列头部（保持顺序），状态切到 WAITING_FOR_REMOTE_KVS，然后 continue 去处理下一个请求。它「占着块，但不占 token 预算」。

worker 侧：六个方法

wait_for_layer_load 与 save_kv_layer 是按层的，这是为了逐层流水：prefill 端第 0 层算完就开始传第 0 层，第 1 层在算的时候，第 0 层的传输可能已经完成。Mooncake 论文里明确写了这一点：prefill 实例「逐层且与 prefill 计算并行地加载/存储 KVCache，以缓解传输开销」。

注意力层里的两个钩子

注意力层里调用这两个钩子的位置很关键。vllm/attention/layer.py 中，unified_attention 在调用后端实现之前插入等待：

1
def wait_for_kv_layer_from_connector(layer_name: str):
2
    ...
3
    connector = get_kv_transfer_group()
4
    forward_context: ForwardContext = get_forward_context()
5
    attn_metadata = forward_context.attn_metadata
6
    if attn_metadata is None:
7
        return
8
    assert isinstance(attn_metadata, dict)
9
    connector.wait_for_layer_load(layer_name)

在执行完后端计算之后插入保存：

1
def unified_attention(...):
2
    ...
3
    output = self.impl.forward(self, query, key, value, kv_cache,
4
                               attn_metadata)
5
    maybe_save_kv_layer_to_connector(layer_name, kv_cache)
6
    return output

attn_metadata is None 就返回，是为了兼容前向上下文缺失的场景（例如预热、纯解码无注意力元数据的情形）；assert isinstance(attn_metadata, dict) 则说明 V1 采用按层名索引的元数据字典——这也是连接器能实现「逐层等待」的前提。

同步与异步：三档加载模式

从接口语义能看出三种加载模式：

1. 同步、立即：在 start_load_kv 里阻塞地搬完，后续各层直接可用。实现最简单，但完全堵住 GPU。
2. 异步、跨步：load_kv_async=True 时，请求进入 WAITING_FOR_REMOTE_KVS，调度器这一步不计算它。传输在后台进行，通常在若干步之后完成，get_finished 报告后请求回到等待队列。
3. 异步、层间流水：start_load_kv 只发起传输，wait_for_layer_load 在每层前等待。需要传输库支持「按描述符批量的非阻塞读写 + 完成查询」。

不同连接器实现覆盖的模式不同，这决定了它们的适用场景。NIXL 属于第 2、3 类；P2P NCCL 属于第 1 类的变体（get_num_new_matched_tokens 返回 False，即同步加载）。

完成与释放：get_finished 的语义

1
@abstractmethod
2
def get_finished(
3
    self, finished_req_ids: set[str]
4
) -> tuple[Optional[set[str]], Optional[set[str]]]:
5
    """
6
    ... Returns:
7
        ids of requests that have finished asynchronous transfer
8
        (requests that previously returned True from request_finished()),
9
        tuple of (sending/saving ids, recving/loading ids).
10
    """

两组 ID 的用途在调度器里被明确分开：

1
def _update_from_kv_xfer_finished(self, model_runner_output):
2
    for req_id in (model_runner_output.finished_recving or ()):
3
        self.finished_recving_kv_req_ids.add(req_id)
4
    for req_id in (model_runner_output.finished_sending or ()):
5
        self._free_blocks(self.requests[req_id])

• 接收完成：记进 finished_recving_kv_req_ids，等待中的请求会在下次调度时检查并转回 WAITING（见下文「远端 KV 到位」小节）；
• 发送完成：这个请求的块此时才能真正回收。之前 request_finished 返回 delay_free_blocks=True，块被「挂起」不释放，就是为了防止传输读到已被覆盖的分页。

这套「延迟释放」是连接器正确性的核心约定之一：KV 池是复用的，谁要读它，谁就得让它多活一会儿。它与 vLLM 源码完全拆解（四）：KVCacheManager 如何分配、复用与回收物理 Block 里的引用计数与 free queue 是同一套内存生命周期思想，区别在于「使用者」变成了一个异步的远端读取者。

远端 KV 到位之后：为什么要减一

1
def _update_waiting_for_remote_kv(self, request: Request) -> bool:
2
    assert self.connector is not None
3
    if request.request_id not in self.finished_recving_kv_req_ids:
4
        return False
5

6
    # Now that the blocks are ready, actually cache them.
7
    (block_ids, ) = self.kv_cache_manager.get_block_ids(request.request_id)
8
    num_computed_tokens = len(block_ids) * self.block_size
9
    # Handle the case where num request tokens less then one block.
10
    num_computed_tokens = min(num_computed_tokens, request.num_tokens)
11
    if num_computed_tokens == request.num_tokens:
12
        num_computed_tokens -= 1
13
    # This will cache the blocks iff caching is enabled.
14
    self.kv_cache_manager.cache_blocks(request, num_computed_tokens)
15

16
    request.num_computed_tokens = num_computed_tokens
17
    self.finished_recving_kv_req_ids.remove(request.request_id)
18
    return True

这段 _update_waiting_for_remote_kv 有几处设计值得解释。

为什么按块数换算 token 数，而不是直接用远端声明的长度？ 因为拉取是按块粒度做的：目标是「把 n_ext 个 token 以整块对齐的数量拉进来」，实际拉进来的 token 数可能比 n_ext 略多（也可能略少——见下文前缀命中时 NIXL 的块选择）。最终「计算到哪」必须由实际拥有的完整块决定，否则最后一个 token 的 KV 可能是不完整的。这与前缀缓存只发布完整块是同一条规则。

为什么 num_computed_tokens == request.num_tokens 时要减一？ 如果整段 prompt 的 KV 都已在池里，还需要一个「query 位置」去生成第一个输出 token：模型的最后一步前向必须以最后一个 prompt token 为输入，拿到它的 logits 才能解码。把最后一个位置标为「未计算」，调度器就会为它安排一次计算——输出的是下一个 token 的 logits，而不是重复计算前缀。

cache_blocks 做什么？ 它把刚刚拉进来的块「正式发布」为可被后续请求前缀复用的块。注意 NIXL 拉取之前给这些块传了 delay_cache_blocks=True，所以拉取期间它们不会被其他请求命中——防止别人读到一半的数据。

request_finished：谁该等下再释放

NIXL 调度器的实现：

1
if (not params.get("do_remote_decode")
2
        or request.status != RequestStatus.FINISHED_LENGTH_CAPPED):
3
    return False, None
4

5
# Get computed blocks.
6
all_full = request.num_computed_tokens % self.block_size == 0
7
computed_block_ids = block_ids if all_full else block_ids[:-1]
8

9
# If prompt < block_size, no xfer so free blocks immediately.
10
delay_free_blocks = len(computed_block_ids) > 0
11

12
return delay_free_blocks, dict(
13
    do_remote_prefill=True,
14
    do_remote_decode=False,
15
    remote_block_ids=computed_block_ids,
16
    remote_engine_id=self.engine_id,
17
    remote_host=self.side_channel_host,
18
    remote_port=self.side_channel_port,
19
    tp_size=self.vllm_config.parallel_config.tensor_parallel_size)

几个细节：

• 只处理「首 token 即长度截断」这一种结束。P 实例在 P/D 分离下只跑 prefill，然后很快以 FINISHED_LENGTH_CAPPED 结束（max_tokens=1 之类的设定），这是唯一需要交接的场景；如果用户在 P 实例上要求继续生成，那就没有「交给 D」这一步。
• 非整块时去掉末尾块。最后一个块可能只装了部分 token，它的内容不完整，对远端没有复用价值；而且远端也只能按完整块对齐读取。
• prompt 不足一块就不延迟释放：没有可传的内容，直接回收。

还有一个「未调度就结束」的防御：如果请求在 update_state_after_alloc 之前就被中止（do_remote_prefill 仍为 True），它会往 _reqs_need_recv 里塞一条空块记录，让 worker 侧给 P 实例发一个「我不需要了」的通知，避免 P 端块被永久占住：

1
if params.get("do_remote_prefill"):
2
    # ... the request must have been aborted before it was scheduled.
3
    # To avoid stranding the prefill blocks in the prefill instance,
4
    # we must add empty block_ids to _reqs_need_recv so that our
5
    # worker side will notify and free blocks in the prefill instance.
6
    self._reqs_need_recv[request.request_id] = (request, [])
7
    params["do_remote_prefill"] = False
8
    return False, None

注册：把分页 KV 变成一组可寻址的块

worker 初始化时调用 register_kv_caches。代码先判断 KV 布局：

1
use_mla = len(first_kv_cache.shape) == 3
2
assert use_mla == self.use_mla
3
if use_mla:
4
    # MLA case.
5
    self.num_blocks = first_kv_cache.shape[0]
6
    block_rank = 2  # [block_size, latent_dim]
7
    block_shape = first_kv_cache.shape[-block_rank:]
8
    block_size, kv_latent_dim = block_shape
9
    self.slot_size_bytes = kv_elem_size * kv_latent_dim
10
else:
11
    # [2 (k and v), num_blocks, ...]
12
    if self._use_flashinfer:
13
        # FlashInfer swaps 2<->num_blocks dimensions.
14
        self.num_blocks = first_kv_cache.shape[0]
15
        block_rank = 4  # [2, block_size, kv_heads, head_dim]
16
    else:
17
        self.num_blocks = first_kv_cache.shape[1]
18
        block_rank = 3  # [block_size, kv_heads, head_dim]

这揭示了一个容易忽略的事实：「一个 block 在显存里有多大」取决于后端。MLA 的每层 KV 只有一个张量、末尾两维是 [block_size, latent_dim]；GQA 的每层是两个张量（K 与 V），非 FlashInfer 后端第一维是 2；FlashInfer 把维度交换成了 [num_blocks, 2, ...]。连接器必须按后端实际的布局计算每个块的字节数，才能为每个块构造正确的描述符。

普通 GQA 模型（每层 K、V 各一份）的块字节数是

MLA 模型只有一个压缩后的 latent KV：

其中 $b_s$ 是分页粒度（每块 token 数），$h_{kv}$ 是 KV 头数，$d_h$ 是每头维度，$d_c$ 是 MLA 的 latent 维度，$s_{\text{elem}}$ 是单个元素的字节数（FP16 为 2、FP8 为 1）。以 V4.1-Flash 量级的 MLA 配置为例，$b_s=64$、$d_c$ 约 500 多、FP8 存储时一个块只有几十 KB；而同样块大小的普通 GQA 模型（$h_{kv}=8$、$d_h=128$、FP16）会到 256 KB 量级。同样叫「一块」，传输成本差一个数量级，这直接决定了 P/D 传输是「可以藏在计算背后」还是「成为 TTFT 的主要项」。

一次传输的下界可以写成

$\alpha$ 是握手与提交的固定开销，$N_{\text{blk}}$ 是本次要搬的块数；分母取链路带宽与显存带宽的较小值——因为源端从 HBM 读、目标端往 HBM 写，任何一侧都可能先饱和。这也解释了 Mooncake 为什么强调「逐层传输」：把 $N_{\text{blk}}$ 拆成每层一小批，$\alpha$ 会略增，但传输可以和该层之后的计算重叠，端到端关键路径上只剩下最后一层的传输。

注册被拆成两级：

1. 区域（region）注册：每个 cache 张量（MLA 每层 1 个、普通每层 2 个）作为一段 (base_addr, region_len, device_index)，交给 nixl_wrapper.get_reg_descs(caches_data, "VRAM") 并 register_memory。注释解释了为什么 K/V 要作为两个区域：这样 MLA（只有一份 KV）与 K/V 非连续的情形都能统一处理，代价是元数据更大（约 8KB 而非 5KB）。
2. 块级描述符（xfer descs）：对每个区域里的每个块编号构造 (addr + block_id * block_len, block_len, tp_rank)，用 get_xfer_descs 注册成传输描述符。每个描述符有一个全局编号，编号规则在 _get_block_descs_ids 里：

1
for reg_id in region_ids:
2
    for block_id in block_ids:
3
        descs_ids.append(reg_id * num_blocks + block_id)

于是「本地块 b 的第 l 层」对应编号 l * num_blocks + b（MLA）或 (2l + {0,1}) * num_blocks + b（普通 GQA 的 K/V）。传输时只需要给出两组编号列表——本地侧的一串和远端侧的一串——按位置配对搬运。

握手：一次交换 agent 元数据

两实例第一次需要传输时，D 端 worker 不知道 P 端的显存地址与 agent 名称。NIXL 用 ZMQ 做一次握手：

1
self._nixl_handshake_listener_t = threading.Thread(...)
2
self._handshake_initiation_executor = ThreadPoolExecutor(
3
    # NIXL is not guaranteed to be thread-safe, limit 1 worker.
4
    max_workers=1,
5
    thread_name_prefix="vllm-nixl-handshake-initiator",
6
)

初始化线程池只有一个 worker，注释写明原因：NIXL 不保证线程安全。握手监听端在 base_port + tp_rank 上起 ROUTER socket，把 NixlAgentMetadata（agent 名称、KV base 地址、块数、块大小等）msgspec 编码后返回给对方；发起端拿到后建立 _remote_agents[engine_id][rank] = agent_name。

两个设计要记下来：

• DP rank 有自己的端口段：side_channel_port = VLLM_NIXL_SIDE_CHANNEL_PORT + dp_rank * tp_size，同一 DP 组内每个 TP rank 用 base_port + tp_rank，避免端口冲突。
• 握手与传输解耦：start_load_kv 遇到未握手的 engine，会把请求挂到 _ready_requests 队列后立即返回，握手的 future 完成回调再把请求放回队列。这样「第一次连接」也不会阻塞整个 step。

拉取：前缀命中的块选择与 TP 异构

start_load_kv 依次处理元数据里的请求：

1
for req_id, meta in metadata.requests.items():
2
    ...
3
    if remote_engine_id not in self._remote_agents:
4
        with self._handshake_lock:
5
            if remote_engine_id not in self._remote_agents:
6
                self._background_nixl_handshake(req_id, remote_engine_id, meta)
7
                continue
8
    self._read_blocks_for_req(req_id, meta)
9
while not self._ready_requests.empty():
10
    self._read_blocks_for_req(*self._ready_requests.get_nowait())

真正构造传输的是 _read_blocks，其中有几处非平凡逻辑：

1
# Number of D TP workers that will read from dst P. Propagate tp_ratio
2
# on notification so that dst worker can wait before freeing blocks.
3
tp_ratio = self._tp_size[self.engine_id] // self._tp_size[dst_engine_id]
4
notif_id = f"{request_id}:{tp_ratio}".encode()
5

6
num_local_blocks = len(local_block_ids)
7
if num_local_blocks == 0:
8
    remote_rank = self.tp_rank // tp_ratio
9
    agent_name = self._remote_agents[dst_engine_id][remote_rank]
10
    self.nixl_wrapper.send_notif(agent_name, notif_msg=notif_id)
11
    return
12

13
num_remote_blocks = len(remote_block_ids)
14
assert num_local_blocks <= num_remote_blocks
15
if num_local_blocks < num_remote_blocks:
16
    remote_block_ids = remote_block_ids[-num_local_blocks:]

要点：

1. D 端本地前缀已经全命中时，local_block_ids 为空。此时不需要读任何东西，但仍要发通知——否则 P 端以为自己还有读者，块永不释放。通知里带上 tp_ratio，因为 P 端需要知道有多少个 D 的 TP worker 会来读它，全部读完才能释放。
2. 部分命中的对齐：D 端已有前 num_remote_blocks - num_local_blocks 个块的内容，只需要远端块的末尾 num_local_blocks 个。所以 remote_block_ids = remote_block_ids[-num_local_blocks:]。这是一条不易发现的细节——如果按前缀对齐读取，就会把已命中的块再读一遍，甚至读到位置错配的数据。
3. TP 异构映射：一个 D rank 对应 tp_ratio = D_tp // P_tp 个 P rank 中的某一个（remote_rank = self.tp_rank // tp_ratio），一个 D worker 只从一个 P worker 拉。这是「异构 TP」下 vLLM 采取的一种简化映射，后续版本中有更完整的分片读取逻辑。

用一个具体例子把「取末尾块」讲清楚。设块大小 $b_s=16$，prompt 有 8 个完整块（128 token），D 端本地前缀缓存已经命中前 2 个块（32 token）。调度器传入 num_computed_tokens=32，NIXL 的 get_num_new_matched_tokens 算出 count = round_down(128,16) - 32 = 96，即还需要 6 个块。update_state_after_alloc 拿到本地新分配的 6 个块编号，_read_blocks 里 local_block_ids 长度是 6，而远端 remote_block_ids 长度是 8。它执行 remote_block_ids[-6:]，也就是取远端第 2 到第 7 块，和本地 6 个块按顺序配对。

为什么不能取前 6 块？因为本地已经有第 0、1 块的内容，真正缺的是第 2 到第 7 块。如果按前缀取，会把第 0、1 块再读一遍，同时漏掉最后两块；更糟的是，目标端会把远端第 0 块写进本地「第 2 个新块」的位置，块表与内容整体错位，decode 时读到的是错位的历史。分页 KV 的传输是按块编号配对的，不是按字节流拼接的，这是连接器实现里最容易写错的一处。

传输描述符的编号也随之确定：远端第 2 块在 MLA 的第 $l$ 层对应编号 l * num_blocks + 2；本地新块（假设编号 40）对应 l * num_blocks + 40。make_prepped_xfer 接收两个等长的编号列表，NIXL 按索引配对，逐层搬完。

最后，按区域组装描述符编号并提交一次 READ 传输：

1
handle = self.nixl_wrapper.make_prepped_xfer(
2
    "READ",
3
    local_xfer_side_handle,
4
    local_block_descs_ids,
5
    remote_xfer_side_handle,
6
    remote_block_descs_ids,
7
    notif_msg=notif_id,
8
)
9
self.nixl_wrapper.transfer(handle)
10
self._recving_transfers[request_id].append(...)

notif_msg 随传输一起发出，意味着通知是在数据完成时由传输层发出的，不由本端代码手动触发。这避免了「通知先到、数据未到」的竞态。

完成：跨 TP rank 的完成计数

get_finished 需要解决一个问题：TP=8 时，8 个 rank 都在收，调度器只应该看到 rank 0 的结果。代码的处理：

1
if self.tp_rank == 0:
2
    for req_id in done_sending:
3
        self._done_sending_count[req_id] += 1
4
    for req_id in done_recving:
5
        self._done_recving_count[req_id] += 1
6
    other_ranks_finished_ids = []
7
    for i in range(1, self.world_size):
8
        other_ranks_finished_ids.extend(self.tp_group.recv_object(src=i))
9
    ...
10
    for req_id in list(self._done_recving_count.keys()):
11
        if self._done_recving_count[req_id] == self.world_size:
12
            ...

每个 rank 完成时把自己完成的 ID 发给 rank 0（tp_group.send_object(finished_req_ids, dst=0)），rank 0 计数，所有 rank 都完成后才返回给调度器。这里有个隐含顺序：rank 0 先处理自己的完成集合，再收其他 rank 的消息，因此其他 rank 报来的 ID 可能是「已完成但因某原因未在 rank 0 集合里」的，代码通过「在 _done_recving_count 或 _recving_transfers 里」来区分收发方向。

LMCache：把逻辑推给外部引擎

先看工厂如何实例化它。V1 连接器既可以来自内置注册表，也可以来自外部模块路径，后者是 LMCache 这类第三方后端接入的标准方式：

1
connector_name = kv_transfer_config.kv_connector
2
if connector_name in cls._registry:
3
    connector_cls = cls._registry[connector_name]()
4
else:
5
    connector_module_path = kv_transfer_config.kv_connector_module_path
6
    if connector_module_path is None:
7
        raise ValueError(
8
            f"Unsupported connector type: {connector_name}")
9
    connector_module = importlib.import_module(connector_module_path)
10
    connector_cls = getattr(connector_module, connector_name)
11
assert issubclass(connector_cls, KVConnectorBase_V1)

kv_connector_module_path 不是事后补的逃生通道，而是与注册表并列的一等入口：内置实现走注册表，外部实现走模块路径，两者在创建前都要通过 issubclass(..., KVConnectorBase_V1) 的类型检查。注册表的注册被刻意集中在一个文件（factory.py）里，注释解释原因——希望只加载当前连接器对应的文件，而不是把所有传输库都 import 进来。对一个同时支持 NCCL、NIXL、sglang 风格存储后端的框架来说，这一点很实际：import 即链接，链接不上的库会让进程起不来。

LMCacheConnectorV1 自身几乎只是转发：

1
def __init__(self, vllm_config, role):
2
    super().__init__(vllm_config=vllm_config, role=role)
3
    self._lmcache_engine = LMCacheConnectorV1Impl(vllm_config, role, self)
4

5
def start_load_kv(self, forward_context, **kwargs):
6
    self._lmcache_engine.start_load_kv(forward_context, **kwargs)

这不是偷懒，而是接口设计的副产品：只要连接器接口足够完整，外部项目就可以完全接管 KV 的存储、淘汰与传输策略。LMCache 借此实现了 CPU 内存、本地磁盘、远端存储的分层，并在调度器侧提供自己的前缀查询。对读源码的人来说，它是一个好例子：一个连接器不需要自己写任何 GPU kernel，只要把「多少个 token 命中」「在哪里」「何时搬完」三件事回答清楚。

SharedStorage：最小可用的参考实现

1
class SharedStorageConnector(KVConnectorBase_V1):
2
    # NOTE: This is Simple debug implementation of the KV connector.
3
    # It save / load the KV cache to / from the disk.
4
    # It does extra work which will overwrite the existing prefix-cache in GPU

它用请求的 token_ids 与槽位映射构造文件名和偏移，把 KV 层存成 safetensors，再读回。它的价值在于给出一个「把接口所有方法都实现」的骨架：ReqMeta.make_meta 里从 block_ids 与 block_size 算出 slot_mapping 的那段代码，是理解「分页块 ↔ 连续 token」映射最直观的例子：

1
block_offsets = torch.arange(0, block_size)
2
slot_mapping = block_offsets.reshape((1, block_size)) + \
3
        block_ids_tensor.reshape((num_blocks, 1)) * block_size
4
slot_mapping = slot_mapping.flatten()[:valid_num_tokens]

MultiConnector：一次命中、多处保存

1
class MultiConnector(KVConnectorBase_V1):
2
    """
3
    The current logic is:
4
    - Load KV from the first connector that advertises available tokens from
5
      get_num_new_matched_tokens(), based on the order in the config.
6
    - Save to all connectors.
7
    """

加载时只从第一个能提供内容的连接器读（避免重复读取同一段 KV），保存时向所有成员分发。实现上的难点在「异步保存计数」：多个成员可能各自异步保存，request_finished 需要等到所有成员的保存都完成才能释放块，所以它用 extra_async_saves 把「除一个之外的额外保存数」随元数据下发到 worker，由 worker 侧统计：

1
if async_saves > 1:
2
    self._extra_async_saves[request.request_id] = async_saves - 1

并且只允许一个成员返回 kv_transfer_params（否则 P/D 信息会冲突）：

1
if kv_txfer_params is not None:
2
    raise RuntimeError(
3
        "Only one connector can produce KV transfer params")

P2P NCCL：同步模式下的显式分工

P2pNcclConnector 用 is_producer / is_consumer 区分实例角色。生产者侧在 build_connector_meta 里把已完成的 prompt 打包发送，并专门处理分块 prefill：块还没凑齐时先记在 self.chunked_prefill 里，等所有块齐了再发。消费者侧先 recv_tensor，再按 slot_mapping 把数据写进分页池：

1
dst_kv_cache_layer[slot_mapping, ...] = src_kv_cache

注意这是按 token 槽位写入而非按块拷贝，所以能兼容「目标端块表与源端块编号不同」的情况，但也意味着它是逐层、逐请求的显式拷贝，无法走 RDMA 直写。

它的调度器侧实现更能说明「同步」模式的含义。消费者端（decode）的 get_num_new_matched_tokens 直接返回差量且 False：

1
if self.is_producer:
2
    return 0, False
3
num_external_tokens = (len(request.prompt_token_ids) - 1 - num_computed_tokens)
4
if num_external_tokens < 0:
5
    num_external_tokens = 0
6
return num_external_tokens, False

因为返回 False，请求不会进入 WAITING_FOR_REMOTE_KVS，而是正常进入本步调度；update_state_after_alloc 把请求放进 _requests_need_load，build_connector_meta 再为它构造 ReqMeta。于是同一步里：start_load_kv 阻塞 recv_tensor，注意力层计算读到的就是完整的 KV。代价很直接——拉取时间全额落在 TTFT 的关键路径上，且它只支持「一次性全部拉完」，没有层间流水。它适合单机多卡、带宽充足且请求长度可控的实验环境；上线到跨机 RDMA 时，通常会换成 NIXL 或 Mooncake 一类支持异步读的后端。

生产者端（prefill）则多了一层 chunked_prefill 记账：只有当本请求的 prompt 全部块被调度执行过（num_tokens >= len(prompt_token_ids)）才发 add_request，否则把已分配的块与完整 prompt token 列表存在 self.chunked_prefill[req_id] 里，等下一次调度接着累积。这段逻辑直观地展示了一个事实：连接器必须理解「一次 prefill 可能被切成多步」，否则会把半截 KV 发出去。

ReqMeta.make_meta 里还有一处值得记下的取舍：它把连续的 token 列表映射为「块偏移 + 块号」的槽位映射，align_to_block_size 把长度向上对齐到块边界，然后 slot_mapping = block_offsets + block_ids * block_size 展开再截断到有效 token 数。槽位映射就是注意力内核写 KV 时用的那张表，P2P 连接器在接收端构造同一张表，把收到的连续 [num_tokens, ...] 张量按槽位散写进分页池。「块表」和「槽位映射」是两件事：前者描述块与块的顺序，后者描述每个 token 落在哪一块的哪个位置，这一点在 vLLM 源码完全拆解（四）：KVCacheManager 如何分配、复用与回收物理 Block 也专门对比过。它的 get_num_new_matched_tokens 返回 (len(prompt) - 1 - num_computed, False)——同步：请求不会进入 WAITING_FOR_REMOTE_KVS，而是一边收一边算。

前缀命中与「外部命中」的叠加顺序

调度器先本地后外部，这个顺序决定了 get_num_new_matched_tokens 的实现必须考虑本地已命中的部分：NIXL 从 num_computed_tokens 开始算差量，LMCache 会查它自己的缓存并跳过本地前缀。如果连接器忽略了 num_computed_tokens，同一个前缀会被计算两遍，块表也会错位。

分块 prefill 下的延迟交接

P 端若开了分块 prefill，一次 schedule 只算 prompt 的一段。request_finished 只在请求结束时触发，因此只有在最后一段完成时 num_computed_tokens 才是 round_down(prompt_len, block_size)。P2P 连接器为此单独维护 chunked_prefill，NIXL 则依赖「结束时取块」，本质上都是「等凑齐再交」。

更麻烦的是 D 端：get_num_new_matched_tokens 里对 num_computed_tokens % block_size != 0 直接断言，因为 NIXL 按块拉取，D 端本地命中必然是整块。

ZOS 与异步调度下的时序

DeepInfer 用延后一拍的 ZOS 把 CPU 后处理移出关键路径；vLLM V1 没有这个名字，但它的调度器一样「先发这一步的执行，再处理上一步的输出」。连接器要适应这种时序：build_connector_meta 生成的是本步要搬的内容，而 get_finished 报告的是上步及之前的搬完状态。任何依赖「本步发起的传输在本步就完成」的假设都会错。

张量并行与异构

• TP 组内每个 rank 独立传输自己负责的头；
• rank 0 汇总完成情况，调度器只看一份结果；
• 异构 TP 需要额外的 tp_ratio 通知与「一个 D rank 对多个 P rank」的映射，这是 NIXL 实现里最复杂的部分之一，也在快速演进；
• 每层的 KV 布局因后端而异（NHD/HND、K/V 分离、MLA），注册与描述符构造都必须按后端实际 shape 走。

失败处理目前仍是薄弱点

NIXL 的握手回调里有一条 TODO：

1
# TODO: handle failure state of future in the
2
# callback, we want to fail the request in this case.
3
def request_ready(_f: Future[Any], entry=(req_id, meta)):
4
    self._ready_requests.put(entry)

握手失败时请求会一直停在 WAITING_FOR_REMOTE_KVS（除非另有超时机制），不会自动降级成本地 prefill。生产部署需要在上层编排层（router / proxy）设置超时与重试，或在连接器里补上失败降级。这是 KV Connector 作为「实验性 API」——KVConnectorBase_V1.__init__ 里那句 This API is experimental and subject to change——的真实含义。

与批次不变性、确定性共享的「同一份 KV」

值得一提的是一个容易被忽视的耦合：vLLM 的 NixlConnector 在读取时按块整块搬运，而 D 端计算时读的仍是自己分页池里的数据——KV 的数值只取决于源端如何计算它，不取决于 D 端 batch。这也意味着，如果 D 端启用了批不变算子，源端也必须启用，否则同一段前缀在不同 P 实例上算出的 KV 会有数值差异。Thinking Machines 实验室 2025 年的博客《Defeating Nondeterminism in LLM Inference》实验展示了同一请求在 batch=1 与 batch=768 下 logprob 可相差约 0.6，并因此让同一 1000 次运行的 completions 产生 18 种类别；他们的 batch_invariant_ops 库通过 set_batch_invariant_mode() 替换 matmul、RMSNorm、注意力等算子实现批不变。两幅对比图直观地展示了「相同输入、不同批次组成」如何分出多条轨迹：

（图源：Thinking Machines Lab 博客 Figure，svgs/nondeterministic.svg）

（图源：Thinking Machines Lab 博客 Figure，svgs/deterministic.svg）

对 P/D 分离而言，这条约束会被放大：同一个请求的 KV 来源可能换了一个 P 实例，如果两实例的算子路径不一致，前缀复用就会产生不可复现的数值漂移。

一次异步加载的逐步时序：把「占块不占预算」拆开看

设 D 实例在第 $t$ 步的调度中遇到请求 R：本地命中 0 token，NIXL 报告远端有 96 token（6 个块），异步加载。

1. 第 $t$ 步：调度器分配 6 个块给 R，标记 delay_cache_blocks=True；R 进入 WAITING_FOR_REMOTE_KVS，不进入本步的 token 预算。update_state_after_alloc 把 R 放进 _reqs_need_recv。
2. 第 $t$ 步末：build_connector_meta 生成一条 ReqMeta，包含本地块编号、远端块编号、远端 engine 地址；同时清空 _reqs_need_recv。元数据随 SchedulerOutput 下发。
3. 第 $t+1$ 步执行前：worker 绑定元数据，start_load_kv 发起 NIXL READ（若首次还要握手，先挂起）。
4. 第 $t+1$ 到 $t+k$ 步：R 一直在等待队列，调度器每步都会调用 _update_waiting_for_remote_kv 检查；未完成就继续跳过它。注意：此时它的 6 个块已经被占用，其他请求无法使用，但也没人读取它们。
5. 第 $t+k$ 步末：get_finished 返回 R；调度器把 R 转回 WAITING，num_computed_tokens=96（若 prompt 恰好 96 token 则减一，变成 95）。
6. 第 $t+k+1$ 步：R 被正常调度，计算最后一个 prompt token 并解码第一个输出 token。

这条时序解释了「异步」的真实代价：它没有阻塞 GPU，但占用了 KV 容量。如果大量请求同时等待远端 KV，分页池会被「占着但不可用」的块迅速吃满，触发其他请求的抢占。生产系统需要根据远端传输的 P99 延迟和并发量预留容量，或者限制同时处于 WAITING_FOR_REMOTE_KVS 的请求数。这是连接器接口之外、部署层必须补的策略。

上线前的检查清单

• 远端与本地块大小一致：block_size 是连接器、KV 管理器、注意力后端共享的常量；改分页粒度会同时改变两端布局。
• 并行拓扑写进传输参数：tp_size 必须随 kv_transfer_params 传给 D 端，否则 tp_ratio 算错，读到的块会缺头或少尾。
• KV 布局按后端校准：FlashInfer 的维度交换、MLA 的 3D 张量、FP8 的 slot_size_bytes，任何一项不一致都会让描述符越界或错位。
• 完成通知必须送达：全命中时也要发 send_notif，否则 P 端块泄漏；这是 P/D 分离最常见的「显存慢漏」来源。
• 失败要能降级：握手失败、传输超时、对端重启，都应该让请求回退到本地 prefill；当前 NIXL 实现里这条路径还不完整。
• 两侧的数值路径一致：如果启用批不变算子，P 与 D 必须同时启用；前缀复用的前提是「同一段 token 算出的 KV 相同」。