fix_pd_expiry

Signed-off-by: underfituu <[email protected]>

Author: underfituu

vllm_ascend/distributed/llmdatadist_c_mgr_connector.py

@@ -351,7 +351,7 @@ def __init__(self, vllm_config: VllmConfig):
         os.environ["HCCL_DETERMINISTIC"] = "true"
         self.done_receiving_counts: defaultdict[str,
                                                 set[int]] = defaultdict(set)
-        self._reqs_to_send: dict[str, float] = {}
+        self.reqs_to_send: dict[str, float] = {}
 
     def listen_for_agent_metadata_req(self, event: threading.Event):
         assert self.local_agent_metadata is not None
@@ -618,7 +618,7 @@ def handle_exception(future):
 
         for future in futures:
             future.add_done_callback(handle_exception)
-        self._reqs_to_send.update(metadata._reqs_need_send)
+        self.reqs_to_send.update(metadata._reqs_need_send)
 
     def add_remote_agent(self, metadata: LLMDataDistCMgrAgentMetadata) -> int:
         assert self.local_agent_metadata is not None
@@ -876,8 +876,8 @@ def get_finished(
         import copy
         now = time.perf_counter()
         with self.thread_lock:
-            while self._reqs_to_send:
-                req_id, expires = next(iter(self._reqs_to_send.items()))
+            while self.reqs_to_send:
+                req_id, expires = next(iter(self.reqs_to_send.items()))
                 if now < expires:
                     break
                 logger.warning(
@@ -886,7 +886,7 @@ def get_finished(
                         )
                 if req_id not in self.finished_reqs:
                     self.finished_reqs.add(req_id)
-                del self._reqs_to_send[req_id]
+                del self.reqs_to_send[req_id]
             req_ids_to_ret = copy.deepcopy(self.finished_reqs)
             self.finished_reqs.clear()
         if self.llm_datadist_role == LLMRole.PROMPT: