Introduction to Queue-Based Throttling

Welcome to the third lesson of the "Throttling API Requests" course! In our previous lessons, we explored various throttling techniques, such as enhancing throttling middleware and implementing the Token Bucket algorithm. Now, we will delve into the concept of queue-based throttling specifically designed for FastAPI applications using asyncio. This technique is crucial for managing API requests by queuing them when the server is busy, preventing server overload, and ensuring fair access to resources. By the end of this lesson, you'll be equipped to implement a queue-based throttling mechanism in your FastAPI REST API using Python's asyncio, enhancing its security and reliability.

What is Queue-Based Throttling?

Queue-based throttling is a technique that limits the number of concurrent requests being processed by placing excess requests in a waiting queue. Unlike other throttling methods that may reject requests immediately when limits are reached, queue-based throttling allows requests to wait for their turn to be processed.

Benefits:

  • Improved User Experience: Instead of immediately rejecting excess requests, users' requests get processed when resources become available
  • Better Resource Utilization: The server processes requests at a consistent, sustainable rate
  • Fairness: Requests are typically processed in a First-In-First-Out (FIFO) manner, ensuring fair treatment
  • Graceful Degradation: When traffic spikes occur, the system degrades gracefully by increasing wait times rather than failing

Drawbacks:

  • Increased Memory Usage: Maintaining a queue of requests consumes memory
  • Request Timeout Challenges: Long-queued requests may time out at the client side before being processed
  • Complexity: Implementation is more complex than simple rate-limiting techniques
  • Potential for Resource Starvation: If improperly configured, a flood of low-priority requests might delay critical ones
Core Components of Queue-Based Throttling

Queue-based throttling in FastAPI with asyncio involves three key components:

  • Request Queue: A data structure that holds incoming requests when the server is busy
  • Maximum Concurrent Requests: The maximum number of requests processed simultaneously
  • Queue Timeout: The maximum time a request can wait in the queue before being timed out
Implementing Queue-Based Throttling: Setting Up the Queue

Let's implement queue-based throttling in our Python REST API using FastAPI. We'll break down the implementation into several key components to make it easier to understand and implement.

First, we need to set up our queue structure and define our configuration:

import asyncio
from dataclasses import dataclass
from fastapi import HTTPException, Request, Response
from typing import List, Optional
import time

# Configuration constants
MAX_CONCURRENT = 3
MAX_QUEUE_SIZE = 10
QUEUE_TIMEOUT_MS = 5000  # 5 seconds timeout

current_requests = 0

@dataclass
class QueuedRequest:
    request: Request
    response: Response
    timestamp: float  # When the request was queued
    ready_event: asyncio.Event  # Signal when request is ready or timed out
    timed_out: bool = False  # Flag to indicate if request timed out

request_queue: List[QueuedRequest] = []
Processing the Queue

The most challenging part of queue-based throttling is managing the queue processing logic using Python's asyncio:

class QueueThrottler:
    def __init__(self):
        self.queue_task: Optional[asyncio.Task] = None
        self.running = False
    
    async def initialize(self):
        """Initialize the queue processing task"""
        if self.queue_task is None:
            self.running = True
            self.queue_task = asyncio.create_task(self._process_queue())
    
    async def _process_queue(self):
        """Continuously process the queue"""
        global current_requests, request_queue
        
        try:
            while self.running:
                await asyncio.sleep(0.1)  # Process every 100ms
                
                now = time.time() * 1000  # Convert to milliseconds
                
                # First, remove expired requests
                i = 0
                while i < len(request_queue):
                    queued_req = request_queue[i]
                    if now - queued_req.timestamp > QUEUE_TIMEOUT_MS:
                        # Request has timed out, signal the handler
                        queued_req.timed_out = True
                        queued_req.ready_event.set()
                        request_queue.pop(i)
                    else:
                        i += 1
                
                # Process eligible requests
                while current_requests < MAX_CONCURRENT and len(request_queue) > 0:
                    queued_req = request_queue.pop(0)  # FIFO - remove from front
                    current_requests += 1
                    
                    # Signal the request is ready to proceed
                    queued_req.ready_event.set()
                    
                    # Create a task to handle the request completion tracking
                    asyncio.create_task(self._track_request_completion())
        
        except asyncio.CancelledError:
            pass
    
    async def _track_request_completion(self):
        """Track when a request completes to decrement the counter"""
        global current_requests
        
        try:
            # Wait for the request to complete (this is a simplified approach)
            # In a real implementation, you'd integrate more closely with FastAPI's lifecycle
            await asyncio.sleep(0.1)  # Small delay to allow request processing
            
        finally:
            current_requests -= 1

