Глава 3. Прогулка по Asyncio

Чтобы применять Asyncio [на повседневной основе], вам необходимо знать о семи функциях.

-- Юрий Селиванов, автор PEP 492

Открывать официальную документацию Asyncio достаточно страшно. Имеется множество разделов с новыми, загадочными словами, а также понятия, с которыми не знакомы даже искушённые программисты Python, поскольку asyncio является очень новой вещью. Мы собираемся поломать всё это и объяснить какой подход следует применять к документации asyncio в дальнейшем, но на текущий момент вам следует знать, что реальная площадь поверхности, о которой вам придётся беспокоиться в имеющейся библиотеке asyncio намного меньше чем может показаться.

Юрий Селиванов, автор PEP 492 и основной разработчик асинхронного Python, объяснил в своём выступлении async/await в Python 3.5 и почему это восхитительно, представленном на PyCon 2016, относительно того, что большая часть API в модуле asyncio на самом деле предназначается для разработчиков инструментальных средств, а не для разработчиков конечных приложений. В э том выступлении он выделил те основные функции, о которых следует беспокоиться конечным пользователям, а эти функции всего лишь составляют небольшое подмножество всего API asyncio.

В данном разделе мы собираемся взглянуть на эти составляющие ядро функции и рассмотреть как вдарить по почве выполнения в цикле с помощью основанного на событиях программирования в Python.

Вам потребуются базовые знания сопрограмм (представленные в разделе, идущим вслед за этим), но помимо этого, если вы хотите иметь возможность использовать обсуждаемую библиотеку asyncio в качестве разработчика конечного приложения (а не как проектировщика инструментальных средств), вот моменты, которые вам необходимо знать в крошечном примере:

# quickstart.py
import time
import asyncio

async def main():
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')
    loop.stop()()

loop = asyncio.get_event_loop()()
loop.create_task(main())()
loop.run_forever()()
pending = asyncio.Task.all_tasks(loop=loop)
group = asyncio.gather(*pending, return_exceptions=True)()
loop.run_until_complete(group)(3)
loop.close()(5)

Вывод:
$ python quickstart.py
Sun Sep 17 14:17:37 2017 Hello!
Sun Sep 17 14:17:38 2017 Goodbye!
 	   

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

asyncio в Python представляет множество базовых механизмов вокруг основного цикла событий - и требует чтобы вы были осведомлены о таких вещах, как сам цикл событий и управление им на протяжении его времени жизни. Именно это является отличием, например, от Node.js, который также содержит некий цикл событий, но сохраняет его где- то вовне в скрытом виде. Однако, как только вы начнёте понемногу работать с asyncio, вы начнёте замечать, что ваш шаблон запуска и останова основного цикла событий не отходит страшно далеко от приведённого выше кода. А в оставшейся части данной книги мы рассмотрим более подробно некоторые нюансы относительно времени жизни цикла.

В приведённом выше примере я кое- что упустил. Самый последний элемент базовой функциональности, о котором вам следует знать, состоит в том как запускать блокирующие функции. Основной момент относительно кооперативной многозадачности состоит в том, что вам требуется все связанные с вводом/ выводом функции..., ну да, кооперировать, а это означает допуск переключения некоторого контекста обратно в данный цикл при помощи особого ключевого слова await. Большая часть имеющегося кода Python, доступного на текущий момент в диком виде, не делает этого, а вместо этого полагается на вас для запуска таких функций в потоках. До тех пор, пока не будет более широко распространённой поддержки функций async def, вы обнаружите, что применение таких библиотек с блокировкой неизбежно.

Для этого asyncio предоставляет некий API, который очень похож на API из пакета concurrent.futures. Данный пакет предоставляет некий ThreadPoolExecutor и какой- тоProcessPoolExecutor. По умолчанию он основан на потоке, но запросто заменяется на базирование на процессе. Я опустил это в своём предыдущем примере, так как это скрыло бы то описание, как подгоняются друг к другу все фундаментальные части. Теперь, когда они были обсуждены, мы можем взглянуть непосредственно на исполнителя.

Есть несколько причуд, о которых следует знать. Давайте рассмотрим пример кода:

# quickstart_exe.py
import time
import asyncio

async def main():
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')
    loop.stop()

def blocking():()
    time.sleep(0.5)()
    print(f"{time.ctime()} Hello from a thread!")

loop = asyncio.get_event_loop()

loop.create_task(main())
loop.run_in_executor(None, blocking)()

loop.run_forever()

pending = asyncio.Task.all_tasks(loop=loop)()
group = asyncio.gather(*pending)
loop.run_until_complete(group)
loop.close()

Вывод:
$ python quickstart_exe.py
Sun Sep 17 14:17:38 2017 Hello!
Sun Sep 17 14:17:38 2017 Hello from a thread!
Sun Sep 17 14:17:39 2017 Goodbye!
 	   

Теперь, когда вы ознакомились с наиболее существенными частями asyncio для потребностей разработки конечных приложений, наступило время объяснять нашу сферу и выстроить перечисленные API asyncio в некий вид иерархии. Это облегчит усвоение и понимание того как получать то что вам требуется из имеющейся документации, но не более того.

Как мы уже отметили в Быстром начале, существует лишь небольшое число команд, которые вам следует знать чтобы быть способным применять asyncio в качестве разработчика конечных приложений. К несчастью, имеющаяся документация для asyncio предоставляет гигантское число различных API, причём они представлены в очень "однообразном" формате, из которого очень трудно определить какие вещи предназначаются для общего применения, а какие стороны предоставляются для разработчиков инструментальных средств.

Когда разработчики инструментальных средств просматривают ту же самую документацию, они ищут точки входа, к которым они могут подключать свои новые инструментальные средства (или новые сторонние библиотеки). В данном разделе мы взглянем на asyncio глазами какого- то разработчика инструментальных средств чтобы дать представление как они могут подходить к построению новой обладающей асинхронностью библиотекой. Надеемся, это поможет нам ещё чётче очертить те функции, о которых вам стоит заботиться в вашей собственной работе.

С этой точки зрения более полезно представлять себе модуль asyncio как отранжированный в виде некоей иерархии, вместо некоего плоского листа, причём всякий последующий уровень строится поверх спецификации предыдущего. К сожалению, это не совсем так, и я взял на себя некие вольности при выравнивании их в Таблице 3-1, но, надеюсь, это снабдит вас неким альтернативным представлением имеющегося API asyncio.

	Предостережение
	Таблица 3-1, а также представляемая здесь названия и нумерация "Уровней" являются исключительно моими собственным изобретением, которое имеет целью добавить некую небольшую структуру в помощь пояснению API asyncio. Это моя выдумка! Искушённый читатель может выстраивать вещи иным образом, и это тоже хорошо!

На самом основополагающем уровне, Уровне 1, у нас имеются сопрограммы (coroutines), которые вы уже видели ранее в этой книге. Это самый нижний уровень, с которого следует начинать размышлять о разработке стороннего инструментального средства, и, что удивительно, он стал чем- то популярным не у какого- то одного, а у двух таких из доступных на сегодняшний день в мире асинхронных инфраструктур: Curio и Trio. Они обе полагаются только на естественные сопрограммы в Python и больше ни на что иное из библиотечного модуля asyncio.

Нашим следующим уровнем является собственно цикл событий. Сопрограммы бесполезны сами по себе: они ничего не могут делать без некоего цикла, в котором они запускаются (более того, с необходимостью, Curio и Trio реализуют свои собственные циклы событий). asyncio предоставляет как некое определение цикла, Abstract​ EventLoop, так и какую- то реализацию, BaseEventLoop.

