Задачи и сопрограммы

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

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

18.5.3.1. Сопрограммы

Сопрограммы, используемые с asyncio, могут быть реализованы с помощью инструкции async def или с использованием генераторов. Тип сопрограммы async def был добавлен в Python 3.5 и рекомендуется, если нет необходимости поддерживать более старые версии Python.

Генераторные сопрограммы должны быть оформлены декоратором @asyncio.coroutine, хотя это не строго обязательно. Декоратор обеспечивает совместимость с async def сопрограммами и также служит документацией. Генераторные сопрограммы используют синтаксис yield from , введённый в PEP 380, вместо исходного синтаксиса yield.

Слово «сопрограмма», подобно слову «генератор», используется для двух разных (хотя и связанных) концепций:

• Функция, определяющая сопрограмму (определение функции с использованием async def или с декоратором @asyncio.coroutine). В случае необходимости разграничения мы будем называть это функцией сопрограммы (iscoroutinefunction() возвращает True).
• Объект, полученный вызовом функции сопрограммы. Этот объект представляет вычисление или операцию ввода-вывода (обычно комбинацию), которые в конечном итоге завершатся. В случае необходимости разграничения мы будем называть его объектом сопрограммы (iscoroutine() возвращает True).

Что может делать сопрограмма:

• result = await future или result = yield from future — приостанавливает сопрограмму до завершения будущего, затем возвращает результат будущего или вызывает исключение, которое будет распространено. (Если будущее отменено, оно вызовет исключение CancelledError). Обратите внимание, что задачи являются будущими, и всё сказанное о будущих также относится к задачам.
• result = await coroutine или result = yield from coroutine — ожидает, пока другая сопрограмма выдаст результат (или вызовет исключение, которое будет распространено). Выражение coroutine должно быть вызовом другой сопрограммы.
• return expression — выдаёт результат сопрограмме, которая ожидает этого с помощью await или yield from.
• raise exception — вызывает исключение в сопрограмме, которая ожидает этого с помощью await или yield from.

Вызов сопрограммы не запускает её код — объект сопрограммы, возвращаемый вызовом, ничего не делает, пока вы не запланируете её выполнение. Есть два основных способа запустить её: вызвать await coroutine или yield from coroutine из другой сопрограммы (предполагая, что другая сопрограмма уже запущена!), или запланировать её выполнение с помощью функции ensure_future() или метода AbstractEventLoop.create_task().

Сопрограммы (и задачи) могут выполняться только при работе цикла событий.

@asyncio.coroutine

Декоратор для обозначения генераторных сопрограмм. Это позволяет генератору использовать yield from для вызова async def сопрограмм, а также позволяет генератору быть вызванным async def сопрограммами, например, с помощью выражения await.

Нет необходимости оформлять async def сопрограммы сами по себе.

Если генератор не возвращает значения перед разрушением, отображается сообщение об ошибке. Смотрите Обнаружение сопрограмм, не запланированных для выполнения.

18.5.3.1.1. Пример: Сопрограмма Hello World

Пример сопрограммы, отображающей "Hello World":

import asyncio

async def hello_world():
    print("Hello World!")

loop = asyncio.get_event_loop()
# Blocking call which returns when the hello_world() coroutine is done
loop.run_until_complete(hello_world())
loop.close()

18.5.3.1.2. Пример: Сопрограмма отображения текущей даты

Пример сопрограммы, отображающей текущую дату каждую секунду в течение 5 секунд с использованием функции sleep():

import asyncio
import datetime

async def display_date(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)

loop = asyncio.get_event_loop()
# Blocking call which returns when the display_date() coroutine is done
loop.run_until_complete(display_date(loop))
loop.close()

18.5.3.1.3. Пример: Цепочка сопрограмм

Пример цепочки сопрограмм:

import asyncio

async def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(1.0)
    return x + y

async def print_sum(x, y):
    result = await compute(x, y)
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()

compute() связана с print_sum(): сопрограмма print_sum() ожидает завершения compute() перед возвращением своего результата.

Диаграмма последовательности примера:

«Задача» создаётся методом AbstractEventLoop.run_until_complete() при получении объекта сопрограммы вместо задачи.

Диаграмма показывает поток управления, она не описывает точно, как всё работает внутри. Например, сопрограмма сна создаёт внутреннее будущее, которое использует AbstractEventLoop.call_later() для пробуждения задачи через 1 секунду.

18.5.3.2. Ошибка состояния

exception asyncio.InvalidStateError

