asyncio.Future

第二次更新，18-12-26

Future 的作用

• 负责终止 loop 的循环。
  • 1、loop 停止循环的唯一条件为 loop._stopping = True
  • 2、将 loop 的 _stopping 设置为 True 的活，一般是 future 负责干。
  • 3、比如官方 sample 里

    import asyncio
    
    # 定义一个协程
    async def slow_operation(future):
        await asyncio.sleep(1)
        future.set_result('Future is done!')
    
    # 获得全局循环事件
    loop = asyncio.get_event_loop()
    # 实例化期物对象
    future = asyncio.Future()
    asyncio.ensure_future(slow_operation(future))
    # loop 的 run_until_complete 会将 _run_until_complete_cb 添加到 future 的完成回调列表中。而 _run_until_complete_cb 中会执行 loop.stop() 方法
    loop.run_until_complete(future)
    print(future.result())
    # 关闭事件循环对象
    loop.close()
    

  • 4、可以发现，loop 事件循环执行最外层一般是 Future 期物对象，期物会包裹其他 task 对象，在 future 对象的状态为 PENDING 时，会一直对 future 包裹的 task 对象调用 send(None) 方法，直到报出 StopIteration 方法。期物包裹的每一个 task 对象在执行完成后都会执行回调，如果期物包裹的所有 task 均执行完毕，则执行期物的 future.set_result() 方法，此方法不仅会将期物的状态设置为 FINISHED,还会执行期物的完成回调方法，执行 loop.stop() 方法。此时 loop 停止循环。

Future 的关键属性和方法

• 属性

  • _state = _PENDING # 表征状态，5 颗星
  • _result = None # 存储结果值，5 颗星
  • _exception
  • _loop # 5 颗星
  • _source_traceback
  • _asyncio_future_blocking = False # 用以在 task._step() 方法中判断是否为期物，以及期物是否已执行完毕，重要指数 5 颗星
  • _log_traceback

• 核心方法

  • __iter__ 方法 # 重要指数，5 颗星

    def __iter__(self):
      if not self.done():
      # 如果期物的状态非 FINISHED，则将其 _asyncio_future_blocking 属性设置为 True,并且 yield self，这个设置是告诉 task，期物未执行完毕，期物的值应该在期物执行完毕后再获取。
          self._asyncio_future_blocking = True
          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.
    

  • set_result 方法，4.5 颗星

    def set_result(self, result):
      if self._state != _PENDING:
          raise InvalidStateError('{}: {!r}'.format(self._state, self))
      self._result = result  # 给期物的结果值属性赋值
      self._state = _FINISHED  # 设置期物的状态
      self._schedule_callbacks()  # 执行期物的完成回调方法
    

  • add_done_callback 方法，4 颗星

      def add_done_callback(self, fn):
        if self._state != _PENDING:
            self._loop.call_soon(fn, self)  # 如果非 pending 状态，则立即执行完成回调方法
        else:
            self._callbacks.append(fn)  # 如果 pending 状态，则将回调添加到回调列表中，等待给期物设置 result 是调用执行
    

  • 其他的方法
    • __init__ 方法：初始化 loop 对象和 _callbacks 列表
    • cancel 方法：将 future 的状态变更为 CANCELLED, 执行完成回调列表
    • _schedule_callbacks 方法：执行 _callbacks 列表的回调

总结（期物的设计思想）

• 1、通过 _state 标记其当前状态
• 2、将标记 loop._stopping 为 True 的动作或者其他一些标记动作放到 _callbacks 中，在期物设置其结果值set_result()时，进行回调，从而达到终止 loop 循环的目的。
• 3、期物因为其__iter__方法的实现方式，
  • 在其 _state 为 PENDING 时，通过标记 _asyncio_future_blocking 为 True,且 yield self 告知 _step()当前期物的状态为 PENDING，需要将 _step() 操作添加到期物的 _callbacks 中，等期物的状态为 FINISHED 时，再获取期物的值。
  • 在 _state 为 FINISHED 时，直接获取期物的 result 值。
• 4、上述官方 sample 中：
  • _run_until_complete_cb 即负责设置 loop._stopping=True,在 loop.run_until_complete(future) 方法中，会将_run_until_complete_cb方法添加到 future 的 _callbacks 中，当 future 有结果值(set_result())时,执行回调，终止 loop 的执行。

