Implement a Lock-Free Concurrent LRU Cache

import threading
import random
import time
from typing import Any, Optional, List

class _Node:
"""
Represents a node in the doubly linked list used for LRU ordering.
Uses slots for memory efficiency.
"""
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:
"""
A custom Read-Write Lock implementation. This lock allows multiple readers
to hold the lock concurrently, but only one writer at a time. A writer
must wait for all readers to finish, and readers must wait for a writer
to finish. This is a readers-preferring implementation, meaning new readers
can acquire the read lock even if a writer is waiting.

This class also provides context managers for cleaner lock acquisition and release.
"""
def __init__(self):
    self._read_lock = threading.Lock() # Protects _readers count and condition variable
    self._write_lock = threading.Lock() # The actual write lock
    self._readers = 0 # Number of active readers
    self._no_readers = threading.Condition(self._read_lock) # Condition for writers to wait for readers

def acquire_read(self):
    """Acquire a read lock. Multiple readers can hold this lock concurrently."""
    with self._read_lock:
        # Wait if a writer currently holds the write lock.
        # New readers will block if a writer is active.
        while self._write_lock.locked():
            self._no_readers.wait()
        self._readers += 1

def release_read(self):
    """Release a read lock."""
    with self._read_lock:
        self._readers -= 1
        # If no readers left, notify any waiting writers
        if self._readers == 0:
            self._no_readers.notify_all()

def acquire_write(self):
    """Acquire a write lock. Only one writer can hold this lock at a time."""
    # Acquire the main write lock first to signal intent to write
    self._write_lock.acquire()
    with self._read_lock: # Then acquire read_lock to wait for active readers
        # Wait until all active readers have finished
        while self._readers > 0:
            self._no_readers.wait()

def release_write(self):
    """Release a write lock."""
    with self._read_lock: # Acquire read_lock to notify potential readers/writers
        self._no_readers.notify_all() # Notify any waiting readers or writers
    self._write_lock.release() # Release the main write lock

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):
    """Returns a context manager for acquiring a read lock."""
    return self._ReadContextManager(self)

def write_locked(self):
    """Returns a context manager for acquiring a write lock."""
    return self._WriteContextManager(self)

class LRUCache:
"""
A thread-safe LRU (Least Recently Used) cache implementation.

This cache uses a ReadWriteLock to manage concurrent access. The design aims
to allow concurrent reads for cache misses and the `keys()` operation.
However, operations that modify the cache state or LRU order (`put`, `delete`,
and cache hits in `get`) acquire a write lock, ensuring atomicity and consistency
for these critical sections. This approach provides better concurrency than a
single global mutex for all operations, especially for read-heavy workloads
with frequent cache misses or `keys()` calls.

The `get` operation employs an "upgrade" pattern: it first attempts a read lock
to check for key existence. If found, it releases the read lock and acquires a
write lock to update the LRU order and retrieve the latest value. This pattern
is carefully implemented to handle race conditions that might occur during the
lock upgrade (e.g., key being deleted or updated by another thread).
"""
def __init__(self, capacity: int):
    if capacity <= 0:
        raise ValueError("Capacity must be a positive integer.")
    self._capacity = capacity
    self._cache = {}  # Stores key -> _Node for O(1) lookup
    self._size = 0

    # Dummy head and tail nodes for the doubly linked list.
    # This simplifies edge cases for adding/removing nodes, as actual data nodes
    # are always between head and tail.
    self._head = _Node("dummy_head", None)
    self._tail = _Node("dummy_tail", None)
    self._head.next = self._tail
    self._tail.prev = self._head

    # Concurrency control: A single ReadWriteLock protects the entire cache state.
    self._rw_lock = ReadWriteLock()

def _add_node(self, node: _Node):
    """Adds a node right after the head (most recently used position).
    Assumes write lock is held."""
    node.prev = self._head
    node.next = self._head.next
    self._head.next.prev = node
    self._head.next = node

def _remove_node(self, node: _Node):
    """Removes a node from the doubly linked list.
    Assumes write lock is held."""
    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):
    """Moves an existing node to the front of the list (most recently used).
    Assumes write lock is held."""
    self._remove_node(node)
    self._add_node(node)

def _pop_tail(self) -> _Node:
    """Removes and returns the least recently used node (the one just before the tail).
    Assumes write lock is held."""
    node = self._tail.prev
    self._remove_node(node)
    return node

def get(self, key: str) -> Optional[Any]:
    """
    Returns the value associated with the key if it exists, and marks it as recently used.
    Returns None if the key is not in the cache.
    
    This operation first acquires a read lock to check for key existence. If the key
    is found, it releases the read lock and then acquires a write lock to update the
    LRU order and retrieve the latest value. This "upgrade" pattern allows concurrent
    `get` misses and `keys()` operations, but serializes `get` hits with other writes.
    """
    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
    
    # If key found, acquire write lock to update LRU order and get the latest value.
    # This is an "upgrade" pattern. A race condition can occur here: another thread
    # might delete or update the key between releasing the read lock and acquiring the write lock.
    with self._rw_lock.write_locked():
        # Re-check if the key still exists and refers to the exact same node object.
        # This is crucial for correctness under contention. If another thread performed
        # a `put` for the same key, `_cache.get(key)` would return a *new* node.
        # If another thread performed a `delete`, it would return `None`.
        current_node = self._cache.get(key)
        if current_node is node_found_under_read_lock: # Ensure it's the exact same node object
            self._move_to_front(node_found_under_read_lock)
            value = current_node.value # Get the latest value after moving
            return value
        else:
            # The node was either deleted or replaced by another 'put' operation
            # between our read-lock and write-lock acquisition. Treat as a miss.
            return None

