Java協程框架--Kilim工作原理

Kilim要解決的問題

 

Kilim協程框架中最核心需要解決的問題：

 

1. 如何暫停處理當前任務，轉而處理其他任務？
2. 如何恢復任務繼續執行？

 

也即如何實現協程本身的 yield / resume的語義特性。

Kilim的解決方案

概括的講，​Kilim框架在實現這個語義特性時，干了以下幾個事情：

 

• 利用字節碼技術(基於ASM字節碼框架)，將普通代碼轉化為支持協程的代碼；
• 調用Pauseable方法的時候，如果暫停了就保存當前方法棧的State，暫停執行當前Task，將控制權交給Scheduler調度器；
• Scheduler調度器負責協調其他就緒的Task；
• 之前暫停的Task恢復的時候，自動恢復State，恢復到上次執行的位置繼續執行；

 

其中，第一點是在編譯期實現，后面三點是在運行期執行。

再稍微詳細一點就是：Kilim通過編譯期字節碼編織，對每一個可暫停（Pauseable）的方法進行字節碼處理，在方法執行前和執行后加上相關的執行上下文的處理，暫停時會保存整個線程堆棧，然后通過特定的字節碼跳轉指令goto跳轉到另外一個Task的執行方法中，恢復時將復原整個線程堆棧，回到上次暫停時的位置繼續往下執行。

Kilim的工作原理

第一個Kilim最神奇的地方在於字節碼增加，那么它是怎樣將普通的Java代碼改寫層支持協程的代碼呢？首先上Kilim官方文檔中的一張圖：

這張圖也即Kilim實現協程語義的精髓所在，我們來一一分析。

左邊是普通的Java函數代碼，與我們常見的函數唯一有所不同的是函數a和b均顯示聲明拋出Pausable異常，而實際上這個異常在運行期間不會拋出，它的實際作用類似於注解，使得Kilim能夠識別哪些代碼需要Weaver工具進行代碼增強。函數拋出Pausable異常即表明該函數是可暫停的，

右邊的代碼即通過字節碼增強后的代碼，與左邊原始的代碼相比，首先函數聲明中額外增加了一個Fiber參數，Fiber可以理解為當前纖程、協程的上下文。Fiber中存儲着協程暫停和恢復時需要用到的函數堆棧、程序計數器以及當前函數的執行狀態。字節碼增強后的代碼以調用Pausable方法a為分界，將整個函數分成幾個代碼塊，也即官網文檔中提及的prelude、pre_call、post_call三部分。

1.在prelude塊中，也即剛進入函數a時將會執行的代碼塊，將根據Fiber中的pc程序計數器跳轉到對應的代碼塊處開始執行。

2.在pre_call塊中，也即在調用函數b之前，將調用Fiber的down方法記錄當前執行狀態和pc程序計數器，標識着函數將進入下一個Pausable方法。

3.在post_call塊中，也即在調用函數b之后，將調用Fiber的up方法計算函數b調用完成后返回的狀態，標識着從一個被調的Pausable方法返回，它既可能是正常的函數b執行完成返回，也可能是函數b執行暫停返回，接着通過這個狀態控制后續的執行流程。

這四種狀態分別為：

 

• NOT_PAUSING__NO_STATE，即被調函數執行完成正常返回，這種情況與即普通的函數執行類似。
• NOT_PAUSING__HAS_STATE，即被調函數執行完成，但還存在上次暫存的棧幀，這種情況一般是函數從上次暫停處恢復執行，且順利執行完成返回，此時需要恢復函數的棧幀，然后goto到RESUME代碼塊繼續執行。
• PAUSING__NO_STATE，即被調函數執行過程中暫停，且還未保存函數棧幀，需要主調函數執行暫存操作，這種情況一般即第一次協程執行到需要暫停處，此時需要采用字節碼暫存函數的棧幀和狀態，然后直接return。
• PAUSING__HAS_STATE，即被調函數執行過程中暫停，且已經保存函數棧幀，這種情況是該Pausable從上次暫停處恢復執行，但是依然沒有預期的結果，需要再次暫停，此時因為之前暫停時函數棧幀和狀態都已經保存過，不需要再做什么，因此直接return即可。

 

 

ok，到目前為止是不是謎底已經大概清晰，不過要把協程與線程的執行過程整個串聯起來，形成一個整體還稍顯迷惑，接下來詳細說明。

上面已經提到協程執行過程中核心的兩個點：一點是調用Pauseable方法的時候，如果暫停了就保存當前方法棧的State，暫停執行當前Task，將控制權交給Scheduler調度器，另外一點是暫停的Task恢復的時候，自動恢復State，恢復到上次執行的位置繼續執行。這兩點的具體過程如下：

上篇博文Java協程框架--Kilim源碼分析中已經講到在Kilim中有幾個核心元素，包括Task、Scheduler、WorkerThread以及Mailbox。其中WorkerThread即實際執行Task任務的工作者線程，Task即具體的可暫停的業務，Task與Task之間通過定制的Mailbox來通信。Kilim將線程run方法體中所有嵌套層級調用的所有Pausable方法組織成一個具有父子關系的調用鏈，形如run->A->B->C，通過Task私有的Fiber來記錄執行到哪一個層級。

通常Kilim中大部分使用Mailbox提供的get、getb、getnb三個不同版本來接收消息，其中最常用的get會阻塞當前Task而不阻塞當前線程。

那么如何實現Task的暫停呢？

 

例如一個運行狀態Task的調用鏈run->A->B->C，其中A、B、C均為Pauseable方法，在函數C中調用了Mailbox的get方法且設置了超時時長，當整個鏈嵌套執行到C的get方法這一行時，因為get本身也是一個Pausable方法，如果沒有接收到消息，將會把Task作為該Mailbox的觀察者，並調用Task.pause(this)方法暫停自身，然后該get方法即直接返回，get調用返回后，C根據Fiber的up計算發現是暫停返回，則也暫停本函數，暫存棧幀和狀態，直接返回，如此逆向直到run方法，從而實現Task的暫停。

那么又如何實現Task的恢復呢？

Task暫停的過程中有一步很關鍵，將該Task作為該Mailbox的觀察者，在當有其他線程把消息通過調用Mailbox的put方法添加到Mailbox中時，或者超時定時器觸發時，將會回調該Mailbox的觀察者，告訴觀察者有新的消息到來。這樣Task的onEvent被回調，onEvent直接調用resume方法，而resume方法實際最重要的一步即調用Scheduler的schedule方法，將該Task加入到Scheduler的可運行任務隊列中，並隨機選擇一個等待運行的工作者線程，並notify該線程，該線程被喚醒后將執行該Task，重復之前的函數調用鏈run->A->B->C執行，由於A、B、C三個函數中均已經保存了之前暫停的函數棧幀和狀態，因此之前已經執行過的代碼塊將不會重復執行，會根據Fiber中狀態選擇性的執行對應的代碼塊。因為Mailbox中已經有消息，因此再重復執行到get方法時能夠直接獲取到消息，正常的往下繼續執行。這樣相當於又走了一次調用鏈，但是並非重復執行已經執行過的代碼，而是恢復執行之前未執行的代碼，從而實現Task的恢復。

 

public void onEvent(EventPublisher ep, Event e) {
    resume();
}