class KVTransferAgent:    def __init__(        self,        rank: int,        local_rank: int,        config: "VllmConfig",    ):        self.config = config        if config.kv_transfer_config is None:            raise ValueError("KVTransferConfig is not set in the VllmConfig,"                             " cannot initialize KVConnector.")        assert self.config.kv_transfer_config.is_kv_transfer_instance, "KV"\            "TransferAgent should only be used when kv_connector is set."        # 工厂类的作用        self.connector = KVConnectorFactory.create_connector_v0(            rank, local_rank, config)

vllm/vllm/distributed/kv_transfer/kv_connector/base.py

KVConnectorFactory.register_connector(    "PyNcclConnector",    "vllm.distributed.kv_transfer.kv_connector.simple_connector",    "SimpleConnector")KVConnectorFactory.register_connector(    "MooncakeConnector",    "vllm.distributed.kv_transfer.kv_connector.simple_connector",    "SimpleConnector")KVConnectorFactory.register_connector(    "LMCacheConnector",    "vllm.distributed.kv_transfer.kv_connector.lmcache_connector",    "LMCacheConnector")KVConnectorFactory.register_connector(    "MooncakeStoreConnector",    "vllm.distributed.kv_transfer.kv_connector.mooncake_store_connector",    "MooncakeStoreConnector")

取值操作操作：input_tokens = torch.tensor([1, 2, 3, 4, 5])roi = torch.tensor([False, True, False, True, False], dtype=torch.bool)result = input_tokens[roi]print(result)  # 输出：tensor([2, 4])POP操作：原始 buffer: [10, 20, 30, 40, 50]input_tokens: tensor([1, 2, 3, 4, 5])roi: tensor([False,  True, False,  True, False])buffer.pop(input_tokens, roi)处理后的 buffer: [10, 30, 50]

class KVPipeBase(ABC):    """    This class provides an interface for sending and receiving tensors, or    None, by distributed communications.    """    @abstractmethod    def send_tensor(self, tensor: Optional[torch.Tensor]) -> None:        """Send a tensor, or None, via the pipe.        """        raise NotImplementedError    @abstractmethod    def recv_tensor(self) -> Optional[torch.Tensor]:        """Receive a tensor (can be None) from the pipeline.        """        raise NotImplementedError    @abstractmethod    def close(self) -> None:        """Close the pipeline and release resources.        """        raise NotImplementedError

# vllm/vllm/v1/core/sched/scheduler.py
# scheduler中的函数调用位置：# KVConnector: update internal state after allocation.                # This information is used to determine if a load is                # needed for this request. if self.connector is not None:self.connector.update_state_after_alloc(request,new_computed_blocks + new_blocks,                        num_external_computed_tokens,)
# ...
# 计算远端已有kv数量，                # Get externally-cached tokens if using a KVConnector.                num_external_computed_tokens, load_kv_async = (                    (0, False) if self.connector is None else                    self.connector.get_num_new_matched_tokens(                        request, num_native_computed_tokens))                # Total computed tokens (local + external).                num_computed_tokens = (num_native_computed_tokens +                                       num_external_computed_tokens +                                       num_prealloc_computed_tokens)# ...        # NOTE(Kuntai): this function is designed for multiple purposes:        # 1. Plan the KV cache store        # 2. Wrap up all the KV cache load / save ops into an opaque object        # 3. Clear the internal states of the connector        if self.connector is not None:            meta = self.connector.build_connector_meta(scheduler_output)            scheduler_output.kv_connector_metadata = meta

#vllm/vllm/attention/layer.py# worker connecotr 使用位置（attention部分）：# ...# 加载逻辑使用def wait_for_kv_layer_from_connector(layer_name: str):    if not has_kv_transfer_group() or not is_v1_kv_transfer_group():        return    connector = get_kv_transfer_group()    forward_context: ForwardContext = get_forward_context()    attn_metadata = forward_context.attn_metadata    if attn_metadata is None:        return    assert isinstance(attn_metadata, dict)    connector.wait_for_layer_load(layer_name)# ...# 保存逻辑使用def maybe_save_kv_layer_to_connector(    layer_name: str,    kv_cache_layer: List[torch.Tensor],):    if not has_kv_transfer_group() or not is_v1_kv_transfer_group():        return    connector = get_kv_transfer_group()    forward_context: ForwardContext = get_forward_context()    attn_metadata = forward_context.attn_metadata    if attn_metadata is None:        return    assert isinstance(attn_metadata, dict)    connector.save_kv_layer(layer_name, kv_cache_layer,                            attn_metadata[layer_name])

# scheduler 的修改# 增加了一种状态：                # P/D: skip request if still waiting for remote kvs.                if request.status == RequestStatus.WAITING_FOR_REMOTE_KVS:                    is_ready = self._update_waiting_for_remote_kv(request)                    if is_ready:                        request.status = RequestStatus.WAITING                    else:                        self.waiting.popleft()                        skipped_waiting_requests.appendleft(request)                        continue# 增加PD的处理逻辑    ########################################################################    # P/D Related Methods    ########################################################################    def get_kv_connector(self) -> Optional[KVConnectorBase_V1]:        return self.connector    def _connector_finished(            self, request: Request) -> tuple[bool, Optional[KVTransferParams]]:        """Invoke the KV connector request_finished() method if applicable."""        if self.connector is None:            return False, None        block_ids = self.kv_cache_manager.get_block_ids(request.request_id)        return self.connector.request_finished(request, block_ids)    def _update_waiting_for_remote_kv(self, request: Request) -> bool:        """        P/D: check if the request_id is finished_recving.        The finished_recving_kv_req_ids list is populated        on the previous steps()'s update_from_output based        on the worker side connector.        When the kv transfer is ready, we cache the blocks        and the request state will be moved back to WAITING from        WAITING_FOR_REMOTE_KV.        """        if request.request_id not in self.finished_recving_kv_req_ids:            return False        # Now that the blocks are ready, actually cache them.        block_ids = self.kv_cache_manager.get_block_ids(request.request_id)        num_computed_tokens = len(block_ids) * self.block_size        if num_computed_tokens == request.num_tokens:            num_computed_tokens -= 1        self.kv_cache_manager.single_type_manager.cache_blocks(            request,            self.kv_cache_manager.req_to_block_hashes[request.request_id],            num_computed_tokens,        )        # Update the request state for scheduling.        request.num_computed_tokens = num_computed_tokens        # Return that we are ready.        self.finished_recving_kv_req_ids.remove(request.request_id)        return True    def _update_from_kv_xfer_finished(self,                                      model_runner_output: ModelRunnerOutput):        """        P/D: update the scheduler state based on the output.        The Worker side connectors add finished_recving and        finished_sending reqs to the output.        * if finished_sending: free the blocks        # if finished_recving: add to state so we can            scheduler the request during the next step.        """        # P/D: update recv and send status from last step.        for req_id in (model_runner_output.finished_recving or ()):            logger.debug("Finished recving KV transfer for request %s", req_id)            self.finished_recving_kv_req_ids.add(req_id)        for req_id in (model_runner_output.finished_sending or ()):            logger.debug("Finished sending KV transfer for request %s", req_id)            self._free_blocks(self.requests[req_id])