Операция запрещена в данном состоянии.

18.5.3.3. Ошибка таймаута

exception asyncio.TimeoutError

Операция превысила заданный срок.

18.5.3.4. Будущее

class asyncio.Future(*, loop=None)

Этот класс почти совместим с concurrent.futures.Future.

Отличия:

• result() и exception() не принимают аргумент timeout и генерируют исключение, если задача ещё не завершена.
• Обратные вызовы, зарегистрированные с помощью add_done_callback(), всегда вызываются через цикл событий asyncio с помощью call_soon().
• Этот класс несовместим с функциями wait() и as_completed() в пакете concurrent.futures.

Этот класс не потокобезопасен.

cancel()

Отменить задачу и запланировать обратные вызовы.

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

cancelled()

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

done()

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

Выполнено означает, что результат/исключение доступны или задача была отменена.

result()

Возвращает результат, представленный этой задачей.

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

exception()

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

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

add_done_callback(fn)

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

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

Используйте functools.partial для передачи параметров в обратный вызов. Например, fut.add_done_callback(functools.partial(print, "Future:", flush=True)) вызовет print("Future:", fut, flush=True).

remove_done_callback(fn)

Удаляет все экземпляры обратного вызова из списка «вызвать при завершении».

Возвращает количество удалённых обратных вызовов.

set_result(result)

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

Если задача уже выполнена при вызове этого метода, генерирует InvalidStateError.

set_exception(exception)

Помечает задачу как выполненную и устанавливает исключение.

Если задача уже выполнена при вызове этого метода, генерирует InvalidStateError.

18.5.3.4.1. Пример: Задача с run_until_complete()

Пример объединения Future и функции-генератора:

import asyncio

async def slow_operation(future):
    await asyncio.sleep(1)
    future.set_result('Future is done!')

loop = asyncio.get_event_loop()
future = asyncio.Future()
asyncio.ensure_future(slow_operation(future))
loop.run_until_complete(future)
print(future.result())
loop.close()

Функция-генератор отвечает за вычисление (которое занимает 1 секунду) и сохраняет результат в задаче. Метод run_until_complete() ожидает завершения задачи.

18.5.3.4.2. Пример: Задача с run_forever()

Предыдущий пример можно переписать, явно описывая поток управления с помощью метода Future.add_done_callback():

import asyncio

async def slow_operation(future):
    await asyncio.sleep(1)
    future.set_result('Future is done!')

def got_result(future):
    print(future.result())
    loop.stop()

loop = asyncio.get_event_loop()
future = asyncio.Future()
asyncio.ensure_future(slow_operation(future))
future.add_done_callback(got_result)
try:
    loop.run_forever()
finally:
    loop.close()

В этом примере задача используется для связывания slow_operation() с got_result(): когда slow_operation() завершается, got_result() вызывается с результатом.

18.5.3.5. Задача

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

Планирование выполнения корутины: обернуть её в будущее. Задача — подкласс Future.

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

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

Отмена задачи отличается от отмены будущего. Вызов cancel() выбросит CancelledError в обернутую корутину. cancelled() возвращает True только если обернутая корутина не перехватила исключение CancelledError или не подняла исключение CancelledError.

Если ожидающая задача уничтожается, выполнение её обернутой корутины не завершено. Вероятно, это ошибка, и будет выведен соответствующий лог-предупреждение: см. Уничтожение ожидающей задачи.

Не создавайте напрямую экземпляры Task: используйте функцию ensure_future() или метод AbstractEventLoop.create_task().

Этот класс не потокобезопасен.

classmethod all_tasks(loop=None)

Возвращает множество всех задач для цикла событий.

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

classmethod current_task(loop=None)

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

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

None возвращается, если вызов не происходит в контексте Task.

cancel()

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

Это организует выброс CancelledError в обернутую корутину на следующей итерации цикла событий. Корутина получает шанс очиститься или даже отклонить запрос, используя try/except/finally.

В отличие от Future.cancel(), это не гарантирует отмены задачи: исключение может быть перехвачено и обработано, задерживая отмену задачи или полностью препятствуя ей. Задача также может вернуть значение или вызвать другое исключение.

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

get_stack(*, limit=None)

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

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

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

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

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

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

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

Выводит вывод, подобный выводу модуля traceback, для кадров, полученных с помощью get_stack(). Аргумент limit передаётся в get_stack(). Аргумент file — поток ввода-вывода, в который записывается вывод; по умолчанию вывод записывается в sys.stderr.

