How to Implement Probabilistic Early Expiration in Python
A Python mock of probabilistic early expiration for caches, using a heap-based expiry queue and random eviction to approximate cache stampede protection.
Python code
54 linesimport heapq
import random
import time
class ProbabilisticEarlyExpirationMock:
def __init__(self, capacity=1024, expiration_probability=0.1):
self.capacity = capacity
self.expiration_probability = expiration_probability
self._items = {}
self._expiry_heap = []
self._next_id = 0
def put(self, value):
if len(self._items) >= self.capacity:
self._evict_expired()
if len(self._items) >= self.capacity:
self._evict_random()
item_id = self._next_id
self._next_id += 1
expires_at = time.monotonic() + random.uniform(0.1, 1.0)
self._items[item_id] = (value, expires_at)
heapq.heappush(self._expiry_heap, (expires_at, item_id))
return item_id
def get(self, item_id):
if item_id not in self._items:
return None
value, expires_at = self._items[item_id]
if time.monotonic() >= expires_at:
del self._items[item_id]
return None
if random.random() < self.expiration_probability:
del self._items[item_id]
return None
return value
def _evict_expired(self):
now = time.monotonic()
while self._expiry_heap and self._expiry_heap[0][0] <= now:
_, item_id = heapq.heappop(self._expiry_heap)
self._items.pop(item_id, None)
def _evict_random(self):
if not self._items:
return
item_id = random.choice(list(self._items.keys()))
del self._items[item_id]
if __name__ == "__main__":
cache = ProbabilisticEarlyExpirationMock(capacity=5, expiration_probability=0.5)
ids = [cache.put(i) for i in range(5)]
print([cache.get(i) for i in ids])
Output
[0, 1, 2, 3, 4] or [0, 1, None, 3, 4] (probabilistic removal can drop an item before its nominal expiry)
How it works
This mock simulates the 'probabilistic early expiration' strategy Redis uses to avoid cache stampedes. Items are given a random expiry between 0.1 and 1.0 seconds, and get has an expiration_probability chance of treating an item as stale even before its real expiry. put maintains a heap of expirations for O(log n) eviction of truly expired items, and falls back to random eviction only when the capacity is full. The time.monotonic() calls are immune to system clock changes, so the mock behaves predictably in tests.
Common mistakes
- Using `random.choice(list(self._items.keys()))` on large caches creates a full copy of keys, hurting performance.
- Forgetting to pop expired entries from the heap in `get`, leaving stale heap entries that slow future evictions.
- Using `time.time()` instead of `time.monotonic()` — wall-clock changes (NTP, DST) break expiry logic.
Variations
- Replace random expiration with a fixed probability constant to make behavior deterministic in unit tests.
- Use a `sortedcontainers.SortedList` instead of a heap for expiry management if you need to delete arbitrary entries.
Real-world use cases
- Protecting a database-backed cache from stampede when 1000s of requests miss the same hot key at once.
- Simulating Redis' `PROBABILISTIC_EARLY_EXPIRATION` behavior in an offline load-test harness.
- Teaching cache invalidation strategies in infrastructure workshops with a lightweight, dependency-free model.
Sponsored
More from Caching & Redis
- Cache Asides in Python with a Read-Through Loader easy
- Cache Data in Redis with Python easy
- Cache Penetration Null Object Mock in Python medium
- Cache Stampede Prevention with SingleFlight in Python medium
- Cache Warming with Python: Preload Hot Keys easy
- Coalescing duplicate in-flight requests: one shared result for concurrent callers hard
Keep learning
Related tutorials and quizzes for this topic.