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

Блокировка

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

Событие

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 если событие установлено.

Условие

class asyncio.Condition(lock=None)

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

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

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

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

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

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

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

locked()

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

notify_all()

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

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

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

release()

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

При вызове на разблокированной блокировке возникает ошибка RuntimeError.

coroutine wait()

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

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

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

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

coroutine wait_for(predicate)

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

Предикат должен быть вызываемым объектом, результат которого будет интерпретироваться как булево значение. Метод будет повторять wait() до тех пор, пока предикат не примет значение 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 сброшен или сломан.