快乐的 asyncio 学习

其实 asyncio 的学习一点也不快乐

一、python 的多线程和多进程

要想理解 asyncio 的异步编程，需要简单了解一下 python 的多线程和多进程知识

• 1、多线程

python 有 GIL 机制，因此，python 的多线程操作并非真正意义的多线程，而仅仅在线程处于睡眠或者等待 I/O 时，才会发挥真正的多线程功能。

• 1.1、睡眠

  • time.sleep()
  • threading.Lock
  • 线程模块其他同步对象

• 1.2、I/O

  • requests
  • open

• 1.3、定期放弃GIL

  • py2 解释器每执行 1000 字节码释放 GIL
  • py3 解释器每执行 15ms 释放 GIL

• 1.4、GIL 全局解释器锁

  • 一个线程运行 Python，而其他 N 个睡眠或者等待 I/O。(保证同一时刻只有一个线程对共享资源进行存取)

• 1.5、GIL 原理

    /* s.connect((host, port)) method */
    static PyObject *
    sock_connect(PySocketSockObject *s, PyObject *addro)
    {
        sock_addr_t addrbuf;
        int addrlen;
        int res;
     
        /* convert (host, port) tuple to C address */
        getsockaddrarg(s, addro, SAS2SA(&addrbuf), &addrlen);
     
        Py_BEGIN_ALLOW_THREADS
        res = connect(s->sock_fd, addr, addrlen);
        Py_END_ALLOW_THREADS
     
        /* error handling and so on .... */
    }
  

  • Py_BEGIN_ALLOW_THREADS 放弃 GIL
  • Py_END_ALLOW_THREADS 重新获取 GIL，一个线程会在这个位置阻塞，等待另一个线程释放锁；一旦出现这个情况，等待的线程会抢夺回锁，并恢复python代码的执行
  • 简而言之：当N个线程在网络 I/O 堵塞，或等待重新获取GIL，而一个线程运行Python

• 1.6、示例

  • 睡眠阻塞

    import time
    from threading import Thread
    from datetime import datetime
    
    def write(i):
        print('{} start write --> {}'.format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), i))
        time.sleep(4)
        print('{} end write --> {}'.format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), i))
    
    def fun():
        print('start ...')
        for i in range(3):
            Thread(target=write, args=(i,), daemon=False).start()
        print('end ...')
    # 输出结果
    start ...
    2018-02-09 23:58:25 start write --> 0
    2018-02-09 23:58:25 start write --> 1
    2018-02-09 23:58:25 start write --> 2
    end ...
    2018-02-09 23:58:29 end write --> 0
    2018-02-09 23:58:29 end write --> 1
    2018-02-09 23:58:29 end write --> 2
  

  • CPU 阻塞

    import time
    from threading import Thread
    from datetime import datetime
    
    def write(n):
        print('{} start write --> {}'.format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), n))
        l, sum_ = list(range(100000000)), 0
        for i in l:
            sum_ += i
        print('{} end write --> {}'.format(datetime.now().strftime('%Y-%m-%d %H:%M:%S'), n))
    
    def fun():
        print('start ...')
        for i in range(3):
            Thread(target=write, args=(i,), daemon=False).start()
        print('end ...')
    # 输出结果
    start ...
    2018-02-10 00:13:55 start write --> 0
    2018-02-10 00:13:58 start write --> 1
    2018-02-10 00:14:02 start write --> 2
    end ...
    2018-02-10 00:14:27 end write --> 0
    2018-02-10 00:14:32 end write --> 1
    2018-02-10 00:14:35 end write --> 2
  

  • 总结
    • 对于睡眠操作或者 I/O 操作，多线程的作用非常明显，明显减少所消耗总时间；
    • 对于 CPU 计算型操作，多线程操作反而因为多线程间获取 GIL 而增加总的消耗时间。

• 2、python 多进程

python多进程即其他语言中的多进程概念，不再累述

二、异步编程思想

• 1、协程(coroutine)
• 2、任务(Task)
• 3、事件循环(loop)