The critical logic here is:

  1. We use asyncio.sleep() to periodically check and process the queue
  2. We first remove expired requests (those waiting too long)
  3. We use asyncio.Event to signal when a request is ready or has timed out
  4. We then process requests up to our concurrency limit
  5. We track request completion through async tasks to free up slots for new requests
Creating the Throttling Middleware

Finally, we implement the actual middleware function that will be used in our FastAPI application:

from contextlib import asynccontextmanager

# Global throttler instance
throttler = QueueThrottler()

@asynccontextmanager
async def queue_throttle(request: Request, response: Response):
    """Queue-based throttling context manager"""
    global current_requests, request_queue
    
    # Initialize throttler if needed
    if throttler.queue_task is None:
        await throttler.initialize()
    
    # Check if queue is full
    if len(request_queue) >= MAX_QUEUE_SIZE:
        raise HTTPException(
            status_code=503,
            detail="Queue is full, please try again later."
        )
    
    # Check if we can process immediately
    if current_requests < MAX_CONCURRENT:
        current_requests += 1
        try:
            yield
        finally:
            current_requests -= 1
    else:
        # Add to queue
        queued_request = QueuedRequest(
            request=request,
            response=response,
            timestamp=time.time() * 1000,
            ready_event=asyncio.Event()
        )
        request_queue.append(queued_request)
        
        # Wait for request to be ready or timeout
        await queued_request.ready_event.wait()
        
        # Check if request timed out
        if queued_request.timed_out:
            raise HTTPException(
                status_code=408,
                detail="Request timeout while waiting in queue"
            )
        
        yield

# Cleanup function
async def cleanup_throttler():
    """Clean up the throttler on shutdown"""
    throttler.running = False
    if throttler.queue_task:
        throttler.queue_task.cancel()
        try:
            await throttler.queue_task
        except asyncio.CancelledError:
            pass
        throttler.queue_task = None
Using the Throttling System

To integrate this with FastAPI routes, you can use it as a context manager:

# Usage in FastAPI route
@router.get("/api/data")
async def get_data(request: Request, response: Response):
    async with queue_throttle(request, response):
        # Your API logic here
        await asyncio.sleep(1)  # Simulate processing
        return {"message": "Data processed successfully"}
Testing the Implementation

When testing this implementation, we should observe specific patterns:

# Key points from testing with 15 concurrent requests
# 1. First 3 requests (MAX_CONCURRENT) complete quickly
# 2. Next 10 requests (MAX_QUEUE_SIZE) are queued and complete with increasing delays
# 3. Remaining 2 requests immediately receive 503 errors (queue full)

The staggered completion times confirm that requests are being queued and processed in order, rather than all being processed simultaneously or immediately rejected.

Real-World Applications and Considerations

Queue-based throttling works well for:

  • APIs with varying processing times: When some requests take longer than others
  • Systems requiring fairness: Where you want to ensure first-come, first-served processing
  • Services with spiky traffic patterns: Where occasional bursts should be handled gracefully

FastAPI and asyncio implementation challenges to consider:

  • Memory management: In high-volume FastAPI systems, the queue size must be carefully monitored
  • Distributed systems: Using Redis or a similar service for centralized queue management across FastAPI instances
  • Request prioritization: Consider adding priority levels to allow critical requests to skip the queue in your FastAPI routes
  • Client timeouts: Ensure queue timeouts are shorter than typical client-side timeouts for FastAPI endpoints
  • AsyncIO context management: FastAPI's async/await patterns require careful handling of request lifecycle and asyncio task management
Summary

Queue-based throttling provides a balanced approach to FastAPI request management, allowing your FastAPI server with asyncio to maintain optimal performance under variable load conditions. By queueing excess requests rather than rejecting them outright, you improve user experience while still protecting your system from overload. The FastAPI implementation using Python's asyncio requires careful consideration of queue size, processing intervals, timeout handling, and async context management, but the benefits of improved resilience and fairness make it worthwhile for many FastAPI applications.

Previous
Next
Sign up
Join the 1M+ learners on CodeSignal
Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal