apps/FastDeploy
1
0
Fork 0
mirror of https://github.com/PaddlePaddle/FastDeploy.git synced 2025-10-06 09:07:10 +08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Fix bug for concurrently visit zmq (#3233)

Browse Source
This commit is contained in:
chenjian
2025-08-06 10:41:10 +08:00
committed by GitHub
parent b20ffe3697
commit a9d231c900
1 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
fastdeploy/splitwise/internal_adapter_utils.py
View File

@@ -43,6 +43,7 @@ class InternalAdapter:
target=self._response_external_module_control_instruct, daemon=True target=self._response_external_module_control_instruct, daemon=True
) )
self.response_external_instruct_thread.start() self.response_external_instruct_thread.start()
self.response_lock = threading.Lock() # prevent to call send_multipart in zmq concurrently
def _get_current_server_info(self): def _get_current_server_info(self):
""" """
@@ -76,6 +77,7 @@ class InternalAdapter:
payload_info = self._get_current_server_info() payload_info = self._get_current_server_info()
result = {"task_id": task_id_str, "result": payload_info} result = {"task_id": task_id_str, "result": payload_info}
logger.info(f"Response for task: {task_id_str}") logger.info(f"Response for task: {task_id_str}")
with self.response_lock:
self.recv_control_cmd_server.response_for_control_cmd(task_id_str, result) self.recv_control_cmd_server.response_for_control_cmd(task_id_str, result)
elif task["cmd"] == "get_metrics": elif task["cmd"] == "get_metrics":
@@ -85,6 +87,7 @@ class InternalAdapter:
) )
result = {"task_id": task_id_str, "result": metrics_text} result = {"task_id": task_id_str, "result": metrics_text}
logger.info(f"Response for task: {task_id_str}") logger.info(f"Response for task: {task_id_str}")
with self.response_lock:
self.recv_control_cmd_server.response_for_control_cmd(task_id_str, result) self.recv_control_cmd_server.response_for_control_cmd(task_id_str, result)
elif task["cmd"] == "connect_rdma": elif task["cmd"] == "connect_rdma":
self.engine.engine_worker_queue.put_connect_rdma_task(task) self.engine.engine_worker_queue.put_connect_rdma_task(task)
@@ -100,6 +103,7 @@ class InternalAdapter:
task_id_str = result_data["task_id"] task_id_str = result_data["task_id"]
result = {"task_id": task_id_str, "result": result_data} result = {"task_id": task_id_str, "result": result_data}
logger.info(f"Response for task: {task_id_str}") logger.info(f"Response for task: {task_id_str}")
with self.response_lock:
self.recv_control_cmd_server.response_for_control_cmd(task_id_str, result) self.recv_control_cmd_server.response_for_control_cmd(task_id_str, result)
else: else:
time.sleep(0.001) time.sleep(0.001)