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

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 сопрограммы.

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

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

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

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 секунду.

2. InvalidStateError

exception asyncio.InvalidStateError

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

3. TimeoutError

exception asyncio.TimeoutError

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

4. Future

class asyncio.Future(*, loop=None)

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

Отличия:

• result() и exception() не принимают аргумент timeout и генерируют исключение, когда задача ещё не завершена.
• Обратные вызовы, зарегистрированные с помощью add_done_callback(), всегда вызываются через call_soon_threadsafe() цикла событий.
• Этот класс несовместим с функциями 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.

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() ожидает завершения задачи.

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() вызывается с результатом.

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.

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

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

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

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

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

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 указан, он возвращается вызывающей стороне при завершении корутины.

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

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

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

Дождитесь завершения фьючера, защитив его от отмены.

Выражение:

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 фьючеров и объектов корутин. Корутины будут обернуты в задачи. Возвращает два набора Future: (завершенные, ожидающие).

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

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

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

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

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

Пример использования:

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

Примечание

Это не вызывает asyncio.TimeoutError! Фьючеры, которые не завершены при наступлении таймаута, возвращаются во второй набор.

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

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

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

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

Если ожидание отменено, фьючер fut также отменяется.

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

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

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