Python Asyncio中Coroutines,Tasks,Future可等待对象的关系及作用_python_

import asyncio async def main(): await asyncio.sleep(0) asyncio.run(main())

class Runner: ... def run(self, coro, *, context=None): """Run a coroutine inside the embedded event loop.""" # 省略代码 ... # 把coroutine转为task task = self._loop.create_task(coro, context=context) # 省略代码 ... try: # 如果传入的是Future或者coroutine，也会专为task return self._loop.run_until_complete(task) except exceptions.CancelledError: # 省略代码 ...

import asyncio async def main(): await asyncio.create_task(asyncio.sleep(0)) asyncio.run(main())

# demo_coro.py import asyncio import time async def main(): await asyncio.sleep(1) await asyncio.sleep(2) s_t = time.time() asyncio.run(main()) print(time.time() - s_t) # // Output: 3.0028765201568604 # demo_task.py import asyncio import time async def main(): task_1 = asyncio.create_task(asyncio.sleep(1)) task_2 = asyncio.create_task(asyncio.sleep(2)) await task_1 await task_2 s_t = time.time() asyncio.run(main()) print(time.time() - s_t) # // Output: 2.0027475357055664

class Future: ... def set_result(self, result): """设置结果，并安排下一个调用""" if self._state != _PENDING: raise exceptions.InvalidStateError(f'{self._state}: {self!r}') self._result = result self._state = _FINISHED self.__schedule_callbacks() def set_exception(self, exception): """设置异常，并安排下一个调用""" if self._state != _PENDING: raise exceptions.InvalidStateError(f'{self._state}: {self!r}') 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() self.__log_traceback = True def __await__(self): """设置为blocking，并接受await或者yield from调用""" if not self.done(): self._asyncio_future_blocking = True yield self # This tells Task to wait for completion. if not self.done(): raise RuntimeError("await wasn't used with future") return self.result() # May raise too. __iter__ = __await__ # make compatible with 'yield from'.

async def demo(future: asyncio.Future): await future print("aha")

class Task(futures._PyFuture): # Inherit Python Task implementation # from a Python Future implementation. _log_destroy_pending = True def __init__(self, coro, *, loop=None, name=None, context=None): super().__init__(loop=loop) # 省略部分初始化代码 ... # 托管的coroutine self._coro = coro if context is None: self._context = contextvars.copy_context() else: self._context = context # 通过loop.call_sonn，在Task初始化后马上就通知事件循环在下次有空的时候执行自己的__step函数 self._loop.call_soon(self.__step, context=self._context) def __step(self, exc=None): coro = self._coro # 方便asyncio自省 _enter_task(self._loop, self) # Call either coro.throw(exc) or coro.send(None). try: if exc is None: # 通过send预激托管的coroutine # 这时候只会得到coroutine yield回来的数据或者收到一个StopIteration的异常 # 对于Future或者Task返回的是Self result = coro.send(None) else: # 发送异常给coroutine result = coro.throw(exc) except StopIteration as exc: # StopIteration代表Coroutine运行完毕 if self._must_cancel: # coroutine在停止之前被执行了取消操作，则需要显示的执行取消操作 self._must_cancel = False super().cancel(msg=self._cancel_message) else: # 把运行完毕的值发送到结果值中 super().set_result(exc.value) # 省略其它异常封装 ... else: # 如果没有异常抛出 blocking = getattr(result, '_asyncio_future_blocking', None) if blocking is not None: # 通过Future代码可以判断，如果带有_asyncio_future_blocking属性，则代表当前result是Future或者是Task # 意味着这个Task里面裹着另外一个的Future或者Task # 省略Future判断 ... if blocking: # 代表这这个Future或者Task处于卡住的状态， # 此时的Task放弃了自己对事件循环的控制权，等待这个卡住的Future或者Task执行完成时唤醒一下自己 result._asyncio_future_blocking = False result.add_done_callback(self.__wakeup, context=self._context) self._fut_waiter = result if self._must_cancel: if self._fut_waiter.cancel(msg=self._cancel_message): self._must_cancel = False else: # 不能被await两次 new_exc = RuntimeError( f'yield was used instead of yield from ' f'in task {self!r} with {result!r}') self._loop.call_soon( self.__step, new_exc, context=self._context) elif result is None: # 放弃了对事件循环的控制权，代表自己托管的coroutine可能有个coroutine在运行，接下来会把控制权交给他和事件循环 # 当前的coroutine里面即使没有Future或者Task,但是子Future可能有 self._loop.call_soon(self.__step, context=self._context) finally: _leave_task(self._loop, self) self = None # Needed to break cycles when an exception occurs. def __wakeup(self, future): # 其它Task和Future完成后会调用到该函数，接下来进行一些处理 try: # 回收Future的状态，如果Future发生了异常，则把异常传回给自己 future.result() except BaseException as exc: # This may also be a cancellation. self.__step(exc) else: # Task并不需要自己托管的Future的结果值，而且如下注释，这样能使调度变得更快 # Don't pass the value of `future.result()` explicitly, # as `Future.__iter__` and `Future.__await__` don't need it. # If we call `_step(value, None)` instead of `_step()`, # Python eval loop would use `.send(value)` method call, # instead of `__next__()`, which is slower for futures # that return non-generator iterators from their `__iter__`. self.__step() self = None # Needed to break cycles when an exception occurs.

# demo_coro.py import asyncio import time async def main(): await asyncio.sleep(1) await asyncio.sleep(2) s_t = time.time() asyncio.run(main()) print(time.time() - s_t) # // Output: 3.0028765201568604