Implemente um cache LRU concorrente sem bloqueios

Resposta: import threading import random import time from typing import Any, Optional, List class _Node: """ Representa um nó na lista duplamente ligada usada para a ordenação LRU. Usa __slots__ para eficiência de memória. """ __slots__ = 'key', 'value', 'prev', 'next' def __init__(self, key: str, value: Any): self.key = key self.value = value self.prev = None self.next = None class ReadWriteLock: """ Uma implementação personalizada de bloqueio de Leitura-Escrita. Este bloqueio permite que múltiplos leitores mantenham o bloqueio simultaneamente, mas apenas um escritor por vez. Um escritor deve esperar que todos os leitores terminem, e os leitores devem esperar que um escritor termine. Esta é uma implementação com preferência para leitores, o que significa que novos leitores podem adquirir o bloqueio de leitura mesmo que um escritor esteja à espera. Esta classe também fornece gestores de contexto para tornar a aquisição e libertação de bloqueios mais limpas. """ def __init__(self): self._read_lock = threading.Lock() # Protege a contagem de _readers e a variável de condição self._write_lock = threading.Lock() # O bloqueio de escrita real self._readers = 0 # Número de leitores ativos self._no_readers = threading.Condition(self._read_lock) # Condição para escritores esperarem pelos leitores def acquire_read(self): """Adquire um bloqueio de leitura. Múltiplos leitores podem manter este bloqueio simultaneamente.""" with self._read_lock: # Espera se um escritor estiver atualmente a manter o bloqueio de escrita. # Novos leitores serão bloqueados se um escritor estiver ativo. while self._write_lock.locked(): self._no_readers.wait() self._readers += 1 def release_read(self): """Liberta um bloqueio de leitura.""" with self._read_lock: self._readers -= 1 # Se não restarem leitores, notifica quaisquer escritores à espera if self._readers == 0: self._no_readers.notify_all() def acquire_write(self): """Adquire um bloqueio de escrita. Apenas um escritor pode manter este bloqueio de cada vez.""" # Adquire primeiro o bloqueio principal de escrita para sinalizar intenção de escrever self._write_lock.acquire() with self._read_lock: # Depois adquire read_lock para esperar pelos leitores ativos # Espera até que todos os leitores ativos terminem while self._readers > 0: self._no_readers.wait() def release_write(self): """Liberta um bloqueio de escrita.""" with self._read_lock: # Adquire read_lock para notificar potenciais leitores/escritores self._no_readers.notify_all() # Notifica quaisquer leitores ou escritores à espera self._write_lock.release() # Liberta o bloqueio principal de escrita class _ReadContextManager: def __init__(self, lock): self.lock = lock def __enter__(self): self.lock.acquire_read() def __exit__(self, exc_type, exc_val, exc_tb): self.lock.release_read() class _WriteContextManager: def __init__(self, lock): self.lock = lock def __enter__(self): self.lock.acquire_write() def __exit__(self, exc_type, exc_val, exc_tb): self.lock.release_write() def read_locked(self): """Retorna um gestor de contexto para adquirir um bloqueio de leitura.""" return self._ReadContextManager(self) def write_locked(self): """Retorna um gestor de contexto para adquirir um bloqueio de escrita.""" return self._WriteContextManager(self) class LRUCache: """ Uma implementação de cache LRU (Least Recently Used) segura para threads. Este cache usa um ReadWriteLock para gerir o acesso concorrente. O design procura permitir leituras concorrentes para falhas de cache e para a operação `keys()`. No entanto, operações que modificam o estado do cache ou a ordem LRU (`put`, `delete`, e acertos de cache em `get`) adquirem um bloqueio de escrita, garantindo atomicidade e consistência para estas secções críticas. Esta abordagem oferece melhor concorrência do que um único mutex global para todas as operações, especialmente para cargas de trabalho com muitas leituras com falhas de cache frequentes ou chamadas a `keys()`. A operação `get` emprega um padrão de "upgrade": primeiro tenta um bloqueio de leitura para verificar a existência da chave. Se encontrada, liberta o bloqueio de leitura e adquire um bloqueio de escrita para atualizar a ordem LRU e obter o valor mais recente. Este padrão é cuidadosamente implementado para lidar com condições de corrida que possam ocorrer durante o upgrade do bloqueio (por exemplo, a chave ser apagada ou atualizada por outra thread). """ def __init__(self, capacity: int): if capacity <= 0: raise ValueError("A capacidade deve ser um inteiro positivo.") self._capacity = capacity self._cache = {} # Armazena key -> _Node para pesquisa O(1) self._size = 0 # Nós fictícios de cabeça e cauda para a lista duplamente ligada. # Isto simplifica casos extremos para adicionar/remover nós, já que os nós de dados reais # estão sempre entre a cabeça e a cauda. self._head = _Node("dummy_head", None) self._tail = _Node("dummy_tail", None) self._head.next = self._tail self._tail.prev = self._head # Controlo de concorrência: um único ReadWriteLock protege todo o estado do cache. self._rw_lock = ReadWriteLock() def _add_node(self, node: _Node): """Adiciona um nó logo após a cabeça (posição mais recentemente usada). Assume que o bloqueio de escrita está ativo.""" node.prev = self._head node.next = self._head.next self._head.next.prev = node self._head.next = node def _remove_node(self, node: _Node): """Remove um nó da lista duplamente ligada. Assume que o bloqueio de escrita está ativo.""" prev_node = node.prev next_node = node.next prev_node.next = next_node next_node.prev = prev_node def _move_to_front(self, node: _Node): """Move um nó existente para a frente da lista (mais recentemente usado). Assume que o bloqueio de escrita está ativo.""" self._remove_node(node) self._add_node(node) def _pop_tail(self) -> _Node: """Remove e retorna o nó menos recentemente usado (o que está imediatamente antes da cauda). Assume que o bloqueio de escrita está ativo.""" node = self._tail.prev self._remove_node(node) return node def get(self, key: str) -> Optional[Any]: """ Retorna o valor associado à chave se ela existir, e marca-a como recentemente usada. Retorna None se a chave não estiver no cache. Esta operação primeiro adquire um bloqueio de leitura para verificar a existência da chave. Se a chave for encontrada, liberta o bloqueio de leitura e depois adquire um bloqueio de escrita para atualizar a ordem LRU e obter o valor mais recente. Este padrão de "upgrade" permite falhas de `get` e operações `keys()` concorrentes, mas serializa acertos de `get` com outras escritas. """ node_found_under_read_lock = None with self._rw_lock.read_locked(): node_found_under_read_lock = self._cache.get(key) if not node_found_under_read_lock: return None # Se a chave foi encontrada, adquire um bloqueio de escrita para atualizar a ordem LRU e obter o valor mais recente. # Este é um padrão de "upgrade". Pode ocorrer uma condição de corrida aqui: outra thread # pode apagar ou atualizar a chave entre a libertação do bloqueio de leitura e a aquisição do bloqueio de escrita. with self._rw_lock.write_locked(): # Verifica novamente se a chave ainda existe e se se refere exatamente ao mesmo objeto nó. # Isto é crucial para a correção sob contenção. Se outra thread tiver executado # um `put` para a mesma chave, `_cache.get(key)` retornaria um *novo* nó. # Se outra thread tiver executado um `delete`, retornaria `None`. current_node = self._cache.get(key) if current_node is node_found_under_read_lock: # Garante que é exatamente o mesmo objeto nó self._move_to_front(node_found_under_read_lock) value = current_node.value # Obtém o valor mais recente depois de mover return value else: # O nó foi apagado ou substituído por outra operação `put` # entre a nossa aquisição do bloqueio de leitura e a do bloqueio de escrita. Trata como falha. return None def put(self, key: str, value: Any) -> None: """ Insere ou atualiza o par chave-valor. Se o cache exceder a capacidade após a inserção, remove o item menos recentemente usado. Esta é uma operação de escrita e adquire um bloqueio de escrita. """ with self._rw_lock.write_locked(): node = self._cache.get(key) if node: # A chave existe: atualiza o valor e move para a frente node.value = value self._move_to_front(node) else: # A chave não existe: adiciona um novo nó new_node = _Node(key, value) self._cache[key] = new_node self._add_node(new_node) self._size += 1 # Remove se a capacidade for excedida if self._size > self._capacity: lru_node = self._pop_tail() del self._cache[lru_node.key] self._size -= 1 def delete(self, key: str) -> bool: """ Remove a chave do cache. Retorna True se a chave estava presente, False caso contrário. Esta é uma operação de escrita e adquire um bloqueio de escrita. """ with self._rw_lock.write_locked(): node = self._cache.get(key) if node: self._remove_node(node) del self._cache[key] self._size -= 1 return True return False def keys(self) -> List[str]: """ Retorna uma lista de todas as chaves atualmente no cache, ordenadas da mais recentemente usada para a menos recentemente usada. Esta é uma operação de leitura e adquire um bloqueio de leitura. """ with self._rw_lock.read_locked(): keys_list = [] current = self._head.next while current is not self._tail: keys_list.append(current.key) current = current.next return keys_list def run_tests(): """ Executa uma série de testes de thread única e multithread para verificar a implementação de LRUCache. """ print("--- A iniciar testes da cache LRU ---") # --- Testes de thread única --- print("\n--- Testes de thread única ---") cache = LRUCache(capacity=3) assert cache.get("a") is None, "ST: Obter chave inexistente deve retornar None" assert cache.keys() == [], "ST: Cache vazio deve ter lista de chaves vazia" cache.put("a", 1) assert cache.get("a") == 1, "ST: Obter chave existente após put" assert cache.keys() == ["a"], "ST: Lista de chaves após um put" cache.put("b", 2) cache.put("c", 3) assert cache.keys() == ["c", "b", "a"], "ST: Lista de chaves após vários puts" assert cache._size == 3, "ST: O tamanho do cache deve ser 3" # Testar remoção por capacidade cache.put("d", 4) # 'a' deve ser removida por ser a LRU assert cache.get("a") is None, "ST: A chave removida 'a' deve ser None" assert cache.get("d") == 4, "ST: A nova chave 'd' deve estar presente" assert cache.keys() == ["d", "c", "b"], "ST: Lista de chaves após remoção" assert cache._size == 3, "ST: O tamanho do cache deve continuar a ser 3 após remoção" # Testar put com chave existente (atualizar valor e mover para a frente) cache.put("b", 22) # 'b' deve mover-se para a frente, valor atualizado assert cache.get("b") == 22, "ST: Valor atualizado para 'b'" assert cache.keys() == ["b", "d", "c"], "ST: Lista de chaves após atualizar e mover 'b'" # Testar delete assert cache.delete("c") is True, "ST: Apagar chave existente 'c'" assert cache.get("c") is None, "ST: A chave apagada 'c' deve ser None" assert cache.keys() == ["b", "d"], "ST: Lista de chaves após apagar 'c'" assert cache._size == 2, "ST: Tamanho do cache após delete" assert cache.delete("z") is False, "ST: Apagar chave inexistente 'z'" assert cache.keys() == ["b", "d"], "ST: Lista de chaves inalterada após apagar chave inexistente" # Testar capacidade 1 cache_cap1 = LRUCache(capacity=1) cache_cap1.put("x", 10) assert cache_cap1.get("x") == 10, "ST: Capacidade 1, obter 'x'" cache_cap1.put("y", 20) # 'x' deve ser removida assert cache_cap1.get("x") is None, "ST: Capacidade 1, 'x' removida" assert cache_cap1.get("y") == 20, "ST: Capacidade 1, obter 'y'" assert cache_cap1.keys() == ["y"], "ST: Capacidade 1, lista de chaves" print("Testes de thread única aprovados!") # --- Testes multithread --- print("\n--- Testes multithread ---") MT_CAPACITY = 10 MT_NUM_THREADS = 8 MT_OPS_PER_THREAD = 2000 MT_KEY_RANGE = 20 # Chaves de 0 a 19 (por ex., "0", "1", ..., "19") mt_cache = LRUCache(capacity=MT_CAPACITY) # Barreira para garantir que todas as threads de trabalho iniciam as suas operações simultaneamente barrier = threading.Barrier(MT_NUM_THREADS + 1) # +1 para a thread principal error_flag = threading.Event() # Definida se alguma thread encontrar um erro # Rastreia chaves que *alguma vez* foram inseridas com sucesso no cache por qualquer thread. # Usado para verificar que `get` nunca retorna um valor para uma chave que realmente nunca foi inserida. global_ever_inserted_keys = set() global_ever_inserted_keys_lock = threading.Lock() def mt_worker(cache_instance, thread_id, barrier_instance, error_event, ever_inserted_keys_set, ever_inserted_keys_lock): """Função worker para testes multithread.""" barrier_instance.wait() # Espera que todas as threads estejam prontas para iniciar operações for i in range(MT_OPS_PER_THREAD): if error_event.is_set(): break # Para se outra thread já encontrou um erro # Escolhe aleatoriamente uma operação com viés para 'get' e 'put' op = random.choices(['get', 'put', 'delete'], weights=[0.5, 0.4, 0.1], k=1)[0] key = str(random.randint(0, MT_KEY_RANGE - 1)) value = f"value_T{thread_id}_K{key}_I{i}" if op == 'put': cache_instance.put(key, value) with ever_inserted_keys_lock: ever_inserted_keys_set.add(key) elif op == 'get': val = cache_instance.get(key) if val is not None: # Afirma que, se um valor for retornado, a chave deve ter sido inserida em algum momento. with ever_inserted_keys_lock: if key not in ever_inserted_keys_set: print(f"MT Error: Thread {thread_id} obteve o valor '{val}' para a chave '{key}' " f"que nunca foi inserida globalmente. Isto não deveria acontecer.") error_event.set() break # Afirma que o valor não está corrompido (verificação simples de formato) if not isinstance(val, str) or not val.startswith("value_T"): print(f"MT Error: Thread {thread_id} obteve um valor corrompido '{val}' para a chave '{key}'.") error_event.set() break elif op == 'delete': cache_instance.delete(key) # Opcional: introduzir um pequeno sleep aleatório para aumentar a troca de contexto e a contenção # time.sleep(random.uniform(0.00001, 0.0001)) # print(f"Thread {thread_id} terminou.") # Descomente para saída detalhada threads = [] for i in range(MT_NUM_THREADS): thread = threading.Thread(target=mt_worker, args=(mt_cache, i, barrier, error_flag, global_ever_inserted_keys, global_ever_inserted_keys_lock)) threads.append(thread) thread.start() # Espera que todas as threads de trabalho cheguem à barreira e iniciem as operações barrier.wait() print(f"Todas as {MT_NUM_THREADS} threads iniciaram as operações.") # Espera que todas as threads de trabalho completem as suas operações for thread in threads: thread.join() print("Todas as threads terminaram.") # Afirmações finais para testes multithread assert not error_flag.is_set(), "MT: Erros detetados durante operações multithread. Verifique a consola para detalhes." assert mt_cache._size <= MT_CAPACITY, f"MT: O tamanho do cache ({mt_cache._size}) excedeu a capacidade ({MT_CAPACITY})." # Verifica que todas as chaves restantes no cache foram inseridas em algum momento final_keys = mt_cache.keys() with global_ever_inserted_keys_lock: for key in final_keys: assert key in global_ever_inserted_keys, f"MT Error: O cache final contém a chave '{key}' que nunca foi inserida globalmente." print("Testes multithread aprovados!") print("\n--- Todos os testes da cache LRU foram aprovados! ---") if __name__ == '__main__': run_tests()

