[P/D] NIXL Updates (vllm-project#25844)

Signed-off-by: Sage Moore <[email protected]>
Signed-off-by: simon-mo <[email protected]>
Signed-off-by: rentianyue-jk <[email protected]>
Signed-off-by: Russell Bryant <[email protected]>
Signed-off-by: Isotr0py <[email protected]>
Signed-off-by: Chenheli Hua <[email protected]>
Signed-off-by: mgoin <[email protected]>
Signed-off-by: Tyler Michael Smith <[email protected]>
Signed-off-by: NickLucche <[email protected]>
Signed-off-by: Roger Wang <[email protected]>
Signed-off-by: Robert Shaw <[email protected]>
Co-authored-by: Sage Moore <[email protected]>
Co-authored-by: Russell Bryant <[email protected]>
Co-authored-by: rentianyue-jk <[email protected]>
Co-authored-by: Isotr0py <[email protected]>
Co-authored-by: Chenheli Hua <[email protected]>
Co-authored-by: Wentao Ye <[email protected]>
Co-authored-by: Michael Goin <[email protected]>
Co-authored-by: Tyler Michael Smith <[email protected]>
Co-authored-by: Nicolò Lucchesi <[email protected]>
Co-authored-by: Roger Wang <[email protected]>
Co-authored-by: Robert Shaw <[email protected]>

Author: 12 people

vllm/distributed/kv_transfer/kv_connector/v1/nixl_connector.py

@@ -474,8 +474,11 @@ def __init__(self, vllm_config: VllmConfig, engine_id: str):
                 "backends", ["UCX"])
         # Agent.
         non_ucx_backends = [b for b in self.nixl_backends if b != "UCX"]
-        config = nixl_agent_config(backends=self.nixl_backends) if len(
-            non_ucx_backends) > 0 and nixl_agent_config is not None else None
+        if nixl_agent_config is None:
+            config = None
+        else:
+            config = nixl_agent_config(backends=self.nixl_backends) if len(
+                non_ucx_backends) > 0 else nixl_agent_config(num_threads=8)
 
         self.nixl_wrapper = NixlWrapper(str(uuid.uuid4()), config)
         # Map of engine_id -> {rank0: agent_name0, rank1: agent_name1..}.

vllm/v1/core/sched/scheduler.py

@@ -1290,4 +1290,9 @@ def _update_from_kv_xfer_finished(self,
             self.finished_recving_kv_req_ids.add(req_id)
         for req_id in (kv_connector_output.finished_sending or ()):
             logger.debug("Finished sending KV transfer for request %s", req_id)
-            self._free_blocks(self.requests[req_id])
+            if req_id not in self.requests:
+                logger.warning(
+                    "Got finished sending KV transfer for request %s,"
+                    "but the request is already freed.", req_id)
+            else:
+                self._free_blocks(self.requests[req_id])