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

Блокировка

class asyncio.Lock

Реализует мьютекс-блокировку для задач asyncio. Не потокобезопасна.

Блокировка asyncio может использоваться для гарантии эксклюзивного доступа к общему ресурсу.

Предпочтительный способ использования Lock — оператор async with:

lock = asyncio.Lock()

# ... later
async with lock:
    # access shared state

что эквивалентно:

lock = asyncio.Lock()

# ... later
await lock.acquire()
try:
    # access shared state
finally:
    lock.release()

Изменено в версии 3.10: Параметр loop удалён.

coroutine acquire()

Захват блокировки.

Этот метод ожидает, пока блокировка не будет разблокирована, устанавливает её в заблокированное состояние и возвращает True.

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

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

release()

Освобождение блокировки.

Когда блокировка заблокирована, она сбрасывается в разблокированное состояние и возвращается значение.

Если блокировка разблокирована, генерируется исключение RuntimeError.

locked()

Возвращает True если блокировка заблокирована.

Событие

class asyncio.Event

Объект события. Не потокобезопасен.

Объект asyncio события может использоваться для уведомления нескольких задач asyncio о том, что какое-то событие произошло.

Объект Event управляет внутренней флагом, который может быть установлен в истинное значение с помощью метода set() и сброшен до ложного с помощью метода clear(). Метод wait() блокируется, пока флаг не будет установлен в истинное значение. Изначально флаг имеет значение ложь.

Изменено в версии 3.10: Параметр loop удалён.

Пример:

async def waiter(event):
    print('waiting for it ...')
    await event.wait()
    print('... got it!')

async def main():
    # Create an Event object.
    event = asyncio.Event()

    # Spawn a Task to wait until 'event' is set.
    waiter_task = asyncio.create_task(waiter(event))

    # Sleep for 1 second and set the event.
    await asyncio.sleep(1)
    event.set()

    # Wait until the waiter task is finished.
    await waiter_task

asyncio.run(main())

coroutine wait()

Ожидание установки события.

Если событие установлено, возвращается True немедленно. В противном случае блокируется до тех пор, пока другая задача не вызовет set().

set()

Установить событие.

Все задачи, ожидающие установки события, будут немедленно разбужены.

clear()

Сбросить (убрать) событие.

Задачи, ожидающие wait(), теперь будут блокироваться до тех пор, пока снова не будет вызван метод set().

is_set()

Возвращает True если событие установлено.

Условие

class asyncio.Condition(lock=None)

Объект Condition. Не потокобезопасен.

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

По сути, объект Condition объединяет функциональность объекта Event и объекта Lock. Возможны ситуации, когда несколько объектов Condition используют одну блокировку, что позволяет координировать эксклюзивный доступ к общему ресурсу между различными задачами, заинтересованными в определённых состояниях этого общего ресурса.

Необязательный аргумент lock должен быть объектом Lock или None. В последнем случае объект Lock создается автоматически.

Изменено в версии 3.10: Параметр loop удалён.

Предпочтительный способ использования Condition — оператор async with:

cond = asyncio.Condition()

# ... later
async with cond:
    await cond.wait()

что эквивалентно:

cond = asyncio.Condition()

# ... later
await cond.acquire()
try:
    await cond.wait()
finally:
    cond.release()

coroutine acquire()

Получить блокировку.

Этот метод ожидает, пока подчинённая блокировка не будет разблокирована, устанавливает её в заблокированное состояние и возвращает True.

notify(n=1)

Разбудить не более n задач (по умолчанию 1), ожидающих этого условия. Метод ничего не делает, если ожидающих задач нет.

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

locked()

Возвращает True если подчинённая блокировка захвачена.

notify_all()

Разбудить все задачи, ожидающие этого условия.

Этот метод действует как notify(), но разбуживает все ожидающие задачи.

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

release()

Освободить подчинённую блокировку.

Если вызов сделан с разблокированной блокировкой, возникает ошибка RuntimeError.

coroutine wait()

Ожидать уведомления.

Если вызывающая задача не получила блокировку при вызове этого метода, возникает ошибка RuntimeError.

Этот метод освобождает подчинённую блокировку, а затем блокируется до тех пор, пока он не разбужен вызовом notify() или notify_all(). После пробуждения Condition снова получает блокировку, и этот метод возвращает True.

coroutine wait_for(predicate)

Ожидать, пока предикат не станет истинным.