1、Task 对象包含 协程(coro) 和协程调用时间2个属性；
2、Loop 对象使用堆承载多个 Task 对象，根据 Task 对象中最小调用时间去执行对应的 coro。如果 coro 没有迭代完，则将此 coro 生成新的 task，然后 push 到 Loop 对象的堆中。

# 简单的调用示例
import asyncio

@asyncio.coroutine
def coro_fun():
    yield from range(10)

loop = asyncio.get_event_loop()

loop.run_until_complete(coro_fun()) 
# or
tasks = [asyncio.ensure_future(coro_fun())]
loop.run_until_complete(asyncio.wait(tasks))

三、源代码分析

• 关于 _ code _.co_flags

  # 每个函数或方法都有 __code__ 魔法方法 以及其对应的 co_flags 值
  # 在 Cpython 中，
  1、生成器函数的标识符为 CO_GENERATOR 即 0x20，
  2、协程函数的标识符为 CO_COROUTINE 即 0x180
  3、CO_ITERABLE_COROUTINE 即 0x100
  
  # 通过对函数对象的 __code__.co_flags 与 对应的标识符做位与运算，如果是真值，则表明函数对象属于生成器函数或协程函数
  
  def gen_fun():
      yield from range(10)
  >>> gen_fun.__code__.co_flags  # 99
  >>> 99 & 0x20  # 32, True
  >>> 99 & 0x180  # 0, False
  
  async def asy_fun():
      await sleep(4)
  >>> asy_fun.__code__.co_flags  # 227
  >>> 99 & 0x20  # 32, True
  >>> 99 & 0x180  # 128, True
  

• 关于类型判断

  from collections import Iterator, Awaitable
  # 判断迭代器 和 Awaitable 对象
  class A:
      def __iter__(self):
          return iter([1,2,3,4,5])
      def __await__(self):
          return iter([1,2,3,4,5])
  a = A()
  >>> isinstance(a, Iterator)  # True
  >>> isinstance(a, Awaitable)  # True
  
  # 判断是否为协程等
  import inspect
  async def asy_fun():
      await a
  >>> inspect.iscoroutine(asy_fun())  # True
  

• @asyncio.coroutine

  def coroutine(func):
      # 将一个生成器标记为协程，如果在destroyed前没有调用，则会记录错误
  
      # 这个方法是使用 inspect.iscoroutinefunction 方法判断是否为协程方法，使用 types.coroutine 装饰的生成器，或 async def 语法定义的函数都会返回 True
      if _inspect_iscoroutinefunction(func):
          return func
  
      # 使用 co_flags 判断是否为生成器
      if inspect.isgeneratorfunction(func):
          coro = func
      else:
          @functools.wraps(func)
          def coro(*args, **kw):
              res = func(*args, **kw)
              
              # 判断 res 是否为期物，生成器 或 协程包装类 实例
              if isinstance(res, futures.Future) or inspect.isgenerator(res) or \
                      isinstance(res, CoroWrapper):
                  res = yield from res
  
              elif _AwaitableABC is not None:
                  # py 3.5 才会有 Awaitable 类
                  try:
                      # 如果有 __await__属性，__await__属性只会返回一个不是协程的迭代器
                      await_meth = res.__await__
                  except AttributeError:
                      pass
                  else:
                      # 如果是 Awaitable 对象
                      if isinstance(res, _AwaitableABC):
                          # 使用 yield from 处理其迭代器
                          res = yield from await_meth()
              return res
  
      # 使用 types.coroutine 包装 coro(注意，多层 @types.coroutine 装饰不会影响,会直接return装饰的值)
      if not _DEBUG:
          if _types_coroutine is None:
              wrapper = coro
          else:
              wrapper = _types_coroutine(coro)
      else:
          @functools.wraps(func)
          def wrapper(*args, **kwds):
            
              # 使用协程包装器处理
              w = CoroWrapper(coro(*args, **kwds), func=func)
              if w._source_traceback:
                  del w._source_traceback[-1]
              # 如果是 py 3.5 则包装增加 协程 对象的属性，否则包装为 生成器 对象的属性
              w.__name__ = getattr(func, '__name__', None)
              w.__qualname__ = getattr(func, '__qualname__', None)
              return w
      
      # 用以别处使用 asyncio.iscoroutinefunction() 判断为 True 的作用
      wrapper._is_coroutine = True  # For iscoroutinefunction().
      return wrapper
  

