updated

Signed-off-by: Robert Shaw <robshaw@redhat.com>

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

@@ -874,13 +874,17 @@ class NixlConnectorWorker:
         Returns:
             set of req_ids that have all done xfers
         """
+        current_time = time.perf_counter()
+
         done_req_ids: set[str] = set()
         for req_id, handles in list(transfers.items()):
             in_progress = False
-            for handle, _xfer_stime in handles:
+            for handle, _xfer_stime, remote_id in handles:
                 xfer_state = self.nixl_wrapper.check_xfer_state(handle)
                 if xfer_state == "DONE":
                     self.nixl_wrapper.release_xfer_handle(handle)
+                    logger.info("========= TRANSFER: req_id %s remote_engine_id %s transfer time: %s",
+                                req_id, remote_id, current_time - _xfer_stime)
                 elif xfer_state == "PROC":
                     in_progress = True
                     continue
@@ -1012,6 +1016,7 @@ class NixlConnectorWorker:
         assert len(local_block_descs_ids) == len(remote_block_descs_ids)
 
         # Prepare transfer with Nixl.
+        start = time.perf_counter()
         handle = self.nixl_wrapper.make_prepped_xfer(
             "READ",
             local_xfer_side_handle,
@@ -1019,15 +1024,20 @@ class NixlConnectorWorker:
             remote_xfer_side_handle,
             remote_block_descs_ids,
             notif_msg=notif_id,
+            skip_desc_merge=True,
         )
 
         # Begin async xfer.
         self.nixl_wrapper.transfer(handle)
+        end = time.perf_counter()
+        # logger.debug(
+        #     "[nixl connector]: req_id %s transfer launch time %s",
+        #     request_id, end - start)
 
         # Use handle to check completion in future step().
         # TODO (NickLucche) surface xfer elapsed time
         self._recving_transfers[request_id].append(
-            (handle, time.perf_counter()))
+            (handle, time.perf_counter(), dst_engine_id))
 
     def _get_block_descs_ids(self,
                              engine_id: str,

vllm/v1/engine/async_llm.py

@@ -60,6 +60,7 @@ class AsyncLLM(EngineClient):
         client_addresses: Optional[dict[str, str]] = None,
         client_index: int = 0,
     ) -> None:
+        print("====== HELLO ======")
         """
         Create an AsyncLLM.