Чёткое отделение описания от реализации делает возможным для сторонних разработчиков выполнять альтернативные реализации своих циклов событий и это уже произошло в проекте uvloop, который предоставляет намного более быструю реализацию цикла чем та, что имеется в стандартном библиотечном модуле asyncio. Что важно: uvloop просто "встраивается" в имеющуюся иерархию и подменяет только саму часть цикла в общем стеке. Такая возможность выполнять подобные вещи по выбору именно то, для чего разрабатывался API asyncio, причём с ясным разделением между всеми различными подвижными частями.

Уровни 3 и 4 привносят нам свойства и задачи, которые очень тесно взаимосвязаны; они разделяются только по той причине, что Task является неким подклассом Future, но их запросто можно рассматривать как находящиеся на одном и том же уровне. Некий экземпляр Future представляет какой- то вид происходящего действия, которое возвратит некий результат посредством уведомления в свой цикл событий, в то время как Task представляет некую сопрограмму, исполняемую в этом цикле событий. Краткая история такова: некое свойство является "осведомлённым о цикле", в то время как конкретная задача одновременно "осведомлена о цикле" и "знает о сопрограмме". Выступая в роли разработчика конечного продукта, вы будете преимущественно работать с задачами, нежели со свойствами, однако для некоторого разработчика инструментальных средств данная пропорция может быть иной в зависимости от имеющихся подробностей.

Уровень 5 предоставляет возможности для запуска, а также снабжения await работ, которые должны исполняться либо в отдельном потоке, либо даже в отдельном процессе.

Уровень 6 представляет дополнительные осведомлённые об асинхронности инструмента, такие как asyncio.Queue. Нам можно было разместить этот уровень после всех сетевых уровне, но я полагаю, что более чётко будет поместить все относящиеся к осведомлённым о сопрограммах API вначале, прежде чем мы взглянем на имеющиеся уровни ввода/ вывода. Те Queue, которые производятся asyncio очень похожи на API имеющихся Queue безопасных потоков в модуле queue, за исключением то8о, что наша версия asyncio требует наличия ключевого слова await для get() и put(). Вы не можете использовать queue.Queue непосредственно внутри сопрограмм, поскольку его get() заблокируют ваш основной поток {main}.

Наконец, у нас имеются уровни сетевого ввода/ вывода с 7 по 9. Для вас, как разработчика конечного продукта, большая часть подходящего вам для работы API это API "потоков" на уровне 9. Я представил этот API потоков на самом верхнем уровне абстракции в нашей башне. Непосредственно под ним расположен уровень "протоколов" (Уровень 8), является более тонко настраиваемым API по сравнению с "потоками"; вы можете применять "протоколы" во всех экземплярах, в которых вы можете применять уровень "потоков", но "потоки" проще. Наконец, вам скорее всего никогда не придётся работать непосредственно с транспортным уровнем (Уровень 7), если только вы не создаёте некое инструментальное средство для других, чтобы применять его и вам необходимо персонализировать настройку имеющихся обменов {Прим. пер.: например, вы пожелаете предоставить потоки на основе протокола, применяющего обмен RDMA для увеличения пропускной способности и снижения латентности, см. например, Рекомендации по разработке высокопроизводительных систем RDMA и Ceph поверх RDMA.}

D Быстром начале мы рассмотрели абсолютно необходимый минимум, который следует знать чтобы начать работу с библиотекой asyncio. Теперь, после того как мы взглянули на собранную воедино библиотеку API asyncio в целом, я бы хотел повторно вернуться к короткому перечню свойств и повторно выделить те части, которые вам скорее всего следует изучить.

Для написания сетевых приложений вам следует сосредоточиться на изучении следующих наиболее важных уровней для применения библиотечного модуля asyncio;

Конечно, если вы применяете некую совместимую с asyncio библиотеку стороннего разработчика, которая обрабатывает всё взаимодействие с нужным вам сокетом, скажем, aiohttp, вам вовсе не требуется напрямую работать с asyncio. В этом случае вам следует в основном полагаться на предоставляемую этими библиотеками документацию.

Это подводит нас к завершению данного раздела. Библиотека asyncio пытается предоставлять достаточную функциональность как для разработчиков конечных приложений, так и для разработчиков инструментальных средств. К сожалению, это означает, что весь API asyncio может показаться слишком размашистым. Я надеюсь, что данный раздел смог снабдить вас сырой "дорожной картой" чтобы помочь вам уцепить только те части, которые вам требуются.

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

Совет

Сайт pythonsheets предоставляет некое более глубокое резюме (или "шпаргалку", если хотите) крупных фрагментов имеющегося API asyncio, где все концепции снабжены короткими отрывками исходного кода. Данная презентация является краткой, поэтому я не рекомендую её начинающим, но если вы опытны в Python и вы из тех людей, которым требуется только "ухватить" новую информацию о программировании в представленном коде, несомненно, это будет полезный вам ресурс.

Давайте начнём с самого начала: что такое сопрограмма?

Я собираюсь позволить вам быстро заглянуть под капот и увидеть некоторые части того механизма, который вы обычно не видите, даже когда применяете на протяжении повседневного программирования. Все приводимые далее примеры могут быть целиком воспроизведены в обычном интерпретаторе Python в интерактивном режиме и я призываю вас поработать с ними самостоятельно набирая их вручную, отслеживая их вывод и, возможно, экспериментируя различными путями с async и await.

Предостережение

asyncio впервые был добавлен в Python 3.4, однако применяемый нами новый синтаксис для сопрограмм с использованием async def и await был добавлен только в Python 3.5. Как люди делают что- то с asyncio в Python 3.4? Они применяют генераторы (generator) очень специальными способами для действия таким образом, как если бы они были сопрограммами. В некоторых более старых базовых кодах вы обнаружите функции генераьора, декорированные @asyncio.coroutine и содержащими операторы yield from. Создаваемые с помощью async def сопрограммы теперь именуются как "натуральные", так как они построены на том же самом языке, что и сопрограммы и ни на чём ином. Данная книга целиком игнорирует более старые сопрограммы, основанные на генераторе.

Давайте начнём с самой простой из возможных вещей:

>>> async def f():()
... return 123
...
>>> type(f)()
<class 'function'>
>>> import inspect()
>>> inspect.iscoroutinefunction(f)()
True
 	   

Возвращаясь к нашей функции async def f(), что же происходит когда мы её вызываем?

>>> coro = f()
>>> type(coro)
<class 'coroutine'>
>>> inspect.iscoroutine(coro)
True
 	   

Итак, теперь вопрос звучит следующим образом: что же такое в точности "сопрограмма"? Сопрограммы очень походи на генераторы. В самом деле, прежде чем с помощью ключевых слов async def и await в Python 3.5 были введены натуральные сопрограммы, они уже были доступны к применению в библиотеке Asyncio Python 3.4 с использованием обычных генераторов при особых декораторах. [Более того, именно таким образом библиотеки открытого исходного кода, такие как Twisted и Tornado выставляли поддержку асинхронности в недалёком прошлом.] Не является неожиданным, что наши новые функции async def (и те сопрограммы, которые они возвращают) ведут себя аналогично генераторам.

Мы можем ещё немного поиграть с сопрограммами, чтобы посмотреть как Python их применяет. Что ещё более важно, мы желаем увидеть как Python способен "переключать " исполнение между сопрограммами. Давайте вначале взглянем на то как может быть получено само значение return.

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

