Method _response_reader

tensorrt_llm/executor/rpc/rpc_client.py:338–413

Task to read responses from the socket and set results on futures.

(self)

                continue

    async def _response_reader(self):
        """Task to read responses from the socket and set results on futures."""
        logger_debug("[client] Response reader started")
        self._reader_asyncio_task = asyncio.current_task()

        try:
            # Add initial delay to ensure socket is fully connected
            # This helps prevent race conditions during initialization
            await asyncio.sleep(0.1)
            logger_debug("[client] Response reader ready to process messages")

            with customized_gc_thresholds(10000):
                last_alive_log = time.time()
                while not self._closed:
                    # Periodic alive logging for debugging
                    if time.time() - last_alive_log > 5.0:
                        logger_debug(
                            "[client] Response reader is alive and waiting for responses"
                        )
                        last_alive_log = time.time()

                    with nvtx_range_debug("response_reader",
                                          color="cyan",
                                          category="RPC"):
                        try:
                            response = await self._wait_for_response()
                            logger_debug(
                                f"[client] [{datetime.now().isoformat()}] Received response: {response}"
                            )

                            nvtx_mark_debug(
                                f"RPC.response.{'streaming' if response.is_streaming else 'sync'}",
                                color="black",
                                category="RPC")

                            # Optimize: Check debug flag before expensive string operations
                            # This avoids holding GIL for f-string evaluation when debug is disabled
                            if enable_llmapi_debug() or logger.level == 'debug':
                                logger_debug(
                                    f"[client] [{datetime.now().isoformat()}] RPC Client received response: request_id={response.request_id}, "
                                    f"is_streaming={response.is_streaming}, "
                                    f"pending_futures={len(self._pending_futures)}"
                                )

                            with nvtx_range_debug("handle_response",
                                                  color="purple",
                                                  category="RPC"):
                                if response.is_streaming:
                                    self._handle_streaming_response(response)
                                else:
                                    self._handle_regular_response(response)

                        except asyncio.CancelledError:
                            # Re-raise cancellation to exit cleanly
                            raise
                        except Exception as e:
                            # Log the error but continue reading unless it's a critical error
                            logger.error(f"Error processing response: {e}",