Design an In-Memory Key-Value Cache
OpenAIMachine Learning Engineer · Technical ScreenLast updated
Implement a thread-safe in-memory key-value cache supporting capacity bounds, LRU eviction, and per-key time-to-live (TTL) expiration.
Problem Statement
Design and implement a robust, in-memory key-value cache with capacity limits, LRU (Least Recently Used) eviction, and optional per-key TTL (Time-To-Live) expiration. Your implementation must be safe for concurrent use by multiple threads.
Class Interface
from typing import Any, Optional
class KeyValueStore:
def __init__(self, capacity: int):
"""
Initialize the store with a maximum capacity of live entries.
"""
pass
def get(self, key: str) -> Optional[Any]:
"""
Retrieve the value for key if present and not expired.
Updates the recency of the key.
Returns None if key is absent or expired.
"""
pass
def put(self, key: str, value: Any, ttl: Optional[float] = None) -> None:
"""
Insert or update a key-value pair with an optional TTL (in seconds).
If key exists, updates its value, TTL, and recency.
If at capacity, evicts expired items first; if still at capacity,
evicts the least recently used live item.
"""
pass
def delete(self, key: str) -> bool:
"""
Explicitly remove a key. Returns True if removed, False otherwise.
"""
pass
def size(self) -> int:
"""
Returns the count of valid (non-expired) items currently stored.
"""
pass
Requirements & Semantics
- Complexity:
get,put, anddeleteshould run inO(1)average time. - Eviction Order: When at
capacityduring aput:- Check and remove expired entries first if possible.
- If the cache remains at
capacity, evict the least recently used (LRU) entry.
- TTL Handling: A key with a TTL expires after
current_time + ttl. Expired entries must never be returned byget. - Concurrency: All operations must be thread-safe.
Example
cache = KeyValueStore(capacity=2)
cache.put("a", 1, ttl=10.0)
cache.put("b", 2, ttl=10.0)
assert cache.get("a") == 1 # "a" is now most recently used
cache.put("c", 3) # Capacity full; "b" is LRU -> evicts "b"
assert cache.get("b") is None
assert cache.get("c") == 3
Constraints & Edge Cases
capacity >= 1ttlis eitherNone(no expiration) or a positive float (ttl > 0).- Handle cases where an entry expires before being accessed or updated.
- Avoid deadlocks during concurrent access.
Hints
Hint 1
Combine a hash map with a doubly linked list to achieve O(1) recency updates and LRU evictions.
Hint 2
Store the absolute expiration timestamp (`time.time() + ttl`) with each entry, and eagerly prune expired items if space is needed during `put`.
Hint 3
Guard shared state modifications with a threading lock (`threading.Lock` or `threading.RLock`) to ensure concurrency safety.
Follow-up questions
- How would you prevent memory leaks from expired keys that are never read again without making `put` or `get` O(n)?
- How would you optimize this implementation for a read-heavy workload (e.g., 99% reads, 1% writes)?
- How would your design change if this cache had to support zero-copy sharing of large numpy tensors across multiple Python processes?
Solution walkthrough is for members
Get the full walkthrough with reference code and complexity analysis, plus Kitty Coach at 300 messages a day. Hints, the code runner and community answers stay free.