def put(self, key: str, value: Any) -> None:
    """
    Inserts or updates the key-value pair. If the cache exceeds capacity after insertion,
    evicts the least recently used item. This is a write operation and acquires a write lock.
    """
    with self._rw_lock.write_locked():
        node = self._cache.get(key)
        if node:
            # Key exists: update value and move to front
            node.value = value
            self._move_to_front(node)
        else:
            # Key does not exist: add new node
            new_node = _Node(key, value)
            self._cache[key] = new_node
            self._add_node(new_node)
            self._size += 1

            # Evict if capacity exceeded
            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:
    """
    Removes the key from the cache. Returns True if the key was present, False otherwise.
    This is a write operation and acquires a write 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]:
    """
    Returns a list of all keys currently in the cache, ordered from most recently used
    to least recently used. This is a read operation and acquires a read lock.
    """
    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():
"""
Runs a series of single-threaded and multi-threaded tests to verify the LRUCache implementation.
"""
print("--- Starting LRU Cache Tests ---")

# --- Single-threaded Tests ---
print("\n--- Single-threaded Tests ---")
cache = LRUCache(capacity=3)
assert cache.get("a") is None, "ST: Get non-existent key should return None"
assert cache.keys() == [], "ST: Empty cache should have empty keys list"

cache.put("a", 1)
assert cache.get("a") == 1, "ST: Get existing key after put"
assert cache.keys() == ["a"], "ST: Keys list after one put"

cache.put("b", 2)
cache.put("c", 3)
assert cache.keys() == ["c", "b", "a"], "ST: Keys list after multiple puts"
assert cache._size == 3, "ST: Cache size should be 3"

# Test capacity eviction
cache.put("d", 4) # 'a' should be evicted as it's LRU
assert cache.get("a") is None, "ST: Evicted key 'a' should be None"
assert cache.get("d") == 4, "ST: New key 'd' should be present"
assert cache.keys() == ["d", "c", "b"], "ST: Keys list after eviction"
assert cache._size == 3, "ST: Cache size should remain 3 after eviction"

# Test put with existing key (update value and move to front)
cache.put("b", 22) # 'b' should move to front, value updated
assert cache.get("b") == 22, "ST: Updated value for 'b'"
assert cache.keys() == ["b", "d", "c"], "ST: Keys list after update and move 'b'"

# Test delete
assert cache.delete("c") is True, "ST: Delete existing key 'c'"
assert cache.get("c") is None, "ST: Deleted key 'c' should be None"
assert cache.keys() == ["b", "d"], "ST: Keys list after deleting 'c'"
assert cache._size == 2, "ST: Cache size after delete"

assert cache.delete("z") is False, "ST: Delete non-existent key 'z'"
assert cache.keys() == ["b", "d"], "ST: Keys list unchanged after deleting non-existent key"

# Test capacity 1
cache_cap1 = LRUCache(capacity=1)
cache_cap1.put("x", 10)
assert cache_cap1.get("x") == 10, "ST: Capacity 1, get 'x'"
cache_cap1.put("y", 20) # 'x' should be evicted
assert cache_cap1.get("x") is None, "ST: Capacity 1, 'x' evicted"
assert cache_cap1.get("y") == 20, "ST: Capacity 1, get 'y'"
assert cache_cap1.keys() == ["y"], "ST: Capacity 1, keys list"

print("Single-threaded tests passed!")

# --- Multi-threaded Tests ---
print("\n--- Multi-threaded Tests ---")
MT_CAPACITY = 10
MT_NUM_THREADS = 8
MT_OPS_PER_THREAD = 2000
MT_KEY_RANGE = 20 # Keys from 0 to 19 (e.g., "0", "1", ..., "19")

mt_cache = LRUCache(capacity=MT_CAPACITY)
# Barrier to ensure all worker threads start their operations simultaneously
barrier = threading.Barrier(MT_NUM_THREADS + 1) # +1 for the main thread
error_flag = threading.Event() # Set if any thread encounters an error

# Tracks keys that have *ever* been successfully put into the cache by any thread.
# Used to verify that 'get' never returns a value for a key that was truly never inserted.
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 function for multi-threaded tests."""
    barrier_instance.wait() # Wait for all threads to be ready to start operations

    for i in range(MT_OPS_PER_THREAD):
        if error_event.is_set():
            break # Stop if another thread already found an error

        # Randomly choose an operation with a bias towards 'get' and '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:
                # Assert that if a value is returned, the key must have been inserted at some point.
                with ever_inserted_keys_lock:
                    if key not in ever_inserted_keys_set:
                        print(f"MT Error: Thread {thread_id} got value '{val}' for key '{key}' "
                              f"which was never inserted globally. This should not happen.")
                        error_event.set()
                        break
                # Assert that the value is not corrupted (simple format check)
                if not isinstance(val, str) or not val.startswith("value_T"):
                    print(f"MT Error: Thread {thread_id} got corrupted value '{val}' for key '{key}'.")
                    error_event.set()
                    break
        elif op == 'delete':
            cache_instance.delete(key)
        
        # Optional: Introduce a small random sleep to increase context switching and contention
        # time.sleep(random.uniform(0.00001, 0.0001))

    # print(f"Thread {thread_id} finished.") # Uncomment for verbose output

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()

