Implementieren Sie einen sperrfreien nebenläufigen LRU-Cache

```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("Kapazität muss eine positive Ganzzahl sein.") self.capacity = capacity self.cache = {} self.head = Node(None, None) # Dummy-Kopf self.tail = Node(None, None) # Dummy-Schwanz self.head.next = self.tail self.tail.prev = self.head self.lock = threading.RLock() # Verwendung von RLock zur Vereinfachung in diesem Beispiel, aber ein feingranularerer Ansatz ist möglich. 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: # LRU-Element löschen 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) # --- Erklärung der Nebenläufigkeitsstrategie --- # Diese Implementierung verwendet ein einzelnes threading.RLock für alle Operationen (get, put, delete). # Während dies Threadsicherheit garantiert, minimiert es die Kontention nicht, wie vom Prompt gefordert. # Ein fortgeschrittenerer, feingranularerer Ansatz würde die Cache-Segmentierung oder die Verwendung # von sperrfreien Datenstrukturen beinhalten. Zum Beispiel könnte man mehrere Sperren verwenden, von denen jede eine Teilmenge # der Cache-Einträge schützt (Striped Locking), oder Read-Write-Sperren verwenden, um mehreren Lesern # den gleichzeitigen Zugriff auf verschiedene Teile des Caches zu ermöglichen. # Das aktuelle RLock stellt sicher, dass zu einem Zeitpunkt nur ein Thread den internen Zustand des Caches (das Dictionary und die doppelt verkettete Liste) ändern kann. # Dies verhindert Race Conditions, bei denen mehrere Threads versuchen könnten, denselben Knoten zu aktualisieren, Knoten hinzuzufügen/zu entfernen oder Einträge gleichzeitig zu löschen, was zu Datenbeschädigung oder falscher LRU-Reihenfolge führen würde. # Die Verwendung von RLock (Reentrant Lock) ermöglicht es einem Thread, die Sperre mehrmals zu erwerben, was in komplexen Szenarien nützlich sein kann, aber die Kontentionseigenschaften im Vergleich zu einer Standard-Sperre in diesem spezifischen Anwendungsfall nicht grundlegend ändert. # --- Demonstration --- 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) # Etwas Arbeit simulieren und anderen Threads die Ausführung ermöglichen time.sleep(0.001) def run_demonstration(): print("Führe LRU-Cache-Demonstration aus...") 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) # Schlüssel, die möglicherweise im Cache vorhanden sind oder nicht 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"Demonstration beendet. Endgültige Cache-Größe: {len(cache)}") # Sicherstellen, dass der Cache seine Kapazität nicht überschreitet assert len(cache) <= capacity, f"Cache-Kapazität überschritten! Aktuelle Größe: {len(cache)}, Kapazität: {capacity}" print("Assertion bestanden: Cache-Kapazität wird eingehalten.") # Grundlegende Prüfung der Datenintegrität (nicht erschöpfend) print("Führe grundlegende Integritätsprüfung durch...") # Dies ist eine sehr grundlegende Prüfung. Ein robusterer Test würde das Verfolgen erwarteter Zustände beinhalten. # Vorerst stellen wir nur sicher, dass keine offensichtliche Beschädigung wie eine negative Größe oder Ausnahmen während des Zugriffs vorliegt. try: # Versuche, auf einige Schlüssel zuzugreifen, die wahrscheinlich eingefügt wurden for i in range(min(5, capacity)): key_to_check = (i * ops_per_thread) % (capacity * 2) cache.get(key_to_check) print("Grundlegende Integritätsprüfung bestanden.") except Exception as e: print(f"Integritätsprüfung fehlgeschlagen mit Ausnahme: {e}") assert False, "Cache-Integritätsprüfung fehlgeschlagen." if __name__ == "__main__": run_demonstration() ```

