Method _execute_on_worker

gunicorn/dirty/arbiter.py:523–581

Execute request on a specific worker (called by consumer). Handles both regular responses and streaming (chunk-based) responses. For streaming, chunk and end messages are forwarded directly to the client_writer as they arrive from the worker.

(self, worker_pid, request, client_writer)

        self.worker_consumers[worker_pid] = task

    async def _execute_on_worker(self, worker_pid, request, client_writer):
        """
        Execute request on a specific worker (called by consumer).

        Handles both regular responses and streaming (chunk-based) responses.
        For streaming, chunk and end messages are forwarded directly to the
        client_writer as they arrive from the worker.
        """
        request_id = request.get("id", "unknown")

        try:
            reader, writer = await self._get_worker_connection(worker_pid)
            await DirtyProtocol.write_message_async(writer, request)

            # Read messages until we get a response, end, or error
            while True:
                try:
                    message = await asyncio.wait_for(
                        DirtyProtocol.read_message_async(reader),
                        timeout=self.cfg.dirty_timeout
                    )
                except asyncio.TimeoutError:
                    response = make_error_response(
                        request_id,
                        DirtyTimeoutError("Worker timeout", self.cfg.dirty_timeout)
                    )
                    await DirtyProtocol.write_message_async(client_writer, response)
                    return

                msg_type = message.get("type")

                # Forward chunk messages to client
                if msg_type == DirtyProtocol.MSG_TYPE_CHUNK:
                    await DirtyProtocol.write_message_async(client_writer, message)
                    continue

                # Forward end message and complete
                if msg_type == DirtyProtocol.MSG_TYPE_END:
                    await DirtyProtocol.write_message_async(client_writer, message)
                    return

                # Forward response or error and complete
                if msg_type in (DirtyProtocol.MSG_TYPE_RESPONSE,
                                DirtyProtocol.MSG_TYPE_ERROR):
                    await DirtyProtocol.write_message_async(client_writer, message)
                    return

                # Unknown message type - log and continue
                self.log.warning("Unknown message type from worker: %s", msg_type)

        except Exception as e:
            self.log.error("Error executing on worker %s: %s", worker_pid, e)
            self._close_worker_connection(worker_pid)
            response = make_error_response(
                request_id,
                DirtyWorkerError(f"Worker communication failed: {e}",
                                 worker_id=worker_pid)
            )