Method route_request

gunicorn/dirty/arbiter.py:445–495

Route a request to an available dirty worker via queue. Each worker has a dedicated queue and consumer task. Requests are submitted to the queue and processed sequentially by the consumer. For streaming responses, messages (chunks) are forwarded directly to

(self, request, client_writer)

                pass

    async def route_request(self, request, client_writer):
        """
        Route a request to an available dirty worker via queue.

        Each worker has a dedicated queue and consumer task. Requests are
        submitted to the queue and processed sequentially by the consumer.

        For streaming responses, messages (chunks) are forwarded directly
        to the client_writer as they arrive from the worker.

        Args:
            request: Request message dict
            client_writer: StreamWriter to send responses to client
        """
        request_id = request.get("id", "unknown")
        app_path = request.get("app_path")

        # Find an available worker (filtered by app if specified)
        worker_pid = await self._get_available_worker(app_path)
        if worker_pid is None:
            # Distinguish between no workers at all vs. no workers for this app
            if not self.workers:
                error = DirtyError("No dirty workers available")
            elif app_path and self.app_specs:
                # Per-app allocation is configured and no workers have this app
                error = DirtyNoWorkersAvailableError(app_path)
            else:
                error = DirtyError("No dirty workers available")
            response = make_error_response(request_id, error)
            await DirtyProtocol.write_message_async(client_writer, response)
            return

        # Get queue (start consumer if needed)
        if worker_pid not in self.worker_queues:
            await self._start_worker_consumer(worker_pid)

        queue = self.worker_queues[worker_pid]
        future = asyncio.get_running_loop().create_future()

        # Submit request to queue with client writer for streaming support
        await queue.put((request, client_writer, future))

        # Wait for completion (streaming messages forwarded by consumer)
        try:
            await future
        except Exception as e:
            response = make_error_response(
                request_id,
                DirtyWorkerError(f"Request failed: {e}", worker_id=worker_pid)
            )
            await DirtyProtocol.write_message_async(client_writer, response)

    async def _start_worker_consumer(self, worker_pid):
        """Start a consumer task for a worker's request queue."""