>>> async def f():
... return 123
>>> coro = f()
>>> try:
... coro.send(None)()
... except StopIteration as e:
... print('The answer was:', e.value)()
...
The answer was: 123
		

Эти два момента, т.е. send() и StopIteration определяют соответственные начало и окончание самого исполнения сопрограммы. До сих пор это выглядело просто как в действительности запутанный способ запуска некоей функции, но всё хорошо: наш цикл событий будет отвечать за продвижение сопрограмм по этим внутренним особенностям нижнего уровня. Со своей точки зрения вы просто планируете сопрограммы к исполнению в имеющемся цикле, а они исполняются сверху вниз почти как обычные функции. [⁷]

Нашим следующим шагом будет рассмотреть как само исполнение такой сопрограммы может быть подвешено.

Данное новое ключевое слово, await, всегда берёт некий параметр и будет принимать единственную вещь, называемую ожидаемой (awaitable), которая определяется как одна из следующих (исключительно!):

Второй из доступных случаев ожиданий выходит за рамки данной книги (он вам никогда не понадобится в повседневном программировании asyncio), тем не менее, самый первый пример достаточно очевиден:

async def f():
    await asyncio.sleep(1.0)
    return 123

async def main():
    result = await f()()
    return result
 	   

Прежде чем мы завершим данный раздел и приступим к нашему циклу событий, полезно взглянуть на то, как могут подаваться исключительные ситуации, которые наиболее часто применяются для завершения: когда вы вызываете task.​cancel(), ваш цикл событий внутренне применит coro.throw() для возбуждения asyncio.CancelledError внутри вашей сопрограммы:

>>> coro = f()()
>>> coro.send(None)
>>> coro.throw(Exception, 'blah')()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 2, in f
Exception: blah
blah
 	   

Метод throw() применяется (внутренним образом в asyncio) для прекращения задачи, что мы также достаточно просто можем продемонстрировать. Мы даже собираемся забежать вперёд и обработать данное завершение внутри некоторой новой сопрограммы, которая и обработает это:

>>> import asyncio
>>> async def f():
...     try:
...         while True: await asyncio.sleep(0)
...     except asyncio.CancelledError:()
...         print('I was cancelled!')()
...     else:
...         return 111
>>> coro = f()
>>> coro.send(None)
>>> coro.send(None)
>>> coro.throw(asyncio.CancelledError)()
I was cancelled!()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration)
 	   

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

>>> async def f():
...     try:
...         while True: await asyncio.sleep(0)
...     except asyncio.CancelledError:
...         print('Nope!')
...         while True: await asyncio.sleep(0)()
...     else:
...         return 111
>>> coro = f()
>>> coro.send(None)
>>> coro.throw(asyncio.CancelledError)()
Nope!
>>> coro.send(None)()
 	   

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

На данный момент мы, естественно, очень устали выступать в роли некоего цикла событий вручную, выполняя все вызовы .send(None), поэтому вместо этого мы привнесём соответствующий цикл, предоставляемый asyncio и очистим предыдущий пример надлежащим образом:

>>> async def f():
...     await asyncio.sleep(0)
...     return 111
>>> loop = asyncio.get_event_loop()()
>>> coro = f()
>>> loop.run_until_complete(coro)()
111
 	   

В своём предыдущем разделе мы показали как методы send() и throw() могут взаимодействовать с некоторой сопрограммой, но это было сделано только для понимания того как структурированы сами по себе сопрограммы. Именно цикл событий в asyncio обрабатывает все переключения между сопрограммами, а также перехватывает необходимые исключительные ситуации StopIteration, и многое другое, например, выполняет ожидание в сокетах и файловых дескрипторах на предмет событий.

Вы можете получить некий цикл событий посредством вызова get_event_loop(), которая является весьма интересной функцией:

>>> loop = asyncio.get_event_loop()
>>> loop2 = asyncio.get_event_loop()
>>> loop is loop2()
True
 	   

Это означает, что если вы внутри некоторой функции сопрограммы, а у вас имеется потребность в доступе к своему экземпляру цикла, будет достаточно вызвать get_event_loop() для его получения. Вам не требуется передавать какой- то параметр loop в явном виде во все свои функции. Эта ситуация меняется коренным образом если вы разработчик инструментального средства: раз так, было бы лучше спроектировать ваши функции на приём некоторого параметра loop, просто на тот случай, когда ваши пользователи делают нечто необычное с политиками цикла событий. Политики выходят за рамки данной книги, и мы больше не будем обсуждать их.

Совет

Метод get_event_loop() работает только в пределах того же самого потока: на самом деле, get_event_loop() выдаст отказ в случае вызова внутри какого- то нового потока, если вы только не создадите намеренно некий новый цикл при помощи new_event_loop() и установите такой новый экземпляр в качестве "основного" цикла для этого потока через вызов set_event_loop(). Большинству из нас никогда не понадобится (и не захочется!) иметь некий отдельный экземпляр цикла запущенным в каом- то отдельном потоке. В первую очередь это касается практически всего асинхронного программирования.

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

async def f():
    # Создадим какие- то задачи!
    loop = asyncio.get_event_loop()
    for i in range():
    loop.create_task(<some other coro>)
 	   

В этом примере замысел состоит в запуске полностью новых задач внутри данной сопрограммы. Без выполнения их ожидания, они запускаются независимо от своего контекста внутри функции сопрограммы f(). На самом деле, f() осуществит выход до того как завершатся запускаемые ею задачи.

Для ситуации, подобной приводимой, вам порой следует посмотреть код, в котором переменная значения цикла передаётся в качестве какого- то параметра, просто чтобы сделать её доступной для возможности вызова соответствующего метода create_task(). Определённой альтернативой является применение неудачно именованного asyncio.ensure_future(), который делает всё то же самое что и create_task(), но не требует некоторой переменной локального цикла:

async def f():
    # Создадим какие- то задачи!
    for i in range():
        asyncio.ensure_future(<some other coro>)
 	   

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

def create_task(coro):
    return asyncio.get_event_loop().create_task(coro)

async def f():
    # Создадим какие- то задачи!
    for i in range():
        create_task(<some other coro>)
 	   

Единственная разница между loop.create_task() и asyncio​.ensure_future() состоит в заголовке и введении в заблуждение вновь прибывших. Мы поясним эти различия в нашем следующем разделе.

В своём предыдущем разделе мы обсудили сопрограммы и как они должны запускаться в некотором цикле чтобы они стали полезными. Здесь мы кратко поговорим об API Task и Future. То, с чем вы будете работать наиболее часто, это Task, поскольку большинство вашей работы будет содержать выполнение сопрограмм при помощи метода loop.create_task(), точно так же, как это было установлено ранее в разделе Быстрое начало. Класс Future на самом деле является суперклассом Task и предоставляет всю функциональность для работы с основным циклом.

Проще всего это воображать себе так: Future (Фьючерс) представляет состояние завершения некоторого действия и управляется самим циклом, в то время как Task в точности то же самое и то, где специфическим "действием" является некая сопрограмма; возможно, одна из созданных вами при помощи async def функция плюс loop.create_task().

Класс Future представляет некое состояние чего- то, что взаимодействует с циклом. Это описание слишком расплывчатое чтобы быть полезным, поэтому представляйте себе вместо этого некий экземпляр Future как такое: это именно то, что выступает переключателем для состояния завершения. Когда некий экземпляр Future создан, такой переключатель "ещё пока не завершён"; однако через некоторое время спустя он завершится. На самом деле некий экземпляр Future имеет метод с названием done():

