Примитивы синхронизации в Python Asyncio: Исчерпывающее руководство

При написании приложений с несколькими потоками или процессами нужно помнить о возможности состояния гонки при использовании неатомарных операций. Даже простая задача для увеличения целого числа на единицу в конкурентной программе может вызвать ошибки, с трудом поддающиеся воспроизведению. Но при использовании asyncio мы всегда работаем в одном потоке (если только явно не задействовали средства многопоточной или многопроцессной обработки), значит, можно не беспокоиться о гонках, правда? На самом деле все не так просто!

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

В этой статье вы узнаете, зачем нужны примитивы синхронизации asyncio, а также о лучших практиках использования нескольких примитивов синхронизации. В конце статьи рассмотрим практический пример использования примитивов синхронизации в действии.

Зачем нужны примитивы синхронизации в asyncio?

Любой, кто использовал многопоточность в Python, знает, что несколько потоков используют один и тот же блок памяти. Поэтому, когда несколько потоков одновременно выполняют неатомарные операции над одной и той же областью, возникает проблема потокобезопасности. Поскольку asyncio работает в один поток, не возникает ли у него аналогичных проблем с потокобезопасностью? Ответ: нет.

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

Чтобы избежать подобных ошибок, в Python asyncio введен примитив синхронизации, аналогичный многопоточности. Это asyncio.Lock, asyncio.Semaphore, asyncio.Event и asyncio.Condition. Кроме того, чтобы избежать одновременного обращения к ресурсу слишком большого числа задач, примитивы синхронизации asyncio позволяют защитить ресурс, ограничив число одновременно обращающихся к нему задач.

Примитивы синхронизации в asyncio

asyncio.Lock

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

import asyncio
import aiohttp

cache = dict()


async def request_remote():
    print("Will request the website to get status.")
    async with aiohttp.ClientSession() as session:
        response = await session.get("https://www.example.com")
        return response.status


async def get_value(key: str):
    if key not in cache:
        print(f"The value of key {key} is not in cache.")
        value = await request_remote()
        cache[key] = value
    else:
        print(f"The value of key {key} is already in cache.")
        value = cache[key]
    print(f"The value of {key} is {value}")
    return value


async def main():
    task_one = asyncio.create_task(get_value("status"))
    task_two = asyncio.create_task(get_value("status"))

    await asyncio.gather(task_one, task_two)


if __name__ == "__main__":
    asyncio.run(main())

Это не соответствует нашему первоначальному замыслу, поэтому на помощь приходит asyncio.Lock. Мы можем проверить наличие данных в кэше, когда параллельные задачи должны получить блокировку первыми, а другие задачи, не получившие блокировку, будут ждать.

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

Вся блок-схема показана ниже:

Давайте поправим наш код:

import asyncio
from asyncio import Lock

import aiohttp

cache = dict()
lock = Lock()


async def request_remote():
    print("Will request the website to get status.")
    async with aiohttp.ClientSession() as session:
        response = await session.get("https://www.example.com")
        return response.status


async def get_value(key: str):
    async with lock:
        if key not in cache:
            print(f"The value of key {key} is not in cache.")
            value = await request_remote()
            cache[key] = value
        else:
            print(f"The value of key {key} is already in cache.")
            value = cache[key]
        print(f"The value of {key} is {value}")
        return value


async def main():
    task_one = asyncio.create_task(get_value("status"))
    task_two = asyncio.create_task(get_value("status"))

    await asyncio.gather(task_one, task_two)


if __name__ == "__main__":
    asyncio.run(main())

asyncio.Semaphore

Иногда нам необходимо получить доступ к ресурсу с ограниченным количеством одновременных запросов. Например, конкретная база данных допускает одновременное открытие только пяти соединений. Или, в зависимости от типа подписки, веб-интерфейс поддерживает только определенное количество одновременных запросов. В этом случае необходимо использовать asyncio.Semaphore. asyncio.Semaphore использует внутренний счетчик, который уменьшается на единицу при каждом получении блокировки Semaphore, пока не достигнет нуля.

Когда счетчик asyncio.Semaphore равен нулю, другие задачи, которым нужна блокировка, будут ждать. При вызове метода release после выполнения других задач счетчик будет увеличен на единицу. Ожидающие задачи могут продолжить выполнение.

Пример кода выглядит следующим образом:

import asyncio
from asyncio import Semaphore
from aiohttp import ClientSession


async def get_url(url: str, session: ClientSession, semaphore: Semaphore):
    print('Waiting to acquire semaphore...')
    async with semaphore:
        print('Semaphore acquired, requesting...')
        response = await session.get(url)
        print('Finishing requesting')
        return response.status


async def main():
    # Хотя мы запускаем 1000 задач, одновременно будут выполняться только 10 задач.
    semaphore: Semaphore = Semaphore(10)
    async with ClientSession() as session:
        tasks = [asyncio.create_task(get_url("https://www.example.com", session, semaphore))
                 for _ in range(1000)]
        await asyncio.gather(*tasks)


if __name__ == "__main__":
    asyncio.run(main())

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

asyncio.BoundedSemaphore

Одна из особенностей семафоров заключается в том, что число вызовов метода release может превышать число вызовов acquire. Если мы всегда используем семафоры в сочетании с блоком async with, то такое невозможно, потому что с каждым acquire автоматически
связывается release. Но если нам требуется более точный контроль над механизмом захвата и освобождения, то возможны проблемы.

Что произойдет, если мы случайно вызовем метод release() несколько раз?

import asyncio
from asyncio import Semaphore


async def acquire(semaphore: Semaphore):
    print("acquire: Ожидание возможности захвата")
    async with semaphore:
        print("acquire: Захвачен...")
        await asyncio.sleep(5)
    print("acquire: Освобождается...")


async def release(semaphore: Semaphore):
    print("release: Одиночное освобождение...")
    semaphore.release()
    print("release: Одиночное освобождение - готово!")


async def main():
    semaphore = Semaphore(2)
    print("Два захвата, три освобождения...")
    await asyncio.gather(asyncio.create_task(acquire(semaphore)),
                         asyncio.create_task(acquire(semaphore)),
                         asyncio.create_task(release(semaphore)))
    print("Три захвата...")
    await asyncio.gather(asyncio.create_task(acquire(semaphore)),
                         asyncio.create_task(acquire(semaphore)),
                         asyncio.create_task(acquire(semaphore)))


if __name__ == "__main__":
    asyncio.run(main())

Здесь мы создаем семафор с пределом 2. Затем дважды вызываем сопрограмму acquire и один раз release, т.е. всего семафор будет освобождён трижды. Первое обращение к gather завершается, по видимости, нормально:

Однако при втором обращении, когда мы захватываем семафор три раза, возникают проблемы – все три захвата происходят сразу! Мы непреднамеренно превысили предел семафора:

Как видно из кода, мы ограничены одновременным выполнением двух задач, но поскольку мы вызвали release более одного раза, то в следующий раз мы сможем выполнить три задачи одновременно. Для решения этой проблемы мы можем использовать asyncio.BoundedSemaphore. Bounded Semaphore - это версия Semaphore, которая вызывает ошибку ValueError в функции release(), если внутренний счетчик увеличивается выше начального значения.

Как мы знаем из исходного кода, при вызове release выдается ошибка ValueError, если значение счетчика больше значения, установленного при инициализации:

import asyncio
from asyncio import BoundedSemaphore


async def main():
    semaphore = BoundedSemaphore(2)

    await semaphore.acquire()
    semaphore.release()
    semaphore.release()


if __name__ == "__main__":
    asyncio.run(main())

Здесь второй вызов release возбудит исключение ValueError, означающее, что мы освободили семафор слишком много раз. Аналогичный результат будет иметь место, если в предыдущем примере использовать BoundedSemaphore вместо Semaphore. Если вы вызываете acquire и release вручную, так что возникает опасность динамически превысить
предел семафора, то лучше работать с BoundedSemaphore, потому что возникшее исключение предупредит об ошибке.

asyncio.Event

Event поддерживает внутреннюю булеву переменную в качестве флага. asyncio.Event имеет три общих метода: wait, set и clear.

Когда задача добегает до event.wait(), она находится в состоянии ожидания. В этот момент можно вызвать event.set(), чтобы установить внутренний маркер в True, и все ожидающие задачи могут продолжить выполнение.

Когда задача завершится, необходимо вызвать метод event.clear(), чтобы сбросить значение маркера в False, вернуть событие в исходное состояние, и можно продолжать использовать событие в следующий раз.

В конце статьи вместо примера кода я покажу, как использовать Event для реализации шины событий.

asyncio.Condition

asyncio.Event хороши, когда нужно просто уведомить о том, что произошло нечто, но бывают ситуации посложнее. Допустим, что по событию требуется получить доступ к разделяемому ресурсу, т.е. захватить блокировку. Или что перед продолжением работы нужно дождаться более сложного сочетания условий, чем простое событие. Или что нужно разбудить не все задачи, а только определенное число. Во всех этих случаях могут выручить asyncio.Condition.

asyncio.Condition похож на asyncio.Lock и asyncio.Event вместе взятые. Сначала мы используем async with, чтобы обеспечить получение блокировки условия, а затем вызываем condition.wait(), чтобы освободить блокировку условия и заставить задачу временно подождать.

