Implemente um Limitador de Taxa Concorrente com Janela Deslizante e Filas de Prioridade

Resposta: ```python import time import threading import collections import heapq import uuid from dataclasses import dataclass, field from typing import Dict, List, Tuple, Any, Optional # --- Configuração --- # Define os níveis com os respetivos limites de taxa e tamanhos de janela. # Um window_seconds menor significa limitação de taxa mais restrita. TIER_CONFIG = { "free": {"max_requests": 5, "window_seconds": 60}, "standard": {"max_requests": 50, "window_seconds": 60}, "premium": {"max_requests": 200, "window_seconds": 60}, } # Quanto tempo manter os dados do cliente antes da limpeza se não houver atividade. CLEANUP_THRESHOLD_SECONDS = 300 # --- Estruturas de Dados --- @dataclass(order=True) class Request: """Representa um pedido com prioridade, para a fila de prioridade.""" # Usamos priority como o primeiro elemento para o heapq ordenar por ele. # Número menor significa prioridade mais alta. priority: int # Usa um ID único para desempatar prioridades e garantir ordenação estável. request_id: str = field(default_factory=lambda: str(uuid.uuid4())) client_id: str timestamp: float payload: Any def __lt__(self, other): # Comparação personalizada para heapq: prioriza por prioridade, depois por request_id para estabilidade. if self.priority != other.priority: return self.priority < other.priority return self.request_id < other.request_id class SlidingWindowCounter: """Implementa um contador de janela deslizante para limitação de taxa.""" def __init__(self, max_requests: int, window_seconds: float): self.max_requests = max_requests self.window_seconds = window_seconds # Usa um deque para armazenar os carimbos temporais dos pedidos dentro da janela. # Deques oferecem appends e pops eficientes em ambas as extremidades. self.request_timestamps: collections.deque[float] = collections.deque() # Lock para acesso thread-safe a request_timestamps. self.lock = threading.Lock() def record_request(self) -> bool: """Regista um pedido e devolve True se estiver dentro do limite, False caso contrário.""" with self.lock: current_time = time.time() # Remove carimbos temporais mais antigos do que a janela. while self.request_timestamps and self.request_timestamps[0] <= current_time - self.window_seconds: self.request_timestamps.popleft() # Verifica se adicionar um novo pedido excede o 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: """Devolve o número atual de pedidos dentro da janela.""" with self.lock: current_time = time.time() # Limpa carimbos temporais antigos antes de devolver a contagem. 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: """Um limitador de taxa thread-safe com múltiplos níveis, janela deslizante e filas de prioridade.""" 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 # Armazena os contadores de janela deslizante para cada nível. self.tier_counters: Dict[str, SlidingWindowCounter] = {} # Armazena a fila de prioridade para pedidos adiados de cada nível. self.tier_queues: Dict[str, List[Request]] = collections.defaultdict(list) # Armazena o carimbo temporal da última atividade de cada cliente para limpeza. self.client_last_activity: Dict[str, float] = {} # Armazena o nível atribuído a cada cliente. self.client_tiers: Dict[str, str] = {} # Lock principal para proteger estruturas de dados partilhadas como client_tiers, client_last_activity e tier_queues. # O SlidingWindowCounter tem o seu próprio lock interno. self.global_lock = threading.Lock() # Inicializa os contadores dos níveis. for tier_name, config in tier_config.items(): self.tier_counters[tier_name] = SlidingWindowCounter( max_requests=config["max_requests"], window_seconds=config["window_seconds"] ) # Inicia uma thread em segundo plano para limpeza. 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: """Regista um cliente com um nível específico. Devolve True se for bem-sucedido, False se o cliente já existir ou se o nível for inválido.""" if tier not in self.tier_config: print(f"[!] Erro: O nível '{tier}' não existe.") return False with self.global_lock: if client_id in self.client_tiers: print(f"[!] Aviso: O cliente '{client_id}' já está registado.") return False self.client_tiers[client_id] = tier self.client_last_activity[client_id] = time.time() print(f"[*] Cliente '{client_id}' registado com o nível '{tier}'.") return True def _get_client_tier(self, client_id: str) -> Optional[str]: """Obtém com segurança o nível para um determinado client_id.""" with self.global_lock: return self.client_tiers.get(client_id) def _update_client_activity(self, client_id: str): """Atualiza o carimbo temporal da última atividade de um cliente.""" with self.global_lock: self.client_last_activity[client_id] = time.time() def allow_request(self, client_id: str) -> bool: """Verifica se um pedido de client_id é permitido com base no limite de taxa do seu nível.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Erro: O cliente '{client_id}' não está registado.") return False counter = self.tier_counters.get(tier) if not counter: # Isto não deve acontecer se tier_config for válido e a inicialização estiver correta. print(f"[!] Erro interno: Contador não encontrado para o nível '{tier}'.") return False if counter.record_request(): self._update_client_activity(client_id) # print(f"[*] Pedido PERMITIDO para o cliente '{client_id}' (Nível: {tier}). Contagem atual: {counter.get_current_count()}") return True else: # print(f"[*] Pedido NEGADO para o cliente '{client_id}' (Nível: {tier}). Contagem atual: {counter.get_current_count()}") return False def enqueue_request(self, client_id: str, priority: int, payload: Any): """Coloca um pedido limitado por taxa na fila de prioridade apropriada.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Erro: O cliente '{client_id}' não está registado. Não é possível enfileirar.") 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"[*] Pedido ENFILEIRADO para o cliente '{client_id}' (Nível: {tier}, Prioridade: {priority}). Tamanho da fila: {len(self.tier_queues[tier])}") self._update_client_activity(client_id) def dequeue_and_process(self, client_id: str) -> Optional[Tuple[Request, bool]]: """Tenta retirar da fila e processar o pedido de maior prioridade para o nível de um cliente se houver capacidade disponível.""" tier = self._get_client_tier(client_id) if not tier: print(f"[!] Erro: O cliente '{client_id}' não está registado. Não é possível desenfileirar.") return None counter = self.tier_counters.get(tier) if not counter: print(f"[!] Erro interno: Contador não encontrado para o nível '{tier}'.") return None # Verifica se há capacidade disponível ANTES de tentar retirar da fila. # Isto evita locking e operações de heap desnecessários se não existir capacidade. if counter.get_current_count() < counter.max_requests: with self.global_lock: queue = self.tier_queues[tier] if queue: # Revalida a capacidade dentro do lock para lidar com race conditions. # É possível que outra thread tenha acabado de preencher a capacidade. if counter.get_current_count() < counter.max_requests: highest_priority_request = heapq.heappop(queue) # Tenta registar o pedido novamente. Isto deve resultar se houver capacidade disponível. if counter.record_request(): self._update_client_activity(highest_priority_request.client_id) print(f"[*] Pedido DESENFILEIRADO e PROCESSADO para o cliente '{highest_priority_request.client_id}' (Nível: {tier}, Prioridade: {highest_priority_request.priority}). Tamanho da fila: {len(queue)}") return highest_priority_request, True else: # Esta é uma race condition rara: havia capacidade disponível, mas record_request falhou. # Volta a colocá-lo na fila e regista um erro. heapq.heappush(queue, highest_priority_request) print(f"[!] Race condition: Falha ao registar o pedido desenfileirado para {highest_priority_request.client_id}. Recolocado na fila.") return highest_priority_request, False else: # A capacidade foi ocupada por outra thread entre a verificação externa e esta verificação interna. # Volta a colocar o pedido na fila se ele tiver sido removido. heapq.heappush(queue, highest_priority_request) # print(f"[*] Tentativa de desenfileiramento falhou para {client_id}: Capacidade preenchida por outra thread.") return highest_priority_request, False else: # A fila está vazia. # print(f"[*] Tentativa de desenfileiramento para {client_id}: A fila está vazia.") return None, False else: # Não há capacidade disponível. # print(f"[*] Tentativa de desenfileiramento para {client_id}: Sem capacidade disponível.") return None, False def _cleanup_task(self): """Tarefa em segundo plano para remover dados de clientes inativos.""" while True: time.sleep(self.cleanup_threshold_seconds / 2) # Verifica periodicamente 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) # Opcionalmente, limpar filas deste cliente se necessário, embora normalmente isso não seja feito. # Por simplicidade, deixamos os pedidos enfileirados como estão. print(f"[*] Cliente inativo '{client_id}' removido na limpeza.") # --- Script de Demonstração --- def simulate_requests(rate_limiter: RateLimiter, client_id: str, num_requests: int, delay: float, priority_offset: int = 0): """Simula uma rajada de pedidos de um único cliente.""" print(f"--- A simular {num_requests} pedidos do cliente '{client_id}' ---") for i in range(num_requests): # Simula prioridades variáveis para pedidos enfileirados priority = i % 5 + priority_offset # Número menor = prioridade mais alta payload = f"data_{client_id}_{i}" if rate_limiter.allow_request(client_id): print(f"[Thread {threading.current_thread().name}] Pedido {i+1}/{num_requests} PERMITIDO para '{client_id}'. Payload: {payload}") else: print(f"[Thread {threading.current_thread().name}] Pedido {i+1}/{num_requests} NEGADO para '{client_id}'. A enfileirar com prioridade {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): """Tenta periodicamente processar pedidos adiados de um cliente.""" print(f"--- A tentar processar pedidos adiados para o cliente '{client_id}' ---") for i in range(num_attempts): time.sleep(1.5) # Espera um pouco para que a capacidade possa potencialmente libertar-se print(f"[Thread do Processador] Tentativa {i+1}/{num_attempts} de desenfileirar para '{client_id}'...") result = rate_limiter.dequeue_and_process(client_id) if result: request, success = result if success: print(f"[Thread do Processador] Processado com sucesso: {request.payload} (Prioridade: {request.priority})") else: print(f"[Thread do Processador] Falha no processamento (problema de capacidade?): {request.payload} (Prioridade: {request.priority}). Reenfileirado.") else: print(f"[Thread do Processador] Nenhum pedido adiado processado para '{client_id}' nesta tentativa (fila vazia ou sem capacidade).") if __name__ == "__main__": print("--- Demonstração do Limitador de Taxa ---") # Inicializa o limitador de taxa limiter = RateLimiter(tier_config=TIER_CONFIG, cleanup_threshold_seconds=CLEANUP_THRESHOLD_SECONDS) # Regista clientes 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--- A simular pedidos concorrentes ---") # Simula uma rajada de pedidos a partir de múltiplas threads threads = [] # Cliente A (nível free) - provavelmente vai atingir o limite de taxa t1 = threading.Thread(target=simulate_requests, args=(limiter, "client_A", 15, 0.1, 0), name="Thread-A") threads.append(t1) # Cliente B (nível standard) - mais capacidade t2 = threading.Thread(target=simulate_requests, args=(limiter, "client_B", 60, 0.05, 1), name="Thread-B") threads.append(t2) # Cliente C (nível premium) - maior capacidade t3 = threading.Thread(target=simulate_requests, args=(limiter, "client_C", 250, 0.01, 2), name="Thread-C") threads.append(t3) # Cliente D (nível free) - outro cliente a atingir o mesmo limite t4 = threading.Thread(target=simulate_requests, args=(limiter, "client_D", 10, 0.15, 3), name="Thread-D") threads.append(t4) # Inicia as threads de simulação for t in threads: t.start() # Dá algum tempo para os pedidos iniciais e o enfileiramento time.sleep(2) print("\n--- A tentar processar pedidos adiados ---") # Inicia threads para processar pedidos adiados para clientes que possam ter sido limitados por taxa # Vamos tentar processar para os clientes A e D (nível free) várias vezes. 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) # Também tenta para o cliente B, por via das dúvidas 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() # Espera que todas as threads de simulação terminem for t in threads: t.join() # Espera que as threads de processamento terminem as suas tentativas for tp in processor_threads: tp.join() print("\n--- Demonstração Concluída ---") print("Verificação do estado final (as contagens são aproximadas devido ao timing):") for tier_name, counter in limiter.tier_counters.items(): print(f"Nível '{tier_name}': Pedidos atuais na janela: {counter.get_current_count()} (Máx: {counter.max_requests})") for tier_name, queue in limiter.tier_queues.items(): print(f"Nível '{tier_name}': Pedidos adiados na fila: {len(queue)}") # --- Explicação das Escolhas de Design --- # 1. Implementação de Janela Deslizante: # - Usamos `collections.deque` para armazenar os carimbos temporais dos pedidos dentro da janela. # - `deque` oferece complexidade O(1) para acrescentar novos carimbos temporais e remover os antigos pela esquerda. # - Quando `record_request` é chamado, primeiro removemos todos os carimbos temporais mais antigos do que `current_time - window_seconds`. # - Isto garante que `len(self.request_timestamps)` reflete com precisão o número de pedidos dentro da janela deslizante atual. # - Trade-off: Esta abordagem é precisa, mas pode envolver múltiplos pops em `record_request` se muitos pedidos chegarem em rápida sucessão e depois parem. No entanto, ao longo do tempo, o tamanho do deque é gerido. # - Segurança em Threads: Cada instância de `SlidingWindowCounter` tem o seu próprio `threading.Lock` para proteger o seu deque `request_timestamps`, garantindo que chamadas concorrentes a `record_request` ou `get_current_count` para o mesmo nível sejam serializadas. # 2. Múltiplos Níveis: # - A classe `RateLimiter` mantém um dicionário `tier_config` que mapeia nomes de níveis para os seus parâmetros de limite de taxa (`max_requests`, `window_seconds`). # - Também mantém instâncias separadas de `SlidingWindowCounter` e filas de prioridade (`tier_queues`) para cada nível. # - A atribuição do nível do cliente é armazenada em `client_tiers`. # - Isto permite configuração flexível e isolamento dos limites de taxa entre diferentes tipos de clientes. # 3. Fila de Prioridade para Pedidos Adiados: # - Para cada nível, é usada uma lista Python padrão como min-heap, gerida pelo módulo `heapq`. # - O dataclass `Request` é concebido com `priority` como chave de ordenação principal (valor menor = prioridade mais alta). # - Um `request_id` único é incluído para garantir ordenação estável (FIFO para prioridades iguais) e evitar erros de comparação se os carimbos temporais forem idênticos. # - `enqueue_request` usa `heapq.heappush` para adicionar pedidos, mantendo a propriedade de heap. # - `dequeue_and_process` usa `heapq.heappop` para obter o pedido de maior prioridade. # - Crucialmente, `dequeue_and_process` verifica primeiro a capacidade disponível *antes* de tentar remover do heap. Se houver capacidade disponível, adquire então o lock global, volta a verificar a capacidade (para lidar com race conditions), remove o pedido e tenta chamar `record_request` novamente. Se `record_request` tiver sucesso, o pedido é processado. Se falhar (devido a uma race condition em que a capacidade acabou de ser preenchida), o pedido é recolocado no heap. # - Trade-off: Usar lista do Python + heapq é eficiente para números moderados de pedidos adiados. Para volumes extremamente altos, poderia ser considerada uma implementação de fila de prioridade concorrente mais especializada, mas isso acrescenta complexidade significativa. # 4. Segurança em Threads: # - `SlidingWindowCounter`: Usa um `threading.Lock` interno para o seu deque `request_timestamps`. # - `RateLimiter`: # - `global_lock` (`threading.Lock`): Protege o estado partilhado dos clientes (`client_tiers`, `client_last_activity`) e as filas de prioridade por nível (`tier_queues`). Este lock é adquirido para operações como `register_client`, `enqueue_request` e dentro de `dequeue_and_process` ao aceder/modificar as filas. # - `allow_request`: Chama o `counter.record_request()` thread-safe. Também atualiza a atividade do cliente sob o lock global. # - `dequeue_and_process`: Gere cuidadosamente os locks. Primeiro verifica a capacidade *sem* o lock global. Se a capacidade *puder* estar disponível, adquire o lock global, volta a verificar a capacidade, realiza a operação de heap e depois chama novamente o `counter.record_request()` thread-safe. # - Trade-off: O uso de múltiplos locks (um por contador + um lock global) visa reduzir a contenção em comparação com um único lock global para todas as operações. No entanto, operações que abrangem o estado do cliente e o estado do nível (como desenfileirar) exigem ordenação cuidadosa dos locks ou aquisição do lock global. # 5. Limpeza: # - Uma thread em segundo plano (`_cleanup_task`) percorre periodicamente `client_last_activity`. # - Identifica clientes cujo carimbo temporal da última atividade é mais antigo do que `cleanup_threshold_seconds`. # - Estes clientes são removidos de `client_tiers` e `client_last_activity` sob o `global_lock`. # - Isto evita que o `RateLimiter` acumule indefinidamente grandes quantidades de estado para clientes inativos. # - Trade-off: A limpeza não é instantânea; é executada periodicamente. O `cleanup_threshold_seconds` determina o equilíbrio entre uso de memória e capacidade de resposta da limpeza. # 6. Casos Limite Tratados: # - Registo duplicado de cliente: `register_client` devolve False e regista um aviso. # - Clientes não registados: `allow_request` e `enqueue_request` verificam `client_tiers` e devolvem/registam erros. # - Filas de prioridade vazias: `dequeue_and_process` lida com casos em que `self.tier_queues[tier]` está vazia. # - Modificações concorrentes: Resolvidas usando `threading.Lock` de forma apropriada. # - Precisão do relógio: `time.time()` é usado, o que geralmente é suficiente. Para requisitos de precisão extremamente elevados, poderiam ser considerados relógios monotónicos específicos da plataforma, mas `time.time()` é o padrão para este tipo de aplicação. # - Race conditions no desenfileiramento: Tratadas voltando a verificar a capacidade e voltando a registar o pedido dentro da secção crítica 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")