從頭造輪子：python3 asyncio 之 sleep （4）

2022 年 3 月 10 日
筆記
asyncio, Python

前言

書接上文：，本文造第四個輪子，也是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）第二次循環run_forever –> run_once，從這裡開始，不一樣的地方來了

從隊列中取出數據，此時_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()階段

上述的函數都沒有執行，所以沒有分析，在後面執行的時候會詳細分析作用

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

yield self self就是之前創建的future對象，並且返回到當初send的地方，對的，就是Task.__step裏面

    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
這一次循環結束了，hello() world()並沒有結束，而是分分掛起，等待他們的子任務await wilsonasyncio.sleep執行結束完之後才會喚醒

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

self._selector.select(timeout) 這裡要詳細描述一下了

核心邏輯就是將一堆等待中的協程任務中，選一個等待時間最小的出來，然後線程sleep
這時的sleep可以用粗暴的time.sleep，但是我們選擇了self._selector.select(timeout)，主要_selector在後面的應用中發揮了巨大的作用，只是在這裡看不出來
self._selector是一個多路復用對象，就是我們熟悉的select() poll() epoll()，我們將一條socket註冊到對象上面，然後調用self._selector.select(timeout)，整個線程就會在timeout期間，監聽socket是否有新的數據，一旦socket有新的數據，就會立即執行回調函數
那socket對象是怎麼來的，event_loop初始化的時候，會通過socket.socketpair()創建一對socket，並且把它註冊進self._selector，所以self._selector.select(timeout)會監聽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...']

五、流程總結

六、小結

● 無法總結。。。。有問題私信、留言
● 本文中的代碼，參考了python 3.7.7中asyncio的源代碼，裁剪而來
● 本文中代碼：代碼

至此，本文結束
在下才疏學淺，有撒湯漏水的，請各位不吝賜教…
更多文章，請關注我：wilson.chai

Tags: asyncio Python