18.5.3. Задачи и корутины¶Tasks and coroutines

18.5.3.1. Корутины¶Coroutines

Корутины, используемые с 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.

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

import asyncio

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

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

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()

18.5.3.1.3. Пример: цепочка корутин¶Example: Chain coroutines

Пример соединения корутин в цепочку:

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¶

exception asyncio.InvalidStateError¶

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

18.5.3.3. TimeoutError¶

exception asyncio.TimeoutError¶

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

18.5.3.4. Future¶

class asyncio.Future(*, loop=None)¶

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

Отличия:

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

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

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

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

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(exception)¶

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

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

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()

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()

18.5.3.5. Задача¶Task

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

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

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

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

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

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

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

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

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)¶

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

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

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

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

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

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

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

Вывод похож на вывод модуля traceback для фреймов, полученных с помощью get_stack(). Аргумент limit передаётся в get_stack(). Аргумент file – это I/O поток, в который записывается вывод; по умолчанию вывод записывается в sys.stderr.

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. Функции задач¶Task functions

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

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

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

Пример:

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

Примечание

Будущие объекты 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, агрегирующий результаты от заданных корутинных объектов или future.

Все 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.

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()

except that if the coroutine containing it is cancelled, the task running in something() is not cancelled. From the point of view of something(), the cancellation did not happen. But its caller is still cancelled, so the yield-from expression still raises CancelledError. Note: If something() is cancelled by other means this will still cancel 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)¶

Ожидает завершения объектов 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	Функция вернёт результат, когда все 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 теперь также отменяется.