Note:asyncio.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()
      future = loop.create_future()
      h = future._loop.call_later(delay,
                                  futures._set_result_unless_cancelled,
                                  future, result)
      try:
          return (yield from future)
      finally:
          h.cancel()
  

  • 1、当 delay 为0时，再次使用 send(None) 迭代执行获取 result 值；当 delay 不为0时，将 _set_result_unless_cancelled方法添加到延时 delay 后执行，_set_result_unless_cancelled是设置 future 的结果值。
  • 2、return (yield from future) 相当于

    data = None
    for _tmp in future:
        data = yield _tmp
    return data
    

    每个future 都会至少一次，至多两次执行 yield 语句来获取其结果值。

如何高效处理异步协程

• 1、关于 sleep(duration)
  • sleep 的设计思想是将回调绑定当前时间+duration实例化 TimeHandle 对象。
  • 将所有的 TimeHandle 对象添加到堆列表 leap 中，TimeHandle 对象使用其时间(time)属性比较大小。
  • loop 在循环执行 leap 列表时，会判定当前时间 current_time 是否小于 leap 的第一个值 near_time，如果小于则 time.sleep(near_time-current_time) ,然后执行第一个回调，否则立即执行第一个回调。
• 2、关于 I/O
  • 生成 socket 对象绑定 ip 和 port 然后进行监听，并且将 socket 对象及其回调注册到 selector 对象中，当 socket 收到可读或可写消息时(socket注册到selector时，可选择读消息或写消息或读写消息)，执行回调。
  • 回调的应用：一般会在回调中使用 accept 获取连接到当前 socket 的 cli_socket,然后将 cli_socket 注册到 selector 对象，客户端即可以和 cli_socket 进行通信。

分割线，第一次更新,18-03-03

• Future

