Implémenter un cache LRU concurrent sans verrou global

Réponse : ```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("La capacité doit être un entier positif.") self.capacity = capacity self.cache = {} self.head = Node(None, None) # Tête factice self.tail = Node(None, None) # Queue factice self.head.next = self.tail self.tail.prev = self.head self.lock = threading.RLock() # Utilisation de RLock pour simplifier dans cet exemple, mais une approche plus fine est possible. 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: # Expulser l'élément 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) # --- Explication de la stratégie de concurrence --- # Cette implémentation utilise un unique threading.RLock pour toutes les opérations (get, put, delete). # Bien que cela garantisse la sûreté des threads, cela ne minimise pas la contention comme demandé par l’invite. # Une approche plus avancée et plus fine impliquerait de segmenter le cache ou d’utiliser # des structures de données sans verrouillage. Par exemple, on pourrait utiliser plusieurs verrous, chacun protégeant un sous-ensemble # des entrées du cache (striped locking), ou utiliser des verrous lecteurs-rédacteurs pour permettre à plusieurs lecteurs # d’accéder simultanément à différentes parties du cache. # Le RLock actuel garantit qu’un seul thread peut modifier l’état interne du cache (le dictionnaire et la liste doublement chaînée) à la fois. # Cela empêche les conditions de concurrence où plusieurs threads pourraient essayer de mettre à jour le même nœud, d’ajouter/supprimer des nœuds, ou d’expulser des entrées simultanément, ce qui entraînerait une corruption des données ou un ordre LRU incorrect. # L’utilisation de RLock (verrou réentrant) permet à un thread d’acquérir le verrou plusieurs fois, ce qui peut être utile dans des scénarios complexes, mais ne change pas fondamentalement les caractéristiques de contention par rapport à un Lock standard dans ce cas d’utilisation spécifique. # --- Démonstration --- 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) # Simuler un peu de travail et permettre à d’autres threads de s’exécuter time.sleep(0.001) def run_demonstration(): print("Exécution de la démonstration du 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) # Clés qui peuvent ou non être dans le 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"Démonstration terminée. Taille finale du cache : {len(cache)}") # Vérifier que le cache ne dépasse jamais sa capacité assert len(cache) <= capacity, f"Capacité du cache dépassée ! Taille actuelle : {len(cache)}, Capacité : {capacity}" print("Assertion réussie : la capacité du cache est maintenue.") # Vérification de base de l’intégrité des données (non exhaustive) print("Exécution d’une vérification de base de l’intégrité...") # Il s’agit d’une vérification très basique. Un test plus robuste impliquerait de suivre les états attendus. # Pour l’instant, nous nous assurons simplement qu’il n’y a aucune corruption évidente, comme une taille négative ou des exceptions lors de l’accès. try: # Essayer d’accéder à quelques clés qui ont probablement été insérées for i in range(min(5, capacity)): key_to_check = (i * ops_per_thread) % (capacity * 2) cache.get(key_to_check) print("La vérification de base de l’intégrité a réussi.") except Exception as e: print(f"La vérification de l’intégrité a échoué avec l’exception : {e}") assert False, "La vérification de l’intégrité du cache a échoué." if __name__ == "__main__": run_demonstration() ```

