基于async关键字的原生协程

# 定义一个简单的原生协程cor

async def cor():

    print('enter cor')

    print('exit cor')

print(type(cor))	# <class 'function'>

print(type(cor()))	# <class 'coroutine'>

可以看到cor的类型<class 'function'>函数类型，说明async关键字修饰的函数也是一个函数而已，跟普通函数在定义上没啥什么差别，差别在于被调用的时候，cor()并不是执行函数体中的语句，而是生成一个<class 'coroutine'>类型的对象，即协程对象，跟生成器类似，协程也有send()方法。

c = cor()

    try:

        c.send(None)

    except StopIteration as e:

        print(e.value)	# None

当c.send(None)时，才会执行cor中的语句，也就是执行print('enter cor')和print('exit cor')这两句，就执行完毕了，执行完毕时会抛出StopIteration异常，这个异常对象的value属性中保存着cor的返回值，这里core没有return语句，其实默认返回None。

原生协程中使用 await 关键字

async def cor():

    print('enter cor')

    rst = await cor1()

    print('rst -->', rst)

    print('exit cor')

async def cor1():

    print('enter cor1')

    print('exit cor1')

    return 'cor1'

c = cor()

try:

    c.send(None)

except StopIteration as e:

    print('cor -->', e.value)  # None

输出如下：

enter cor

enter cor1

exit cor1

rst --> cor1

exit cor

cor --> None

await关键字后面必须是一个实现了__awwit__方法的对象，不一定是协程对象，只要实现了这个方法，就会进入到此方法中，调用该方法中的逻辑。比如后面要介绍的Future对象，它也是实现了__await__方法的。

await 关键字的语义是表示挂起当前的cor协程的执行，进入到cor1协程中执行cor1的逻辑，直到cor1执行完毕，然后执行流程又回到cor挂起的地方，继续执行cor中await后面的语句。直到最后抛出StopIteration异常。

基于生成器的协程，也就是非原生协程

import types

async def cor():

    print('enter cor')

    rst = await cor2()

    print('rst --> ', rst)

    print('exit cor')

@types.coroutine

def cor2():

    print('enter cor2')

    yield

    print('exit cor2')

    return 'cor2'

c = cor()

try:

    c.send(None)

    c.send(None)

except StopIteration as e:

    print('cor -->', e.value)  # None

输出如下：

enter cor

enter cor2

exit cor2

rst -->  cor2

exit cor

cor --> None

与上面的原生协程的嵌套不同的是，调用了两次c.send(None)，执行第一次c.send(None)时，会在cor2的yield关键字处挂起，第二次c.send(None)则会在yield挂起的地方，接着往下执行，然后core2返回'cor2'，赋值给rst变量，继续执行cor中await后面的语句，直到最后抛出StopIteration异常。

总结

async 是一个关键字，async def 定义的类型还是一个function类型，只有当它被调用时才返回一个协程对象

async def 跟def定义的方法在没被调用时没有任何区别，不必看得很神秘，它也可以有return语句，这点也正常，

因为python中没有return语句的函数实际上默认是返回None的，所以只是显式的return和隐式return None的区别

对于协程的send(None)方法，跟生成器的send(None)类似，不同的是，原生协程的send方法被调用的时候，会一直执行到碰

到await语句，但是不会停下来，会直接再进入到await EXPR的EXPR中，其实EXPR是一个awaitable对象，会调用该对象的

__await__()执行该方法的里面的逻辑，如果该awaitable对象是一个原生协程对象，那么它的__await__()方法中的逻辑就

是在定义此协程时async def 下面的逻辑，执行完毕后，该协程对象就关闭了，执行流程就再次跳转到当前挂起的协程中，

执行该协程中余下的逻辑，最后执行完毕，抛出StopIteration异常。

对于原生生协程来说，调用send()方法时，会一直执行到出现StopIteration异常为止，只有在 __await__()方法中有yield语句时才

会挂起在那里，如果__await__()方法中没有yield语句，不会挂起，会返回__await__的返回值，继续执行，直到抛出StopIteration异常。

