Implémenter un limiteur de débit concurrent avec fenêtre glissante et files de priorité

Réponse : ```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 --- # Définir les niveaux avec leurs limites de débit et tailles de fenêtre respectives. # Une valeur de window_seconds plus faible signifie une limitation de débit plus stricte. TIER_CONFIG = { "free": {"max_requests": 5, "window_seconds": 60}, "standard": {"max_requests": 50, "window_seconds": 60}, "premium": {"max_requests": 200, "window_seconds": 60}, } # Durée de conservation des données client avant nettoyage en l’absence d’activité. CLEANUP_THRESHOLD_SECONDS = 300 # --- Structures de données --- @dataclass(order=True) class Request: """Représente une requête avec une priorité, pour la file de priorité.""" # Nous utilisons priority comme premier élément pour que heapq effectue le tri dessus. # Un nombre plus faible signifie une priorité plus élevée. priority: int # Utiliser un ID unique pour départager les égalités de priorité et garantir un tri stable. request_id: str = field(default_factory=lambda: str(uuid.uuid4())) client_id: str timestamp: float payload: Any def __lt__(self, other): # Comparaison personnalisée pour heapq : prioriser par priorité, puis par request_id pour la stabilité. if self.priority != other.priority: return self.priority < other.priority return self.request_id < other.request_id class SlidingWindowCounter: """Implémente un compteur à fenêtre glissante pour la limitation de débit.""" def __init__(self, max_requests: int, window_seconds: float): self.max_requests = max_requests self.window_seconds = window_seconds # Utiliser une deque pour stocker les horodatages des requêtes dans la fenêtre. # Les deque offrent des ajouts et suppressions efficaces aux deux extrémités. self.request_timestamps: collections.deque[float] = collections.deque() # Verrou pour un accès thread-safe à request_timestamps. self.lock = threading.Lock() def record_request(self) -> bool: """Enregistre une requête et renvoie True si elle reste dans la limite, False sinon.""" with self.lock: current_time = time.time() # Supprimer les horodatages plus anciens que la fenêtre. while self.request_timestamps and self.request_timestamps[0] <= current_time - self.window_seconds: self.request_timestamps.popleft() # Vérifier si l’ajout d’une nouvelle requête dépasse la limite. 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: """Renvoie le nombre actuel de requêtes dans la fenêtre.""" with self.lock: current_time = time.time() # Nettoyer les anciens horodatages avant de renvoyer le compte. 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: """Un limiteur de débit thread-safe avec plusieurs niveaux, fenêtre glissante et files de priorité.""" 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 # Stocke les compteurs à fenêtre glissante pour chaque niveau. self.tier_counters: Dict[str, SlidingWindowCounter] = {} # Stocke la file de priorité des requêtes différées pour chaque niveau. self.tier_queues: Dict[str, List[Request]] = collections.defaultdict(list) # Stocke l’horodatage de la dernière activité pour chaque client en vue du nettoyage. self.client_last_activity: Dict[str, float] = {} # Stocke le niveau attribué à chaque client. self.client_tiers: Dict[str, str] = {} # Verrou principal pour protéger les structures de données partagées comme client_tiers, client_last_activity et tier_queues. # SlidingWindowCounter possède son propre verrou interne. self.global_lock = threading.Lock() # Initialiser les compteurs par niveau. for tier_name, config in tier_config.items(): self.tier_counters[tier_name] = SlidingWindowCounter( max_requests=config["max_requests"], window_seconds=config["window_seconds"] ) # Démarrer un thread d’arrière-plan pour le nettoyage. 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: """Enregistre un client avec un niveau spécifique. Renvoie True en cas de succès, False si le client existe déjà ou si le niveau est invalide.""" if tier not in self.tier_config: print(f"[!] Erreur : le niveau '{tier}' n’existe pas.") return False with self.global_lock: if client_id in self.client_tiers: print(f"[!] Avertissement : le client '{client_id}' est déjà enregistré.") return False self.client_tiers[client_id] = tier self.client_last_activity[client_id] = time.time() print(f"[*] Client '{client_id}' enregistré avec le niveau '{tier}'.") return True def _get_client_tier(self, client_id: str) -> Optional[str]: """Récupère en toute sécurité le niveau pour un client_id donné.""" with self.global_lock: return self.client_tiers.get(client_id) def _update_client_activity(self, client_id: str): """Met à jour l’horodatage de dernière activité d’un client.""" with self.global_lock: self.client_last_activity[client_id] = time.time() def allow_request(self, client_id: str) -> bool: """Vérifie si une requête provenant de client_id est autorisée selon la limite de débit de son niveau.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Erreur : le client '{client_id}' n’est pas enregistré.") return False counter = self.tier_counters.get(tier) if not counter: # Cela ne devrait pas arriver si tier_config est valide et si l’initialisation est correcte. print(f"[!] Erreur interne : compteur introuvable pour le niveau '{tier}'.") return False if counter.record_request(): self._update_client_activity(client_id) # print(f"[*] Requête AUTORISÉE pour le client '{client_id}' (Niveau : {tier}). Compte actuel : {counter.get_current_count()}") return True else: # print(f"[*] Requête REFUSÉE pour le client '{client_id}' (Niveau : {tier}). Compte actuel : {counter.get_current_count()}") return False def enqueue_request(self, client_id: str, priority: int, payload: Any): """Place une requête limitée par le débit dans la file de priorité appropriée.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Erreur : le client '{client_id}' n’est pas enregistré. Impossible de mettre en file.") 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"[*] Requête MISE EN FILE pour le client '{client_id}' (Niveau : {tier}, Priorité : {priority}). Taille de la file : {len(self.tier_queues[tier])}") self._update_client_activity(client_id) def dequeue_and_process(self, client_id: str) -> Optional[Tuple[Request, bool]]: """Tente de retirer de la file et de traiter la requête de plus haute priorité pour le niveau d’un client si de la capacité est disponible.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Erreur : le client '{client_id}' n’est pas enregistré. Impossible de retirer de la file.") return None counter = self.tier_counters.get(tier) if not counter: print(f"[!] Erreur interne : compteur introuvable pour le niveau '{tier}'.") return None # Vérifier si de la capacité est disponible AVANT d’essayer de retirer de la file. # Cela évite des verrous inutiles et des opérations sur le tas si aucune capacité n’existe. if counter.get_current_count() < counter.max_requests: with self.global_lock: queue = self.tier_queues[tier] if queue: # Revérifier la capacité à l’intérieur du verrou pour gérer les conditions de concurrence. # Il est possible qu’un autre thread ait juste rempli la capacité. if counter.get_current_count() < counter.max_requests: highest_priority_request = heapq.heappop(queue) # Tenter d’enregistrer à nouveau la requête. Cela devrait réussir si de la capacité est disponible. if counter.record_request(): self._update_client_activity(highest_priority_request.client_id) print(f"[*] Requête RETIRÉE DE LA FILE et TRAITÉE pour le client '{highest_priority_request.client_id}' (Niveau : {tier}, Priorité : {highest_priority_request.priority}). Taille de la file : {len(queue)}") return highest_priority_request, True else: # Il s’agit d’une rare condition de concurrence : la capacité était disponible, mais record_request a échoué. # La remettre dans la file et consigner une erreur. heapq.heappush(queue, highest_priority_request) print(f"[!] Condition de concurrence : échec de l’enregistrement de la requête retirée de la file pour {highest_priority_request.client_id}. Réinsérée.") return highest_priority_request, False else: # La capacité a été prise par un autre thread entre la vérification externe et cette vérification interne. # Remettre la requête dans la file si elle a été retirée. heapq.heappush(queue, highest_priority_request) # print(f"[*] Tentative de retrait de file échouée pour {client_id} : capacité remplie par un autre thread.") return highest_priority_request, False else: # La file est vide. # print(f"[*] Tentative de retrait de file pour {client_id} : la file est vide.") return None, False else: # Aucune capacité disponible. # print(f"[*] Tentative de retrait de file pour {client_id} : aucune capacité disponible.") return None, False def _cleanup_task(self): """Tâche d’arrière-plan pour supprimer les données des clients inactifs.""" while True: time.sleep(self.cleanup_threshold_seconds / 2) # Vérifier périodiquement 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) # Optionnellement, vider les files de ce client si nécessaire, bien que cela ne soit généralement pas fait. # Par simplicité, nous laissons les requêtes mises en file telles quelles. print(f"[*] Nettoyage du client inactif '{client_id}'.") # --- Script de démonstration --- def simulate_requests(rate_limiter: RateLimiter, client_id: str, num_requests: int, delay: float, priority_offset: int = 0): """Simule une rafale de requêtes provenant d’un seul client.""" print(f"--- Simulation de {num_requests} requêtes du client '{client_id}' ---") for i in range(num_requests): # Simuler des priorités variables pour les requêtes mises en file priority = i % 5 + priority_offset # Nombre plus faible = priorité plus élevée payload = f"data_{client_id}_{i}" if rate_limiter.allow_request(client_id): print(f"[Thread {threading.current_thread().name}] Requête {i+1}/{num_requests} AUTORISÉE pour '{client_id}'. Payload : {payload}") else: print(f"[Thread {threading.current_thread().name}] Requête {i+1}/{num_requests} REFUSÉE pour '{client_id}'. Mise en file avec priorité {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): """Essaie périodiquement de traiter les requêtes différées d’un client.""" print(f"--- Tentative de traitement des requêtes différées pour le client '{client_id}' ---") for i in range(num_attempts): time.sleep(1.5) # Attendre un peu afin qu’une capacité puisse potentiellement se libérer print(f"[Thread de traitement] Tentative {i+1}/{num_attempts} de retrait de file pour '{client_id}'...") result = rate_limiter.dequeue_and_process(client_id) if result: request, success = result if success: print(f"[Thread de traitement] Traitement réussi : {request.payload} (Priorité : {request.priority})") else: print(f"[Thread de traitement] Échec du traitement (problème de capacité ?) : {request.payload} (Priorité : {request.priority}). Réinsérée dans la file.") else: print(f"[Thread de traitement] Aucune requête différée traitée pour '{client_id}' lors de cette tentative (file vide ou aucune capacité).") if __name__ == "__main__": print("--- Démonstration du limiteur de débit ---") # Initialiser le limiteur de débit limiter = RateLimiter(tier_config=TIER_CONFIG, cleanup_threshold_seconds=CLEANUP_THRESHOLD_SECONDS) # Enregistrer les 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--- Simulation de requêtes concurrentes ---") # Simuler une rafale de requêtes depuis plusieurs threads threads = [] # Client A (niveau free) - atteindra probablement la limite de débit t1 = threading.Thread(target=simulate_requests, args=(limiter, "client_A", 15, 0.1, 0), name="Thread-A") threads.append(t1) # Client B (niveau standard) - plus de capacité t2 = threading.Thread(target=simulate_requests, args=(limiter, "client_B", 60, 0.05, 1), name="Thread-B") threads.append(t2) # Client C (niveau premium) - capacité la plus élevée t3 = threading.Thread(target=simulate_requests, args=(limiter, "client_C", 250, 0.01, 2), name="Thread-C") threads.append(t3) # Client D (niveau free) - un autre client atteignant la même limite t4 = threading.Thread(target=simulate_requests, args=(limiter, "client_D", 10, 0.15, 3), name="Thread-D") threads.append(t4) # Démarrer les threads de simulation for t in threads: t.start() # Laisser un peu de temps pour les requêtes initiales et la mise en file time.sleep(2) print("\n--- Tentative de traitement des requêtes différées ---") # Démarrer des threads pour traiter les requêtes différées des clients qui ont pu être limités par le débit # Nous essaierons plusieurs fois pour les clients A et D (niveau free). 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) # Essayer aussi pour le client B, au cas où 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() # Attendre que tous les threads de simulation aient terminé for t in threads: t.join() # Attendre que les threads de traitement aient terminé leurs tentatives for tp in processor_threads: tp.join() print("\n--- Démonstration terminée ---") print("Vérification de l’état final (les comptes sont approximatifs en raison du timing) :") for tier_name, counter in limiter.tier_counters.items(): print(f"Niveau '{tier_name}' : requêtes actuelles dans la fenêtre : {counter.get_current_count()} (Max : {counter.max_requests})") for tier_name, queue in limiter.tier_queues.items(): print(f"Niveau '{tier_name}' : requêtes différées dans la file : {len(queue)}") # --- Explication des choix de conception --- # 1. Implémentation de la fenêtre glissante : # - Nous utilisons `collections.deque` pour stocker les horodatages des requêtes dans la fenêtre. # - `deque` fournit une complexité O(1) pour l’ajout de nouveaux horodatages et la suppression des anciens à gauche. # - Lorsque `record_request` est appelé, nous supprimons d’abord tous les horodatages plus anciens que `current_time - window_seconds`. # - Cela garantit que `len(self.request_timestamps)` reflète avec précision le nombre de requêtes dans la fenêtre glissante actuelle. # - Compromis : cette approche est précise, mais peut impliquer plusieurs suppressions dans `record_request` si de nombreuses requêtes arrivent en succession rapide puis s’arrêtent. Cependant, au fil du temps, la taille de la deque reste maîtrisée. # - Thread safety : chaque instance de `SlidingWindowCounter` possède son propre `threading.Lock` pour protéger sa deque `request_timestamps`, garantissant que les appels concurrents à `record_request` ou `get_current_count` pour un même niveau sont sérialisés. # 2. Niveaux multiples : # - La classe `RateLimiter` maintient un dictionnaire `tier_config` qui associe les noms de niveaux à leurs paramètres de limitation de débit (`max_requests`, `window_seconds`). # - Elle maintient aussi des instances séparées de `SlidingWindowCounter` et des files de priorité (`tier_queues`) pour chaque niveau. # - L’attribution du niveau client est stockée dans `client_tiers`. # - Cela permet une configuration flexible et une isolation des limites de débit entre différents types de clients. # 3. File de priorité pour les requêtes différées : # - Pour chaque niveau, une liste Python standard est utilisée comme tas min, géré par le module `heapq`. # - La dataclass `Request` est conçue avec `priority` comme clé de tri principale (valeur plus faible = priorité plus élevée). # - Un `request_id` unique est inclus pour garantir un tri stable (FIFO pour des priorités identiques) et éviter les erreurs de comparaison si les horodatages sont identiques. # - `enqueue_request` utilise `heapq.heappush` pour ajouter des requêtes tout en maintenant la propriété du tas. # - `dequeue_and_process` utilise `heapq.heappop` pour récupérer la requête de plus haute priorité. # - Point crucial : `dequeue_and_process` vérifie d’abord la capacité disponible *avant* d’essayer de retirer un élément du tas. Si de la capacité est disponible, il acquiert alors le verrou global, revérifie la capacité (pour gérer les conditions de concurrence), retire la requête et tente ensuite à nouveau `record_request`. Si `record_request` réussit, la requête est traitée. Si cela échoue (en raison d’une condition de concurrence où la capacité vient juste d’être remplie), la requête est remise dans le tas. # - Compromis : l’utilisation de la liste Python + heapq est efficace pour un nombre modéré de requêtes différées. Pour des volumes extrêmement élevés, on pourrait envisager une implémentation plus spécialisée de file de priorité concurrente, mais cela ajoute une complexité significative. # 4. Thread safety : # - `SlidingWindowCounter` : utilise un `threading.Lock` interne pour sa deque `request_timestamps`. # - `RateLimiter` : # - `global_lock` (`threading.Lock`) : protège l’état client partagé (`client_tiers`, `client_last_activity`) et les files de priorité par niveau (`tier_queues`). Ce verrou est acquis pour des opérations comme `register_client`, `enqueue_request` et à l’intérieur de `dequeue_and_process` lors de l’accès/modification des files. # - `allow_request` : appelle `counter.record_request()`, qui est thread-safe. Il met également à jour l’activité client sous le verrou global. # - `dequeue_and_process` : gère soigneusement les verrous. Il vérifie d’abord la capacité *sans* le verrou global. Si de la capacité *peut* être disponible, il acquiert le verrou global, revérifie la capacité, effectue l’opération sur le tas, puis appelle à nouveau `counter.record_request()`, qui est thread-safe. # - Compromis : l’utilisation de plusieurs verrous (un par compteur + un verrou global) vise à réduire la contention par rapport à un verrou global unique pour toutes les opérations. Cependant, les opérations qui couvrent à la fois l’état client et l’état du niveau (comme le retrait de file) nécessitent un ordre de verrouillage soigneux ou l’acquisition du verrou global. # 5. Nettoyage : # - Un thread d’arrière-plan (`_cleanup_task`) parcourt périodiquement `client_last_activity`. # - Il identifie les clients dont l’horodatage de dernière activité est plus ancien que `cleanup_threshold_seconds`. # - Ces clients sont supprimés de `client_tiers` et `client_last_activity` sous `global_lock`. # - Cela empêche `RateLimiter` d’accumuler indéfiniment une quantité de plus en plus importante d’état pour des clients inactifs. # - Compromis : le nettoyage n’est pas instantané ; il s’exécute périodiquement. La valeur de `cleanup_threshold_seconds` détermine le compromis entre l’utilisation mémoire et la réactivité du nettoyage. # 6. Cas limites pris en charge : # - Enregistrement de client en double : `register_client` renvoie False et consigne un avertissement. # - Clients non enregistrés : `allow_request` et `enqueue_request` vérifient `client_tiers` et renvoient/consignent des erreurs. # - Files de priorité vides : `dequeue_and_process` gère les cas où `self.tier_queues[tier]` est vide. # - Modifications concurrentes : traitées par une utilisation appropriée de `threading.Lock`. # - Précision de l’horloge : `time.time()` est utilisé, ce qui est généralement suffisant. Pour des exigences de très haute précision, des horloges monotones spécifiques à la plateforme pourraient être envisagées, mais `time.time()` est standard pour ce type d’application. # - Conditions de concurrence lors du retrait de file : gérées en revérifiant la capacité et en réenregistrant la requête dans la section critique de `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")