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

Исходный код: Lib/asyncio/locks.py

---

Примитивы синхронизации asyncio спроектированы так, чтобы быть похожими на примитивы из модуля threading, с двумя важными оговорками:

• Примитивы asyncio не являются потокобезопасными, поэтому их не следует использовать для синхронизации потоков ОС (используйте для этого threading);

• Методы этих примитивов синхронизации не принимают аргумент тайм-аут; используйте функцию asyncio.wait_for() для выполнения операций с тайм-аутами.

asyncio имеет следующие базовые примитивы синхронизации:

• Lock

• Event

• Condition

• Semaphore

• BoundedSemaphore

• Barrier

---

LockБлокировка

class asyncio.Lock¶

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

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

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

lock = asyncio.Lock()

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

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

lock = asyncio.Lock()

# ... later
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 управляет внутренним флагом, который может быть установлен в истину с помощью метода set() и сброшен в ложь с помощью метода clear(). Метод wait() блокируется до тех пор, пока флаг не будет установлен в истину. Изначально флаг установлен в ложь.

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

Пример:

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

async def main():
    # Создать объект Event.
    event = asyncio.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 должен быть объектом 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)¶

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

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

SemaphoreСемафор

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:
    # работа с общим ресурсом

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

sem = asyncio.Semaphore(10)

# ... later
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)¶

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

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

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

BarrierБарьер

class asyncio.Barrier(parties)¶

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

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

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

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

Пример:

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

   # создать 2 новые ожидающие задачи
   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.11.

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

---

Изменено в версии 3.9: Блокировка с помощью await lock или yield from lock и/или оператора with (with await lock, with (yield from lock)) была удалена. Используйте async with lock вместо этого.