# Wait for all worker threads to reach the barrier and start operations
barrier.wait()
print(f"All {MT_NUM_THREADS} threads started operations.")

# Wait for all worker threads to complete their operations
for thread in threads:
    thread.join()

print("All threads completed.")

# Final assertions for multi-threaded tests
assert not error_flag.is_set(), "MT: Errors detected during multi-threaded operations. Check console for details."
assert mt_cache._size <= MT_CAPACITY, f"MT: Cache size ({mt_cache._size}) exceeded capacity ({MT_CAPACITY})."

# Check that all keys remaining in the cache were at some point inserted
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 Error: Final cache contains key '{key}' that was never globally inserted."

print("Multi-threaded tests passed!")
print("\n--- All LRU Cache Tests Passed! ---")

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:
"""
A simple writer-preferred readers-writer lock.

Concurrency strategy:
- Multiple readers may hold the lock concurrently.
- Writers are exclusive.
- Once a writer is waiting, new readers are blocked so writers are not starved.

This is used to protect the dictionary that maps keys to nodes.
Reads that only need membership/value lookup can proceed concurrently.
Operations that mutate the index or the LRU list take the write lock.

Note:
Python does not provide true lock-free primitives for building a fully
lock-free LRU in the standard library. This implementation instead avoids
a single global mutex for all access by using:
- a readers-writer lock for the key->node index
- a separate lock for the doubly linked LRU list
- per-node locks for node-local state

That gives substantially better concurrency than a single global lock while
preserving correctness under 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:
"""
Thread-safe LRU cache with fine-grained synchronization.

Design overview
---------------
Data structures:
- self._map: dict[str, _Node]
  Maps keys to nodes. Protected by self._index_lock (RWLock).
- Doubly linked list with sentinels self._head (MRU side) and self._tail (LRU side)
  Protected by self._list_lock.

Locking rules
-------------
1. Dictionary/index reads use self._index_lock.acquire_read().
2. Dictionary/index mutations use self._index_lock.acquire_write().
3. Any pointer changes in the linked list use self._list_lock.
4. Per-node locks serialize access to a node's value/deleted state.

Ordering rule to avoid deadlock
--------------------------------
When multiple locks are needed, operations follow a stable order:
  index lock -> list lock -> node lock(s)

Concurrency trade-offs
----------------------
- get() needs to update recency, so it cannot be a pure read. It first performs
  a concurrent index lookup under a read lock, then briefly takes the list lock
  to move the located node to the MRU position.
- put(), delete(), and eviction modify both index and list, so they take the
  index write lock and the list lock.
- This is not strictly lock-free; Python's standard library does not make a
  practical lock-free linked-list LRU feasible. It is, however, more concurrent
  than a design using one global mutex for every operation.
"""

def __init__(self, capacity: int):
    """Initialize the cache with a positive maximum capacity."""
    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:
    """Insert node immediately after head. Caller must hold 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:
    """Detach node from list. Caller must hold 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:
    """
    Move an existing node to MRU position if it is still live.

    Uses the list lock plus the node lock. The map read lock held by caller
    ensures the node object remains reachable from the index during lookup.
    """
    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]:
    """
    Return the value for key if present and mark it most recently used.
    Return None if absent.
    """
    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:
    """
    Insert or update a key-value pair.
    If capacity is exceeded, evict the least recently used item.
    """
    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:
    """
    Remove key from the cache.
    Return True if key existed, otherwise 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]:
    """
    Return keys from most recently used to least recently used.

    We take the index read lock and list lock to obtain a consistent snapshot.
    """
    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:
    """Internal helper for tests."""
    self._index_lock.acquire_read()
    try:
        return len(self._map)
    finally:
        self._index_lock.release_read()

def _debug_validate(self) -> None:
    """
    Internal consistency checker for tests.

    Verifies:
    - list links are well-formed
    - every live list node is present in the map
    - no duplicate keys in the list
    - size never exceeds capacity
    """
    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:
"""
Run single-threaded and multi-threaded tests.

Multi-threaded test requirements covered:
- at least 8 threads
- mix of get/put/delete on overlapping keys
- assert cache never exceeds capacity
- assert get never returns a value for a key that was never inserted
"""

# 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()

# Concurrent 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)

# Focused same-key contention test
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()