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

前言

書接上文：，本文造第四個輪子，也是asyncio包里面非常常用，并且非常復雜的一個函式sleep

一、知識準備

● time.sleep直接讓當前執行緒睡覺，但是這種方式顯然是不能接受的，如果當前執行緒睡覺，那我們所有的協程任務都會被卡主，并發也就無從談起了
● 理解socket.socketpair()創建的套接字物件
● 理解selectors的應用
● 理解最小堆以及heapq的應用
● 理解物件比較
● 這一小結的基礎知識很多，希望大家優先了解上述的知識再開始閱讀，否則很容易不知所云

二、環境準備

三、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

2）代碼概覽：

eventloops.py

tasks.py

futures.py

新加了很多的函式，后面我們邊走流程，邊講解他們的用途

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

本文來自博客園，作者：wilson排球，轉載請注明原文鏈接：https://www.cnblogs.com/MrVolleyball/p/15988557.html

標籤：Python

上一篇：25~30K的學員面試考題，10分鐘就寫完？

下一篇：Spring的介面集合注入功能