Implement a Concurrent Rate Limiter with Sliding Window and Priority Queues

```python import time import threading import collections import heapq import uuid from dataclasses import dataclass, field from typing import Dict, List, Tuple, Any, Optional # --- Configuration --- # Define tiers with their respective rate limits and window sizes. # Lower window_seconds means tighter rate limiting. TIER_CONFIG = { "free": {"max_requests": 5, "window_seconds": 60}, "standard": {"max_requests": 50, "window_seconds": 60}, "premium": {"max_requests": 200, "window_seconds": 60}, } # How long to keep client data before cleaning up if no activity. CLEANUP_THRESHOLD_SECONDS = 300 # --- Data Structures --- @dataclass(order=True) class Request: """Represents a request with priority, for the priority queue.""" # We use priority as the first element for heapq to sort by it. # Lower number means higher priority. priority: int # Use a unique ID to break ties in priority and ensure stable sorting. request_id: str = field(default_factory=lambda: str(uuid.uuid4())) client_id: str timestamp: float payload: Any def __lt__(self, other): # Custom comparison for heapq: prioritize by priority, then by request_id for stability. if self.priority != other.priority: return self.priority < other.priority return self.request_id < other.request_id class SlidingWindowCounter: """Implements a sliding window counter for rate limiting.""" def __init__(self, max_requests: int, window_seconds: float): self.max_requests = max_requests self.window_seconds = window_seconds # Use a deque to store timestamps of requests within the window. # Deques offer efficient appends and pops from both ends. self.request_timestamps: collections.deque[float] = collections.deque() # Lock for thread-safe access to request_timestamps. self.lock = threading.Lock() def record_request(self) -> bool: """Records a request and returns True if it's within the limit, False otherwise.""" with self.lock: current_time = time.time() # Remove timestamps older than the window. while self.request_timestamps and self.request_timestamps[0] <= current_time - self.window_seconds: self.request_timestamps.popleft() # Check if adding a new request exceeds the limit. if len(self.request_timestamps) < self.max_requests: self.request_timestamps.append(current_time) return True else: return False def get_current_count(self) -> int: """Returns the current number of requests within the window.""" with self.lock: current_time = time.time() # Clean up old timestamps before returning count. while self.request_timestamps and self.request_timestamps[0] <= current_time - self.window_seconds: self.request_timestamps.popleft() return len(self.request_timestamps) class RateLimiter: """A thread-safe rate limiter with multiple tiers, sliding window, and priority queues.""" def __init__(self, tier_config: Dict[str, Dict[str, Any]], cleanup_threshold_seconds: float): self.tier_config = tier_config self.cleanup_threshold_seconds = cleanup_threshold_seconds # Stores the sliding window counters for each tier. self.tier_counters: Dict[str, SlidingWindowCounter] = {} # Stores the priority queue for deferred requests for each tier. self.tier_queues: Dict[str, List[Request]] = collections.defaultdict(list) # Stores the last activity timestamp for each client for cleanup. self.client_last_activity: Dict[str, float] = {} # Stores the tier assigned to each client. self.client_tiers: Dict[str, str] = {} # Main lock to protect shared data structures like client_tiers, client_last_activity, and tier_queues. # The SlidingWindowCounter has its own internal lock. self.global_lock = threading.Lock() # Initialize tier counters. for tier_name, config in tier_config.items(): self.tier_counters[tier_name] = SlidingWindowCounter( max_requests=config["max_requests"], window_seconds=config["window_seconds"] ) # Start a background thread for cleanup. self.cleanup_thread = threading.Thread(target=self._cleanup_task, daemon=True) self.cleanup_thread.start() def register_client(self, client_id: str, tier: str) -> bool: """Registers a client with a specific tier. Returns True if successful, False if client already exists or tier is invalid.""" if tier not in self.tier_config: print(f"[!] Error: Tier '{tier}' does not exist.") return False with self.global_lock: if client_id in self.client_tiers: print(f"[!] Warning: Client '{client_id}' already registered.") return False self.client_tiers[client_id] = tier self.client_last_activity[client_id] = time.time() print(f"[*] Client '{client_id}' registered with tier '{tier}'.") return True def _get_client_tier(self, client_id: str) -> Optional[str]: """Safely retrieves the tier for a given client_id.""" with self.global_lock: return self.client_tiers.get(client_id) def _update_client_activity(self, client_id: str): """Updates the last activity timestamp for a client.""" with self.global_lock: self.client_last_activity[client_id] = time.time() def allow_request(self, client_id: str) -> bool: """Checks if a request from client_id is allowed based on their tier's rate limit.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Error: Client '{client_id}' not registered.") return False counter = self.tier_counters.get(tier) if not counter: # This should not happen if tier_config is valid and initialization is correct. print(f"[!] Internal Error: Counter not found for tier '{tier}'.") return False if counter.record_request(): self._update_client_activity(client_id) # print(f"[*] Request ALLOWED for client '{client_id}' (Tier: {tier}). Current count: {counter.get_current_count()}") return True else: # print(f"[*] Request DENIED for client '{client_id}' (Tier: {tier}). Current count: {counter.get_current_count()}") return False def enqueue_request(self, client_id: str, priority: int, payload: Any): """Enqueues a rate-limited request into the appropriate priority queue.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Error: Client '{client_id}' not registered. Cannot enqueue.") return request = Request(priority=priority, client_id=client_id, timestamp=time.time(), payload=payload) with self.global_lock: heapq.heappush(self.tier_queues[tier], request) print(f"[*] Request ENQUEUED for client '{client_id}' (Tier: {tier}, Priority: {priority}). Queue size: {len(self.tier_queues[tier])}") self._update_client_activity(client_id) def dequeue_and_process(self, client_id: str) -> Optional[Tuple[Request, bool]]: """Attempts to dequeue and process the highest priority request for a client's tier if capacity is available.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Error: Client '{client_id}' not registered. Cannot dequeue.") return None counter = self.tier_counters.get(tier) if not counter: print(f"[!] Internal Error: Counter not found for tier '{tier}'.") return None # Check if capacity is available BEFORE trying to dequeue. # This prevents unnecessary locking and heap operations if no capacity exists. if counter.get_current_count() < counter.max_requests: with self.global_lock: queue = self.tier_queues[tier] if queue: # Re-check capacity inside the lock to handle race conditions. # It's possible another thread just filled the capacity. if counter.get_current_count() < counter.max_requests: highest_priority_request = heapq.heappop(queue) # Attempt to record the request again. This should succeed if capacity is available. if counter.record_request(): self._update_client_activity(highest_priority_request.client_id) print(f"[*] Request DEQUEUED and PROCESSED for client '{highest_priority_request.client_id}' (Tier: {tier}, Priority: {highest_priority_request.priority}). Queue size: {len(queue)}") return highest_priority_request, True else: # This is a rare race condition: capacity was available, but record_request failed. # Push it back and log an error. heapq.heappush(queue, highest_priority_request) print(f"[!] Race condition: Failed to record dequeued request for {highest_priority_request.client_id}. Pushed back.") return highest_priority_request, False else: # Capacity was taken by another thread between the outer check and this inner check. # Push the request back if it was popped. heapq.heappush(queue, highest_priority_request) # print(f"[*] Dequeue attempt failed for {client_id}: Capacity filled by another thread.") return highest_priority_request, False else: # Queue is empty. # print(f"[*] Dequeue attempt for {client_id}: Queue is empty.") return None, False else: # No capacity available. # print(f"[*] Dequeue attempt for {client_id}: No capacity available.") return None, False def _cleanup_task(self): """Background task to remove inactive client data.""" while True: time.sleep(self.cleanup_threshold_seconds / 2) # Check periodically current_time = time.time() clients_to_remove = [] with self.global_lock: for client_id, last_activity in self.client_last_activity.items(): if current_time - last_activity > self.cleanup_threshold_seconds: clients_to_remove.append(client_id) for client_id in clients_to_remove: tier = self.client_tiers.pop(client_id, None) if tier: self.client_last_activity.pop(client_id, None) # Optionally, clear queues for this client if needed, though typically not done. # For simplicity, we leave enqueued requests as they are. print(f"[*] Cleaned up inactive client '{client_id}'.") # --- Demonstration Script --- def simulate_requests(rate_limiter: RateLimiter, client_id: str, num_requests: int, delay: float, priority_offset: int = 0): """Simulates a burst of requests from a single client.""" print(f"--- Simulating {num_requests} requests from client '{client_id}' ---") for i in range(num_requests): # Simulate varying priorities for enqueued requests priority = i % 5 + priority_offset # Lower number = higher priority payload = f"data_{client_id}_{i}" if rate_limiter.allow_request(client_id): print(f"[Thread {threading.current_thread().name}] Request {i+1}/{num_requests} ALLOWED for '{client_id}'. Payload: {payload}") else: print(f"[Thread {threading.current_thread().name}] Request {i+1}/{num_requests} DENIED for '{client_id}'. Enqueuing with priority {priority}.") rate_limiter.enqueue_request(client_id, priority, payload) time.sleep(delay) def process_deferred_requests(rate_limiter: RateLimiter, client_id: str, num_attempts: int): """Periodically tries to process deferred requests for a client.""" print(f"--- Attempting to process deferred requests for client '{client_id}' ---") for i in range(num_attempts): time.sleep(1.5) # Wait a bit for capacity to potentially free up print(f"[Processor Thread] Attempt {i+1}/{num_attempts} to dequeue for '{client_id}'...") result = rate_limiter.dequeue_and_process(client_id) if result: request, success = result if success: print(f"[Processor Thread] Successfully processed: {request.payload} (Priority: {request.priority})") else: print(f"[Processor Thread] Failed to process (capacity issue?): {request.payload} (Priority: {request.priority}). Re-enqueued.") else: print(f"[Processor Thread] No deferred requests processed for '{client_id}' this attempt (queue empty or no capacity).") if __name__ == "__main__": print("--- Rate Limiter Demonstration ---") # Initialize the rate limiter limiter = RateLimiter(tier_config=TIER_CONFIG, cleanup_threshold_seconds=CLEANUP_THRESHOLD_SECONDS) # Register clients clients = { "client_A": "free", "client_B": "standard", "client_C": "premium", "client_D": "free", "client_E": "standard", } for cid, tier in clients.items(): limiter.register_client(cid, tier) print("\n--- Simulating concurrent requests ---") # Simulate a burst of requests from multiple threads threads = [] # Client A (free tier) - likely to hit rate limit t1 = threading.Thread(target=simulate_requests, args=(limiter, "client_A", 15, 0.1, 0), name="Thread-A") threads.append(t1) # Client B (standard tier) - more capacity t2 = threading.Thread(target=simulate_requests, args=(limiter, "client_B", 60, 0.05, 1), name="Thread-B") threads.append(t2) # Client C (premium tier) - highest capacity t3 = threading.Thread(target=simulate_requests, args=(limiter, "client_C", 250, 0.01, 2), name="Thread-C") threads.append(t3) # Client D (free tier) - another client hitting the same limit t4 = threading.Thread(target=simulate_requests, args=(limiter, "client_D", 10, 0.15, 3), name="Thread-D") threads.append(t4) # Start simulation threads for t in threads: t.start() # Give some time for initial requests and enqueuing time.sleep(2) print("\n--- Attempting to process deferred requests ---") # Start threads to process deferred requests for clients that might have been rate-limited # We'll try processing for clients A and D (free tier) multiple times. processor_threads = [] tp1 = threading.Thread(target=process_deferred_requests, args=(limiter, "client_A", 5), name="Processor-A") processor_threads.append(tp1) tp2 = threading.Thread(target=process_deferred_requests, args=(limiter, "client_D", 5), name="Processor-D") processor_threads.append(tp2) # Also try for client B, just in case tp3 = threading.Thread(target=process_deferred_requests, args=(limiter, "client_B", 3), name="Processor-B") processor_threads.append(tp3) for tp in processor_threads: tp.start() # Wait for all simulation threads to complete for t in threads: t.join() # Wait for processor threads to complete their attempts for tp in processor_threads: tp.join() print("\n--- Demonstration Complete ---") print("Final state check (counts are approximate due to timing):") for tier_name, counter in limiter.tier_counters.items(): print(f"Tier '{tier_name}': Current requests in window: {counter.get_current_count()} (Max: {counter.max_requests})") for tier_name, queue in limiter.tier_queues.items(): print(f"Tier '{tier_name}': Deferred requests in queue: {len(queue)}") # --- Design Choices Explanation --- # 1. Sliding Window Implementation: # - We use `collections.deque` to store timestamps of requests within the window. # - `deque` provides O(1) complexity for appending new timestamps and popping old ones from the left. # - When `record_request` is called, we first remove all timestamps older than `current_time - window_seconds`. # - This ensures that `len(self.request_timestamps)` accurately reflects the number of requests within the current sliding window. # - Trade-off: This approach is precise but can involve multiple pops in `record_request` if many requests arrive in quick succession and then stop. However, over time, the deque size is managed. # - Thread Safety: Each `SlidingWindowCounter` instance has its own `threading.Lock` to protect its `request_timestamps` deque, ensuring that concurrent calls to `record_request` or `get_current_count` for the same tier are serialized. # 2. Multiple Tiers: # - The `RateLimiter` class maintains a dictionary `tier_config` mapping tier names to their rate limit parameters (`max_requests`, `window_seconds`). # - It also maintains separate `SlidingWindowCounter` instances and priority queues (`tier_queues`) for each tier. # - Client tier assignment is stored in `client_tiers`. # - This allows for flexible configuration and isolation of rate limits between different client types. # 3. Priority Queue for Deferred Requests: # - For each tier, a standard Python list is used as a min-heap, managed by the `heapq` module. # - The `Request` dataclass is designed with `priority` as the primary sorting key (lower value = higher priority). # - A unique `request_id` is included to ensure stable sorting (FIFO for same priorities) and prevent comparison errors if timestamps are identical. # - `enqueue_request` uses `heapq.heappush` to add requests, maintaining the heap property. # - `dequeue_and_process` uses `heapq.heappop` to retrieve the highest priority request. # - Crucially, `dequeue_and_process` first checks available capacity *before* attempting to pop from the heap. If capacity is available, it then acquires the global lock, re-checks capacity (to handle race conditions), pops the request, and attempts to `record_request` again. If `record_request` succeeds, the request is processed. If it fails (due to a race condition where capacity was just filled), the request is pushed back onto the heap. # - Trade-off: Using Python's list + heapq is efficient for moderate numbers of deferred requests. For extremely high volumes, a more specialized concurrent priority queue implementation might be considered, but this adds significant complexity. # 4. Thread Safety: # - `SlidingWindowCounter`: Uses an internal `threading.Lock` for its `request_timestamps` deque. # - `RateLimiter`: # - `global_lock` (`threading.Lock`): Protects shared client state (`client_tiers`, `client_last_activity`) and the tier priority queues (`tier_queues`). This lock is acquired for operations like `register_client`, `enqueue_request`, and within `dequeue_and_process` when accessing/modifying the queues. # - `allow_request`: Calls the thread-safe `counter.record_request()`. It also updates client activity under the global lock. # - `dequeue_and_process`: Carefully manages locks. It first checks capacity *without* the global lock. If capacity *might* be available, it acquires the global lock, re-checks capacity, performs the heap operation, and then calls the thread-safe `counter.record_request()` again. # - Trade-off: The use of multiple locks (one per counter + one global lock) aims to reduce contention compared to a single global lock for all operations. However, operations that span client state and tier state (like dequeueing) require careful lock ordering or acquisition of the global lock. # 5. Cleanup: # - A background thread (`_cleanup_task`) periodically scans `client_last_activity`. # - It identifies clients whose last activity timestamp is older than `cleanup_threshold_seconds`. # - These clients are removed from `client_tiers` and `client_last_activity` under the `global_lock`. # - This prevents the `RateLimiter` from accumulating indefinitely large amounts of state for inactive clients. # - Trade-off: The cleanup is not instantaneous; it runs periodically. The `cleanup_threshold_seconds` determines the trade-off between memory usage and responsiveness of cleanup. # 6. Edge Cases Handled: # - Duplicate client registration: `register_client` returns False and logs a warning. # - Unregistered clients: `allow_request` and `enqueue_request` check `client_tiers` and return/log errors. # - Empty priority queues: `dequeue_and_process` handles cases where `self.tier_queues[tier]` is empty. # - Concurrent modifications: Addressed by using `threading.Lock` appropriately. # - Clock precision: `time.time()` is used, which is generally sufficient. For extremely high-precision requirements, platform-specific monotonic clocks might be considered, but `time.time()` is standard for this type of application. # - Race conditions in dequeue: Handled by re-checking capacity and re-recording the request within the critical section of `dequeue_and_process`. ```