先抛出一个结论，在asyncio中协程的流程的挂起操作，实际上还是是通过yield关键字来实现的，并不是await关键字， async和await关键字只不过是语法糖。

asyncio的执行基本流程分析

import asyncio

async def cor():

    print('enter cor ...')

    await asyncio.sleep(2)

    print('exit cor ...')

    return 'cor'

loop = asyncio.get_event_loop()

task = loop.create_task(cor())

rst = loop.run_until_complete(task)

print(rst)

从这个简单的例子入手，逐步分析协程在事件循环中的执行流程。

• 第1步 async def cor()定义了一个cor协程。

• 第2步 loop = asyncio.get_event_loop()得到一个事件循环对象loop，这个loop在一个线程中只有唯一的一个实例，只要在同一个线程中调用此方法，得到的都是同一个loop对象。

• 第3步 task = loop.create_task(cor())把cor协程包装成一个task对象。

• 第4步 rst = loop.run_until_complete(task)把这个task对象添加到事件循环中去。

首先看第3步的loop.create_task(cor())这个方法

class BaseEventLoop(events.AbstractEventLoop):

	...

    def __init__(self):

		...

		# 用来保存包裹task.step方法的handle对象的对端队列

        self._ready = collections.deque()

		# 用来保存包裹延迟回调函数的handle对象的二叉堆，是一个最小二叉堆

        self._scheduled = []

		...

    def create_task(self, coro):

        """Schedule a coroutine object.

        Return a task object.

        """

        self._check_closed()

		# self._task_factory 默认是None

        if self._task_factory is None:

			# 创建一个task对象

            task = tasks.Task(coro, loop=self)

            if task._source_traceback:

                del task._source_traceback[-1]

        else:

            task = self._task_factory(self, coro)

		# 返回这个task对象

        return task

	def call_soon(self, callback, *args):

        self._check_closed()

        if self._debug:

            self._check_thread()

            self._check_callback(callback, 'call_soon')

		# 关键代码callback就是task._step方法，args是task._step的参数

        handle = self._call_soon(callback, args)

        if handle._source_traceback:

            del handle._source_traceback[-1]

        return handle

    def _call_soon(self, callback, args):

		# 1 handle是一个包裹了task._step方法和args参数的对象

        handle = events.Handle(callback, args, self)

        if handle._source_traceback:

            del handle._source_traceback[-1]

		# 2 关键代码，把handle添加到列表self._ready中

        self._ready.append(handle)

        return handle

loop.create_task(cor())实际上是创建了一个Task类的实例。再来看一下Task这个类

class Task(futures.Future):

	...

    def __init__(self, coro, *, loop=None):

        assert coroutines.iscoroutine(coro), repr(coro)

		# 调用父类的__init__获得线程中唯一的loop对象

        super().__init__(loop=loop)

        if self._source_traceback:

            del self._source_traceback[-1]

        self._coro = coro

        self._fut_waiter = None

        self._must_cancel = False

		# 关键代码，把_step方法注册到loop对象中去

        self._loop.call_soon(self._step)

        self.__class__._all_tasks.add(self)

	def _step(self, exc=None):

        ...

		result = coro.send(None)

		...

task实例化时，调用了self._loop.call_soon(self._step) --> loop.call_soon(callback, *args) --> loop._call_soon(callback, args)，实际上是把handle(task._step)这个对象放到了loop._ready队列中，放在这个队列中有什么用呢？先告诉大家，_step方法会在loop对象的循环中被调用，也就是会执行coro.send(None)这句。coro.send(None)实际上就是执行上面定义的cor协程的里面的语句。

也就是说到第3步执行完时，loop对象已经实例化，task对象也实例化，并且task对象的_step方法被封装成handle对象放入了loop对象的_ready队列中去了。

再来看第4步loop.run_until_complete(task)

