Как гарантировать единственную загрузку в asyncio без дубликатов

Как в asyncio избежать дублирующей загрузки: double-checked locking, проверка под замком и пример кода для корутин без гонок и лишней работы. На практике.

Asyncio: как гарантировать единственную загрузку без дублирования работы

Когда сервис отслеживает метку времени в Redis и при её изменении обновляет данные в памяти, задача проста: в каждый момент загрузку должна выполнять только одна корутина. Загвоздка возникает, когда несколько корутин одновременно приходят к одной и той же точке await и поочерёдно проделывают одну и ту же работу. Исправление минимальное, но принципиально важное.

Постановка задачи

Приведённый сниппет опрашивает хранилище за последней меткой времени и, если она отличается от хранящейся в памяти, запускает загрузку. Предполагается, что загрузку выполняет только одна корутина, а остальные либо ждут, либо временно используют устаревшие данные.

import asyncio
class SyncAgent:
    def __init__(self):
        self.gate = asyncio.Lock()
        self.mark = None
    async def _pull_mark_from_store(self):
        # запрос к БД
        ...
    async def apply_if_needed(self):
        latest = await self._pull_mark_from_store()
        if self.mark == latest:
            return
        if self.gate.locked():
            return  # загрузка уже идёт; сейчас допустимо использовать имеющиеся данные
        async with self.gate:
            # выполнить загрузку
            self.mark = latest

Что идёт не так и почему

Проверка того, занят ли замок, выполняется до входа в критическую секцию. Несколько корутин могут увидеть, что замок свободен, и одновременно дойти до строки async with gate. После этого они войдут внутрь по очереди и повторят загрузку. Корневая причина — гонка между предварительными проверками и захватом замка. Сравниваемое значение может измениться во время паузы на await, поэтому одной проверки снаружи замка недостаточно.

Исправление: двойная проверка под замком

Рабочий паттерн здесь — double-checked locking: сначала сделать быструю предварительную проверку, чтобы не брать замок, когда ничего не изменилось, а затем повторить проверку после получения замка. Так только первая корутина, заметившая изменения, действительно выполнит загрузку; остальные придут позже, перепроверят и увидят, что делать больше нечего.

Перенести проверку внутрь замка — правильно.

import asyncio
class SyncAgent:
    def __init__(self):
        self.gate = asyncio.Lock()
        self.mark = None
    async def _pull_mark_from_store(self):
        ...
    async def apply_if_needed(self):
        latest = await self._pull_mark_from_store()
        if self.mark == latest:
            return
        async with self.gate:
            if self.mark == latest:
                return
            # выполнить загрузку
            self.mark = latest

Так сохраняется желаемое поведение. Внешняя проверка избавляет от лишнего захвата замка, когда изменений нет. Внутренняя проверка гарантирует корректность, когда несколько корутин сходятся к одной и той же точке await.

Необязательный быстрый отказ, если загрузка уже идёт

Иногда вместо ожидания захвата замка лучше сразу выйти, если загрузка уже запущена. Такой подход избавляет от очереди на замке, но по-прежнему опирается на внутреннюю проверку для корректности. Он уместен в asyncio, где конкуренция строится вокруг конкретных точек await, и не предназначен для многопоточной конкуренции.

import asyncio
class SyncAgent:
    def __init__(self):
        self.gate = asyncio.Lock()
        self.mark = None
    async def _pull_mark_from_store(self):
        ...
    async def apply_if_needed(self):
        latest = await self._pull_mark_from_store()
        if self.mark == latest:
            return
        if self.gate.locked():
            return  # не ждать; другая корутина уже загружает
        async with self.gate:
            if self.mark == latest:
                return
            # выполнить загрузку
            self.mark = latest

Почему это важно

Без внутренней проверки несколько корутин могут выполнить одну и ту же дорогую загрузку. Двойная проверка под замком сохраняет код простым и не допускает лишней работы. Этот паттерн здесь уместен, потому что состояние может меняться между точками await, а вторая проверка закрывает окно гонки.

Итоги

Координируя обновления между корутинами в asyncio, полагайтесь на замок, который защищает и работу, и проверку. Сначала проверяйте снаружи, чтобы не захватывать замок без нужды, затем повторяйте проверку внутри, чтобы продолжила только одна корутина. Если не хотите ждать в очереди на замок, допустимо сразу возвращаться, когда он уже удерживается, но внутреннюю проверку оставляйте. Так вы исключите дублирующие загрузки и сделаете путь обновления предсказуемым.