從頭造輪子：python3 asyncio之 gather （3）

• 2022 年 1 月 17 日
• 筆記
• asyncio, Python

前言

書接上文：，本文造第三個輪子，也是asyncio包裏面非常常用的一個函數gather

一、知識準備

● 相對於前兩個函數，gather的使用頻率更高，因為它支持多個協程任務「同時」執行
● 理解__await__ __iter__的使用
● 理解關鍵字async/await，async/await是3.5之後的語法，和yield/yield from異曲同工
● 今天的文章有點長，請大家耐心看完

二、環境準備

三、run的實現

先來看下官方gather的使用方法：

|># more main.py
import asyncio

async def hello():
    print('enter hello ...')
    return 'return hello ...'

async def world():
    print('enter world ...')
    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)
    
|># python3 main.py
enter helloworld
enter hello ...
enter world ...
exit helloworld
['return hello ...', 'return world ...']

來看下造的輪子的使用方式：

▶ more main.py
import wilsonasyncio

async def hello():
    print('enter hello ...')
    return 'return hello ...'

async def world():
    print('enter world ...')
    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)

    
▶ python3 main.py
enter helloworld
enter hello ...
enter world ...
exit helloworld
['return hello ...', 'return world ...']

自己造的輪子也很好的運行了，下面我們來看下輪子的代碼

四、代碼解析

輪子代碼

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)

• ret = wilsonasyncio.run(helloworld())使用run，參數是用戶函數helloworld()，進入run，run的流程可以參考上一小節
• run –> run_until_complete

3.2）事件循環啟動，同之前

3.3）第一次循環run_forever –> run_once

• 將_ready隊列的內容（即：task.__step）取出來執行，這裡的coro是helloworld()

    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

• __step較之前的代碼有改動
• result = coro.send(None)，進入用戶定義函數

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret

• ret = await wilsonasyncio.gather(hello(), world())，這裡沒啥可說的，進入gather函數

def gather(*coros_or_futures, loop=None):
    loop = get_event_loop()

    def _done_callback(fut):
        nonlocal nfinished
        nfinished += 1

        if nfinished == nfuts:
            results = []
            for fut in children:
                res = fut.result()
                results.append(res)

            outer.set_result(results)

    children = []
    nfuts = 0
    nfinished = 0

    for arg in coros_or_futures:
        fut = tasks.ensure_future(arg, loop=loop)
        nfuts += 1
        fut.add_done_callback(_done_callback)

        children.append(fut)

    outer = _GatheringFuture(children, loop=loop)
    return outer

• loop = get_event_loop()獲取事件循環
• def _done_callback(fut)這個函數是回調函數，細節後面分析，現在只需要知道任務（hello()與world()）執行完之後就會回調就行
• for arg in coros_or_futures for循環確保每一個任務都是Future對象，並且add_done_callback將回調函數設置為_done_callback ，還有將他們加入到_ready隊列等待下一次循環調度
• 3個重要的變量：
         children 裏面存放的是每一個異步任務，在本例是hello()與world()
       nfuts 存放是異步任務的數量，在本例是2
         nfinished 存放的是異步任務完成的數量，目前是0，完成的時候是2
• 繼續往下，來到了_GatheringFuture ，看看源碼：

class _GatheringFuture(Future):

    def __init__(self, children, *, loop=None):
        super().__init__(loop=loop)
        self._children = children

• _GatheringFuture 最主要的作用就是將多個異步任務放入self._children，然後用_GatheringFuture 這個對象來管理。需要注意，這個對象繼承了Future
• 至此，gather完成初始化，返回了outer，其實就是_GatheringFuture
• 總結一下gather，初始化了3個重要的變量，後面用來存放狀態；給每一個異步任務添加回調函數；將多個異步子任務合併，並且使用一個Future對象去管理

3.3.1）gather完成，回到helloworld()

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret

• ret = await wilsonasyncio.gather(hello(), world()) gather返回_GatheringFuture ，隨後使用await，就會進入Future.__await__

    def __await__(self):
        if self._state == _PENDING:
            self._asyncio_future_blocking = True
            yield self
        return self.result()

• 由於_GatheringFuture 的狀態是_PENDING，所以進入if，遇到yield self，將self，也就是_GatheringFuture 返回（這裡注意yield的用法，流程控制的功能）
• 那yield回到哪兒去了呢？從哪兒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