from __future__ import annotations import random import threading import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional class RWLock: """ Um bloqueio leitores-escritor simples com preferência para escritores. Estratégia de concorrência: - Vários leitores podem manter o bloqueio simultaneamente. - Escritores são exclusivos. - Assim que um escritor estiver esperando, novos leitores são bloqueados para que os escritores não sofram starvation. Isto é usado para proteger o dicionário que mapeia chaves para nós. Leituras que só precisam de verificação de pertencimento/consulta de valor podem prosseguir concorrentemente. Operações que modificam o índice ou a lista LRU usam o bloqueio de escrita. Observação: O Python não fornece primitivas realmente lock-free para construir uma LRU totalmente lock-free na biblioteca padrão. Em vez disso, esta implementação evita um único mutex global para todo acesso usando: - um bloqueio leitores-escritor para o índice chave->nó - um bloqueio separado para a lista LRU duplamente encadeada - bloqueios por nó para estado local ao nó Isso oferece concorrência substancialmente melhor do que um único bloqueio global enquanto preserva a correção sob contenção. """ def __init__(self) -> None: self._cond = threading.Condition(threading.Lock()) self._readers = 0 self._writer = False self._writers_waiting = 0 def acquire_read(self) -> None: with self._cond: while self._writer or self._writers_waiting > 0: self._cond.wait() self._readers += 1 def release_read(self) -> None: with self._cond: self._readers -= 1 if self._readers == 0: self._cond.notify_all() def acquire_write(self) -> None: with self._cond: self._writers_waiting += 1 try: while self._writer or self._readers > 0: self._cond.wait() self._writer = True finally: self._writers_waiting -= 1 def release_write(self) -> None: with self._cond: self._writer = False self._cond.notify_all() @dataclass class _Node: key: str value: Any prev: Optional["_Node"] = None next: Optional["_Node"] = None deleted: bool = False lock: threading.RLock = field(default_factory=threading.RLock) class ConcurrentLRUCache: """ Cache LRU thread-safe com sincronização de granularidade fina. Visão geral do design --------------------- Estruturas de dados: - self._map: dict[str, _Node] Mapeia chaves para nós. Protegido por self._index_lock (RWLock). - Lista duplamente encadeada com sentinelas self._head (lado MRU) e self._tail (lado LRU) Protegida por self._list_lock. Regras de bloqueio ------------------ 1. Leituras do dicionário/índice usam self._index_lock.acquire_read(). 2. Mutações do dicionário/índice usam self._index_lock.acquire_write(). 3. Quaisquer mudanças de ponteiros na lista encadeada usam self._list_lock. 4. Bloqueios por nó serializam o acesso ao valor/estado deleted de um nó. Regra de ordenação para evitar deadlock --------------------------------------- Quando vários bloqueios são necessários, as operações seguem uma ordem estável: bloqueio do índice -> bloqueio da lista -> bloqueio(s) do nó Trade-offs de concorrência -------------------------- - get() precisa atualizar a recência, então não pode ser uma leitura pura. Primeiro ele faz uma consulta concorrente ao índice sob um bloqueio de leitura, depois adquire brevemente o bloqueio da lista para mover o nó localizado para a posição MRU. - put(), delete() e eviction modificam tanto o índice quanto a lista, então adquirem o bloqueio de escrita do índice e o bloqueio da lista. - Isto não é estritamente lock-free; a biblioteca padrão do Python não torna uma LRU prática baseada em lista encadeada lock-free viável. Ainda assim, é mais concorrente do que um design que usa um mutex global para cada operação. """ def __init__(self, capacity: int): """Inicializa o cache com uma capacidade máxima positiva.""" if not isinstance(capacity, int) or capacity <= 0: raise ValueError("capacity must be a positive integer") self._capacity = capacity self._map: Dict[str, _Node] = {} self._index_lock = RWLock() self._list_lock = threading.RLock() self._head = _Node("__HEAD__", None) self._tail = _Node("__TAIL__", None) self._head.next = self._tail self._tail.prev = self._head def _insert_after_head_unlocked(self, node: _Node) -> None: """Insere o nó imediatamente após head. O chamador deve manter self._list_lock.""" first = self._head.next node.prev = self._head node.next = first self._head.next = node if first is not None: first.prev = node def _detach_unlocked(self, node: _Node) -> None: """Desanexa o nó da lista. O chamador deve manter self._list_lock.""" prev_node = node.prev next_node = node.next if prev_node is not None: prev_node.next = next_node if next_node is not None: next_node.prev = prev_node node.prev = None node.next = None def _move_to_front_if_present(self, node: _Node) -> None: """ Move um nó existente para a posição MRU se ele ainda estiver ativo. Usa o bloqueio da lista mais o bloqueio do nó. O bloqueio de leitura do mapa mantido pelo chamador garante que o objeto nó permaneça alcançável a partir do índice durante a consulta. """ with self._list_lock: with node.lock: if node.deleted: return if node.prev is self._head: return if node.prev is not None and node.next is not None: self._detach_unlocked(node) self._insert_after_head_unlocked(node) def get(self, key: str) -> Optional[Any]: """ Retorna o valor de key se presente e o marca como o mais recentemente usado. Retorna None se ausente. """ self._index_lock.acquire_read() try: node = self._map.get(key) if node is None: return None with node.lock: if node.deleted: return None value = node.value self._move_to_front_if_present(node) return value finally: self._index_lock.release_read() def put(self, key: str, value: Any) -> None: """ Insere ou atualiza um par chave-valor. Se a capacidade for excedida, remove o item menos recentemente usado. """ self._index_lock.acquire_write() try: with self._list_lock: existing = self._map.get(key) if existing is not None: with existing.lock: if not existing.deleted: existing.value = value if existing.prev is not self._head: if existing.prev is not None and existing.next is not None: self._detach_unlocked(existing) self._insert_after_head_unlocked(existing) return else: self._map.pop(key, None) node = _Node(key, value) self._map[key] = node self._insert_after_head_unlocked(node) while len(self._map) > self._capacity: victim = self._tail.prev if victim is None or victim is self._head: break with victim.lock: if victim.deleted: if victim.prev is not None and victim.next is not None: self._detach_unlocked(victim) self._map.pop(victim.key, None) continue victim.deleted = True self._detach_unlocked(victim) self._map.pop(victim.key, None) finally: self._index_lock.release_write() def delete(self, key: str) -> bool: """ Remove key do cache. Retorna True se key existia; caso contrário, False. """ self._index_lock.acquire_write() try: with self._list_lock: node = self._map.pop(key, None) if node is None: return False with node.lock: if node.deleted: return False node.deleted = True if node.prev is not None and node.next is not None: self._detach_unlocked(node) return True finally: self._index_lock.release_write() def keys(self) -> List[str]: """ Retorna as chaves da mais recentemente usada para a menos recentemente usada. Adquirimos o bloqueio de leitura do índice e o bloqueio da lista para obter um snapshot consistente. """ result: List[str] = [] self._index_lock.acquire_read() try: with self._list_lock: cur = self._head.next while cur is not None and cur is not self._tail: with cur.lock: if not cur.deleted: result.append(cur.key) cur = cur.next return result finally: self._index_lock.release_read() def _debug_size(self) -> int: """Helper interno para testes.""" self._index_lock.acquire_read() try: return len(self._map) finally: self._index_lock.release_read() def _debug_validate(self) -> None: """ Verificador de consistência interna para testes. Verifica: - os links da lista estão bem formados - todo nó ativo da lista está presente no mapa - não há chaves duplicadas na lista - o tamanho nunca excede a capacidade """ self._index_lock.acquire_read() try: with self._list_lock: assert len(self._map) <= self._capacity, "cache exceeded capacity" seen = set() cur = self._head assert self._head.prev is None assert self._tail.next is None while cur is not None: nxt = cur.next if nxt is not None: assert nxt.prev is cur, "broken backward pointer" if cur not in (self._head, self._tail): with cur.lock: assert not cur.deleted, "deleted node remained in list" assert cur.key in self._map, "live list node missing from map" assert self._map[cur.key] is cur, "map points to wrong node" assert cur.key not in seen, "duplicate key in list" seen.add(cur.key) if cur is self._tail: break cur = nxt assert seen == set(self._map.keys()), "map/list key mismatch" finally: self._index_lock.release_read() def run_tests() -> None: """ Executa testes single-threaded e multi-threaded. Requisitos cobertos pelos testes multi-threaded: - pelo menos 8 threads - mistura de get/put/delete em chaves sobrepostas - garantir que o cache nunca exceda a capacidade - garantir que get nunca retorne um valor para uma chave que nunca foi inserida """ # Testes single-threaded c = ConcurrentLRUCache(2) assert c.get("missing") is None c.put("a", 1) c.put("b", 2) assert c.keys() == ["b", "a"] assert c.get("a") == 1 assert c.keys() == ["a", "b"] c.put("c", 3) assert c.get("b") is None assert c.keys() == ["c", "a"] c.put("a", 10) assert c.get("a") == 10 assert c.keys()[0] == "a" assert c.delete("a") is True assert c.delete("a") is False assert c.get("a") is None assert c.keys() == ["c"] c1 = ConcurrentLRUCache(1) c1.put("x", 1) assert c1.keys() == ["x"] c1.put("y", 2) assert c1.get("x") is None assert c1.get("y") == 2 assert c1.keys() == ["y"] c1.put("y", 3) assert c1.get("y") == 3 assert c1.delete("z") is False c1._debug_validate() # Testes concorrentes capacity = 16 cache = ConcurrentLRUCache(capacity) inserted_ever = set() inserted_lock = threading.Lock() errors: List[str] = [] errors_lock = threading.Lock() stop_flag = threading.Event() key_space = [f"k{i}" for i in range(24)] def record_error(msg: str) -> None: with errors_lock: errors.append(msg) def worker(tid: int) -> None: rnd = random.Random(1000 + tid) try: for i in range(2000): key = key_space[rnd.randrange(len(key_space))] op = rnd.random() if op < 0.45: value = (tid, i) cache.put(key, value) with inserted_lock: inserted_ever.add(key) elif op < 0.80: value = cache.get(key) if value is not None: with inserted_lock: if key not in inserted_ever: record_error(f"get returned value for never-inserted key: {key}") else: cache.delete(key) if i % 100 == 0: size = cache._debug_size() if size > capacity: record_error(f"capacity exceeded: {size} > {capacity}") cache._debug_validate() except Exception as exc: record_error(f"worker {tid} failed: {exc!r}") finally: if stop_flag.is_set(): return def validator() -> None: try: for _ in range(400): cache._debug_validate() size = cache._debug_size() if size > capacity: record_error(f"capacity exceeded in validator: {size} > {capacity}") time.sleep(0.002) except Exception as exc: record_error(f"validator failed: {exc!r}") finally: stop_flag.set() threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)] threads += [threading.Thread(target=validator)] for t in threads: t.start() for t in threads: t.join() cache._debug_validate() assert cache._debug_size() <= capacity assert not errors, "\n".join(errors) # Teste focado de contenção na mesma chave hot = ConcurrentLRUCache(4) hot.put("shared", 0) def same_key_worker(tid: int) -> None: rnd = random.Random(2000 + tid) for i in range(1000): if rnd.random() < 0.5: hot.put("shared", (tid, i)) else: _ = hot.get("shared") if rnd.random() < 0.1: hot.delete("shared") hot.put("shared", (tid, i, "reinsert")) ts = [threading.Thread(target=same_key_worker, args=(i,)) for i in range(8)] for t in ts: t.start() for t in ts: t.join() hot._debug_validate() assert hot._debug_size() <= 4 assert hot.get("shared") is not None print("All tests passed.") if __name__ == "__main__": run_tests()