Когда condition.wait() проходит, мы возвращаем блокировку условия, чтобы гарантировать одновременное выполнение только одной задачи. Пока задача временно освобождает блокировку и переходит в состояние ожидания по методу condition.wait(), другая задача может либо асинхронизироваться с блокировкой условия и уведомить все ожидающие задачи о необходимости продолжить выполнение по методу condition.notify_all().

Блок-схема показана ниже:

Мы можем продемонстрировать действие asyncio.Condition на примере кода:

import asyncio
from asyncio import Condition


async def do_work(condition: Condition):
    print("do_work: захватываю блокировку условия...")
    async with condition:
        print("do_work: Блокировка захвачена, освобождаю и жду выполнения условия...")
        await condition.wait()
        print("do_work: Условие выполнено, вновь захватываю блокировку и начинаю работать...")
        await asyncio.sleep(1)
        print("do_work: Работа закончена, блокировка освобождена.")

        
async def fire_event(condition: Condition):
    await asyncio.sleep(5)
    print("fire_event: захватываю блокировку условия...")
    async with condition:
        print("fire_event: Блокировка захвачена, уведомляю всех исполнителей")
        condition.notify_all()
        print("fire_event: Исполнители уведомлены, освобождаю блокировку.")

        
async def main():
    condition = Condition()
    asyncio.create_task(fire_event(condition))

    await asyncio.gather(do_work(condition), do_work(condition))


if __name__ == "__main__":
    asyncio.run(main())

Иногда нам требуется, чтобы asyncio.Condition ждал наступления определенного события, прежде чем перейти к следующему шагу. Мы можем вызвать метод condition.wait_for() и передать в качестве аргумента метод. При каждом вызове condition.notify_all метод condition.wait_for проверяет результат выполнения метода-параметра и завершает ожидание, если он равен True, или продолжает ожидание, если он равен False.

condition.wait_for мы можем продемонстрировать на примере. В следующем коде мы смоделируем подключение к базе данных. Перед выполнением SQL-оператора код проверит, инициализировано ли соединение с базой данных, и выполнит запрос, если инициализация соединения завершена, или подождет, пока соединение не завершит инициализацию:

import asyncio
from asyncio import Condition
from enum import Enum


class ConnectionState(Enum):
    WAIT_INIT = 0
    INITIALING = 1
    INITIALIZED = 2


class Connection:
    def __init__(self):
        self._state = ConnectionState.WAIT_INIT
        self._condition = Condition()

    async def initialize(self):
        print("initialize: Preparing initialize the connection.")
        await self._change_state(ConnectionState.INITIALING)
        await asyncio.sleep(5)
        print("initialize: Connection initialized")
        await self._change_state(ConnectionState.INITIALIZED)

    async def execute(self, query: str):
        async with self._condition:
            print("execute: Waiting for connection initialized")
            await self._condition.wait_for(self._is_initialized)
            print(f"execute: Connection initialized, executing query: {query}")
            await asyncio.sleep(5)
            print("execute: Execute finished.")

    async def _change_state(self, state: ConnectionState):
        print(f"_change_state: Will change state from {self._state} to {state}")
        self._state = state
        print("_change_state: Change the state and notify all..")
        async with self._condition:
            self._condition.notify_all()

    def _is_initialized(self):
        if self._state is not ConnectionState.INITIALIZED:
            print("_is_initialized: The connection is not initialized.")
            return False
        print("_is_initialized: The connection is ready.")
        return True


async def main():
    connection = Connection()
    task_one = asyncio.create_task(connection.execute("SELECT * FROM table"))
    task_two = asyncio.create_task(connection.execute("SELECT * FROM other_table"))

    asyncio.create_task(connection.initialize())
    await asyncio.gather(task_one, task_two)


if __name__ == "__main__":
    asyncio.run(main())

Некоторые советы по использованию примитивов синхронизации asyncio

Не забывайте использовать таймаут или отмену при необходимости

При использовании примитивов синхронизации мы, как правило, ожидаем завершения определенной операции ввода-вывода. Однако из-за флуктуаций в сети или по другим неизвестным причинам операция ввода-вывода в задаче может выполняться дольше, чем в других.В этом случае необходимо установить таймаут для операции, чтобы при слишком большом времени выполнения освободить блокировку и дать возможность другим задачам выполнить ее вовремя.

В другом случае мы можем перебирать задачи в цикле. Это может заставить некоторые задачи ждать в фоновом режиме и помешать правильному завершению программы. В этом случае не забудьте воспользоваться командой cancel, чтобы прервать циклическое выполнение задачи.

Избегайте использования примитивов синхронизации или блокировки только наименьшего количества ресурсов

Все мы знаем, что преимущество asyncio заключается в том, что задача может переключаться на выполнение другой задачи в ожидании возврата IO. Но задача asyncio часто содержит как операции, связанные с IO, так и операции, связанные с процессором. Если заблокировать на задаче слишком много кода, то она не сможет вовремя переключиться на другую задачу, что скажется на производительности. Поэтому, если в этом нет необходимости, старайтесь не использовать примитивы синхронизации или блокировать только наименьшее количество ресурсов.

