Implementiere einen sperrfreien konkurrierenden LRU-Cache

Antwort: import threading import random import time from typing import Any, Optional, List class _Node: """ Repräsentiert einen Knoten in der doppelt verketteten Liste, die für die LRU-Reihenfolge verwendet wird. Verwendet __slots__ für Speichereffizienz. """ __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: """ Eine benutzerdefinierte Read-Write-Lock-Implementierung. Dieses Lock erlaubt mehreren Lesern, das Lock gleichzeitig zu halten, aber jeweils nur einem Schreiber. Ein Schreiber muss warten, bis alle Leser fertig sind, und Leser müssen warten, bis ein Schreiber fertig ist. Dies ist eine leserbevorzugende Implementierung, was bedeutet, dass neue Leser das Leselock erwerben können, selbst wenn ein Schreiber wartet. Diese Klasse stellt außerdem Context Manager für einen saubereren Erwerb und die Freigabe von Locks bereit. """ def __init__(self): self._read_lock = threading.Lock() # Schützt den _readers-Zähler und die Bedingungsvariable self._write_lock = threading.Lock() # Das eigentliche Schreib-Lock self._readers = 0 # Anzahl aktiver Leser self._no_readers = threading.Condition(self._read_lock) # Bedingung, damit Schreiber auf Leser warten können def acquire_read(self): """Erwirbt ein Leselock. Mehrere Leser können dieses Lock gleichzeitig halten.""" with self._read_lock: # Warten, falls ein Schreiber derzeit das Schreib-Lock hält. # Neue Leser blockieren, wenn ein Schreiber aktiv ist. while self._write_lock.locked(): self._no_readers.wait() self._readers += 1 def release_read(self): """Gibt ein Leselock frei.""" with self._read_lock: self._readers -= 1 # Wenn keine Leser mehr übrig sind, alle wartenden Schreiber benachrichtigen if self._readers == 0: self._no_readers.notify_all() def acquire_write(self): """Erwirbt ein Schreib-Lock. Jeweils nur ein Schreiber kann dieses Lock gleichzeitig halten.""" # Zuerst das Haupt-Schreib-Lock erwerben, um die Schreibabsicht zu signalisieren self._write_lock.acquire() with self._read_lock: # Dann read_lock erwerben, um auf aktive Leser zu warten # Warten, bis alle aktiven Leser fertig sind while self._readers > 0: self._no_readers.wait() def release_write(self): """Gibt ein Schreib-Lock frei.""" with self._read_lock: # read_lock erwerben, um potenzielle Leser/Schreiber zu benachrichtigen self._no_readers.notify_all() # Alle wartenden Leser oder Schreiber benachrichtigen self._write_lock.release() # Das Haupt-Schreib-Lock freigeben 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): """Gibt einen Context Manager zum Erwerb eines Leselocks zurück.""" return self._ReadContextManager(self) def write_locked(self): """Gibt einen Context Manager zum Erwerb eines Schreib-Locks zurück.""" return self._WriteContextManager(self) class LRUCache: """ Eine thread-sichere LRU-Cache-Implementierung (Least Recently Used). Dieser Cache verwendet ein ReadWriteLock, um gleichzeitigen Zugriff zu verwalten. Das Design zielt darauf ab, gleichzeitige Lesezugriffe für Cache-Misses und die Operation `keys()` zu ermöglichen. Operationen jedoch, die den Cache-Zustand oder die LRU-Reihenfolge verändern (`put`, `delete` und Cache-Hits in `get`), erwerben ein Schreib-Lock und gewährleisten damit Atomarität und Konsistenz für diese kritischen Abschnitte. Dieser Ansatz bietet bessere Parallelität als ein einzelner globaler Mutex für alle Operationen, insbesondere bei leseintensiven Workloads mit häufigen Cache-Misses oder `keys()`-Aufrufen. Die Operation `get` verwendet ein "Upgrade"-Muster: Sie versucht zunächst mit einem Leselock, die Existenz des Schlüssels zu prüfen. Wird er gefunden, gibt sie das Leselock frei und erwirbt ein Schreib-Lock, um die LRU-Reihenfolge zu aktualisieren und den neuesten Wert abzurufen. Dieses Muster ist sorgfältig implementiert, um Race Conditions zu behandeln, die während des Lock-Upgrades auftreten können (z. B. wenn der Schlüssel von einem anderen Thread gelöscht oder aktualisiert wird). """ def __init__(self, capacity: int): if capacity <= 0: raise ValueError("Capacity must be a positive integer.") self._capacity = capacity self._cache = {} # Speichert key -> _Node für O(1)-Lookup self._size = 0 # Dummy-Knoten für Kopf und Ende der doppelt verketteten Liste. # Das vereinfacht Randfälle beim Hinzufügen/Entfernen von Knoten, da sich echte Datenknoten # immer zwischen Kopf und Ende befinden. self._head = _Node("dummy_head", None) self._tail = _Node("dummy_tail", None) self._head.next = self._tail self._tail.prev = self._head # Parallelitätskontrolle: Ein einzelnes ReadWriteLock schützt den gesamten Cache-Zustand. self._rw_lock = ReadWriteLock() def _add_node(self, node: _Node): """Fügt einen Knoten direkt nach dem Kopf hinzu (Position des zuletzt verwendeten Elements). Geht davon aus, dass das Schreib-Lock gehalten wird.""" node.prev = self._head node.next = self._head.next self._head.next.prev = node self._head.next = node def _remove_node(self, node: _Node): """Entfernt einen Knoten aus der doppelt verketteten Liste. Geht davon aus, dass das Schreib-Lock gehalten wird.""" 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): """Verschiebt einen vorhandenen Knoten an den Anfang der Liste (zuletzt verwendet). Geht davon aus, dass das Schreib-Lock gehalten wird.""" self._remove_node(node) self._add_node(node) def _pop_tail(self) -> _Node: """Entfernt und gibt den am längsten nicht verwendeten Knoten zurück (den direkt vor dem Ende). Geht davon aus, dass das Schreib-Lock gehalten wird.""" node = self._tail.prev self._remove_node(node) return node def get(self, key: str) -> Optional[Any]: """ Gibt den mit dem Schlüssel verknüpften Wert zurück, falls er existiert, und markiert ihn als kürzlich verwendet. Gibt None zurück, wenn sich der Schlüssel nicht im Cache befindet. Diese Operation erwirbt zunächst ein Leselock, um die Existenz des Schlüssels zu prüfen. Wird der Schlüssel gefunden, gibt sie das Leselock frei und erwirbt dann ein Schreib-Lock, um die LRU-Reihenfolge zu aktualisieren und den neuesten Wert abzurufen. Dieses "Upgrade"-Muster erlaubt gleichzeitige `get`-Misses und `keys()`-Operationen, serialisiert aber `get`-Hits mit anderen Schreibvorgängen. """ 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 # Wenn der Schlüssel gefunden wurde, Schreib-Lock erwerben, um die LRU-Reihenfolge zu aktualisieren # und den neuesten Wert zu erhalten. # Dies ist ein "Upgrade"-Muster. Hier kann eine Race Condition auftreten: Ein anderer Thread # könnte den Schlüssel zwischen dem Freigeben des Leselocks und dem Erwerb des Schreib-Locks löschen oder aktualisieren. with self._rw_lock.write_locked(): # Erneut prüfen, ob der Schlüssel noch existiert und auf exakt dasselbe Knotenobjekt verweist. # Das ist für Korrektheit unter Konkurrenz entscheidend. Wenn ein anderer Thread # ein `put` für denselben Schlüssel ausgeführt hat, würde `_cache.get(key)` einen *neuen* Knoten zurückgeben. # Wenn ein anderer Thread ein `delete` ausgeführt hat, würde `None` zurückgegeben. current_node = self._cache.get(key) if current_node is node_found_under_read_lock: # Sicherstellen, dass es exakt dasselbe Knotenobjekt ist self._move_to_front(node_found_under_read_lock) value = current_node.value # Nach dem Verschieben den neuesten Wert holen return value else: # Der Knoten wurde zwischen dem Erwerb unseres Lese-Locks und Schreib-Locks # entweder gelöscht oder durch einen anderen 'put'-Vorgang ersetzt. Als Miss behandeln. return None def put(self, key: str, value: Any) -> None: """ Fügt das Schlüssel-Wert-Paar ein oder aktualisiert es. Wenn der Cache nach dem Einfügen die Kapazität überschreitet, wird das am längsten nicht verwendete Element entfernt. Dies ist eine Schreiboperation und erwirbt ein Schreib-Lock. """ with self._rw_lock.write_locked(): node = self._cache.get(key) if node: # Schlüssel existiert: Wert aktualisieren und an den Anfang verschieben node.value = value self._move_to_front(node) else: # Schlüssel existiert nicht: neuen Knoten hinzufügen new_node = _Node(key, value) self._cache[key] = new_node self._add_node(new_node) self._size += 1 # Entfernen, wenn Kapazität überschritten wurde 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: """ Entfernt den Schlüssel aus dem Cache. Gibt True zurück, wenn der Schlüssel vorhanden war, andernfalls False. Dies ist eine Schreiboperation und erwirbt ein Schreib-Lock. """ 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]: """ Gibt eine Liste aller derzeit im Cache befindlichen Schlüssel zurück, geordnet von zuletzt verwendet bis am längsten nicht verwendet. Dies ist eine Leseoperation und erwirbt ein Leselock. """ 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(): """ Führt eine Reihe von Single-Thread- und Multi-Thread-Tests aus, um die LRUCache-Implementierung zu überprüfen. """ print("--- Starte LRU-Cache-Tests ---") # --- Single-Thread-Tests --- print("\n--- Single-Thread-Tests ---") cache = LRUCache(capacity=3) assert cache.get("a") is None, "ST: Abruf eines nicht existierenden Schlüssels sollte None zurückgeben" assert cache.keys() == [], "ST: Ein leerer Cache sollte eine leere Schlüsselliste haben" cache.put("a", 1) assert cache.get("a") == 1, "ST: Vorhandenen Schlüssel nach put abrufen" assert cache.keys() == ["a"], "ST: Schlüsselliste nach einem put" cache.put("b", 2) cache.put("c", 3) assert cache.keys() == ["c", "b", "a"], "ST: Schlüsselliste nach mehreren puts" assert cache._size == 3, "ST: Cache-Größe sollte 3 sein" # Test der Kapazitätsverdrängung cache.put("d", 4) # 'a' sollte entfernt werden, da es LRU ist assert cache.get("a") is None, "ST: Entferntem Schlüssel 'a' sollte None entsprechen" assert cache.get("d") == 4, "ST: Neuer Schlüssel 'd' sollte vorhanden sein" assert cache.keys() == ["d", "c", "b"], "ST: Schlüsselliste nach Verdrängung" assert cache._size == 3, "ST: Cache-Größe sollte nach Verdrängung 3 bleiben" # put mit vorhandenem Schlüssel testen (Wert aktualisieren und an den Anfang verschieben) cache.put("b", 22) # 'b' sollte nach vorn verschoben und der Wert aktualisiert werden assert cache.get("b") == 22, "ST: Aktualisierter Wert für 'b'" assert cache.keys() == ["b", "d", "c"], "ST: Schlüsselliste nach Aktualisierung und Verschiebung von 'b'" # delete testen assert cache.delete("c") is True, "ST: Vorhandenen Schlüssel 'c' löschen" assert cache.get("c") is None, "ST: Gelöschter Schlüssel 'c' sollte None sein" assert cache.keys() == ["b", "d"], "ST: Schlüsselliste nach dem Löschen von 'c'" assert cache._size == 2, "ST: Cache-Größe nach delete" assert cache.delete("z") is False, "ST: Nicht existierenden Schlüssel 'z' löschen" assert cache.keys() == ["b", "d"], "ST: Schlüsselliste nach dem Löschen eines nicht existierenden Schlüssels unverändert" # Kapazität 1 testen cache_cap1 = LRUCache(capacity=1) cache_cap1.put("x", 10) assert cache_cap1.get("x") == 10, "ST: Kapazität 1, 'x' abrufen" cache_cap1.put("y", 20) # 'x' sollte entfernt werden assert cache_cap1.get("x") is None, "ST: Kapazität 1, 'x' entfernt" assert cache_cap1.get("y") == 20, "ST: Kapazität 1, 'y' abrufen" assert cache_cap1.keys() == ["y"], "ST: Kapazität 1, Schlüsselliste" print("Single-Thread-Tests bestanden!") # --- Multi-Thread-Tests --- print("\n--- Multi-Thread-Tests ---") MT_CAPACITY = 10 MT_NUM_THREADS = 8 MT_OPS_PER_THREAD = 2000 MT_KEY_RANGE = 20 # Schlüssel von 0 bis 19 (z. B. "0", "1", ..., "19") mt_cache = LRUCache(capacity=MT_CAPACITY) # Barriere, um sicherzustellen, dass alle Worker-Threads ihre Operationen gleichzeitig starten barrier = threading.Barrier(MT_NUM_THREADS + 1) # +1 für den Haupt-Thread error_flag = threading.Event() # Wird gesetzt, wenn ein Thread auf einen Fehler stößt # Verfolgt Schlüssel, die *jemals* von irgendeinem Thread erfolgreich in den Cache eingefügt wurden. # Wird verwendet, um zu überprüfen, dass 'get' niemals einen Wert für einen Schlüssel zurückgibt, der tatsächlich nie eingefügt wurde. 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): """Worker-Funktion für Multi-Thread-Tests.""" barrier_instance.wait() # Warten, bis alle Threads bereit sind, Operationen zu starten for i in range(MT_OPS_PER_THREAD): if error_event.is_set(): break # Stoppen, wenn ein anderer Thread bereits einen Fehler gefunden hat # Zufällig eine Operation auswählen, mit einer Gewichtung zugunsten von 'get' und '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: # Sicherstellen, dass der Schlüssel, wenn ein Wert zurückgegeben wird, irgendwann eingefügt worden sein muss. with ever_inserted_keys_lock: if key not in ever_inserted_keys_set: print(f"MT-Fehler: Thread {thread_id} erhielt Wert '{val}' für Schlüssel '{key}', " f"der global nie eingefügt wurde. Das sollte nicht passieren.") error_event.set() break # Sicherstellen, dass der Wert nicht beschädigt ist (einfache Formatprüfung) if not isinstance(val, str) or not val.startswith("value_T"): print(f"MT-Fehler: Thread {thread_id} erhielt beschädigten Wert '{val}' für Schlüssel '{key}'.") error_event.set() break elif op == 'delete': cache_instance.delete(key) # Optional: Eine kleine zufällige Schlafzeit einführen, um Kontextwechsel und Konkurrenz zu erhöhen # time.sleep(random.uniform(0.00001, 0.0001)) # print(f"Thread {thread_id} beendet.") # Für ausführliche Ausgabe einkommentieren 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() # Warten, bis alle Worker-Threads die Barriere erreicht haben und mit den Operationen beginnen barrier.wait() print(f"Alle {MT_NUM_THREADS} Threads haben ihre Operationen gestartet.") # Warten, bis alle Worker-Threads ihre Operationen abgeschlossen haben for thread in threads: thread.join() print("Alle Threads abgeschlossen.") # Abschließende Assertions für Multi-Thread-Tests assert not error_flag.is_set(), "MT: Während der Multi-Thread-Operationen wurden Fehler erkannt. Details in der Konsole prüfen." assert mt_cache._size <= MT_CAPACITY, f"MT: Cache-Größe ({mt_cache._size}) überschritt die Kapazität ({MT_CAPACITY})." # Prüfen, dass alle im Cache verbleibenden Schlüssel irgendwann eingefügt wurden 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-Fehler: Finaler Cache enthält Schlüssel '{key}', der global nie eingefügt wurde." print("Multi-Thread-Tests bestanden!") print("\n--- Alle LRU-Cache-Tests bestanden! ---") 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: """ Eine einfache, schreibzugriff-bevorzugende Lese-/Schreibsperre. Konkurrenzstrategie: - Mehrere Leser können die Sperre gleichzeitig halten. - Schreiber sind exklusiv. - Sobald ein Schreiber wartet, werden neue Leser blockiert, damit Schreiber nicht ausgehungert werden. Dies wird verwendet, um das Wörterbuch zu schützen, das Schlüssel zu Knoten zuordnet. Lesevorgänge, die nur Mitgliedschafts-/Werteabfragen benötigen, können gleichzeitig ausgeführt werden. Operationen, die den Index oder die LRU-Liste ändern, übernehmen die Schreibsperre. Hinweis: Python bietet keine echten sperrenfreien Primitiven zum Erstellen einer vollständig sperrenfreien LRU in der Standardbibliothek. Diese Implementierung vermeidet stattdessen eine einzelne globale Mutex für allen Zugriff, indem sie verwendet: - eine Lese-/Schreibsperre für den Schlüssel->Knoten-Index - eine separate Sperre für die doppelt verkettete LRU-Liste - Knotenspezifische Sperren für knotenlokalen Zustand Dies bietet eine erheblich bessere Nebenläufigkeit als eine einzelne globale Sperre und wahrt gleichzeitig die Korrektheit unter Last. """ 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: """ Threadsichere LRU-Cache mit feingranularer Synchronisation. Designübersicht --------------- Datenstrukturen: - self._map: dict[str, _Node] Ordnet Schlüssel Knoten zu. Geschützt durch self._index_lock (RWLock). - Doppelt verkettete Liste mit Sentinel-Knoten self._head (MRU-Seite) und self._tail (LRU-Seite) Geschützt durch self._list_lock. Sperrregeln ------------- 1. Wörterbuch-/Index-Lesevorgänge verwenden self._index_lock.acquire_read(). 2. Wörterbuch-/Index-Mutationen verwenden self._index_lock.acquire_write(). 3. Jegliche Zeigeränderungen in der verketteten Liste verwenden self._list_lock. 4. Knotenspezifische Sperren serialisieren den Zugriff auf den Wert/den gelöschten Zustand eines Knotens. Reihenfolgeregel zur Vermeidung von Deadlocks -------------------------------- Wenn mehrere Sperren benötigt werden, folgen Operationen einer stabilen Reihenfolge: Index-Sperre -> Listen-Sperre -> Knoten-Sperre(n) Nebenläufigkeits-Kompromisse ---------------------- - get() muss die Aktualität aktualisieren, daher kann es kein reiner Lesevorgang sein. Es führt zuerst eine gleichzeitige Indexabfrage unter einer Lesesperre durch und übernimmt dann kurzzeitig die Listensperre, um den gefundenen Knoten an die MRU-Position zu verschieben. - put(), delete() und Eviction ändern sowohl den Index als auch die Liste, daher übernehmen sie die Index-Schreibsperre und die Listensperre. - Dies ist nicht streng sperrenfrei; die Standardbibliothek von Python macht eine praktische sperrenfreie LRU-Liste nicht machbar. Es ist jedoch nebenläufiger als ein Design, das eine globale Mutex für jede Operation verwendet. """ def __init__(self, capacity: int): """Initialisiert den Cache mit einer positiven maximalen Kapazität.""" 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: """Fügt den Knoten direkt nach dem Kopf ein. Der Aufrufer muss self._list_lock halten.""" 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: """Löst den Knoten aus der Liste. Der Aufrufer muss self._list_lock halten.""" 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: """ Verschiebt einen vorhandenen Knoten an die MRU-Position, wenn er noch aktiv ist. Verwendet die Listensperre plus die Knotensperre. Die vom Aufrufer gehaltene Map-Lesesperre stellt sicher, dass das Knotenobjekt während der Abfrage über den Index erreichbar bleibt. """ 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]: """ Gibt den Wert für den Schlüssel zurück, wenn er vorhanden ist, und markiert ihn als zuletzt verwendet. Gibt None zurück, wenn er nicht vorhanden ist. """ 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: """ Fügt ein Schlüssel-Wert-Paar ein oder aktualisiert es. Wenn die Kapazität überschritten wird, wird das am wenigsten zuletzt verwendete Element entfernt. """ 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: """ Entfernt den Schlüssel aus dem Cache. Gibt True zurück, wenn der Schlüssel vorhanden war, andernfalls 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]: """ Gibt die Schlüssel vom zuletzt verwendeten zum am wenigsten zuletzt verwendeten zurück. Wir nehmen die Index-Lesesperre und die Listensperre, um einen konsistenten Schnappschuss zu erhalten. """ 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: """Interner Helfer für Tests.""" self._index_lock.acquire_read() try: return len(self._map) finally: self._index_lock.release_read() def _debug_validate(self) -> None: """ Interner Konsistenzprüfer für Tests. Überprüft: - Listenverknüpfungen sind wohlgeformt - Jeder aktive Listenknoten ist im Map vorhanden - Keine doppelten Schlüssel in der Liste - Die Größe überschreitet niemals die Kapazität """ 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: """ Führt Single-Threaded- und Multi-Threaded-Tests aus. Anforderungen an Multi-Threaded-Tests: - mindestens 8 Threads - Mischung aus get/put/delete auf überlappenden Schlüsseln - Assert, dass der Cache niemals die Kapazität überschreitet - Assert, dass get niemals einen Wert für einen Schlüssel zurückgibt, der nie eingefügt wurde """ # Single-Threaded-Tests 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() # Konkurrente Tests 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) # Fokussierter Test auf gleichzeitigen Zugriff auf denselben Schlüssel 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()