How to Stop a Cache Stampede in Python With Single-Flight Loads, TTL Jitter, and Background Refresh
Reproduce a cache stampede in Python asyncio, then fix it with single-flight loads, TTL jitter, stale-while-revalidate and XFetch early refresh, with measured results and a pytest suite.
What you will build, and why cache stampedes matter
Picture a page that costs your database three seconds to build, shown to ten visitors every second. A cache hides that cost. The first visitor pays for the three seconds and the next thousand get the stored copy for free. Then the stored copy expires. For the next three seconds, every visitor who arrives finds the cache empty, and every one of them starts the same three-second job. That is thirty copies of the same query running at once. The database slows down under the extra load, so the jobs take longer than three seconds, more visitors pile in behind them, and the cache never gets refilled.
Table Of Content
- What you will build, and why cache stampedes matter
- Seven terms to know before we start
- Prerequisites
- Create the project
- Step 1: Set up the lab and reproduce a stampede
- 1.1 Check your Python
- 1.2 Build a slow backend
- 1.3 Write the naive cache
- 1.4 Trigger the stampede
- Step 2: Let one request do the work with single-flight
- 2.1 Write the first version
- 2.2 Run the same burst
- Step 3: Three ways the first version breaks, and the fixed class
- 3.1 An await between the check and the registration
- 3.2 One caller gives up
- 3.3 The load fails once
- What the output shows
- 3.4 The fixed class
- 3.5 Check the fix
- Step 4: Spread expiry out with TTL jitter
- 4.1 Measure what jitter does
- Step 5: Hide the wait with stale-while-revalidate
- 5.1 Write the class
- 5.2 Watch it work
- Step 6: Refresh early with XFetch
- 6.1 The rule
- 6.2 Write the class
- 6.3 Check the probabilities
- Step 7: Compare all four and combine them
- 7.1 Run the comparison
- Reading the table
- Choosing a strategy
- 7.2 Combine the pieces
- 7.3 Lock it in with tests
- Step 8: What about threads, and why not just functools.lru_cache?
- 8.1 Measure it
- 8.2 A trap with async functions
- Check that everything works end to end
- Where this stops working, and what to learn next
This failure is called a cache stampede, or dog-piling. Wikipedia describes it like this: “A cache stampede is a type of cascading failure that can occur when massively parallel computing systems with caching mechanisms come under a very high load.” The arithmetic above is the example used by the authors of a 2015 VLDB paper on the problem: “if the cache item is accessed 10 times per second, and the recomputation of the item takes 3 seconds, then 30 requests will recompute the item.”
In this tutorial you will reproduce a stampede in a small Python program and then build four defenses, measuring each one. Single-flight makes sure only one request does the expensive work while the others wait for its answer. TTL jitter stops thousands of different entries from expiring in the same second. Stale-while-revalidate hands callers the old value instantly while one background task fetches a new one. XFetch refreshes a hot entry a little before it expires, using a probability rule from the paper above. You will also break your first version of single-flight on purpose, in three different ways, because the bugs teach more than the happy path. The code uses only the Python standard library and pytest, and every number printed below came from running it.
Seven terms to know before we start
Cache hit and cache miss. A hit means the cache holds a usable answer and returns it. A miss means it does not, so the program has to do the real work.
TTL (time to live). How long a stored answer counts as fresh. After that the entry has expired.
Loader. The function that does the expensive work on a miss, such as a database query or an API call. In this tutorial it is an async function.
Stampede. Many requests missing on the same key at the same moment and each running the loader.
Single-flight. Letting one request run the loader for a key while everyone else waits for that run and shares its result. The name comes from the Go ecosystem, where the singleflight package is described as “a duplicate function call suppression mechanism.”
Stale. An entry past its TTL. It is out of date but still holds an answer that may be good enough for a short while.
Jitter. A small random change applied on purpose, here to how long each entry lives, so that things which would happen together happen a little apart.
Prerequisites
- Python 3.12 or newer. Everything here ran on Python 3.13.14 on Windows 11. One test uses eager tasks, a feature added in Python 3.12, and everything else uses features that are older.
- Basic comfort with
asyncandawait: anawaitis the point where a coroutine pauses and lets other coroutines run. You do not need to know anything about asyncio internals; the parts that matter are explained when they appear. - pytest 9.1.1 for the test suite in Step 7. The demos themselves need nothing beyond the standard library.
- A terminal and an empty folder. There is no Redis, no database and no network access in this tutorial; the “backend” is a small class that pretends to be slow. The last section explains what changes when your cache lives in several processes.
Create the project
Make a folder, create a virtual environment inside it, activate it, and install pytest:
mkdir herd
cd herd
python -m venv .venv
.venv\Scripts\activate
python -m pip install pytest==9.1.1On macOS and Linux the activation line is source .venv/bin/activate. Every file in the rest of this tutorial goes in the herd folder, and every command is run from there.
Step 1: Set up the lab and reproduce a stampede
Before fixing anything you need to see the problem happen. This step builds a pretend database and a cache that does what most first attempts do.
1.1 Check your Python
Save this small script and run it with python step00_versions.py. It prints your Python version and operating system so you can compare them with the ones used here.
# step00_versions.py
import platform
import sys
print("Python", sys.version.split()[0], "on", platform.system())Expected output on the machine used for this article:
Python 3.13.14 on Windows1.2 Build a slow backend
The class below stands in for a database. Each query takes latency seconds, and the server can only run capacity queries at the same moment; a query that arrives while it is full is refused with an error, the way a database with a full connection pool refuses a new connection. The class counts every call it receives and remembers the most queries it ever ran at once, which is the evidence we will use all through this tutorial. fail_next lets later steps make the next few calls fail on demand. Each successful answer carries an as_of timestamp, so a test can tell old answers from new ones.
# backend.py
import asyncio
import time
class BackendOverloaded(Exception):
pass
class SlowBackend:
"""A pretend database. Each query takes `latency` seconds, and the server
refuses new queries while `capacity` of them are already running."""
def __init__(self, latency=0.2, capacity=20):
self.latency = latency
self.capacity = capacity
self.fail_next = 0 # make the next N calls fail at once
self.in_flight = 0
self.reset()
def reset(self):
self.calls = 0
self.max_in_flight = 0
async def fetch(self, key):
self.calls += 1
if self.fail_next > 0:
self.fail_next -= 1
raise BackendOverloaded(f"backend refused {key!r}")
if self.in_flight >= self.capacity:
raise BackendOverloaded(f"{self.in_flight} queries already running")
self.in_flight += 1
self.max_in_flight = max(self.max_in_flight, self.in_flight)
try:
await asyncio.sleep(self.latency)
finally:
self.in_flight -= 1
return {"key": key, "as_of": time.monotonic()}asyncio.sleep(self.latency) is the stand-in for waiting on a real query. While one task sleeps, the event loop runs other tasks, exactly as it would while a real database call is in flight.
1.3 Write the naive cache
Now the cache. It is the pattern from Wikipedia’s “typical cache usage” pseudocode: look in the cache, and on a miss call the loader and store the answer. An Entry keeps the value together with the clock time at which it expires. The clock argument defaults to time.monotonic, a clock that only moves forward, and exists so that tests can later pass in a fake one.
# naive.py
import time
from dataclasses import dataclass
@dataclass
class Entry:
value: object
expires_at: float
class NaiveCache:
def __init__(self, loader, *, ttl, clock=time.monotonic):
self._loader = loader
self._ttl = ttl
self._clock = clock
self._data = {}
async def get(self, key):
entry = self._data.get(key)
if entry is not None and entry.expires_at > self._clock():
return entry.value
value = await self._loader(key)
self._data[key] = Entry(value, self._clock() + self._ttl)
return value1.4 Trigger the stampede
This script warms the cache, waits for the entry to expire, and then fires 200 requests for the same key at the same instant, the way a burst of visitors would. asyncio.gather starts all 200 coroutines together, and return_exceptions=True keeps failures as results instead of stopping at the first one. The backend can only run 20 queries at once.
# step01_stampede.py
import asyncio
from backend import SlowBackend
from naive import NaiveCache
async def main():
backend = SlowBackend(latency=0.2, capacity=20)
cache = NaiveCache(backend.fetch, ttl=0.5)
await cache.get("home") # warm the cache
print("warm-up backend calls:", backend.calls)
await asyncio.sleep(0.6) # let the entry expire
backend.reset()
results = await asyncio.gather(
*(cache.get("home") for _ in range(200)), return_exceptions=True
)
errors = [r for r in results if isinstance(r, Exception)]
print("requests:", len(results))
print("backend calls:", backend.calls)
print(f"most queries running at once: {backend.max_in_flight} (capacity {backend.capacity})")
print("requests that failed:", len(errors))
print("first error:", repr(errors[0]))
asyncio.run(main())Run it with python step01_stampede.py:
warm-up backend calls: 1
requests: 200
backend calls: 200
most queries running at once: 20 (capacity 20)
requests that failed: 180
first error: BackendOverloaded('20 queries already running')One value was needed, and the backend was asked for it 200 times. Twenty queries were accepted, and the other 180 requests failed. Here is why. Each of the 200 tasks runs get and finds nothing fresh. Each then calls the loader, which contains an await. An await is where a task steps aside so that others can run, so all 200 tasks get past the if check before the first load has finished. Nothing in the code lets one task notice that another is already doing the same work. In a real service the refused requests would be timeouts or a growing queue instead of instant errors, but the cause is the same: demand for one key multiplied by the time one load takes.
Step 2: Let one request do the work with single-flight
The fix is to give the work an owner. On a miss, the first request starts the load as its own task, which is a unit of work that asyncio schedules independently, and records that task in a dictionary keyed by the cache key. A later request that also misses looks in the dictionary, finds the running task, and simply waits for it instead of starting another. When the task finishes, everyone waiting receives the same result.
Wikipedia files this under the “locking” family of fixes: “To prevent multiple simultaneous recomputations of the same value, upon a cache miss a process will attempt to acquire the lock for that cache key and recompute it only if it acquires it.” The Go documentation for Do describes the shared-result variant used here: “Do executes and returns the results of the given function, making sure that only one execution is in-flight for a given key at a time. If a duplicate comes in, the duplicate caller waits for the original to complete and receives the same results.”
2.1 Write the first version
Save this as single_flight_v1.py. It is deliberately a first draft, and Step 3 will show where it breaks. The dictionary _inflight maps each key to the task loading it right now.
# single_flight_v1.py
import asyncio
import time
from naive import Entry
class SingleFlightV1:
def __init__(self, loader, *, ttl, clock=time.monotonic):
self._loader = loader
self._ttl = ttl
self._clock = clock
self._data = {}
self._inflight = {} # key -> the task that is loading it right now
async def get(self, key):
entry = self._data.get(key)
if entry is not None and entry.expires_at > self._clock():
return entry.value
task = self._inflight.get(key)
if task is None:
task = asyncio.create_task(self._load(key))
self._inflight[key] = task
return await task
async def _load(self, key):
value = await self._loader(key)
self._data[key] = Entry(value, self._clock() + self._ttl)
self._inflight.pop(key, None)
return valueNotice that there is no lock anywhere. That is safe in asyncio because tasks only switch places at an await. Between reading self._inflight.get(key) and writing self._inflight[key] = task there is no await, so no other task can run in that gap. Step 3 shows what happens when someone adds one.
2.2 Run the same burst
This is the Step 1 script with the new class. Save it as step02_coalesce.py and run it.
# step02_coalesce.py
import asyncio
from backend import SlowBackend
from single_flight_v1 import SingleFlightV1
async def main():
backend = SlowBackend(latency=0.2, capacity=20)
cache = SingleFlightV1(backend.fetch, ttl=0.5)
await cache.get("home") # warm the cache
await asyncio.sleep(0.6) # let the entry expire
backend.reset()
results = await asyncio.gather(
*(cache.get("home") for _ in range(200)), return_exceptions=True
)
errors = [r for r in results if isinstance(r, Exception)]
print("requests:", len(results))
print("backend calls:", backend.calls)
print("most queries running at once:", backend.max_in_flight)
print("requests that failed:", len(errors))
print("distinct values returned:", len({id(r) for r in results}))
asyncio.run(main())requests: 200
backend calls: 1
most queries running at once: 1
requests that failed: 0
distinct values returned: 1The same 200 requests now cause one backend call, one query running at a time, no failures, and one shared answer: the last line counts the distinct objects returned, and all 200 callers received the same one. That is single-flight working.
Step 3: Three ways the first version breaks, and the fixed class
The first version passes the happy-path test, which is how bugs like these slip past code review. Each one below is a real mistake that is easy to make, and each is reproduced here with output. They share a helper module, pitfalls.py, which you will reuse in 3.5 to check the fixed class.
3.1 An await between the check and the registration
Suppose a colleague adds a metrics call, such as counting cache misses, in the middle of get. The subclass below simulates that with await asyncio.sleep(0), which does nothing except give other tasks a turn. The helper then sends 200 requests to a class and prints how many backend calls resulted.
# pitfalls.py
import asyncio
from backend import BackendOverloaded, SlowBackend
from single_flight_v1 import SingleFlightV1
class AwaitBeforeRegister(SingleFlightV1):
"""The first version plus one innocent-looking await between check and register."""
async def get(self, key):
entry = self._data.get(key)
if entry is not None and entry.expires_at > self._clock():
return entry.value
task = self._inflight.get(key)
if task is None:
await asyncio.sleep(0) # imagine metrics.incr("cache.miss") here
task = asyncio.create_task(self._load(key))
self._inflight[key] = task
return await task
async def check_register_race(cls):
backend = SlowBackend(latency=0.2, capacity=1000)
cache = cls(backend.fetch, ttl=5)
await asyncio.gather(*(cache.get("home") for _ in range(200)))
print(f"{cls.__name__}: 200 requests made {backend.calls} backend call(s)")3.2 One caller gives up
Real callers time out or disconnect, and many async frameworks respond by cancelling the task that was handling the request. This helper starts one impatient caller that gives up after 50 milliseconds, using asyncio.wait_for, and 20 patient callers that wait as long as it takes. The load takes 200 milliseconds. At the end it makes one more request to see what state the cache was left in.
# pitfalls.py (continued)
async def check_cancellation(cls):
backend = SlowBackend(latency=0.2, capacity=1000)
cache = cls(backend.fetch, ttl=5)
async def impatient():
return await asyncio.wait_for(cache.get("home"), timeout=0.05)
results = await asyncio.gather(
impatient(), *(cache.get("home") for _ in range(20)), return_exceptions=True
)
patient = results[1:]
got_value = sum(isinstance(r, dict) for r in patient)
cancelled = sum(isinstance(r, asyncio.CancelledError) for r in patient)
print(f"{cls.__name__}: the impatient caller got {type(results[0]).__name__}")
print(f" of 20 patient callers, {got_value} got a value and {cancelled} were cancelled")
print(f" backend calls: {backend.calls}")
try:
await cache.get("home")
print(" the next request got a value")
except asyncio.CancelledError:
print(" the next request raised CancelledError")3.3 The load fails once
This helper makes the backend fail on its next call, makes a request, lets the backend recover, and makes a second request.
# pitfalls.py (continued)
async def check_failure(cls):
backend = SlowBackend(latency=0.05, capacity=1000)
cache = cls(backend.fetch, ttl=5)
backend.fail_next = 1
try:
await cache.get("home")
except BackendOverloaded as exc:
print(f"{cls.__name__}: first request failed: {exc}")
await asyncio.sleep(0.1)
print(f" the backend is healthy again; calls so far: {backend.calls}")
try:
await cache.get("home")
print(" second request: got a value")
except BackendOverloaded as exc:
print(f" second request failed too: {exc}")
print(f" backend calls: {backend.calls}")The script below runs all three checks against the first version. Save it as step03_pitfalls.py and run it.
# step03_pitfalls.py
import asyncio
from pitfalls import (
AwaitBeforeRegister,
check_cancellation,
check_failure,
check_register_race,
)
from single_flight_v1 import SingleFlightV1
async def main():
print("=== pitfall 1: an await between the check and the registration")
await check_register_race(SingleFlightV1)
await check_register_race(AwaitBeforeRegister)
print("=== pitfall 2: one caller gives up")
await check_cancellation(SingleFlightV1)
print("=== pitfall 3: the load fails once")
await check_failure(SingleFlightV1)
asyncio.run(main())=== pitfall 1: an await between the check and the registration
SingleFlightV1: 200 requests made 1 backend call(s)
AwaitBeforeRegister: 200 requests made 200 backend call(s)
=== pitfall 2: one caller gives up
SingleFlightV1: the impatient caller got TimeoutError
of 20 patient callers, 0 got a value and 20 were cancelled
backend calls: 1
the next request raised CancelledError
=== pitfall 3: the load fails once
SingleFlightV1: first request failed: backend refused 'home'
the backend is healthy again; calls so far: 1
second request failed too: backend refused 'home'
backend calls: 1What the output shows
Pitfall 1. The first version made one backend call for 200 requests; the version with one extra await made 200. All 200 tasks checked _inflight, found it empty, and then paused at the await before any of them had registered a task. The check and the registration must happen in one stretch of code with no await in it. Anything awaitable has to go before the check or after the registration.
Pitfall 2. The impatient caller got a TimeoutError, which is correct. The surprise is the 20 patient callers: none got a value, and all 20 were cancelled, even though only one caller gave up. Only one backend call was made, and it was cancelled with them. The reason is how asyncio cancels a task. This is the code in CPython 3.13.14 (Lib/asyncio/tasks.py), inside Task.cancel:
if self._fut_waiter is not None:
if self._fut_waiter.cancel(msg=msg):When a task is cancelled while it is waiting on another task, the cancellation is forwarded to the task it is waiting on. The impatient caller was waiting on the shared load task, so the shared load was cancelled, and every caller waiting on that same task received a CancelledError. The last line of the output adds a second problem: the next request also raised CancelledError, although nobody had cancelled it. The cancelled task was still registered in _inflight and nothing ever removed it, so the key was stuck.
Pitfall 3. The first request failed, which is expected. The backend then recovered, yet the second request failed with the very same error and the call count stayed at 1: the backend was never asked again. The cleanup line sits at the end of _load, after await self._loader(key). When the loader raises, the function exits early and the failed task stays in _inflight. Every later request joins that dead task and receives its stored exception. The key is poisoned until the process restarts.
Pitfalls 2 and 3 have one root cause, cleanup that is not guaranteed to run. Pitfall 2 also needs a second change, so that one caller’s cancellation cannot reach the shared task.
3.4 The fixed class
Here is the fixed version. Save it as single_flight.py; it replaces the first draft for the rest of the tutorial.
# single_flight.py
import asyncio
import time
from dataclasses import dataclass
@dataclass
class Entry:
value: object
expires_at: float
load_seconds: float = 0.0 # how long the load took (Step 6 uses this)
class SingleFlightCache:
def __init__(self, loader, *, ttl, clock=time.monotonic):
self._loader = loader
self._ttl = ttl
self._clock = clock
self._data = {}
self._inflight = {} # key -> the task that is loading it right now
def _ttl_for(self, key):
return self._ttl # a hook that Step 4 overrides
async def get(self, key):
entry = self._data.get(key)
if entry is not None and entry.expires_at > self._clock():
return entry.value
return await asyncio.shield(self._start_load(key))
def _start_load(self, key):
task = self._inflight.get(key)
if task is None:
task = asyncio.create_task(self._load(key))
self._inflight[key] = task
task.add_done_callback(lambda done: self._finish(key, done))
return task
async def _load(self, key):
started = self._clock()
value = await self._loader(key)
now = self._clock()
self._data[key] = Entry(value, now + self._ttl_for(key), now - started)
return value
def _finish(self, key, task):
if self._inflight.get(key) is task:
del self._inflight[key]
if not task.cancelled():
task.exception() # mark any error as seenEach change answers one pitfall:
getwraps the shared task inasyncio.shield. The Python documentation says that withshield, “except that if the coroutine containing it is cancelled, the Task running in something() is not cancelled. From the point of view of something(), the cancellation did not happen.” The caller that gave up still receives itsCancelledError, but the load carries on for everyone else._start_loadis an ordinary function with noawait, so checking for a running task and registering a new one cannot be interrupted. That fixes pitfall 1. The dictionary also doubles as the strong reference that the documentation asks you to keep: “The event loop only keeps weak references to tasks.”_finishis attached as a done callback, which asyncio calls when the task ends for any reason: success, failure or cancellation. It removes the task from_inflightonly if it is still the registered one, and that fixes pitfall 3. Its last two lines mark any error as seen. Without them asyncio logs “Task exception was never retrieved” when a background load fails and nobody was waiting for it. I checked: with those lines removed, a failed background refresh produced that message, and with them it did not.Entrynow also recordsload_seconds, how long the load took, which Step 6 uses, and_ttl_foris a small hook that Step 4 will override.
You might wonder why cleanup lives in a done callback and not in a try/finally inside _load. Python 3.12 added eager tasks. With asyncio.eager_task_factory, the documentation says, “coroutines begin execution synchronously during Task construction.” A loader that finishes without ever waiting would then be completely done, with its finally block already run, before _start_load stores the task. The finished task would sit in _inflight for good. I ran that variant under eager tasks: the second request after expiry returned the old value and the loader was never called again, while the callback version reloaded as it should. A test in Step 7 locks this in.
3.5 Check the fix
This script runs the same three checks against the fixed class. Save it as step03b_fixed.py and run it.
# step03b_fixed.py
import asyncio
from pitfalls import check_cancellation, check_failure, check_register_race
from single_flight import SingleFlightCache
async def main():
print("=== check 1: 200 requests, one load")
await check_register_race(SingleFlightCache)
print("=== check 2: one caller gives up")
await check_cancellation(SingleFlightCache)
print("=== check 3: the load fails once")
await check_failure(SingleFlightCache)
asyncio.run(main())=== check 1: 200 requests, one load
SingleFlightCache: 200 requests made 1 backend call(s)
=== check 2: one caller gives up
SingleFlightCache: the impatient caller got TimeoutError
of 20 patient callers, 20 got a value and 0 were cancelled
backend calls: 1
the next request got a value
=== check 3: the load fails once
SingleFlightCache: first request failed: backend refused 'home'
the backend is healthy again; calls so far: 1
second request: got a value
backend calls: 2Two hundred requests still make one backend call. The impatient caller still gets its TimeoutError, but all 20 patient callers get a value and the next request is served normally. After a failure the next request really calls the backend again, and the call count goes from 1 to 2. All three bugs are gone, and the failure case shows another property worth knowing: when a shared load fails, every caller waiting on it receives that one failure, so a struggling backend is asked once and not 200 times.
Step 4: Spread expiry out with TTL jitter
Single-flight merges requests for the same key. It cannot merge requests for different keys, and a real cache holds thousands of keys. They are often filled at the same moment, for example when a service restarts and warms its cache or when a bulk job loads a whole table, and often with the same TTL. Then they also expire at the same moment, and the backend receives thousands of different loads at once. Each one is for a different key, so single-flight can do nothing about it.
The cure is to give each entry a slightly different lifetime. Save this as jitter.py. The function returns the TTL plus or minus up to jitter of itself, so 0.1 means up to 10 percent either way. The subclass overrides the _ttl_for hook from Step 3, so every entry gets its own random lifetime when it is stored, and it refuses a jitter of 1 or more because that could produce a lifetime of zero or less.
# jitter.py
import random
from single_flight import SingleFlightCache
def jittered_ttl(ttl, jitter, rng=random):
"""ttl plus or minus up to `jitter` of itself (0.1 means 10 percent)."""
return ttl * rng.uniform(1.0 - jitter, 1.0 + jitter)
class JitteredCache(SingleFlightCache):
def __init__(self, loader, *, jitter=0.1, rng=random, **kwargs):
if not 0 <= jitter < 1:
raise ValueError("jitter must be at least 0 and below 1")
super().__init__(loader, **kwargs)
self._jitter = jitter
self._rng = rng
def _ttl_for(self, key):
return jittered_ttl(self._ttl, self._jitter, self._rng)4.1 Measure what jitter does
The simulation below models 5,000 keys that are all filled at second 0 with a TTL of 300 seconds. Each key is requested every second, so it is reloaded in the first second after it expires, and the reload gets a fresh jittered lifetime. It prints the busiest second in each of three 5-minute cycles, and the average load over a stretch of seconds, to show that the total work does not change. This is a model, not a measurement of a real system, but it is the same arithmetic your cache would perform.
# step04_jitter.py
import math
import random
from collections import Counter
from jitter import jittered_ttl
KEYS, TTL, HORIZON = 5000, 300, 1600 # seconds; every key is requested every second
def loads_per_second(jitter, seed=7):
rng = random.Random(seed)
loads = Counter()
for _ in range(KEYS):
expires = jittered_ttl(TTL, jitter, rng) # the cache was warmed at second 0
while expires < HORIZON:
second = math.ceil(expires) # the first request after expiry reloads the key
loads[second] += 1
expires = second + jittered_ttl(TTL, jitter, rng)
return loads
def busiest(loads, start, stop):
return max(loads[s] for s in range(start, stop))
print(f"{KEYS} keys warmed at second 0, ttl {TTL} s")
print("busiest second in each cycle, and the average load over seconds 600 to 1499")
print(f"{'jitter':>8}{'cycle 1':>9}{'cycle 2':>9}{'cycle 3':>9}{'average per second':>20}")
for jitter in (0.0, 0.05, 0.1, 0.2):
loads = loads_per_second(jitter)
average = sum(loads[s] for s in range(600, 1500)) / 900
print(
f"{jitter:>8.2f}{busiest(loads, 0, 450):>9}{busiest(loads, 450, 750):>9}"
f"{busiest(loads, 750, 1050):>9}{average:>20.1f}"
)5000 keys warmed at second 0, ttl 300 s
busiest second in each cycle, and the average load over seconds 600 to 1499
jitter cycle 1 cycle 2 cycle 3 average per second
0.00 5000 5000 5000 16.7
0.05 193 174 145 16.6
0.10 98 103 76 16.6
0.20 60 50 40 16.6Without jitter, all 5,000 keys come due in the same second, at 300, again at 600 and again at 900, and the backend sees 5,000 loads in one second. With 5 percent jitter the busiest second is about 190 loads in the first cycle, with 10 percent about 100, and with 20 percent about 60. By the third cycle the busiest second is lower than in the first cycle in every jittered row, because each reload adds a new random offset and the keys drift further apart. The last column is the point that surprises people: the average load is 16.7 loads per second without jitter and 16.6 with it. Jitter does not reduce the total work. It changes the shape of the work from a spike into a smooth stream.
It has a price, which is that entries can now live up to jitter longer than the TTL. With 10 percent jitter, a 300-second TTL can serve an answer up to 330 seconds old. Pick the smallest jitter that flattens your own peak.
Step 5: Hide the wait with stale-while-revalidate
Single-flight leaves one problem unsolved. Even with exactly one load running, the callers who arrive during that load still have to wait for it. The next idea comes from HTTP caching. RFC 5861 defines a stale-while-revalidate extension and describes it like this: “The stale-while-revalidate HTTP Cache-Control extension allows a cache to immediately return a stale response while it revalidates it in the background, thereby hiding latency (both in the network and on the server) from clients.” That RFC is informational; it says so itself: “This document is not an Internet Standards Track specification; it is published for informational purposes.” It describes HTTP caches, but the idea works just as well inside your program.
5.1 Write the class
Save this as swr.py. It extends the fixed single-flight class with a stale window, stale_for seconds after the TTL during which an expired entry may still be served. The get method now has three outcomes: a fresh entry is returned; an entry that is expired but inside the stale window is returned immediately while _start_load makes sure a refresh is running in the background; and if there is nothing usable, the caller waits for the load as before.
# swr.py
import asyncio
from single_flight import SingleFlightCache
class StaleWhileRevalidate(SingleFlightCache):
def __init__(self, loader, *, stale_for, **kwargs):
super().__init__(loader, **kwargs)
self._stale_for = stale_for
async def get(self, key):
entry = self._data.get(key)
now = self._clock()
if entry is not None:
if now < entry.expires_at:
return entry.value # fresh
if now < entry.expires_at + self._stale_for:
self._start_load(key) # refresh in the background
return entry.value # stale, but instant
return await asyncio.shield(self._start_load(key)) # nothing usable: waitBecause _start_load reuses the running task, a thousand requests during the stale window start exactly one refresh. Nobody waits for it, so nobody needs shield in the stale branch.
5.2 Watch it work
This script goes through the whole life of an entry with a 0.5-second TTL and a 2-second stale window: a cold start, a burst of 200 requests just after expiry, a refresh that succeeds, a refresh that fails, and a request after the stale window has closed.
# step05_swr.py
import asyncio
import time
from backend import SlowBackend
from swr import StaleWhileRevalidate
async def timed(coro):
started = time.monotonic()
value = await coro
return value, (time.monotonic() - started) * 1000
async def main():
backend = SlowBackend(latency=0.2, capacity=20)
cache = StaleWhileRevalidate(backend.fetch, ttl=0.5, stale_for=2.0)
first, ms = await timed(cache.get("home"))
print(f"cold start: waited {ms:.0f} ms, backend calls {backend.calls}")
await asyncio.sleep(0.6) # past the ttl, inside the stale window
backend.reset()
started = time.monotonic()
results = await asyncio.gather(*(cache.get("home") for _ in range(200)))
ms = (time.monotonic() - started) * 1000
print(f"burst of 200 after expiry: all returned in {ms:.0f} ms,"
f" all the old value: {all(r is first for r in results)},"
f" backend calls: {backend.calls}")
await asyncio.sleep(0.3) # the background refresh finishes
second, ms = await timed(cache.get("home"))
print(f"next request: waited {ms:.0f} ms, new value: {second['as_of'] > first['as_of']}")
await asyncio.sleep(0.6) # expired again, and this time the refresh will fail
backend.fail_next = 1
stale, ms = await timed(cache.get("home"))
await asyncio.sleep(0.05)
print(f"failed refresh: the caller still got the old value in {ms:.0f} ms: {stale is second}")
again, ms = await timed(cache.get("home"))
print(f"next request: old value again: {again is second} (it starts a retry in the background)")
await asyncio.sleep(0.3)
third, ms = await timed(cache.get("home"))
print(f"after the retry: new value: {third['as_of'] > second['as_of']}")
await asyncio.sleep(3.0) # beyond ttl + stale_for
value, ms = await timed(cache.get("home"))
print(f"after the stale window: waited {ms:.0f} ms (a hard miss)")
asyncio.run(main())cold start: waited 216 ms, backend calls 1
burst of 200 after expiry: all returned in 4 ms, all the old value: True, backend calls: 1
next request: waited 0 ms, new value: True
failed refresh: the caller still got the old value in 0 ms: True
next request: old value again: True (it starts a retry in the background)
after the retry: new value: True
after the stale window: waited 209 ms (a hard miss)The cold start has to wait for the load, because there is no old value to show. The burst of 200 requests after expiry all returned within a few milliseconds, every one of them received the old value, and the backend was called once, in the background. The next request got the new value without waiting. When the refresh failed, the caller still received the old value instantly, and the following request started another attempt in the background, which succeeded. This is a bonus you did not have to write: RFC 5861 also defines stale-if-error, which “allows a cache to return a stale response when an error” is encountered and, in the RFC’s words, “This improves availability.” Inside its stale window, this cache behaves the same way, because a failed refresh leaves the old entry in place. After the stale window closed, the last request waited for the full load again, which is called a hard miss.
The trade-off is freshness. Callers can now see data that is up to ttl + stale_for seconds old, so choose stale_for by asking how wrong an old answer is allowed to be, not by how long the load takes.
Step 6: Refresh early with XFetch
Stale-while-revalidate starts the refresh after the entry has expired. The VLDB paper’s approach starts it before. In Wikipedia’s summary of it: “each process may recompute the cache value before its expiration by making an independent probabilistic decision, where the probability of performing the early recomputation increases as we get closer to the expiration of the value.” The paper calls its algorithm XFetch, short for exponential fetch, and says that in its simplest form it “requires no parameter tuning to perform effectively.”
6.1 The rule
On every request for a fresh entry, draw a random number and decide whether to refresh now. The paper’s test is Time() - delta * beta * log(rand()) >= expiry. Here delta is how long the last load took, which is why Step 3 stored load_seconds. beta is a knob that defaults to 1. Wikipedia’s page explains that “The variable delta represents the time to recompute the value and is used to scale the probability distribution appropriately.” The logarithm of a random number between 0 and 1 is negative, so -delta * beta * log(rand()) is a positive random gap, usually small and occasionally large. A request refreshes the entry if that gap would carry it past the expiry time. A request long before the expiry rarely does; a request close to it usually does, and one at the expiry itself always does. The slower the load, the earlier the refreshes start.
6.2 Write the class
Save this as xfetch.py. Two details are worth a note. Python’s random.random() returns values from 0.0 up to but not including 1.0, so it can return exactly 0.0 and log(0) is an error; 1.0 - random() can never be 0, and has the same distribution. And I run the early refresh in the background with _start_load so that no caller waits for it, which differs from the paper’s pseudocode, where the request that wins the draw recomputes inline. The decision rule is the paper’s.
# xfetch.py
import asyncio
import math
import random
from single_flight import SingleFlightCache
class XFetchCache(SingleFlightCache):
def __init__(self, loader, *, beta=1.0, rand=random.random, **kwargs):
super().__init__(loader, **kwargs)
self._beta = beta
self._rand = rand
def should_refresh_early(self, entry, now):
# 1.0 - rand() is never 0.0, so the logarithm is always defined
gap = -entry.load_seconds * self._beta * math.log(1.0 - self._rand())
return now + gap >= entry.expires_at
async def get(self, key):
entry = self._data.get(key)
now = self._clock()
if entry is not None and now < entry.expires_at:
if self.should_refresh_early(entry, now):
self._start_load(key) # recompute early, in the background
return entry.value
return await asyncio.shield(self._start_load(key))6.3 Check the probabilities
This script asks the class’s decision method 100,000 times at several distances from the expiry, for a load that takes 0.1 seconds, and compares the share of “yes” answers with the formula exp(-t / delta), which is the chance that a random gap is at least t seconds long when beta is 1.
# step06_xfetch_table.py
import math
import random
from single_flight import Entry
from xfetch import XFetchCache
rng = random.Random(42)
cache = XFetchCache(None, ttl=2.0, beta=1.0, rand=rng.random) # no loader needed for this check
entry = Entry(value="x", expires_at=10.0, load_seconds=0.1)
print("load takes 0.1 s, beta = 1; 100,000 requests at each distance from expiry")
print(f"{'seconds before expiry':>22}{'start a refresh':>17}{'exp(-t/delta)':>15}")
for before in (0.5, 0.3, 0.2, 0.1, 0.05, 0.0):
starts = sum(cache.should_refresh_early(entry, 10.0 - before) for _ in range(100_000))
formula = math.exp(-before / 0.1)
print(f"{before:>22.2f}{starts / 1000:>16.1f}%{formula * 100:>14.1f}%")load takes 0.1 s, beta = 1; 100,000 requests at each distance from expiry
seconds before expiry start a refresh exp(-t/delta)
0.50 0.7% 0.7%
0.30 5.0% 5.0%
0.20 13.5% 13.5%
0.10 37.0% 36.8%
0.05 60.7% 60.7%
0.00 100.0% 100.0%Half a second before expiry, fewer than 1 request in 100 starts a refresh. At 0.2 seconds it is about 1 in 7, at 0.05 seconds about 3 in 5, and at the expiry itself every request does. The measured shares match the formula to within a few tenths of a percent. At a steady 100 requests per second, that ramp means a refresh almost always starts before the entry actually dies.
Step 7: Compare all four and combine them
7.1 Run the comparison
This harness sends 100 requests per second to one key for 20 seconds, with a 2-second TTL and a query that takes 0.1 seconds, and measures five strategies at the same time, each with its own backend. It reports the number of backend calls, the most queries running at once, the number of slow requests (any request that waited more than half the query time, 50 milliseconds), and the age of the oldest value any request received.
# step06_compare.py
import asyncio
import random
import sys
import time
from backend import SlowBackend
from naive import NaiveCache
from single_flight import SingleFlightCache
from swr import StaleWhileRevalidate
from xfetch import XFetchCache
RATE, SECONDS, TTL, LATENCY = 100, 20, 2.0, 0.1
STRATEGIES = {
"naive": lambda b: NaiveCache(b.fetch, ttl=TTL),
"single-flight": lambda b: SingleFlightCache(b.fetch, ttl=TTL),
"stale-while-revalidate": lambda b: StaleWhileRevalidate(b.fetch, ttl=TTL, stale_for=1.0),
"xfetch beta=1": lambda b: XFetchCache(b.fetch, ttl=TTL, beta=1.0),
"xfetch beta=2": lambda b: XFetchCache(b.fetch, ttl=TTL, beta=2.0),
}
async def one_request(cache, records):
started = time.monotonic()
value = await cache.get("home")
finished = time.monotonic()
records.append((finished - started, finished - value["as_of"]))
async def run(name, make):
backend = SlowBackend(latency=LATENCY, capacity=1000)
cache = make(backend)
await cache.get("home") # warm up, then count from zero
backend.reset()
records, tasks = [], []
start = time.monotonic()
for i in range(RATE * SECONDS):
delay = start + i / RATE - time.monotonic()
if delay > 0:
await asyncio.sleep(delay)
tasks.append(asyncio.create_task(one_request(cache, records)))
await asyncio.gather(*tasks)
slow = sum(waited > LATENCY / 2 for waited, _ in records)
oldest = max(age for _, age in records)
return name, backend.calls, backend.max_in_flight, slow, oldest
async def main():
random.seed(int(sys.argv[1]) if len(sys.argv) > 1 else 1)
rows = await asyncio.gather(*(run(name, make) for name, make in STRATEGIES.items()))
print(f"{RATE} requests/s for {SECONDS} s, ttl {TTL} s, one query takes {LATENCY} s")
print(f"{'strategy':<24}{'backend calls':>14}{'peak queries':>14}{'slow requests':>15}{'oldest value':>14}")
for name, calls, peak, slow, oldest in rows:
print(f"{name:<24}{calls:>14}{peak:>14}{slow:>15}{oldest:>13.2f}s")
asyncio.run(main())Save it as step06_compare.py and run python step06_compare.py 1. The argument is the random seed. It takes about 20 seconds.
100 requests/s for 20 s, ttl 2.0 s, one query takes 0.1 s
strategy backend calls peak queries slow requests oldest value
naive 103 12 103 2.00s
single-flight 9 1 57 2.00s
stale-while-revalidate 9 1 0 2.11s
xfetch beta=1 11 1 0 1.96s
xfetch beta=2 16 1 0 1.60sReading the table
The naive row is the stampede: 103 backend calls, up to 12 queries at once, and every one of those 103 requests was slow. The paper’s formula predicts it. A request rate of 100 per second multiplied by a 0.1-second load gives about 10 loads at every expiry, and the run had about 10 expiries. Single-flight cut the load to 9 calls and never more than one query at a time, but 57 requests were still slow, because those that arrived while the one load was running had to wait for it. Stale-while-revalidate kept the 9 calls and had no slow requests at all, at the cost of serving a value up to 2.11 seconds old when the TTL is 2.0. XFetch with beta 1 needed 11 calls, because it refreshes slightly early, had no slow requests, and kept every value it served within the TTL (the oldest was 1.96 seconds against a TTL of 2.0). XFetch with beta 2 started refreshes even earlier: 16 calls, no slow requests, and the freshest data of any row, at 1.60 seconds.
I ran the comparison with three different seeds. The naive cache made 100 to 103 calls each time, single-flight made 9 and had 57 to 60 slow requests, the stale-while-revalidate row had 0 slow requests every time, XFetch with beta 1 made 11 calls and had 0 slow requests every time, and XFetch with beta 2 made 15 or 16 calls. Expect small differences on your machine, since the timing depends on the operating system’s scheduler. XFetch is also a random rule, so an unlucky run could occasionally start a refresh too late and produce a few slow requests; none of my three runs did.
Choosing a strategy
Start with single-flight, since the other two ideas are built on it and it fixes the damage to the backend. Add jitter whenever many keys share a TTL. Then choose how to treat the callers who arrive during a refresh. Stale-while-revalidate gives them no wait at all and is deterministic, but it lets data go past its TTL. XFetch never serves an entry after its expiry time and also leaves nobody waiting, at the price of a few extra loads and a probabilistic rule. Wikipedia notes beta can be set above 1 to favor earlier recomputations, and the table shows the cost: more backend calls for more margin.
Neither of these helps the very first request for a key, or the first request after a long quiet period, because there is no old value to serve. Single-flight is what protects the backend in that case.
7.2 Combine the pieces
Because each idea overrides only one or two methods, they combine with ordinary Python inheritance. Save this as herdcache.py. When Python looks for a method it tries HerdCache first, then JitteredCache, then StaleWhileRevalidate, then SingleFlightCache. So the jittered lifetime comes from JitteredCache, get comes from StaleWhileRevalidate, and the shared loading comes from the base class. Each __init__ takes the arguments it knows and passes the rest up with **kwargs.
# herdcache.py
from jitter import JitteredCache
from swr import StaleWhileRevalidate
class HerdCache(JitteredCache, StaleWhileRevalidate):
"""One load per key at a time, jittered lifetimes, and stale-while-revalidate."""Using it looks like this: cache = HerdCache(load_report, ttl=60, jitter=0.1, stale_for=30), followed by await cache.get(report_id) wherever you used to call the loader directly.
7.3 Lock it in with tests
The test suite below needs no running backend. A FakeClock replaces time.monotonic, so a test can jump ahead 12 seconds instantly, and make_loader builds loaders that count their calls, can fail on chosen calls, and can advance the fake clock to pretend that a load took time. Each test is a normal function that runs one async scenario with asyncio.run. Save the three parts that follow as one file, test_stampede.py. First the helpers and the single-flight tests, including one for each pitfall from Step 3:
# test_stampede.py
import asyncio
import random
import sys
import pytest
from backend import SlowBackend
from herdcache import HerdCache
from jitter import JitteredCache
from naive import NaiveCache
from single_flight import SingleFlightCache
from swr import StaleWhileRevalidate
from xfetch import XFetchCache
class FakeClock:
def __init__(self):
self.now = 1000.0
def __call__(self):
return self.now
def advance(self, seconds):
self.now += seconds
def make_loader(clock=None, seconds=0.0, fail_on=()):
calls = []
async def loader(key):
calls.append(key)
number = len(calls)
await asyncio.sleep(0)
if clock is not None:
clock.advance(seconds)
if number in fail_on:
raise ValueError(f"load {number} failed")
return f"{key}-v{number}"
loader.calls = calls
return loader
def test_naive_cache_loads_once_per_caller():
async def scenario():
loader = make_loader()
cache = NaiveCache(loader, ttl=10)
await asyncio.gather(*(cache.get("a") for _ in range(50)))
assert len(loader.calls) == 50
asyncio.run(scenario())
def test_concurrent_misses_share_one_load():
async def scenario():
loader = make_loader()
cache = SingleFlightCache(loader, ttl=10)
results = await asyncio.gather(*(cache.get("a") for _ in range(100)))
assert len(loader.calls) == 1
assert set(results) == {"a-v1"}
asyncio.run(scenario())
def test_different_keys_load_independently():
async def scenario():
loader = make_loader()
cache = SingleFlightCache(loader, ttl=10)
await asyncio.gather(cache.get("a"), cache.get("b"), cache.get("a"))
assert sorted(loader.calls) == ["a", "b"]
asyncio.run(scenario())
def test_expired_entry_is_loaded_again():
async def scenario():
clock = FakeClock()
loader = make_loader()
cache = SingleFlightCache(loader, ttl=10, clock=clock)
assert await cache.get("a") == "a-v1"
clock.advance(9)
assert await cache.get("a") == "a-v1"
clock.advance(2)
assert await cache.get("a") == "a-v2"
asyncio.run(scenario())
def test_a_cancelled_caller_does_not_cancel_the_load():
async def scenario():
loader = make_loader()
cache = SingleFlightCache(loader, ttl=10)
first = asyncio.create_task(cache.get("a"))
second = asyncio.create_task(cache.get("a"))
await asyncio.sleep(0) # let both start waiting
first.cancel()
assert await second == "a-v1"
with pytest.raises(asyncio.CancelledError):
await first
assert len(loader.calls) == 1
asyncio.run(scenario())
def test_a_failure_reaches_every_waiter_after_one_attempt():
async def scenario():
loader = make_loader(fail_on=(1,))
cache = SingleFlightCache(loader, ttl=10)
results = await asyncio.gather(
*(cache.get("a") for _ in range(10)), return_exceptions=True
)
assert all(isinstance(r, ValueError) for r in results)
assert len(loader.calls) == 1
asyncio.run(scenario())
def test_a_failure_is_not_remembered():
async def scenario():
loader = make_loader(fail_on=(1,))
cache = SingleFlightCache(loader, ttl=10)
with pytest.raises(ValueError):
await cache.get("a")
assert await cache.get("a") == "a-v2"
asyncio.run(scenario())
@pytest.mark.skipif(sys.version_info < (3, 12), reason="eager tasks need Python 3.12")
def test_works_when_tasks_start_eagerly():
async def scenario():
asyncio.get_running_loop().set_task_factory(asyncio.eager_task_factory)
clock = FakeClock()
calls = []
async def loader(key): # finishes without ever suspending
calls.append(key)
return "x"
cache = SingleFlightCache(loader, ttl=1, clock=clock)
assert await cache.get("a") == "x"
await asyncio.sleep(0)
clock.advance(5)
assert await cache.get("a") == "x"
assert len(calls) == 2
asyncio.run(scenario())Next the tests for jitter and for stale-while-revalidate. The last of them checks that a failed refresh keeps the old value and that the following request retries:
# test_stampede.py (continued)
def test_jitter_stays_inside_its_band():
cache = JitteredCache(make_loader(), ttl=100, jitter=0.1, rng=random.Random(5))
lifetimes = [cache._ttl_for("k") for _ in range(2000)]
assert min(lifetimes) >= 90 and max(lifetimes) <= 110
assert max(lifetimes) - min(lifetimes) > 15
def test_zero_jitter_is_exact():
cache = JitteredCache(make_loader(), ttl=100, jitter=0.0)
assert {cache._ttl_for("k") for _ in range(50)} == {100.0}
def test_jitter_of_one_or_more_is_refused():
with pytest.raises(ValueError):
JitteredCache(make_loader(), ttl=100, jitter=1.0)
def test_stale_entry_is_served_at_once_and_refreshed_once():
async def scenario():
clock = FakeClock()
loader = make_loader()
cache = StaleWhileRevalidate(loader, ttl=10, stale_for=5, clock=clock)
assert await cache.get("a") == "a-v1"
clock.advance(12) # expired, inside the stale window
results = await asyncio.gather(*(cache.get("a") for _ in range(50)))
assert set(results) == {"a-v1"} # nobody waited for the new value
await asyncio.sleep(0.01) # let the background refresh finish
assert len(loader.calls) == 2
assert await cache.get("a") == "a-v2"
asyncio.run(scenario())
def test_a_hard_miss_waits_after_the_stale_window():
async def scenario():
clock = FakeClock()
loader = make_loader()
cache = StaleWhileRevalidate(loader, ttl=10, stale_for=5, clock=clock)
assert await cache.get("a") == "a-v1"
clock.advance(16) # past ttl + stale_for
assert await cache.get("a") == "a-v2"
asyncio.run(scenario())
def test_a_failed_refresh_keeps_serving_the_stale_value():
async def scenario():
clock = FakeClock()
loader = make_loader(fail_on=(2,))
cache = StaleWhileRevalidate(loader, ttl=10, stale_for=5, clock=clock)
assert await cache.get("a") == "a-v1"
clock.advance(12)
assert await cache.get("a") == "a-v1" # the refresh starts and fails
await asyncio.sleep(0.01)
assert await cache.get("a") == "a-v1" # still stale; this call retries
await asyncio.sleep(0.01)
assert await cache.get("a") == "a-v3"
assert len(loader.calls) == 3
asyncio.run(scenario())Finally XFetch and the combined class. The XFetch tests replace the random source with a function that always returns the same number, so the “dice” are loaded and the outcome is predictable:
# test_stampede.py (continued)
def test_xfetch_refreshes_before_expiry_when_the_dice_say_so():
async def scenario():
clock = FakeClock()
loader = make_loader(clock=clock, seconds=0.5) # each load "takes" 0.5 s
cache = XFetchCache(loader, ttl=10, beta=1.0, rand=lambda: 0.999999, clock=clock)
assert await cache.get("a") == "a-v1"
clock.advance(4) # the early gap is about 6.9 s, so we are inside it
assert await cache.get("a") == "a-v1" # still fresh, served at once
await asyncio.sleep(0.01)
assert len(loader.calls) == 2
assert await cache.get("a") == "a-v2"
asyncio.run(scenario())
def test_xfetch_waits_for_expiry_when_the_dice_say_never():
async def scenario():
clock = FakeClock()
loader = make_loader(clock=clock, seconds=0.5)
cache = XFetchCache(loader, ttl=10, beta=1.0, rand=lambda: 0.0, clock=clock)
assert await cache.get("a") == "a-v1"
clock.advance(9)
assert await cache.get("a") == "a-v1"
assert len(loader.calls) == 1
clock.advance(2)
assert await cache.get("a") == "a-v2"
asyncio.run(scenario())
def test_the_combined_cache_uses_all_three_ideas():
async def scenario():
clock = FakeClock()
loader = make_loader()
cache = HerdCache(
loader, ttl=10, jitter=0.1, stale_for=5, rng=random.Random(1), clock=clock
)
assert await cache.get("a") == "a-v1"
lifetime = cache._data["a"].expires_at - clock.now
assert 9 <= lifetime <= 11
clock.advance(11.5) # past every possible lifetime, inside the stale window
assert await cache.get("a") == "a-v1"
await asyncio.sleep(0.01)
assert len(loader.calls) == 2
asyncio.run(scenario())
def test_slow_backend_refuses_work_beyond_its_capacity():
async def scenario():
backend = SlowBackend(latency=0.05, capacity=3)
results = await asyncio.gather(
*(backend.fetch("a") for _ in range(10)), return_exceptions=True
)
assert sum(isinstance(r, Exception) for r in results) == 7
asyncio.run(scenario())Run python -m pytest -q:
.................. [100%]
18 passed in 0.30sStep 8: What about threads, and why not just functools.lru_cache?
Many Python services are not asyncio programs. They run on threads, with Flask or Django workers for example, and then the first thing people reach for is functools.lru_cache. It is a good cache, and my LRU cache tutorial builds one, but it does not stop stampedes. The Python documentation says: “It is possible for the wrapped function to be called more than once if another thread makes an additional call before the initial call has been completed and cached.” And: “Call-once behavior is not guaranteed because locks are not held during the function call.”
8.1 Measure it
The script below starts 20 threads at the same instant, using a Barrier, and calls the same function from all of them. It also contains a thread-based single-flight cache: a lock for each key, and a second look at the cache after the lock is acquired, because another thread may have loaded the value while this one waited. It runs three experiments: lru_cache, the lock version with a healthy backend, and the lock version with a backend that is down.
# step08_threads.py
import threading
import time
from functools import lru_cache
from single_flight import Entry
THREADS = 20
class Counter:
def __init__(self):
self.calls = 0
self._lock = threading.Lock()
def hit(self):
with self._lock:
self.calls += 1
def run_threads(fn):
barrier = threading.Barrier(THREADS)
def worker():
barrier.wait() # release all threads at the same moment
try:
fn("home")
except RuntimeError:
pass
threads = [threading.Thread(target=worker) for _ in range(THREADS)]
for t in threads:
t.start()
for t in threads:
t.join()
class ThreadSingleFlight:
def __init__(self, loader, *, ttl):
self._loader = loader
self._ttl = ttl
self._data = {}
self._locks = {}
self._guard = threading.Lock()
def _lock_for(self, key):
with self._guard:
return self._locks.setdefault(key, threading.Lock())
def _fresh(self, key):
entry = self._data.get(key)
if entry is not None and entry.expires_at > time.monotonic():
return entry
return None
def get(self, key):
entry = self._fresh(key)
if entry is not None:
return entry.value
with self._lock_for(key):
entry = self._fresh(key) # someone may have loaded it while we waited
if entry is not None:
return entry.value
value = self._loader(key)
self._data[key] = Entry(value, time.monotonic() + self._ttl)
return value
def demo_lru_cache():
counter = Counter()
@lru_cache(maxsize=None)
def load(key):
counter.hit()
time.sleep(0.2)
return key.upper()
run_threads(load)
print(f"lru_cache: {THREADS} threads, the function ran {counter.calls} times")
def demo_lock(fail):
counter = Counter()
def loader(key):
counter.hit()
time.sleep(0.2)
if fail:
raise RuntimeError("backend down")
return key.upper()
run_threads(ThreadSingleFlight(loader, ttl=5).get)
label = "lock, backend down:" if fail else "lock, backend up: "
print(f"{label} {THREADS} threads, the loader ran {counter.calls} times")
demo_lru_cache()
demo_lock(fail=False)
demo_lock(fail=True)lru_cache: 20 threads, the function ran 20 times
lock, backend up: 20 threads, the loader ran 1 times
lock, backend down: 20 threads, the loader ran 20 timeslru_cache ran the function 20 times, once per thread, exactly as the documentation warns. The lock version ran the loader once. With the backend down, though, it ran the loader 20 times. The threads take the lock one after another, each finds nothing cached, and each tries and fails on its own, so a struggling backend still receives 20 attempts, one at a time. The asyncio class from Step 3 does better here, because all the callers waiting on a shared load receive that one failure. A test in Step 7 checks that behavior.
8.2 A trap with async functions
Some people try to put lru_cache on an async def function. It caches the coroutine object and not its result, and a coroutine can be awaited only once:
# step08b_async_lru.py
import asyncio
from functools import lru_cache
@lru_cache(maxsize=None)
async def load(key):
await asyncio.sleep(0.01)
return key.upper()
async def main():
print("first call:", await load("home"))
try:
print("second call:", await load("home"))
except RuntimeError as exc:
print("second call raised RuntimeError:", exc)
asyncio.run(main())first call: HOME
second call raised RuntimeError: cannot reuse already awaited coroutineThe first call works and the second raises RuntimeError. If you need single-flight for asyncio code, use a class like the one you built here.
Check that everything works end to end
From the herd folder with the virtual environment active, run every script in order. The comments show what you should see; counts in the comparison will differ by a few on your machine.
python step01_stampede.py # 200 backend calls, 180 failed requests
python step02_coalesce.py # 1 backend call, 0 failures
python step03_pitfalls.py # 200 calls, cancelled callers, a poisoned key
python step03b_fixed.py # 1 call, 20 of 20 values, second request works
python step04_jitter.py # busiest second falls from 5000 to about 100
python step05_swr.py # the burst returns in a few milliseconds
python step06_xfetch_table.py # shares match exp(-t / delta)
python step06_compare.py 1 # the comparison table, about 20 seconds
python step08_threads.py # 20, 1 and 20 loader runs
python step08b_async_lru.py # RuntimeError on the second call
python -m pytest -q # 18 passedIf step03b_fixed.py reports 200 calls in its first check, you probably left an await in _start_load. If a script cannot find backend or single_flight, you are running it from a different folder than the one that holds the files.
Where this stops working, and what to learn next
The cache you built is small on purpose, so here are the limits that matter before you rely on one like it.
- It lives in one process. Single-flight here protects the backend from the callers of one Python process. If you run eight worker processes, each has its own
_inflight, so up to eight loads per key can still run at once. To get one load across all processes you need a shared lock or an external process that recomputes the value. Wikipedia lists both and is honest about the cost of the lock: “the main drawback is a correct implementation of the locking mechanism which also takes care of edge cases including failure of the process acquiring the lock, tuning of a time-to-live for the lock, race-conditions, and so on.” - It never evicts. Entries stay in
_datauntil they are overwritten, so a cache over millions of distinct keys grows without bound. Combine it with a size limit, as in the LRU cache tutorial. - Failures are retried immediately. After a failed load, the next request starts a new attempt at once. One attempt at a time per key is far better than 200, but a backend that is down still receives a steady stream of attempts. Add a delay between attempts with exponential backoff and jitter, or stop calling a failing backend for a while with a circuit breaker.
- Background errors are silent. The class marks failed refreshes as seen so asyncio does not complain, but it does not log them either. In production, log or count them inside
_finish. - It does not limit load. It reduces duplicate work, but a backend can still be overwhelmed by many different keys. The backpressure and load shedding tutorial and the bulkhead tutorial cover that side.
- Cache nodes can cause stampedes too. When a cache server joins or leaves a pool, a large share of keys can move and miss at once. That is a different cause of the same symptom, and the consistent hashing tutorial shows how to limit it.
To keep learning, read the XFetch paper for the analysis behind the formula, the asyncio task documentation for shield, cancellation and task references, and RFC 5861 if you want to see the same ideas applied to HTTP caches. If your “cache” revalidates with an HTTP server, the ETag and If-Match tutorial shows how conditional requests make a revalidation cheap. You now have a working cache, a test suite that checks each behavior you built, and a way to measure whether the next defense you add is worth its cost.
No Comment! Be the first one.