Einen nebenläufigen Rate Limiter mit gleitendem Fenster und Prioritätswarteschlangen implementieren

Antwort: ```python import time import threading import collections import heapq import uuid from dataclasses import dataclass, field from typing import Dict, List, Tuple, Any, Optional # --- Konfiguration --- # Definiere Tiers mit ihren jeweiligen Ratenbegrenzungen und Fenstergrößen. # Ein kleineres window_seconds bedeutet eine strengere Ratenbegrenzung. TIER_CONFIG = { "free": {"max_requests": 5, "window_seconds": 60}, "standard": {"max_requests": 50, "window_seconds": 60}, "premium": {"max_requests": 200, "window_seconds": 60}, } # Wie lange Client-Daten ohne Aktivität aufbewahrt werden sollen, bevor sie bereinigt werden. CLEANUP_THRESHOLD_SECONDS = 300 # --- Datenstrukturen --- @dataclass(order=True) class Request: """Repräsentiert eine Anfrage mit Priorität für die Prioritätswarteschlange.""" # Wir verwenden priority als erstes Element, damit heapq danach sortiert. # Eine kleinere Zahl bedeutet eine höhere Priorität. priority: int # Verwende eine eindeutige ID, um Gleichstände bei der Priorität aufzulösen und stabiles Sortieren sicherzustellen. request_id: str = field(default_factory=lambda: str(uuid.uuid4())) client_id: str timestamp: float payload: Any def __lt__(self, other): # Benutzerdefinierter Vergleich für heapq: priorisiere nach priority, dann nach request_id für Stabilität. if self.priority != other.priority: return self.priority < other.priority return self.request_id < other.request_id class SlidingWindowCounter: """Implementiert einen Sliding-Window-Zähler zur Ratenbegrenzung.""" def __init__(self, max_requests: int, window_seconds: float): self.max_requests = max_requests self.window_seconds = window_seconds # Verwende eine deque, um Zeitstempel von Anfragen innerhalb des Fensters zu speichern. # Deques bieten effiziente appends und pops an beiden Enden. self.request_timestamps: collections.deque[float] = collections.deque() # Lock für threadsicheren Zugriff auf request_timestamps. self.lock = threading.Lock() def record_request(self) -> bool: """Erfasst eine Anfrage und gibt True zurück, wenn sie innerhalb des Limits liegt, andernfalls False.""" with self.lock: current_time = time.time() # Entferne Zeitstempel, die älter als das Fenster sind. while self.request_timestamps and self.request_timestamps[0] <= current_time - self.window_seconds: self.request_timestamps.popleft() # Prüfe, ob das Hinzufügen einer neuen Anfrage das Limit überschreitet. 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: """Gibt die aktuelle Anzahl von Anfragen innerhalb des Fensters zurück.""" with self.lock: current_time = time.time() # Bereinige alte Zeitstempel, bevor der Zählerstand zurückgegeben wird. 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: """Ein threadsicherer Rate-Limiter mit mehreren Tiers, Sliding Window und Prioritätswarteschlangen.""" 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 # Speichert die Sliding-Window-Zähler für jedes Tier. self.tier_counters: Dict[str, SlidingWindowCounter] = {} # Speichert die Prioritätswarteschlange für aufgeschobene Anfragen für jedes Tier. self.tier_queues: Dict[str, List[Request]] = collections.defaultdict(list) # Speichert den Zeitstempel der letzten Aktivität für jeden Client zur Bereinigung. self.client_last_activity: Dict[str, float] = {} # Speichert das jedem Client zugewiesene Tier. self.client_tiers: Dict[str, str] = {} # Haupt-Lock zum Schutz gemeinsam genutzter Datenstrukturen wie client_tiers, client_last_activity und tier_queues. # Der SlidingWindowCounter hat seinen eigenen internen Lock. self.global_lock = threading.Lock() # Initialisiere Tier-Zähler. for tier_name, config in tier_config.items(): self.tier_counters[tier_name] = SlidingWindowCounter( max_requests=config["max_requests"], window_seconds=config["window_seconds"] ) # Starte einen Hintergrund-Thread für die Bereinigung. 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: """Registriert einen Client mit einem bestimmten Tier. Gibt True zurück, wenn erfolgreich, andernfalls False, wenn der Client bereits existiert oder das Tier ungültig ist.""" if tier not in self.tier_config: print(f"[!] Fehler: Tier '{tier}' existiert nicht.") return False with self.global_lock: if client_id in self.client_tiers: print(f"[!] Warnung: Client '{client_id}' ist bereits registriert.") return False self.client_tiers[client_id] = tier self.client_last_activity[client_id] = time.time() print(f"[*] Client '{client_id}' mit Tier '{tier}' registriert.") return True def _get_client_tier(self, client_id: str) -> Optional[str]: """Ruft sicher das Tier für eine gegebene client_id ab.""" with self.global_lock: return self.client_tiers.get(client_id) def _update_client_activity(self, client_id: str): """Aktualisiert den Zeitstempel der letzten Aktivität eines Clients.""" with self.global_lock: self.client_last_activity[client_id] = time.time() def allow_request(self, client_id: str) -> bool: """Prüft, ob eine Anfrage von client_id basierend auf der Ratenbegrenzung ihres Tiers erlaubt ist.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Fehler: Client '{client_id}' ist nicht registriert.") return False counter = self.tier_counters.get(tier) if not counter: # Dies sollte nicht passieren, wenn tier_config gültig ist und die Initialisierung korrekt war. print(f"[!] Interner Fehler: Kein Zähler für Tier '{tier}' gefunden.") return False if counter.record_request(): self._update_client_activity(client_id) # print(f"[*] Anfrage ERLAUBT für Client '{client_id}' (Tier: {tier}). Aktuelle Anzahl: {counter.get_current_count()}") return True else: # print(f"[*] Anfrage ABGELEHNT für Client '{client_id}' (Tier: {tier}). Aktuelle Anzahl: {counter.get_current_count()}") return False def enqueue_request(self, client_id: str, priority: int, payload: Any): """Stellt eine rate-limitierte Anfrage in die passende Prioritätswarteschlange ein.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Fehler: Client '{client_id}' ist nicht registriert. Einreihen nicht möglich.") 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"[*] Anfrage EINGEREIHT für Client '{client_id}' (Tier: {tier}, Priorität: {priority}). Warteschlangengröße: {len(self.tier_queues[tier])}") self._update_client_activity(client_id) def dequeue_and_process(self, client_id: str) -> Optional[Tuple[Request, bool]]: """Versucht, die Anfrage mit der höchsten Priorität für das Tier eines Clients aus der Warteschlange zu nehmen und zu verarbeiten, falls Kapazität verfügbar ist.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Fehler: Client '{client_id}' ist nicht registriert. Ausreihen nicht möglich.") return None counter = self.tier_counters.get(tier) if not counter: print(f"[!] Interner Fehler: Kein Zähler für Tier '{tier}' gefunden.") return None # Prüfe, ob Kapazität verfügbar ist, BEVOR versucht wird, aus der Warteschlange zu nehmen. # Das verhindert unnötiges Locking und Heap-Operationen, wenn keine Kapazität vorhanden ist. if counter.get_current_count() < counter.max_requests: with self.global_lock: queue = self.tier_queues[tier] if queue: # Prüfe die Kapazität innerhalb des Locks erneut, um Race Conditions zu behandeln. # Es ist möglich, dass ein anderer Thread die Kapazität gerade gefüllt hat. if counter.get_current_count() < counter.max_requests: highest_priority_request = heapq.heappop(queue) # Versuche erneut, die Anfrage zu erfassen. Das sollte gelingen, wenn Kapazität verfügbar ist. if counter.record_request(): self._update_client_activity(highest_priority_request.client_id) print(f"[*] Anfrage AUSGEREIHT und VERARBEITET für Client '{highest_priority_request.client_id}' (Tier: {tier}, Priorität: {highest_priority_request.priority}). Warteschlangengröße: {len(queue)}") return highest_priority_request, True else: # Dies ist eine seltene Race Condition: Kapazität war verfügbar, aber record_request ist fehlgeschlagen. # Lege die Anfrage zurück und protokolliere einen Fehler. heapq.heappush(queue, highest_priority_request) print(f"[!] Race Condition: Ausgereihte Anfrage für {highest_priority_request.client_id} konnte nicht erfasst werden. Zurückgelegt.") return highest_priority_request, False else: # Die Kapazität wurde von einem anderen Thread zwischen der äußeren und dieser inneren Prüfung verbraucht. # Lege die Anfrage zurück, falls sie entfernt wurde. heapq.heappush(queue, highest_priority_request) # print(f"[*] Ausreihversuch fehlgeschlagen für {client_id}: Kapazität von einem anderen Thread gefüllt.") return highest_priority_request, False else: # Warteschlange ist leer. # print(f"[*] Ausreihversuch für {client_id}: Warteschlange ist leer.") return None, False else: # Keine Kapazität verfügbar. # print(f"[*] Ausreihversuch für {client_id}: Keine Kapazität verfügbar.") return None, False def _cleanup_task(self): """Hintergrundaufgabe zum Entfernen inaktiver Client-Daten.""" while True: time.sleep(self.cleanup_threshold_seconds / 2) # Regelmäßig prüfen 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) # Optional könnten Warteschlangen für diesen Client geleert werden, falls nötig, was jedoch typischerweise nicht gemacht wird. # Der Einfachheit halber belassen wir eingereihte Anfragen wie sie sind. print(f"[*] Inaktiver Client '{client_id}' wurde bereinigt.") # --- Demonstrationsskript --- def simulate_requests(rate_limiter: RateLimiter, client_id: str, num_requests: int, delay: float, priority_offset: int = 0): """Simuliert einen Burst von Anfragen von einem einzelnen Client.""" print(f"--- Simuliere {num_requests} Anfragen von Client '{client_id}' ---") for i in range(num_requests): # Simuliere unterschiedliche Prioritäten für eingereihte Anfragen priority = i % 5 + priority_offset # Kleinere Zahl = höhere Priorität payload = f"data_{client_id}_{i}" if rate_limiter.allow_request(client_id): print(f"[Thread {threading.current_thread().name}] Anfrage {i+1}/{num_requests} ERLAUBT für '{client_id}'. Payload: {payload}") else: print(f"[Thread {threading.current_thread().name}] Anfrage {i+1}/{num_requests} ABGELEHNT für '{client_id}'. Einreihen mit Priorität {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): """Versucht periodisch, aufgeschobene Anfragen für einen Client zu verarbeiten.""" print(f"--- Versuche, aufgeschobene Anfragen für Client '{client_id}' zu verarbeiten ---") for i in range(num_attempts): time.sleep(1.5) # Warte etwas, damit möglicherweise Kapazität frei wird print(f"[Processor-Thread] Versuch {i+1}/{num_attempts}, für '{client_id}' auszureihen...") result = rate_limiter.dequeue_and_process(client_id) if result: request, success = result if success: print(f"[Processor-Thread] Erfolgreich verarbeitet: {request.payload} (Priorität: {request.priority})") else: print(f"[Processor-Thread] Verarbeitung fehlgeschlagen (Kapazitätsproblem?): {request.payload} (Priorität: {request.priority}). Erneut eingereiht.") else: print(f"[Processor-Thread] In diesem Versuch wurden keine aufgeschobenen Anfragen für '{client_id}' verarbeitet (Warteschlange leer oder keine Kapazität).") if __name__ == "__main__": print("--- Rate-Limiter-Demonstration ---") # Initialisiere den Rate-Limiter limiter = RateLimiter(tier_config=TIER_CONFIG, cleanup_threshold_seconds=CLEANUP_THRESHOLD_SECONDS) # Registriere 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--- Simuliere gleichzeitige Anfragen ---") # Simuliere einen Burst von Anfragen aus mehreren Threads threads = [] # Client A (free tier) - wird wahrscheinlich das Rate-Limit erreichen t1 = threading.Thread(target=simulate_requests, args=(limiter, "client_A", 15, 0.1, 0), name="Thread-A") threads.append(t1) # Client B (standard tier) - mehr Kapazität t2 = threading.Thread(target=simulate_requests, args=(limiter, "client_B", 60, 0.05, 1), name="Thread-B") threads.append(t2) # Client C (premium tier) - höchste Kapazität t3 = threading.Thread(target=simulate_requests, args=(limiter, "client_C", 250, 0.01, 2), name="Thread-C") threads.append(t3) # Client D (free tier) - ein weiterer Client, der dasselbe Limit erreicht t4 = threading.Thread(target=simulate_requests, args=(limiter, "client_D", 10, 0.15, 3), name="Thread-D") threads.append(t4) # Starte Simulations-Threads for t in threads: t.start() # Gib etwas Zeit für erste Anfragen und das Einreihen time.sleep(2) print("\n--- Versuche, aufgeschobene Anfragen zu verarbeiten ---") # Starte Threads, um aufgeschobene Anfragen für Clients zu verarbeiten, die möglicherweise rate-limitiert wurden # Wir versuchen die Verarbeitung für Clients A und D (free tier) mehrfach. 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) # Versuche es auch für Client B, nur für den Fall 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() # Warte, bis alle Simulations-Threads abgeschlossen sind for t in threads: t.join() # Warte, bis die Processor-Threads ihre Versuche abgeschlossen haben for tp in processor_threads: tp.join() print("\n--- Demonstration abgeschlossen ---") print("Endgültige Zustandsprüfung (Anzahlen sind aufgrund des Timings ungefähr):") for tier_name, counter in limiter.tier_counters.items(): print(f"Tier '{tier_name}': Aktuelle Anfragen im Fenster: {counter.get_current_count()} (Max: {counter.max_requests})") for tier_name, queue in limiter.tier_queues.items(): print(f"Tier '{tier_name}': Aufgeschobene Anfragen in der Warteschlange: {len(queue)}") # --- Erklärung der Designentscheidungen --- # 1. Sliding-Window-Implementierung: # - Wir verwenden `collections.deque`, um Zeitstempel von Anfragen innerhalb des Fensters zu speichern. # - `deque` bietet O(1)-Komplexität für das Anhängen neuer Zeitstempel und das Entfernen alter Zeitstempel von links. # - Wenn `record_request` aufgerufen wird, entfernen wir zunächst alle Zeitstempel, die älter sind als `current_time - window_seconds`. # - Dadurch wird sichergestellt, dass `len(self.request_timestamps)` die Anzahl der Anfragen innerhalb des aktuellen Sliding Windows korrekt widerspiegelt. # - Trade-off: Dieser Ansatz ist präzise, kann aber mehrere pops in `record_request` erfordern, wenn viele Anfragen in kurzer Folge eintreffen und dann aufhören. Mit der Zeit bleibt die Größe der deque jedoch kontrolliert. # - Thread-Sicherheit: Jede `SlidingWindowCounter`-Instanz besitzt ihren eigenen `threading.Lock`, um ihre `request_timestamps`-deque zu schützen und sicherzustellen, dass gleichzeitige Aufrufe von `record_request` oder `get_current_count` für dasselbe Tier serialisiert werden. # 2. Mehrere Tiers: # - Die Klasse `RateLimiter` verwaltet ein Dictionary `tier_config`, das Tier-Namen auf ihre Parameter für die Ratenbegrenzung (`max_requests`, `window_seconds`) abbildet. # - Sie verwaltet außerdem separate `SlidingWindowCounter`-Instanzen und Prioritätswarteschlangen (`tier_queues`) für jedes Tier. # - Die Zuweisung des Client-Tiers wird in `client_tiers` gespeichert. # - Dies ermöglicht flexible Konfiguration und eine Isolierung der Ratenbegrenzungen zwischen verschiedenen Client-Typen. # 3. Prioritätswarteschlange für aufgeschobene Anfragen: # - Für jedes Tier wird eine standardmäßige Python-Liste als Min-Heap verwendet, verwaltet durch das Modul `heapq`. # - Die `Request`-Dataclass ist so gestaltet, dass `priority` der primäre Sortierschlüssel ist (kleinerer Wert = höhere Priorität). # - Eine eindeutige `request_id` ist enthalten, um stabiles Sortieren sicherzustellen (FIFO bei gleicher Priorität) und Vergleichsfehler zu verhindern, falls Zeitstempel identisch sind. # - `enqueue_request` verwendet `heapq.heappush`, um Anfragen hinzuzufügen und dabei die Heap-Eigenschaft aufrechtzuerhalten. # - `dequeue_and_process` verwendet `heapq.heappop`, um die Anfrage mit der höchsten Priorität abzurufen. # - Entscheidend ist, dass `dequeue_and_process` zunächst die verfügbare Kapazität *vor* dem Versuch prüft, aus dem Heap zu entfernen. Falls Kapazität verfügbar ist, erwirbt es dann den globalen Lock, prüft die Kapazität erneut (um Race Conditions zu behandeln), entfernt die Anfrage, und versucht dann erneut `record_request()`. Wenn `record_request` erfolgreich ist, wird die Anfrage verarbeitet. Wenn es fehlschlägt (aufgrund einer Race Condition, bei der die Kapazität gerade gefüllt wurde), wird die Anfrage zurück auf den Heap gelegt. # - Trade-off: Die Verwendung von Python-Liste + heapq ist für moderate Zahlen aufgeschobener Anfragen effizient. Bei extrem hohen Volumina könnte eine stärker spezialisierte gleichzeitige Prioritätswarteschlangen-Implementierung in Betracht gezogen werden, was jedoch die Komplexität erheblich erhöht. # 4. Thread-Sicherheit: # - `SlidingWindowCounter`: Verwendet einen internen `threading.Lock` für seine `request_timestamps`-deque. # - `RateLimiter`: # - `global_lock` (`threading.Lock`): Schützt gemeinsam genutzten Client-Zustand (`client_tiers`, `client_last_activity`) und die Tier-Prioritätswarteschlangen (`tier_queues`). Dieser Lock wird bei Operationen wie `register_client`, `enqueue_request` und innerhalb von `dequeue_and_process` beim Zugriff auf bzw. Ändern der Warteschlangen erworben. # - `allow_request`: Ruft das threadsichere `counter.record_request()` auf. Außerdem aktualisiert es die Client-Aktivität unter dem globalen Lock. # - `dequeue_and_process`: Verwaltet Locks sorgfältig. Zuerst wird die Kapazität *ohne* den globalen Lock geprüft. Wenn Kapazität *möglicherweise* verfügbar ist, wird der globale Lock erworben, die Kapazität erneut geprüft, die Heap-Operation ausgeführt und anschließend das threadsichere `counter.record_request()` erneut aufgerufen. # - Trade-off: Die Verwendung mehrerer Locks (einer pro Zähler + ein globaler Lock) zielt darauf ab, Contention im Vergleich zu einem einzelnen globalen Lock für alle Operationen zu reduzieren. Allerdings erfordern Operationen, die sich über Client-Zustand und Tier-Zustand erstrecken (wie das Ausreihen), eine sorgfältige Lock-Reihenfolge oder den Erwerb des globalen Locks. # 5. Bereinigung: # - Ein Hintergrund-Thread (`_cleanup_task`) durchsucht periodisch `client_last_activity`. # - Er identifiziert Clients, deren Zeitstempel der letzten Aktivität älter ist als `cleanup_threshold_seconds`. # - Diese Clients werden unter dem `global_lock` aus `client_tiers` und `client_last_activity` entfernt. # - Dadurch wird verhindert, dass der `RateLimiter` unbegrenzt große Zustandsmengen für inaktive Clients ansammelt. # - Trade-off: Die Bereinigung erfolgt nicht sofort; sie läuft periodisch. `cleanup_threshold_seconds` bestimmt den Trade-off zwischen Speichernutzung und Reaktionsfähigkeit der Bereinigung. # 6. Behandelte Randfälle: # - Doppelte Client-Registrierung: `register_client` gibt False zurück und protokolliert eine Warnung. # - Nicht registrierte Clients: `allow_request` und `enqueue_request` prüfen `client_tiers` und geben Fehler zurück bzw. protokollieren sie. # - Leere Prioritätswarteschlangen: `dequeue_and_process` behandelt Fälle, in denen `self.tier_queues[tier]` leer ist. # - Gleichzeitige Änderungen: Werden durch geeignete Verwendung von `threading.Lock` behandelt. # - Uhrengenauigkeit: Es wird `time.time()` verwendet, was im Allgemeinen ausreichend ist. Für Anforderungen an extrem hohe Präzision könnten plattformspezifische monotone Uhren in Betracht gezogen werden, aber `time.time()` ist für diese Art von Anwendung Standard. # - Race Conditions beim Ausreihen: Werden durch erneute Prüfung der Kapazität und erneutes Erfassen der Anfrage innerhalb des kritischen Abschnitts von `dequeue_and_process` behandelt. ```

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")