>>> from asyncio import Future
>>> f = Future()
>>> f.done()
False
		

Некий экземпляр Future также может:

Даже несмотря на то, что Tasks более распространены, вы не можете полностью избежать Future: например, запуск некоей функции в каком- то исполнителе возвратит экземпляр Future, а не Tasks. {Прим. пер.: как определяет словарь, фьючерс- программная конструкция, указывающая на то, что результат некоего действия будет использоваться в программе позже, но само вычисление может планироваться системой в любой произвольный момент времени.} Давайте быстро взглянем на код примера чтобы получить представление о том, как работать с неким экземпляром Future напрямую:

>>> import asyncio
>>> async def main(f: asyncio.Future):()
...     await asyncio.sleep(1)
...     f.set_result('I have finished.')()
>>> loop = asyncio.get_event_loop()
>>> fut = asyncio.Future()()
>>> print(fut.done())()
False
>>> loop.create_task(main(fut))()
<Task pending coro=<main() running at <ast>:4>>
>>> loop.run_until_complete(fut)()
'I have finished.'
>>> print(fut.done())
True
>>> print(fut.result())()
I have finished.
		

Конечно, маловероятно что вы будете работать с Future напрямую, как мы показали это выше. Данный пример кода приводится исключительно для обучения. Большая часть ваших контактов с asyncio будет происходить через экземпляры Task.

Один из самых последних примеров, чтобы доказать что некая Task на самом деле какое- то небольшое декорирование в Future: мы можем повторить точно тот же самый пример что и выше, но вместо этого воспользовавшись Task:

>>> import asyncio
>>> async def main(f: asyncio.Future):
...     await asyncio.sleep(1)
...     f.set_result('I have finished.')
>>> loop = asyncio.get_event_loop()
>>> fut = asyncio.Task(asyncio.sleep(1_000_000))()
>>> print(fut.done())
False
>>> loop.create_task(main(fut))
<Task pending coro=<main() running at <ast>:4>>
>>> loop.run_until_complete(fut)
'I have finished.'
>>> print(fut.done())
True
>>> print(fut.result())
I have finished.
		

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

В Быстром начале мы сказали, что верный путь для запуска сопрограмм состоит в использовании loop.create_task(coro). Это возвращает нас к тому, что этого же можно достичь другой функцией уровня модуля: asyncio.ensure_future().

В процессе исследований для данной книги я убедился, что данный метод asyncio.ensure_future() несёт ответственность за большую часть распространённого недоразумения относительно самой библиотеки asyncio. Основная часть имеющегося API является достаточно ясной, однако имеется пара в которых можно споткнуться при обучении, и именно это один из них. Когда вы сталкиваетесь с ensure_future(), ваше сознание очень усердно работает над тем как интегрировать его в умственное построение модели того как следует применять asyncio - и терпит неудачу!

Основная проблема с ensure_future() лучше всего высвечивается этим имеющим теперь дурную репутацию пояснением из документации asyncio самого Python:

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

Планирование собственно исполнения некоторого объекта сопрограммы: его обёртывание в фьючерсе. Возвращает некий объект Task. Если в качестве аргумента выступает Future, он возвращается непосредственно.

-- Документация Python 3.6 {Прим. пер.: см. изменения в Python 3.7}

Что-а!? К счастью, кот более ясное описание ensure_future():

Давайте подробнее рассмотрим как это работает:

import asyncio

async def f():()
    pass

coro = f()()
loop = asyncio.get_event_loop()()

task = loop.create_task(coro)()
assert isinstance(task, asyncio.Task)()

new_task = asyncio.ensure_future(coro)()
assert isinstance(new_task, asyncio.Task)

