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

Блокировка

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 удалён.

async acquire()

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

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

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

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

release()

Освобождает блокировку.

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

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

locked()

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

Событие

class asyncio.Event

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

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

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

Изменено в версии 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())

async wait()

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

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

set()

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

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

clear()

Сбрасывает (снимает) событие.

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

is_set()

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

Условие

class asyncio.Condition(lock=None)

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

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

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

Необязательный аргумент 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()

async acquire()

Захватывает базовую блокировку.

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

notify(n=1)

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

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

locked()

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

notify_all()

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

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

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

release()

Освобождает базовую блокировку.

При вызове для разблокированной блокировки возникает исключение RuntimeError.

async wait()

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

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

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

async wait_for(predicate)

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

Предикат должен быть вызываемым объектом, результат которого будет интерпретироваться как логическое значение. Метод будет повторно вызывать wait(), пока предикат не вернёт true. Возвращаемым значением будет итоговый результат.

Семафор

class asyncio.Semaphore(value=1)

Объект семафора. Не является потокобезопасным.

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

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

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

Рекомендуемый способ использования Semaphore — оператор 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()

async 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(). После этого все ожидающие задачи одновременно продолжат выполнение.

В качестве альтернативы ожиданию вызова wait() можно использовать async with.

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

Пример:

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.

async 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, если задача отменена.

async reset()

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

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

async abort()

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

parties

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

n_waiting

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

broken

Логическое значение, равное True, если барьер находится в сломанном состоянии.

exception asyncio.BrokenBarrierError

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