Method _process_request

tensorrt_llm/executor/rpc/rpc_server.py:447–495

Process a request. Returns None for streaming requests (handled separately).

(self, req: RPCRequest)

        return adjusted_timeout

    async def _process_request(self, req: RPCRequest) -> Optional[RPCResponse]:
        """Process a request. Returns None for streaming requests (handled separately)."""
        func = self._functions[req.method_name]

        # Calculate adjusted timeout based on pending overhead
        adjusted_timeout = self._calculate_adjusted_timeout(req)

        try:
            if inspect.iscoroutinefunction(func):
                # Execute async function directly in event loop, no need to run in executor due to the GIL
                logger_debug(
                    f"[server] RPC Server running async task {req.method_name} in dispatcher"
                )
                result = await asyncio.wait_for(func(*req.args, **req.kwargs),
                                                timeout=adjusted_timeout)
            else:
                # Execute sync function in thread executor
                loop = asyncio.get_running_loop()

                def call_with_kwargs():
                    return func(*req.args, **req.kwargs)

                logger_debug(
                    f"[server] RPC Server running async task {req.method_name} in worker"
                )
                # TODO: let num worker control the pool size
                result = await asyncio.wait_for(loop.run_in_executor(
                    self._executor, call_with_kwargs),
                                                timeout=adjusted_timeout)

            response = RPCResponse(req.request_id, result=result)

        except asyncio.TimeoutError:
            response = RPCResponse(
                req.request_id,
                result=None,
                error=RPCTimeout(
                    f"Method '{req.method_name}' timed out after {req.timeout} seconds",
                    traceback=traceback.format_exc()))

        except Exception as e:
            response = RPCResponse(req.request_id,
                                   result=None,
                                   error=RPCError(
                                       str(e),
                                       cause=e,
                                       traceback=traceback.format_exc()))

        return response

    async def _process_streaming_request(self, req: RPCRequest) -> None:
        """Process a streaming request by sending multiple responses."""