编程语言 - Python 3 的 AsyncIO 库

并发和并行

Wikipedia: https://en.wikipedia.org/wiki/Concurrency_(computer_science)

In computer science, concurrency refers to the ability of different parts or units of a program,
algorithm, or problem to be executed out-of-order or in partial order, without affecting the
final outcome...In more technical terms, concurrency refers to decomposability property of a
program, algorithm, or problem into order-independent or partially-ordered components or units.

Wikipedia: https://en.wikipedia.org/wiki/Parallel_computing

Parallel computing is a type of computation in which many calculations or the execution of
processes are carried out simultaneously.

• 并发是指一次处理多件事，并行是指一次做多件事。一个关于结构，一个关于执行。并发用于制定方案，用来解 决可能（但未必）并行的问题。（ concurrency is not parallelism by Rob Pike）

同步和异步

Wikipedia: https://en.wikipedia.org/wiki/Asynchronous_I/O

In computer science, asynchronous operation means that a process operates independently of other
processes, whereas synchronous operation means that the process runs only as a result of some
other process being completed or handing off operation. And asynchronous I/O is a form of
input/output processing that permits other processing to continue before the transmission has
finished.

• 从概念上讲，「同步」是指各个任务按顺序一个接一个执行，某个任务结束之前，并不会开始其它任务；而「异 步」是指一个任务的执行流程（任务开始、运行和结束）独立于其它任务，并不会阻塞其它任务（其它任务无 需等待该任务执行结束）。
• 「同步」和「异步」表达的是任务之间的关系。那么谈「异步」和「同步」时，需要先明确我们是在「何种上下 文」中「如何定义任务」的。比如，在 "One request per thread" 的并发模式中，如果我们在整个进程上下文， 将 每个请求定义为一个任务时，那么多个任务间就是 异步 关系，他们独立执行，互不阻塞；而如果我们在 单个线 程上下文中，将每次 IO 操作定义为一个任务时， 那么多 个任务间就是同步关系，一个任务执行完成 后才能开 始下一个任务。
  • 任务的「异步」和「同步」关系并非一成不变。例如，在多线程并发的应用中，我们将每个线程定义为任务 时，它们互不通信的独立运行阶段时，我们可以认为他们是「异步」关系；而线程需要同时互斥访问共享数据 时，它们通过「同步原语」保证访问顺序，此时我们可以认为任务间就是「同步」关系。
  • 一般来说，我们根据以下层次识别和定义上下文：硬件、内核、系统调用、函数库、框架、应用代码。
• 「IO 多路复用」vs. 「异步 IO」- 在 IO 操作，也就是系统 IO 调用的上下文里，我们区分一下这两个概念：
  • 在 IO 多路复用技术中，IO 事件的发生时机相对于应用程序执行流程来说是不确定的（在等待事件期间，应 用程序可以做其它事情）。但是，IO 事件的处理及 IO 读写操作依然在当前执行流程中完成，所以从 IO 操 作角度 讲，IO 多路复用依然是「同步 IO」。
  • 而「异步 IO」一般指 IO 读写操作由独立执行流程（a thread of execution）完成，随后将数据交给其它 执行流程。
• 「异步」和「非阻塞 IO」没有必然关系：前者指任务间的关系，而后者是一项 IO 技术，两者不是同一层次的 概念。但是「异步」和「非阻塞」一般被认为是两个可互换使用的词汇 - 因为异步流程之间并不会相互阻塞。
• 在 Python 领域，许多框架声称的「异步」是指在运行于该框架上的任务之间的关系，一个任务的 IO 操作阻塞 与否并不影响其它任务的处理流程。

vs. Gevent/Eventlet

Gevent/eventlet 的工作原理和 asyncio 类似，在性能和运行效率上也相差无几（非数量级差别）。它们的协程 并未使用 Python 原生语法，而是借助 greenlet 库实现。

和 asyncio 相比，gevent 和 evenlet 的竞态条件（race condition）处理方式较为晦涩：