class BaseEventLoop(events.AbstractEventLoop):

    def run_until_complete(self, future):

		...

		# future就是task对象，下面2句是为了确保future是一个Future类实例对象

        new_task = not futures.isfuture(future)

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

        if new_task:

            # An exception is raised if the future didn't complete, so there

            # is no need to log the "destroy pending task" message

            future._log_destroy_pending = False

        # 添加回调方法_run_until_complete_cb到当前的task对象的callbacks列表中，_run_until_complete_cb就是最后

        # 把loop的_stop属性设置为ture的，用来结束loop循环的

        future.add_done_callback(_run_until_complete_cb)

        try:

            # 开启无线循环

            self.run_forever()

        except:

			...

            raise

        finally:

			...

		# 执行完毕返回cor的返回值

        return future.result()

    def run_forever(self):

		...

        try:

            events._set_running_loop(self)

            while True:

                # 每次运行一次循环，判断下_stopping是否为true，也就是是否结束循环

                self._run_once()

                if self._stopping:

                    break

        finally:

			...

    def _run_once(self):

        # loop的_scheduled是一个最小二叉堆，用来存放延迟执行的回调函数，根据延迟的大小，把这些回调函数构成一个最小堆，然后再每次从对顶弹出延迟最小的回调函数放入_ready双端队列中，

        # loop的_ready是双端队列，所有注册到loop的回调函数，最终是要放入到这个队列中，依次取出然后执行的

		# 1. self._scheduled是否为空

        sched_count = len(self._scheduled)

        if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and

            self._timer_cancelled_count / sched_count >

                _MIN_CANCELLED_TIMER_HANDLES_FRACTION):

            # Remove delayed calls that were cancelled if their number

            # is too high

            new_scheduled = []

            for handle in self._scheduled:

                if handle._cancelled:

                    handle._scheduled = False

                else:

                    new_scheduled.append(handle)

            heapq.heapify(new_scheduled)

            self._scheduled = new_scheduled

            self._timer_cancelled_count = 0

        else:

            # Remove delayed calls that were cancelled from head of queue.

            while self._scheduled and self._scheduled[0]._cancelled:

                self._timer_cancelled_count -= 1

                handle = heapq.heappop(self._scheduled)

                handle._scheduled = False

		# 2. 给timeout赋值，self._scheduled为空，timeout就为None

        timeout = None

		# 只要self._ready不为空，timeout就为0

        if self._ready or self._stopping:

            timeout = 0

		# 只要self._scheduled不为空

        elif self._scheduled:

            # Compute the desired timeout.

            # 用堆顶的回调函数的延迟时间作为timeout的等待时间，也就是说用等待时间最短的回调函数的时间作为timeout的等待时间

            when = self._scheduled[0]._when

            timeout = max(0, when - self.time())

		、

        if self._debug and timeout != 0:

            ...

		# 3. 关注else分支，这是关键代码

        else:

            # timeout=None --> 一直阻塞，只要有io事件产生，立马返回event_list事件列表，否则一直阻塞着

            # timeout=0 --> 不阻塞，有io事件产生，就立马返回event_list事件列表，没有也返空列表

            # timeout=2 --> 阻塞等待2s，在这2秒内只要有io事件产生，立马返回event_list事件列表，没有io事件就阻塞2s，然后返回空列表

            event_list = self._selector.select(timeout)

        #  用来处理真正的io事件的函数

        self._process_events(event_list)

        # Handle 'later' callbacks that are ready.

        end_time = self.time() + self._clock_resolution

		# 4. 依次取出堆顶的回调函数handle添加到_ready队列中

        while self._scheduled:

            handle = self._scheduled[0]

            # 当_scheduled[]中有多个延迟回调时，通过handle._when >= end_time来阻止没有到时间的延迟函数被弹出，

            # 也就是说，当有n个延迟回调时，会产生n个timeout，对应n次run_once循环的调用

            if handle._when >= end_time:

                break

            # 从堆中弹出堆顶最小的回调函数，放入 _ready 队列中

            handle = heapq.heappop(self._scheduled)

            handle._scheduled = False

            self._ready.append(handle)

		# 5. 执行self._ready队列中所有的回调函数handle对象

        ntodo = len(self._ready)

        for i in range(ntodo):

            handle = self._ready.popleft()

            if handle._cancelled:

                continue

            if self._debug:

                try:

                    self._current_handle = handle

                    t0 = self.time()

                    handle._run()

                    dt = self.time() - t0

                    if dt >= self.slow_callback_duration:

                        logger.warning('Executing %s took %.3f seconds',

                                       _format_handle(handle), dt)

                finally:

                    self._current_handle = None

            else:

				# handle._run()实际上就是执行task._step()，也就是执行cor.send(None)

                handle._run()

        handle = None  # Needed to break cycles when an exception occurs.

