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

Блокировка¶Lock

class asyncio.Lock¶

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

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

The preferred way to use a Lock is an async with statement:

lock = asyncio.Lock()

# ... позже
async with lock:
    # доступ к общему состоянию

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

lock = asyncio.Lock()

# ... позже
await lock.acquire()
try:
    # доступ к общему состоянию
finally:
    lock.release()

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

async 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():
    # Создать объект Event.
    event = asyncio.Event()

    # Создать задачу, ожидающую установки 'event'.
    waiter_task = asyncio.create_task(waiter(event))

    # Подождать 1 секунду и установить событие.
    await asyncio.sleep(1)
    event.set()

    # Дождаться завершения ожидающей задачи.
    await waiter_task

asyncio.run(main())

async wait()¶

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

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

set()¶

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

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

clear()¶

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

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

is_set()¶

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

Condition¶

class asyncio.Condition(lock=None)¶

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

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

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

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

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

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

cond = asyncio.Condition()

# ... позже
async with cond:
    await cond.wait()

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

cond = asyncio.Condition()

# ... позже
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(). После пробуждения условие снова захватывает свою блокировку, и метод возвращает True.

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

async wait_for(predicate)¶

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

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

Semaphore¶

class asyncio.Semaphore(value=1)¶

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

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

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

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

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

sem = asyncio.Semaphore(10)

# ... позже
async with sem:
    # работать с общим ресурсом

это эквивалентно следующему:

sem = asyncio.Semaphore(10)

# ... позже
await sem.acquire()
try:
    # работать с общим ресурсом
finally:
    sem.release()

async acquire()¶

Захватывает семафор.

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

locked()¶

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

release()¶

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

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

BoundedSemaphore¶

class asyncio.BoundedSemaphore(value=1)¶

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

Bounded Semaphore – это версия Semaphore, которая возбуждает ValueError в release(), если увеличивает внутренний счётчик выше начального value.

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

Barrier¶

class asyncio.Barrier(parties)¶

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

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

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

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

Пример:

async def example_barrier():
   # барьер с тремя участниками
   b = asyncio.Barrier(3)

   # создать две новые ожидающие задачи
   asyncio.create_task(b.wait())
   asyncio.create_task(b.wait())

   await asyncio.sleep(0)
   print(b)

   # Третий вызов .wait() проходит барьер
   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.12.

async wait()¶

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

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

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

...
async with barrier as position:
   if position == 0:
      # Только одна задача выводит это
      print('End of *draining phase*')

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

async reset()¶

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

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

async abort()¶

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

parties¶

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

n_waiting¶

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

broken¶

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

exception asyncio.BrokenBarrierError¶

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