Docker容器蓉城devops

从头造轮子:python3 asyncio 之 sleep (4

2022-03-10  本文已影响0人  wilsonchai

从头造轮子: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._readyself._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._callbackself._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_soonself.__step加入self._ready队列
self._coro 用户定义的函数主体
__step Task类的核心函数
__wakeup 唤醒任务
ensure_future 如果对象是一个Future对象,就返回,否则就会调用create_task返回,并且加入到_ready队列

futures.py

类/函数 方法 对象 作用 描述
Future 主要负责与用户函数进行交互
__init__ 初始化两个重要对象 self._loopself._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>

    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()

<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
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...'
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
    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
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
    def __await__(self):
        if self._state == _PENDING:
            self._asyncio_future_blocking = True
            yield self
        return self.result()
    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

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()

<font color=red size=4>self._selector.select(timeout) 这里要详细描述一下了</font>


简单解释了一下,我们喝一口水,继续。。。

        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()

3.6)第四次循环run_forever --> run_once

if not self._ready and self._scheduled:
async def world():
    print('enter world ...')
    await wilsonasyncio.sleep(3)
    print('world sleep end...')
    return 'return world...'

3.7)第五次循环run_forever --> run_once

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)
    def __wakeup(self, future):
        try:
            future.result()
        except Exception as exc:
            raise exc
        else:
            self.__step()
        self = None

3.8)第七次循环run_forever --> run_once

async def hello():
    print('enter hello ...')
    await wilsonasyncio.sleep(5)
    print('hello sleep end...')
    return 'return hello...'

3.9)第八次循环run_forever --> run_once

3.10)第九次循环run_forever --> run_once

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的源代码,裁剪而来
● 本文中代码:代码


至此,本文结束
在下才疏学浅,有撒汤漏水的,请各位不吝赐教...

上一篇下一篇

猜你喜欢

热点阅读