# http://asyncio.readthedocs.io/en/latest/why_asyncio.html
Code written with asyncio is less error-prone: by just looking at the code, it is possible
to identify which parts of the code are under our controls and where the event loop takes over
the control flow and is able to run other tasks when our task is waiting for something.

gevent and eventlet are designed to hide the asynchronous programming. For nonexpert, and
sometimes even for experts, it is really hard to guess where the event loop is allowed to
suspend the task and run other tasks in background. It is even worse. A modification in a third
party library can change the behaviour of our code, introduce a new point where the task is
suspended.

# The Zen of Python
Explicit is better than implicit.

兼容功能库

• https://github.com/aio-libs
• https://github.com/timofurrer/awesome-asyncio

框架性能

uvloop 是一个速度很快的、可以直接替换 asyncio 内建事件循 环的事件循环。uvloop 底层基于 libuv，通过 Cython 封装为 Python 模块。

uvloop: Blazing fast Python networking 一文中使用 uvloop 替换 asyncio 自带的事件循环，然后根据 基准测试的结果得出如下结论：

使用了 `uvloop <https://github.com/MagicStack/uvloop>`_ 事件循环的 *asyncio* 应用性能至少比
nodejs， gevent 及其它 Python 异步框架快 2 倍以上，性能和 Go 应用接近。

具体测试环境和工具可参见原文。下面只简单摘录几种场景下的基准测试数据图。

Benchmarking Result of Simple TCP Echo Servers

Benchmarking Result of Simple HTTP Servers

生成器、基于生成器的协程和原生协程

先来简单回顾一下 Python 语言里协程的发展历史：

Python 2.2 引入了生成器的语法，可以用它暂停代码执行。
Python 2.5 为生成器增加了 ``.send()`` 方法，可以通过此方法向生成器传递数据，Python 协程概念初具雏形。
Python 3.3 中引入 ``yield from`` 句法，降低了生成器协同工作复杂度。
Python 3.4 增的 asyncio 包，它提供的 ``asyncio.coroutine`` 装饰器将生成器函数正式打上了「协程」的标签。
Python 3.5 正式提供了协程的抽象基类 ``collections.abc.Coroutine`` 和用于创建和使用原生协程的句法：
  ``async def`` 和 ``await`` 。同时，也增加了 ``types.coroutine`` 装饰器，用于将基于生成器的协程封装为
  ``coroutine`` 对象。

从句法上看，基于生成器的协程和生成器类似，都是定义体中包含 yield （或 yield from ）关键字的 函数。

将生成器函数作为协程使用时，yield 关键字通常（并不绝对）出现在表达式的右边，用于从协程调用方接 收数据（协程调用方一般使用 .send(datum) 方法向协程传递数据）。并且从逻辑上讲，用于实现协程的 yield 除了用于和调用者互相传递数据外，主要用于 控制代码流程 。

由于两者在语言层面并没有明显的区别，将生成器函数视为迭代器还是协程，完全依赖使用场景和上下文。在 PyCon US 2009 期间举办的一直播著名的课程中（ http://www.dabeaz.com/coroutines/ ），David Beazley （可有是 Python 社区中在协程 方面最多产的作者和演讲者）提醒道：

• 生成器用于生成供迭代的数据
• 协程是数据的消费者
• 为了避免脑袋炸裂，不能把这两个概念混为一谈
• 协程与迭代无关
• 注意，虽然在协程中会使用 yield 产出值，但这与迭代无关

yield from 句法可以用来实现功能完备的协程（见下节）， asyncio 要求传递给它 API 的协程都使用 该句法实现。也就是说，在 asyncio 上下文中， 基于生成器的协程 （generator-based coroutine）特指 使用 yield from 句法的协程。

如果将使用 ``yield`` 定义的协程传给 *asyncio* 的 API 时，*asyncio* 会抛出
``RuntimeError: yield was used instead of yield from for generatror...`` 这
样的运行时异常。

Python 3.5 又增加了用于创建 原生协程 的句法结构（见下面） async def ，以便从句法上和基于 生成器的协程进行区分。 同时，为了让两种生成器可以互相操作，Python 3.5 还提供了 types.coroutine 装饰器函数对基于生成器的协程进行了包装：

