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

Блокировка

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)

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

Примитив условия 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), ожидающих этого условия. Если ожидающих задач меньше, чем n, будут разбужены все.

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

locked()

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

notify_all()

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

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

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

release()

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

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

async wait()

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

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

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

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

async wait_for(predicate)

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

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

Семафор

class asyncio.Semaphore(value=1)

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

Семафор управляет внутренним счётчиком, который уменьшается при каждом вызове 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 сбрасывается или ломается.