Cache stampede on a hot key

import threading
import time
from collections.abc import Callable

import redis

r = redis.Redis(decode_responses=True)
DB_SECONDS = 0.25
ARRIVALS = [10, 10, 10, 10, 5, 5]  # requests per 100 ms slot
db_reads = 0
lock = threading.Lock()


def load_from_db() -> str:
    global db_reads
    with lock:
        db_reads += 1
    time.sleep(DB_SECONDS)
    return "report-v2"


def no_protection() -> None:
    if r.get("report") is None:
        r.set("report", load_from_db(), ex=60)


def single_flight() -> None:
    while r.get("report") is None:
        if r.set("report:lock", "1", nx=True, px=5000):
            if r.get("report") is None:  # re-check under the lock
                r.set("report", load_from_db(), ex=60)
            r.delete("report:lock")
        else:
            time.sleep(0.02)


def refresh() -> None:
    r.set("report", load_from_db())
    r.set("report:fresh", "1", ex=60)
    r.delete("report:lock")


def stale_while_revalidate() -> None:
    value = r.get("report")  # stale copy is still there
    stale = r.get("report:fresh") is None
    if stale and r.set("report:lock", "1", nx=True, px=5000):
        threading.Thread(target=refresh).start()
    assert value is not None


def run(name: str, handler: Callable[[], None], stale: bool) -> None:
    global db_reads
    r.flushall()
    db_reads = 0
    if stale:
        r.set("report", "report-v1")
    threads = []
    for slot, n in enumerate(ARRIVALS):
        for _ in range(n):
            t = threading.Timer(slot * 0.1, handler)
            threads.append(t)
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    time.sleep(DB_SECONDS + 0.1)
    print(f"{name}: {db_reads}")


run("No protection", no_protection, False)
run("Single-flight", single_flight, False)
run("Stale-while-revalidate", stale_while_revalidate, True)