A new function ``coroutine(fn)`` is added to the ``types`` module. It allows interoperability
between existing *generator-based coroutines* in asyncio and native coroutines ... The function
applies ``CO_ITERABLE_COROUTINE`` flag to generator-function's code object, making it return a
``coroutine`` object.

总结 : yield from 在 Python 3.3 中引入； asyncio 库在 Python 3.4 中引入；原生协程语法在 Python 3.5 中引入。那么，在进行 asyncio 相关开发时：

• 如果代码是为 Python 3.4+ 编写的，那么只能使用基于生成器的协程，并使用 asyncio.coroutine 对协 程函数进行装饰（非强制要求，但有利于提高可读性和调试便利性）。
• 如果代码是为 Python 3.5+ 编写的，并不需要向前兼容的话，尽量全部使用原生协程语法。如果需要混合使用 基于生成器的协程和原生协程的话，那么需要使用 types.coroutine 装饰基于生成器的协程。

yield from

yield from 语法在 PEP 380 - Syntax for Delegating to a Subgenerator 提案中提出，并于 Python 3.3 正式加入 Python 语言。

A Python generator is a form of coroutine, but has the limitation that it can only yield to
its immediate caller. This means that a piece of code containing a ``yield`` cannot be factored
out and put into a separate function in the same way as other code. Performing such a
factoring cuases the called function to itself become a generator, and it is necessary to
explicitly iterate over this second generator and re-yield any values that it produces.

生成器可以使用 yield from 句法将部分操作委托给另一个生成器，也就是说，可以将包含 yield 的 代码逻辑重构为另一个生成器。 yield from 句法会创建一个「通道」，把内层生成器（子生成器）与外层生 成器的客户端联系起来，这样二者可以直接发送和产出值，还可以直接传入异常。这样一来，就省去了之前把生 成器的工作委托给子生成器所需要的大量样板代码。

``yield from <expr>``

