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

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

Корутина – это генератор, который следует определённым соглашениям. Для целей документации все корутины следует декорировать с помощью @asyncio.coroutine, но это не может быть строго обязательным.

Корутины используют синтаксис yield from, введённый в PEP 380, вместо оригинального синтаксиса yield.

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

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

Что может делать корутина:

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

Вызов корутины не запускает её код – это всего лишь генератор, и объект корутины, возвращаемый вызовом, на самом деле является объектом генератора, который ничего не делает, пока по нему не пройти итерацией. В случае объекта корутины есть два основных способа запустить её выполнение: вызвать yield from корутина из другой корутины (если другая корутина уже выполняется!), или запланировать её выполнение с помощью функции async() или метода BaseEventLoop.create_task().

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

@asyncio.coroutine¶

Декоратор для пометки корутин.

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

import asyncio

@asyncio.coroutine
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

@asyncio.coroutine
def display_date(loop):
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        yield from 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

@asyncio.coroutine
def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    yield from asyncio.sleep(1.0)
    return x + y

@asyncio.coroutine
def print_sum(x, y):
    result = yield from 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() не завершится, прежде чем вернуть свой результат.

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

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

Диаграмма показывает поток управления, она не описывает точно, как всё работает внутри. Например, корутина sleep создаёт внутренний future, который использует BaseEventLoop.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_threadsafe().
• Этот класс несовместим с функциями 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

@asyncio.coroutine
def slow_operation(future):
    yield from 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

@asyncio.coroutine
def slow_operation(future):
    yield from 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. Задача (Task) является подклассом Future.

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

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

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

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

Don’t directly create Task instances: use the async() function or the BaseEventLoop.create_task() method.

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

classmethod all_tasks(loop=None)¶

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

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

classmethod current_task(loop=None)¶

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

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

None возвращается при вызове не в контексте задачи.

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

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

loop = asyncio.get_event_loop()
tasks = [
    asyncio.ensure_future(factorial("A", 2)),
    asyncio.ensure_future(factorial("B", 3)),
    asyncio.ensure_future(factorial("C", 4))]
loop.run_until_complete(asyncio.wait(tasks))
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' может возбуждать исключение
    # Используйте результат

Примечание

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

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

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

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

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

См. также

Метод BaseEventLoop.create_task().

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

Устаревший псевдоним ensure_future().

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

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

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

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

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

asyncio.iscoroutine(obj)¶

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

asyncio.iscoroutinefunction(obj)¶

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

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.4.4,: 3.5.1

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

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

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

Эта функция – корутина.

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

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

Возвращает менеджер контекста, который отменяет выполнение блока по истечении timeout:

with timeout(1.5):
    yield from inner()

1. Если inner() выполняется быстрее, чем за 1.5 секунд, ничего не происходит.
2. В противном случае inner() отменяется внутри, но asyncio.TimeoutError возбуждается за пределами области действия менеджера контекста.

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

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

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

timeout можно использовать для задания максимального времени ожидания в секундах перед возвратом. timeout может быть целым числом (int) или числом с плавающей точкой (float). Если 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.

Корутина будет обёрнута в Задача.

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

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

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

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

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