Problem 87 · Concurrency & Python

Thread-Safe LRU Cache Decorator with TTL

HardClassic DSA & systems

01Problem

Implement a memoizing decorator with a max size (LRU eviction) and per-entry time-to-live that is safe to call from multiple threads.

02Key insight

OrderedDict gives O(1) move_to_end/popitem for LRU; store (value, expires_at). Hold the lock only for cache bookkeeping and compute outside it so one slow call does not serialize all callers (accepting possible duplicate computes - the 'thundering herd' trade-off).

03Solution

solution.pypython43 lines

import functools
import threading
import time
from collections import OrderedDict
from typing import Any, Callable, Hashable


def ttl_lru_cache(maxsize: int = 128, ttl: float = 60.0,
                  clock: Callable[[], float] = time.monotonic) -> Callable:
    def decorator(fn: Callable) -> Callable:
        cache: OrderedDict[Hashable, tuple[Any, float]] = OrderedDict()
        lock = threading.Lock()

        @functools.wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            key = (args, tuple(sorted(kwargs.items())))
            with lock:
                entry = cache.get(key)
                if entry is not None:
                    if entry[1] > clock():
                        cache.move_to_end(key)
                        return entry[0]
                    del cache[key]  # expired
            value = fn(*args, **kwargs)  # compute outside the lock
            with lock:
                cache[key] = (value, clock() + ttl)
                cache.move_to_end(key)
                while len(cache) > maxsize:
                    cache.popitem(last=False)  # evict least recently used
            return value

        def cache_info() -> int:
            with lock:
                return len(cache)

        def cache_clear() -> None:
            with lock:
                cache.clear()

        wrapper.cache_info = cache_info  # type: ignore[attr-defined]
        wrapper.cache_clear = cache_clear  # type: ignore[attr-defined]
        return wrapper
    return decorator

Try it first.

TimeO(1) per lookup

SpaceO(maxsize)

tests.py · 3 passing

tests.pypython22 lines

import threading
now = [0.0]
calls = []
@ttl_lru_cache(maxsize=2, ttl=10, clock=lambda: now[0])
def sq(x):
    calls.append(x)
    return x * x
assert sq(2) == 4 and sq(2) == 4 and calls == [2]
sq(3); sq(2); sq(4)            # 3 is LRU -> evicted
assert sq.cache_info() == 2
sq(3); assert calls == [2, 3, 4, 3]
now[0] = 100.0                  # everything expired
sq(4); assert calls[-1] == 4 and len(calls) == 5
ts = [threading.Thread(target=lambda: [sq(i % 2) for i in range(300)]) for _ in range(4)]
for t in ts: t.start()
for t in ts: t.join()
assert sq.cache_info() <= 2
ts = [threading.Thread(target=lambda: [sq(i % 5) for i in range(300)]) for _ in range(3)]
ts.append(threading.Thread(target=lambda: [sq.cache_clear() for _ in range(300)]))
for t in ts: t.start()
for t in ts: t.join()
sq.cache_clear(); assert sq.cache_info() == 0

04Staff-level follow-up

How do you stop N threads from computing the same expensive miss? Single-flight: store a per-key Future/Event in the cache on miss so later callers wait on it; for a fleet, use a distributed lock or request coalescing at the cache tier.