Cache LRU concurrente (LRU strict) avec contention minimisée. Stratégie de concurrence (bref) : - Le cache utilise un dict pour clé->Nœud et une liste doublement chaînée pour l'ordre LRU global. - Pour réduire la contention pour les opérations clé-valeur, l'espace des clés est réparti sur plusieurs verrous. Les mises à jour d'une clé particulière (insertion/mise à jour/suppression/lecture de la correspondance et des champs du nœud) utilisent uniquement le verrou de bande de cette clé. - Pour préserver les sémantiques LRU globales strictes, toutes les mutations de la liste LRU sont sérialisées avec un verrou dédié `lru_lock`. Cela évite les conditions de concurrence qui pourraient corrompre la liste et garantit un ordre total unique des opérations « toucher ». - L'éviction nécessite une coordination : sous `lru_lock`, nous sélectionnons la queue actuelle (la moins récemment utilisée), puis acquérons le verrou de bande de cette clé pour vérifier qu'il s'agit toujours du même nœud et le supprimer de la map. Cela évite d'évincer incorrectement une clé mise à jour de manière concurrente. Pourquoi c'est correct : - La liste LRU n'est mutée que sous `lru_lock`, elle ne peut donc pas être corrompue et chaque toucher (get/put pour les clés existantes) est totalement ordonné, ce qui donne un LRU strict. - L'entrée de la map et la valeur du nœud sont protégées par le verrou de bande pour cette clé, empêchant les mises à jour/suppressions concurrentes pour la même clé de corrompre l'état. - Les évictions sont basées sur la queue réelle au moment où la décision d'éviction est prise sous `lru_lock`; si le nœud candidat change en raison d'un toucher/suppression concurrent, l'éviction est retentée. Note : Python ne fournit pas de primitives véritablement sans verrou pour des structures partagées complexes ; cette implémentation évite un seul verrou global autour de tout en séparant les préoccupations : - Les opérations au niveau de la clé sont réparties. - Seule la maintenance de l'ordre de la liste utilise un seul verrou. Une démonstration est incluse dans __main__. ```python 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 # défini sur False lors de la suppression/éviction 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] = {} # Têtes/queues sentinelles pour la liste doublement chaînée (MRU à head.next, LRU à tail.prev) self._head = _Node("<HEAD>", None) self._tail = _Node("<TAIL>", None) self._head.next = self._tail self._tail.prev = self._head # Sérialise les opérations de liste et les décisions de taille/éviction self._lru_lock = threading.Lock() # Verrous répartis pour les opérations de map/nœud au niveau de la clé self._stripes_n = int(stripes) self._stripes: List[threading.Lock] = [threading.Lock() for _ in range(self._stripes_n)] # Taille suivie sous lru_lock self._size = 0 def _stripe_lock(self, key: Any) -> threading.Lock: return self._stripes[hash(key) % self._stripes_n] # ----- Aide à la liste doublement chaînée (nécessite de détenir 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: # le nœud doit être dans la liste self._unlink(node) self._link_front(node) def _current_lru_node(self) -> Optional[_Node]: # retourne le nœud LRU actuel (tail.prev) ou None s'il est vide n = self._tail.prev if n is None or n is self._head: return None return n # ----- API publique ----- 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 # Move-to-front nécessite le verrou de liste ; faites-le en dehors du verrou de bande pour réduire la contention. with self._lru_lock: # Re-vérifier l'appartenance sous le verrou de bande pour éviter de déplacer un nœud supprimé/remplacé. with lock: cur = self._map.get(key) if cur is not node or cur is None or not cur.alive: return value # la valeur était valide au moment de la lecture ; le toucher a été perdu en raison d'une course. # Sous lru_lock, nous pouvons déplacer ce nœud en toute sécurité. self._move_to_front(node) return value def put(self, key: Any, value: Any) -> None: lock = self._stripe_lock(key) # D'abord, vérifier si la clé existe et mettre à jour sous le verrou de bande. with lock: node = self._map.get(key) if node is not None and node.alive: node.value = value # Toucher (déplacer vers l'avant) sous lru_lock. # Libérer rapidement le verrou de bande ? Nous pouvons le garder court : la mise à jour est terminée. pass else: node = None if node is not None: with self._lru_lock: # S'assurer qu'il est toujours présent et qu'il s'agit du même nœud 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 # Insérer un nouveau nœud new_node = _Node(key, value) with self._lru_lock: # Acquérir le verrou de bande tout en détenant lru_lock pour assurer une insertion/liaison cohérente. with lock: existing = self._map.get(key) if existing is not None and existing.alive: # Quelqu'un a inséré de manière concurrente ; mettre à jour celui-ci à la place. existing.value = value # Le toucher self._move_to_front(existing) return # Insérer la correspondance et lier self._map[key] = new_node self._link_front(new_node) self._size += 1 # Évincer si la capacité est dépassée while self._size > self.capacity: if not self._evict_one_locked(): # Si l'éviction a échoué en raison de courses, réessayer continue def _evict_one_locked(self) -> bool: """Évince un seul nœud LRU. Nécessite de détenir self._lru_lock. Retourne True si une éviction a eu lieu, False si un nouveau essai est nécessaire. """ 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: # la victime a changé ou a déjà été supprimée ; réessayer return False # Supprimer d'abord de la map del self._map[vkey] victim.alive = False # Maintenant, supprimer de la liste (nous détenons lru_lock) self._unlink(victim) self._size -= 1 return True def delete(self, key: Any) -> bool: lock = self._stripe_lock(key) # Acquérir d'abord lru_lock pour éviter les interblocages avec le chemin d'éviction/insertion. 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 # Détacher de la liste 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: """Routine de débogage/vérification : vérifie la cohérence interne.""" with self._lru_lock: # Vérifier la structure de la liste et compter les nœuds 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 # La map et la liste sont d'accord (meilleur effort ; les verrous de bande doivent être parfaits). # Sous lru_lock, la map peut changer sous les verrous de bande ; nous faisons donc uniquement une vérification conservative : # chaque nœud de liste doit exister dans la map. 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) # Vérifier occasionnellement que la taille ne dépasse pas la capacité 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() # Exécuter un peu time.sleep(1.5) # Arrêter et joindre stop.set() for t in threads: t.join(timeout=2.0) # Aucune erreur de travailleur if errors: raise errors[0] # Vérifications finales des invariants assert len(cache) <= cap cache._assert_invariants() # Sanité de base : les put/get répétés sur la même clé à partir de plusieurs threads ne doivent pas corrompre 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.") ```