where <expr> is an expression evaluating to an iterable, from which an iterator is extracted.
The iterator is run to exhaustion, during which time it yields and receives values directly to
or from the caller of the generator containing the ``yield from`` expression (the "delegating
generator").

Furthermore, when the iterator is another generator, the subgenerator is allowed to execute a
``return`` statement with a value, and that value becomes the value of the ``yield from``
expression.

>>> def chainfor(*iterables):
...     for i in iterables:
...         for j in i:
...             yield j
>>> list(chainfor("ABC", "012"))
['A', 'B', 'C', '0', '1', '2']
>>> def chain(*iterables):
...     for i in iterables:
...         yield from i
...
>>> list(chain("ABC", "012"))
["A", "B", "C", 0, 1, 2]

刚刚创建的协程处于 GEN_CREATED 状态中，调用者需要使用 next(my_coro) 或者 my_coro.send(None) 启动/激活这个协程，被激活的协程开始运行，直到第一个 yield 表达式，之后 这个协程就准备好作用活跃的协程使用了（这一步通常称为 “预激” 协程）。使用 yield from 句法调用协 程时，会自动预激协程。Python 3.4 标准库里的 asyncio.coroutine 装饰器不会预激协程，因此能兼容 yield from 句法。

若想使用 yield from 结构，就要大幅改动代码，为了说明需要改动的部分，PEP 380 使用了一些专门的术语：

• 委派生成器 （delegating generator）- 包含 yield from <iterable> 表达式的生成器函数。
• 子生成器 （subgenerator）- 从 <iterable> 部分获取的生成器。
• 调用方 （caller）- 调用委派生成器的客户端代码。

同时，PEP 380 中对 yield from 的行为做了如下说明：

• 子生成器产出的值都直接传给委派生成器的调用方。
• 使用 send() 方法发给委派生成器的值都直接传给子生成器。如果发送的值是 None ，那么会调用子 生成器的 __next__() 方法。如果发送的值不是 None ，那么会调用子生成器的 send() 方法。如 果子生成器抛出 StopIteration 异常，那委派生成器恢复运行。任何其它异常都会向上冒泡，传给委派生 成器。
• 生成器退出时，生成器（或子生成器）中的 return expr 表达式会触发 StopIteration(expr) 异常 抛出。
• yield from 表达式的值是子生成器终止时传给 StopIteration 异常的第一个参数值。
• 传入委派生成器的异常，除了 GeneratorExit 之外都传给子生成器的 throw() 方法。如果调用 throw() 方法时抛出了 StopIteration 异常，委派生成器恢复运行。 StopIteration 之外的异 常会向上冒泡，传给委派生成器。
• 如果把 GeneratorExit 异常传入委派生成器，或者在委派生成器上调用 close() 方法，那么（委派 生成器）调用子生成器的 close() 方法（如果子生成器提供了该方法的话）。如果子生成器的 close() 方法导致异常抛出，那么异常会向上冒泡，传给委派生成器。如果子生成器的 close() 方法 正常执行的话，委派生成器随后向上抛出 GeneratorExit 异常。

接下来，看一段代码，然后思考一下它会有什么样的输出结果：

def sub():
    try:
        yield 10
        yield 11
    except BaseException as e:
        print("sub", type(e))
        raise e


def bar():
    try:
        yield from sub()
    except BaseException as e:
        print("bar", type(e))
        raise e


def foo():
    try:
        yield from bar()
    except BaseException as e:
        print("foo", type(e))
        raise e


co = foo()
co.send(None)
co.throw(GeneratorExit)

# Output
10
sub <class 'GeneratorExit'>
bar <class 'GeneratorExit'>
foo <class 'GeneratorExit'>
Traceback (most recent call last):
  File "delegation.py", line 31, in <module>
    co.throw(GeneratorExit)
  File "delegation.py", line 26, in foo
    raise e
  File "delegation.py", line 23, in foo
    yield from bar()
GeneratorExit

await 和 async

Python 3.5 增加了两个关键字 await 和 async ，以便在语法上强调协程概念，降低因 yield from 和 yield 的相似性造成的「协程」和「生成器」定义形式混淆。

PEP 0492 - Coroutines with async and await syntax:

The growth of internet and general connectivity has triggered the proportionate need for
responsive and scalable code. This proposal aims to answer that need by making writing
explicitly asynchronous, concurrent Python code easier and more Pythonic.

It is proposed to make *conroutines* a proper standalone concept in Python, and introduce new
supporting syntax. The ultimate goal is to help establish a common, easily approachable, mental
model of asynchronous programming in Python and make it as close to synchronous programming as
possible.

这个提案将「协程」作为 Python 原生语言特性和「生成器」明确区分开来，为用户提供了不依赖特定第三方库的 可靠的协程定义方法。原生协程和相关的新语法特性也使得定义异步方式的「上下文管理器」和「迭代器协议」成为 了可能。

# native coroutine
async def co1():
    pass

# generator-based coroutine
@types.coroutine
def co2():
    yield from 1

print(type(co1()))          # <class 'coroutine'>
print(type(co2()))          # <class 'generator'>

原生协程（native coroutine）有如下关键特征：

• async def functions are always coroutines, even if they do not contain await expressions.

• It is a SyntaxError to have yield or yield from expressions in an async def function.

• Internally, two new code object flags were introduced:

  • CO_COROUTINE is used to mark native coroutines (defined with new syntax).
  • CO_ITERABLE_COROUTINE is used to make generator-based coroutines compatible with native coroutines.

• Regular generators, when called, return a generator object; similarly, coroutines return a coroutine object.

• StopIteration exceptions are not propagated out of coroutines, and are replaced with a RuntimeError. For regular generators such behavior requires a future import (see PEP 479).

• When a native coroutine is garbage collected, a RuntimeWarning is raised if it was never awaited on (see also Debugging Features).

原生协程（native coroutine）和 基于生成器的协程（generator-based coroutine）有如下区别：

• 原生协程对象并未实现 __iter__ 和 __next__ 方法，所以它们并不能用于需要可迭代对象的场景。
• 普通生成器函数内部不能使用 yield from 原生协程 句法。
• 基于生成器的协程函数（代码用于 asyncio 时，需用 asyncio.coroutine ）内部可以使用 yield from 原生协程 句法。
• inspect.isgenerator() 函数和 inspect.isgeneratorfunction() 函数作用于原生协程时，返回 False 。

await 和 yield from 功能相似，它会挂起（suspend execution）调用「协程」，等待 awaitable 对象处理完后后，恢复被挂起的「协程」执行流程，并将 awaitable 对象返回值作为 await 表达式的结果。

async def read_data(db):
    data = await db.fetch("SELECT ...")
    ...

await 关键字可以处理的 awaitable 对象有：

• A native coroutine object returned from a native coroutine function.
• A generator-based coroutine object returned from a function decorated with types.coroutine().
• An object with an __await__ method returning an iterator.
• Objects defined with CPython C API with a tp_as_async.am_await function, returning an iterator (similar to __await__ method).

NOTES ：await 句法只能出现在 async def 定义的函数内（和 yield 句法只能出现在 def 定义的函数内的类似）。

「异步上下文管理器」允许在它的 enter 和 exit 方法中挂起协程执行流。为了定义「异步上下文管理器」， PEP 0492 提案为 Python 对象新增了两个特殊方法： __aenter__ 和 __aexit__ 。这两个方法必须返 回 awaitable 对象。

class AsyncContextManager:
    async def __aenter__(self):
        await log("entering context")

    async def __aexit__(self, exc_type, exc, tb):
        await log("exiting context")

「异步上下文管理器」可以被 async with 句法使用：

async with lock:
    ...

# instead of

with (yield from lock):
    ...

NOTES ：async with 句法只能出现在 async def 定义的函数内；普通上下文管理器也不能用在 async with 句法中。

「异步迭代器」 （asynchronous iterator）和 「异步可迭代对象」（asynchronous iterable）：

• An object must implement an __aiter__ method (or, if defined with CPython C API, tp_as_async.am_aiter slot) returning an asynchronous iterator object.
• An asynchronous iterator object must implment an __anext__ method (or, if defined with CPython C API, tp_as_async.am_anext slot) returning an awaitable.
• To stop iteration __anext__ must raise a StopAsyncIteration exception.

异步迭代操作由 async for 句法驱动，示例如下：

class AsyncIterable:
    def __aiter__(self):
        return self

    async def __anext__(self):
        data = await self.fetch_data()
        if data:
            return data:
        else:
            raise StopAsyncIteration
    ...

# Iteraing through asynchronous iterator
async for item in AsyncIterable():
    ...
else:
    ...

NOTES ： async for 句法只能出现在 async def 定义的函数内；普通可迭代对象也不能用在 async with 句法中。

asyncio.Future

<流畅的 Python> 中译版作者将 Future 翻译为「期物」（「物」指「物体」、也即对象，「期」取期货、期权和 期权中「期」字的意思），相当贴切，我们下文也遵循这个译法。另外，future （期物） ， promise ， delay 和 deferred 等术语含义相似，一般可以互换使用。

期物是尚未完成的异步操作的处理结果的占位符，当该操作完成后，可以从期物中获取处理结果。简单来说， 期物代表一个异步操作 。

从 Python 3.4 起，标准库中有两个名为 Future 的类： concurrent.futures.Future 和 asyncio.Future 。这两个类接口相似，但是目前还不能互换使用（PEP 3156 中提到，它们未来可能会 兼容）。

它们是 concurrent.futures 模块和 asyncio 包的重要组件，但是，作为这两个库的用户，通常情况 下自己不应该创建期物，而只能由并发框架实例化。客户端代码也不应该改变期物的状态，并发框架在期物表示 的延迟计算结束后会改变期特的状态，而我们无法控制计算何时结束。

这两种期物都有 .done() 方法，这个方法不阻塞，返回值是布尔值，指明期物链接的可调用对象是否已经 执行。客户端代码通常不会询问期特是否运行结束，而是会等待通知。因此，两个 Future 类都有 .add_done_callback() 方法：这个方法只有一个参数，类型是可调用对象，期物运行结束后会调用指定的 可调用对象。

此外，还有 .result() 方法。在期物运行结束后调用的话，这个方法在两个 Future 类中的作用相同： 返回可调用对象的结果，或者重新抛出执行可调用的对象时抛出的异常。

不同的是，asyncio.Future 类的 .result() 方法没有参数，因此不能指定超时时间。此外，如果
调用 .result() 方法时期物还没有运行完毕，那么 .result() 方法不会阻塞去等持结果，而是抛
出 asyncio.InvalidStateError 异常。

所以，通常会使用 yield from 或 .add_done_callback 获取期物的执行结果。使用 yield from 时：

1. 当前协程被挂起，事件循环得到控制权；
2. 异步操作结束后，事件循环不会触发回调对象，而是设置期物返回值；
3. yield from 表达式则在暂停的协程中生成返回值，恢复协程执行流程。

这两个库中有几个函数会返回期物，其他函数则使用期物。比如， asyncio 提供的如下几个函数：

# Create an ``Future`` object attached to the loop.
AbstractEventLoop.create_future()

# Schedule the execution of a coroutine boject: wrap it in a future. Return a Task object.
AbstraceEventLoop.create_task(coro)

# Run until the ``Future`` is done. If the argument is a coroutine object, it is wrapped
# by ``ensure_future()``.
AbstractEventLoop.run_until_complete(future)

# Wrap a ``concurrent.futures.Future`` object in a ``Future`` object.
asyncio.wrap_future(future, *, loop=None)

# Schedule the execution of a coroutine object: wrap it in a future. Return a Task object.
asyncio.ensure_future(coro_or_future, *, loop=None)

# Return a future aggregating results from the given coroutine object or futures.
async.gather(*coros_or_futures, loop=None, return_exceptions=False)

# Waiting for the futures and coroutine objects given by the sequence futures to complete.
# Coroutines will be wrapped in Tasks. Returns two sets of Futures: (done, pending)
async.wait(futures, *, loop=None, timeout=None, return_when=ALL_COMPLETED)

那么，在 asyncio 中，事件循环是如何使用期物的呢？

https://snarky.ca/how-the-heck-does-async-await-work-in-python-3-5/
... With this concrete definition of a coroutine, you then used ``yield from`` on any
``asyncio.Future`` object to pass it down to the event loop, pausing execution of the coroutine
while you waited for somthing to happen. Once the future object reached the evelt loop it was
monitored there until the future object was done doing whatever it needed to do. Once the
future was done doing its thing, the event loop noticed and the coroutine that was paused
waiting for the future's result started again with its result send back into the corouting
using its ``send()`` method.

asyncio 中的协程是由事件循环驱动的：协程链中需要挂起执行流程的部分返回 Future 对象，事件循环 拿到这个对象后，监听它的执行进度。Future 代表的异步操作完成后，事件循环才恢复协程链的执行。

Hello World!

import asyncio


async def say(what, when):
    print("Gonna sleep for {} seconds".format(when))
    await asyncio.sleep(when)
    print(what)


loop = asyncio.get_event_loop()
loop.run_until_complete(say("Hello world", 3))
loop.close()

HTTP 客户端

import asyncio
import aiohttp


async def fetch_page(session, url):
    with aiohttp.Timeout(10):
        async with session.get(url) as response:
            return await response.read()


loop = asyncio.get_event_loop()
with aiohttp.ClientSession(loop=loop) as session:
    content = loop.run_until_complete(
        fetch_page(session, "https://www.baidu.com")
    )
    print(content)

loop.close()

线程池

import asyncio


def compute_pi(digits):
    return 3.14


async def main(loop):
    digits = await loop.run_in_executor(None, compute_pi, 20000)
    print("pi: %s" % digits)


loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
loop.close()

使用 uvloop

import asyncio
import uvloop


asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# Or
loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)