执行run_until_complete方法时future.add_done_callback(_run_until_complete_cb)这一步其实是task.add_done_callback(_run_until_complete_cb)，也就是把_run_until_complete_cb回调函数注册到task对象中去了，这个回调函数的作用是当cor协程执行完毕时，回调_run_until_complete_cb把loop对象的 _stopping属性设为True，然后_run_once执行完毕时，判断_stopping为True就跳出while循环，run_until_complete才能返回task.result

从上面的函数调用流程

run_until_complete() --> run_forever() --> _run_once()，重点看_run_once这个方法的执行流程。

此时：

• cor协程还未开始执行。

• loop._ready = [handle(task._step)]，loop._scheduled = []。

第一轮_run_once()的调用执行开始

注意这里的第1,2,3,4,5步是在_run_once()上标记的1,2,3,4,5注释

• 第1,2步的逻辑判断，timeout = 0。

• 第3步 event_list = self._selector.select(0)，此时立马返回空[]。

• 第4步 由于loop._scheduled = []，不执行第4步中的语句。

• 第5步 依次从_ready队列中取出回调函数handle，执行handle._run()。

执行handle._run()方法，也就是调用task._step()，来看看task._step()的执行逻辑：

class Task(futures.Future):

	...

	def _step(self, exc=None):

	    """

	    _step方法可以看做是task包装的coroutine对象中的代码的直到yield的前半部分逻辑

	    """

	    ...

	    try:

	        if exc is None:

	            # 1.关键代码

	            result = coro.send(None)

	        else:

	            result = coro.throw(exc)

		# 2. coro执行完毕会抛出StopIteration异常

	    except StopIteration as exc:

	        if self._must_cancel:

	            # Task is cancelled right before coro stops.

	            self._must_cancel = False

	            self.set_exception(futures.CancelledError())

	        else:

	            # result为None时，调用task的callbasks列表中的回调方法，在调用loop.run_until_complite，结束loop循环

	            self.set_result(exc.value)

	    except futures.CancelledError:

	        super().cancel()  # I.e., Future.cancel(self).

	    except Exception as exc:

	        self.set_exception(exc)

	    except BaseException as exc:

	        self.set_exception(exc)

	        raise

	    # 3. result = coro.send(None)不抛出异常，说明coro协程遇到 yield 被挂起

	    else:

			# 4. 查看result是否含有_asyncio_future_blocking属性

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

	        if blocking is not None:

	            # Yielded Future must come from Future.__iter__().

	            if result._loop is not self._loop:

	                self._loop.call_soon(

	                    self._step,

	                    RuntimeError(

	                        'Task {!r} got Future {!r} attached to a '

	                        'different loop'.format(self, result)))

	            elif blocking:

	                if result is self:

	                    self._loop.call_soon(

	                        self._step,

	                        RuntimeError(

	                            'Task cannot await on itself: {!r}'.format(

	                                self)))

	                # 4.1. 如果result是一个future对象时，blocking会被设置成true

	                else:

	                    result._asyncio_future_blocking = False

	                    # 把_wakeup回调函数设置到此future对象中，当此future对象调用set_result()方法时，就会调用_wakeup方法

	                    result.add_done_callback(self._wakeup)

	                    self._fut_waiter = result

	                    if self._must_cancel:

	                        if self._fut_waiter.cancel():

	                            self._must_cancel = False

	            else:

	                self._loop.call_soon(

	                    self._step,

	                    RuntimeError(

	                        'yield was used instead of yield from '

	                        'in task {!r} with {!r}'.format(self, result)))

			# 5. 如果result是None，则注册task._step到loop对象中去，在下一轮_run_once中被回调

	        elif result is None:

	            # Bare yield relinquishes control for one event loop iteration.

	            self._loop.call_soon(self._step)

			# --------下面的代码可以暂时不关注了--------

	        elif inspect.isgenerator(result):

	            # Yielding a generator is just wrong.

	            self._loop.call_soon(

	                self._step,

	                RuntimeError(

	                    'yield was used instead of yield from for '

	                    'generator in task {!r} with {}'.format(

	                        self, result)))

	        else:

	            # Yielding something else is an error.

	            self._loop.call_soon(

	                self._step,

	                RuntimeError(

	                    'Task got bad yield: {!r}'.format(result)))

	    finally:

	        self.__class__._current_tasks.pop(self._loop)

	        self = None  # Needed to break cycles when an exception occurs.

    def _wakeup(self, future):

        try:

            future.result()

        except Exception as exc:

            # This may also be a cancellation.

            self._step(exc)

        else:

            # 这里是关键代码，上次的_step()执行到第一次碰到yield的地方挂住了，此时再次执行_step(),

            # 也就是再次执行 result = coro.send(None) 这句代码，也就是从上次yield的地方继续执行yield后面的逻辑

            self._step()

        self = None  # Needed to break cycles when an exception occurs.

