asyncio源码分析之基本执行流程

原生协程是基于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挂起的地方,继续执行corawait后面的语句。直到最后抛出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)时,会在cor2yield关键字处挂起,第二次c.send(None)则会在yield挂起的地方,接着往下执行,然后core2返回'cor2',赋值给rst变量,继续执行corawait后面的语句,直到最后抛出StopIteration异常。

总结

async 是一个关键字,async def 定义的类型还是一个function类型,只有当它被调用时才返回一个协程对象
async defdef定义的方法在没被调用时没有任何区别,不必看得很神秘,它也可以有return语句,这点也正常,
因为python中没有return语句的函数实际上默认是返回None的,所以只是显式的return和隐式return None的区别

对于协程的send(None)方法,跟生成器的send(None)类似,不同的是,原生协程的send方法被调用的时候,会一直执行到碰
await语句,但是不会停下来,会直接再进入到await EXPREXPR中,其实EXPR是一个awaitable对象,会调用该对象的
__await__()执行该方法的里面的逻辑,如果该awaitable对象是一个原生协程对象,那么它的__await__()方法中的逻辑就
是在定义此协程时async def 下面的逻辑,执行完毕后,该协程对象就关闭了,执行流程就再次跳转到当前挂起的协程中,
执行该协程中余下的逻辑,最后执行完毕,抛出StopIteration异常。

对于原生生协程来说,调用send()方法时,会一直执行到出现StopIteration异常为止,只有在 __await__()方法中有yield语句时才
会挂起在那里,如果__await__()方法中没有yield语句,不会挂起,会返回await的返回值,继续执行,直到抛出StopIteration异常。

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

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和self._scheduled中有一个不为空,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执行完毕时,判断_stoppingTrue就跳出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)不抛出异常
        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._stoppingTrue。第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._readyloop._scheduledloop对象中非常重要的两个属性

  • 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停止的位置,也就是corawait处,更准确的说是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语句,直到碰到futureyield语句,这一次循环的result = cor.send(None)逻辑才算执行完成了。如果
await后压根没有返回futureresult = 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调试,来帮助自己理解。)

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 212,222评论 6 493
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,455评论 3 385
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 157,720评论 0 348
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,568评论 1 284
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,696评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,879评论 1 290
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,028评论 3 409
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,773评论 0 268
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,220评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,550评论 2 327
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,697评论 1 341
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,360评论 4 332
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 40,002评论 3 315
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,782评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,010评论 1 266
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,433评论 2 360
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,587评论 2 350

推荐阅读更多精彩内容