Предикат должен быть вызываемым объектом, результат которого будет интерпретирован как булево значение. Конечное значение — это возвращаемое значение.

Семафор

class asyncio.Semaphore(value=1)

Объект Семафора. Не потокобезопасен.

Семафор управляет внутренней переменной счётчика, которая уменьшается при каждом вызове acquire() и увеличивается при каждом вызове release(). Счётчик никогда не может стать меньше нуля; когда acquire() обнаруживает, что он равен нулю, он блокируется, ожидая, пока какая-то задача не вызовет release().

Необязательный аргумент value задаёт начальное значение внутреннего счётчика (по умолчанию 1). Если указанное значение меньше 0, генерируется исключение ValueError.

Изменено в версии 3.10: Параметр loop удалён.

Рекомендуемый способ использования семафора — оператор async with:

sem = asyncio.Semaphore(10)

# ... later
async with sem:
    # work with shared resource

что эквивалентно:

sem = asyncio.Semaphore(10)

# ... later
await sem.acquire()
try:
    # work with shared resource
finally:
    sem.release()

coroutine acquire()

Получить семафор.

Если внутренний счётчик больше нуля, уменьшите его на единицу и верните True сразу. Если он равен нулю, дождитесь вызова release() и верните True.

locked()

Возвращает True , если семафор не может быть получен сразу.

release()

Освободить семафор, увеличив внутренний счётчик на единицу. Может разбудить задачу, ожидающую получения семафора.

В отличие от BoundedSemaphore, Semaphore позволяет делать больше release() вызовов, чем acquire() вызовов.

Ограниченный семафор

class asyncio.BoundedSemaphore(value=1)

Объект ограниченного семафора. Не потокобезопасен.

Ограниченный семафор — это версия Semaphore, которая генерирует ValueError в release(), если она увеличивает внутренний счётчик сверх начального значения value.

Изменено в версии 3.10: Параметр loop удалён.

Барьер

class asyncio.Barrier(parties)

Объект барьера. Не потокобезопасен.

Барьер — это простой синхронизационный примитив, позволяющий заблокировать выполнение, пока parties задач не дождутся его. Задачи могут ожидать на методе wait() и будут заблокированы, пока указанное количество задач не дождутся wait(). В этот момент все ожидающие задачи разблокируются одновременно.

Можно использовать async with как альтернативу ожиданию на wait().

Барьер можно повторно использовать любое количество раз.

Пример:

async def example_barrier():
   # barrier with 3 parties
   b = asyncio.Barrier(3)

   # create 2 new waiting tasks
   asyncio.create_task(b.wait())
   asyncio.create_task(b.wait())

   await asyncio.sleep(0)
   print(b)

   # The third .wait() call passes the barrier
   await b.wait()
   print(b)
   print("barrier passed")

   await asyncio.sleep(0)
   print(b)

asyncio.run(example_barrier())

Результат этого примера:

<asyncio.locks.Barrier object at 0x... [filling, waiters:2/3]>
<asyncio.locks.Barrier object at 0x... [draining, waiters:0/3]>
barrier passed
<asyncio.locks.Barrier object at 0x... [filling, waiters:0/3]>

Новое в версии 3.11.

coroutine wait()

Пройти барьер. Когда все задачи, участвующие в барьере, вызовут эту функцию, они все разблокируются одновременно.

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

Возвращаемое значение — целое число в диапазоне от 0 до parties-1, разное для каждой задачи. Это можно использовать для выбора задачи для выполнения специальной утилитарной работы, например:

...
async with barrier as position:
   if position == 0:
      # Only one task prints this
      print('End of *draining phase*')

Этот метод может генерировать исключение BrokenBarrierError, если барьер сломан или сброшен, в то время как задача ожидает. Он может генерировать исключение CancelledError, если задача отменена.

coroutine reset()

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

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

coroutine abort()

Перевести барьер в состояние «сломан». Это заставляет любые активные или будущие вызовы wait() завершиться неудачно с исключением BrokenBarrierError. Используйте это, например, если одна из задач должна прерваться, чтобы избежать бесконечного ожидания задач.

parties

Число задач, необходимых для прохождения барьера.

n_waiting

Число задач, которые в настоящее время ожидают в барьере при заполнении.

broken

Булево значение, которое True , если барьер находится в состоянии «сломан».

exception asyncio.BrokenBarrierError

Это исключение, подкласс RuntimeError, генерируется, когда объект Barrier сброшен или сломан.