当task._step()执行时，调用core.send(None)，即调用：

async def cor():

	# 1.

    print('enter cor ...')

	# 2.

    await asyncio.sleep(2)

	# 3.

    print('exit cor ...')

    # 4.

    return 'cor'

注意这里的第1,2,3,4步是在cor上标记的1,2,3,4注释

• 第1步 print('enter cor ...')

• 第2步 await asyncio.sleep(2)，sleep是一个非原生协程，前面介绍过，await语句挂起当前的协程也就是cor，然后会进入到sleep协程中的。注意，此时执行流程已经在sleep协程中了，我们来看一下sleep协程的代码逻辑。

看一下sleep协程实现

@coroutine

def sleep(delay, result=None, *, loop=None):

    """Coroutine that completes after a given time (in seconds)."""

    if delay == 0:

        yield

        return result

    if loop is None:

        loop = events.get_event_loop()

    # 1. 创建一个future对象，用来衔接前一个task

    future = loop.create_future()

    # 2. 添加一个延迟执行的回调函数futures._set_result_unless_cancelled 到当前loop对象的_scheduled二叉堆中，这个堆中的回调函数按照delay的大小，形成一个最小堆

    h = future._loop.call_later(delay,

                                futures._set_result_unless_cancelled,

                                future, result)

    try:

        # 3. 执行 yield from future 语句，此时会调用future的 __iter__() 方法，然后在 yield future 处挂住，返回future本身，self._asyncio_future_blocking = True

        return (yield from future)

    finally:

        h.cancel()

sleep是一个非原生协程，delay=2

注意这里的第1,2,3步是在sleep上标记的1,2,3注释

• 第1步 生成一个新Future对象，这个对象不同于跟task是不同的对象，他们都是Future类型的对象，因为Task类继承自Future类。

• 第2步 loop对象中注册了一个futures._set_result_unless_cancelled的延迟回调函数handle
  
  对象，前面介绍过，延迟回调函数handle对象是放在loop._scheduled这个最小二叉堆中的，此时，loop对象的_scheduled最小堆中只有一个延迟回调函数handle。到sleep中的第2步完成为止，loop._scheduled=[handle(delay_2s__set_result_unless_cancelled)]，loop._ready=[]，注意正在执行的handle._run()的流程还没走完，现在是进入到了sleep协程中的第2步中。

• 第3步 执行(yield from future)会调用future的 __iter__方法，进入到__iter__方法中，首先把self._asyncio_future_blocking 赋值为 True了，，然后yield self，注意，此时cor协程的执行流程就挂起在了yield处，返回self也就是Future对象自己，也就是说执行result= core.send(None)最终挂起在新的Future对象的yield self处，返回得到了一个Future对象赋值给result。即result就是在sleep（)协程中新生成的一个Future对象了。

我们看一下Future对象的这个方法。

class Future:

	...

    def __iter__(self):

		# self.done()返回False，

        if not self.done():

            self._asyncio_future_blocking = True

			# 把Future对象自己返回出去

            yield self  # This tells Task to wait for completion.

        assert self.done(), "yield from wasn't used with future"

        return self.result()  # May raise too.

    if compat.PY35:

        __await__ = __iter__ # make compatible with 'await' expression

