router recv reqs update

Author: shihaobai

lightllm/server/router/manager.py

@@ -79,7 +79,7 @@ def __init__(self, args: StartArgs, router_port, detokenization_port, metric_por
         self.eos_id = args.eos_id
         self.has_wait_tokens = 0
         self.max_wait_tokens = args.router_max_wait_tokens
-        context = zmq.asyncio.Context(2)
+        context = zmq.Context(2)
         self.recv_from_httpserver = context.socket(zmq.PULL)
         self.recv_from_httpserver.bind(f"{args.zmq_mode}127.0.0.1:{router_port}")
 
@@ -442,13 +442,21 @@ def get_used_tokens(self, dp_index):
         else:
             return self.max_total_token_num - self.read_only_statics_mem_manager.get_unrefed_token_num(dp_index)
 
+    def recv_reqs(self):
+        while True:
+            try:
+                recv_req: GroupReqIndexes = self.recv_from_httpserver.recv_pyobj(zmq.NOBLOCK)
+                if isinstance(recv_req, GroupReqIndexes):
+                    self.add_req(recv_req)
+                else:
+                    assert False, f"Error Req Inf {recv_req}"
+            except zmq.ZMQError:
+                return
+
     async def loop_for_netio_req(self):
         while True:
-            recv_req: GroupReqIndexes = await self.recv_from_httpserver.recv_pyobj()
-            if isinstance(recv_req, GroupReqIndexes):
-                self.add_req(recv_req)
-            else:
-                assert False, f"Error Req Inf {recv_req}"
+            self.recv_reqs()
+            await asyncio.sleep(0.01)
 
     def clean_up(self):
         return