• @types.coroutine

  def coroutine(func):
    # 将一个普通的生成器函数转化为协程
  
    if not callable(func):
        raise TypeError('types.coroutine() expects a callable')
  
    if (func.__class__ is FunctionType and
        getattr(func, '__code__', None).__class__ is CodeType):
  
        # 获取函数的 co_flags
        co_flags = func.__code__.co_flags
  
        # 检查是否为协程函数
        if co_flags & 0x180:
            return func
  
        # 检查是否为生成器函数，此步主要作用是将生成器的 co_flags 同 0x100 做位或运算，将其标识变更为协程标识
        if co_flags & 0x20:
            # TODO: Implement this in C.
            co = func.__code__
            func.__code__ = CodeType(
                co.co_argcount, co.co_kwonlyargcount, co.co_nlocals,
                co.co_stacksize,
                co.co_flags | 0x100,  # 0x100 == CO_ITERABLE_COROUTINE
                co.co_code,
                co.co_consts, co.co_names, co.co_varnames, co.co_filename,
                co.co_name, co.co_firstlineno, co.co_lnotab, co.co_freevars,
                co.co_cellvars)
            return func
  
    # 用以支持类似生成器的对象
  
    @_functools.wraps(func)
    def wrapped(*args, **kwargs):
        coro = func(*args, **kwargs)
  
        # 协程或 co_flags 大于 256 的生成器对象，直接返回
        if (coro.__class__ is CoroutineType or
            coro.__class__ is GeneratorType and coro.gi_code.co_flags & 0x100):
            return coro
        if (isinstance(coro, _collections_abc.Generator) and
            not isinstance(coro, _collections_abc.Coroutine)):
            # 实现了生成器抽象类的方法，使用生成器包装器处理成生成器
            return _GeneratorWrapper(coro)
        # 协程抽象类实例或其他对象
        return coro
  
    return wrapped
  

