Кэш и одновременные промахи

Решение

import asyncio


class SingleFlightCache:
    def __init__(self, fetch):
        self._fetch = fetch
        self._values = {}
        self._in_flight = {}      # ключ -> задача, которая его вычисляет

    async def get(self, key):
        if key in self._values:
            return self._values[key]

        task = self._in_flight.get(key)
        if task is None:
            # создаём задачу синхронно: до первого await никто не вклинится
            task = asyncio.create_task(self._load(key))
            self._in_flight[key] = task

        return await asyncio.shield(task)

    async def _load(self, key):
        try:
            value = await self._fetch(key)
            self._values[key] = value
            return value
        finally:
            self._in_flight.pop(key, None)

Почему так

Почему обычной проверки «нет в кэше — сходи в базу» недостаточно

• Между промахом и записью в кэш есть await — за это время сто корутин успеют промахнуться и пойти в базу
• Это и называется cache stampede: популярный ключ протух, и вся нагрузка разом ушла в базу
• Особенно больно при рестарте: кэш пуст, а трафик прежний

Почему задача в словаре, а не Lock на ключ

• Задача — это и блокировка, и место для результата: ожидающим достаточно await по ней
• С Lock после его освобождения все ждавшие пошли бы проверять кэш заново — лишний код и лишние гонки
• Словарь блокировок ещё и надо чистить, иначе он растёт по числу ключей

Почему create_task, а потом shield

• create_task между проверкой и записью в словарь выполняется без await — вклиниться туда невозможно, поэтому гонки нет
• shield защищает общую задачу от отмены: если один из ожидающих отвалится по таймауту, остальные не потеряют вычисление
• finally снимает ключ из in_flight даже при ошибке — иначе следующий запрос вечно ждал бы мёртвую задачу