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

Блокировка

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)

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

Примитив условия 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)

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

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

Семафор

class asyncio.Semaphore(value=1)

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

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