Потоки и задачи

Потоки

Потоки объявленные с помощью синтаксиса async/await являются предпочтительным способом написания приложений asyncio. Например, следующий фрагмент кода выводит “hello”, ждёт 1 секунду, а затем выводит “world”:

>>> import asyncio

>>> async def main():
...     print('hello')
...     await asyncio.sleep(1)
...     print('world')

>>> asyncio.run(main())
hello
world

Обратите внимание, что просто вызов потока не запланирует его выполнение:

>>> main()
<coroutine object main at 0x1053bb7c8>

Для фактического запуска потока asyncio предоставляет три основных механизма:

• Функция asyncio.run() для запуска точки входа верхнего уровня — функции «main()» (см. пример выше).

• Ожидание потока. Следующий фрагмент кода выведет «hello» после ожидания 1 секунды, а затем выведет «world» после ожидания ещё 2 секунд:

  import asyncio
  import time
  
  async def say_after(delay, what):
      await asyncio.sleep(delay)
      print(what)
  
  async def main():
      print(f"started at {time.strftime('%X')}")
  
      await say_after(1, 'hello')
      await say_after(2, 'world')
  
      print(f"finished at {time.strftime('%X')}")
  
  asyncio.run(main())
  

  Ожидаемый вывод:

  started at 17:13:52
  hello
  world
  finished at 17:13:55
  

• Функция asyncio.create_task() для одновременного запуска потоков как asyncio Tasks.

  Давайте изменим предыдущий пример и запустим два say_after потока одновременно:

  async def main():
      task1 = asyncio.create_task(
          say_after(1, 'hello'))
  
      task2 = asyncio.create_task(
          say_after(2, 'world'))
  
      print(f"started at {time.strftime('%X')}")
  
      # Wait until both tasks are completed (should take
      # around 2 seconds.)
      await task1
      await task2
  
      print(f"finished at {time.strftime('%X')}")
  

  Обратите внимание, что ожидаемый вывод теперь показывает, что фрагмент выполняется на 1 секунду быстрее, чем раньше:

  started at 17:14:32
  hello
  world
  finished at 17:14:34
  

Awaitables

Мы говорим, что объект является awaitable объектом, если он может быть использован в выражении await. Многие API asyncio разработаны для приема awaitables.

Существует три основных типа awaitable объектов: потоки, задачи и будущие.

Потоки

Python-потоки являются awaitables и, следовательно, могут быть ожидаемыми другими потоками:

import asyncio

async def nested():
    return 42

async def main():
    # Nothing happens if we just call "nested()".
    # A coroutine object is created but not awaited,
    # so it *won't run at all*.
    nested()

    # Let's do it differently now and await it:
    print(await nested())  # will print "42".

asyncio.run(main())

asyncio также поддерживает устаревшие потоки на основе генераторов.

Задачи

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

Когда поток оборачивается в задачу с помощью функций, таких как asyncio.create_task(), поток автоматически планируется для запуска в ближайшее время:

import asyncio

async def nested():
    return 42

async def main():
    # Schedule nested() to run soon concurrently
    # with "main()".
    task = asyncio.create_task(nested())

    # "task" can now be used to cancel "nested()", or
    # can simply be awaited to wait until it is complete:
    await task

asyncio.run(main())

Будущие

A Future — это специальный низкоуровневый awaitable-объект, представляющий конечный результат асинхронной операции.

Когда ожидается объект Future, это означает, что поток будет ждать, пока Future не будет разрешен в другом месте.

Объекты Future в asyncio необходимы для использования кода на основе обратных вызовов с async/await.

Обычно нет необходимости создавать объекты Future на уровне кода приложения.

Объекты Future, иногда предоставляемые библиотеками и некоторыми API asyncio, могут быть ожидаемыми:

async def main():
    await function_that_returns_a_future_object()

    # this is also valid:
    await asyncio.gather(
        function_that_returns_a_future_object(),
        some_python_coroutine()
    )

Хороший пример низкоуровневой функции, возвращающей объект Future, — loop.run_in_executor().

Запуск программы asyncio

asyncio.run(coro, *, debug=False)

Выполнить поток coro и вернуть результат.

Эта функция запускает переданный поток, позаботившись об управлении циклом событий asyncio, завершении асинхронных генераторов и закрытии пула потоков.

Этот метод не может быть вызван, когда в одном потоке работает другой цикл событий asyncio.

Если debug True, цикл событий будет запущен в отладочном режиме.