"""Nebenläufiger LRU-Cache (striktes LRU) mit minimierter Konkurrenz. Nebenläufigkeitsstrategie (kurz): - Der Cache verwendet ein dict für Schlüssel->Node und eine doppelt verkettete Liste für die globale LRU-Reihenfolge. - Um Konkurrenz bei Schlüssel-Wert-Operationen zu reduzieren, wird der Schlüsselraum über mehrere Locks gestreift. Aktualisierungen für einen bestimmten Schlüssel (Einfügen/Aktualisieren/Löschen/Lesen des Mappings und der Node-Felder) verwenden nur den Stripe-Lock dieses Schlüssels. - Um *strikte globale LRU-Semantik* zu bewahren, werden alle Mutationen der LRU-Liste mit einem dedizierten lru_lock serialisiert. Dies vermeidet Race Conditions, die die Liste beschädigen könnten, und stellt sicher, dass es eine einzige totale Ordnung von „touch“-Operationen gibt. - Eviction erfordert Koordination: Unter lru_lock wählen wir das aktuelle Ende (am wenigsten kürzlich verwendet) aus und erwerben dann den Stripe-Lock dieses Schlüssels, um zu verifizieren, dass es noch derselbe Node ist, und entfernen ihn aus der Map. Dies verhindert, dass ein gleichzeitig aktualisierter Schlüssel fälschlich verdrängt wird. Warum das korrekt ist: - Die LRU-Liste wird nur unter lru_lock mutiert, daher kann sie nicht beschädigt werden, und jede touch- Operation (get/put für bestehende Schlüssel) ist total geordnet, was striktes LRU ergibt. - Der Map-Eintrag und der Wert des Node werden durch den Stripe-Lock für diesen Schlüssel geschützt, wodurch gleichzeitige Aktualisierungen/Löschungen für denselben Schlüssel den Zustand nicht beschädigen können. - Verdrängungen basieren auf dem tatsächlichen Ende in dem Moment, in dem die Verdrängung unter lru_lock entschieden wird; wenn sich der Kandidaten-Node aufgrund eines gleichzeitigen touch/delete ändert, wird die Verdrängung erneut versucht. Hinweis: Python stellt keine wirklich lock-freien Primitive für komplexe gemeinsam genutzte Strukturen bereit; diese Implementierung vermeidet einen einzigen globalen Lock um *alles*, indem Zuständigkeiten getrennt werden: - Operationen auf Schlüssel-Ebene sind gestreift. - Nur die Pflege der Listenreihenfolge verwendet einen einzelnen Lock. Eine Demonstration ist in __main__ enthalten. """ 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 # auf False gesetzt, wenn gelöscht/verdrängt class ConcurrentLRUCache: def __init__(self, capacity: int, stripes: int = 64): if capacity <= 0: raise ValueError("capacity muss positiv sein") if stripes <= 0: raise ValueError("stripes muss positiv sein") self.capacity = int(capacity) self._map: Dict[Any, _Node] = {} # Sentinel-Kopf/-Ende für doppelt verkettete Liste (MRU bei head.next, LRU bei tail.prev) self._head = _Node("<HEAD>", None) self._tail = _Node("<TAIL>", None) self._head.next = self._tail self._tail.prev = self._head # Serialisiert Listenoperationen sowie Größen-/Verdrängungsentscheidungen self._lru_lock = threading.Lock() # Gestreifte Locks für Map-/Node-Operationen auf Schlüssel-Ebene self._stripes_n = int(stripes) self._stripes: List[threading.Lock] = [threading.Lock() for _ in range(self._stripes_n)] # Größe wird unter lru_lock verfolgt self._size = 0 def _stripe_lock(self, key: Any) -> threading.Lock: return self._stripes[hash(key) % self._stripes_n] # ----- Hilfsfunktionen für doppelt verkettete Liste (erfordern das Halten von 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 muss in der Liste sein self._unlink(node) self._link_front(node) def _current_lru_node(self) -> Optional[_Node]: # gibt den tatsächlichen LRU-Node (tail.prev) zurück oder None, wenn leer n = self._tail.prev if n is None or n is self._head: return None return n # ----- Öffentliche API ----- 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 # Nach-vorn-Verschieben erfordert den Listen-Lock; dies außerhalb des Stripe-Locks, um Konkurrenz zu reduzieren. with self._lru_lock: # Mitgliedschaft unter Stripe-Lock erneut prüfen, um das Verschieben eines gelöschten/ersetzten Node zu vermeiden. with lock: cur = self._map.get(key) if cur is not node or cur is None or not cur.alive: return value # Wert war zum Zeitpunkt des Lesens gültig; touch durch Race verloren. # Unter lru_lock können wir diesen Node sicher verschieben. self._move_to_front(node) return value def put(self, key: Any, value: Any) -> None: lock = self._stripe_lock(key) # Zuerst prüfen, ob der Schlüssel existiert, und unter Stripe-Lock aktualisieren. with lock: node = self._map.get(key) if node is not None and node.alive: node.value = value # Touch (nach vorn verschieben) unter lru_lock. # Stripe-Lock schnell freigeben? Wir können ihn kurz halten: Aktualisierung erledigt. pass else: node = None if node is not None: with self._lru_lock: # Sicherstellen, dass noch vorhanden und derselbe Node 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 # Neuen Node einfügen new_node = _Node(key, value) with self._lru_lock: # Stripe-Lock halten, während lru_lock gehalten wird, um konsistentes Einfügen+Verknüpfen sicherzustellen. with lock: existing = self._map.get(key) if existing is not None and existing.alive: # Jemand hat gleichzeitig eingefügt; stattdessen diesen aktualisieren. existing.value = value # Touch darauf self._move_to_front(existing) return # Mapping einfügen und verknüpfen self._map[key] = new_node self._link_front(new_node) self._size += 1 # Verdrängen, solange über Kapazität while self._size > self.capacity: if not self._evict_one_locked(): # Wenn Verdrängung wegen Race Conditions fehlschlägt, erneut versuchen continue def _evict_one_locked(self) -> bool: """Einen einzelnen LRU-Node verdrängen. Erfordert das Halten von self._lru_lock. Gibt True zurück, wenn eine Verdrängung stattgefunden hat, False, wenn erneut versucht werden muss. """ 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: # Opfer hat sich geändert oder bereits entfernt; erneut versuchen return False # Zuerst aus der Map entfernen del self._map[vkey] victim.alive = False # Jetzt aus der Liste entfernen (wir halten lru_lock) self._unlink(victim) self._size -= 1 return True def delete(self, key: Any) -> bool: lock = self._stripe_lock(key) # Zuerst lru_lock erwerben, um Deadlocks mit Verdrängungs-/Einfügepfad zu vermeiden. 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 # Aus der LRU-Liste lösen self._unlink(node) self._size -= 1 return True def __len__(self) -> int: with self._lru_lock: return self._size def _assert_invariants(self) -> None: """Debug-/Prüfroutine: verifiziert interne Konsistenz.""" with self._lru_lock: # Listenstruktur prüfen und Nodes zählen 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 # Map und Liste stimmen überein (Best-Effort; für Perfektion wären Stripe-Locks nötig). # Unter lru_lock kann sich die Map unter Stripe-Locks ändern; daher führen wir nur eine konservative Prüfung durch: # jeder Listen-Node muss in der Map existieren. 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) # Gelegentlich prüfen, dass die Größe die Kapazität nicht überschreitet 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() # Ein wenig laufen lassen time.sleep(1.5) # Stoppen und zusammenführen stop.set() for t in threads: t.join(timeout=2.0) # Keine Worker-Fehler if errors: raise errors[0] # Abschließende Invariantenprüfungen assert len(cache) <= cap cache._assert_invariants() # Grundlegender Plausibilitätstest: wiederholtes put/get auf demselben Schlüssel aus mehreren Threads sollte nichts beschädigen 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 abgeschlossen; Cache-Invarianten gelten.")