Implement a Lock-Free Concurrent 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("Capacity must be a positive integer.") self.capacity = capacity self.cache = {} self.head = Node(None, None) # Dummy head self.tail = Node(None, None) # Dummy tail self.head.next = self.tail self.tail.prev = self.head self.lock = threading.RLock() # Using RLock for simplicity in this example, but a more fine-grained approach is 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: # Evict LRU item 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) # --- Concurrency Strategy Explanation --- # This implementation uses a single threading.RLock for all operations (get, put, delete). # While this guarantees thread safety, it does not minimize contention as requested by the prompt. # A more advanced, fine-grained approach would involve segmenting the cache or using # lock-free data structures. For example, one could use multiple locks, each protecting a subset # of the cache entries (striped locking), or use read-write locks to allow multiple readers # to access different parts of the cache concurrently. # The current RLock ensures that only one thread can modify the cache's internal state (the dictionary and the doubly linked list) at a time. # This prevents race conditions where multiple threads might try to update the same node, add/remove nodes, or evict entries simultaneously, leading to data corruption or incorrect LRU ordering. # The use of RLock (Reentrant Lock) allows a thread to acquire the lock multiple times, which can be useful in complex scenarios but doesn't fundamentally change the contention characteristics compared to a standard Lock in this specific usage. # --- 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) # Simulate some work and allow other threads to run time.sleep(0.001) def run_demonstration(): print("Running LRU Cache Demonstration...") 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) # Keys that might or might not be in 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"Demonstration finished. Final cache size: {len(cache)}") # Assert that the cache never exceeds its capacity assert len(cache) <= capacity, f"Cache capacity exceeded! Current size: {len(cache)}, Capacity: {capacity}" print("Assertion passed: Cache capacity is maintained.") # Basic check for data integrity (not exhaustive) print("Performing basic integrity check...") # This is a very basic check. A more robust test would involve tracking expected states. # For now, we just ensure no obvious corruption like negative size or exceptions during access. try: # Try accessing a few keys that were likely put for i in range(min(5, capacity)): key_to_check = (i * ops_per_thread) % (capacity * 2) cache.get(key_to_check) print("Basic integrity check passed.") except Exception as e: print(f"Integrity check failed with exception: {e}") assert False, "Cache integrity check failed." if __name__ == "__main__": run_demonstration() ```

"""Concurrent LRU Cache (strict LRU) with minimized contention. Concurrency strategy (brief): - The cache uses a dict for key->Node and a doubly-linked list for global LRU order. - To reduce contention for key-value operations, the key space is striped across multiple locks. Updates to a particular key (insert/update/delete/read of the mapping and node fields) use only that key’s stripe lock. - To preserve *strict global LRU semantics*, all mutations of the LRU list are serialized with a dedicated lru_lock. This avoids races that could corrupt the list and ensures there is a single total order of “touch” operations. - Eviction requires coordination: under lru_lock we select the current tail (least recently used) and then acquire that key’s stripe lock to verify it is still the same node and remove it from the map. This avoids evicting a concurrently-updated key incorrectly. Why this is correct: - The LRU list is only mutated under lru_lock, so it cannot become corrupted and each touch (get/put for existing keys) is totally ordered, yielding strict LRU. - The map entry and node’s value are protected by the stripe lock for that key, preventing concurrent updates/deletes for the same key from corrupting state. - Evictions are based on the actual tail at the moment eviction is decided under lru_lock; if the candidate node changes due to a concurrent touch/delete, eviction retries. Note: Python does not provide true lock-free primitives for complex shared structures; this implementation avoids a single global lock around *everything* by separating concerns: - Key-level operations are striped. - Only list-order maintenance uses a single lock. A demonstration is included in __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 # set False when deleted/evicted 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] = {} # Sentinel head/tail for doubly-linked list (MRU at head.next, LRU at tail.prev) self._head = _Node("<HEAD>", None) self._tail = _Node("<TAIL>", None) self._head.next = self._tail self._tail.prev = self._head # Serializes list operations and size/eviction decisions self._lru_lock = threading.Lock() # Striped locks for key-level map/node operations self._stripes_n = int(stripes) self._stripes: List[threading.Lock] = [threading.Lock() for _ in range(self._stripes_n)] # Size tracked under lru_lock self._size = 0 def _stripe_lock(self, key: Any) -> threading.Lock: return self._stripes[hash(key) % self._stripes_n] # ----- Doubly-linked list helpers (require holding 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 must be in the list self._unlink(node) self._link_front(node) def _current_lru_node(self) -> Optional[_Node]: # returns actual LRU node (tail.prev) or None if empty n = self._tail.prev if n is None or n is self._head: return None return n # ----- Public 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 # Move-to-front requires the list lock; do it outside stripe lock to reduce contention. with self._lru_lock: # Re-check membership under stripe lock to avoid moving a deleted/replaced node. with lock: cur = self._map.get(key) if cur is not node or cur is None or not cur.alive: return value # value was valid at time of read; touch lost due to race. # Under lru_lock, we can safely move this node. self._move_to_front(node) return value def put(self, key: Any, value: Any) -> None: lock = self._stripe_lock(key) # First, check if key exists and update under stripe lock. with lock: node = self._map.get(key) if node is not None and node.alive: node.value = value # Touch (move to front) under lru_lock. # Release stripe lock quickly? We can keep it short: update done. pass else: node = None if node is not None: with self._lru_lock: # Ensure still present and same 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 # Insert new node new_node = _Node(key, value) with self._lru_lock: # Acquire stripe lock while holding lru_lock to ensure consistent insertion+link. with lock: existing = self._map.get(key) if existing is not None and existing.alive: # Someone inserted concurrently; update that instead. existing.value = value # Touch it self._move_to_front(existing) return # Insert mapping and link self._map[key] = new_node self._link_front(new_node) self._size += 1 # Evict while over capacity while self._size > self.capacity: if not self._evict_one_locked(): # If eviction failed due to races, retry continue def _evict_one_locked(self) -> bool: """Evict a single LRU node. Requires holding self._lru_lock. Returns True if an eviction happened, False if needs retry. """ 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: # victim changed or already removed; retry return False # Remove from map first del self._map[vkey] victim.alive = False # Now remove from list (we hold lru_lock) self._unlink(victim) self._size -= 1 return True def delete(self, key: Any) -> bool: lock = self._stripe_lock(key) # Acquire lru_lock first to avoid deadlocks with eviction/insert path. 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 # Unlink from LRU list 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/check routine: verifies internal consistency.""" with self._lru_lock: # Check list structure and count nodes 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 and list agree (best-effort; need stripe locks to be perfect). # Under lru_lock, map may be changing under stripe locks; so we only do a conservative check: # every list node must exist in 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) # Occasionally assert size not exceeding capacity 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() # Run a bit time.sleep(1.5) # Stop and join stop.set() for t in threads: t.join(timeout=2.0) # No worker errors if errors: raise errors[0] # Final invariant checks assert len(cache) <= cap cache._assert_invariants() # Basic sanity: repeated put/get on same key from multiple threads should not corrupt 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.")