到此为止，result = core.send(None)的调用得到了一个Future对象，然后执行流程继续往下走也就是由于result是Future对象，因此进入到_step方法的第4.1步，这里看一下代码片段

# 4.1. 如果result是一个future对象时，blocking会被设置成true

else:

    result._asyncio_future_blocking = False

    # 把_wakeup回调函数设置到此future对象中，当此future对象调用set_result()方法时，就会调用_wakeup方法

    result.add_done_callback(self._wakeup)

result.add_done_callback(self._wakeup)实际上就是把task._wakeup方法注册到了新Futrue对象的回调方法列表_callbacks = [task._wakeup,]中，到此为止，task._step方法才彻底执行完毕。第一轮_run_once()的调用执行结束了。此时 loop._stopping = Fasle，然后继续执行下一轮的_run_once()。

此时：

• cor协程的执行流程挂起在sleep协程的中产生的新Future对象的__iter__方法的yield处。cor协程的执行了cor中标记的第1,2步，第3,4步未执行。

• 新Future对象的_callbacks = [task._wakeup,]。

• loop._scheduled=[handle(delay_2s__set_result_unless_cancelled)]，loop._ready=[]。

第二轮_run_once()的调用执行开始

进入到_run_once()方法中，由于loop._scheduled=[handle(delay_2s__set_result_unless_cancelled)]

注意这里的第1,2,3,4,5步是在_run_once()上标记的1,2,3,4,5注释

• 第1,2步的逻辑判断，timeout = 2

• 第3步 event_list = self._selector.select(2)，也就是说阻塞2s中，注意，此时因为我们编写的那个cor协程是没有io事件的，是一个通过sleep协程模拟耗时操作的，不涉及到真正的io事件，所以这个时候，selector.select(2)会完整的阻塞2秒钟。

• 第4步 依次取出 _scheduled的延迟回调函数handle，放入到 _ready队列中。

• 第5步 依次从_ready队列中取出延迟回调函数handle，执行handle._run()。

第5步中的回调函数就是sleep协程中注册到loop对象的futures._set_result_unless_cancelled函数

def _set_result_unless_cancelled(fut, result):

    """Helper setting the result only if the future was not cancelled."""

    if fut.cancelled():

        return

    # fut就是sleep中新生成的Future对象，调用set_result()方法

    fut.set_result(result)

Future对象的set_result方法

class Future:

	...

    def set_result(self, result):

        if self._state != _PENDING:

            raise InvalidStateError('{}: {!r}'.format(self._state, self))

        self._result = result

        self._state = _FINISHED

		# 回调Future对象中添加的所有回调函数

        self._schedule_callbacks()

    def _schedule_callbacks(self):

        callbacks = self._callbacks[:]

        if not callbacks:

            return

        self._callbacks[:] = []

		# 依次取出注册到Future对象中的所有回调函数，放入到loop._ready队列中去，等待下一轮 _run_onece的调用时，执行这些回调

        for callback in callbacks:

            self._loop.call_soon(callback, self)

fut.set_result(result) --> fut._schedule_callbacks() --> fut._loop.call_soon(callback, self)， callback实际上是新Future对象的_callbacks 中的task._wakeup方法，task._wakeup又被添加到loop._ready队列中去了。到此为止handle._run()执行完毕，第二轮的_run_once()执行完毕。

此时：

• cor协程的执行流程挂起在sleep协程的中产生的新Future对象的__iter__方法的yield处。

• 新Future对象的_callbacks = []。

• loop._ready = [handle(task._wakeup)]， loop._scheduled=[]。

第三轮_run_once()的调用执行开始

注意这里的第1,2,3,4,5步是在_run_once()上标记的1,2,3,4,5注释

• 第1,2步的逻辑判断，timeout = 0。

• 第3步 event_list = self._selector.select(0)，也就是说立即返回。

• 第4步 由于loop._scheduled=[]，因此不执行第4步中的逻辑。

• 第5步 依次从_ready队列中取出回调函数handle，执行handle._run()

