Implémenter un cache LRU concurrent sans verrou global

import threading import random import time from typing import Any, Optional, List class _Node: """ Représente un nœud dans la liste doublement chaînée utilisée pour l'ordre LRU. Utilise __slots__ pour l'efficacité de la mémoire. """ __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: """ Une implémentation personnalisée de verrou Lecture-Écriture. Ce verrou permet à plusieurs lecteurs de détenir le verrou simultanément, mais à un seul écrivain à la fois. Un écrivain doit attendre que tous les lecteurs aient terminé, et les lecteurs doivent attendre qu'un écrivain ait terminé. Il s'agit d'une implémentation privilégiant les lecteurs, ce qui signifie que les nouveaux lecteurs peuvent acquérir le verrou de lecture même si un écrivain attend. Cette classe fournit également des gestionnaires de contexte pour une acquisition et une libération plus propres du verrou. """ def __init__(self): self._read_lock = threading.Lock() # Protège le compteur _readers et la variable de condition self._write_lock = threading.Lock() # Le verrou d'écriture réel self._readers = 0 # Nombre de lecteurs actifs self._no_readers = threading.Condition(self._read_lock) # Condition pour que les écrivains attendent les lecteurs def acquire_read(self): """Acquiert un verrou de lecture. Plusieurs lecteurs peuvent détenir ce verrou simultanément.""" with self._read_lock: # Attend si un écrivain détient actuellement le verrou d'écriture. # Les nouveaux lecteurs seront bloqués si un écrivain est actif. while self._write_lock.locked(): self._no_readers.wait() self._readers += 1 def release_read(self): """Libère un verrou de lecture.""" with self._read_lock: self._readers -= 1 # S'il ne reste plus de lecteurs, notifier les écrivains en attente if self._readers == 0: self._no_readers.notify_all() def acquire_write(self): """Acquiert un verrou d'écriture. Un seul écrivain peut détenir ce verrou à la fois.""" # Acquiert d'abord le verrou d'écriture principal pour signaler l'intention d'écrire self._write_lock.acquire() with self._read_lock: # Puis acquiert le verrou de lecture pour attendre les lecteurs actifs # Attend jusqu'à ce que tous les lecteurs actifs aient terminé while self._readers > 0: self._no_readers.wait() def release_write(self): """Libère un verrou d'écriture.""" with self._read_lock: # Acquiert le verrou de lecture pour notifier les lecteurs/écrivains potentiels self._no_readers.notify_all() # Notifie les lecteurs ou écrivains en attente self._write_lock.release() # Libère le verrou d'écriture principal 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): """Retourne un gestionnaire de contexte pour acquérir un verrou de lecture.""" return self._ReadContextManager(self) def write_locked(self): """Retourne un gestionnaire de contexte pour acquérir un verrou d'écriture.""" return self._WriteContextManager(self) class LRUCache: """ Une implémentation de cache LRU (Least Recently Used) thread-safe. Ce cache utilise un ReadWriteLock pour gérer l'accès concurrent. La conception vise à permettre des lectures concurrentes pour les misses de cache et l'opération `keys()`. Cependant, les opérations qui modifient l'état du cache ou l'ordre LRU (`put`, `delete`, et les hits de cache dans `get`) acquièrent un verrou d'écriture, garantissant l'atomicité et la cohérence pour ces sections critiques. Cette approche offre une meilleure concurrence qu'un seul mutex global pour toutes les opérations, en particulier pour les charges de travail fortement axées sur la lecture avec des misses de cache fréquents ou des appels à `keys()`. L'opération `get` utilise un modèle de "mise à niveau" : elle tente d'abord un verrou de lecture pour vérifier l'existence de la clé. Si elle est trouvée, elle libère le verrou de lecture et acquiert un verrou d'écriture pour mettre à jour l'ordre LRU et récupérer la dernière valeur. Ce modèle est soigneusement implémenté pour gérer les conditions de concurrence qui pourraient survenir lors de la mise à niveau du verrou (par exemple, la clé étant supprimée ou mise à jour par un autre thread). """ def __init__(self, capacity: int): if capacity <= 0: raise ValueError("La capacité doit être un entier positif.") self._capacity = capacity self._cache = {} # Stocke clé -> _Node pour une recherche en O(1) self._size = 0 # Nœuds tête et queue factices pour la liste doublement chaînée. # Cela simplifie les cas limites pour l'ajout/suppression de nœuds, car les nœuds de données réels # sont toujours entre la tête et la queue. self._head = _Node("dummy_head", None) self._tail = _Node("dummy_tail", None) self._head.next = self._tail self._tail.prev = self._head # Contrôle de concurrence : un seul ReadWriteLock protège l'état complet du cache. self._rw_lock = ReadWriteLock() def _add_node(self, node: _Node): """Ajoute un nœud juste après la tête (position la plus récemment utilisée). Suppose que le verrou d'écriture est détenu.""" node.prev = self._head node.next = self._head.next self._head.next.prev = node self._head.next = node def _remove_node(self, node: _Node): """Supprime un nœud de la liste doublement chaînée. Suppose que le verrou d'écriture est détenu.""" 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): """Déplace un nœud existant vers l'avant de la liste (le plus récemment utilisé). Suppose que le verrou d'écriture est détenu.""" self._remove_node(node) self._add_node(node) def _pop_tail(self) -> _Node: """Supprime et retourne le nœud le moins récemment utilisé (celui juste avant la queue). Suppose que le verrou d'écriture est détenu.""" node = self._tail.prev self._remove_node(node) return node def get(self, key: str) -> Optional[Any]: """ Retourne la valeur associée à la clé si elle existe, et la marque comme récemment utilisée. Retourne None si la clé n'est pas dans le cache. Cette opération acquiert d'abord un verrou de lecture pour vérifier l'existence de la clé. Si la clé est trouvée, elle libère le verrou de lecture puis acquiert un verrou d'écriture pour mettre à jour l'ordre LRU et récupérer la dernière valeur. Ce "modèle de mise à niveau" permet des opérations `get` concurrentes manquantes et des opérations `keys()`, mais sérialise les `get` réussis avec les autres écritures. """ 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 # Si la clé est trouvée, acquiert le verrou d'écriture pour mettre à jour l'ordre LRU et obtenir la dernière valeur. # C'est un "modèle de mise à niveau". Une condition de concurrence peut survenir ici : un autre thread # pourrait supprimer ou mettre à jour la clé entre la libération du verrou de lecture et l'acquisition du verrou d'écriture. with self._rw_lock.write_locked(): # Revérifie si la clé existe toujours et fait référence au même objet nœud. # Ceci est crucial pour la correction sous contention. Si un autre thread a effectué # un `put` pour la même clé, `_cache.get(key)` retournerait un *nouveau* nœud. # Si un autre thread a effectué un `delete`, il retournerait `None`. current_node = self._cache.get(key) if current_node is node_found_under_read_lock: # S'assure que c'est le même objet nœud self._move_to_front(node_found_under_read_lock) value = current_node.value # Obtient la dernière valeur après le déplacement return value else: # Le nœud a été soit supprimé, soit remplacé par une autre opération `put` # entre l'acquisition de notre verrou de lecture et de verrou d'écriture. Traite comme un miss. return None def put(self, key: str, value: Any) -> None: """ Insère ou met à jour la paire clé-valeur. Si le cache dépasse la capacité après l'insertion, il évince l'élément le moins récemment utilisé. C'est une opération d'écriture et elle acquiert un verrou d'écriture. """ with self._rw_lock.write_locked(): node = self._cache.get(key) if node: # La clé existe : met à jour la valeur et déplace vers l'avant node.value = value self._move_to_front(node) else: # La clé n'existe pas : ajoute un nouveau nœud new_node = _Node(key, value) self._cache[key] = new_node self._add_node(new_node) self._size += 1 # Évince si la capacité est dépassée 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: """ Supprime la clé du cache. Retourne True si la clé était présente, False sinon. C'est une opération d'écriture et elle acquiert un verrou d'écriture. """ 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]: """ Retourne une liste de toutes les clés actuellement dans le cache, ordonnées de la plus récemment utilisée à la moins récemment utilisée. C'est une opération de lecture et elle acquiert un verrou de lecture. """ 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(): """ Exécute une série de tests mono-thread et multi-thread pour vérifier l'implémentation LRUCache. """ print("--- Démarrage des tests du cache LRU ---") # --- Tests mono-thread --- print("\n--- Tests mono-thread ---") cache = LRUCache(capacity=3) assert cache.get("a") is None, "ST: Obtenir une clé inexistante doit retourner None" assert cache.keys() == [], "ST: Un cache vide doit avoir une liste de clés vide" cache.put("a", 1) assert cache.get("a") == 1, "ST: Obtenir une clé existante après put" assert cache.keys() == ["a"], "ST: Liste des clés après un put" cache.put("b", 2) cache.put("c", 3) assert cache.keys() == ["c", "b", "a"], "ST: Liste des clés après plusieurs puts" assert cache._size == 3, "ST: La taille du cache doit être 3" # Test d'éviction de capacité cache.put("d", 4) # 'a' doit être évincé car c'est le LRU assert cache.get("a") is None, "ST: La clé évincée 'a' doit être None" assert cache.get("d") == 4, "ST: La nouvelle clé 'd' doit être présente" assert cache.keys() == ["d", "c", "b"], "ST: Liste des clés après éviction" assert cache._size == 3, "ST: La taille du cache doit rester 3 après éviction" # Test de put avec clé existante (mise à jour de la valeur et déplacement vers l'avant) cache.put("b", 22) # 'b' doit se déplacer vers l'avant, valeur mise à jour assert cache.get("b") == 22, "ST: Valeur mise à jour pour 'b'" assert cache.keys() == ["b", "d", "c"], "ST: Liste des clés après mise à jour et déplacement de 'b'" # Test de suppression assert cache.delete("c") is True, "ST: Supprimer la clé existante 'c'" assert cache.get("c") is None, "ST: La clé supprimée 'c' doit être None" assert cache.keys() == ["b", "d"], "ST: Liste des clés après suppression de 'c'" assert cache._size == 2, "ST: Taille du cache après suppression" assert cache.delete("z") is False, "ST: Supprimer la clé inexistante 'z'" assert cache.keys() == ["b", "d"], "ST: Liste des clés inchangée après suppression de clé inexistante" # Test de capacité 1 cache_cap1 = LRUCache(capacity=1) cache_cap1.put("x", 10) assert cache_cap1.get("x") == 10, "ST: Capacité 1, obtenir 'x'" cache_cap1.put("y", 20) # 'x' doit être évincé assert cache_cap1.get("x") is None, "ST: Capacité 1, 'x' évincé" assert cache_cap1.get("y") == 20, "ST: Capacité 1, obtenir 'y'" assert cache_cap1.keys() == ["y"], "ST: Capacité 1, liste des clés" print("Tests mono-thread réussis !") # --- Tests multi-thread --- print("\n--- Tests multi-thread ---") MT_CAPACITY = 10 MT_NUM_THREADS = 8 MT_OPS_PER_THREAD = 2000 MT_KEY_RANGE = 20 # Clés de 0 à 19 (par exemple, "0", "1", ..., "19") mt_cache = LRUCache(capacity=MT_CAPACITY) # Barrière pour s'assurer que tous les threads de travail démarrent leurs opérations simultanément barrier = threading.Barrier(MT_NUM_THREADS + 1) # +1 pour le thread principal error_flag = threading.Event() # Défini si un thread rencontre une erreur # Suit les clés qui ont *jamais* été insérées avec succès dans le cache par un thread. # Utilisé pour vérifier que `get` ne retourne jamais une valeur pour une clé qui n'a jamais été réellement insérée. 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): """Fonction de travail pour les tests multi-thread.""" barrier_instance.wait() # Attend que tous les threads soient prêts à démarrer les opérations for i in range(MT_OPS_PER_THREAD): if error_event.is_set(): break # Arrête si un autre thread a déjà trouvé une erreur # Choisit aléatoirement une opération avec un biais vers 'get' et '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: # Vérifie que si une valeur est retournée, la clé doit avoir été insérée à un moment donné. with ever_inserted_keys_lock: if key not in ever_inserted_keys_set: print(f"Erreur MT : Le thread {thread_id} a obtenu la valeur '{val}' pour la clé '{key}' " f"qui n'a jamais été insérée globalement. Cela ne devrait pas arriver.") error_event.set() break # Vérifie que la valeur n'est pas corrompue (vérification simple du format) if not isinstance(val, str) or not val.startswith("value_T"): print(f"Erreur MT : Le thread {thread_id} a obtenu une valeur corrompue '{val}' pour la clé '{key}'.") error_event.set() break elif op == 'delete': cache_instance.delete(key) # Optionnel : Introduire un petit sommeil aléatoire pour augmenter le changement de contexte et la contention # time.sleep(random.uniform(0.00001, 0.0001)) # print(f"Le thread {thread_id} a terminé.") # Décommenter pour une sortie détaillée theads = [] 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() # Attend que tous les threads de travail atteignent la barrière et démarrent les opérations barrier.wait() print(f"Les {MT_NUM_THREADS} threads ont démarré les opérations.") # Attend que tous les threads de travail terminent leurs opérations for thread in threads: thread.join() print("Tous les threads ont terminé.") # Assertions finales pour les tests multi-thread assert not error_flag.is_set(), "MT : Des erreurs ont été détectées pendant les opérations multi-thread. Vérifiez la console pour les détails." assert mt_cache._size <= MT_CAPACITY, f"MT : La taille du cache ({mt_cache._size}) a dépassé la capacité ({MT_CAPACITY})." # Vérifie que toutes les clés restantes dans le cache ont été insérées à un moment donné final_keys = mt_cache.keys() with global_ever_inserted_keys_lock: for key in final_keys: assert key in global_ever_inserted_keys, f"Erreur MT : Le cache final contient la clé '{key}' qui n'a jamais été insérée globalement." print("Tests multi-thread réussis !") print("\n--- Tous les tests du cache LRU ont réussi ! ---") 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: """ Un simple verrou lecteur-écrivain préférentiel pour les écrivains. Stratégie de concurrence : - Plusieurs lecteurs peuvent détenir le verrou simultanément. - Les écrivains sont exclusifs. - Une fois qu'un écrivain attend, les nouveaux lecteurs sont bloqués afin que les écrivains ne soient pas affamés. Ceci est utilisé pour protéger le dictionnaire qui mappe les clés aux nœuds. Les lectures qui n'ont besoin que de la vérification d'appartenance/valeur peuvent se dérouler simultanément. Les opérations qui modifient l'index ou la liste LRU prennent le verrou d'écriture. Remarque : Python ne fournit pas de primitives véritablement sans verrou pour construire une LRU entièrement sans verrou dans la bibliothèque standard. Cette implémentation évite plutôt un seul mutex global pour tout accès en utilisant : - un verrou lecteur-écrivain pour l'index clé->nœud - un verrou séparé pour la liste LRU doublement chaînée - des verrous par nœud pour l'état local du nœud Cela offre une concurrence substantiellement meilleure qu'un seul verrou global tout en préservant la correction sous contention. """ 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 avec synchronisation fine. Aperçu de la conception ---------------------- Structures de données : - self._map: dict[str, _Node] Mappe les clés aux nœuds. Protégé par self._index_lock (RWLock). - Liste chaînée double avec sentinelles self._head (côté MRU) et self._tail (côté LRU) Protégé par self._list_lock. Règles de verrouillage ---------------------- 1. Les lectures du dictionnaire/index utilisent self._index_lock.acquire_read(). 2. Les mutations du dictionnaire/index utilisent self._index_lock.acquire_write(). 3. Tout changement de pointeur dans la liste chaînée utilise self._list_lock. 4. Les verrous par nœud sérialisent l'accès à la valeur/état supprimé d'un nœud. Règle d'ordonnancement pour éviter les interblocages -------------------------------- Lorsque plusieurs verrous sont nécessaires, les opérations suivent un ordre stable : verrou d'index -> verrou de liste -> verrou(s) de nœud Compromis de concurrence ---------------------- - get() doit mettre à jour la récence, il ne peut donc pas être une lecture pure. Il effectue d'abord une recherche d'index concurrente sous un verrou de lecture, puis prend brièvement le verrou de liste pour déplacer le nœud localisé vers la position MRU. - put(), delete() et l'éviction modifient à la fois l'index et la liste, ils prennent donc le verrou d'écriture de l'index et le verrou de liste. - Ce n'est pas strictement sans verrou ; la bibliothèque standard Python ne rend pas une LRU à liste chaînée pratique sans verrou réalisable. C'est, cependant, plus concurrent qu'une conception utilisant un seul mutex global pour chaque opération. """ def __init__(self, capacity: int): """Initialise le cache avec une capacité maximale positive.""" 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: """Insère le nœud juste après la tête. L'appelant doit détenir 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: """Détache le nœud de la liste. L'appelant doit détenir 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: """ Déplace un nœud existant vers la position MRU s'il est toujours actif. Utilise le verrou de liste plus le verrou de nœud. Le verrou de lecture de la carte détenu par l'appelant assure que l'objet nœud reste accessible depuis l'index pendant la recherche. """ 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]: """ Retourne la valeur pour la clé si elle est présente et la marque comme la plus récemment utilisée. Retourne None si absente. """ 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: """ Insère ou met à jour une paire clé-valeur. Si la capacité est dépassée, évince l'élément le moins récemment utilisé. """ 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: """ Supprime la clé du cache. Retourne True si la clé existait, sinon 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]: """ Retourne les clés de la plus récemment utilisée à la moins récemment utilisée. Nous prenons le verrou de lecture de l'index et le verrou de liste pour obtenir un instantané cohérent. """ 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: """Aide interne pour les tests.""" self._index_lock.acquire_read() try: return len(self._map) finally: self._index_lock.release_read() def _debug_validate(self) -> None: """ Vérificateur de cohérence interne pour les tests. Vérifie : - les liens de liste sont bien formés - chaque nœud de liste actif est présent dans la carte - aucune clé dupliquée dans la liste - la taille ne dépasse jamais la capacité """ 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: """ Exécute des tests mono-thread et multi-thread. Exigences des tests multi-thread couvertes : - au moins 8 threads - mélange de get/put/delete sur des clés qui se chevauchent - assertion que le cache ne dépasse jamais la capacité - assertion que get ne retourne jamais une valeur pour une clé qui n'a jamais été insérée """ # Tests mono-thread 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() # Tests concurrentiels 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) # Test de contention sur la même clé focalisé 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()