从头造轮子:python3 asyncio 之 sleep (4
从头造轮子:asyncio之sleep
(4)
前言
书接上文:,本文造第四个轮子,也是asyncio包里面非常常用,并且非常复杂的一个函数sleep
一、知识准备
● time.sleep
直接让当前线程睡觉,但是这种方式显然是不能接受的,如果当前线程睡觉,那我们所有的协程任务都会被卡主,并发也就无从谈起了
● 理解socket.socketpair()
创建的套接字对象
● 理解selectors
的应用
● 理解最小堆以及heapq
的应用
● 理解对象比较
● 这一小结的基础知识很多,希望大家优先了解上述的知识再开始阅读,否则很容易不知所云
二、环境准备
组件 | 版本 |
---|---|
python | 3.7.7 |
三、
sleep
的实现
先来看下官方sleep的使用方法:
|># more main.py
import asyncio
async def hello():
print('enter hello ...')
await asyncio.sleep(5)
print('hello sleep end...')
return 'return hello...'
async def world():
print('enter world ...')
await asyncio.sleep(3)
print('world sleep end...')
return 'return world...'
async def helloworld():
print('enter helloworld')
ret = await asyncio.gather(hello(), world())
print('exit helloworld')
return ret
if __name__ == "__main__":
ret = asyncio.run(helloworld())
print(ret)
|># time python3 main.py
enter helloworld
enter hello ...
enter world ...
world sleep end...
hello sleep end...
exit helloworld
['return hello...', 'return world...']
real 0m5.256s
user 0m0.077s
sys 0m0.020s
来看下造的轮子的使用方式:
▶ more main.py
async def hello():
print('enter hello ...')
await wilsonasyncio.sleep(5)
print('hello sleep end...')
return 'return hello...'
async def world():
print('enter world ...')
await wilsonasyncio.sleep(3)
print('world sleep end...')
return 'return world...'
async def helloworld():
print('enter helloworld')
ret = await wilsonasyncio.gather(hello(), world())
print('exit helloworld')
return ret
if __name__ == "__main__":
ret = wilsonasyncio.run(helloworld())
print(ret)
▶ time python3 main.py
enter helloworld
enter hello ...
enter world ...
world sleep end...
hello sleep end...
exit helloworld
['return hello...', 'return world...']
python3 main.py 0.06s user 0.04s system 1% cpu 5.406 total
都是用了5s左右,自己造的轮子也很好的运行了,下面我们来看下轮子的代码
四、代码解析
1)代码组成
|># tree
.
├── eventloops.py
├── futures.py
├── main.py
├── tasks.py
├── wilsonasyncio.py
文件 | 作用 |
---|---|
eventloops.py | 事件循环 |
futures.py | futures对象 |
tasks.py | tasks对象 |
wilsonasyncio.py | 可调用方法集合 |
main.py | 入口 |
2)代码概览:
eventloops.py
类/函数 | 方法 | 对象 | 作用 | 描述 |
---|---|---|---|---|
Eventloop | 事件循环,一个线程只有运行一个 | |||
__init__ |
初始化两个重要对象 self._ready 与 self._stopping
|
|||
self._ready |
所有的待执行任务都是从这个队列取出来,非常重要 | |||
self._scheduled |
待调度的任务队列,这个队列的任务会在合适的时机进入self._ready 中 |
新增 | ||
self._stopping |
事件循环完成的标志 | |||
current_time |
从线程启动到现在经历的秒数 | 新增 | ||
call_later |
调用该方法会经历一个延时delay 之后,再将任务添加到待执行队列 |
新增 | ||
call_at |
调用该方法会在指定的时间将任务添加到待执行队列 | 新增 | ||
call_soon |
调用该方法会立即将任务添加到待执行队列 | |||
run_once |
被run_forever 调用,从self._ready 队列里面取出任务执行 |
重新改造,添加了大量逻辑 | ||
run_forever |
死循环,若self._stopping 则退出循环 |
|||
run_until_complete |
非常重要的函数,任务的起点和终点(后面详细介绍) | |||
create_task |
将传入的函数封装成task 对象,这个操作会将task.__step 添加到__ready 队列 |
|||
create_future |
返回Future 对象 |
新增 | ||
Handle |
所有的任务进入待执行队列(Eventloop.call_soon )之前都会封装成Handle对象 |
|||
__init__ |
初始化两个重要对象 self._callback 与 self._args
|
|||
self._callback |
待执行函数主体 | |||
self._args |
待执行函数参数 | |||
_run |
待执行函数执行 | |||
TimerHandle |
带时间戳的对象 | 新增 | ||
__init__ |
初始化重要对象 self._when
|
新增 | ||
self._when |
调度的时间 | 新增 | ||
__lt__``__gt__``__eq__ |
一系列魔术方法,实现对象比较 | 新增 | ||
get_event_loop |
获取当前线程的事件循环 | |||
fake_socket |
创建一对套接字对象,并且将一种一条套接字注册到多路复用对象sel ,返回sel
|
新增 | ||
_complete_eventloop |
将事件循环的_stopping 标志置位True |
|||
run |
入口函数 | |||
gather |
可以同时执行多个任务的入口函数 | |||
_GatheringFuture |
将每一个任务组成列表,封装成一个新的类 | |||
sleep |
入口函数 | 新增 |
tasks.py
类/函数 | 方法 | 对象 | 作用 | 描述 |
---|---|---|---|---|
Task | 继承自Future,主要用于整个协程运行的周期 | |||
__init__ |
初始化对象 self._coro ,并且call_soon 将self.__step 加入self._ready 队列 |
|||
self._coro |
用户定义的函数主体 | |||
__step |
Task类的核心函数 | |||
__wakeup |
唤醒任务 | |||
ensure_future |
如果对象是一个Future对象,就返回,否则就会调用create_task 返回,并且加入到_ready 队列 |
futures.py
类/函数 | 方法 | 对象 | 作用 | 描述 |
---|---|---|---|---|
Future | 主要负责与用户函数进行交互 | |||
__init__ |
初始化两个重要对象 self._loop 与 self._callbacks
|
|||
self._loop |
事件循环 | |||
self._callbacks |
回调队列,任务暂存队列,等待时机成熟(状态不是PENDING ),就会进入_ready 队列 |
|||
add_done_callback |
添加任务回调函数,状态_PENDING ,就虎进入_callbacks 队列,否则进入_ready 队列 |
|||
set_result |
获取任务执行结果并存储至_result ,将状态置位_FINISH ,调用__schedule_callbacks
|
|||
__schedule_callbacks |
将回调函数放入_ready ,等待执行 |
|||
result |
获取返回值 | |||
__await__ |
使用await就会进入这个方法 | |||
__iter__ |
使用yield from就会进入这个方法 | |||
set_result_unless_cancelled |
其实就是Future.set_result ,只不过调用场景与调用方式不一样 |
新增 |
新加了很多的函数,后面我们边走流程,边讲解他们的用途
3)执行过程
3.1)入口函数
main.py
if __name__ == "__main__":
ret = wilsonasyncio.run(helloworld())
print(ret)
3.2)事件循环启动,同gather
,不再赘述
3.3)第一次循环run_forever
--> run_once
,同gather
,不再赘述
3.3.1)gather
完成,回到helloworld()
,同gather
,不再赘述
3.4)<font color=red size=4>第二次循环run_forever
--> run_once
,从这里开始,不一样的地方来了</font>
- 从队列中取出数据,此时
_ready
队列有两个任务,hello()``world()
,在gather
的for循环时添加的
def run_once(self):
timeout = 0
if not self._ready and self._scheduled:
heapq.heapify(self._scheduled)
when = self._scheduled[0]._when
timeout = min(max(0, when - self.current_time()), 60)
self._selector.select(timeout)
end_time = self.current_time()
while self._scheduled:
handle = self._scheduled[0]
if handle._when >= end_time:
break
handle = heapq.heappop(self._scheduled)
self._ready.append(handle)
ntodo = len(self._ready)
for _ in range(ntodo):
handle = self._ready.popleft()
handle._run()
-
run_once
做了改动,由于if not self._ready and self._scheduled
不满足(self._ready
有内容) -
timeout=0
,self._selector.select(timeout)
不会等待,直接跳过 -
while self._scheduled
不满足,不会进入while循环 - 直接进入
handle._run()
阶段
<font color=red size=4>上述的函数都没有执行,所以没有分析,在后面执行的时候会详细分析作用</font>
tasks.py
def __step(self, exc=None):
coro = self._coro
try:
if exc is None:
result = coro.send(None)
else:
result = coro.throw(exc)
except StopIteration as exc:
super().set_result(exc.value)
else:
blocking = getattr(result, '_asyncio_future_blocking', None)
if blocking:
result._asyncio_future_blocking = False
result.add_done_callback(self.__wakeup, result)
finally:
self = None
- 经过
coro.send(None)
回到各自的函数本体中
async def hello():
print('enter hello ...')
await wilsonasyncio.sleep(5)
return 'return hello...'
async def world():
print('enter world ...')
await wilsonasyncio.sleep(3)
return 'return world...'
- 由于
hello()``world()
使用await
调用了sleep
,我们开看看sleep
的源码:
async def sleep(delay, result=None, *, loop=None):
if loop is None:
loop = get_event_loop()
future = loop.create_future()
loop.call_later(delay, set_result_unless_cancelled, future, result)
return await future
-
future = loop.create_future()
创建了一个Future
对象,随后调用call_later
def call_later(self, delay, callback, *args):
timer = self.call_at(self.current_time() + delay, callback, *args)
return timer
def call_at(self, when, callback, *args):
timer = TimerHandle(when, callback, *args)
heapq.heappush(self._scheduled, timer)
return timer
-
call_later``call_at
,主要的逻辑:
a)self.current_time()
获取当前的时间线,再加上传入的delay
(hello(5)``world(3)
)计算出延时
b)timer = TimerHandle(when, callback, *args)
将延时、callback(即set_result_unless_cancelled
)再加上参数封装成TimerHandle
对象
c)heapq.heappush(self._scheduled, timer)
将对象推入self._scheduled
队列等待合适的时间调度到self._ready
d) 第一点需要注意的是,self._scheduled
是一个最小堆
e) 第二点需要注意的是,TimerHandle
实现了__lt__``__gt__``__eq__
,所以会通过self._when
进行对象比较。在本例中,hello()
的延时是current_time
+5,world()
的延时是current_time
+3,所以world()
会优先调度 -
返回到
sleep
async def sleep(delay, result=None, *, loop=None):
if loop is None:
loop = get_event_loop()
future = loop.create_future()
loop.call_later(delay, set_result_unless_cancelled, future, result)
return await future
-
return await future
在gather
小节中描述过,一旦调用了await,就会来到__await__
def __await__(self):
if self._state == _PENDING:
self._asyncio_future_blocking = True
yield self
return self.result()
- <font color=red size=4>
yield self
self就是之前创建的future
对象,并且返回到当初send
的地方,对的,就是Task.__step
里面</font>
def __step(self, exc=None):
coro = self._coro
try:
if exc is None:
result = coro.send(None)
else:
result = coro.throw(exc)
except StopIteration as exc:
super().set_result(exc.value)
else:
blocking = getattr(result, '_asyncio_future_blocking', None)
if blocking:
result._asyncio_future_blocking = False
result.add_done_callback(self.__wakeup, result)
finally:
self = None
-
由于他们去
__await__
溜达了一圈,所以_asyncio_future_blocking=True
,所以hello()``world()
的回调函数是self.__wakeup
-
<font color=red size=4>这一次循环结束了,
hello()``world()
并没有结束,而是分分挂起,等待他们的子任务await wilsonasyncio.sleep
执行结束完之后才会唤醒</font>
3.5)第三次循环run_forever
--> run_once
eventloops:
def run_once(self):
timeout = 0
if not self._ready and self._scheduled:
heapq.heapify(self._scheduled)
when = self._scheduled[0]._when
timeout = min(max(0, when - self.current_time()), 60)
self._selector.select(timeout)
end_time = self.current_time()
while self._scheduled:
handle = self._scheduled[0]
if handle._when >= end_time:
break
handle = heapq.heappop(self._scheduled)
self._ready.append(handle)
ntodo = len(self._ready)
for _ in range(ntodo):
handle = self._ready.popleft()
handle._run()
-
run_once
函数进行了改造 -
self._ready
为空,并且self._scheduled
队列不为空,里面的内容是hello()``world()
2个函数中调用await wilsonasyncio.sleep()
时添加进self._scheduled
,分别是两个TimerHandle
对象,并且是按照最小堆排序 - 由于是按照最小堆排序,直接取出来肯定就是时间最小的,
timeout = min(max(0, when - self.current_time()), 60)
,获取timeout
-
self._selector.select(timeout)
核心代码,主要的逻辑就会等待timeout
的时间,是整个线程等待,所有的协程任务都会卡主,但是当前队列里面所有的任务都是等待,并且当前的timeout
是等待时间最小的。在本例中,timeout
就是world()
的等待时间:3s
<font color=red size=4>self._selector.select(timeout)
这里要详细描述一下了</font>
- 核心逻辑就是将一堆等待中的协程任务中,选一个等待时间最小的出来,然后线程sleep
- 这时的sleep可以用粗暴的
time.sleep
,但是我们选择了self._selector.select(timeout)
,主要_selector
在后面的应用中发挥了巨大的作用,只是在这里看不出来 -
self._selector
是一个多路复用对象,就是我们熟悉的select()``poll()``epoll()
,我们将一条socket注册到对象上面,然后调用self._selector.select(timeout)
,整个线程就会在timeout
期间,监听socket是否有新的数据,一旦socket有新的数据,就会立即执行回调函数 -
sel.register(_ssock.fileno(), selectors.EVENT_READ, None)
当然,我们的回调函数是None
,没有注册任何回调,在本例中,我们只需要它的等待功能 - 肯定会有人问了,那如果这时候出现了一个需要立即执行的任务,怎么办?这是后面解答的问题,在后面的轮子中肯定会解答这个问题,而在本例中,不可能会出现这种情况(如果是立即执行的任务,那么一定出现在
self._ready
队列里面,在第一次循环就已经执行完成了)
简单解释了一下,我们喝一口水,继续。。。
-
end_time = self.current_time()
记录一下当前的时间
while self._scheduled:
handle = self._scheduled[0]
if handle._when >= end_time:
break
handle = heapq.heappop(self._scheduled)
self._ready.append(handle)
- 将
_when
大于当前时间的协程任务取出来,放入self._ready
队列,等待下一次循环调度 -
_when
小于当前事件的协程任务留在self._scheduled
,他们还需要继续睡觉 - 全部搞完,继续往下走
ntodo = len(self._ready)
for _ in range(ntodo):
handle = self._ready.popleft()
handle._run()
- 熟悉的流程,
handle
是world()
的sleep
,执行其回调函数set_result_unless_cancelled
,就是set_result
- 将
__wakeup
加入到self._ready
队列,world()
等待被唤醒
3.6)第四次循环run_forever
--> run_once
if not self._ready and self._scheduled:
-
timeout=0
,self._ready
有协程任务,不会等待 -
__wakeup
-->__step
-->send(None)
--> 回到world()
async def world():
print('enter world ...')
await wilsonasyncio.sleep(3)
print('world sleep end...')
return 'return world...'
-
world()
协程完成,将回调_done_callback
加入到self._ready
队列,同gather
3.7)第五次循环run_forever
--> run_once
-
gather._done_callback()
,由于条件不满足,不会添加回调到self._ready
,同gather
3.8)第六次循环run_forever
--> run_once
if not self._ready and self._scheduled:
heapq.heapify(self._scheduled)
when = self._scheduled[0]._when
timeout = min(max(0, when - self.current_time()), 60)
-
timeout = min(max(0, when - self.current_time()), 60)
,由于刚才等了3s,hello()
的delay
是5s,所以这里timeout=2
(这之间还有很多代码的执行时间,不过执行速度非常快) -
self._selector.select(timeout)
等待2s之后,重复了world()
的流程 -
handle
是hello()
的sleep
,执行其回调函数set_result_unless_cancelled
,就是set_result
-
将
__wakeup
加入到self._ready
队列,hello()
等待被唤醒 -
results.append(res)
将子任务的结果取出来,放进父任务的results里面 -
子任务执行完成,终于到了唤醒父任务的时候了
task.__wakeup
def __wakeup(self, future):
try:
future.result()
except Exception as exc:
raise exc
else:
self.__step()
self = None
3.8)第七次循环run_forever
--> run_once
-
__wakeup
-->__step
-->send(None)
--> 回到hello()
async def hello():
print('enter hello ...')
await wilsonasyncio.sleep(5)
print('hello sleep end...')
return 'return hello...'
-
hello()
协程完成,将回调_done_callback
加入到self._ready
队列,同gather
3.9)第八次循环run_forever
--> run_once
-
gather._done_callback()
,由于条件满足,将回调到self._ready
,helloworld
等待被唤醒。同上一小节gather
3.10)第九次循环run_forever
--> run_once
- 循环结束
- 回到
run
3.11)回到主函数,获取返回值
if __name__ == "__main__":
ret = wilsonasyncio.run(helloworld())
print(ret)
3.12)执行结果
▶ python3 main.py
enter helloworld
enter hello ...
enter world ...
world sleep end...
hello sleep end...
exit helloworld
['return hello...', 'return world...']
sleep.jpg五、流程总结
六、小结
● 无法总结。。。。有问题私信、留言
● 本文中的代码,参考了python 3.7.7中asyncio的源代码,裁剪而来
● 本文中代码:代码
至此,本文结束
在下才疏学浅,有撒汤漏水的,请各位不吝赐教...