18.5.3.5.1. Пример: Параллельное выполнение задач

Пример выполнения 3 задач (A, B, C) параллельно:

import asyncio

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

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(
    factorial("A", 2),
    factorial("B", 3),
    factorial("C", 4),
))
loop.close()

Вывод:

Task A: Compute factorial(2)...
Task B: Compute factorial(2)...
Task C: Compute factorial(2)...
Task A: factorial(2) = 2
Task B: Compute factorial(3)...
Task C: Compute factorial(3)...
Task B: factorial(3) = 6
Task C: Compute factorial(4)...
Task C: factorial(4) = 24

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

18.5.3.6. Функции задач

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

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

Вызывает asyncio.TimeoutError, если таймаут наступает до завершения всех будущих объектов.

Пример:

for f in as_completed(fs):
    result = yield from f  # The 'yield from' may raise
    # Use result

Примечание

Будущие f необязательно являются членами fs.

asyncio.ensure_future(coro_or_future, *, loop=None)

Планирование выполнения объекта корутины: обернуть его в будущее. Возвращает объект Task.

Если аргумент является Future, он возвращается напрямую.

Введено в версии 3.4.4.

Изменено в версии 3.5.1: Функция принимает любой awaitable объект.

См. также

Метод AbstractEventLoop.create_task().

asyncio.async(coro_or_future, *, loop=None)

Устаревшее алиас для ensure_future().

Устарело начиная с версии 3.4.4.

asyncio.wrap_future(future, *, loop=None)

Оборачивает concurrent.futures.Future объект в Future объект.

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

Возвращает будущее, агрегирующее результаты заданных корутин или будущих объектов.

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

Отмена: если внешнее будущее отменяется, все дочерние элементы (которые ещё не завершены) также отменяются. Если какая-либо дочерняя задача отменяется, это обрабатывается так, как если бы она подняла CancelledError — внешнее будущее не отменяется в этом случае. (Это предотвращает отмену одного дочернего элемента, чтобы вызвать отмену других дочерних элементов.)

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

asyncio.iscoroutine(obj)

Возвращает True , если obj является объектом корутины, который может быть основан на генераторе или async def корутине.

asyncio.iscoroutinefunction(func)

Возвращает True, если func определён как функция-генератор корутины, которая может быть декорированной функцией-генератором или функцией async def.

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('The coroutine raised an exception: {!r}'.format(exc))
else:
    print('The coroutine returned: {!r}'.format(result))

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

Примечание

В отличие от других функций модуля, run_coroutine_threadsafe() требует явного указания аргумента loop.

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

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

Создаёт корутину, которая завершается после заданного времени (в секундах). Если result предоставлен, он передаётся вызывающей стороне при завершении корутины.

Точность засыпания зависит от точности цикла событий.

Данная функция является корутиной.

coroutine asyncio.shield(arg, *, loop=None)

Ожидает завершения future, защищая его от отмены.

Выражение:

res = yield from shield(something())

совершенно эквивалентно выражению:

res = yield from something()

кроме того, если корутина, содержащая его, отменяется, задача, выполняемая в something(), не отменяется. С точки зрения something(), отмена не произошла. Но вызывающая сторона всё ещё отменена, поэтому выражение yield-from всё ещё вызывает CancelledError. Примечание: если something() отменяется другими способами, это всё равно отменит shield().

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

try:
    res = yield from shield(something())
except CancelledError:
    res = None

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

Ожидает завершения объектов Futures и корутин, заданных последовательностью futures. Корутины будут обернуты в Tasks. Возвращает два множества Future: (выполненные, ожидающие).

Последовательность futures не должна быть пустой.

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

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

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

Данная функция является корутиной.

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

done, pending = yield from asyncio.wait(fs)

Примечание

Это не вызывает asyncio.TimeoutError! Future, которые не завершены к моменту истечения времени ожидания, возвращаются во втором множестве.

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

Ожидает завершения единственной Future или объекта корутины с ограничением по времени. Если timeout равен None, ожидать завершения future до завершения.

Корутина будет обернута в Task.

Возвращает результат Future или корутины. При истечении времени ожидания задача отменяется и возникает asyncio.TimeoutError. Чтобы избежать отмены задачи, оберните её в shield().

Если ожидание отменено, future fut также отменяется.

Данная функция является корутиной, пример использования:

result = yield from asyncio.wait_for(fut, 60.0)

Изменено в версии 3.4.3: Если ожидание отменено, future fut теперь также отменяется.