执行handle._run()就是执行task._wakeup()，又要回到task._wakeup()代码中看看

class Task(futures.Future):

    def _wakeup(self, future):

        try:

			# future为sleep协程中生成的新的Future对象

            future.result()

        except Exception as exc:

            # This may also be a cancellation.

            self._step(exc)

        else:

            # 这里是关键代码，上次的_step()执行到第一次碰到yield的地方挂住了，此时再次执行_step(),

            # 也就是再次执行 result = coro.send(None) 这句代码，也就是从上次yield的地方继续执行yield后面的逻辑

            self._step()

        self = None  # Needed to break cycles when an exception occurs.

调用task._wakeup()实际上又是执行task._step()也就是再次执行result = core.send(None)这行代码，前面提到过，core协程被挂起在Future对象的__iter__方法的yield处，此时再次执行result = core.send(None)，就是执行yield后面的语句

class Future:

	...

    def __iter__(self):

		# self.done()返回False，

        if not self.done():

            self._asyncio_future_blocking = True

			# 把Future对象自己返回出去

            yield self  # This tells Task to wait for completion.

        assert self.done(), "yield from wasn't used with future"

		# 调用task._wakeup再次进入到core挂起的地方执行yield后面的语句

        return self.result()  # May raise too.

    if compat.PY35:

        __await__ = __iter__ # make compatible with 'await' expression

self.result() 返回的是None，所以 cor协程的await asyncio.sleep(2)返回的是None，到此为止cor协程的第三步await asyncio.sleep(2)才真正的执行完毕，也就是说sleep协程执行完毕了，然后继续执行cor协程await下面的语句print('exit cor ...')最后返回'cor'，到此为止cor协程就完全执行完毕了。

async def cor():

    print('enter cor ...')

    await asyncio.sleep(2) # 上次在这里挂起

    print('exit cor ...')

    return 'cor'

前面介绍了，原生协程在执行结束时会抛出StopIteration异常，并且把返回值存放在异常的的value属性中，因此在task._step()的第2步捕捉到StopIteration异常

	# 2. coro执行完毕会抛出StopIteration异常

    except StopIteration as exc:

        if self._must_cancel:

            # Task is cancelled right before coro stops.

            self._must_cancel = False

            self.set_exception(futures.CancelledError())

        else:

			# exc.value是'cor'

            # 调用task.set_result('cor')

            self.set_result(exc.value)

task.set_result('cor')其实就是把task中的存放的回调函数又放到loop._ready队列中去，task中的回调函数就是前面介绍的_run_until_complete_cb函数。到此为止第3轮的_run_once()执行完毕。

此时：

• cor协程的执行完毕。

• 新Future对象的_callbacks = []。

• loop._ready = [handle(_run_until_complete_cb)]， loop._scheduled=[]。

第四轮_run_once()开始执行###

注意这里的第1,2,3,4,5步是在_run_once()上标记的1,2,3,4,5注释

• 第1,2步的逻辑判断，timeout = 0。

• 第3步 event_list = self._selector.select(0)，也就是说立即返回。

• 第4步 由于loop._scheduled=[]，因此不执行第4步中的逻辑。

• 第5步 依次从_ready队列中取出回调函数handle，执行handle._run()

执行handle._run()就是执行_run_until_complete_cb(task)

def _run_until_complete_cb(fut):

    exc = fut._exception

    if (isinstance(exc, BaseException)

    and not isinstance(exc, Exception)):

        # Issue #22429: run_forever() already finished, no need to

        # stop it.

        return

	# fut就是task对象，_loop.stop()就是把loop._stopping赋值为Ture

    fut._loop.stop()

loop._stopping为True。第4轮_run_once()执行完毕。

def run_forever(self):

		...

        try:

            events._set_running_loop(self)

            while True:

				# 第4轮_run_once()结束

                self._run_once()

				# _stopping为true，跳出循环，run_forever()执行结束

                if self._stopping:

                    break

        finally:

			...

跳出while循环， run_forever()执行结束，run_until_complete()也就执行完毕了，最后把cor协程的返回值'cor'返回出来赋值给rst变量。

到此为止所有整个task任务执行完毕，loop循环关闭。