• 這裡是本函數的第一個核心點，流程控制/跳轉，需要非常的清晰，如果搞不清楚的同學，再詳細的去閱讀有關yield/yield from的文章
• 繼續往下走，由於用戶函數helloworld()沒有結束，所以不會拋異常，所以來到了else分支
• blocking = getattr(result, '_asyncio_future_blocking', None)這裡有一個重要的狀態，那就是_asyncio_future_blocking ，只有調用__await__，才會有這個參數，默認是true，這個參數主要的作用：一個異步函數，如果調用了多個子異步函數，那證明該異步函數沒有結束（後面詳細講解），就需要添加「喚醒」回調
• result._asyncio_future_blocking = False將參數置位False，並且添加self.__wakeup回調等待喚醒
• __step函數完成

這裡需要詳細講解一下_asyncio_future_blocking 的作用

• 如果在異步函數裏面出現了await，調用其他異步函數的情況，就會走到Future.__await__將_asyncio_future_blocking 設置為true

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret
    
class Future:
    def __await__(self):
        if self._state == _PENDING:
            self._asyncio_future_blocking = True
            yield self
        return self.result()

• 這樣做了之後，在task.__step中就會把該任務的回調函數設置為__wakeup
• 為啥要__wakeup，因為helloworld()並沒有執行完成，所以需要再次__wakeup來喚醒helloworld()

這裡揭示了，在Eventloop裏面，只要使用await調用其他異步任務，就會掛起父任務，轉而去執行子任務，直至子任務完成之後，回到父任務繼續執行

先喝口水，休息一下，下面更複雜。。。

3.4）第二次循環run_forever –> run_once

eventloops.py

    def run_once(self):
        ntodo = len(self._ready)
        for _ in range(ntodo):
            handle = self._ready.popleft()
            handle._run()

• 從隊列中取出數據，此時_ready隊列有兩個任務，hello() world()，在gather的for循環時添加的

async def hello():
    print('enter hello ...')
    return 'return hello ...'

async def world():
    print('enter world ...')
    return 'return world ...'

• 由於hello() world()沒有await調用其他異步任務，所以他們的執行比較簡單，分別一次task.__step就結束了，到達set_result()處
• set_result()將回調函數放入_ready隊列，等待下次循環執行

3.5）第三次循環run_forever –> run_once

• 我們來看下回調函數

    def _done_callback(fut):
        nonlocal nfinished
        nfinished += 1

        if nfinished == nfuts:
            results = []
            for fut in children:
                res = fut.result()
                results.append(res)

            outer.set_result(results)

• 沒錯，這是本文的第二個核心點，我們來仔細分析一下
• 這段代碼最主要的邏輯，其實就是，只有當所有的子任務執行完之後，才會啟動父任務的回調函數，本文中只有hello() world()都執行完之後if nfinished == nfuts: ，才會啟動父任務_GatheringFuture的回調outer.set_result(results)
• 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.6）第四次循環run_forever –> run_once

• future.result()從_GatheringFuture取出結果，然後進入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

• result = coro.send(None)其實就是helloworld() --> send又要跳回到當初yield的地方，那就是Future.__await__

    def __await__(self):
        if self._state == _PENDING:
            self._asyncio_future_blocking = True
            yield self
        return self.result()

• return self.result()終於返回到helloworld()函數裏面去了

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret

• helloworld終於也執行完了，返回了ret

3.7）第五次循環run_forever –> run_once

• 循環結束
• 回到run

3.8）回到主函數，獲取返回值

if __name__ == "__main__":
    ret = wilsonasyncio.run(helloworld())
    print(ret)

3.9）執行結果

▶ python3 main.py
enter helloworld
enter hello ...
enter world ...
exit helloworld
['return hello ...', 'return world ...']

五、流程總結

六、小結

● 終於結束了，這是一個非常長的小節了，但是我感覺很多細節還是沒有說到，大家有問題請及時留言探討
● _GatheringFuture一個非常重要的對象，它不但追蹤了hello() world()的執行狀態，喚醒helloworld()，並且將返回值傳遞給helloworld
● await async yield對流程的控制需要特別關注
● 本文中的代碼，參考了python 3.7.7中asyncio的源代碼，裁剪而來
● 本文中代碼：代碼

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