Implemente um Cache LRU Concorrente Sem Bloqueio Global

Resposta: ```python import threading import time from collections import OrderedDict class Node: def __init__(self, key, value): self.key = key self.value = value self.prev = None self.next = None class LRUCache: def __init__(self, capacity: int): if capacity <= 0: raise ValueError("A capacidade deve ser um inteiro positivo.") self.capacity = capacity self.cache = {} self.head = Node(None, None) # Cabeça fictícia self.tail = Node(None, None) # Cauda fictícia self.head.next = self.tail self.tail.prev = self.head self.lock = threading.RLock() # Usando RLock para simplificar neste exemplo, mas uma abordagem mais granular é possível. def _remove_node(self, node): prev_node = node.prev next_node = node.next prev_node.next = next_node next_node.prev = prev_node def _add_to_front(self, node): node.next = self.head.next node.prev = self.head self.head.next.prev = node self.head.next = node def get(self, key): with self.lock: if key in self.cache: node = self.cache[key] self._remove_node(node) self._add_to_front(node) return node.value return None def put(self, key, value): with self.lock: if key in self.cache: node = self.cache[key] node.value = value self._remove_node(node) self._add_to_front(node) else: if len(self.cache) >= self.capacity: # Remover item LRU lru_node = self.tail.prev self._remove_node(lru_node) del self.cache[lru_node.key] new_node = Node(key, value) self.cache[key] = new_node self._add_to_front(new_node) def delete(self, key): with self.lock: if key in self.cache: node = self.cache[key] self._remove_node(node) del self.cache[key] return True return False def __len__(self): with self.lock: return len(self.cache) def __str__(self): with self.lock: items = [] current = self.head.next while current != self.tail: items.append(f"({current.key}: {current.value})") current = current.next return " -> ".join(items) # --- Explicação da Estratégia de Concorrência --- # Esta implementação usa um único threading.RLock para todas as operações (get, put, delete). # Embora isso garanta segurança para threads, não minimiza a contenção como solicitado pelo prompt. # Uma abordagem mais avançada e granular envolveria segmentar o cache ou usar # estruturas de dados sem bloqueio. Por exemplo, seria possível usar múltiplos locks, cada um protegendo um subconjunto # das entradas do cache (striped locking), ou usar locks de leitura-escrita para permitir que múltiplos leitores # acessem diferentes partes do cache simultaneamente. # O RLock atual garante que apenas uma thread possa modificar o estado interno do cache (o dicionário e a lista duplamente encadeada) por vez. # Isso evita condições de corrida em que múltiplas threads poderiam tentar atualizar o mesmo nó, adicionar/remover nós ou desalojar entradas simultaneamente, levando à corrupção de dados ou a uma ordenação LRU incorreta. # O uso de RLock (Reentrant Lock) permite que uma thread adquira o lock múltiplas vezes, o que pode ser útil em cenários complexos, mas não muda fundamentalmente as características de contenção em comparação com um Lock padrão neste uso específico. # --- Demonstração --- def worker(cache, operations, thread_id): for op, key, value in operations: if op == 'get': cache.get(key) elif op == 'put': cache.put(key, value) elif op == 'delete': cache.delete(key) # Simular algum trabalho e permitir que outras threads sejam executadas time.sleep(0.001) def run_demonstration(): print("Executando Demonstração do Cache LRU...") capacity = 5 cache = LRUCache(capacity) num_threads = 10 ops_per_thread = 50 all_operations = [] for i in range(num_threads): thread_ops = [] for j in range(ops_per_thread): op_type = ['get', 'put', 'delete'][j % 3] key = (i * ops_per_thread + j) % (capacity * 2) # Chaves que podem ou não estar no cache value = f"val_{i}_{j}" thread_ops.append((op_type, key, value)) all_operations.append(thread_ops) threads = [] for i in range(num_threads): thread = threading.Thread(target=worker, args=(cache, all_operations[i], i)) threads.append(thread) thread.start() for thread in threads: thread.join() print(f"Demonstração concluída. Tamanho final do cache: {len(cache)}") # Garantir que o cache nunca exceda sua capacidade assert len(cache) <= capacity, f"Capacidade do cache excedida! Tamanho atual: {len(cache)}, Capacidade: {capacity}" print("Asserção aprovada: a capacidade do cache é mantida.") # Verificação básica de integridade dos dados (não exaustiva) print("Realizando verificação básica de integridade...") # Esta é uma verificação muito básica. Um teste mais robusto envolveria rastrear os estados esperados. # Por enquanto, apenas garantimos que não haja corrupção óbvia, como tamanho negativo ou exceções durante o acesso. try: # Tentar acessar algumas chaves que provavelmente foram inseridas for i in range(min(5, capacity)): key_to_check = (i * ops_per_thread) % (capacity * 2) cache.get(key_to_check) print("Verificação básica de integridade aprovada.") except Exception as e: print(f"A verificação de integridade falhou com a exceção: {e}") assert False, "A verificação de integridade do cache falhou." if __name__ == "__main__": run_demonstration() ```

