18.5.3. Задачи и корутины

Исходный код: 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 – приостанавливает корутину до завершения future, затем возвращает результат future или возбуждает исключение, которое будет распространено. (Если future отменён, возникнет исключение CancelledError.) Обратите внимание, что задачи являются future, и всё сказанное о future относится также к задачам.

• 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.

If the generator is not yielded from before it is destroyed, an error message is logged. See Detect coroutines never scheduled.

Примечание

В этой документации некоторые методы описаны как корутины, даже если они являются обычными функциями Python, возвращающими Future. Это сделано намеренно, чтобы иметь возможность в будущем изменять реализацию этих функций. Если такую функцию необходимо использовать в коде в стиле колбэков, оберните её результат с помощью ensure_future().

18.5.3.1.1. Пример: корутина Hello World

Пример корутины, выводящей "Hello World":

import asyncio

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

loop = asyncio.get_event_loop()
# Блокирующий вызов, который возвращается после завершения корутины hello_world()
loop.run_until_complete(hello_world())
loop.close()

Смотрите также

Пример Hello World с call_soon() использует метод AbstractEventLoop.call_soon() для планирования колбэка.

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()
# Блокирующий вызов, который возвращается после завершения корутины display_date()
loop.run_until_complete(display_date(loop))
loop.close()

Смотрите также

Пример отображение текущей даты с call_later() использует колбэк с методом AbstractEventLoop.call_later().

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

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

18.5.3.2. InvalidStateError

exceptionasyncio.InvalidStateError¶

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

18.5.3.3. TimeoutError

exceptionasyncio.TimeoutError¶

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

Примечание

Это исключение отличается от встроенного исключения TimeoutError!

18.5.3.4. Future

classasyncio.Future(*, loop=None)¶

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

Различия:

• result() и exception() не принимают аргумент timeout и вызывают исключение, если future ещё не завершён.

• Колбэки, зарегистрированные с помощью add_done_callback(), всегда вызываются через call_soon() цикла событий.

• Этот класс несовместим с функциями wait() и as_completed() в пакете concurrent.futures.

Этот класс является не потокобезопасным.

cancel()¶

Отменяет future и планирует вызов колбэков.

Если future уже завершён или отменён, возвращает False. В противном случае меняет состояние future на «отменён», планирует колбэки и возвращает True.

cancelled()¶

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

done()¶

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

«Завершён» означает, что либо доступны результат или исключение, либо future был отменён.

result()¶

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

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

exception()¶

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

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

add_done_callback(fn)¶

Добавляет колбэк, который будет вызван, когда future станет завершённым.

Колбэк вызывается с одним аргументом – объектом future. Если future уже завершён на момент вызова, колбэк планируется с помощью 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)¶

Помечает future как завершённый и устанавливает его результат.

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

set_exception(исключение)¶

Помечает future как завершённый и устанавливает исключение.

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

18.5.3.4.1. Пример: Future с 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 секунду) и сохраняет результат в future. Метод run_until_complete() ожидает завершения future.

Примечание

Метод run_until_complete() внутри использует метод add_done_callback() для получения уведомления о том, что future завершён.

18.5.3.4.2. Пример: Future с 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()

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

18.5.3.5. Задача

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

Планирует выполнение корутины: оборачивает её в future. Задача – это подкласс Future.

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

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

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

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

Не создавайте экземпляры 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. Функции задач

Примечание

В функциях ниже необязательный аргумент loop позволяет явно задать объект цикла событий, используемый базовой задачей или корутиной. Если он не указан, используется цикл событий по умолчанию.

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

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

Вызывает asyncio.TimeoutError, если время ожидания истекает до завершения всех Future.

Пример:

for f in as_completed(fs):
    result = yield from f  # Выражение 'yield from' может возбуждать исключение
    # Использовать результат

Примечание

Объекты Future f не обязательно являются элементами fs.

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

Планирует выполнение объекта-корутины: оборачивает его в future. Возвращает объект Task.

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

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

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

Смотрите также

Метод 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)¶

Возвращает future, объединяющий результаты из переданных объектов корутин или futures.

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

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

Изменено в версии 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 для доступа к результату.

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

# Создать корутину
coro = asyncio.sleep(1, result=3)
# Отправить корутину в заданный цикл событий
future = asyncio.run_coroutine_threadsafe(coro, loop)
# Ожидать результат с необязательным аргументом тайм-аута
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))

Смотрите раздел concurrency and multithreading документации.

Примечание

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

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

coroutineasyncio.sleep(delay, result=None, *, loop=None)¶

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

Точность ожидания зависит от гранулярности цикла событий.

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

coroutineasyncio.shield(arg, *, loop=None)¶

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

Выражение:

res = yield from shield(something())

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

res = yield from something()

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

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

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

coroutineasyncio.wait(futures, *, loop=None, timeout=None, return_when=ALL_COMPLETED)¶

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

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

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

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

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

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

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

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

Примечание

Это не возбуждает asyncio.TimeoutError! Объекты Future, которые не завершены при наступлении тайм-аута, возвращаются во втором наборе.

coroutineasyncio.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 теперь также отменяется.