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

Блокировка (Lock)

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, если блокировка заблокирована.

Событие (Event)

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())

coroutine wait()

Ожидать, пока событие не будет установлено.

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

set()

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

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

clear()

Сбросить (убрать установку) события.

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

is_set()

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

Условие (Condition)

class asyncio.Condition(lock=None)

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

Примитив 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)

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

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

Семафор

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 сбрасывается или разрывается.