Method _process_requests

tensorrt_llm/executor/rpc/rpc_server.py:335–420

Process incoming requests directly from the socket.

(self)

        return True

    async def _process_requests(self) -> None:
        """Process incoming requests directly from the socket."""
        assert self._client_socket is not None, "Client socket is not bound"

        while not self._shutdown_event.is_set():
            try:
                #logger_debug(f"[server] Worker waiting for request", color="green")
                # Read request directly from socket with timeout
                req, routing_id = await asyncio.wait_for(
                    self._client_socket.get_async_noblock(return_identity=True),
                    timeout=2)
                req.routing_id = routing_id
                logger_debug(f"[server] Worker got request: {req}",
                             color="green")
            except asyncio.TimeoutError:
                continue
            except asyncio.CancelledError:
                logger_debug("[server] RPC worker cancelled")
                break
            except Exception as e:
                if self._shutdown_event.is_set():
                    break
                logger.error(f"RPC worker caught an exception: {e}")
                logger.error(traceback.format_exc())
                continue

            # shutdown methods depend on _num_pending_requests, so
            # they should not be counted
            if req.method_name not in ["_rpc_shutdown", "shutdown"]:
                self._num_pending_requests += 1
                logger_debug(
                    f"[server] Worker received request {req}, pending: {self._num_pending_requests}"
                )

            # Check if we should cancel due to shutdown
            if await self._handle_shutdown_request(req):
                continue

            # Check if the method exists
            if req.method_name not in self._functions:
                logger.error(
                    f"Method '{req.method_name}' not found in RPC server.")
                self._num_pending_requests -= 1

                error = RPCStreamingError if req.is_streaming else RPCError
                await self._send_error_response(
                    req,
                    error(
                        f"Method '{req.method_name}' not found in RPC server.",
                        traceback=traceback.format_exc()))
                continue

            func = self._functions[req.method_name]

            # Final shutdown check before processing
            if await self._handle_shutdown_request(req):
                continue