Чтобы избежать ситуаций с конкурентной блокировкой

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

Пример: Шина событий на базе Asyncio

Надеюсь к этому моменту вы смогли разобраться и понять как работую примитивы синхронизации в Asyncio. Давайте рассмотрим более пракичный пример, рассмотрев реализацию шины событий. Как обычно, первым шагом является проектирование EventBus API.

import asyncio
from asyncio import Event
import inspect
from typing import Callable


class EventBus:
    def __init__(self):
        self._event_dict = dict()

    async def on(self, event_name: str, fn: Callable):
        PASS

    def trigger(self, event_name: str, *args, **kwargs):
        PASS

    def _get_event(self, event_name: str):
        PASS

Давайте реализуем эти методы. Поскольку EventBus взаимодействует с помощью строк, а внутри я намерен использовать asyncio.Event для реализации событий, соответствующих каждой строке, начнем с реализации метода _get_event:

def _get_event(self, event_name: str):
        if event_name in self._event_dict:
            print("event already inited...")
            event = self._event_dict.get(event_name)
        else:
            print(f"need to init a new event for {event_name}")
            event = Event()
            self._event_dict[event_name] = event
        return event

Метод on привязывает функцию обратного вызова к определенному событию:

async def on(self, event_name: str, fn: Callable):
        event = self._get_event(event_name)
        while True:
            await event.wait()
            print("event fired")
            result = fn(*event.args, **event.kwargs)
            if inspect.isawaitable(result):
                await result
            
            # Since the callback function is likely a synchronous method, 
            # we must perform an await here to allow other tasks to execute.
            await asyncio.sleep(0.1)
            event.clear()

Метод trigger может вручную вызвать событие и передать соответствующие данные:

def trigger(self, event_name: str, *args, **kwargs):
    event = self._get_event(event_name)

    event.args = args
    event.kwargs = kwargs
    event.set()

Наконец, напишем метод main, чтобы проверить действие EventBus:

def a_sync_callback(data):
    print(f"A sync callback with data {data} is triggered")


async def a_async_callback(data):
    await asyncio.sleep(1)
    print(f"A async callback with data {data} is triggered")


async def main():
    event_bus = EventBus()
    task_one = asyncio.create_task(event_bus.on("some_event", a_async_callback))
    task_two = asyncio.create_task(event_bus.on("some_event", a_sync_callback))

    event_bus.trigger("some_event", {id: 1})
    await asyncio.wait([task_one, task_two], timeout=20.0)


asyncio.run(main())
  

В конце метода main не забудьте использовать таймаут, чтобы программа не выполнялась постоянно, как я уже предупреждал.

import asyncio
from asyncio import Event
import inspect
from typing import Callable


class EventBus:
    def __init__(self):
        self._event_dict = dict()

    def _get_event(self, event_name: str):
        if event_name in self._event_dict:
            print("event already inited...")
            event = self._event_dict.get(event_name)
        else:
            print(f"need to init a new event for {event_name}")
            event = Event()
            self._event_dict[event_name] = event
        return event

    async def on(self, event_name: str, fn: Callable):
        event = self._get_event(event_name)
        while True:
            await event.wait()
            print("event fired")
            result = fn(*event.args, **event.kwargs)
            if inspect.isawaitable(result):
                await result

            # Поскольку функция обратного вызова, скорее всего, является синхронным методом,
            # мы должны выполнить здесь await, чтобы дать возможность другим задачам выполниться.
            await asyncio.sleep(0.1)
            event.clear()

    def trigger(self, event_name: str, *args, **kwargs):
        event = self._get_event(event_name)

        event.args = args
        event.kwargs = kwargs
        event.set()


def a_sync_callback(data):
    print(f"A sync callback with data {data} is triggered")


async def a_async_callback(data):
    await asyncio.sleep(1)
    print(f"A async callback with data {data} is triggered")


async def main():
    event_bus = EventBus()
    task_one = asyncio.create_task(event_bus.on("some_event", a_async_callback))
    task_two = asyncio.create_task(event_bus.on("some_event", a_sync_callback))

    event_bus.trigger("some_event", {id: 1})
    await asyncio.wait([task_one, task_two], timeout=20)

asyncio.run(main())

Выводы

В этой статье сначала было рассказано о том, зачем Python asyncio нужны примитивы синхронизации. Затем я представил лучшие практики для Lock, Semaphore, Event и Condition и дал несколько советов по их правильному использованию.

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

Не стесняйтесь комментировать, делиться опытом или обсуждать со мной темы, связанные с asyncio.