Initial benchmark suite: 8 graded models + cyberpunk dashboard generator
- prompts/: LFU cache exam + 5-pillar grading rubric - outputs/: 8 model .py outputs (local + cloud baseline) - data/benchmark_history.json: graded results (scores, metrics, bugs, patches) - generate_dashboard.py: builds dashboard.html + pages/*.html from JSON - Dockerfile + DEPLOY.md: Gitea→Coolify deploy (build-step, nginx static) - .gitignore: generated HTML excluded (built on deploy) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -0,0 +1,262 @@
|
||||
import asyncio
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class Node:
|
||||
key: str
|
||||
value: Any
|
||||
ttl_expiry: float
|
||||
freq: int = 1
|
||||
|
||||
def __repr__(self):
|
||||
return f"Node({self.key}, freq={self.freq})"
|
||||
|
||||
|
||||
class DoublyLinkedListList:
|
||||
def __init__(self):
|
||||
self.head = Node("", "", 0) # Dummy head
|
||||
self.tail = Node("", "", 0) # Dummy tail
|
||||
self.head.next = self.tail
|
||||
self.tail.prev = self.head
|
||||
|
||||
def add_to_front(self, node: Node):
|
||||
node.next = self.head.next
|
||||
node.prev = self.head
|
||||
self.head.next = node
|
||||
node.next.prev = node
|
||||
|
||||
def remove(self, node: Node):
|
||||
node.prev.next = node.next
|
||||
node.next.prev = node.prev
|
||||
|
||||
def move_to_front(self, node: Node):
|
||||
self.remove(node)
|
||||
self.add_to_front(node)
|
||||
|
||||
def pop_tail(self) -> Optional[Node]:
|
||||
if self.tail == self.tail.prev: # Only dummy tail left
|
||||
return None
|
||||
node = self.tail.prev
|
||||
self.remove(node)
|
||||
return node
|
||||
|
||||
|
||||
class ConcurrentLFUCache:
|
||||
def __init__(self, capacity: int):
|
||||
if capacity <= 0:
|
||||
raise ValueError("Capacity must be positive")
|
||||
self.capacity = capacity
|
||||
self.cache: Dict[str, Node] = {} # key -> node
|
||||
self.freq_buckets: Dict[int, DoublyLinkedListList] = {} # freq -> list
|
||||
self.min_freq = 0
|
||||
self.global_lock = asyncio.Lock() # Protects global cache structure
|
||||
self.evictor_task: Optional[asyncio.Task] = None
|
||||
|
||||
def _update_freq(self, node: Node):
|
||||
old_freq = node.freq
|
||||
bucket = self.freq_buckets[old_freq]
|
||||
bucket.remove(node)
|
||||
|
||||
if old_freq == self.min_freq and bucket.tail == bucket.head:
|
||||
self.min_freq += 1
|
||||
|
||||
node.freq += 1
|
||||
if node.freq not in self.freq_buckets:
|
||||
self.freq_buckets[node.freq] = DoublyLinkedListList()
|
||||
|
||||
bucket = self.freq_buckets[node.freq]
|
||||
bucket.add_to_front(node)
|
||||
|
||||
async def get(self, key: str) -> Optional[Any]:
|
||||
now = time.time()
|
||||
async with self.global_lock:
|
||||
node = self.cache.get(key)
|
||||
if not node:
|
||||
return None
|
||||
|
||||
if now > node.ttl_expiry: # Lazy eviction
|
||||
bucket = self.freq_buckets[node.freq]
|
||||
bucket.remove(node)
|
||||
del self.cache[key]
|
||||
return None
|
||||
|
||||
self._update_freq(node)
|
||||
return node.value
|
||||
|
||||
async def put(self, key: str, value: Any, ttl_seconds: float):
|
||||
now = time.time()
|
||||
expiry = now + ttl_seconds
|
||||
|
||||
async with self.global_lock:
|
||||
if key in self.cache:
|
||||
node = self.cache[key]
|
||||
bucket = self.freq_buckets[node.freq]
|
||||
bucket.remove(node)
|
||||
node.value = value
|
||||
node.ttl_expiry = expiry
|
||||
self._update_freq(node)
|
||||
else:
|
||||
if len(self.cache) >= self.capacity:
|
||||
bucket = self.freq_buckets[self.min_freq]
|
||||
evicted = bucket.pop_tail()
|
||||
if evicted:
|
||||
del self.cache[evicted.key]
|
||||
|
||||
new_node = Node(key, value, expiry)
|
||||
self.cache[key] = new_node
|
||||
if self.min_freq == 0:
|
||||
self.min_freq = 1
|
||||
bucket = self.freq_buckets[self.min_freq]
|
||||
bucket.add_to_front(new_node)
|
||||
|
||||
async def start_evictor(self, interval: float = 1.0):
|
||||
def evict_loop():
|
||||
while True:
|
||||
time.sleep(interval)
|
||||
now = time.time()
|
||||
expired_keys = [k for k, n in self.cache.items() if now > n.ttl_expiry]
|
||||
for k in expired_keys:
|
||||
asyncio.run(self.put(k, self.cache[k].value, 0)) # Re-insert with 0 TTL
|
||||
|
||||
self.evictor_task = asyncio.create_task(evict_loop())
|
||||
|
||||
async def stop_evictor(self):
|
||||
if self.evictor_task:
|
||||
self.evictor_task.cancel()
|
||||
|
||||
|
||||
class Transaction:
|
||||
def __init__(self, cache):
|
||||
self.cache = cache
|
||||
self.local_cache: Dict[str, Node] = {} # key -> node
|
||||
self.local_freq: Dict[int, DoublyLinkedListList] = {} # freq -> list
|
||||
self.local_min_freq = 0
|
||||
|
||||
def _update_local_freq(self, node: Node):
|
||||
old_freq = node.freq
|
||||
bucket = self.local_freq[old_freq]
|
||||
bucket.remove(node)
|
||||
|
||||
if old_freq == self.local_min_freq and bucket.tail == bucket.head:
|
||||
self.local_min_freq += 1
|
||||
|
||||
node.freq += 1
|
||||
if node.freq not in self.local_freq:
|
||||
self.local_freq[node.freq] = DoublyLinkedListList()
|
||||
|
||||
bucket = self.local_freq[node.freq]
|
||||
bucket.add_to_front(node)
|
||||
|
||||
async def get(self, key: str) -> Optional[Any]:
|
||||
if key in self.local_cache:
|
||||
node = self.local_cache[key]
|
||||
self._update_local_freq(node)
|
||||
return node.value
|
||||
|
||||
async with self.cache.global_lock:
|
||||
node = self.cache.cache.get(key)
|
||||
if not node:
|
||||
return None
|
||||
|
||||
# Copy the node so local changes don't affect global cache until commit
|
||||
new_node = Node(key, node.value, node.ttl_expiry, node.freq)
|
||||
self.local_cache[key] = new_node
|
||||
if self.local_min_freq == 0:
|
||||
self.local_min_freq = new_node.freq
|
||||
bucket = self.local_freq[new_node.freq]
|
||||
bucket.add_to_front(new_node)
|
||||
return new_node.value
|
||||
|
||||
async def put(self, key: str, value: Any, ttl_seconds: float):
|
||||
now = time.time()
|
||||
expiry = now + ttl_seconds
|
||||
|
||||
if key in self.local_cache:
|
||||
node = self.local_cache[key]
|
||||
bucket = self.local_freq[node.freq]
|
||||
bucket.remove(node)
|
||||
node.value = value
|
||||
node.ttl_expiry = expiry
|
||||
self._update_local_freq(node)
|
||||
else:
|
||||
if len(self.local_cache) >= self.cache.capacity:
|
||||
bucket = self.local_freq[self.local_min_freq]
|
||||
evicted = bucket.pop_tail()
|
||||
if evicted:
|
||||
del self.local_cache[evicted.key]
|
||||
|
||||
new_node = Node(key, value, expiry)
|
||||
self.local_cache[key] = new_node
|
||||
if self.local_min_freq == 0:
|
||||
self.local_min_freq = new_node.freq
|
||||
bucket = self.local_freq[new_node.freq]
|
||||
bucket.add_to_front(new_node)
|
||||
|
||||
async def commit(self):
|
||||
async with self.cache.global_lock:
|
||||
for key, node in self.local_cache.items():
|
||||
if key in self.cache.cache:
|
||||
old_node = self.cache.cache[key]
|
||||
bucket = self.cache.freq_buckets[old_node.freq]
|
||||
bucket.remove(old_node)
|
||||
self.cache.cache[key] = node
|
||||
if old_node.freq == self.cache.min_freq and bucket.tail == bucket.head:
|
||||
self.cache.min_freq += 1
|
||||
node.freq = node.freq # freq is already updated in local_cache
|
||||
bucket = self.cache.freq_buckets[node.freq]
|
||||
if node.freq not in self.cache.freq_buckets:
|
||||
self.cache.freq_buckets[node.freq] = DoublyLinkedListList()
|
||||
bucket.add_to_front(node)
|
||||
|
||||
async def rollback(self):
|
||||
self.local_cache = {}
|
||||
self.local_freq = {}
|
||||
self.local_min_freq = 0
|
||||
|
||||
|
||||
async def main():
|
||||
cache = ConcurrentLFUCache(capacity=3)
|
||||
|
||||
# a. O(1) LFU eviction order
|
||||
await cache.put("a", "val_a", 60) # freq=1
|
||||
await cache.put("b", "val_b", 60) # freq=1
|
||||
await cache.put("c", "val_c", 60) # freq=1
|
||||
await cache.get("a") # a:freq=2, b/c:freq=1
|
||||
await cache.get("b") # a:freq=2, b:freq=2, c:freq=1
|
||||
await cache.put("d", "val_d", 60) # c is evicted (lowest freq=1)
|
||||
assert await cache.get("c") is None
|
||||
|
||||
# b. Lazy vs Background eviction
|
||||
await cache.put("e", "val_e", 0) # expire immediately
|
||||
assert await cache.get("e") is None # lazy eviction
|
||||
|
||||
# c. Transaction commit vs rollback
|
||||
tx = await cache.begin_transaction() # helper below
|
||||
await tx.put("f", "val_f", 60)
|
||||
assert await tx.get("f") == "val_f" # Read Your Own Writes
|
||||
assert await cache.get("f") is None # Global not yet committed
|
||||
|
||||
await tx.rollback()
|
||||
assert await cache.get("f") is None # Rollback restored
|
||||
|
||||
await tx.commit()
|
||||
assert await cache.get("f") == "val_f" # Committed
|
||||
|
||||
# d. Stress test
|
||||
async def worker(i):
|
||||
await cache.put(f"key_{i}", i, 10)
|
||||
await cache.get(f"key_{i}")
|
||||
|
||||
tasks = [worker(i) for i in range(50)]
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
def begin_transaction(): # Helper for the test suite
|
||||
return Transaction(ConcurrentLFUCache(capacity=3))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
@@ -0,0 +1,343 @@
|
||||
import asyncio
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, Optional, Generic, TypeVar, Set
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
# --- Internal Data Structures for O(1) LFU ---
|
||||
|
||||
@dataclass
|
||||
class Node:
|
||||
"""A node in the LFU frequency doubly linked list."""
|
||||
key: Any
|
||||
value: Any
|
||||
freq: int = 1
|
||||
expiry: Optional[float] = None
|
||||
prev: Optional['Node'] = None
|
||||
next: Optional['Node'] = None
|
||||
|
||||
class DoublyLinkedList:
|
||||
"""A standard DLL to maintain nodes of the same frequency."""
|
||||
def __init__(self):
|
||||
self.head: Optional[Node] = None
|
||||
self.tail: Optional[Node] = None
|
||||
self.size = 0
|
||||
|
||||
def append(self, node: Node):
|
||||
"""Add node to the front (most recently used in this freq bucket)."""
|
||||
node.next = self.head
|
||||
node.prev = None
|
||||
if self.head:
|
||||
self.head.prev = node
|
||||
self.head = node
|
||||
if not self.tail:
|
||||
self.tail = node
|
||||
self.size += 1
|
||||
|
||||
def pop_tail(self) -> Optional[Node]:
|
||||
"""Remove and return the least recently used node in this freq bucket."""
|
||||
if not self.tail:
|
||||
return None
|
||||
node = self.tail
|
||||
self.remove(node)
|
||||
return node
|
||||
|
||||
def remove(self, node: Node):
|
||||
"""Remove a specific node from the list in O(1)."""
|
||||
if node.prev:
|
||||
node.prev.next = node.next
|
||||
else:
|
||||
self.head = node.next
|
||||
|
||||
if node.next:
|
||||
node.next.prev = node.prev
|
||||
else:
|
||||
self.tail = node.prev
|
||||
|
||||
node.next = None
|
||||
node.prev = None
|
||||
self.size -= 1
|
||||
|
||||
# --- Cache Implementation ---
|
||||
|
||||
class LFUCache:
|
||||
def __init__(self, capacity: int):
|
||||
if capacity <= 0:
|
||||
raise ValueError("Capacity must be greater than 0")
|
||||
|
||||
self.capacity = capacity
|
||||
self.cache: Dict[Any, Node] = {} # key -> Node
|
||||
self.freq_map: Dict[int, DoublyLinkedList] = {} # freq -> DLL
|
||||
self.min_freq = 0
|
||||
self.lock = asyncio.Lock()
|
||||
self._evictor_task: Optional[asyncio.Task] = None
|
||||
|
||||
def _get_now(self) -> float:
|
||||
return time.time()
|
||||
|
||||
def _is_expired(self, node: Node) -> bool:
|
||||
return node.expiry is not None and self._get_now() > node.expiry
|
||||
|
||||
async def _update_freq(self, node: Node):
|
||||
"""Moves a node to the next frequency bucket in O(1)."""
|
||||
old_freq = node.freq
|
||||
dll = self.freq_map[old_freq]
|
||||
dll.remove(node)
|
||||
|
||||
if old_freq == self.min_freq and dll.size == 0:
|
||||
self.min_freq += 1
|
||||
|
||||
node.freq += 1
|
||||
if node.freq not in self.freq_map:
|
||||
self.freq_map[node.freq] = DoublyLinkedList()
|
||||
self.freq_map[node.freq].append(node)
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
"""O(1) Access with Lazy TTL check."""
|
||||
async with self.lock:
|
||||
if key not in self.cache:
|
||||
return None
|
||||
|
||||
node = self.cache[key]
|
||||
if self._is_expired(node):
|
||||
await self._delete_node(node)
|
||||
return None
|
||||
|
||||
await self._update_freq(node)
|
||||
return node.value
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: Optional[float] = None):
|
||||
"""O(1) Insertion with LFU eviction and Lazy TTL check."""
|
||||
async with self.lock:
|
||||
expiry = self._get_now() + ttl_seconds if ttl_seconds else None
|
||||
|
||||
if key in self.cache:
|
||||
node = self.cache[key]
|
||||
node.value = value
|
||||
node.expiry = expiry
|
||||
await self._update_freq(node)
|
||||
return
|
||||
|
||||
# Evict if capacity reached
|
||||
if len(self.cache) >= self.capacity:
|
||||
# Priority 1: Try to find an expired key first (optimistic)
|
||||
# Otherwise: Evict the LFU (min_freq tail)
|
||||
evicted = self.freq_map[self.min_freq].pop_tail()
|
||||
if evicted:
|
||||
del self.cache[evicted.key]
|
||||
|
||||
# Create new node
|
||||
new_node = Node(key=key, value=value, expiry=expiry)
|
||||
self.cache[key] = new_node
|
||||
self.min_freq = 1
|
||||
if 1 not in self.freq_map:
|
||||
self.freq_map[1] = DoublyLinkedList()
|
||||
self.freq_map[1].append(new_node)
|
||||
|
||||
async def delete(self, key: Any):
|
||||
"""O(1) deletion."""
|
||||
async with self.lock:
|
||||
if key in self.cache:
|
||||
await self._delete_node(self.cache[key])
|
||||
|
||||
async def _delete_node(self, node: Node):
|
||||
"""Internal helper to remove node from all tracking structures."""
|
||||
dll = self.freq_map.get(node.freq)
|
||||
if dll:
|
||||
dll.remove(node)
|
||||
if node.key in self.cache:
|
||||
del self.cache[node.key]
|
||||
|
||||
# --- Background Evictor ---
|
||||
|
||||
async def start_evictor(self, interval: float = 1.0, batch_size: int = 100):
|
||||
"""Starts a non-blocking background task to purge expired keys."""
|
||||
if self._evictor_task:
|
||||
return
|
||||
|
||||
async def evict_loop():
|
||||
while True:
|
||||
await asyncio.sleep(interval)
|
||||
# Collect keys to check outside the lock to prevent blocking reads
|
||||
keys_to_check = list(self.cache.keys())
|
||||
|
||||
# Process in small batches to yield to the event loop
|
||||
for i in range(0, len(keys_to_check), batch_size):
|
||||
batch = keys_to_check[i : i + batch_size]
|
||||
async with self.lock:
|
||||
for k in batch:
|
||||
node = self.cache.get(k)
|
||||
if node and self._is_expired(node):
|
||||
await self._delete_node(node)
|
||||
# Yield control back to asyncio
|
||||
await asyncio.sleep(0)
|
||||
|
||||
self._evictor_task = asyncio.create_task(evict_loop())
|
||||
|
||||
async def stop_evictor(self):
|
||||
"""Stops the background eviction task."""
|
||||
if self._evictor_task:
|
||||
self._evictor_task.cancel()
|
||||
try:
|
||||
await self._evictor_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._evictor_task = None
|
||||
|
||||
# --- Transaction Support ---
|
||||
|
||||
def begin_transaction(self) -> 'Transaction':
|
||||
return Transaction(self)
|
||||
|
||||
async def apply_transaction_changes(self, changes: 'TransactionState'):
|
||||
"""Atomically apply buffered transaction writes to the global state."""
|
||||
async with self.lock:
|
||||
for key, action in changes.writes.items():
|
||||
if action == "DELETE":
|
||||
if key in self.cache:
|
||||
await self._delete_node(self.cache[key])
|
||||
else:
|
||||
val, ttl = action
|
||||
# Using put logic inside the lock
|
||||
expiry = self._get_now() + ttl if ttl else None
|
||||
if key in self.cache:
|
||||
node = self.cache[key]
|
||||
node.value = val
|
||||
node.expiry = expiry
|
||||
await self._update_freq(node)
|
||||
else:
|
||||
if len(self.cache) >= self.capacity:
|
||||
evicted = self.freq_map[self.min_freq].pop_tail()
|
||||
if evicted:
|
||||
del self.cache[evicted.key]
|
||||
new_node = Node(key=key, value=val, expiry=expiry)
|
||||
self.cache[key] = new_node
|
||||
self.min_freq = 1
|
||||
if 1 not in self.freq_map:
|
||||
self.freq_map[1] = DoublyLinkedList()
|
||||
self.freq_map[1].append(new_node)
|
||||
|
||||
@dataclass
|
||||
class TransactionState:
|
||||
writes: Dict[Any, Any] = None # key -> (value, ttl) or "DELETE"
|
||||
|
||||
class Transaction:
|
||||
def __init__(self, cache: LFUCache):
|
||||
self._cache = cache
|
||||
self._state = TransactionState(writes={})
|
||||
self._committed = False
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: Optional[float] = None):
|
||||
if self._committed: raise RuntimeError("Transaction already committed")
|
||||
self._state.writes[key] = (value, ttl_seconds)
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
if self._committed: raise RuntimeError("Transaction already committed")
|
||||
|
||||
# Read-Your-Own-Writes (RYOW)
|
||||
if key in self._state.writes:
|
||||
action = self._state.writes[key]
|
||||
if action == "DELETE":
|
||||
return None
|
||||
return action[0]
|
||||
|
||||
# Otherwise read from global cache
|
||||
return await self._cache.get(key)
|
||||
|
||||
async def delete(self, key: Any):
|
||||
if self._committed: raise RuntimeError("Transaction already committed")
|
||||
self._state.writes[key] = "DELETE"
|
||||
|
||||
async def commit(self):
|
||||
if self._committed: raise RuntimeError("Transaction already committed")
|
||||
await self._cache.apply_transaction_changes(self._state)
|
||||
self._committed = True
|
||||
|
||||
async def rollback(self):
|
||||
if self._committed: raise RuntimeError("Transaction already committed")
|
||||
self._state.writes.clear()
|
||||
self._committed = True
|
||||
|
||||
# --- Executable Unit Tests ---
|
||||
|
||||
async def main():
|
||||
print("🚀 Starting LFU Cache Suite...")
|
||||
|
||||
# 1. Test O(1) LFU Eviction
|
||||
print("\nTesting LFU Eviction Order...", end=" ")
|
||||
cache = LFUCache(capacity=2)
|
||||
await cache.put("a", 1)
|
||||
await cache.put("b", 2)
|
||||
await cache.get("a") # freq(a)=2, freq(b)=1
|
||||
await cache.put("c", 3) # b should be evicted
|
||||
assert await cache.get("b") is None
|
||||
assert await cache.get("a") == 1
|
||||
assert await cache.get("c") == 3
|
||||
print("✅ PASSED")
|
||||
|
||||
# 2. Test Dual-Layer TTL
|
||||
print("\nTesting TTL Eviction...", end=" ")
|
||||
ttl_cache = LFUCache(capacity=10)
|
||||
await ttl_cache.put("lazy", "val", ttl_seconds=0.1)
|
||||
await ttl_cache.put("async", "val", ttl_seconds=0.1)
|
||||
|
||||
# Lazy check
|
||||
await asyncio.sleep(0.2)
|
||||
assert await ttl_cache.get("lazy") is None, "Lazy eviction failed"
|
||||
|
||||
# Background check
|
||||
await ttl_cache.start_evictor(interval=0.1, batch_size=10)
|
||||
await asyncio.sleep(0.2)
|
||||
async with ttl_cache.lock:
|
||||
assert "async" not in ttl_cache.cache, "Background evictor failed"
|
||||
await ttl_cache.stop_evictor()
|
||||
print("✅ PASSED")
|
||||
|
||||
# 3. Test Atomic Transactions
|
||||
print("\nTesting Atomic Transactions...", end=" ")
|
||||
tx_cache = LFUCache(capacity=10)
|
||||
await tx_cache.put("base", 100)
|
||||
|
||||
# Test Commit
|
||||
tx1 = tx_cache.begin_transaction()
|
||||
await tx1.put("base", 200)
|
||||
await tx1.put("new", 300)
|
||||
assert await tx1.get("base") == 200 # RYOW
|
||||
assert await tx_cache.get("base") == 100 # Isolation
|
||||
await tx1.commit()
|
||||
assert await tx_cache.get("base") == 200
|
||||
assert await tx_cache.get("new") == 300
|
||||
|
||||
# Test Rollback
|
||||
tx2 = tx_cache.begin_transaction()
|
||||
await tx2.put("base", 999)
|
||||
await tx2.get("base")
|
||||
await tx2.rollback()
|
||||
assert await tx_cache.get("base") == 200
|
||||
print("✅ PASSED")
|
||||
|
||||
# 4. Stress Test: Concurrent Async Tasks
|
||||
print("\nStress Testing 50 concurrent tasks...", end=" ")
|
||||
stress_cache = LFUCache(capacity=20)
|
||||
|
||||
async def worker(worker_id):
|
||||
for i in range(100):
|
||||
key = f"key_{i % 30}" # Contention on keys
|
||||
if i % 3 == 0:
|
||||
await stress_cache.put(key, f"val_{worker_id}_{i}")
|
||||
elif i % 3 == 1:
|
||||
await stress_cache.get(key)
|
||||
else:
|
||||
tx = stress_cache.begin_transaction()
|
||||
await tx.put(key, f"tx_{worker_id}_{i}")
|
||||
await tx.commit()
|
||||
|
||||
await asyncio.gather(*(worker(i) for i in range(50)))
|
||||
print("✅ PASSED")
|
||||
|
||||
print("\n✨ All tests passed successfully!")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
@@ -0,0 +1,339 @@
|
||||
import asyncio
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, Optional, Generic, TypeVar, Set
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
@dataclass
|
||||
class Node:
|
||||
"""A node in the LFU doubly linked list."""
|
||||
key: Any
|
||||
value: Any
|
||||
freq: int = 1
|
||||
expiry: float = float('inf')
|
||||
prev: Optional['Node'] = None
|
||||
next: Optional['Node'] = None
|
||||
|
||||
class DoublyLinkedList:
|
||||
"""Standard O(1) doubly linked list for maintaining frequency buckets."""
|
||||
def __init__(self):
|
||||
self.head = Node(None, None) # Sentinel head
|
||||
self.tail = Node(None, None) # Sentinel tail
|
||||
self.head.next = self.tail
|
||||
self.tail.prev = self.head
|
||||
self.size = 0
|
||||
|
||||
def append(self, node: Node):
|
||||
"""Adds a node to the end (most recently used in this frequency)."""
|
||||
node.next = self.tail
|
||||
node.prev = self.tail.prev
|
||||
self.tail.prev.next = node
|
||||
self.tail.prev = node
|
||||
self.size += 1
|
||||
|
||||
def remove(self, node: Node):
|
||||
"""Removes a specific node from the list."""
|
||||
if node.prev:
|
||||
node.prev.next = node.next
|
||||
if node.next:
|
||||
node.next.prev = node.prev
|
||||
node.prev = None
|
||||
node.next = None
|
||||
self.size -= 1
|
||||
|
||||
def pop_front(self) -> Optional[Node]:
|
||||
"""Removes and returns the oldest node (LFU candidate)."""
|
||||
if self.size == 0:
|
||||
return None
|
||||
node = self.head.next
|
||||
self.remove(node)
|
||||
return node
|
||||
|
||||
def is_empty(self) -> bool:
|
||||
return self.size == 0
|
||||
|
||||
|
||||
class LFUCache:
|
||||
"""
|
||||
In-Memory Concurrent LFU Cache with Async TTL Eviction and Atomic Transactions.
|
||||
Time Complexity: O(1) for get and put.
|
||||
Space Complexity: O(N).
|
||||
"""
|
||||
def __init__(self, capacity: int):
|
||||
if capacity <= 0:
|
||||
raise ValueError("Capacity must be greater than 0")
|
||||
|
||||
self.capacity = capacity
|
||||
self.cache: Dict[Any, Node] = {} # Key -> Node
|
||||
self.freq_map: Dict[int, DoublyLinkedList] = {} # Freq -> DLL
|
||||
self.min_freq = 0
|
||||
self._lock = asyncio.Lock()
|
||||
self._evictor_task: Optional[asyncio.Task] = None
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
async with self._lock:
|
||||
return await self._get_internal(key)
|
||||
|
||||
async def _get_internal(self, key: Any) -> Optional[Any]:
|
||||
"""Internal get without lock for transaction use."""
|
||||
if key not in self.cache:
|
||||
return None
|
||||
|
||||
node = self.cache[key]
|
||||
|
||||
# Lazy Eviction check
|
||||
if time.time() > node.expiry:
|
||||
await self._delete_internal(key)
|
||||
return None
|
||||
|
||||
self._update_frequency(node)
|
||||
return node.value
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: Optional[float] = None):
|
||||
async with self._lock:
|
||||
await self._put_internal(key, value, ttl_seconds)
|
||||
|
||||
async def _put_internal(self, key: Any, value: Any, ttl_seconds: Optional[float] = None):
|
||||
"""Internal put without lock for transaction use."""
|
||||
expiry = time.time() + ttl_seconds if ttl_seconds is not None else float('inf')
|
||||
|
||||
if key in self.cache:
|
||||
node = self.cache[key]
|
||||
node.value = value
|
||||
node.expiry = expiry
|
||||
self._update_frequency(node)
|
||||
else:
|
||||
# Capacity Management
|
||||
if len(self.cache) >= self.capacity:
|
||||
await self._evict_lfu()
|
||||
|
||||
new_node = Node(key, value, freq=1, expiry=expiry)
|
||||
self.cache[key] = new_node
|
||||
self._add_to_freq_bucket(1, new_node)
|
||||
self.min_freq = 1
|
||||
|
||||
async def delete(self, key: Any):
|
||||
async with self._lock:
|
||||
await self._delete_internal(key)
|
||||
|
||||
async def _delete_internal(self, key: Any):
|
||||
"""Internal delete without lock for transaction use."""
|
||||
if key in self.cache:
|
||||
node = self.cache.pop(key)
|
||||
self.freq_map[node.freq].remove(node)
|
||||
# Note: We don't strictly need to update min_freq here because
|
||||
# _evict_lfu handles empty buckets by incrementing.
|
||||
|
||||
def _update_frequency(self, node: Node):
|
||||
"""Moves a node to the next frequency bucket in O(1)."""
|
||||
old_freq = node.freq
|
||||
self.freq_map[old_freq].remove(node)
|
||||
|
||||
if old_freq == self.min_freq and self.freq_map[old_freq].is_empty():
|
||||
self.min_freq += 1
|
||||
|
||||
node.freq += 1
|
||||
self._add_to_freq_bucket(node.freq, node)
|
||||
|
||||
def _add_to_freq_bucket(self, freq: int, node: Node):
|
||||
if freq not in self.freq_map:
|
||||
self.freq_map[freq] = DoublyLinkedList()
|
||||
self.freq_map[freq].append(node)
|
||||
|
||||
async def _evict_lfu(self):
|
||||
"""Evicts the least frequently used (and oldest within that freq) item."""
|
||||
while self.min_freq not in self.freq_map or self.freq_map[self.min_freq].is_empty():
|
||||
# This handles cases where items were deleted manually
|
||||
if not self.cache: return
|
||||
self.min_freq += 1 # Search for the next available freq bucket
|
||||
|
||||
evicted_node = self.freq_map[self.min_freq].pop_front()
|
||||
if evicted_node:
|
||||
del self.cache[evicted_node.key]
|
||||
|
||||
# --- Background Evictor ---
|
||||
|
||||
async def start_evictor(self, interval: float = 1.0):
|
||||
"""Starts the background task that purges expired keys."""
|
||||
if self._evictor_task is not None:
|
||||
return
|
||||
self._evictor_task = asyncio.create_task(self._eviction_loop(interval))
|
||||
|
||||
async def stop_evictor(self):
|
||||
"""Stops the background eviction task."""
|
||||
if self._evictor_task:
|
||||
self._evictor_task.cancel()
|
||||
try:
|
||||
await self._evictor_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._evictor_task = None
|
||||
|
||||
async def _eviction_loop(self, interval: float):
|
||||
while True:
|
||||
await asyncio.sleep(interval)
|
||||
# Process in small batches to avoid blocking the main lock for too long
|
||||
now = time.time()
|
||||
keys_to_check = list(self.cache.keys())
|
||||
|
||||
batch_size = 100
|
||||
for i in range(0, len(keys_to_check), batch_size):
|
||||
async with self._lock:
|
||||
batch = keys_to_check[i : i + batch_size]
|
||||
for k in batch:
|
||||
if k in self.cache and now > self.cache[k].expiry:
|
||||
await self._delete_internal(k)
|
||||
|
||||
# --- Transaction Logic ---
|
||||
|
||||
def begin_transaction(self):
|
||||
return Transaction(self)
|
||||
|
||||
|
||||
class Transaction:
|
||||
"""
|
||||
Provides ACID-like atomic updates to the LFUCache.
|
||||
Implements 'Read Your Own Writes' (RYOW).
|
||||
"""
|
||||
def __init__(self, cache: LFUCache):
|
||||
self._cache = cache
|
||||
self._pending_puts: Dict[Any, tuple[Any, Optional[float]]] = {}
|
||||
self._pending_deletes: Set[Any] = set()
|
||||
self._committed = False
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
if self._committed: raise RuntimeError("Transaction already committed")
|
||||
|
||||
# 1. Check local deletes first
|
||||
if key in self._pending_deletes:
|
||||
return None
|
||||
|
||||
# 2. Check local writes (RYOW)
|
||||
if key in self._pending_puts:
|
||||
return self._pending_puts[key][0]
|
||||
|
||||
# 3. Fallback to global cache
|
||||
return await self._cache.get(key)
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: Optional[float] = None):
|
||||
if self._committed: raise RuntimeError("Transaction already committed")
|
||||
self._pending_deletes.discard(key)
|
||||
self._pending_puts[key] = (value, ttl_seconds)
|
||||
|
||||
async def delete(self, key: Any):
|
||||
if self._committed: raise RuntimeError("Transaction already committed")
|
||||
self._pending_puts.pop(key, None)
|
||||
self._pending_deletes.add(key)
|
||||
|
||||
async def commit(self):
|
||||
"""Applies all pending changes to the global cache atomically."""
|
||||
if self._committed: return
|
||||
|
||||
async with self._cache._lock:
|
||||
# Apply deletes first
|
||||
for key in self._pending_deletes:
|
||||
await self._cache._delete_internal(key)
|
||||
# Apply puts
|
||||
for key, (val, ttl) in self._pending_puts.items():
|
||||
await self._cache._put_internal(key, val, ttl)
|
||||
|
||||
self._committed = True
|
||||
|
||||
async def rollback(self):
|
||||
"""Discards all pending changes."""
|
||||
self._pending_puts.clear()
|
||||
self._pending_deletes.clear()
|
||||
self._committed = True
|
||||
|
||||
|
||||
# ==========================================
|
||||
# Unit Tests
|
||||
# ==========================================
|
||||
|
||||
async def main():
|
||||
print("Starting LFU Cache Comprehensive Test Suite...")
|
||||
|
||||
# 1. Test O(1) LFU Eviction Order
|
||||
print("\nTesting LFU Eviction Order...", end=" ")
|
||||
cache = LFUCache(capacity=2)
|
||||
await cache.put("a", 1) # freq 1
|
||||
await cache.put("b", 2) # freq 1
|
||||
await cache.get("a") # freq 2
|
||||
await cache.put("c", 3) # "b" should be evicted as it's the LFU (freq 1)
|
||||
assert await cache.get("b") is None, "Key 'b' should have been evicted"
|
||||
assert await cache.get("a") == 1, "Key 'a' should still exist"
|
||||
assert await cache.get("c") == 3, "Key 'c' should still exist"
|
||||
print("PASSED")
|
||||
|
||||
# 2. Test TTL Eviction (Lazy & Background)
|
||||
print("Testing Dual-Layer TTL...", end=" ")
|
||||
cache = LFUCache(capacity=10)
|
||||
await cache.start_evictor(interval=0.1)
|
||||
|
||||
await cache.put("lazy", "val", ttl_seconds=0.1)
|
||||
await asyncio.sleep(0.2)
|
||||
# Lazy check: get() should trigger deletion
|
||||
assert await cache.get("lazy") is None, "Lazy eviction failed"
|
||||
|
||||
await cache.put("bg", "val", ttl_seconds=0.1)
|
||||
await asyncio.sleep(0.3)
|
||||
# Background check: key should be gone from internal map without calling get()
|
||||
async with cache._lock:
|
||||
assert "bg" not in cache.cache, "Background eviction failed"
|
||||
await cache.stop_evictor()
|
||||
print("PASSED")
|
||||
|
||||
# 3. Test Atomic Transactions (Commit vs Rollback)
|
||||
print("Testing Transaction Isolation...", end=" ")
|
||||
cache = LFUCache(capacity=10)
|
||||
await cache.put("shared", "original")
|
||||
|
||||
tx = cache.begin_transaction()
|
||||
await tx.put("shared", "modified")
|
||||
await tx.put("new", "added")
|
||||
|
||||
# Verify Isolation: Global reader shouldn't see changes
|
||||
assert await cache.get("shared") == "original"
|
||||
assert await cache.get("new") is None
|
||||
|
||||
# Verify RYOW: Transaction should see its own writes
|
||||
assert await tx.get("shared") == "modified"
|
||||
assert await tx.get("new") == "added"
|
||||
|
||||
await tx.commit()
|
||||
assert await cache.get("shared") == "modified", "Commit failed to update global state"
|
||||
assert await cache.get("new") == "added", "Commit failed to add new key"
|
||||
|
||||
tx2 = cache.begin_transaction()
|
||||
await tx2.put("shared", "rollback_this")
|
||||
assert await tx2.get("shared") == "rollback_this"
|
||||
await tx2.rollback()
|
||||
assert await cache.get("shared") == "modified", "Rollback mutated global state"
|
||||
print("PASSED")
|
||||
|
||||
# 4. Stress Test: Concurrent Access
|
||||
print("Testing High Concurrency (50 tasks)...", end=" ")
|
||||
cache = LFUCache(capacity=100)
|
||||
|
||||
async def worker(worker_id):
|
||||
for i in range(100):
|
||||
key = f"key_{i % 20}" # Create contention on keys
|
||||
if i % 3 == 0:
|
||||
await cache.put(key, f"val_{worker_id}_{i}", ttl_seconds=0.5)
|
||||
elif i % 3 == 1:
|
||||
await cache.get(key)
|
||||
else:
|
||||
tx = cache.begin_transaction()
|
||||
await tx.put(f"tx_{worker_id}", i)
|
||||
await tx.commit()
|
||||
|
||||
tasks = [worker(i) for i in range(50)]
|
||||
await asyncio.gather(*tasks)
|
||||
print("PASSED")
|
||||
|
||||
print("\nAll tests passed successfully!")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
@@ -0,0 +1,550 @@
|
||||
"""
|
||||
In-Memory Concurrent LFU Cache with Async TTL Eviction and Atomic Transactions.
|
||||
|
||||
Pure Python 3.11+ implementation — no third-party dependencies.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import weakref
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Optional
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Doubly-Linked List (O(1) move/remove/add)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class _DLLNode:
|
||||
"""Element inside a frequency bucket's doubly-linked list."""
|
||||
key: Any
|
||||
value: Any
|
||||
ttl_seconds: float
|
||||
expiry_time: float
|
||||
freq: int = 0
|
||||
prev: Optional[_DLLNode] = field(default=None, repr=False)
|
||||
next: Optional[_DLLNode] = field(default=None, repr=False)
|
||||
|
||||
|
||||
class _DoublyLinkedList:
|
||||
"""
|
||||
Doubly-linked list with sentinel head/tail for O(1) prepend, pop, and remove.
|
||||
The *tail* side holds the least-recently-used node within a frequency tier,
|
||||
which is also the LFU candidate for eviction.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._head = _DLLNode(None, None, 0.0, 0.0) # sentinel
|
||||
self._tail = _DLLNode(None, None, 0.0, 0.0) # sentinel
|
||||
self._head.next = self._tail
|
||||
self._tail.prev = self._head
|
||||
self.size = 0
|
||||
|
||||
# -- internal helpers ---------------------------------------------------
|
||||
|
||||
def _insert_after(self, node: _DLLNode, prev: _DLLNode) -> None:
|
||||
nxt = prev.next
|
||||
prev.next = node
|
||||
node.prev = prev
|
||||
node.next = nxt
|
||||
nxt.prev = node
|
||||
|
||||
def _unlink(self, node: _DLLNode) -> None:
|
||||
prev, nxt = node.prev, node.next
|
||||
prev.next = nxt
|
||||
nxt.prev = prev
|
||||
node.prev = node.next = None
|
||||
|
||||
# -- public API ---------------------------------------------------------
|
||||
|
||||
def push_front(self, node: _DLLNode) -> None:
|
||||
"""Insert *node* right after the head sentinel (most-recent)."""
|
||||
self._insert_after(node, self._head)
|
||||
self.size += 1
|
||||
|
||||
def pop_tail(self) -> Optional[_DLLNode]:
|
||||
"""Remove and return the node just before the tail sentinel (LRU)."""
|
||||
if self.size == 0:
|
||||
return None
|
||||
node = self._tail.prev
|
||||
self._unlink(node)
|
||||
return node
|
||||
|
||||
def remove(self, node: _DLLNode) -> None:
|
||||
"""Remove an arbitrary node from the list."""
|
||||
self._unlink(node)
|
||||
|
||||
def is_empty(self) -> bool:
|
||||
return self.size == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Cache Node (wraps the DLL node + TTL metadata)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class _CacheNode:
|
||||
key: Any
|
||||
value: Any
|
||||
ttl_seconds: float
|
||||
expiry_time: float
|
||||
freq: int = 1
|
||||
dll_node: Optional[_DLLNode] = field(default=None, repr=False)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Transaction Handle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class Transaction:
|
||||
"""
|
||||
Represents an isolated sub-session on the cache.
|
||||
|
||||
- ``tx.put(key, value, ttl)`` buffers a write locally.
|
||||
- ``tx.get(key)`` reads from local buffer first, then falls back to the
|
||||
global cache (without mutating global frequency state).
|
||||
- ``tx.delete(key)`` buffers a deletion locally.
|
||||
- ``await tx.commit()`` atomically applies all buffered writes to the global cache.
|
||||
- ``tx.rollback()`` discards everything.
|
||||
"""
|
||||
|
||||
def __init__(self, cache: "LFUCache") -> None:
|
||||
self._cache = cache
|
||||
self._pending_writes: dict[Any, tuple[Any, float]] = {} # key -> (value, expiry)
|
||||
self._pending_deletes: set[Any] = set()
|
||||
self._committed = False
|
||||
|
||||
# -- read ----------------------------------------------------------------
|
||||
|
||||
def get(self, key: Any) -> Optional[Any]:
|
||||
"""Read with read-your-own-writes semantics."""
|
||||
if self._committed:
|
||||
raise RuntimeError("Transaction already committed")
|
||||
|
||||
# 1. Check local pending writes first
|
||||
if key in self._pending_writes:
|
||||
return self._pending_writes[key][0]
|
||||
if key in self._pending_deletes:
|
||||
return None
|
||||
|
||||
# 2. Fall back to global cache (read-only, no frequency bump)
|
||||
return self._cache._get_raw(key)
|
||||
|
||||
# -- write ---------------------------------------------------------------
|
||||
|
||||
def put(self, key: Any, value: Any, ttl_seconds: float = 60.0) -> None:
|
||||
"""Buffer a write locally; not visible to others until commit."""
|
||||
if self._committed:
|
||||
raise RuntimeError("Transaction already committed")
|
||||
expiry = time.monotonic() + ttl_seconds
|
||||
self._pending_writes[key] = (value, expiry)
|
||||
# If previously deleted in this txn, re-add overrides the delete.
|
||||
self._pending_deletes.discard(key)
|
||||
|
||||
def delete(self, key: Any) -> None:
|
||||
"""Buffer a deletion locally."""
|
||||
if self._committed:
|
||||
raise RuntimeError("Transaction already committed")
|
||||
self._pending_deletes.add(key)
|
||||
self._pending_writes.pop(key, None)
|
||||
|
||||
# -- commit / rollback ---------------------------------------------------
|
||||
|
||||
async def commit(self) -> None:
|
||||
"""Atomically apply all buffered changes to the global cache."""
|
||||
if self._committed:
|
||||
raise RuntimeError("Transaction already committed")
|
||||
async with self._cache._lock:
|
||||
for key, (value, expiry) in self._pending_writes.items():
|
||||
await self._cache._put_internal(key, value, expiry)
|
||||
for key in self._pending_deletes:
|
||||
await self._cache._delete_internal(key)
|
||||
self._committed = True
|
||||
|
||||
def rollback(self) -> None:
|
||||
"""Discard all pending changes."""
|
||||
self._pending_writes.clear()
|
||||
self._pending_deletes.clear()
|
||||
self._committed = True # mark so further ops raise
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# LFU Cache
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class LFUCache:
|
||||
"""
|
||||
In-memory concurrent LFU cache with O(1) get/put, async TTL eviction,
|
||||
and atomic transaction support.
|
||||
"""
|
||||
|
||||
def __init__(self, capacity: int = 1024) -> None:
|
||||
if capacity < 1:
|
||||
raise ValueError("capacity must be >= 1")
|
||||
|
||||
self._capacity = capacity
|
||||
self._lock = asyncio.Lock()
|
||||
self._evictor_task: Optional[asyncio.Task[None]] = None
|
||||
|
||||
# Core O(1) structures
|
||||
self._cache_map: dict[Any, _CacheNode] = {} # key -> CacheNode
|
||||
self._freq_map: dict[int, _DoublyLinkedList] = {} # freq -> DLL
|
||||
self._min_freq: int = 1
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
"""O(1) lookup with lazy TTL eviction."""
|
||||
async with self._lock:
|
||||
node = self._cache_map.get(key)
|
||||
if node is None:
|
||||
return None
|
||||
# Lazy TTL check
|
||||
if time.monotonic() > node.expiry_time:
|
||||
await self._evict_node(key, node)
|
||||
return None
|
||||
# Bump frequency — O(1)
|
||||
await self._bump_freq(node)
|
||||
return node.value
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: float = 60.0) -> None:
|
||||
"""O(1) insert/update with lazy TTL eviction of LRU-LFU victim if needed."""
|
||||
expiry = time.monotonic() + ttl_seconds
|
||||
async with self._lock:
|
||||
await self._put_internal(key, value, expiry)
|
||||
|
||||
async def delete(self, key: Any) -> bool:
|
||||
"""O(1) deletion."""
|
||||
async with self._lock:
|
||||
return await self._delete_internal(key)
|
||||
|
||||
def begin_transaction(self) -> Transaction:
|
||||
"""Start a new isolated transaction."""
|
||||
return Transaction(self)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# TTL Eviction Loop
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def start_evictor(self, interval_seconds: float = 1.0) -> None:
|
||||
"""Start the background async TTL sweep task."""
|
||||
if self._evictor_task is not None and not self._evictor_task.done():
|
||||
return
|
||||
self._evictor_task = asyncio.create_task(self._eviction_loop(interval_seconds))
|
||||
|
||||
def stop_evictor(self) -> None:
|
||||
"""Stop the background eviction task."""
|
||||
if self._evictor_task is not None:
|
||||
self._evictor_task.cancel()
|
||||
self._evictor_task = None
|
||||
|
||||
async def _eviction_loop(self, interval: float) -> None:
|
||||
"""Periodically purge expired entries in small batches."""
|
||||
try:
|
||||
while True:
|
||||
await asyncio.sleep(interval)
|
||||
async with self._lock:
|
||||
now = time.monotonic()
|
||||
# Collect expired keys in a snapshot to avoid dict-changed-size
|
||||
expired = [
|
||||
k for k, n in self._cache_map.items()
|
||||
if now > n.expiry_time
|
||||
]
|
||||
for k in expired:
|
||||
node = self._cache_map.get(k)
|
||||
if node is not None and now > node.expiry_time:
|
||||
await self._evict_node(k, node)
|
||||
except asyncio.CancelledError:
|
||||
return
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def _get_raw(self, key: Any) -> Optional[Any]:
|
||||
"""Read-only global lookup — does NOT bump frequency. For transactions."""
|
||||
node = self._cache_map.get(key)
|
||||
if node is None:
|
||||
return None
|
||||
if time.monotonic() > node.expiry_time:
|
||||
await self._evict_node(key, node)
|
||||
return None
|
||||
return node.value
|
||||
|
||||
async def _put_internal(self, key: Any, value: Any, expiry: float) -> None:
|
||||
"""Core put logic (must be called under lock)."""
|
||||
# If key already exists, update in place
|
||||
if key in self._cache_map:
|
||||
node = self._cache_map[key]
|
||||
# Remove from old freq list, update value/ttl/freq
|
||||
self._freq_map[node.freq].remove(node.dll_node)
|
||||
if self._freq_map[node.freq].is_empty():
|
||||
del self._freq_map[node.freq]
|
||||
if node.freq == self._min_freq:
|
||||
self._min_freq += 1
|
||||
node.value = value
|
||||
node.ttl_seconds = expiry - time.monotonic()
|
||||
node.expiry_time = expiry
|
||||
node.freq = 1
|
||||
self._ensure_freq_list(1).push_front(node.dll_node)
|
||||
return
|
||||
|
||||
# Evict if at capacity
|
||||
if len(self._cache_map) >= self._capacity:
|
||||
await self._evict_lfu()
|
||||
|
||||
# Insert new node
|
||||
dll_node = _DLLNode(key, value, expiry - time.monotonic(), expiry)
|
||||
cache_node = _CacheNode(key, value, expiry - time.monotonic(), expiry, freq=1, dll_node=dll_node)
|
||||
self._cache_map[key] = cache_node
|
||||
self._freq_map[1].push_front(dll_node)
|
||||
self._min_freq = 1
|
||||
|
||||
async def _delete_internal(self, key: Any) -> bool:
|
||||
"""Core delete logic (must be called under lock)."""
|
||||
node = self._cache_map.get(key)
|
||||
if node is None:
|
||||
return False
|
||||
await self._evict_node(key, node)
|
||||
return True
|
||||
|
||||
async def _evict_node(self, key: Any, node: _CacheNode) -> None:
|
||||
"""Remove a single node from all structures."""
|
||||
dll = self._freq_map.get(node.freq)
|
||||
if dll is not None:
|
||||
dll.remove(node.dll_node)
|
||||
if dll.is_empty():
|
||||
del self._freq_map[node.freq]
|
||||
if node.freq == self._min_freq:
|
||||
# Find new min freq
|
||||
new_min = min(self._freq_map) if self._freq_map else 1
|
||||
self._min_freq = new_min
|
||||
del self._cache_map[key]
|
||||
|
||||
async def _evict_lfu(self) -> None:
|
||||
"""Evict the least-frequently-used (and least-recently-used within that freq) node."""
|
||||
if not self._freq_map:
|
||||
return
|
||||
dll = self._freq_map.get(self._min_freq)
|
||||
if dll is None or dll.is_empty():
|
||||
# Shouldn't happen, but safeguard
|
||||
self._min_freq += 1
|
||||
await self._evict_lfu()
|
||||
return
|
||||
victim = dll.pop_tail()
|
||||
if victim is not None:
|
||||
await self._evict_node(victim.key, self._cache_map[victim.key])
|
||||
|
||||
async def _bump_freq(self, node: _CacheNode) -> None:
|
||||
"""Move a node from freq f to freq f+1. O(1)."""
|
||||
old_freq = node.freq
|
||||
old_dll = self._freq_map[old_freq]
|
||||
old_dll.remove(node.dll_node)
|
||||
if old_dll.is_empty():
|
||||
del self._freq_map[old_freq]
|
||||
if old_freq == self._min_freq:
|
||||
self._min_freq += 1
|
||||
|
||||
node.freq += 1
|
||||
new_dll = self._ensure_freq_list(node.freq)
|
||||
new_dll.push_front(node.dll_node)
|
||||
|
||||
def _ensure_freq_list(self, freq: int) -> _DoublyLinkedList:
|
||||
"""Return (or create) the DLL for *freq*."""
|
||||
if freq not in self._freq_map:
|
||||
self._freq_map[freq] = _DoublyLinkedList()
|
||||
return self._freq_map[freq]
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Diagnostics
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@property
|
||||
def size(self) -> int:
|
||||
return len(self._cache_map)
|
||||
|
||||
@property
|
||||
def capacity(self) -> int:
|
||||
return self._capacity
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Executable Test Suite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def main() -> None:
|
||||
passed = 0
|
||||
failed = 0
|
||||
|
||||
def _check(name: str, condition: bool) -> None:
|
||||
nonlocal passed, failed
|
||||
if condition:
|
||||
passed += 1
|
||||
print(f" ✓ {name}")
|
||||
else:
|
||||
failed += 1
|
||||
print(f" ✗ {name}")
|
||||
|
||||
# ======================================================================
|
||||
# (a) O(1) LFU eviction order
|
||||
# ======================================================================
|
||||
print("\n=== (a) LFU Eviction Order ===")
|
||||
|
||||
cache = LFUCache(capacity=3)
|
||||
|
||||
# Insert 3 items
|
||||
await cache.put("a", 1, ttl_seconds=60.0)
|
||||
await cache.put("b", 2, ttl_seconds=60.0)
|
||||
await cache.put("c", 3, ttl_seconds=60.0)
|
||||
|
||||
# Access "a" and "b" once each → freq=2; "c" stays at freq=1
|
||||
await cache.get("a")
|
||||
await cache.get("b")
|
||||
|
||||
# Insert "d" — should evict "c" (lowest freq)
|
||||
await cache.put("d", 4, ttl_seconds=60.0)
|
||||
|
||||
_check("a still present after eviction", await cache.get("a") == 1)
|
||||
_check("b still present after eviction", await cache.get("b") == 2)
|
||||
_check("c evicted (lowest freq)", await cache.get("c") is None)
|
||||
_check("d present", await cache.get("d") == 4)
|
||||
|
||||
# Now access "a" again → freq=3; "b" and "d" at freq=2
|
||||
await cache.get("a")
|
||||
# Insert "e" — should evict either "b" or "d" (both freq=2, LRU wins)
|
||||
await cache.put("e", 5, ttl_seconds=60.0)
|
||||
|
||||
_check("a still present", await cache.get("a") == 1)
|
||||
_check("e present", await cache.get("e") == 5)
|
||||
|
||||
# ======================================================================
|
||||
# (b) Lazy TTL vs Background Async Sweep
|
||||
# ======================================================================
|
||||
print("\n=== (b) TTL Eviction (Lazy + Background) ===")
|
||||
|
||||
cache2 = LFUCache(capacity=10)
|
||||
|
||||
await cache2.put("lazy_key", "lazy_val", ttl_seconds=0.1)
|
||||
await cache2.put("bg_key", "bg_val", ttl_seconds=0.1)
|
||||
|
||||
# Lazy eviction: access lazy_key after expiry
|
||||
await asyncio.sleep(0.15)
|
||||
_check("Lazy eviction: get returns None after TTL", await cache2.get("lazy_key") is None)
|
||||
_check("Lazy eviction: bg_key still there (not accessed)", await cache2.get("bg_key") == "bg_val")
|
||||
|
||||
# Start background evictor
|
||||
cache2.start_evictor(interval_seconds=0.2)
|
||||
await asyncio.sleep(0.3) # let background sweep run
|
||||
|
||||
_check("Background eviction: bg_key purged by sweep", await cache2.get("bg_key") is None)
|
||||
cache2.stop_evictor()
|
||||
|
||||
# ======================================================================
|
||||
# (c) Transaction commit visibility vs rollback
|
||||
# ======================================================================
|
||||
print("\n=== (c) Atomic Transactions ===")
|
||||
|
||||
cache3 = LFUCache(capacity=10)
|
||||
await cache3.put("x", 10, ttl_seconds=60.0)
|
||||
await cache3.put("y", 20, ttl_seconds=60.0)
|
||||
|
||||
# --- Commit test ---
|
||||
tx1 = cache3.begin_transaction()
|
||||
tx1.put("x", 99, ttl_seconds=60.0) # local write
|
||||
tx1.put("z", 30, ttl_seconds=60.0) # new key
|
||||
|
||||
_check("TX: read-your-own-write (x)", tx1.get("x") == 99)
|
||||
_check("TX: read-your-own-write (z)", tx1.get("z") == 30)
|
||||
_check("TX: global still sees old x", await cache3.get("x") == 10)
|
||||
|
||||
await tx1.commit()
|
||||
_check("TX: after commit, global sees x=99", await cache3.get("x") == 99)
|
||||
_check("TX: after commit, global sees z=30", await cache3.get("z") == 30)
|
||||
|
||||
# --- Rollback test ---
|
||||
tx2 = cache3.begin_transaction()
|
||||
tx2.put("x", -1, ttl_seconds=60.0)
|
||||
tx2.delete("y")
|
||||
_check("TX rollback: local sees x=-1", tx2.get("x") == -1)
|
||||
_check("TX rollback: local sees y deleted", tx2.get("y") is None)
|
||||
_check("TX rollback: global still sees x=99", await cache3.get("x") == 99)
|
||||
_check("TX rollback: global still sees y=20", await cache3.get("y") == 20)
|
||||
|
||||
tx2.rollback()
|
||||
_check("TX rollback: global unchanged after rollback", await cache3.get("x") == 99)
|
||||
_check("TX rollback: y still present after rollback", await cache3.get("y") == 20)
|
||||
|
||||
# --- Double commit / rollback raises ---
|
||||
tx3 = cache3.begin_transaction()
|
||||
await tx3.commit()
|
||||
try:
|
||||
tx3.put("x", 1)
|
||||
_check("TX: double commit raises", False)
|
||||
except RuntimeError:
|
||||
_check("TX: double commit raises", True)
|
||||
|
||||
tx4 = cache3.begin_transaction()
|
||||
tx4.rollback()
|
||||
try:
|
||||
tx4.put("x", 1)
|
||||
_check("TX: op after rollback raises", False)
|
||||
except RuntimeError:
|
||||
_check("TX: op after rollback raises", True)
|
||||
|
||||
# ======================================================================
|
||||
# (d) Stress test: 50 concurrent async tasks
|
||||
# ======================================================================
|
||||
print("\n=== (d) Stress Test — 50 Concurrent Tasks ===")
|
||||
|
||||
cache4 = LFUCache(capacity=200)
|
||||
errors: list[str] = []
|
||||
|
||||
async def worker(task_id: int, base_key: int) -> None:
|
||||
try:
|
||||
for i in range(50):
|
||||
key = f"t{task_id}_k{i}"
|
||||
val = task_id * 1000 + i
|
||||
await cache4.put(key, val, ttl_seconds=5.0)
|
||||
result = await cache4.get(key)
|
||||
if result != val:
|
||||
errors.append(f"task={task_id} key={key} expected={val} got={result}")
|
||||
# Occasional transaction
|
||||
if i % 10 == 0:
|
||||
tx = cache4.begin_transaction()
|
||||
tx.put(f"tx_{task_id}_{i}", val * 2, ttl_seconds=5.0)
|
||||
r = tx.get(f"tx_{task_id}_{i}")
|
||||
if r != val * 2:
|
||||
errors.append(f"task={task_id} tx key mismatch")
|
||||
await tx.commit()
|
||||
except Exception as e:
|
||||
errors.append(f"task={task_id} exception: {e}")
|
||||
|
||||
tasks = [asyncio.create_task(worker(tid, tid)) for tid in range(50)]
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
_check("Stress: no errors", len(errors) == 0)
|
||||
_check("Stress: cache size within capacity", cache4.size <= 200)
|
||||
if errors:
|
||||
for e in errors[:5]:
|
||||
print(f" error: {e}")
|
||||
|
||||
# ======================================================================
|
||||
# Summary
|
||||
# ======================================================================
|
||||
total = passed + failed
|
||||
print(f"\n{'='*50}")
|
||||
print(f"Results: {passed}/{total} passed, {failed} failed")
|
||||
if failed == 0:
|
||||
print("All tests passed! ✓")
|
||||
else:
|
||||
print(f"{failed} test(s) FAILED ✗")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
@@ -0,0 +1,365 @@
|
||||
"""
|
||||
In-Memory Concurrent LFU Cache with Async TTL Eviction and Atomic Transactions.
|
||||
Pure Python 3.11+ implementation using only standard library modules.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
from asyncio import Lock
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Internal Data Structures (O(1) LFU Core)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _Node:
|
||||
"""Doubly linked list node tracking key, value, frequency, and TTL."""
|
||||
__slots__ = ('key', 'value', 'freq', 'expires_at', 'prev', 'next')
|
||||
def __init__(self, key: Any, value: Any, freq: int = 1, expires_at: float = 0.0):
|
||||
self.key = key
|
||||
self.value = value
|
||||
self.freq = freq
|
||||
self.expires_at = expires_at
|
||||
self.prev = None
|
||||
self.next = None
|
||||
|
||||
class _FreqList:
|
||||
"""Doubly linked list maintaining nodes of a specific frequency tier."""
|
||||
__slots__ = ('head', 'tail', 'size')
|
||||
def __init__(self):
|
||||
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.size = 0
|
||||
|
||||
def add(self, node: _Node):
|
||||
"""Add node to tail (most recently used in this frequency)."""
|
||||
last = self.tail.prev
|
||||
last.next = node
|
||||
node.prev = last
|
||||
node.next = self.tail
|
||||
self.tail.prev = node
|
||||
self.size += 1
|
||||
|
||||
def remove(self, node: _Node):
|
||||
"""Remove node from the list in O(1)."""
|
||||
node.prev.next = node.next
|
||||
node.next.prev = node.prev
|
||||
node.prev = None
|
||||
node.next = None
|
||||
self.size -= 1
|
||||
|
||||
def pop(self) -> _Node:
|
||||
"""Remove and return node from head (least recently used)."""
|
||||
node = self.head.next
|
||||
self.remove(node)
|
||||
return node
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main Cache Implementation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class LFUCache:
|
||||
"""
|
||||
O(1) In-Memory Concurrent LFU Cache with Async TTL Eviction.
|
||||
Uses a hash map for key lookup and a hash map of frequency-buckets
|
||||
with doubly linked lists for O(1) frequency updates and eviction.
|
||||
"""
|
||||
def __init__(self, capacity: int, evict_interval: float = 1.0):
|
||||
self.capacity = capacity
|
||||
self.evict_interval = evict_interval
|
||||
self.key_to_node: Dict[Any, _Node] = {}
|
||||
self.freq_to_list: Dict[int, _FreqList] = {}
|
||||
self.min_freq: int = 1
|
||||
self.lock: Lock = asyncio.Lock()
|
||||
self._evict_task: Optional[asyncio.Task] = None
|
||||
self._running = False
|
||||
|
||||
async def start_evictor(self):
|
||||
"""Start the non-blocking background TTL eviction loop."""
|
||||
self._running = True
|
||||
self._evict_task = asyncio.create_task(self._eviction_loop())
|
||||
|
||||
async def stop_evictor(self):
|
||||
"""Gracefully stop the background eviction task."""
|
||||
self._running = False
|
||||
if self._evict_task:
|
||||
self._evict_task.cancel()
|
||||
try: await self._evict_task
|
||||
except asyncio.CancelledError: pass
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
"""Retrieve value in O(1). Performs lazy TTL eviction and frequency update."""
|
||||
async with self.lock:
|
||||
node = self.key_to_node.get(key)
|
||||
if not node:
|
||||
return None
|
||||
|
||||
# Lazy TTL Eviction
|
||||
if node.expires_at > 0 and node.expires_at <= time.time():
|
||||
self._remove_node(node)
|
||||
return None
|
||||
|
||||
# O(1) Frequency Update
|
||||
self._update_freq(node)
|
||||
return node.value
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: float = 0.0):
|
||||
"""Insert/update value in O(1). Handles capacity eviction and TTL."""
|
||||
async with self.lock:
|
||||
node = self.key_to_node.get(key)
|
||||
if node:
|
||||
# Update existing node
|
||||
node.value = value
|
||||
if ttl_seconds > 0:
|
||||
node.expires_at = time.time() + ttl_seconds
|
||||
self._update_freq(node)
|
||||
else:
|
||||
# Evict if capacity reached
|
||||
if len(self.key_to_node) >= self.capacity:
|
||||
self._evict()
|
||||
# Insert new node with freq=1
|
||||
node = _Node(
|
||||
key, value, freq=1,
|
||||
expires_at=float('inf') if ttl_seconds <= 0 else time.time() + ttl_seconds
|
||||
)
|
||||
self.key_to_node[key] = node
|
||||
self._add_to_freq(node)
|
||||
self.min_freq = 1 # New keys always start at freq 1
|
||||
|
||||
async def begin_transaction(self) -> 'Transaction':
|
||||
"""Create an isolated transaction session."""
|
||||
return Transaction(self)
|
||||
|
||||
# --- Internal Cache Helpers (O(1)) ---
|
||||
|
||||
def _remove_node(self, node: _Node):
|
||||
"""Remove node from global structures."""
|
||||
del self.key_to_node[node.key]
|
||||
self.freq_to_list[node.freq].remove(node)
|
||||
if self.freq_to_list[node.freq].size == 0:
|
||||
del self.freq_to_list[node.freq]
|
||||
if self.min_freq == node.freq:
|
||||
self.min_freq += 1
|
||||
|
||||
def _update_freq(self, node: _Node):
|
||||
"""Move node to next frequency bucket in O(1)."""
|
||||
old_freq = node.freq
|
||||
self.freq_to_list[old_freq].remove(node)
|
||||
if self.freq_to_list[old_freq].size == 0:
|
||||
del self.freq_to_list[old_freq]
|
||||
if self.min_freq == old_freq:
|
||||
self.min_freq += 1
|
||||
|
||||
node.freq += 1
|
||||
self._add_to_freq(node)
|
||||
|
||||
def _add_to_freq(self, node: _Node):
|
||||
"""Add node to its frequency bucket."""
|
||||
if node.freq not in self.freq_to_list:
|
||||
self.freq_to_list[node.freq] = _FreqList()
|
||||
self.freq_to_list[node.freq].add(node)
|
||||
|
||||
def _evict(self):
|
||||
"""Evict least frequently used (then LRU) node in O(1)."""
|
||||
if not self.freq_to_list:
|
||||
return
|
||||
evict_list = self.freq_to_list[self.min_freq]
|
||||
node = evict_list.pop()
|
||||
self._remove_node(node)
|
||||
|
||||
async def _eviction_loop(self):
|
||||
"""Background async task that purges expired keys in small batches."""
|
||||
while self._running:
|
||||
await asyncio.sleep(self.evict_interval)
|
||||
async with self.lock:
|
||||
# Batch scan limited to avoid blocking reads
|
||||
keys_to_check = list(self.key_to_node.keys())[:50]
|
||||
for k in keys_to_check:
|
||||
node = self.key_to_node.get(k)
|
||||
if node and node.expires_at > 0 and node.expires_at <= time.time():
|
||||
self._remove_node(node)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Transaction Isolation Layer
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class Transaction:
|
||||
"""
|
||||
Atomic transaction session providing Read-Your-Own-Writes isolation.
|
||||
Global readers remain blind to uncommitted changes until commit().
|
||||
"""
|
||||
def __init__(self, cache: LFUCache):
|
||||
self.cache = cache
|
||||
self.buffer: List[Dict] = []
|
||||
self.committed = False
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
"""Read with local isolation. Does not update global frequencies."""
|
||||
# Read-Your-Own-Writes: check buffer first
|
||||
for op in reversed(self.buffer):
|
||||
if op['key'] == key:
|
||||
return None if op['op'] == 'delete' else op['value']
|
||||
|
||||
# Global read (no freq update to preserve isolation)
|
||||
async with self.cache.lock:
|
||||
node = self.cache.key_to_node.get(key)
|
||||
if not node:
|
||||
return None
|
||||
# Lazy TTL check without mutation
|
||||
if node.expires_at > 0 and node.expires_at <= time.time():
|
||||
return None
|
||||
return node.value
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: float = 0.0):
|
||||
"""Buffer a put operation without mutating global state."""
|
||||
original = self.cache.key_to_node.get(key)
|
||||
self.buffer.append({'op': 'put', 'key': key, 'value': value, 'ttl': ttl_seconds, 'original': original})
|
||||
|
||||
async def delete(self, key: Any):
|
||||
"""Buffer a delete operation without mutating global state."""
|
||||
original = self.cache.key_to_node.get(key)
|
||||
self.buffer.append({'op': 'delete', 'key': key, 'original': original})
|
||||
|
||||
async def commit(self):
|
||||
"""Apply buffered changes to global cache atomically."""
|
||||
async with self.cache.lock:
|
||||
# Merge buffer: keep only the latest operation per key
|
||||
merged = {}
|
||||
for op in self.buffer:
|
||||
merged[op['key']] = op
|
||||
|
||||
for op in merged.values():
|
||||
if op['op'] == 'put':
|
||||
await self._apply_put(op['key'], op['value'], op['ttl'], self.cache.key_to_node.get(op['key']))
|
||||
elif op['op'] == 'delete':
|
||||
await self._apply_delete(op['key'], self.cache.key_to_node.get(op['key']))
|
||||
|
||||
self.buffer.clear()
|
||||
self.committed = True
|
||||
|
||||
async def rollback(self):
|
||||
"""Discard all pending changes without affecting global frequencies or TTL."""
|
||||
self.buffer.clear()
|
||||
self.committed = False
|
||||
|
||||
async def _apply_put(self, key: Any, value: Any, ttl: float, original_node: Optional[_Node]):
|
||||
"""Apply buffered put to global cache."""
|
||||
if original_node:
|
||||
original_node.value = value
|
||||
if ttl > 0:
|
||||
original_node.expires_at = time.time() + ttl
|
||||
self.cache._update_freq(original_node)
|
||||
else:
|
||||
if len(self.cache.key_to_node) >= self.cache.capacity:
|
||||
self.cache._evict()
|
||||
node = _Node(
|
||||
key, value, freq=1,
|
||||
expires_at=float('inf') if ttl <= 0 else time.time() + ttl
|
||||
)
|
||||
self.cache.key_to_node[key] = node
|
||||
self.cache._add_to_freq(node)
|
||||
self.cache.min_freq = 1
|
||||
|
||||
async def _apply_delete(self, key: Any, original_node: Optional[_Node]):
|
||||
"""Apply buffered delete to global cache."""
|
||||
if original_node:
|
||||
self.cache._remove_node(original_node)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Executable Unit Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def main():
|
||||
print("🧪 Running LFU Cache Test Suite...\n")
|
||||
await test_lfu_eviction()
|
||||
await test_ttl_eviction()
|
||||
await test_transactions()
|
||||
await test_stress()
|
||||
print("✅ All tests passed successfully!")
|
||||
|
||||
async def test_lfu_eviction():
|
||||
"""Verifies strict O(1) LFU eviction order when capacity is reached."""
|
||||
cache = LFUCache(3)
|
||||
await cache.put(1, 1)
|
||||
await cache.put(2, 2)
|
||||
await cache.put(3, 3)
|
||||
|
||||
# Access 1 -> freq becomes 2
|
||||
assert await cache.get(1) == 1
|
||||
|
||||
# Insert 4 -> evicts key 2 (freq=1, LRU among freq=1)
|
||||
await cache.put(4, 4)
|
||||
|
||||
assert await cache.get(2) is None, "Key 2 should be evicted"
|
||||
assert await cache.get(1) == 1
|
||||
assert await cache.get(3) == 3
|
||||
assert await cache.get(4) == 4
|
||||
print(" ✓ O(1) LFU eviction order verified")
|
||||
|
||||
async def test_ttl_eviction():
|
||||
"""Verifies Lazy vs Background Async Sweep eviction."""
|
||||
cache = LFUCache(10, evict_interval=0.1)
|
||||
await cache.start_evictor()
|
||||
|
||||
# Lazy eviction on access
|
||||
await cache.put('a', 1, ttl_seconds=0.2)
|
||||
assert await cache.get('a') == 1
|
||||
await asyncio.sleep(0.3)
|
||||
assert await cache.get('a') is None, "Lazy eviction failed"
|
||||
|
||||
# Background async sweep
|
||||
await cache.put('b', 2, ttl_seconds=0.1)
|
||||
await asyncio.sleep(0.2) # Wait for background loop
|
||||
assert await cache.get('b') is None, "Background sweep failed"
|
||||
|
||||
await cache.stop_evictor()
|
||||
print(" ✓ Dual-layer TTL eviction (Lazy + Background) verified")
|
||||
|
||||
async def test_transactions():
|
||||
"""Verifies Transaction commit visibility vs rollback state restoration."""
|
||||
cache = LFUCache(5)
|
||||
await cache.put('x', 10)
|
||||
|
||||
# Test 1: Isolation & Read-Your-Own-Writes
|
||||
tx = await cache.begin_transaction()
|
||||
await tx.put('x', 20)
|
||||
assert await tx.get('x') == 20, "Transaction failed Read-Your-Own-Writes"
|
||||
assert await cache.get('x') == 10, "Global cache leaked uncommitted write"
|
||||
|
||||
await tx.commit()
|
||||
assert await cache.get('x') == 20, "Commit failed to apply changes"
|
||||
|
||||
# Test 2: Rollback restores state
|
||||
tx2 = await cache.begin_transaction()
|
||||
await tx2.put('y', 30)
|
||||
await tx2.rollback()
|
||||
assert await cache.get('y') is None, "Rollback failed to discard changes"
|
||||
print(" ✓ Transaction commit/rollback isolation verified")
|
||||
|
||||
async def test_stress():
|
||||
"""Stress test with 50 concurrent async tasks reading/writing simultaneously."""
|
||||
cache = LFUCache(100)
|
||||
await cache.start_evictor()
|
||||
|
||||
tasks = []
|
||||
for i in range(50):
|
||||
async def worker(idx):
|
||||
tx = await cache.begin_transaction()
|
||||
await tx.put(f'key_{idx}', idx)
|
||||
await asyncio.sleep(0.01)
|
||||
val = await tx.get(f'key_{idx}')
|
||||
assert val == idx, f"Transaction read mismatch for key_{idx}"
|
||||
await tx.commit()
|
||||
# Verify global visibility post-commit
|
||||
assert await cache.get(f'key_{idx}') == idx, f"Global read mismatch for key_{idx}"
|
||||
tasks.append(asyncio.create_task(worker(i)))
|
||||
|
||||
await asyncio.gather(*tasks)
|
||||
await cache.stop_evictor()
|
||||
print(" ✓ Stress test (50 concurrent tasks) passed without race conditions")
|
||||
|
||||
if __name__ == '__main__':
|
||||
asyncio.run(main())
|
||||
|
||||
@@ -0,0 +1,426 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
In-Memory Concurrent LFU Cache with Async TTL Eviction and Atomic Transactions.
|
||||
Pure Python 3.11+ implementation. No external dependencies.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Optional, Dict, Set, List
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Core Data Structures for O(1) Operations
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class Node:
|
||||
key: Any
|
||||
value: Any
|
||||
freq: int
|
||||
expires_at: Optional[float]
|
||||
prev: Optional['Node'] = None
|
||||
next: Optional['Node'] = None
|
||||
|
||||
|
||||
class DoublyLinkedList:
|
||||
"""Sentinel-based doubly linked list supporting O(1) add/remove/pop."""
|
||||
def __init__(self):
|
||||
self.head = Node(None, None, 0, None)
|
||||
self.tail = Node(None, None, 0, None)
|
||||
self.head.next = self.tail
|
||||
self.tail.prev = self.head
|
||||
self.size = 0
|
||||
|
||||
def add_to_tail(self, node: Node):
|
||||
node.prev = self.tail.prev
|
||||
node.next = self.tail
|
||||
self.tail.prev.next = node
|
||||
self.tail.prev = node
|
||||
self.size += 1
|
||||
|
||||
def remove_node(self, node: Node):
|
||||
if node.prev is None or node.next is None:
|
||||
return
|
||||
node.prev.next = node.next
|
||||
node.next.prev = node.prev
|
||||
node.prev = node.next = None
|
||||
self.size -= 1
|
||||
|
||||
def pop_head(self) -> Optional[Node]:
|
||||
if self.size == 0:
|
||||
return None
|
||||
node = self.head.next
|
||||
self.remove_node(node)
|
||||
return node
|
||||
|
||||
def is_empty(self) -> bool:
|
||||
return self.size == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Transaction Isolation Layer
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class Transaction:
|
||||
"""
|
||||
Provides ACID-like sub-sessions with Read-Your-Own-Writes isolation.
|
||||
Changes remain local until commit(), where they are applied atomically.
|
||||
"""
|
||||
def __init__(self, cache: 'LFUCache'):
|
||||
self._cache = cache
|
||||
self._local_writes: Dict[Any, tuple] = {} # key -> (value, ttl)
|
||||
self._local_deletes: Set[Any] = set()
|
||||
self._local_freq_bumps: Set[Any] = set() # Track accesses for LFU bump
|
||||
self._committed = False
|
||||
self._rolled_back = False
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: Optional[float] = None):
|
||||
if self._committed or self._rolled_back:
|
||||
raise RuntimeError("Transaction already finished")
|
||||
self._local_writes[key] = (value, ttl_seconds)
|
||||
self._local_deletes.discard(key)
|
||||
self._local_freq_bumps.add(key)
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
if self._committed or self._rolled_back:
|
||||
raise RuntimeError("Transaction already finished")
|
||||
# Read Your Own Writes
|
||||
if key in self._local_writes:
|
||||
val, _ = self._local_writes[key]
|
||||
self._local_freq_bumps.add(key)
|
||||
return val
|
||||
if key in self._local_deletes:
|
||||
return None
|
||||
# Read Global State
|
||||
val = await self._cache.get(key)
|
||||
if val is not None:
|
||||
self._local_freq_bumps.add(key)
|
||||
return val
|
||||
|
||||
async def delete(self, key: Any) -> bool:
|
||||
if self._committed or self._rolled_back:
|
||||
raise RuntimeError("Transaction already finished")
|
||||
self._local_deletes.add(key)
|
||||
self._local_writes.pop(key, None)
|
||||
return True
|
||||
|
||||
async def commit(self) -> bool:
|
||||
if self._committed or self._rolled_back:
|
||||
raise RuntimeError("Transaction already finished")
|
||||
self._committed = True
|
||||
|
||||
async with self._cache._lock:
|
||||
# 1. Apply Writes
|
||||
for key, (value, ttl) in self._local_writes.items():
|
||||
if key in self._cache.store:
|
||||
node = self._cache.store[key]
|
||||
if self._cache._is_expired(node):
|
||||
self._cache._remove_node(key)
|
||||
else:
|
||||
node.value = value
|
||||
node.expires_at = time.time() + ttl if ttl else None
|
||||
self._cache._update_freq(node)
|
||||
else:
|
||||
self._cache._add_node(key, value, ttl)
|
||||
|
||||
# 2. Apply Deletes
|
||||
for key in self._local_deletes:
|
||||
self._cache._remove_node(key)
|
||||
|
||||
# 3. Apply Frequency Bumps (for reads during transaction)
|
||||
for key in self._local_freq_bumps:
|
||||
if key in self._cache.store and key not in self._local_deletes:
|
||||
node = self._cache.store[key]
|
||||
self._cache._update_freq(node)
|
||||
|
||||
self._cache._cleanup_freq_lists()
|
||||
return True
|
||||
|
||||
async def rollback(self):
|
||||
if self._committed or self._rolled_back:
|
||||
raise RuntimeError("Transaction already finished")
|
||||
self._rolled_back = True
|
||||
# Completely discard pending changes without mutating global state
|
||||
self._local_writes.clear()
|
||||
self._local_deletes.clear()
|
||||
self._local_freq_bumps.clear()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main Cache Implementation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class LFUCache:
|
||||
"""
|
||||
Concurrent LFU Cache with O(1) get/put, dual-layer TTL eviction,
|
||||
and atomic transaction support.
|
||||
"""
|
||||
def __init__(self, capacity: int, eviction_interval: float = 0.5):
|
||||
if capacity <= 0:
|
||||
raise ValueError("Cache capacity must be positive")
|
||||
self.capacity = capacity
|
||||
self.store: Dict[Any, Node] = {}
|
||||
self.freq_map: Dict[int, DoublyLinkedList] = {}
|
||||
self.min_freq: int = 0
|
||||
self.eviction_interval = eviction_interval
|
||||
self._lock = asyncio.Lock()
|
||||
self._evictor_task: Optional[asyncio.Task] = None
|
||||
self._evictor_running = False
|
||||
self._ttl_keys: Set[Any] = set()
|
||||
|
||||
def begin_transaction(self) -> Transaction:
|
||||
return Transaction(self)
|
||||
|
||||
async def start_evictor(self):
|
||||
"""Starts the background async TTL eviction loop."""
|
||||
if self._evictor_running:
|
||||
return
|
||||
self._evictor_running = True
|
||||
self._evictor_task = asyncio.create_task(self._eviction_loop())
|
||||
|
||||
async def stop_evictor(self):
|
||||
"""Stops the background async TTL eviction loop."""
|
||||
self._evictor_running = False
|
||||
if self._evictor_task:
|
||||
self._evictor_task.cancel()
|
||||
try:
|
||||
await self._evictor_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._evictor_task = None
|
||||
|
||||
async def _eviction_loop(self):
|
||||
while self._evictor_running:
|
||||
await asyncio.sleep(self.eviction_interval)
|
||||
if not self._evictor_running:
|
||||
break
|
||||
await self._evict_batch()
|
||||
|
||||
async def _evict_batch(self):
|
||||
"""Background sweep: acquires lock briefly, removes expired keys in batches."""
|
||||
async with self._lock:
|
||||
expired_keys = [
|
||||
k for k in list(self._ttl_keys)
|
||||
if k in self.store and self.store[k].expires_at and time.time() > self.store[k].expires_at
|
||||
]
|
||||
batch = expired_keys[:100] # Small batch to avoid lock contention
|
||||
for k in batch:
|
||||
self._remove_node(k)
|
||||
self._ttl_keys.discard(k)
|
||||
self._cleanup_freq_lists()
|
||||
|
||||
def _cleanup_freq_lists(self):
|
||||
"""Remove empty frequency buckets and update min_freq."""
|
||||
empty_freqs = [f for f, dll in self.freq_map.items() if dll.is_empty()]
|
||||
for f in empty_freqs:
|
||||
del self.freq_map[f]
|
||||
if self.freq_map:
|
||||
self.min_freq = min(self.freq_map.keys())
|
||||
else:
|
||||
self.min_freq = 0
|
||||
|
||||
def _is_expired(self, node: Node) -> bool:
|
||||
return node.expires_at is not None and time.time() > node.expires_at
|
||||
|
||||
def _remove_node(self, key: Any):
|
||||
"""Removes a key from store and its frequency linked list."""
|
||||
if key not in self.store:
|
||||
return
|
||||
node = self.store.pop(key)
|
||||
self._ttl_keys.discard(key)
|
||||
freq = node.freq
|
||||
if freq in self.freq_map:
|
||||
self.freq_map[freq].remove_node(node)
|
||||
|
||||
def _update_freq(self, node: Node):
|
||||
"""Moves node to next frequency bucket in O(1)."""
|
||||
freq = node.freq
|
||||
self.freq_map[freq].remove_node(node)
|
||||
if self.freq_map[freq].is_empty():
|
||||
del self.freq_map[freq]
|
||||
if self.min_freq == freq:
|
||||
self.min_freq += 1
|
||||
node.freq += 1
|
||||
new_freq = node.freq
|
||||
if new_freq not in self.freq_map:
|
||||
self.freq_map[new_freq] = DoublyLinkedList()
|
||||
self.freq_map[new_freq].add_to_tail(node)
|
||||
|
||||
async def get(self, key: Any) -> Optional[Any]:
|
||||
"""O(1) get with lazy TTL eviction and frequency bump."""
|
||||
async with self._lock:
|
||||
if key not in self.store:
|
||||
return None
|
||||
node = self.store[key]
|
||||
if self._is_expired(node):
|
||||
self._remove_node(key)
|
||||
return None
|
||||
self._update_freq(node)
|
||||
return node.value
|
||||
|
||||
async def put(self, key: Any, value: Any, ttl_seconds: Optional[float] = None):
|
||||
"""O(1) put with capacity eviction and TTL handling."""
|
||||
async with self._lock:
|
||||
if key in self.store:
|
||||
node = self.store[key]
|
||||
if self._is_expired(node):
|
||||
self._remove_node(key)
|
||||
else:
|
||||
node.value = value
|
||||
node.expires_at = time.time() + ttl_seconds if ttl_seconds else None
|
||||
self._update_freq(node)
|
||||
return
|
||||
if len(self.store) >= self.capacity:
|
||||
self._evict()
|
||||
self._add_node(key, value, ttl_seconds)
|
||||
|
||||
def _evict(self):
|
||||
"""Evicts the least frequently used item (O(1) via min_freq bucket)."""
|
||||
if self.min_freq in self.freq_map:
|
||||
dll = self.freq_map[self.min_freq]
|
||||
node = dll.pop_head()
|
||||
if node:
|
||||
self.store.pop(node.key, None)
|
||||
self._ttl_keys.discard(node.key)
|
||||
self._cleanup_freq_lists()
|
||||
|
||||
def _add_node(self, key: Any, value: Any, ttl_seconds: Optional[float]):
|
||||
"""Inserts new node into freq 1 bucket."""
|
||||
node = Node(key, value, 1, time.time() + ttl_seconds if ttl_seconds else None)
|
||||
self.store[key] = node
|
||||
if ttl_seconds:
|
||||
self._ttl_keys.add(key)
|
||||
if 1 not in self.freq_map:
|
||||
self.freq_map[1] = DoublyLinkedList()
|
||||
self.min_freq = 1
|
||||
self.freq_map[1].add_to_tail(node)
|
||||
|
||||
async def delete(self, key: Any) -> bool:
|
||||
async with self._lock:
|
||||
if key in self.store:
|
||||
self._remove_node(key)
|
||||
return True
|
||||
return False
|
||||
|
||||
async def size(self) -> int:
|
||||
async with self._lock:
|
||||
return len(self.store)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Executable Unit Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def test_lfu_eviction_order():
|
||||
"""a) O(1) LFU eviction order when capacity is reached."""
|
||||
cache = LFUCache(3)
|
||||
await cache.put(1, 'a')
|
||||
await cache.put(2, 'b')
|
||||
await cache.put(3, 'c')
|
||||
|
||||
# Access 1 twice to make it most frequent
|
||||
await cache.get(1)
|
||||
await cache.get(1)
|
||||
|
||||
# Insert 4 -> should evict 2 (freq 1, least used)
|
||||
await cache.put(4, 'd')
|
||||
|
||||
assert await cache.get(2) is None, "Failed: Least frequent key should be evicted"
|
||||
assert await cache.get(1) == 'a', "Failed: Key 1 should still exist"
|
||||
assert await cache.get(3) == 'c', "Failed: Key 3 should still exist"
|
||||
assert await cache.get(4) == 'd', "Failed: Key 4 should exist"
|
||||
print("✅ Test a) LFU eviction order: PASSED")
|
||||
|
||||
|
||||
async def test_ttl_eviction():
|
||||
"""b) Lazy TTL vs. Background Async Sweep eviction."""
|
||||
cache = LFUCache(10)
|
||||
await cache.start_evictor()
|
||||
|
||||
# Lazy eviction
|
||||
await cache.put('lazy', 'val', ttl_seconds=0.1)
|
||||
assert await cache.get('lazy') == 'val', "Failed: Should return value before TTL"
|
||||
await asyncio.sleep(0.15)
|
||||
assert await cache.get('lazy') is None, "Failed: Lazy eviction should trigger on next get"
|
||||
|
||||
# Background sweep eviction
|
||||
await cache.put('bg', 'val2', ttl_seconds=0.1)
|
||||
await asyncio.sleep(0.25) # Wait for background task cycle
|
||||
assert await cache.get('bg') is None, "Failed: Background sweep should evict expired key"
|
||||
|
||||
await cache.stop_evictor()
|
||||
print("✅ Test b) TTL eviction (Lazy & Background): PASSED")
|
||||
|
||||
|
||||
async def test_transaction_isolation():
|
||||
"""c) Transaction commit visibility vs. rollback state restoration."""
|
||||
cache = LFUCache(10)
|
||||
await cache.put('a', 1)
|
||||
|
||||
# Commit test
|
||||
tx = cache.begin_transaction()
|
||||
await tx.put('a', 2)
|
||||
await tx.put('b', 3)
|
||||
|
||||
# Read your own writes
|
||||
assert await tx.get('a') == 2, "Failed: Should see local write"
|
||||
assert await tx.get('b') == 3, "Failed: Should see local write"
|
||||
|
||||
# Global shouldn't see uncommitted changes
|
||||
assert await cache.get('a') == 1, "Failed: Global should not see uncommitted write"
|
||||
|
||||
await tx.commit()
|
||||
assert await cache.get('a') == 2, "Failed: Global should see committed write"
|
||||
assert await cache.get('b') == 3, "Failed: Global should see committed write"
|
||||
|
||||
# Rollback test
|
||||
tx2 = cache.begin_transaction()
|
||||
await tx2.put('c', 4)
|
||||
assert await tx2.get('c') == 4, "Failed: Should see local write"
|
||||
await tx2.rollback()
|
||||
assert await cache.get('c') is None, "Failed: Rollback should discard changes"
|
||||
print("✅ Test c) Transaction isolation (Commit & Rollback): PASSED")
|
||||
|
||||
|
||||
async def test_stress_concurrency():
|
||||
"""d) Stress test with 50 concurrent async tasks reading/writing simultaneously."""
|
||||
cache = LFUCache(100)
|
||||
await cache.start_evictor()
|
||||
|
||||
tasks = []
|
||||
for i in range(50):
|
||||
async def worker(idx: int):
|
||||
for j in range(10):
|
||||
key = f"k{idx}"
|
||||
await cache.put(key, f"v{idx}_{j}")
|
||||
val = await cache.get(key)
|
||||
assert val == f"v{idx}_{j}", f"Failed: Read mismatch for {key}"
|
||||
tasks.append(asyncio.create_task(worker(i)))
|
||||
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
# All 50 keys should exist
|
||||
assert await cache.size() == 50, f"Failed: Expected 50 keys, got {await cache.size()}"
|
||||
|
||||
# Verify concurrent reads/writes didn't corrupt data
|
||||
for i in range(50):
|
||||
assert await cache.get(f"k{i}") == f"v{i}_9", f"Failed: Data corruption for k{i}"
|
||||
|
||||
await cache.stop_evictor()
|
||||
print("✅ Test d) Stress concurrency (50 tasks): PASSED")
|
||||
|
||||
|
||||
async def main():
|
||||
print("🚀 Running LFU Cache Unit Tests...\n")
|
||||
await test_lfu_eviction_order()
|
||||
await test_ttl_eviction()
|
||||
await test_transaction_isolation()
|
||||
await test_stress_concurrency()
|
||||
print("\n🎉 All tests passed successfully.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
@@ -0,0 +1,398 @@
|
||||
"""
|
||||
In-Memory Concurrent LFU Cache with Async TTL Eviction & Atomic Transactions.
|
||||
Pure Python 3.11+ implementation using asyncio, dataclasses, and collections.OrderedDict.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, Optional, Set, Tuple
|
||||
from collections import OrderedDict
|
||||
|
||||
|
||||
@dataclass
|
||||
class _Node:
|
||||
"""Internal node storing cache value and TTL expiry timestamp."""
|
||||
key: str
|
||||
value: Any
|
||||
ttl_expiry: float
|
||||
|
||||
|
||||
class LFUCache:
|
||||
"""
|
||||
O(1) Least Frequently Used Cache with dual-layer TTL eviction and async concurrency.
|
||||
|
||||
Data Structures:
|
||||
- self.cache: Dict[str, _Node] -> Direct O(1) key lookup
|
||||
- self.freq_map: Dict[int, OrderedDict[str, None]] -> Frequency buckets maintaining insertion order (LRU within same freq)
|
||||
- self.key_freq: Dict[str, int] -> Tracks current frequency of each key for O(1) updates
|
||||
- self.ttl_map: Dict[str, float] -> Stores absolute TTL expiry timestamps
|
||||
"""
|
||||
|
||||
def __init__(self, capacity: int):
|
||||
self.capacity = max(0, capacity)
|
||||
self.cache: Dict[str, _Node] = {}
|
||||
self.freq_map: Dict[int, OrderedDict[str, None]] = {}
|
||||
self.key_freq: Dict[str, int] = {}
|
||||
self.ttl_map: Dict[str, float] = {}
|
||||
self.min_freq: int = 0
|
||||
self._lock = asyncio.Lock()
|
||||
self._evictor_task: Optional[asyncio.Task] = None
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# PUBLIC API #
|
||||
# ------------------------------------------------------------------ #
|
||||
|
||||
async def start_evictor(self) -> None:
|
||||
"""Start the non-blocking background TTL eviction loop."""
|
||||
if self._evictor_task is not None:
|
||||
return
|
||||
self._evictor_task = asyncio.create_task(self._background_loop())
|
||||
|
||||
async def stop_evictor(self) -> None:
|
||||
"""Gracefully stop the background eviction task."""
|
||||
if self._evictor_task:
|
||||
self._evictor_task.cancel()
|
||||
try:
|
||||
await self._evictor_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._evictor_task = None
|
||||
|
||||
async def get(self, key: str) -> Optional[Any]:
|
||||
"""
|
||||
Retrieve value by key. O(1) average time complexity.
|
||||
Performs lazy TTL eviction upon access.
|
||||
"""
|
||||
current_time = time.monotonic()
|
||||
expiry = self.ttl_map.get(key)
|
||||
|
||||
# Lazy TTL Eviction
|
||||
if expiry is not None and current_time >= expiry:
|
||||
await self._remove_key(key)
|
||||
return None
|
||||
|
||||
async with self._lock:
|
||||
if key not in self.cache:
|
||||
return None
|
||||
|
||||
node = self.cache[key]
|
||||
|
||||
# Frequency Increment & Bucket Migration (O(1))
|
||||
old_freq = self.key_freq[key]
|
||||
new_freq = old_freq + 1
|
||||
self.key_freq[key] = new_freq
|
||||
|
||||
old_bucket = self.freq_map[old_freq]
|
||||
del old_bucket[key]
|
||||
if not old_bucket:
|
||||
del self.freq_map[old_freq]
|
||||
if self.min_freq == old_freq:
|
||||
self.min_freq = min(self.freq_map.keys()) if self.freq_map else 0
|
||||
|
||||
new_bucket = self.freq_map.setdefault(new_freq, OrderedDict())
|
||||
new_bucket[key] = None # Store key reference in bucket
|
||||
|
||||
return node.value
|
||||
|
||||
async def put(self, key: str, value: Any, ttl_seconds: float) -> None:
|
||||
"""
|
||||
Insert or update key-value pair with TTL. O(1) average time complexity.
|
||||
Performs lazy TTL eviction before insertion if needed.
|
||||
"""
|
||||
current_time = time.monotonic()
|
||||
expiry = self.ttl_map.get(key)
|
||||
|
||||
# Lazy TTL Eviction for stale keys
|
||||
if expiry is not None and current_time >= expiry:
|
||||
await self._remove_key(key)
|
||||
|
||||
async with self._lock:
|
||||
if key in self.cache:
|
||||
# Update existing: increment frequency & migrate bucket
|
||||
old_freq = self.key_freq[key]
|
||||
new_freq = old_freq + 1
|
||||
self.key_freq[key] = new_freq
|
||||
|
||||
old_bucket = self.freq_map[old_freq]
|
||||
del old_bucket[key]
|
||||
if not old_bucket:
|
||||
del self.freq_map[old_freq]
|
||||
if self.min_freq == old_freq:
|
||||
self.min_freq = min(self.freq_map.keys()) if self.freq_map else 0
|
||||
|
||||
new_bucket = self.freq_map.setdefault(new_freq, OrderedDict())
|
||||
new_bucket[key] = None
|
||||
else:
|
||||
# Insert new: evict LFU if at capacity
|
||||
if len(self.cache) >= self.capacity and self.capacity > 0:
|
||||
await self._evict_lfu()
|
||||
|
||||
freq = 1
|
||||
self.key_freq[key] = freq
|
||||
bucket = self.freq_map.setdefault(freq, OrderedDict())
|
||||
bucket[key] = None
|
||||
self.min_freq = 1
|
||||
|
||||
# Update node & TTL map
|
||||
self.cache[key] = _Node(key=key, value=value, ttl_expiry=current_time + ttl_seconds)
|
||||
self.ttl_map[key] = current_time + ttl_seconds
|
||||
|
||||
async def delete(self, key: str) -> bool:
|
||||
"""Delete a key from the cache. O(1)."""
|
||||
async with self._lock:
|
||||
if key not in self.cache:
|
||||
return False
|
||||
await self._remove_key(key)
|
||||
return True
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# INTERNAL HELPERS #
|
||||
# ------------------------------------------------------------------ #
|
||||
|
||||
async def _evict_lfu(self) -> None:
|
||||
"""Evict the least frequently used key (oldest among ties). O(1)."""
|
||||
if not self.freq_map or self.min_freq == 0:
|
||||
return
|
||||
|
||||
# Ensure min_freq points to a valid tier
|
||||
while self.min_freq in self.freq_map and self.min_freq < max(self.freq_map.keys()):
|
||||
pass
|
||||
if self.min_freq not in self.freq_map:
|
||||
self.min_freq = min(self.freq_map.keys()) if self.freq_map else 0
|
||||
|
||||
bucket = self.freq_map[self.min_freq]
|
||||
evict_key, _ = bucket.popitem(last=False)
|
||||
await self._remove_key(evict_key)
|
||||
|
||||
async def _remove_key(self, key: str) -> None:
|
||||
"""Remove key from all internal structures. O(1)."""
|
||||
if key not in self.cache:
|
||||
return
|
||||
|
||||
node = self.cache.pop(key)
|
||||
self.ttl_map.pop(key, None)
|
||||
|
||||
freq = self.key_freq.pop(key)
|
||||
bucket = self.freq_map[freq]
|
||||
del bucket[key]
|
||||
|
||||
if not bucket:
|
||||
del self.freq_map[freq]
|
||||
if self.min_freq == freq:
|
||||
# Find next valid minimum frequency
|
||||
self.min_freq = min(self.freq_map.keys()) if self.freq_map else 0
|
||||
|
||||
async def _background_loop(self) -> None:
|
||||
"""Non-blocking background task that purges expired keys in batches."""
|
||||
while True:
|
||||
await asyncio.sleep(0.1) # Check interval
|
||||
current_time = time.monotonic()
|
||||
|
||||
async with self._lock:
|
||||
# Collect expired keys safely
|
||||
expired_keys = [k for k, exp in list(self.ttl_map.items()) if current_time >= exp]
|
||||
|
||||
# Purge in batch (yields control between removals implicitly via await)
|
||||
for key in expired_keys:
|
||||
await self._remove_key(key)
|
||||
|
||||
async def apply_transaction(self, tx: "Transaction") -> None:
|
||||
"""Atomically apply transaction buffers to global state."""
|
||||
async with self._lock:
|
||||
# 1. Apply local deletes first (frees capacity for puts)
|
||||
for key in list(tx._local_deletes):
|
||||
if key in self.cache:
|
||||
await self._remove_key(key)
|
||||
tx._local_puts.pop(key, None)
|
||||
|
||||
# 2. Apply local puts
|
||||
for key, (value, expiry) in tx._local_puts.items():
|
||||
if key in self.cache:
|
||||
old_freq = self.key_freq[key]
|
||||
new_freq = old_freq + 1
|
||||
self.key_freq[key] = new_freq
|
||||
|
||||
old_bucket = self.freq_map[old_freq]
|
||||
del old_bucket[key]
|
||||
if not old_bucket:
|
||||
del self.freq_map[old_freq]
|
||||
if self.min_freq == old_freq:
|
||||
self.min_freq = min(self.freq_map.keys()) if self.freq_map else 0
|
||||
|
||||
new_bucket = self.freq_map.setdefault(new_freq, OrderedDict())
|
||||
new_bucket[key] = None
|
||||
else:
|
||||
if len(self.cache) >= self.capacity and self.capacity > 0:
|
||||
await self._evict_lfu()
|
||||
|
||||
freq = 1
|
||||
self.key_freq[key] = freq
|
||||
bucket = self.freq_map.setdefault(freq, OrderedDict())
|
||||
bucket[key] = None
|
||||
self.min_freq = 1
|
||||
|
||||
self.cache[key] = _Node(key=key, value=value, ttl_expiry=expiry)
|
||||
self.ttl_map[key] = expiry
|
||||
|
||||
|
||||
class Transaction:
|
||||
"""
|
||||
ACID-like sub-session handle supporting Read-Your-Own-Writes and isolation.
|
||||
Global readers do not see uncommitted writes until commit().
|
||||
"""
|
||||
|
||||
def __init__(self, cache: LFUCache):
|
||||
self.cache = cache
|
||||
self._local_puts: Dict[str, Tuple[Any, float]] = {} # key -> (value, absolute_expiry)
|
||||
self._local_deletes: Set[str] = set()
|
||||
|
||||
async def get(self, key: str) -> Optional[Any]:
|
||||
"""Read with local buffer priority (Read-Your-Own-Writes)."""
|
||||
if key in self._local_deletes:
|
||||
return None
|
||||
if key in self._local_puts:
|
||||
val, _ = self._local_puts[key]
|
||||
return val
|
||||
# Fall back to global cache (handles lazy eviction & lock)
|
||||
return await self.cache.get(key)
|
||||
|
||||
async def put(self, key: str, value: Any, ttl_seconds: float) -> None:
|
||||
"""Buffer write locally. Does not affect global state until commit."""
|
||||
current_time = time.monotonic()
|
||||
self._local_puts[key] = (value, current_time + ttl_seconds)
|
||||
if key in self._local_deletes:
|
||||
self._local_deletes.remove(key)
|
||||
|
||||
async def delete(self, key: str) -> None:
|
||||
"""Buffer deletion locally."""
|
||||
self._local_deletes.add(key)
|
||||
self._local_puts.pop(key, None)
|
||||
|
||||
async def commit(self) -> None:
|
||||
"""Atomically apply all buffered changes to the global cache."""
|
||||
await self.cache.apply_transaction(self)
|
||||
|
||||
def rollback(self) -> None:
|
||||
"""Discard all pending local changes without mutating global state."""
|
||||
self._local_puts.clear()
|
||||
self._local_deletes.clear()
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# TEST SUITE #
|
||||
# ------------------------------------------------------------------ #
|
||||
|
||||
async def main():
|
||||
print("🧪 Starting LFU Cache Test Suite...\n")
|
||||
|
||||
# a) O(1) LFU Eviction Order
|
||||
print("[a] Testing O(1) LFU Eviction Order...")
|
||||
cache = LFUCache(capacity=3)
|
||||
await cache.put("A", 1, ttl_seconds=60)
|
||||
await cache.put("B", 2, ttl_seconds=60)
|
||||
await cache.put("C", 3, ttl_seconds=60)
|
||||
|
||||
# Access A twice to increase its frequency
|
||||
await cache.get("A")
|
||||
await cache.get("A")
|
||||
|
||||
# Insert D. Should evict B or C (both freq=1). LFU policy guarantees one of them is gone.
|
||||
await cache.put("D", 4, ttl_seconds=60)
|
||||
|
||||
val_b = await cache.get("B")
|
||||
val_c = await cache.get("C")
|
||||
assert val_d := await cache.get("D"), "D should exist"
|
||||
assert val_a := await cache.get("A"), "A should exist (highest freq)"
|
||||
assert val_b is None or val_c is None, f"LFU eviction failed: B={val_b}, C={val_c}"
|
||||
print(f" ✅ LFU Eviction verified. Evicted key had lower frequency than A & D.")
|
||||
|
||||
# b) Lazy TTL vs Background Async Sweep
|
||||
print("\n[b] Testing Dual-Layer TTL Eviction...")
|
||||
cache2 = LFUCache(capacity=10)
|
||||
|
||||
# Lazy Eviction Test
|
||||
await cache2.put("lazy_key", "val", ttl_seconds=0.2)
|
||||
assert await cache2.get("lazy_key") == "val"
|
||||
await asyncio.sleep(0.3)
|
||||
assert await cache2.get("lazy_key") is None, "Lazy eviction failed"
|
||||
|
||||
# Background Eviction Test
|
||||
await cache2.start_evictor()
|
||||
await cache2.put("bg_key", "val", ttl_seconds=0.15)
|
||||
await asyncio.sleep(0.3) # Wait past TTL without accessing key
|
||||
assert await cache2.get("bg_key") is None, "Background async sweep failed"
|
||||
await cache2.stop_evictor()
|
||||
print(" ✅ Lazy & Background TTL eviction verified.")
|
||||
|
||||
# c) Transaction Commit Visibility vs Rollback
|
||||
print("\n[c] Testing Transaction Isolation & Rollback...")
|
||||
cache3 = LFUCache(capacity=10)
|
||||
|
||||
tx1 = cache3.begin_transaction() if hasattr(cache3, 'begin_transaction') else None
|
||||
class TxWrapper:
|
||||
def __init__(self, c): self.c = c
|
||||
def begin(self): return Transaction(self.c)
|
||||
tw = TxWrapper(cache3)
|
||||
|
||||
# Commit visibility
|
||||
tx_put = tw.begin()
|
||||
await tx_put.put("committed", 100, ttl_seconds=60)
|
||||
assert await cache3.get("committed") is None, "Uncommitted write should be invisible"
|
||||
await tx_put.commit()
|
||||
assert await cache3.get("committed") == 100, "Committed write should be visible globally"
|
||||
|
||||
# Rollback state restoration
|
||||
tx_roll = tw.begin()
|
||||
await tx_roll.put("rolled_back", 200, ttl_seconds=60)
|
||||
await tx_roll.rollback()
|
||||
assert await cache3.get("rolled_back") is None, "Rolled back write should not persist"
|
||||
|
||||
# Read-Your-Own-Writes inside transaction
|
||||
tx_ryo = tw.begin()
|
||||
await tx_ryo.put("local", 999, ttl_seconds=60)
|
||||
assert await tx_ryo.get("local") == 999, "Transaction should see its own writes"
|
||||
print(" ✅ Transaction commit visibility & rollback verified.")
|
||||
|
||||
# d) Stress Test: 50 Concurrent Tasks
|
||||
print("\n[d] Running Stress Test (50 concurrent async tasks)...")
|
||||
cache4 = LFUCache(capacity=100)
|
||||
await cache4.start_evictor()
|
||||
|
||||
errors = []
|
||||
results = {"gets": 0, "puts": 0}
|
||||
|
||||
async def worker(task_id: int):
|
||||
try:
|
||||
for i in range(20):
|
||||
key = f"stress_{task_id}_{i}"
|
||||
op_type = i % 3
|
||||
|
||||
if op_type == 0:
|
||||
await cache4.put(key, f"v_{task_id}_{i}", ttl_seconds=1.0)
|
||||
results["puts"] += 1
|
||||
elif op_type == 1:
|
||||
val = await cache4.get(key)
|
||||
if val is not None:
|
||||
results["gets"] += 1
|
||||
else:
|
||||
tx = Transaction(cache4)
|
||||
await tx.put(f"tx_{task_id}_{i}", "tx_val", ttl_seconds=0.5)
|
||||
assert await tx.get(f"tx_{task_id}_{i}") == "tx_val"
|
||||
await tx.commit()
|
||||
except Exception as e:
|
||||
errors.append((task_id, str(e)))
|
||||
|
||||
tasks = [asyncio.create_task(worker(i)) for i in range(50)]
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
assert len(errors) == 0, f"Stress test failed with errors: {errors}"
|
||||
print(f" ✅ Stress test passed. Processed {results['puts']} puts & {results['gets']} gets across 50 tasks without race conditions.")
|
||||
await cache4.stop_evictor()
|
||||
|
||||
print("\n🎉 All tests passed successfully!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
Reference in New Issue
Block a user