在笔记1中追踪了 call_soon这中调用方式是如何运行的,这次看一看 run_until_complete是怎么工作的.

还是由一个简单的例子开始:

import asyncio

async def hi(loop):

print("Hello World")

loop = asyncio.get_event_loop()

# Blocking call which returns when the display_date() coroutine is done

loop.run_until_complete(hi(loop))

loop.stop()

loop.close()

一开始还是通过asyncio.get_event_loop()得到事件循环, 之后调用了 run_until_complete,而调用的参数是一个 async 修饰过的函数的返回值,

async def hi():

print("Hello World")

a = hi()

print(type(a))

#Out[3]: coroutine

可以通过简单的调用 type 来得到它的返回类型, 至于到底是什么, 可参见 asyncio.coroutine , 如果非要追究到底,其实在cpython的源码中是有写的,

typedef struct {

/* The gi_ prefix is intended to remind of generator-iterator. */

_PyGenObject_HEAD(gi)

} PyGenObject;

typedef struct {

_PyGenObject_HEAD(cr)

} PyCoroObject;

可以看出 coroutine和 生成器 其实是一个根源.因此通过 dir() 得到的结果也是很相似的.

回到 run_until_complete, 我们将一个 coroutine 传给它,之后发生了什么?

def run_until_complete(self, future):

self._check_closed()

future = tasks.ensure_future(future, loop=self)

future.add_done_callback(_run_until_complete_cb)

try:

self.run_forever()

except:

...

future.remove_done_callback(_run_until_complete_cb)

if not future.done():

raise RuntimeError('Event loop stopped before Future completed.')

return future.result()

去点注释和非关键部分就如上所示.可以看到传进去的 coroutine 又传给了 tasks.ensure_future, 继续追踪,发现其实这个 coroutine传到了 tasks.Task中用来构造 Task 对象了.而在构造 Task 对象之后或者准确的说的是构造的末尾, 通过 self._loop.call_soon(self._step)调用了 Task 对象的 _step 方法. 在其中,通过

result = coro.send(None)

开始执行我们传递进去的 coroutine, 由于出自生成器, 所以 send 的调用作用是一致的, 都是执行到 挂起 的位置返回. 比如下面简单的示例,

async def hi():

print("Hello World")

a = hi()

try:

result = a.send(None)

except StopIteration as exc:

print(exc.value)

#输出结果如下

Hello World

None

同理,这里的 result = coro.send(None) 也是如此, 如果触发了 StopIteration, 说明 coroutine 已经运行完了.如果没有呢, 比如在 coroutine中遇到了 挂起 呢?

blocking = getattr(result, '_asyncio_future_blocking', None)

首先会检查是否阻塞.先说没有阻塞即 blocking is None 的情况:

情况一 inspect.isgenerator(result) result是一个生成器:

这种在新的语法中是不允许的, 及在 coroutine中不允许再使用yield, 而应使用 yield from

情况二 result is None:

这种是​ coroutine返回了一次迭代但是并没有执行完, 这种可以看到是直接通过

self._loop.call_soon(self._step)进行了下一次迭代.

剩余情况 :

错误情况,直接抛出异常

那blocking  非 None的情况呢? 首先要知道什么 coroutine会返回 blocking is not None.

我们可以简单的从字面看出, coroutine 中需要有阻塞才可以. 可以简单的尝试三个情景:

import asyncio

async def no_block_no_return():

print("haha")

async def no_block_with_return():

print("haha")

return 1

async def with_block():

await asyncio.sleep(1.0)

print("haha")

return 1

first = no_block_no_return()

second = no_block_with_return()

third = with_block()

try:

first_result = first.send(None)

except StopIteration as exc:

print(exc.value)

try:

second_result = second.send(None)

except StopIteration as exc:

print(exc.value)

try:

third_result = third.send(None)

except StopIteration as exc:

print(exc.value)

'''

result:

haha

None

haha

1

'''

再检查,发现只有 third_result 是有定义的, 而且是 _asyncio.Future, 这时检查 blocking

assert getattr(third_result, '_asyncio_future_blocking', None)

会发现这就是我们要找的 blocking is not None.也就是在 coroutine中触发另一个coroutine.

这个 _asyncio.Future是c实现的 Future ,可以在 CPython 源码的 Modules/_asynciomodule.c中找到相关的定义, 其实和 asyncio.Future是类似的.所以我们这里参考python实现的 Future.

那么 await到底发生了什么?

其实要理解await到底做了啥也不难.前面已经说过,coroutine 和生成器类似.比如下面的代码:

async def with_block():

await asyncio.sleep(1.0)

print("haha")

return 1

翻译成没有await关键字的python代码就成了:

def fake_with_block():

#asyncio.sleep 示意如下

loop = asyncio.get_event_loop()

future = loop.create_future()

h = future._loop.call_later(

1.0, futures._set_result_unless_cancelled, future,None)

def _await():

while not future.done():

yield future

yield from _await()

#以上就是await的不完全等价的一个示意,注:只适用于此特殊情景,不能套用

print("haha")

return 1

也就是await后面的(同一行)语句(表达式)悄悄的执行了并和生成器似的返回了,不过返回了一个future对象.

所以在blocking is not None 的语块内可以看到:

result.add_done_callback(self._wakeup)

也就是添加了future结束后的操作.那怎么知道 await后面的(同一行)语句(表达式)运行完了呢?于是asyncio.future中有了set_result的方法.通过set_result来表明future执行完了并执行绑定的call_back.而这里的sleep是通过call_later延迟调用的方式执行set_result.

于是所有的操作就都合并到了loop._run_once中了.这样就完整的循环起来了.

可await后面(后面的行)还有语句要执行啊?添加的那个call_back,就是负责await部分完成了后面怎么运行的.

def _wakeup(self, future):

try:

future.result()

except Exception as exc:

self._step(exc)

else:

self._step()

self = None

可以看到,await执行完,如果没有异常,还是继续_step也就是coro.send(None),也就是继续await后面(后面的行)的内容,这样一个coro就能完整的运行了.

至此,future = tasks.ensure_future(future, loop=self)就剖析完了.后面的

future.add_done_callback(_run_until_complete_cb)

try:

self.run_forever()

except:

...

future.remove_done_callback(_run_until_complete_cb)

就是异常处理和开始事件循环,在笔记1中已经提过了就不再赘余.