• asyncio.get_event_loop
  • 1、DefaultEventLoopPolicy().get_event_loop()

    # DefaultEventLoopPolicy = _UnixDefaultEventLoopPolicy
    
    class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
    # 事件循环且监听子进程
    _loop_factory = _UnixSelectorEventLoop
    
    def __init__(self):
      super().__init__()
      self._watcher = None
    
    def _init_watcher(self):
      with events._lock:
          if self._watcher is None:  # pragma: no branch
              self._watcher = SafeChildWatcher()
              if isinstance(threading.current_thread(),
                            threading._MainThread):
                  self._watcher.attach_loop(self._local._loop)
    
    def set_event_loop(self, loop):
      # 如果子监听已经设置，那么在主线程中调用 .set_event_loop() 会在子监听中调用 .attach_loop(loop)
      super().set_event_loop(loop)
    
      if self._watcher is not None and \
          isinstance(threading.current_thread(), threading._MainThread):
          self._watcher.attach_loop(loop)
    
    def get_child_watcher(self):
      """
      Get the watcher for child processes.
      If not yet set, a SafeChildWatcher object is automatically created.
      """
      if self._watcher is None:
          self._init_watcher()
    
      return self._watcher
    
    def set_child_watcher(self, watcher):
      """Set the watcher for child processes."""
    
      assert watcher is None or isinstance(watcher, AbstractChildWatcher)
    
      if self._watcher is not None:
          self._watcher.close()
    
      self._watcher = watcher
    

  • 2、父类BaseDefaultEventLoopPolicy().get_event_loop()

    class BaseDefaultEventLoopPolicy(AbstractEventLoopPolicy):
    
    # 此 policy 下，每一个线程拥有独立的事件循环，然而我们仅会默认在主线程创建一个事件循环，其他线程默认没有事件循环
    # 其他 policy 下有不同的规则(e.g. 一个全局的事件循环，或者每一个线程自动创建一个事件循环，或者使用其他上下文将事件循环关联起来)
    
    _loop_factory = None
    
    class _Local(threading.local):
        _loop = None
        _set_called = False
    
    def __init__(self):
        self._local = self._Local()
    
    def get_event_loop(self):
        # 如果主线程 且 self._local._loop == None 且 self._local._set_called == False
        if (self._local._loop is None and
            not self._local._set_called and
            isinstance(threading.current_thread(), threading._MainThread)):
            self.set_event_loop(self.new_event_loop())
        if self._local._loop is None:
            raise RuntimeError('There is no current event loop in thread %r.'
                               % threading.current_thread().name)
        return self._local._loop
    
    def set_event_loop(self, loop):
        # 设置事件循环
        self._local._set_called = True
        assert loop is None or isinstance(loop, AbstractEventLoop)
        self._local._loop = loop
    
    def new_event_loop(self):
        # 调用 set_event_loop() 设置，因为还牵扯到 _local._set_called 的设置
        return self._loop_factory()
    

  • 3、_UnixSelectorEventLoop 事件循环工厂类

    class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
    """Unix event loop.
    
    Adds signal handling and UNIX Domain Socket support to SelectorEventLoop.
    """
    
       def __init__(self, selector=None):
           super().__init__(selector)
           self._signal_handlers = {}
    
       def _socketpair(self):
           return socket.socketpair()
    
       def close(self):
           super().close()
           for sig in list(self._signal_handlers):
               self.remove_signal_handler(sig)
    
       def _process_self_data(self, data):
           for signum in data:
               if not signum:
                   # ignore null bytes written by _write_to_self()
                   continue
               self._handle_signal(signum)
    
       def add_signal_handler(self, sig, callback, *args):
           """Add a handler for a signal.  UNIX only.
    
           Raise ValueError if the signal number is invalid or uncatchable.
           Raise RuntimeError if there is a problem setting up the handler.
           """
           if (coroutines.iscoroutine(callback)
           or coroutines.iscoroutinefunction(callback)):
               raise TypeError("coroutines cannot be used "
                               "with add_signal_handler()")
           self._check_signal(sig)
           self._check_closed()
           try:
               # set_wakeup_fd() raises ValueError if this is not the
               # main thread.  By calling it early we ensure that an
               # event loop running in another thread cannot add a signal
               # handler.
               signal.set_wakeup_fd(self._csock.fileno())
           except (ValueError, OSError) as exc:
               raise RuntimeError(str(exc))
    
           handle = events.Handle(callback, args, self)
           self._signal_handlers[sig] = handle
    
           try:
               # Register a dummy signal handler to ask Python to write the signal
               # number in the wakup file descriptor. _process_self_data() will
               # read signal numbers from this file descriptor to handle signals.
               signal.signal(sig, _sighandler_noop)
    
               # Set SA_RESTART to limit EINTR occurrences.
               signal.siginterrupt(sig, False)
           except OSError as exc:
               del self._signal_handlers[sig]
               if not self._signal_handlers:
                   try:
                       signal.set_wakeup_fd(-1)
                   except (ValueError, OSError) as nexc:
                       logger.info('set_wakeup_fd(-1) failed: %s', nexc)
    
               if exc.errno == errno.EINVAL:
                   raise RuntimeError('sig {} cannot be caught'.format(sig))
               else:
                   raise
    
       def _handle_signal(self, sig):
           """Internal helper that is the actual signal handler."""
           handle = self._signal_handlers.get(sig)
           if handle is None:
               return  # Assume it's some race condition.
           if handle._cancelled:
               self.remove_signal_handler(sig)  # Remove it properly.
           else:
               self._add_callback_signalsafe(handle)
    
       def remove_signal_handler(self, sig):
           """Remove a handler for a signal.  UNIX only.
    
           Return True if a signal handler was removed, False if not.
           """
           self._check_signal(sig)
           try:
               del self._signal_handlers[sig]
           except KeyError:
               return False
    
           if sig == signal.SIGINT:
               handler = signal.default_int_handler
           else:
               handler = signal.SIG_DFL
    
           try:
               signal.signal(sig, handler)
           except OSError as exc:
               if exc.errno == errno.EINVAL:
                   raise RuntimeError('sig {} cannot be caught'.format(sig))
               else:
                   raise
    
           if not self._signal_handlers:
               try:
                   signal.set_wakeup_fd(-1)
               except (ValueError, OSError) as exc:
                   logger.info('set_wakeup_fd(-1) failed: %s', exc)
    
           return True
    
       def _check_signal(self, sig):
           """Internal helper to validate a signal.
    
           Raise ValueError if the signal number is invalid or uncatchable.
           Raise RuntimeError if there is a problem setting up the handler.
           """
           if not isinstance(sig, int):
               raise TypeError('sig must be an int, not {!r}'.format(sig))
    
           if not (1 <= sig < signal.NSIG):
               raise ValueError(
                   'sig {} out of range(1, {})'.format(sig, signal.NSIG))
    
       def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
                                     extra=None):
           return _UnixReadPipeTransport(self, pipe, protocol, waiter, extra)
    
       def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
                                      extra=None):
           return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra)
    
       @coroutine
       def _make_subprocess_transport(self, protocol, args, shell,
                                      stdin, stdout, stderr, bufsize,
                                      extra=None, **kwargs):
           with events.get_child_watcher() as watcher:
               waiter = self.create_future()
               transp = _UnixSubprocessTransport(self, protocol, args, shell,
                                                 stdin, stdout, stderr, bufsize,
                                                 waiter=waiter, extra=extra,
                                                 **kwargs)
    
               watcher.add_child_handler(transp.get_pid(),
                                         self._child_watcher_callback, transp)
               try:
                   yield from waiter
               except Exception as exc:
                   # Workaround CPython bug #23353: using yield/yield-from in an
                   # except block of a generator doesn't clear properly
                   # sys.exc_info()
                   err = exc
               else:
                   err = None
    
               if err is not None:
                   transp.close()
                   yield from transp._wait()
                   raise err
    
           return transp
    
       def _child_watcher_callback(self, pid, returncode, transp):
           self.call_soon_threadsafe(transp._process_exited, returncode)
    
       @coroutine
       def create_unix_connection(self, protocol_factory, path, *,
                                  ssl=None, sock=None,
                                  server_hostname=None):
           assert server_hostname is None or isinstance(server_hostname, str)
           if ssl:
               if server_hostname is None:
                   raise ValueError(
                       'you have to pass server_hostname when using ssl')
           else:
               if server_hostname is not None:
                   raise ValueError('server_hostname is only meaningful with ssl')
    
           if path is not None:
               if sock is not None:
                   raise ValueError(
                       'path and sock can not be specified at the same time')
    
               sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
               try:
                   sock.setblocking(False)
                   yield from self.sock_connect(sock, path)
               except:
                   sock.close()
                   raise
    
           else:
               if sock is None:
                   raise ValueError('no path and sock were specified')
               sock.setblocking(False)
    
           transport, protocol = yield from self._create_connection_transport(
               sock, protocol_factory, ssl, server_hostname)
           return transport, protocol
    
       @coroutine
       def create_unix_server(self, protocol_factory, path=None, *,
                              sock=None, backlog=100, ssl=None):
           if isinstance(ssl, bool):
               raise TypeError('ssl argument must be an SSLContext or None')
    
           if path is not None:
               if sock is not None:
                   raise ValueError(
                       'path and sock can not be specified at the same time')
    
               sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    
               try:
                   sock.bind(path)
               except OSError as exc:
                   sock.close()
                   if exc.errno == errno.EADDRINUSE:
                       # Let's improve the error message by adding
                       # with what exact address it occurs.
                       msg = 'Address {!r} is already in use'.format(path)
                       raise OSError(errno.EADDRINUSE, msg) from None
                   else:
                       raise
               except:
                   sock.close()
                   raise
           else:
               if sock is None:
                   raise ValueError(
                       'path was not specified, and no sock specified')
    
               if sock.family != socket.AF_UNIX:
                   raise ValueError(
                       'A UNIX Domain Socket was expected, got {!r}'.format(sock))
    
           server = base_events.Server(self, [sock])
           sock.listen(backlog)
           sock.setblocking(False)
           self._start_serving(protocol_factory, sock, ssl, server)
           return server
    

