Build a thread-safe in-memory cache in Python with the following requirements: Core Features:
put(key, value, ttl=None) → Add item with optional TTL (in seconds).
get(key) → Return value if present and not expired. Also update access order for LRU.
delete(key) → Remove a key.
clear() → Clear entire cache.
Internal Design:
Use a hash map (dictionary) for O(1) lookup.
Implement LRU eviction using a doubly linked list.
Track TTL expiration using timestamps.
Support per-entry TTL and default TTL.
Use a background thread to periodically remove expired items.
LRU Logic:
Maintain a max cache size (max_size).
If size exceeded, evict least recently used item in O(1) time.
Thread-Safe:
All operations must be thread-safe.
Use threading.Lock or RLock to protect shared structures.
Avoid deadlocks and race conditions.
Statistics:
Implement get_stats() returning:
{
"hits": int,
"misses": int,
"hit_rate": float,
"total_requests": int,
"current_size": int,
"evictions": int,
"expired_removals": int
}
Tests:
Add test cases for:
Basic operations (put, get, delete, clear)
LRU eviction after exceeding size
TTL expiration
Concurrent access using threading.Thread
Extras:
Add logging to show evictions, TTL expiry, and background cleaner actions.
Make sure expired entries are removed on access and in background cleanup.
Output:
Clean, modular Python code.
Include comments and error handling.
Include README.md explaining design, usage, and how to run the tests.
I'll build a comprehensive thread-safe in-memory cache system in Python with all the features you requested. Let me break this down into modular components.
I've created a comprehensive thread-safe in-memory cache system with all the features you requested:
put()
, get()
, delete()
, clear()
with O(1) performancethreading.RLock
for all operationsThe system includes extensive examples and tests demonstrating:
You can run the tests to verify functionality:
python -m tests.test_cache
python -m tests.test_performance
python examples/usage_example.py
The implementation is production-ready with proper error handling, logging, and documentation. All operations are thread-safe and the system handles edge cases gracefully.
how should I show this to an interviewer, do we have to make a front end part for this?
Great question! For a technical interview, you don't need a frontend for this type of system-level component. Here's how to effectively present this cache implementation to an interviewer:
# Create a simple demo script that shows key features
def demo_for_interviewer():
print("=== Thread-Safe Cache Demo ===\n")
# Initialize cache
cache = ThreadSafeCache(max_size=3, default_ttl=None)
print("1. Basic Operations:")
cache.put("user:1", {"name": "Alice", "role": "admin"})
cache.put("user:2", {"name": "Bob", "role": "user"})
print(f" Retrieved user:1 -> {cache.get('user:1')}")
print(f" Cache size: {cache.size()}")
print("\n2. LRU Eviction Demo:")
cache.put("user:3", {"name": "Charlie", "role": "user"})
cache.get("user:1") # Make user:1 most recently used
cache.put("user:4", {"name": "David", "role": "user"}) # Should evict user:2
print(f" user:2 after eviction: {cache.get('user:2')}") # None
print(f" user:1 still exists: {cache.get('user:1')}") # Still there
print("\n3. TTL Demo:")
cache.put("session:abc", "active", ttl=2)
print(f" Session immediately: {cache.get('session:abc')}")
time.sleep(2.1)
print(f" Session after 2s: {cache.get('session:abc')}") # None
print("\n4. Statistics:")
stats = cache.get_stats()
print(f" Hit rate: {stats['hit_rate']:.1%}")
print(f" Total requests: {stats['total_requests']}")
print(f" Evictions: {stats['evictions']}")
cache.shutdown()
Show the interviewer how to use it interactively:
>>> from cache import ThreadSafeCache
>>> cache = ThreadSafeCache(max_size=5)
>>> cache.put("key1", "value1")
>>> cache.get("key1")
'value1'
>>> cache.get_stats()
{'hits': 1, 'misses': 0, 'hit_rate': 1.0, ...}
"I built a thread-safe in-memory cache with these requirements:
- O(1) get/put operations
- LRU eviction when full
- TTL support with background cleanup
- Thread-safe for concurrent access
- Statistics tracking
Key design decisions:
- HashMap + Doubly Linked List for LRU
- RLock for thread safety
- Background daemon thread for cleanup"
Show the file structure and explain:
cache/
├── __init__.py # Public API
├── thread_safe_cache.py # Main implementation
├── lru_list.py # Doubly linked list
├── node.py # Cache node structure
└── stats.py # Statistics tracking
Run your demo script showing:
# Test everything works
python examples/usage_example.py
python -m tests.test_cache
python -m tests.test_performance
Q: "How would you make this distributed?" A: "Add consistent hashing, network protocol, replication..."
Q: "What about persistence?" A: "Could add write-through/write-behind to disk or database..."
Q: "Performance bottlenecks?" A: "Lock contention is main bottleneck. Could use lock-free structures or sharding..."
Q: "How to monitor in production?" A: "Export metrics to Prometheus, add health checks, log eviction patterns..."
You could quickly create a simple monitoring script:
def monitor_cache():
cache = ThreadSafeCache(max_size=10)
while True:
stats = cache.get_stats()
print(f"Size: {stats['current_size']}/10 | "
f"Hit Rate: {stats['hit_rate']:.1%} | "
f"Evictions: {stats['evictions']}")
time.sleep(1)
Bottom Line: Your comprehensive implementation with tests, examples, and documentation is exactly what technical interviewers want to see. No frontend required!
what we have done here how will i explain to interviewer