总结

loop._ready和loop._scheduled是loop对象中非常重要的两个属性

• loop._ready是一个双端队列deque，用来存放调用loop.call_soon方法时中放入的回调函数。

• loop._scheduled是一个最小堆，根据延迟时间的大小构建的最小堆，用来存放调用loop.call_later时存放的延迟回调函数。

• loop._scheduled中有延迟调用函数是，timeout被赋值为堆顶的延迟函数的等待时间，这样会使得select(timeout)阻塞等待timeout秒。到时间后loop._scheduled中的回调函数最终还是会被转移到loop._ready队列中去执行。

每一个协程都会被封装到一个task对象中，task在初始化时就把task._step方法添加到loop._ready队列中，同时添加一个停止loop的事件循环的回调函数到task._callback列表中。

task._step函数就是驱动协程的，里面执行cor.send(None)，分两种情况：

• 第一种：cor中有await语句

  cor执行到await处，await的后面如果是另一个协程cor1，则进入到cor1中执行，cor1执行完毕，如果cor1返回的是None，
  
  则task._step方法会对result进行判断，返回None，执行task.set_result()，这个方法里
  
  面会调取task._callback列表中的所有回调方法，依次执行，此时，task._callback列表中只注册了一个停止loop事件
  
  循环的回调，此时就调用该回调函数，把loop._stopping设置成Ture，使得loop停止。

  如果cor1返回的是一个future对象，也就是task._step函数执行到cor1协程返回的一个cor1_future时，则task._step方
  
  法会对result进行判断，返回类型是future则对cor1_future添加一个回调函数task._wakeup，当cor1_future.set_result()
  
  在某一时刻被调用时（比如像sleep协程n秒后被调用，或者正在的IO事件触发的调用），会调用cor1_future添加的回调函数
  
  也就是task._wakeup函数，task._wakeup里面又会调用task._step来驱动上次cor停止的位置，也就是cor的await处，更准确的说是cor1_future.__iter__方法的yield self处，继续
  
  执行yield后面的语句，await cor1()才算真正的执行完毕，然后接着执行await cor1()下面的语句，直到cor执行完毕时会抛出StopIteration异常，cor返回值会保存在StopIteration异常对象的value
  
  属性中，与上面的逻辑一样，task会调用之前调用的停止loop的回调函数，停止loop循环

• 第二种：cor中没有await语句

  如果没有await语句，task._step执行后，result = cor.send(None)抛出StopIteration异常，与
  
  上面的逻辑一样，task会调用之前调用的停止loop的回调函数，停止loop循环

本质上，在一次loop循环中会执行一次result = cor.send(None)语句，也就是cor中的用户书写的逻辑，碰到await expr，再进

入到expr语句，直到碰到future的yield语句，这一次循环的result = cor.send(None)逻辑才算执行完成了。如果

await后压根没有返回future，result = cor.send(None)则会直接执行完await expr语句后再接着执行await expr下面的语句，直到

抛出StopIteration异常。

通过await future或者yield from future语句对cor的执行逻辑进行分割，yield之前的语句被封装到task._step方法中，在一次loop循环中被调用，yield之后的逻辑封装在task._wakeup方法中，在下一次的loop循环中被执行，而这个task._wakeup是由future.set_result()把它注册到loop._ready队列中的。

sleep协程模拟耗时IO操作，通过向loop中注册一个延迟回调函数，明确的控制select(timeout)中的timeout超时时间 + future对象的延迟timeout秒调用future.set_result()函数来实现一个模拟耗时的操作。

这个其实每一个真实的耗时的IO操作都会对应一个future对象。只不过sleep中的回调明确的传入了延迟回调的时间，而真实的IO操作时的future.set_result()的调用则是由真实的IO事件，也就是select(timeout)返回的socket对象的可读或者可写事件来触发的，一旦有相应的事件产生，就会回调对应的可读可写事件的回调函数，而在这些回调函数中又会去触发future.set_result()方法。

(上面的总结都是根据本人自己的理解所写，限于本人的表达能力有限，很多表达可能会产生一些歧义，还望各位看官一边看此文一边开启debug调试，来帮助自己理解。)