• BaseEventLoop().run_until_complete

  def run_until_complete(self, future):
        """ 
        1、参数为协程，使用asyncio.wait()值也是协程
        2、会将协程包装成期物
        3、运行直到期物状态为 done
        """
  
        # 检查 loop 是否 close
        self._check_closed()
  
        new_task = not isinstance(future, futures.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
  
        # 添加 done 回调
        future.add_done_callback(_run_until_complete_cb)
        try:
            self.run_forever()
        except:
            if new_task and future.done() and not future.cancelled():
                # The coroutine raised a BaseException. Consume the exception
                # to not log a warning, the caller doesn't have access to the
                # local task.
                future.exception()
            raise
        future.remove_done_callback(_run_until_complete_cb)
        if not future.done():
            raise RuntimeError('Event loop stopped before Future completed.')
  
        return future.result()
  

• asyncio.ensure_future

  def ensure_future(coro_or_future, *, loop=None):
    
    # 把一个协程或者awaitable对象包装成期物(Future)
    if isinstance(coro_or_future, futures.Future):
        if loop is not None and loop is not coro_or_future._loop:
            raise ValueError('loop argument must agree with Future')
        # 直接返回
        return coro_or_future
  
    # 判断是否为生成器或协程包装器
    elif coroutines.iscoroutine(coro_or_future):
        if loop is None:
            loop = events.get_event_loop()
        # 创建 task 对象
        task = loop.create_task(coro_or_future)
        if task._source_traceback:
            del task._source_traceback[-1]
        return task
    
    # py 3.5下，判断对象是否为 awaitable 对象
    elif compat.PY35 and inspect.isawaitable(coro_or_future):
        return ensure_future(_wrap_awaitable(coro_or_future), loop=loop)
    else:
        raise TypeError('A Future, a coroutine or an awaitable is required')
  

• asyncio.wait

  @coroutine
  def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED):
    """
    fs 参数是 futures 或者 coroutines 序列，且不能为空
    return two sets of Future: (done, pending).
    用法:
        done, pending = yield from asyncio.wait(fs)
    """
    if isinstance(fs, futures.Future) or coroutines.iscoroutine(fs):
        raise TypeError("expect a list of futures, not %s" % type(fs).__name__)
    if not fs:
        raise ValueError('Set of coroutines/Futures is empty.')
    if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
        raise ValueError('Invalid return_when value: {}'.format(return_when))
  
    if loop is None:
        loop = events.get_event_loop()
  
    # 把 fs 中的参数转化为期物 future 对象的集合
    fs = {ensure_future(f, loop=loop) for f in set(fs)}
  
    return (yield from _wait(fs, timeout, return_when, loop))
  