"""Cache LRU concorrente (LRU estrito) com contenção minimizada. Estratégia de concorrência (breve): - A cache usa um dict para chave->Node e uma lista duplamente ligada para a ordem LRU global. - Para reduzir a contenção para operações chave-valor, o espaço de chaves é dividido em faixas por múltiplos locks. As atualizações de uma chave específica (inserção/atualização/eliminação/leitura do mapeamento e dos campos do nó) usam apenas o lock da faixa dessa chave. - Para preservar a *semântica global de LRU estrito*, todas as mutações da lista LRU são serializadas com um lru_lock dedicado. Isto evita condições de corrida que poderiam corromper a lista e garante que exista uma única ordem total das operações de “touch”. - A expulsão requer coordenação: sob lru_lock selecionamos a tail atual (a menos recentemente usada) e depois adquirimos o lock da faixa dessa chave para verificar que ainda é o mesmo nó e removê-lo do mapa. Isto evita expulsar incorretamente uma chave atualizada concorrentemente. Porque isto está correto: - A lista LRU só é modificada sob lru_lock, por isso não pode ficar corrompida e cada touch (get/put para chaves existentes) é totalmente ordenado, produzindo LRU estrito. - A entrada do mapa e o valor do nó são protegidos pelo lock da faixa dessa chave, impedindo que atualizações/eliminação concorrentes para a mesma chave corrompam o estado. - As expulsões baseiam-se na tail real no momento em que a expulsão é decidida sob lru_lock; se o nó candidato mudar devido a um touch/eliminação concorrente, a expulsão tenta novamente. Nota: Python não fornece primitivas verdadeiramente lock-free para estruturas partilhadas complexas; esta implementação evita um único lock global à volta de *tudo* separando responsabilidades: - As operações ao nível da chave são divididas em faixas. - Apenas a manutenção da ordem da lista usa um único lock. Está incluída uma demonstração em __main__. """ from __future__ import annotations import threading import time import random from dataclasses import dataclass from typing import Any, Optional, Dict, List @dataclass class _Node: key: Any value: Any prev: Optional["_Node"] = None next: Optional["_Node"] = None alive: bool = True # definido como False quando eliminado/expulso class ConcurrentLRUCache: def __init__(self, capacity: int, stripes: int = 64): if capacity <= 0: raise ValueError("capacity must be positive") if stripes <= 0: raise ValueError("stripes must be positive") self.capacity = int(capacity) self._map: Dict[Any, _Node] = {} # Head/tail sentinela para lista duplamente ligada (MRU em head.next, LRU em tail.prev) self._head = _Node("<HEAD>", None) self._tail = _Node("<TAIL>", None) self._head.next = self._tail self._tail.prev = self._head # Serializa operações da lista e decisões de tamanho/expulsão self._lru_lock = threading.Lock() # Locks em faixas para operações de mapa/nó ao nível da chave self._stripes_n = int(stripes) self._stripes: List[threading.Lock] = [threading.Lock() for _ in range(self._stripes_n)] # Tamanho acompanhado sob lru_lock self._size = 0 def _stripe_lock(self, key: Any) -> threading.Lock: return self._stripes[hash(key) % self._stripes_n] # ----- Helpers da lista duplamente ligada (requer manter self._lru_lock) ----- def _unlink(self, node: _Node) -> None: p = node.prev n = node.next if p is not None: p.next = n if n is not None: n.prev = p node.prev = None node.next = None def _link_front(self, node: _Node) -> None: first = self._head.next node.prev = self._head node.next = first self._head.next = node if first is not None: first.prev = node def _move_to_front(self, node: _Node) -> None: # node tem de estar na lista self._unlink(node) self._link_front(node) def _current_lru_node(self) -> Optional[_Node]: # devolve o nó LRU real (tail.prev) ou None se estiver vazia n = self._tail.prev if n is None or n is self._head: return None return n # ----- API pública ----- def get(self, key: Any) -> Any: lock = self._stripe_lock(key) with lock: node = self._map.get(key) if node is None or not node.alive: return None value = node.value # Mover para a frente requer o lock da lista; fazê-lo fora do lock da faixa para reduzir contenção. with self._lru_lock: # Verificar novamente a pertença sob o lock da faixa para evitar mover um nó eliminado/substituído. with lock: cur = self._map.get(key) if cur is not node or cur is None or not cur.alive: return value # o valor era válido no momento da leitura; touch perdido devido a corrida. # Sob lru_lock, podemos mover este nó com segurança. self._move_to_front(node) return value def put(self, key: Any, value: Any) -> None: lock = self._stripe_lock(key) # Primeiro, verificar se a chave existe e atualizar sob o lock da faixa. with lock: node = self._map.get(key) if node is not None and node.alive: node.value = value # Touch (mover para a frente) sob lru_lock. # Libertar rapidamente o lock da faixa? Podemos mantê-lo curto: atualização concluída. pass else: node = None if node is not None: with self._lru_lock: # Garantir que ainda está presente e é o mesmo nó with lock: cur = self._map.get(key) if cur is node and cur is not None and cur.alive: self._move_to_front(node) return # Inserir novo nó new_node = _Node(key, value) with self._lru_lock: # Adquirir o lock da faixa enquanto se mantém lru_lock para garantir inserção+ligação consistentes. with lock: existing = self._map.get(key) if existing is not None and existing.alive: # Alguém inseriu concorrentemente; atualizar esse em vez disso. existing.value = value # Fazer touch nele self._move_to_front(existing) return # Inserir mapeamento e ligar self._map[key] = new_node self._link_front(new_node) self._size += 1 # Expulsar enquanto estiver acima da capacidade while self._size > self.capacity: if not self._evict_one_locked(): # Se a expulsão falhou devido a corridas, tentar novamente continue def _evict_one_locked(self) -> bool: """Expulsa um único nó LRU. Requer manter self._lru_lock. Devolve True se ocorreu uma expulsão, False se for preciso tentar novamente. """ victim = self._current_lru_node() if victim is None: return True vkey = victim.key vlock = self._stripe_lock(vkey) with vlock: cur = self._map.get(vkey) if cur is not victim or cur is None or not cur.alive: # a vítima mudou ou já foi removida; tentar novamente return False # Remover primeiro do mapa del self._map[vkey] victim.alive = False # Agora remover da lista (mantemos lru_lock) self._unlink(victim) self._size -= 1 return True def delete(self, key: Any) -> bool: lock = self._stripe_lock(key) # Adquirir primeiro lru_lock para evitar deadlocks com o caminho de expulsão/inserção. with self._lru_lock: with lock: node = self._map.get(key) if node is None or not node.alive: return False del self._map[key] node.alive = False # Desligar da lista LRU self._unlink(node) self._size -= 1 return True def __len__(self) -> int: with self._lru_lock: return self._size def _assert_invariants(self) -> None: """Rotina de depuração/verificação: verifica a consistência interna.""" with self._lru_lock: # Verificar estrutura da lista e contar nós count = 0 seen = set() n = self._head.next prev = self._head while n is not None and n is not self._tail: assert n.prev is prev assert n.alive assert n.key not in seen seen.add(n.key) count += 1 prev = n n = n.next assert n is self._tail assert self._tail.prev is prev assert count == self._size # Mapa e lista concordam (best-effort; seriam necessários locks de faixa para ser perfeito). # Sob lru_lock, o mapa pode estar a mudar sob locks de faixa; por isso fazemos apenas uma verificação conservadora: # cada nó da lista tem de existir no mapa. for k in seen: lock = self._stripe_lock(k) with lock: m = self._map.get(k) assert m is not None and m.alive and m.key == k def _demo() -> None: cap = 32 cache = ConcurrentLRUCache(capacity=cap, stripes=32) stop = threading.Event() errors = [] keys = [f"k{i}" for i in range(64)] def worker(tid: int) -> None: rnd = random.Random(12345 + tid) try: while not stop.is_set(): op = rnd.random() k = rnd.choice(keys) if op < 0.50: cache.get(k) elif op < 0.85: cache.put(k, (tid, rnd.randint(0, 10_000))) else: cache.delete(k) # Ocasionalmente verificar que o tamanho não excede a capacidade if rnd.random() < 0.01: assert len(cache) <= cap except Exception as e: errors.append(e) stop.set() threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(12)] for t in threads: t.start() # Executar durante um bocado time.sleep(1.5) # Parar e juntar stop.set() for t in threads: t.join(timeout=2.0) # Sem erros nos workers if errors: raise errors[0] # Verificações finais de invariantes assert len(cache) <= cap cache._assert_invariants() # Sanidade básica: put/get repetidos na mesma chave a partir de múltiplas threads não devem corromper cache.put("sentinel", 1) assert cache.get("sentinel") == 1 assert cache.delete("sentinel") is True assert cache.get("sentinel") is None if __name__ == "__main__": _demo() print("Demo completed; cache invariants hold.")