Эта функция всегда создает новый цикл событий и закрывает его в конце. Она должна использоваться в качестве основной точки входа для программ asyncio и, желательно, вызываться только один раз.

Пример:

async def main():
    await asyncio.sleep(1)
    print('hello')

asyncio.run(main())

Новое в версии 3.7.

Изменено в версии 3.9: Обновлено для использования loop.shutdown_default_executor().

Примечание

Исходный код для asyncio.run() можно найти в Lib/asyncio/runners.py.

Создание задач

asyncio.create_task(coro, *, name=None)

Оборачивает поток coro в Task и планирует его выполнение. Возвращает объект Task.

Если name не None, он устанавливается в качестве имени задачи с помощью Task.set_name().

Задача выполняется в цикле, возвращаемом get_running_loop(). RuntimeError генерируется, если в текущем потоке нет работающего цикла.

Важно

Сохраните ссылку на результат этой функции, чтобы избежать исчезновения задачи во время выполнения.

Новое в версии 3.7.

Изменено в версии 3.8: Добавлен параметр name.

Ожидание

coroutine asyncio.sleep(delay, result=None, *, loop=None)

Ожидание delay секунд.

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

sleep() всегда приостанавливает текущую задачу, позволяя другим задачам работать.

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

Устаревшее с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Пример потока, отображающего текущую дату каждую секунду в течение 5 секунд:

import asyncio
import datetime

async def display_date():
    loop = asyncio.get_running_loop()
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(1)

asyncio.run(display_date())

Выполнение задач одновременно

awaitable asyncio.gather(*aws, loop=None, return_exceptions=False)

Запустить объекты-awaitables в последовательности aws одновременно.

Если какой-либо awaitable в aws является корутиной, она автоматически планируется как задача (Task).

Если все awaitables завершились успешно, результатом является агрегированный список возвращаемых значений. Порядок значений результата соответствует порядку awaitables в aws.

Если return_exceptions равно False (по умолчанию), первое поднятое исключение сразу же передаётся задаче, которая ожидает на gather(). Другие awaitables в последовательности aws не будут отменены и продолжат выполняться.

Если return_exceptions равно True, исключения обрабатываются так же, как и успешные результаты, и агрегируются в списке результатов.

Если gather() отменена, все отправленные awaitables (которые ещё не завершились) также отменяются.

Если какая-либо задача (Task) или будущее (Future) из последовательности aws отменена, она обрабатывается так, как если бы она подняла CancelledError – вызов gather() в этом случае не отменяется. Это сделано для предотвращения отмены одной отправленной задачи/будущего, которая могла бы вызвать отмену других задач/будущих.

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Пример:

import asyncio

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task {name}: Compute factorial({number}), currently i={i}...")
        await asyncio.sleep(1)
        f *= i
    print(f"Task {name}: factorial({number}) = {f}")
    return f

async def main():
    # Schedule three calls *concurrently*:
    L = await asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )
    print(L)

asyncio.run(main())

# Expected output:
#
#     Task A: Compute factorial(2), currently i=2...
#     Task B: Compute factorial(3), currently i=2...
#     Task C: Compute factorial(4), currently i=2...
#     Task A: factorial(2) = 2
#     Task B: Compute factorial(3), currently i=3...
#     Task C: Compute factorial(4), currently i=3...
#     Task B: factorial(3) = 6
#     Task C: Compute factorial(4), currently i=4...
#     Task C: factorial(4) = 24
#     [2, 6, 24]

Примечание

Если return_exceptions равно False, отмена gather() после того, как она была помечена как завершённая, не отменяет отправленных awaitables. Например, gather может быть помечена как завершённая после передачи исключения вызывающей стороне, поэтому вызов gather.cancel() после перехвата исключения (поднятого одним из awaitables) от gather не отменит другие awaitables.

Изменено в версии 3.7: Если само gather отменено, отмена распространяется независимо от return_exceptions.

Защита от отмены

awaitable asyncio.shield(aw, *, loop=None)

Защитить объект-awaitable от cancelled.

Если aw является корутиной, она автоматически планируется как задача (Task).

Выражение:

res = await shield(something())

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

res = await something()

за исключением того, что если корутина, содержащая его, отменена, задача, выполняющаяся в something() не отменяется. С точки зрения something(), отмена не произошла. Хотя вызывающая её сторона всё равно отменена, поэтому выражение «await» всё равно поднимет CancelledError.

Если something() отменена другими средствами (например, изнутри), это также отменит shield().

Если необходимо полностью игнорировать отмену (не рекомендуется), функция shield() должна быть комбинирована с блоком try/except, как показано ниже:

try:
    res = await shield(something())
except CancelledError:
    res = None

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Таймауты

coroutine asyncio.wait_for(aw, timeout, *, loop=None)

Подождать завершения aw awaitable с таймаутом.

Если aw является корутиной, она автоматически планируется как задача (Task).

timeout может быть None или числом с плавающей точкой или целым числом в секундах для ожидания. Если timeout None, заблокировать до завершения будущего.

Если таймаут истекает, задача отменяется и поднимается asyncio.TimeoutError.

Чтобы избежать отмены задачи cancellation, оберните её в shield().

Функция будет ждать, пока будущее не будет фактически отменено, поэтому общее время ожидания может превысить timeout. Если во время отмены происходит исключение, оно распространяется.

Если ожидание отменено, будущее aw также отменяется.

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Пример:

async def eternity():
    # Sleep for one hour
    await asyncio.sleep(3600)
    print('yay!')

async def main():
    # Wait for at most 1 second
    try:
        await asyncio.wait_for(eternity(), timeout=1.0)
    except asyncio.TimeoutError:
        print('timeout!')

asyncio.run(main())

# Expected output:
#
#     timeout!

Изменено в версии 3.7: Когда aw отменяется из-за таймаута, wait_for ждёт, пока aw будет отменено. Раньше оно поднимало asyncio.TimeoutError немедленно.

Примитивы ожидания

coroutine asyncio.wait(aws, *, loop=None, timeout=None, return_when=ALL_COMPLETED)

Запустить объекты-awaitables в итерируемом объекте aws одновременно и заблокировать до выполнения условия, заданного return_when.

Итерируемый объект aws не должен быть пустым.

Возвращает два набора задач/будущих: (done, pending).

Использование:

done, pending = await asyncio.wait(aws)

timeout (число с плавающей точкой или целое число), если указано, может использоваться для управления максимальным числом секунд ожидания перед возвратом.

Обратите внимание, что эта функция не поднимает asyncio.TimeoutError. Будущие или задачи, которые не завершены к моменту истечения таймаута, просто возвращаются во второй набор.

return_when указывает, когда эта функция должна вернуть значение. Она должна быть одним из следующих констант:

Константа	Описание
FIRST_COMPLETED	Функция вернётся, когда какое-либо будущее завершится или будет отменено.
FIRST_EXCEPTION	Функция вернётся, когда какое-либо будущее завершится, подняв исключение. Если ни одно будущее не поднимет исключение, то это эквивалентно ALL_COMPLETED.
ALL_COMPLETED	Функция вернётся, когда все будущие завершаться или отменяются.

В отличие от wait_for(), wait() не отменяет будущие при истечении таймаута.

Устарело начиная с версии 3.8: Если какой-либо awaitable в aws является корутиной, она автоматически планируется как задача (Task). Передача объектов корутин в wait() напрямую устарела, так как это приводит к неясном поведению.

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Примечание

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

async def foo():
    return 42

coro = foo()
done, pending = await asyncio.wait({coro})

if coro in done:
    # This branch will never be run!

Вот как можно исправить указанный фрагмент:

async def foo():
    return 42

task = asyncio.create_task(foo())
done, pending = await asyncio.wait({task})

if task in done:
    # Everything will work as expected now.

Устарело начиная с версии 3.8, будет удалено в версии 3.11: Передача объектов корутин в wait() напрямую устарела.

asyncio.as_completed(aws, *, loop=None, timeout=None)

Запустить объекты-awaitables в итерируемом объекте aws одновременно. Возвращает итератор корутин. Каждая возвращённая корутина может быть ожидаема, чтобы получить самый ранний следующий результат из итерируемого объекта оставшихся awaitables.

Поднимает asyncio.TimeoutError, если таймаут происходит до завершения всех Futures.

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

Пример:

for coro in as_completed(aws):
    earliest_result = await coro
    # ...

Выполнение в потоках

coroutine asyncio.to_thread(func, /, *args, **kwargs)

Асинхронно выполнить функцию func в отдельном потоке.

Любые *args и **kwargs, переданные для этой функции, непосредственно передаются в func. Также текущий contextvars.Context распространяется, позволяя переменным контекста из потока цикла событий быть доступными в отдельном потоке.

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

Эта функция-корутина предназначена в первую очередь для выполнения функций/методов, связанных с вводом-выводом, которые в противном случае заблокируют цикл событий, если они выполняются в основном потоке. Например:

def blocking_io():
    print(f"start blocking_io at {time.strftime('%X')}")
    # Note that time.sleep() can be replaced with any blocking
    # IO-bound operation, such as file operations.
    time.sleep(1)
    print(f"blocking_io complete at {time.strftime('%X')}")

async def main():
    print(f"started main at {time.strftime('%X')}")

    await asyncio.gather(
        asyncio.to_thread(blocking_io),
        asyncio.sleep(1))

    print(f"finished main at {time.strftime('%X')}")


asyncio.run(main())

# Expected output:
#
# started main at 19:50:53
# start blocking_io at 19:50:53
# blocking_io complete at 19:50:54
# finished main at 19:50:54

Прямой вызов blocking_io() в любой корутине заблокирует цикл событий на всё время её выполнения, что приведёт к дополнительной 1 секунде времени выполнения. Вместо этого, используя asyncio.to_thread(), мы можем выполнить его в отдельном потоке без блокирования цикла событий.

Примечание

Из-за GIL, asyncio.to_thread() обычно может использоваться только для того, чтобы сделать функции, связанные с вводом-выводом, неголосующими. Однако для расширяемых модулей, которые выпускают GIL, или альтернативных реализаций Python, у которых его нет, asyncio.to_thread() также можно использовать для функций, связанных с центральным процессором.

Добавлена в версии 3.9.

Планирование из других потоков

asyncio.run_coroutine_threadsafe(coro, loop)

Отправляет корутину в указанную событийную петлю. Потокобезопасный.

Возвращает concurrent.futures.Future для ожидания результата из другого потока ОС.

Эта функция предназначена для вызова из другого потока ОС, чем тот, в котором работает событийная петля. Пример:

# Create a coroutine
coro = asyncio.sleep(1, result=3)

# Submit the coroutine to a given loop
future = asyncio.run_coroutine_threadsafe(coro, loop)

# Wait for the result with an optional timeout argument
assert future.result(timeout) == 3

Если в корутине возникает исключение, возвращаемая Future будет уведомлена. Также её можно использовать для отмены задачи в событийной петле:

try:
    result = future.result(timeout)
except asyncio.TimeoutError:
    print('The coroutine took too long, cancelling the task...')
    future.cancel()
except Exception as exc:
    print(f'The coroutine raised an exception: {exc!r}')
else:
    print(f'The coroutine returned: {result!r}')

См. раздел конкурентности и многопоточности документации.

В отличие от других функций asyncio, эта функция требует явного передачи аргумента loop.

Новая в версии 3.5.1.

Интроспекция

asyncio.current_task(loop=None)

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

Если loop является None, используется get_running_loop() для получения текущей петли.

Новая в версии 3.7.

asyncio.all_tasks(loop=None)

Возвращает множество ещё не завершенных Task объектов, запущенных петлей.

Если loop является None, используется get_running_loop() для получения текущей петли.

Новая в версии 3.7.

Объект задачи

class asyncio.Task(coro, *, loop=None, name=None)

Объект Future-like, выполняющий Python-корутину. Не потокобезопасен.

Задачи используются для запуска корутин в циклах событий. Если корутина ожидает завершения задачи, задача приостанавливает выполнение корутины и ожидает завершения задачи. Когда задача завершается, выполнение обернутой корутины возобновляется.

Циклы событий используют кооперативное планирование: цикл событий выполняет одну задачу за раз. Пока задача ожидает завершения задачи, цикл событий выполняет другие задачи, обратные вызовы или выполняет операции ввода-вывода.

Используйте высокоуровневую функцию asyncio.create_task() для создания задач или низкоуровневые функции loop.create_task() или ensure_future(). Ручное создание задач не рекомендуется.

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

Можно использовать cancelled() для проверки, была ли задача отменена. Метод возвращает True , если обернутая корутина не подавила исключение CancelledError и была фактически отменена.

asyncio.Task наследует все свои API от Future, за исключением Future.set_result() и Future.set_exception().

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

Изменено в версии 3.7: Добавлена поддержка модуля contextvars.

Изменено в версии 3.8: Добавлен параметр name.

Устарело начиная с версии 3.8, будет удалено в версии 3.10: Параметр loop.

cancel(msg=None)

Запрос на отмену задачи.

Это организует бросок исключения CancelledError в обернутую корутину в следующем цикле цикла событий.

Корутина тогда имеет возможность выполнить очистку или даже отклонить запрос, подавив исключение с помощью блока try … … except CancelledError … finally. Поэтому, в отличие от Future.cancel(), Task.cancel() не гарантирует, что задача будет отменена, хотя полное подавление отмены не является распространенным и активно не рекомендуется.

