最近在看 ayncio 这块内容
于是打算从 asyncio.run 这个 函数作为入口看下内部是如何工作的
下面是源码以及一些分析
如有问题, 欢迎交流
另外 有没得大佬晓得 博客园的 markdown 怎么显示行号啊
def run(main, *, debug=False):
if events._get_running_loop() is not None: # 获取当前是否含有 loop
raise RuntimeError(
"asyncio.run() cannot be called from a running event loop")
if not coroutines.iscoroutine(main): # 判断是否是一个 coroutine
raise ValueError("a coroutine was expected, got {!r}".format(main))
loop = events.new_event_loop() # 获取一个 loop
"""
下面是获取一个 loop 的逻辑
"""
# get_event_loop_policy().new_event_loop()
# 其中 get_event_loop_policy 是获取 loop_policy
# 在 linux 下 默认是使用 _UnixDefaultEventLoopPolicy # asyncio.unix_events._UnixDefaultEventLoopPolicy
# loop 为 _UnixSelectorEventLoop 的一个实例 # asyncio.unix_events._UnixSelectorEventLoop
"""
获取 loop end
"""
try:
events.set_event_loop(loop) # get_event_loop_policy().set_event_loop(loop)
"""
get_event_loop_policy() -> _UnixDefaultEventLoopPolicy 一个实例
set_event_loop() # asyncio.events.BaseDefaultEventLoopPolicy#set_event_loop
给 loop_policy 设置 loop
放在 loop_policy 的 ThreadLocal 中 (引用变量)
ps. 最近被引用变量坑的不行...有点怕(弱引用)
"""
loop.set_debug(debug)
"""
# todo 暂时不处理
"""
return loop.run_until_complete(main)
"""
asyncio.base_events.BaseEventLoop#run_until_complete
"""
"""以下是 run_until_complete 的逻辑"""
def run_until_complete(self, future):
self._check_closed() # 暂时不管
new_task = not futures.isfuture(future) # 判断是否是 future toto task???
# todo asyncio.ensure_future asyncio.create_task 两者的差别
future = tasks.ensure_future(future, loop=self) # create task
"""
asyncio.tasks.ensure_future
看了下如果 future 是一个 coroutine 则会调用 create_task
asyncio.base_events.BaseEventLoop#create_task
# todo task 内容暂时不看
future 是 _asyncio.Task 或者是 _asyncio.Future -----asyncio.futures.Future(划掉)
Task 为 Future 的一个子类
"""
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
future.add_done_callback(_run_until_complete_cb)
"""
add_done_callback 是写在 so 文件中了 我并不知道他干了点啥 ~~~~(>_<)~~~~
可能类似这个吧:
def add_done_callback(self, fn, *, context=None):
'''Add a callback to be run when the future becomes done.
The callback is called with a single argument - the future object. If
the future is already done when this is called, the callback is
scheduled with call_soon.
'''
if self._state != _PENDING:
self._loop.call_soon(fn, self, context=context)
else:
if context is None:
context = contextvars.copy_context()
"""
# todo 有时间再看
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
finally:
future.remove_done_callback(_run_until_complete_cb)
if not future.done():
raise RuntimeError('Event loop stopped before Future completed.')
return future.result()
"""-------- run_until_complete --------- end"""
finally:
try:
_cancel_all_tasks(loop)
loop.run_until_complete(loop.shutdown_asyncgens())
finally:
events.set_event_loop(None)
loop.close()