從頭造輪子：python3 asyncio之 run_until_complete （1）

前言

今天開始聊一聊python3的asyncio。關於asyncio，大家肯定都有自己的理解，並且網上大神也把基礎概念也解釋的比較透徹。
本文寫作的初衷，主要是理解asyncio的原理並且實現一遍。
話不多說，我們開始！

一、知識准備

● 理解進程、線程、協程。簡單來說，這三個都是為了解決多任務同時進行的問題
  1）進程是操作資源分配的最小單位，多任務的實現主要是極快地在進程間來回切換，而進程切換消耗時間最長（系統調用）
  2）線程依賴於進程，多個線程共享了父進程的一部分資源，線程切換時間相對於進程來說消耗時間大大減少，但是由於python gil的存在，並不存在多線程（系統調用）
  3）協程依賴於線程，由於進程/線程切換都是系統調用，開銷是巨大的。而協程是在用戶空間內完成任務切換，不會切換到操作系統資源（寄存器、信號量、堆棧等），所以這種方式開銷最小。python的協程核心在於，遇到等待事件，就交出cpu控制權，轉而讓其他協程執行


● 理解python生成器，yield/yield from
  這里就不班門弄斧了，直接推薦大佬的blog


● 理解關鍵字async/await，async/await是3.5之后的語法，和yield/yield from異曲同工

二、環境准備

組件	版本
python	3.7.7

三、run_until_complete的實現

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

|># more main.py
import asyncio
async def hello():
    print('enter hello ...')
    return 'world'

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
    
|># python3 main.py
enter hello ...
world

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

▶ more main.py
from wilsonasyncio import get_event_loop
async def hello():
    print('enter hello ...')
    return 'world'


if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
    
▶ python3 main.py
enter hello ...
world

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

四、代碼解析

輪子代碼

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._stopping	事件循環完成的標志
	call_soon		調用該方法會立即將任務添加到待執行隊列
	run_once		被run_forever調用，從self._ready隊列里面取出任務執行
	run_forever		死循環，若self._stopping則退出循環
	run_until_complete		非常重要的函數，任務的起點和終點（后面詳細介紹）
	create_task		將傳入的函數封裝成task對象，這個操作會將task.__step添加到__ready隊列
Handle			所有的任務進入待執行隊列（Eventloop.call_soon）之前都會封裝成Handle對象
	__init__		初始化兩個重要對象 self._callback 與 self._args
		self._callback	待執行函數主體
		self._args	待執行函數參數
	_run		待執行函數執行
get_event_loop			獲取當前線程的事件循環
_complete_eventloop			將事件循環的_stopping標志置位True

tasks.py

類/函數	方法	對象	作用	描述
Task			繼承自Future，主要用於整個協程運行的周期
	__init__		初始化對象 self._coro ，並且call_soon將self.__step加入self._ready隊列
		self._coro	用戶定義的函數主體
	__step		Task類的核心函數

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		獲取返回值

3）執行過程

3.1）入口函數

main.py

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

if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)

• loop = get_event_loop()獲取事件循環
• task = loop.create_task(hello())將用戶函數hello()封裝成協程，我們看下create_task的源碼

    def create_task(self, coro):
        task = tasks.Task(coro, loop=self)
        return task

    初始化一個Task對象，從代碼概覽得知，初始化對象之后會立即將__step添加到_ready隊列等待執行

• rst = loop.run_until_complete(task)開始執行事件循環的第一個函數run_until_complete

3.2）事件循環啟動

eventloops.py

    def run_until_complete(self, future):
        future.add_done_callback(_complete_eventloop, future)
        self.run_forever()
        return future.result()

• future.add_done_callback(_complete_eventloop, future)為future添加回調函數（future就是task對象，而task對象里的任務就是hello() ），回調函數是_complete_eventloop 。就是future執行完成之后執行_complete_eventloop
• self.run_forever()啟動事件循環

3.3）第一次循環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隊列的內容（task.__step）取出來執行

tasks.py

    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                coro.send(None)
            else:
                coro.throw(exc)
        except StopIteration as exc:
            super().set_result(exc.value)
        finally:
            self = None

• coro.send(None)核心代碼，跳轉回到用戶函數hello()

main.py

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

• 用戶函數非常簡單，打印一行數據，以及返回一個字符串world，執行完成之后回到task.__step()
• super().set_result(exc.value)由於用戶函數執行完成，會拋出異常StopIteration，捕獲之后執行set_result
• 由代碼概覽得知set_result 的作用在於將任務狀態置位_FINISHED，並且將回調函數（_complete_eventloop ）寫入_ready隊列

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

• 繼續循環，handle封裝了_complete_eventloop

def _complete_eventloop(fut):
    fut._loop.stop()

• 調用stop，設置停止標志

3.5）第三次循環run_forever

    def run_forever(self):
        while True:
            self.run_once() 
            if self._stopping:
                break

• 跳出事件循環，回到run_until_complete

    def run_until_complete(self, future):
        future.add_done_callback(_complete_eventloop, future)
        self.run_forever()
        return future.result()

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

if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)

• rst = loop.run_until_complete(task)獲取返回值

3.7）執行結果

▶ python3 main.py
enter hello ...
return world ...

五、流程總結

六、小結

● task對象與future有什么區別？主要用於整個協程運行的周期，主要負責與用戶函數進行交互
● 本文從asyncio的第一個函數run_until_complete，介紹了asyncio的基本流程：用戶函數並不是立即執行，而是進入隊列，然后根據eventloop在合適的時機進行統一調度
● 本文中的代碼，參考了python 3.7.7中asyncio的源代碼，裁剪而來
● 本文中代碼：代碼

---

至此，本文結束
在下才疏學淺，有撒湯漏水的，請各位不吝賜教...