Main server loop that handles incoming requests directly.
(self)
| 252 | f"[server] Drained {drained_count} requests after shutdown") |
| 253 | |
| 254 | async def _run_server(self) -> None: |
| 255 | """Main server loop that handles incoming requests directly.""" |
| 256 | assert self._client_socket is not None, "Client socket is not bound" |
| 257 | |
| 258 | logger_debug("[server] RPC Server main loop started") |
| 259 | |
| 260 | # Create worker tasks |
| 261 | for i in range(self._num_workers): |
| 262 | task = asyncio.create_task(self._process_requests()) |
| 263 | self._worker_tasks.append(task) |
| 264 | |
| 265 | try: |
| 266 | # Wait for all worker tasks to complete |
| 267 | await asyncio.gather(*self._worker_tasks) |
| 268 | except asyncio.CancelledError: |
| 269 | logger_debug("[server] RPC Server main loop cancelled") |
| 270 | # Cancel all worker tasks |
| 271 | for task in self._worker_tasks: |
| 272 | if not task.done(): |
| 273 | task.cancel() |
| 274 | # Wait for all tasks to finish cancellation |
| 275 | await asyncio.gather(*self._worker_tasks, return_exceptions=True) |
| 276 | except Exception as e: |
| 277 | logger.error(f"RPC Server main loop error: {e}") |
| 278 | logger.error(traceback.format_exc()) |
| 279 | finally: |
| 280 | logger_debug("[server] RPC Server main loop exiting") |
| 281 | |
| 282 | # TODO optimization: resolve the sequential scheduling for the remote calls |
| 283 | # Suppose tons of submit remote call block the FIFO queue, and the later get_stats remote calls may be blocked |
no test coverage detected