import threading import time import heapq from collections import deque from dataclasses import dataclass from typing import Any, Dict, Optional, Tuple import itertools import random # Request dataclass holds minimal required fields @dataclass(order=True) class Request: # ordering will be driven primarily by the priority in heaps priority: int client_id: str timestamp: float payload: Any = None class RateLimiter: """ Thread-safe rate limiter with sliding window per client and per-tier priority queues. Design choices (high-level): - Sliding window implemented per client using a deque of request timestamps. This gives exact counts for the sliding window (precision) at the cost of storing one timestamp per request. - Per-tier priority queues implemented with heapq. Each queued Request is pushed as (priority, counter, request) to preserve FIFO among same-priority items. Lower priority number = higher priority. - Concurrency: use a combination of locks: * self.clients_lock protects the client registry structure. * each client has its own lock to protect its timestamp deque and last_seen metadata (reduces contention). * each tier has its own lock to protect its priority queue. Lock ordering is important to avoid deadlocks: when both client lock and tier lock need to be acquired, the implementation always acquires the client lock first, then the tier lock. All methods follow this order. - For dequeue_for_client, we must find the highest-priority waiting request for that client in the per-tier heap. Because heapq does not support efficient arbitrary-item removal, we pop items until we find one for the client, buffering others and pushing them back afterwards. This can be O(n) in the queue size in the worst case — a trade-off favoring simpler data structures and correctness over per-operation worst-case performance. - Time functions use time.monotonic() for durations to avoid system clock jumps affecting rate limiting. Trade-offs: - Precision vs space: deque per client is precise for sliding window but stores each request timestamp. - Performance vs complexity: scanning heaps for dequeue_for_client is simple but can be costly if queues are large. """ def __init__(self, tiers: Dict[str, Dict[str, float]]): """ tiers: mapping tier_name -> { 'max_requests': int, 'window_seconds': float } """ self.tiers = {} self.tiers_lock = threading.Lock() self._counter = itertools.count() # Initialize tier structures for name, conf in tiers.items(): if 'max_requests' not in conf or 'window_seconds' not in conf: raise ValueError(f"Tier {name} missing configuration") self.tiers[name] = { 'max_requests': int(conf['max_requests']), 'window_seconds': float(conf['window_seconds']), 'queue': [], # heap: elements (priority, counter, Request) 'lock': threading.Lock(), } # Clients registry: client_id -> { 'tier': str, 'timestamps': deque, 'lock': Lock, 'last_seen': float } self.clients: Dict[str, Dict[str, Any]] = {} self.clients_lock = threading.Lock() def register_client(self, client_id: str, tier_name: str): """Register a client into a tier. Raises ValueError for duplicate registration or unknown tier. Thread-safe. """ with self.clients_lock: if client_id in self.clients: raise ValueError(f"Client {client_id} already registered") if tier_name not in self.tiers: raise ValueError(f"Unknown tier {tier_name}") self.clients[client_id] = { 'tier': tier_name, 'timestamps': deque(), 'lock': threading.Lock(), 'last_seen': time.monotonic(), } def _remove_old_timestamps(self, client_entry: Dict[str, Any], window_seconds: float, now: float): # remove timestamps older than sliding window timestamps: deque = client_entry['timestamps'] cutoff = now - window_seconds while timestamps and timestamps[0] <= cutoff: timestamps.popleft() def allow_request(self, client_id: str, priority: int = 100, payload: Any = None, timestamp: Optional[float] = None) -> Tuple[bool, Optional[Request]]: """ Attempt to allow a request for client_id. If under the sliding window limit, the request is recorded and allowed (returns (True, None)). If the client is at capacity, the request is enqueued into the client's tier priority queue and (False, request) is returned. Raises KeyError if client is not registered. Thread-safe. """ if timestamp is None: now = time.monotonic() else: # If caller supplied an external timestamp, convert it to monotonic-relative by using monotonic() now # and assuming external timestamp is in same epoch; for demonstration we accept timestamp as-is. now = timestamp # Get client entry with self.clients_lock: client_entry = self.clients.get(client_id) if client_entry is None: raise KeyError(f"Client {client_id} is not registered") tier_name = client_entry['tier'] tier = self.tiers[tier_name] # Acquire client lock first (lock ordering rule), then possibly tier lock. with client_entry['lock']: self._remove_old_timestamps(client_entry, tier['window_seconds'], now) if len(client_entry['timestamps']) < tier['max_requests']: # allow request client_entry['timestamps'].append(now) client_entry['last_seen'] = now return True, None else: # at capacity -> enqueue into tier priority queue req = Request(priority=priority, client_id=client_id, timestamp=now, payload=payload) # Acquire tier lock to push into heap with tier['lock']: count = next(self._counter) heapq.heappush(tier['queue'], (req.priority, count, req)) client_entry['last_seen'] = now return False, req def enqueue_request(self, req: Request): """Directly enqueue a Request into its client's tier queue. Raises if client not registered. Thread-safe. """ with self.clients_lock: client_entry = self.clients.get(req.client_id) if client_entry is None: raise KeyError(f"Client {req.client_id} is not registered") tier_name = client_entry['tier'] tier = self.tiers[tier_name] with tier['lock']: count = next(self._counter) heapq.heappush(tier['queue'], (req.priority, count, req)) def dequeue_for_client(self, client_id: str, timestamp: Optional[float] = None) -> Optional[Request]: """ If capacity is available for client_id, dequeue and return the highest-priority queued Request for the client's tier. The returned Request is considered processed and will be counted toward the window. If no queued request is present for the client or capacity isn't available, return None. Raises KeyError if client not registered. Thread-safe. """ if timestamp is None: now = time.monotonic() else: now = timestamp with self.clients_lock: client_entry = self.clients.get(client_id) if client_entry is None: raise KeyError(f"Client {client_id} is not registered") tier_name = client_entry['tier'] tier = self.tiers[tier_name] # Acquire client lock first, then tier lock (consistent ordering) with client_entry['lock']: self._remove_old_timestamps(client_entry, tier['window_seconds'], now) if len(client_entry['timestamps']) >= tier['max_requests']: # no capacity return None # There is capacity; try to find highest-priority queued request for this client with tier['lock']: queue = tier['queue'] if not queue: return None temp = [] found_req = None try: while queue: priority, count, req = heapq.heappop(queue) if req.client_id == client_id: found_req = req break else: temp.append((priority, count, req)) finally: # push back any popped non-matching requests for item in temp: heapq.heappush(queue, item) if found_req is None: return None # mark the request as processed: append timestamp to client's timestamps client_entry['timestamps'].append(now) client_entry['last_seen'] = now return found_req def cleanup(self, cleanup_threshold_seconds: float): """ Remove tracking data for clients who have not made requests in the last cleanup_threshold_seconds. Does not unregister the client (keeps registration), but removes timestamp history to free memory. If a client has queued deferred requests, we keep their tracking data. Thread-safe. """ now = time.monotonic() cutoff = now - cleanup_threshold_seconds removed = [] # iterate clients safely with self.clients_lock: for client_id, client_entry in list(self.clients.items()): # acquire client lock to inspect safely with client_entry['lock']: # if the client has no activity recently and no queued requests, clear timestamps last_seen = client_entry.get('last_seen', 0) tier_name = client_entry['tier'] tier = self.tiers[tier_name] has_queued = False # check if any queued requests exist for this client in tier queue with tier['lock']: for _, _, req in tier['queue']: if req.client_id == client_id: has_queued = True break if last_seen < cutoff and not has_queued: # clear timestamps to free memory client_entry['timestamps'].clear() removed.append(client_id) return removed # Demonstration script if __name__ == "__main__": # Create a rate limiter with two tiers: free and premium tiers_conf = { 'free': {'max_requests': 3, 'window_seconds': 5.0}, 'premium': {'max_requests': 10, 'window_seconds': 5.0}, } rl = RateLimiter(tiers_conf) # Register clients clients = [ ('alice', 'free'), ('bob', 'free'), ('carol', 'premium'), ] for cid, tier in clients: rl.register_client(cid, tier) print(f"Registered client {cid} as tier {tier}") stop_event = threading.Event() # Worker threads simulate bursts of requests def client_worker(client_id: str, burst: int, sleep_between: float): for i in range(burst): allowed, req = rl.allow_request(client_id, priority=random.randint(1, 10), payload={'i': i}) ts = time.monotonic() if allowed: print(f"[{ts:.3f}] {client_id}: request {i} ALLOWED") else: print(f"[{ts:.3f}] {client_id}: request {i} ENQUEUED (priority={req.priority})") time.sleep(sleep_between) # Background processor that periodically tries to dequeue queued requests for all clients def deferred_processor(): while not stop_event.is_set(): # iterate all clients and try to dequeue one request if capacity exists with rl.clients_lock: client_ids = list(rl.clients.keys()) progressed = False for cid in client_ids: try: req = rl.dequeue_for_client(cid) except KeyError: continue if req is not None: ts = time.monotonic() print(f"[{ts:.3f}] Processed deferred request for {cid} (priority={req.priority}, payload={req.payload})") progressed = True if not progressed: # nothing processed; sleep a bit time.sleep(0.2) # Start threads: bursty clients threads = [] t1 = threading.Thread(target=client_worker, args=('alice', 8, 0.4)) t2 = threading.Thread(target=client_worker, args=('bob', 6, 0.3)) t3 = threading.Thread(target=client_worker, args=('carol', 12, 0.1)) threads.extend([t1, t2, t3]) processor_thread = threading.Thread(target=deferred_processor) processor_thread.daemon = True processor_thread.start() for t in threads: t.start() # Let threads finish for t in threads: t.join() # Allow background processor to handle remaining queued items for some time time.sleep(3.0) stop_event.set() processor_thread.join(timeout=1.0) # Show cleanup removed = rl.cleanup(cleanup_threshold_seconds=1.0) print(f"Cleanup removed timestamp history for clients: {removed}") # Show queued items left in each tier for tier_name, tier in rl.tiers.items(): with tier['lock']: qsize = len(tier['queue']) print(f"Tier '{tier_name}' has {qsize} queued requests left") print("Demo complete")