• asyncio._wait

  @coroutine
  def _wait(fs, timeout, return_when, loop):
    # fs 必须是集合
    assert fs, 'Set of Futures is empty.'
    
    # 创建一个 future 对象，附加到当前 loop 上
    waiter = loop.create_future()
    timeout_handle = None
  
    if timeout is not None:
        # 在给定的时间，进行超时回调
        timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
    counter = len(fs)
  
    def _on_completion(f):
        # 声明使用外层变量 counter
        nonlocal counter
        counter -= 1
        """ 
        以下三种状态均表示 waiter 期物运行结束
        1、所有的期物都运行完成 
        2、return_when 值为 FIRST_COMPLETED
        3、return_when 值为 FIRST_EXCEPTION 并且期物没有被取消，且 f.exception()值存在
        """
        if (counter <= 0 or
            return_when == FIRST_COMPLETED or
            return_when == FIRST_EXCEPTION and (not f.cancelled() and
                                                f.exception() is not None)):
            # 如果超时回调函数存在，则取消
            if timeout_handle is not None:
                timeout_handle.cancel()
            # 变更 waiter 期物的状态为结束
            if not waiter.done():
                waiter.set_result(None)
  
    for f in fs:
        # 给期物添加运行完成后的回调
        f.add_done_callback(_on_completion)
  
    try:
        yield from waiter
    finally:
        if timeout_handle is not None:
            timeout_handle.cancel()
  
    done, pending = set(), set()
    for f in fs:
        # 将 _on_completion 从 done 回调列表移除
        f.remove_done_callback(_on_completion)
        if f.done():
            done.add(f)
        else:
            pending.add(f)
    return done, pending