mystery_meat = asyncio.ensure_future(task()
assert mystery_meat is task()
 	   

Итак: что именно является непосредственным пунктом передачи экземпляров Future? И зачем делать два разных дела в одной функции? Ответ состоит в том, что данная функция, ensure_future(), предназначается для применения авторами инструментальных средств для предоставления API разработчиком конечных приложений и которая может обрабатывать оба вида параметров {даже три!}. Не верите мне? Вот что следует от самого Благословенного Диктатора Всея Жизни (BDFL):

Основной момент ensure_future() состоит в том, что если у вас есть нечто, что может быть либо сопрограммой, либо неким Future (и этот последний содержит некую Task, поскольку она является субклассом Future), и вы желаете иметь возможность вызывать некий метод в нём, который определён только в Future (скорее всего, единственным полезным примером является cancel()). Когда это уже Future (или Task) не делается ничего; когда это некая сопрограмма, она обёртывается некоторой Task.

Если вы знаете что у вас имеется некая сопрограмма и вы желаете её запланировать, более верным API для применения будет create_task(). Единственный случай, в котором вам следует вызывать ensure_future() это когда вы предоставляете некий API (как это делает большая часть владельцев API), который принимает либо некую сопрограмму, либо какой- то Future и вам требуется делать нечто с этим, и что требует чтобы у вас имелся Future_[10].

-- Гвидо ван Россум, github.com/python/asyncio/issues/477

Итак, суммируем: наш API asyncio.ensure_future() является некоторой вспомогательной функцией предназначенной для разработчиков инструментальных средств. Она проще всего поясняется по аналогии с более распространённым видом функции: если у вас есть за плечами пять лет опыта программирования, вы вы уже могли видеть функции подобные приводимой ниже listify():

def listify(x: Any) ­> List:
    """ Try hard to convert x into a list """
    if isinstance(x, (str, bytes)):
        return [x]

    try:
        return [_ for _ in x]
    except TypeError:
        return [x]
 	   

Эта функция пытается преобразовать полученный аргумент в некий перечень, вне зависимости от того что она получает. Такой вид функция часто применяется в API и инструментальных средствах для принуждения вводов в некий известный тип, что упрощает последующий код, так как вы знаете что этот параметр (выводимый из listify()) всегда будет неким списком.

Если я переименую свою функцию listify() в ensure_list(), то вы должны начать замечать определённую параллель с asyncio.ensure_future(), не так ли? Он всегда пытается принудительно преобразовывать получаемый аргумент в некий тип Future. Эта утилитарная функция чтобы делать жизнь более простой для разработчиков инструментальных средств, а не для разработчиков окончательных продуктов, коими являемся вы и я.

И на самом деле, даже стандартная библиотека asyncio сама по себе применяет ensure_future() в точности для этой цели. Когда вы в следующий раз будете просматривать API, вы повсеместно будете видеть некий параметр функции, описываемый как coro_or_future, и именно его, скорее всего, ensure_future() будет применять внутри себя для принудительного преобразования данного параметра. Например, имеющаяся функция asyncio.gather() имеет такуб отличительную особенность:

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

Внутри себя gather() применяет using ensure_future() для принудительного приведения типа и это делает для вас возможным передавать ему сопрограммы, задачи или фьючерсы.

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

К сожалению, это не самый конец данной истории. Прямо сейчас некоторые разработчики конечных продуктов предпочитают применять asyncio.ensure_future() вместо loop.create_task() по одной простой и очень прагматичной причине: так меньше работы! Чтобы вызвать create_task() вам либо требуется некий экземпляр loop, доступный в вашем локальном пространстве имён, в котором вы пишите код, или же вам необходим некий дополнительный вызов asyncio.get_event_loop() для получения соответствующей ссылки на loop; в противоположность этому, ensure_future() может вызываться как есть.

В приведённом выше примере у нас имелось три функции сопрограмм внутри которых будут запущены некие фоновые сопрограммы. Основная цель данного упражнения состояла в сопоставлении эстетических преимуществ применения create_task() по отношению к ensure_future() при планировании сопрограмм в основном цикле.

import asyncio

async def background_job():
    pass

async def option_A(loop):()
    loop.create_task(background_job())

async def option_B():()
    asyncio.ensure_future(background_job())

async def option_C():()
    loop = asyncio.get_event_loop()
    loop.create_task(background_job())

loop = asyncio.get_event_loop()

loop.create_task(option_A(loop))(1)
loop.create_task(option_B)(2)
loop.create_task(option_C)(3)
 	   

Что в действительности следует иметь доступным, так это вспомогательную функцию asyncio.create_task​ (coro), которая делает в точности то же самое что и наш "Вариант C"? приведённый выше, но предопределённой в имеющейся стандартной библиотеке. Это низложит удобство ensure_future() и сохранит ясность loop.create_task(). Эта потребность не замечалась командой разработки Python, но я счастлив сообщить, что asyncio.create_task() в конечном счёте будет доступна в Python 3.7!

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

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

Основным ключом к пониманию async with является осознание того, что определённая операция некоторого диспетчера контекста управляется вызовами метода; с последующим рассмотрением соображения: а что если бы такие методы были бы функциями сопрограмм? И на самом деле, это в точности так и работает. Вот псевдокод, демонстрирующий всё это:

class Connection:
    def __init__(self, host, port):
        self.host = host
        self.port = port
    async def __aenter__(self):()
        self.conn = await get_conn(self.host, self.port)
        return conn
    async def __aexit__(self, exc_type, exc, tb):()
        await self.conn.close()

async with Connection('localhost', 9001) as conn:
    <do stuff with conn>
 	   

	Предостережение
	просто потому что вы должны применять asyncio в своей программе, это не означает, что все ваши диспетчеры контекста обязаны синхронизироваться именно так! Такой порядок полезен только если вам требуется выполнять await чего- то внутри своих методов enter и exit. Если нет никаого кода ввода/ вывода с блокировкой, просто применяйте диспетчеры контекста.

А теперь - только между нами - мне не сильно нравится такой стиль применения диспетчера контекста в явном виде, в то время как имеется великолепный декоратор @contextmanager в соответствующем модуле contextlib стандартной библиотеки! Как вы можете догадаться, имеется и некая асинхронная версия, @asynccontextmanager, но, к сожалению, она доступна только начиная с Python 3.7, который ещё не был доступен на момент написания данной книги {Прим. пер.: На момент перевода, всё уже хорошо, Python 3.7 доступен.} Тем не менее, достаточно просто скомпилировать Python из исходного кода и в нашем следующем разделе покажем как @asynccontextmanager будет работать в Python 3.7.

Он аналогичен декоратору @contextmanager из соответствующего модуля contextlib стандартной библиотеки. Для краткого повтора, давайте вначале рассмотрим соответствующий вариант с блокировкой:

from contextlib import contextmanager

@contextmanager()
def web_page(url):
    data = download_webpage(url)()
    yield data
    update_stats(url)()

with web_page('google.com') as data:()
    process(data)()
 	   

Давайте сопоставим в точности тот же самый пример, но на этот раз и применением новой вспомогательной функции, осведомлённой об асинхронности, привносимой Python 3.7:

from contextlib import asynccontextmanager

@asynccontextmanage()
async def web_page(url):()
    data = await download_webpage(url)()
    yield data()
    await update_stats(url)()

async with web_page('google.com') as data:()
    process(data)
 	   

К счастью, данный пример показывает, что наш новый @asynccontextmanager в точности аналогичен имевшемуся ранее декоратору @contextmanager.

В сноске 3 предыдущего примера я сказал, что было необходимо изменить некоторые функции для возврата сопрограмм; ими были download_webpage() иupdate_stats(). Обычно это не так просто сделать, так как необходимо добавить поддержку асинхронности на нижнем уровне сокета.

Основной фокус предыдущего примера был нацелен на то что бы просто показать наш новый @asynccontextmanager, а не показывать как преобразовывать функции с блокировкой в неблокирующие. Более общий случай состоит в том, что вам может понадобиться применять в своей программе некую функцию с блокировкой, но не будет возможности изменять имеющийся в таких функциях код.

Такая ситуация часто происходит с библиотеками сторонних разработчиков и прекрасным примером может послужить соответствующая библиотека requests, которая внутри себя применяет вызовы с блокировкой [Может оказаться весьма сложным добавление поддержки Async в некое имеющееся инструментальное средство по данной причине, так как могут потребоваться большие структурные изменения. Это обсуждается в эссе для Requests github]. Хорошо, раз вы не можете изменить подлежащий вызову код, имеется иной путь, и это удобное место показать как именно для этого можно применять Исполнитель (executor):

from contextlib import asynccontextmanager

@asynccontextmanager
async def web_page(url):()
    loop = asyncio.get_event_loop()
    data = await loop.run_in_executor(
        None, download_webpage, url)()
    yield data
    await loop.run_in_executor(None, update_stats, url)()

async with web_page('google.com') as data:
    process(data)
 	   

Скорее всего асинхронные диспетчеры контекста станут широко применяться во многих кодах, основанных на asyncio, следовательно достаточно важно иметь о них хорошее представление. Вы можете прочесть дополнительные подробности о новом асинхронном декораторе диспетчера контекста в документации Python 3.7.

Замечание

asyncio всё ещё находится в стадии активной разработки командой разработчиков Python и в Python 3.7 доступны некие прочие небольшие, но существенные улучшения помимо asynccontextmanager, включая следующие небольшие примеры: asyncio.run(): некая новая вспомогательная функция, предназначенная для применения в качестве основной точки входа для программ asyncio. asyncio.create_task(): создаёт некую задачу без необходимости указания в явном виде идентификатора цикла. AbstractEventLoop.sock_sendfile(): отправляет некий файл поверх сокета TCP с применением высокопроизводительного API os.sendfile(). AbstractEventLoop.start_tls(): обновляет некое имеющееся соединение до безопасного транспортного уровня (TLS). asyncio.Server.serve_forever(): API очистки для запуска сетевых серверов asyncio.

Помимо async def и await имеется ещё несколько расширений в синтаксисе языка Python. Первым из них является асинхронная версия для цикла "for". Проще всего понять как это работает если вы вначале поймёте что простая итерация - так же как и многие другие функции языка программирования - реализуется при помощи специальных методов, отличающих их двойными подчёркиваниями в соответствующих названиях.

Например, вот как некий стандартный (не асинхронный) итератор определяется посредством применения методов __iter__() и __next__():

>>> class A:
...     def __iter__(self):()
...         self.x = 0()
...         return self()
...     def __next__(self):()
...         if self.x > 2:
...             raise StopIteration()
...         else:
...             self.x += 1
...             return self.x()
>>> for i in A():
...     print(i)
123
 	   

Теперь вы зададите справедливый вопрос: что произойдёт если вы объявите свой особый метод __next__() в качестве некоторой функции сопрограммы async def? Это позволит ему выполнять await некоторого вида операций, ограниченных вводом/ выводом; а также это во многом в точности повторяет то, как работает async for, за исключением маленьких деталей вокруг именований. Имеющаяся спецификация (в PEP 492) показывает, что чтобы применять async for в качестве какого- то асинхронного итератора в самом этом асинхронном итераторе требуются определённые моменты:

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

import asyncio
from aioredis import create_redis

async def main():()
    redis = await create_redis(('localhost', 6379))()
    keys = ['Americas', 'Africa', 'Europe', 'Asia']()

    async for value in OneAtATime(redis, keys):()
        await do_something_with(value)()

class OneAtATime:
    def __init__(self, redis, keys):()
        self.redis = redis
        self.keys = keys
    def __aiter__(self):()
        self.ikeys = iter(self.keys)
        return self
    async def __anext__(self):()
        try:
            k = next(self.ikeys)()
        except StopIteration:()
            raise StopAsyncIteration

        value = await redis.get(k)()
        return value

asyncio.get_event_loop().run_until_complete(main())
 	   

Надеемся, что этот пример является понятным: async for предоставляет возможность оставаться в удобном простом цикле for, даже при итерации данных при которых сама по себе итерация выполняет ввод/ вывод. Основное преимущество от этого состоит в том что вы можете обрабатывать гигантские объёмы данных, причём всё это делать в отдельном цикле, так как вам придётся всякий раз иметь дело только с каждым фрагментом в небольших партиях.

Асинхронные генераторы отвечают на законный вопрос: "Что произойдёт если применить yield внутри естественной функции сопрограммы async def?" Эта концепция может вводить в заблуждение если у вас имеется некий опыт применения генераторов как если бы они были сопрограммами, так как это происходит в инструментальном средстве Twisted или в инфраструктуре Tornado, или даже как в случае yield from из asyncio Python 3.4.

Следовательно, прежде чем мы продолжим данный раздел, было бы лучше, чтобы вы убедили себя что:

Наш пример, использованный в предыдущем разделе для демонстрации какого- то асинхронного итератора при итерациях по Redis оказывается намного проще если мы установим некий асинхронный генератор:

import asyncio
from aioredis import create_redis

async def main():()
    redis = await create_redis(('localhost', 6379))
    keys = ['Americas', 'Africa', 'Europe', 'Asia']

    async for value in one_at_a_time(redis, keys):()
        await do_something_with(value)

async def one_at_a_time(redis, keys):()
    for k in keys:
        value = await redis.get(k)()
        yield value()

asyncio.get_event_loop().run_until_complete(main())
 	   

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

Теперь, когда мы рассмотрели как Python поддерживает асинхронные итерации, следующим естественным вопросом будет спросить понимает ли он также списки в for - и наш ответ ДА! Данная поддержка была введена в PEP 530 и я рекомендую вам просмотреть этот PEP самостоятельно. Он достаточно короткий и хорошо читаемый.

>>> import asyncio
>>> async def doubler(n):
...     for i in range(n):
...         yield i, i * 2()
...         await asyncio.sleep(0.1)()
>>> async def main():
...     result = [x async for x in doubler(3)]()
...     print(result)
...
...     result = {x: y async for x, y in doubler(3)}()
...     print(result)
...
...     result = {x async for x in doubler(3)}()
...     print(result)
>>> asyncio.get_event_loop().run_until_complete(main())
[(0, 0), (1, 2), (2, 4)]
{0: 0, 1: 2, 2: 4}
{(1, 2), (0, 0), (2, 4)}
 	   

Обратной стороной нашей монеты, как выводит PEP 530, является применение await внутри выразительных средств. Но это не совсем так: await <coro> является обычным выражением и оно может применяться в большинстве мест, в которых вы это можете ожидать.

Именно async for является тем, что превращает выразительность в асинхронную выразительность, а не наличие await. Всё что вам требуется для допустимости await (внутри некоторого поглощения), это то, что вы находитесь внутри самого тела какой- то функции сопрограммы, то есть некоторой функции, объявленной с помощью async def. Поэтому хотя применение await и async for внутри того же самого поглощаемого списка в действительности сочетает два различных подхода, давайте сделаем это в любом случае чтобы продолжить снижать порого чувствительности вхождения в комфортное состояние от применения синтаксиса асинхронного языка программирования:

>>> import asyncio
>>> async def f(x):()
...   await asyncio.sleep(0.1)
...   return x + 100
>>> async def factory(n):()
...   for x in range(n):
...     await asyncio.sleep(0.1)
...     yield f, x()
>>> async def main():
...   results = [await f(x) async for f, x in factory(3)]()
...   print('results = ', results)
>>> asyncio.get_event_loop().run_until_complete(main())
results = [100, 101, 102]
 	   

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

Из этих двоих, запуск намного проще. Самый стандартный способ запуска некоторого приложения asyncio состоит в создании какой- то задачи, с последующим вызовом loop.run_forever(), как это было показано в нашем примере Hello World из раздела Быстрое начало.

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

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

Останов является намного более интригующим.

Для останова вначале мы рассмотрим соответствующий танец, который следует за тем, когда нечто останавливает имеющийся цикл событий. Когда некий исполняемый цикл останавливается, соответствующий вызов run_forever() выполняет снятие блокировки и выполняется появляющийся после этого код. В этот момент вы обязаны:

Только после этого останов завершится.

Обряд посвящения в построение ваших первых нескольких прикладных программ asyncio будет состоять из попыток избавления от сообщений подобных Task was destroyed but it is pending! {Задача была уничтожена, но она приостановлена!} на протяжении останова. Это происходит по причине пропуска одного или более перечисленных выше этапов. Вот пример данной ошибки:

# taskwarning.py
import asyncio

async def f(delay):
    await asyncio.sleep(delay)

loop = asyncio.get_event_loop()
t1 = loop.create_task(f(1))()
t2 = loop.create_task(f(2))()
loop.run_until_complete(t1)()
loop.close()
 	   

Вывод:

$ python taskwarning.py
Task was destroyed but it is pending!
task: <Task pending coro=<f() done, defined at [...snip...]>>
		

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

Давайте рассмотрим более подробно код из своего примера Быстрого начала и рассмотрим эти фазы вновь. Мы проделаем это через мини- обучение с помощью эхо сервера на основе telnet.

На Рисунке 3-1 наш сервер запущен с левой стороны панели. Затем, справа, некий сеанс telnet выполняет подключение к нашему серверу. Мы набираем несколько команд, которые возвращаются к нам обратно (все приведённые в верхний регистр) и после этого мы отключаемся. Теперь мы рассмотрим тот код, который управляет данной программой.

 

Рисунок 3-1

Взаимодействие Telnet при помощи эхо сервера

from asyncio import (()
    get_event_loop, start_server, CancelledError,
    StreamReader, StreamWriter, Task, gather)

async def echo(reader: StreamReader, writer: StreamWriter):()
    print('New connection.')
    try:
        while True:()
            data: bytes = await reader.readline()()
            if data in [b'', b'quit']:
                break
            writer.write(data.upper())()
            await writer.drain()
        print('Leaving Connection.')
    # except CancelledError:()
    # writer.write_eof()
    # print('Cancelled')
    finally:
        writer.close()

loop = get_event_loop()
coro = start_server(echo, '127.0.0.1', 8888, loop=loop)()
server = loop.run_until_complete(coro)()

try:
    loop.run_forever()()
except KeyboardInterrupt:
    print('Shutting down!')

server.close()()
loop.run_until_complete(server.wait_closed())()

tasks = Task.all_tasks()()
for t in tasks:
    t.cancel()
group = gather(*tasks, return_exceptions=True)()
loop.run_until_complete(group)()
loop.close()
 	   

Надеюсь, вы начинаете различать знакомый шаблон.

Совет

Один самый последний момент прежде чем мы переместимся: Если вы перехватываете CancelledError внутри некоторой сопрограммы, вам следует быть аккуратеым и не создавать никаких новых задач внутри данного обработчика прерываний. И вот почему: тот вызов all_tasks(), который выдаётся в 12 не собирается быть информированным о каких- либо новых задачах, создаваемых на протяжении данной фазы run_until_complete() при вызове 14 - что происходит когда данный код внутри ваших обработчиков прекращения когда они будут исполняться. Поэтому основное правило гласит: никаких новых задач внутри обработчиков исключительной ситуации CancelledError, если только вы также не снабжаете их await внутри какого- то обработчика исключительной ситуации CancelledError.

И запомните: если вы применяете некую библиотеку или какое- то инструментальное средство, убедитесь что вы следуете их документации относительно того как выполнять запуск и останов. Инструментальные средства сторонних разработчиков обычно предоставляют свои собственные функции запуска и останова, к тому же они предоставляют особые точки входа событий для индивидуализации. Вы можете увидеть это в примере из данной книги с инструментальным средством Sanic позднее в этой книге в Примере: недостоверность кэширования.

Вы могли заметить, что я установил значение аргумента с ключевым словом return_exceptions=True в своём вызове gather() при исполнении 13 в нашем предыдущем примере кода. Я также это делал ранее в Быстром начале, причём я очень подло не сказал ничего об этом в тот раз. Настал момент для пояснений.

Значением по умолчанию является gather(..., return_exceptions=False). Такое значение по умолчанию является проблематичным для нашего процесса останова. Это слегка сложно объяснить напрямую; вместо этого давайте пошагово пройдём некую последовательность фактов,которая сделает более простым понимание этого:

И здесь у вас именно это: необходимая взаимосвязь между return_exceptions=True и run_until_complete()!

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

import asyncio

async def f(delay):
    await asyncio.sleep(1 / delay)()
    return delay

loop = asyncio.get_event_loop()
for i in range(10):
    loop.create_task(f(i))
pending = asyncio.Task.all_tasks()
group = asyncio.gather(*pending, return_exceptions=True)
results = loop.run_until_complete(group)
print(f'Results: {results}')
loop.close()
 	   

Вывод:

$ python alltaskscomplete.py
Results: [6, 9, 3, 7, ...
          ZeroDivisionError('division by zero',), 4, ...
          8, 1, 5, 2]
		

Без return_exceptions=True, соответствующая ZeroDivisionError была бы возбуждена из run_until_complete(), останавливая весь цикл и таким образом не позволяя завершиться прочим задачам.

В нашем следующем разделе мы рассмотрим обработку сигналов (помимо Keyboard​Interrupt), но прежде чем мы перейдём к этому, было бы неплохо принять во внимание, что аккуратный останов является одной из самых сложных сторон сетевого программирования и это остаётся справедливым и для asyncio. Информация в этом разделе - это только начало. Я призываю вас провести особые проверки для чистого останова в ваших собственных автоматизированных комплектах. Различные приложения часто требуют различных стратегий.

В индексе пакетов Python я опубликовал некий крошечный пакет aiorun, преимущественно на основе своих собственных экспериментов и исследований, относящихся в останову и он содержит множество идей из данного раздела. Также может оказаться полезным для вас повозиться с этим кодом и поэкспериментировать со своими собственными мыслями относительно вариантов останова asyncio.

В своём предыдущем примере мы показали как наш цикл событий будет остановлен с помощью KeyboardInterrupt, например, нажатием Ctrl-C. Такое возбуждение KeyboardInterrupt действенно снимет блокировку имеющегося вызова run_forever() и позволит произойти дальнейшей последовательности останова.

KeyboardInterrupt соответствует сигналу SIGINT. Для сетевых служб более распространёнными в действительности являются сигнал для обработки завершения SIGTERM, и именно он также является установленным по умолчанию сигналом,который вы применяете в соответствующей команде kill() из оболочки UNIX.

Совет

Данная команда kill() в системах Unix имеет обманчивое название: всё что она делает, это отправка сигналов в некий процесс! Без аргументов, $ kill <PID> отправит некий сигнал TERM: ваш процесс может получить этот сигнал и аккуратно выполнить останов или просто проигнорировать его! Однако, это плохая идея, так как если ваш процесс не может быть остановлен в конце концов, следующий момент который выполняет данный убивец, это $ kill ­s KILL <PID>, который отправляет соответствующий сигнал KILL. Вот он уже остановит вас и ваша программа ничего с этим не сможет поделать.

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

$ python shell_signal01.py
<Your app is running>
<Your app is running>
<Your app is running>
<Your app is running>
^CGot signal: SIGINT, shutting down.
		

Для останова данной программы я нажал Ctrl-C, что видно из последней строки. Вот сам код:

# shell_signal01.py
import asyncio

async def main():()
    while True:
        print('')
        await asyncio.sleep(1)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.create_task(main())()
    try:
        loop.run_forever()
    except KeyboardInterrupt:()
        print('Got signal: SIGINT, shutting down.')
    tasks = asyncio.Task.all_tasks()
    for t in tasks:
        t.cancel()
    group = asyncio.gather(*tasks, return_exceptions=True)
    loop.run_until_complete(group)
    loop.close()
 	   

До сих пор всё было достаточно прямолинейным. Теперь мы собираемся усложнить действия:

asyncio предоставляет достаточную степень детализации в своём API для обработки подобных ситуаций. Я изменил свой код приведённого ранее примера для ввода этих моментов:

# shell_signal02.py
import asyncio
from signal import SIGINT, SIGTERM()

async def main():
    try:
        while True:
            print('<Your app is running>')
            await asyncio.sleep(1)
    except asyncio.CancelledError:()
        for i in range(3):
            print('<Ваша прикладная программа завершается...>')
            await asyncio.sleep(1)

def handler(sig):()
    loop.stop()()
    print(f'Got signal: {sig!s}, shutting down.')
    loop.remove_signal_handler(SIGTERM)()
    loop.add_signal_handler(SIGINT, lambda: None)()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    for sig in (SIGTERM, SIGINT):()
        loop.add_signal_handler(sig, handler, sig)
    loop.create_task(main())
    loop.run_forever()()
    tasks = asyncio.Task.all_tasks()
    for t in tasks:
        t.cancel()
    group = asyncio.gather(*tasks, return_exceptions=True)
    loop.run_until_complete(group)
    loop.close()
 	   

Вывод:

$ python shell_signal02.py
<Your app is running>
<Your app is running>
<Your app is running>
<Your app is running>
<Your app is running>
^CGot signal: Signals.SIGINT, shutting down.
<Your app is shutting down...>
^C<Your app is shutting down...>()
^C<Your app is shutting down...>
		

Возвращаясь к Быстрому началу, в котором мы представили основы интерфейса Исполнителя в Примере 2. В Примере 2 мы указали, что соответствующий вызов time.sleep() был короче чем соответствующий вызов asyncio.sleep() - к счастью для нас - так как это означает, что собственная задача Исполнителя завершается быстрее чем его сопрограмма main() и в следствии этого данная программа останавливается корректно.

В этом разделе мы изучим что происходит в процессе останова когда задания Исполнителя требуют больше времени чем все приостановленные экземпляры Task. Самый короткий ответ таков: без вмешательства будьте готовы получить такой вид ошибок:

# quickstart.py
import time
import asyncio

async def main():
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')
    loop.stop()

def blocking():
    time.sleep(1.5)()
    print(f"{time.ctime()} Hello from a thread!")


loop = asyncio.get_event_loop()

loop.create_task(main())

loop.run_in_executor(None, blocking)
loop.run_forever()
tasks = asyncio.Task.all_tasks(loop=loop)
group = asyncio.gather(*tasks, return_exceptions=True)
loop.run_until_complete(group)
loop.close()
 	   

Fri Sep 15 16:25:08 2017 Hello!
Fri Sep 15 16:25:09 2017 Goodbye!
exception calling callback for <Future at [...snip...]>
Traceback (most recent call last):

<big nasty traceback>

RuntimeError: Event loop is closed
Fri Sep 15 16:25:09 2017 Hello from a thread!
 	   

Что здесь происходит, так это то, что за сценой, run_in_executor() не создаёт некий экземпляр Task, он возвращает некий Future. Это означает, что он не содержится в наборе "активных задач", возвращаемых из asyncio.Task.all_tasks(), и таким образом, run_until_complete() не ждёт завершения задачи своего Исполнителя!

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

Самая первая идея состоит в самая простая реализация состоит в постоянном выполнении await некоторой задачи Исполнителя внутри какой- то сопрограммы:

# quickstart.py
import time
import asyncio

async def main():
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')
    loop.stop()

def blocking():
    time.sleep(2.0)
    print(f"{time.ctime()} Hello from a thread!")

async def run_blocking():()
    await loop.run_in_executor(None, blocking)

loop = asyncio.get_event_loop()
loop.create_task(main())
loop.create_task(run_blocking())()
loop.run_forever()
tasks = asyncio.Task.all_tasks(loop=loop)
group = asyncio.gather(*tasks, return_exceptions=False)
loop.run_until_complete(group)
loop.close()
 	   

Приведённый выше код выглядит великолепно, за исключением одного момента: он не способен обрабатывать прекращение! Если вы посмотрите повнимательнее, вы увидите, что я опустил тот цикл прекращения задач, который появлялся в наших предыдущих примерах. Если вы вернёте его обратно, мы получим jib,rb "Event loop is closed" {Цикл событий закрыт}, как и ранее. Мы даже не можем обработать CancelledError внутри run_blocking() чтобы попытаться повторно ожидать данный фьючерс. Вне зависимости от того что мы пытаемся сделать, та задача, которая обёртывает run_blocking() не может быть прекращена, однако соответствующее задание Исполнителя будет исполняться пока не завершится его внутренний time.sleep(). Давайте перейдём в нашей следующей идее.

Следующая, приводимая ниже идея, является слегка более хитрой. Основная стратегия состоит в сборе всех приостановленных задач, с последующим прекращением только их, однако перед этим мы вызываем run_until_complete(), который мы добавили в соответствующий фьючерс из run_in_executor().

# quickstart.py
import time
import asyncio

async def main():
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')
    loop.stop()

def blocking():
    time.sleep(2.0)
    print(f"{time.ctime()} Hello from a thread!")

loop = asyncio.get_event_loop()
loop.create_task(main())
future = loop.run_in_executor(None, blocking)()
loop.run_forever()
tasks = asyncio.Task.all_tasks(loop=loop)()
for t in tasks:
    t.cancel()()
group_tasks = asyncio.gather(*tasks, return_exceptions=True)
group = asyncio.gather(group_tasks, future)()
loop.run_until_complete(group)
loop.close()
 	   

Это решение ведёт себя лучшим образом в процессе останова, но всё ещё имеет недостатки. В целом, будет достаточно неудобно собирать все имеющиеся фьючерсы Исполнителя на протяжении всей вашей программы в некую коллекцию, которую вы можете применять в своей последовательности останова; затем создавать некую группу задачи- плюс- фьючерсы; а потом запускать это вплоть до завершения. Это работает, но имеется лучший способ.

# quickstart.py
import time
import asyncio

from concurrent.futures import ThreadPoolExecutor as Executor

async def main():
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')
    loop.stop()

def blocking():
    time.sleep(2.0)
    print(f"{time.ctime()} Hello from a thread!")

loop = asyncio.get_event_loop()
executor = Executor()()
loop.set_default_executor(executor)()
loop.create_task(main())
future = loop.run_in_executor(None, blocking)()
loop.run_forever()
tasks = asyncio.Task.all_tasks(loop=loop)
for t in tasks:
    t.cancel()
group = asyncio.gather(*tasks, return_exceptions=True)
loop.run_until_complete(group)
executor.shutdown(wait=True)()
loop.close()
 	   

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

Я настоятельно рекомендую вам поэкспериментировать со всеми показанными здесь примерами кода и испытайте различные стратегии для создания задач и заданий Исполнителя, заменяйте их поочерёдно во времени попытайтесь выполнять завершения чисто.

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

Однако, при написании проверок элементов, вы приходится работать со множеством циклов событий, так как каждая проверка элемента обычно требует некоторого нового экземпляра цикла. Вновь возникает основной вопрос: следует ли вам писать код, который требует передачи циклов событий в параметрах функций для поддержки проверок элементов?

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

import asyncio
from typing import Callable

async def f(notify: Callable[[str], None]):()
    # <...some code...>
    loop = asyncio.get_event_loop()()
    loop.call_soon(notify, 'Alert!')()
    # <...some code...>
 	   

Лучший способ для переброски pytest в ваш код asyncio состоит в предоставлении некоторого цикла событий через какое- то приспособление. Pytest встраивает такое в каждый ваш тест в виде некоего параметра функции. Это звучит более ложно, чем есть на самом деле, поэтому вот пример приспособления для предоставления некоторого цикла событий:

# conftest.py()
import pytest

@pytest.fixture(scope='function')()

def loop():
    loop = asyncio.new_event_loop()()
    try:
        yield loop
    finally:
        loop.close()()
 	   

Приведённый выше код вызывает некую проблему, так что не применяйте его. Эта проблема едва уловима, но она также является основной сутью данного раздела, поэтому в ближайшее время мы обсудим её подробно. Вначале давайте рассмотрим все проверки для функции сопрограммы f:

from somewhere import f()

def test_f(loop):()

    collected_msgs = []()

    def dummy_notify(msg):()
        collected_msgs.append(msg)

    loop.create_task(f(dummy_notify))()
    loop.call_later(1, loop.stop)()
    loop.run_forever()

    assert collected_msgs[0] == 'Alert!'()
 	   

Помните я сказал, что у ас была некая проблема? Вот она: внутри функции сопрограммы f наш возвращаемый из get_event_loop() экземпляр цикла является иным по отношению к тому циклу событий, который был предоставлен нам нашим приспособлением. Данная проверка завершается по этой причине неудачей; на самом деле данный цикл из get_event_loop() даже никогда не будет запущен.

Наше решение этой дилеммы состоит в добавлении некоторого параметра loop в явном виде в нашу функцию сопрограмму f. Затем вы всегда можете удостовериться что был использован правильный экземпляр цикла. На практике, однако, это может вызывать страшную головную боль, так как большие базовые коды потребуют гигантского числа функций для передачи по имеющемуся циклу.

Вот некий лучший способ: когда в игре некий новый цикл, установите этот цикл чтобы он был "тем самым" циклом связанным с нашим текущим потоком, что означает что всякий вызов к get_event_loop() возвратит совершенно новый образец. Это работает очень хорошо для тестирования. Всё что вам требуется сделать, так это слегка изменить наше приспособление pytest:

# conftest.py
import pytest

@pytest.fixture(scope='function')
def loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)()
    try:
        yield loop
    finally:
        loop.close()
 	   

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