Direct answer
Use one OrderedDict per user as the O(1) LRU (least recently used) structure: move_to_end(key) marks an item as most recent (moves it to the newest end of the dictionary's order), popitem(last=False) removes the oldest (the entry at the other end), and both are O(1). Wrap a map from user to that structure in a fixed number of shards (stripes), each with its own threading.Lock, picking the shard by hash(user) % shards. Every get and put takes exactly one shard lock for its whole lookup, reorder and eviction. Bound memory twice: at most N items per user, and a cap on the number of users with the same LRU policy applied to users. Test with a stress run that checks invariants (conditions that must always hold, such as "no user holds more than N items") and a mutation check (deliberately breaking the code, here by removing the lock, to prove the test notices).
Implementation
The thread-safety requirement matters even with the GIL (the lock that lets one thread run Python bytecode at a time): a get that checks key in items and then calls move_to_end(key) is two steps, and another thread can evict the key in between, raising KeyError. For example, thread A checks that key k is present; before A's next line, thread B inserts enough new items that k is evicted; A's move_to_end(k) then raises KeyError. Putting the whole sequence under one lock removes that window.
python
import threading
from collections import OrderedDict
_MISSING = object()
class _Shard:
def __init__(self):
self.lock = threading.Lock()
self.users = OrderedDict() # user -> OrderedDict(key -> value); least recently used first
class PerUserLRU:
"""Keeps the `per_user` most recently used items for each user, and at most
about `max_users` users in total. get and put are O(1) and thread-safe."""
def __init__(self, per_user, max_users, shards=16):
if per_user < 1 or max_users < 1 or shards < 1:
raise ValueError("per_user, max_users and shards must be >= 1")
self._per_user = per_user
self._max_users_per_shard = -(-max_users // shards) # ceiling division
self._shards = [_Shard() for _ in range(shards)]
def _shard(self, user):
return self._shards[hash(user) % len(self._shards)]
def get(self, user, key, default=None):
shard = self._shard(user)
with shard.lock: # lookup, recency update and return are one critical section
items = shard.users.get(user)
if items is None or key not in items:
return default
items.move_to_end(key)
shard.users.move_to_end(user)
return items[key]
def put(self, user, key, value):
shard = self._shard(user)
with shard.lock:
items = shard.users.get(user)
if items is None:
items = shard.users[user] = OrderedDict()
items[key] = value
items.move_to_end(key)
if len(items) > self._per_user:
items.popitem(last=False) # drop this user's least recently used item
shard.users.move_to_end(user)
while len(shard.users) > self._max_users_per_shard:
shard.users.popitem(last=False) # drop the least recently used user
def stats(self):
"""(number of users, largest per-user size, total items), taken shard by shard."""
users = largest = total = 0
for shard in self._shards:
with shard.lock:
users += len(shard.users)
for items in shard.users.values():
largest = max(largest, len(items))
total += len(items)
return users, largest, total
Choosing the locking granularity
- One global lock: simplest and obviously correct, but every operation for every user serialises. Fine for low traffic; it becomes the bottleneck when many threads touch the cache.
- One lock per user: maximum parallelism between users, but you now need a lock protecting the user-to-lock map, and evicting a whole user while another thread holds or waits on that user's lock is a lifetime race (a bug where an object is removed while someone still holds a reference to it). Concretely: thread A is waiting for user U's lock; thread B holds it, inserts for other users until U is the least recently used user and evicts it, then releases; A acquires the lock of an entry that is no longer in the map and writes its item there, so the write is silently lost and a later
put creates a second, different lock for U.
- Striped locks (chosen): a fixed array of shards. Contention drops roughly in proportion to the shard count when users are spread evenly, each user lives in exactly one shard so per-user order is exact, and user eviction happens under the same lock as the user's operations, so there is no lifetime race. The cost is that two unrelated users in one shard still wait for each other, and the user cap is enforced per shard.
- Keep critical sections short: the lock covers only dictionary operations. Never compute a value, call the database or do I/O while holding it; compute first, then
put.
Per-shard user eviction means the global bound is shards x ceil(max_users / shards) users. The expression -(-max_users // shards) is ceiling division: Python's // rounds toward minus infinity, so negating before and after rounds up instead. With max_users=64 and 4 shards it is 16 per shard, 64 in total. With max_users=10 and 4 shards, -(-10 // 4) is 3 per shard, so the real bound is 4 x 3 = 12 users (the cap is "about" max_users, as the class docstring says). The same applies to the number of items: each user holds at most per_user, so 64 users and 64 x 4 = 256 items for the test's numbers. stats() locks shard by shard, so it is not an atomic snapshot of the whole cache, which is fine for monitoring.
Memory bound and eviction
- Items per user are capped at
per_user: inserting beyond it removes that user's least recently used item.
- Users are capped at
max_users: inserting for a new user in a full shard removes the least recently used user, with all of that user's items. A user touched by get or put becomes the most recent user.
- Entries do not expire by time. To add a time-to-live, store
(value, expires_at) and treat an expired entry as a miss on get, deleting it then; a periodic sweep keeps idle users from holding memory.
Testing under concurrent load
The test file checks two things. A sequential test pins the LRU semantics (a get refreshes recency, the third insert evicts the least recently used item, the user cap evicts the oldest user). A stress test runs eight threads released together by a Barrier, each doing 40,000 random puts and gets over 200 users and 12 keys, with sys.setswitchinterval(1e-6) to force frequent thread switches (a Barrier makes all threads wait until everyone has arrived, so they start at the same moment). Each value is a pure function of (user, key), so any get that returns something else exposes a mixed-up entry, and after the run the cache must still satisfy its size bounds.
python
import random
import sys
import threading
from per_user_lru import PerUserLRU
def test_sequential_semantics():
c = PerUserLRU(per_user=2, max_users=2, shards=1)
c.put("u1", "a", 1)
c.put("u1", "b", 2)
assert c.get("u1", "a") == 1 # a is now most recent, b is least
c.put("u1", "c", 3) # evicts b
assert c.get("u1", "b") is None
assert c.get("u1", "a") == 1 and c.get("u1", "c") == 3
c.put("u2", "x", 9)
c.put("u3", "y", 7) # users capped at 2 in the single shard: u1 is oldest, goes
assert c.get("u1", "a") is None
assert c.get("u2", "x") == 9 and c.get("u3", "y") == 7
assert c.get("u2", "missing", "d") == "d"
def stress(threads=8, ops=40_000, per_user=4, max_users=64):
sys.setswitchinterval(1e-6)
cache = PerUserLRU(per_user, max_users, shards=4)
errors = []
start = threading.Barrier(threads)
def worker(seed):
rng = random.Random(seed)
start.wait()
try:
for _ in range(ops):
user = f"user{rng.randrange(200)}"
key = rng.randrange(12)
if rng.random() < 0.5:
cache.put(user, key, (user, key)) # the value is a pure function of (user, key)
else:
got = cache.get(user, key)
if got is not None and got != (user, key):
errors.append(f"wrong value {got} for {(user, key)}")
except Exception as e: # a lost race typically shows up as KeyError
errors.append(f"{type(e).__name__}: {e}")
ts = [threading.Thread(target=worker, args=(i,)) for i in range(threads)]
for t in ts:
t.start()
for t in ts:
t.join()
users, largest, total = cache.stats()
# per-shard user cap is ceil(64/4) = 16, so at most 64 users and 64 * 4 items overall
if largest > per_user:
errors.append(f"a user holds {largest} items, limit {per_user}")
if users > max_users or total > max_users * per_user:
errors.append(f"bound broken: {users} users, {total} items")
return errors
if __name__ == "__main__":
test_sequential_semantics()
print("sequential semantics: ok")
failed = 0
for round_no in range(10):
errors = stress()
if errors:
failed += 1
print("round", round_no, "FAILED:", errors[0], f"(+{len(errors) - 1} more)")
print(f"concurrent stress: {10 - failed} of 10 rounds clean")
Run python test_per_user_lru.py with the two files in one directory (Python 3.12):
text
sequential semantics: ok
concurrent stress: 10 of 10 rounds clean
Reading what a failure would mean: a KeyError (or other exception) means a thread saw a structure change between two steps, which is a lost race. A wrong value message means one entry's data got mixed with another's, since each value is a function of its own (user, key). A a user holds ... items or bound broken message means eviction failed to keep a size limit, for example because two threads each saw room and both inserted. The final stats() check runs after all threads have joined, so it is a single-threaded look at the end state.
Repeated runs with the real lock print 10 of 10 rounds clean. The harness can fail: replacing the shard lock with a do-nothing context manager (contextlib.nullcontext(), which supports with but excludes nobody, in place of threading.Lock()) makes the rounds fail with KeyError, typically all of them, so the summary line reads far below 10 of 10 rounds clean. A passing stress test is evidence, not proof; the proof is the argument that every access to shared state happens under its shard's lock.
Trade-offs and pitfalls
functools.lru_cache is not a substitute. Its cache is thread-safe, but it has one global capacity keyed on function arguments, no per-user bound, and the Python documentation notes the wrapped function may be called more than once when threads miss on the same key at the same time.- Duplicate computation on a simultaneous miss is the same: the cache above does not coordinate loaders. If computing is expensive, store a future or per-key lock for in-flight loads, at the cost of more code and a place where one slow loader blocks others.
- The cache is per process. Several worker processes each hold their own copy, so hit rates drop and entries can disagree. If consistency matters, use a shared store such as Redis with an eviction policy.
- Do not hold the lock across callbacks. Values and keys must be cheap to hash and compare; a custom
__hash__ or __eq__ that calls into other code runs under the lock.
- What would change the design: if profiling shows shard locks contended, increase the shard count; if per-user ordering across a user's requests is not needed, an approximate algorithm with batched recency updates would cut lock hold time.