class Future:
    """
    这个类同 concurrent.futures.Future基本上是兼容的
    
    Difference:
    
    - result() 和 exception() 不提供 timeout 参数，并且如果期物没有执行完毕会 raise 异常
    
    - 使用 add_done_callback() 注册的回调只能通过循环事件的 call_soon_threadsafe() 执行
    
    - 同 wait() 和 as_completed() 方法不兼容
    
    (Python 3.4 以及后面可能会统一这一块的处理)
    """

    # Class variables serving as defaults for instance variables.
    _state = _PENDING
    _result = None
    _exception = None
    _loop = None
    _source_traceback = None

    _blocking = False  # proper use of future (yield vs yield from)

    _log_traceback = False   # Used for Python 3.4 and later
    _tb_logger = None        # Used for Python 3.3 only

    def __init__(self, *, loop=None):
        """
        使用 loop 初始化期物实例
        """
        if loop is None:
            self._loop = events.get_event_loop()
        else:
            self._loop = loop
        self._callbacks = []
        if self._loop.get_debug():
            self._source_traceback = traceback.extract_stack(sys._getframe(1))

    # 格式化展示回调函数
    def __format_callbacks(self):
        cb = self._callbacks
        size = len(cb)
        if not size:
            cb = ''

        def format_cb(callback):
            return events._format_callback_source(callback, ())

        if size == 1:
            cb = format_cb(cb[0])
        elif size == 2:
            cb = '{}, {}'.format(format_cb(cb[0]), format_cb(cb[1]))
        elif size > 2:
            cb = '{}, <{} more>, {}'.format(format_cb(cb[0]),
                                            size-2,
                                            format_cb(cb[-1]))
        return 'cb=[%s]' % cb

    # 格式化展示信息
    def _repr_info(self):
        info = [self._state.lower()]
        if self._state == _FINISHED:
            if self._exception is not None:
                info.append('exception={!r}'.format(self._exception))
            else:
                # use reprlib to limit the length of the output, especially
                # for very long strings
                result = reprlib.repr(self._result)
                info.append('result={}'.format(result))
        if self._callbacks:
            info.append(self.__format_callbacks())
        if self._source_traceback:
            frame = self._source_traceback[-1]
            info.append('created at %s:%s' % (frame[0], frame[1]))
        return info

    def __repr__(self):
        info = self._repr_info()
        return '<%s %s>' % (self.__class__.__name__, ' '.join(info))

    # On Python 3.3 and older, objects with a destructor part of a reference
    # cycle are never destroyed. It's not more the case on Python 3.4 thanks
    # to the PEP 442.
    if compat.PY34:
        def __del__(self):
            if not self._log_traceback:
                # set_exception() was not called, or result() or exception()
                # has consumed the exception
                return
            exc = self._exception
            context = {
                'message': ('%s exception was never retrieved'
                            % self.__class__.__name__),
                'exception': exc,
                'future': self,
            }
            if self._source_traceback:
                context['source_traceback'] = self._source_traceback
            self._loop.call_exception_handler(context)

    def cancel(self):
        """
        取消期物，并且执行回调。如果期物已经 done 或者 cancelled，返回 False。
        变更期物的状态，通知事件循环去安排回调，并且返回 True
        """
        if self._state != _PENDING:
            return False
        self._state = _CANCELLED
        self._schedule_callbacks()
        return True

    def _schedule_callbacks(self):
        """
        通知事件循环尽快的执行回调，清空 callback 列表
        """
        callbacks = self._callbacks[:]
        if not callbacks:
            return

        self._callbacks[:] = []
        for callback in callbacks:
            self._loop.call_soon(callback, self)

    def cancelled(self):
        return self._state == _CANCELLED

    # Don't implement running(); see http://bugs.python.org/issue18699

    def done(self):
        """
        期物已经运行结束或取消都是 done 的状态
        """
        
        return self._state != _PENDING

    def result(self):
        """
        不会阻塞等待期物的执行，如果期物的状态非 _FINISHED，则 raise 对应的错误。如果 _exception 值存在，则 raise 对应的值，否则返回 _result值
        """
        if self._state == _CANCELLED:
            raise CancelledError
        if self._state != _FINISHED:
            raise InvalidStateError('Result is not ready.')
        self._log_traceback = False
        if self._tb_logger is not None:
            self._tb_logger.clear()
            self._tb_logger = None
        if self._exception is not None:
            raise self._exception
        return self._result

    def exception(self):
        """
        不会阻塞等待期物的执行，如果期物的状态非 _FINISHED，则 raise 对应的错误。返回 _exception 值
        """
        if self._state == _CANCELLED:
            raise CancelledError
        if self._state != _FINISHED:
            raise InvalidStateError('Exception is not set.')
        self._log_traceback = False
        if self._tb_logger is not None:
            self._tb_logger.clear()
            self._tb_logger = None
        return self._exception

    def add_done_callback(self, fn):
        """
        给期物运行完毕添加方法回调
        如果期物已经运行完毕，则立即回调
        """
        if self._state != _PENDING:
            self._loop.call_soon(fn, self)
        else:
            self._callbacks.append(fn)

    # New method not in PEP 3148.

    def remove_done_callback(self, fn):
        """
        移除指定回调
        """
        filtered_callbacks = [f for f in self._callbacks if f != fn]
        removed_count = len(self._callbacks) - len(filtered_callbacks)
        if removed_count:
            self._callbacks[:] = filtered_callbacks
        return removed_count

    # So-called internal methods (note: no set_running_or_notify_cancel()).

    def set_result(self, result):
        """
        标记期物运行结束，设置 _result 值，如果在期物运行结束时调用这个方法，则 raise InvalidStateError
        """
        if self._state != _PENDING:
            raise InvalidStateError('{}: {!r}'.format(self._state, self))
        self._result = result
        self._state = _FINISHED
        self._schedule_callbacks()  # 运行回调

    def set_exception(self, exception):
        """
        标记期物运行结束，设置 _exception 值，如果在期物运行结束时调用这个方法，则 raise InvalidStateError
        """idStateError.
        if self._state != _PENDING:
            raise InvalidStateError('{}: {!r}'.format(self._state, self))
        if isinstance(exception, type):
            exception = exception()
        if type(exception) is StopIteration:
            raise TypeError("StopIteration interacts badly with generators "
                            "and cannot be raised into a Future")
        self._exception = exception
        self._state = _FINISHED
        self._schedule_callbacks()
        if compat.PY34:
            self._log_traceback = True
        else:
            self._tb_logger = _TracebackLogger(self, exception)
            # Arrange for the logger to be activated after all callbacks
            # have had a chance to call result() or exception().
            self._loop.call_soon(self._tb_logger.activate)

    def __iter__(self):
        if not self.done():
            self._blocking = True
            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:
        # 兼容 async/await 方法
        __await__ = __iter__ # make compatible with 'await' expression