Изменено в версии 3.9: Добавлен параметр msg.

Следующий пример демонстрирует, как корутины могут перехватывать запрос на отмену:

async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        # Wait for 1 hour
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')

async def main():
    # Create a "cancel_me" Task
    task = asyncio.create_task(cancel_me())

    # Wait for 1 second
    await asyncio.sleep(1)

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me is cancelled now")

asyncio.run(main())

# Expected output:
#
#     cancel_me(): before sleep
#     cancel_me(): cancel sleep
#     cancel_me(): after sleep
#     main(): cancel_me is cancelled now

cancelled()

Возвращает True , если задача отменена.

Задача отменена, когда отмена была запрошена с помощью cancel() и обернутая корутина распространила исключение CancelledError ,брошенное в неё.

done()

Возвращает True , если задача завершена.

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

result()

Возвращает результат задачи.

Если задача завершена, возвращается результат обернутой корутины (или, если корутина вызвала исключение, это исключение перевызывается).

Если задача была отменена, этот метод вызывает исключение CancelledError.

Если результат задачи еще недоступен, этот метод вызывает исключение InvalidStateError.

exception()

Возвращает исключение задачи.

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

Если задача была отменена, этот метод вызывает исключение CancelledError.

Если задача еще не завершена, этот метод вызывает исключение InvalidStateError.

add_done_callback(callback, *, context=None)

Добавляет обратный вызов, который будет выполнен, когда задача завершена.

Этот метод следует использовать только в низкоуровневом коде, основанном на обратных вызовах.

См. документацию Future.add_done_callback() для получения дополнительных сведений.

remove_done_callback(callback)

Удаляет callback из списка обратных вызовов.

Этот метод следует использовать только в низкоуровневом коде, основанном на обратных вызовах.

См. документацию Future.remove_done_callback() для получения дополнительных сведений.

get_stack(*, limit=None)

Возвращает список кадров стека для этой задачи.

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

Кадры всегда упорядочены от старейшего к новейшему.

Для приостановленной корутины всегда возвращается один кадр стека.

Необязательный аргумент limit задаёт максимальное количество возвращаемых кадров; по умолчанию возвращаются все доступные кадры. Порядок возвращаемого списка отличается в зависимости от того, возвращается ли стек или трассировка стека: возвращаются самые новые кадры стека, но самые старые кадры трассировки стека. (Это соответствует поведению модуля traceback.)

print_stack(*, limit=None, file=None)

Выводит стек или трассировку стека для этой задачи.

Это генерирует вывод, аналогичный тому, который генерирует модуль traceback для кадров, полученных методом get_stack().

Аргумент limit передаётся в get_stack() напрямую.

Аргумент file — поток ввода-вывода, в который записывается вывод; по умолчанию вывод записывается в sys.stderr.

get_coro()

Возвращает объект корутины, обернутый в Task.

Новое в версии 3.8.

get_name()

Возвращает имя задачи.

Если задаче явно не было присвоено имя, по умолчанию реализация asyncio Task генерирует имя по умолчанию во время создания.

Новое в версии 3.8.

set_name(value)

Устанавливает имя задачи.

Аргумент value может быть любым объектом, который затем преобразуется в строку.

В реализации задачи по умолчанию имя будет видно в выводе repr() объекта задачи.

Новое в версии 3.8.

Генераторные корутины

Генераторные корутины предшествуют синтаксису async/await. Это генераторы Python, которые используют выражения yield from для ожидания выполнения задач и других корутин.

Генераторные корутины должны быть декорированы с помощью @asyncio.coroutine, хотя это и не обязательно.

@asyncio.coroutine

Декоратор для маркировки генераторных корутин.

Этот декоратор позволяет корутинам на основе генераторов в стиле старого кода взаимодействовать с кодом async/await:

@asyncio.coroutine
def old_style_coroutine():
    yield from asyncio.sleep(1)

async def main():
    await old_style_coroutine()

Этот декоратор не следует использовать для корутин с async def.

Устарело начиная с версии 3.8, будет удалено в версии 3.11: Используйте async def вместо этого.

asyncio.iscoroutine(obj)

Возвращает True если obj является объектом корутины.

Этот метод отличается от inspect.iscoroutine(), так как возвращает True для генераторных корутин.

asyncio.iscoroutinefunction(func)

Возвращает True если func является функцией корутины.

Этот метод отличается от inspect.iscoroutinefunction(), так как возвращает True для функций генераторных корутин, декорированных с помощью @coroutine.