01

概述

盡管 Tokio 目前已經是 Rust 異步運行時的事實標準，但要實現極致性能的網絡中間件還有一定距離。為了這個目標，CloudWeGo Rust Team 探索基于 io-uring 為 Rust 提供異步支持，并在此基礎上研發通用網關。

本文包括以下內容：

介紹 Rust 異步 Runtime；

Monoio 的一些設計精要；

Runtime 對比選型與應用。

02

Rust 異步機制

借助 Rustc 和 llvm，Rust 可以生成足夠高效且安全的機器碼。但是一個應用程序除了計算邏輯以外往往還有 IO，特別是對于網絡中間件，IO 其實是占了相當大比例的。

程序做 IO 需要和操作系統打交道，編寫異步程序通常并不是一件簡單的事情，在 Rust 中是怎么解決這兩個問題的呢？比如，在 C++里面，可能經常會寫一些 callback ，但是我們并不想在 Rust 里面這么做，這樣的話會遇到很多生命周期相關的問題。
Rust 允許自行實現 Runtime 來調度任務和執行 syscall；并提供了 Future 等統一的接口；另外內置了 async-await 語法糖從面向 callback 編程中解放出來。

Example

這里從一個簡單的例子入手，看一看這套系統到底是怎么工作的。
當并行下載兩個文件時，在任何語言中都可以啟動兩個 Thread，分別下載一個文件，然后等待 thread 執行結束；但并不想為了 IO 等待啟動多余的線程，如果需要等待 IO，我們希望這時線程可以去干別的，等 IO 就緒了再做就好。
這種基于事件的觸發機制在 cpp 里面常常會以 callback 的形式遇見。Callback 會打斷我們的連續邏輯，導致代碼可讀性變差，另外也容易在 callback 依賴的變量的生命周期上踩坑，比如在 callback 執行前提前釋放了它會引用的變量。
但在 Rust 中只需要創建兩個 task 并等待 task 執行結束即可。

這個例子相比線程的話，異步 task 會高效很多，但編程上并沒有因此復雜多少。

第二個例子，現在 mock 一個異步函數 do_http，這里直接返回一個 1，其實里面可能是一堆異步的遠程請求；在此之上還想對這些異步函數做一些組合，這里假設是做兩次請求，然后把兩次的結果加起來，最后再加一個 1 ，就是這個例子里面的 sum 函數。通過 Async 和 Await 語法可以非常友好地把這些異步函數給嵌套起來。

#[inline(never)]
asyncfndo_http()->i32{
//dohttprequestinasyncway
1
}

pubasyncfnsum()->i32{
do_http().await+do_http().await+1
}

這個過程和寫同步函數是非常像的，也就說是在面向過程編程，而非面向狀態編程。利用這種機制可以避開寫一堆 callback 的問題，帶來了編程的非常大的便捷性。

Async Await 背后的秘密

通過這兩個例子可以得知 Rust 的異步是怎么用的，以及它寫起來確實非常方便。那么它背后到底是什么原理呢？

#[inline(never)]
asyncfndo_http()->i32{
//dohttprequestinasyncway
1
}

pubasyncfnsum()->i32{
do_http().await+do_http().await+1
}

剛才的例子使用 Async + Await 編寫，其生成結構最終實現 Future trait 。

Async + Await 其實是語法糖，可以在 HIR 階段被展開為 Generator 語法，然后 Generator 又會在 MIR 階段被編譯器展開成狀態機。

Future 抽象

Future trait 是標準庫里定義的。它的接口非常簡單，只有一個關聯類型和一個 poll 方法。

pubtraitFuture{
typeOutput;
fnpoll(self:Pin<&mut?Self>,cx:&mutContext<'_>)->Poll;
}

pubenumPoll{
Ready(T),
Pending,
}

Future 描述狀態機對外暴露的接口：

推動狀態機執行：Poll 方法顧名思義就是去推動狀態機執行，給定一個任務，就會推動這個任務做狀態轉換。

返回執行結果：

遇到了阻塞：Pending

執行完畢：Ready + 返回值

可以看出，異步 task 的本質就是實現 Future 的狀態機。程序可以利用 Poll 方法去操作它，它可能會告訴程序現在遇到阻塞，或者說任務執行完了并返回結果。

既然有了 Future trait，我們完全可以手動地去實現 Future。這樣一來，實現出來的代碼要比 Async、Await 語法糖去展開的要易讀。下面是手動生成狀態機的樣例。如果用 Async 語法寫，可能直接一個 async 函數返回一個 1 就可以；我們手動編寫需要自定義一個結構體，并為這個結構體實現 Future。

//autogenerate
asyncfndo_http()->i32{
//dohttprequestinasyncway
1
}

//manuallyimpl
fndo_http()->DOHTTPFuture{DoHTTPFuture}

structDoHTTPFuture;
implFutureforDoHTTPFuture{
typeOutput=i32;
fnpoll(self:Pin<&mut?Self>,_cx:&mutContext<'_>)->Poll{
Poll::Ready(1)
}
}

Async fn 的本質就是返回一個實現了 Future 的匿名結構，這個類型由編譯器自動生成，所以它的名字不會暴露給我們。而我們手動實現就定義一個 Struct DoHTTPFuture，并為它實現 Future，它的 Output 和 Async fn 的返回值是一樣的，都是 i32 。這兩種寫法是等價的。

由于這里只需要立刻返回一個數字 1，不涉及任何等待，那么我們只需要在 poll 實現上立刻返回 Ready(1) 即可。前面舉了 sum 的例子，它做的事情是異步邏輯的組合：調用兩次 do http，最后再把兩個結果再加一起。這時候如果要手動去實現的話，就會稍微復雜一些，因為會涉及到兩個 await 點。一旦涉及到 await，其本質上就變成一個狀態機。

為什么是狀態機呢？因為每次 await 等待都有可能會卡住，而線程此時是不能停止工作并等待在這里的，它必須切出去執行別的任務；為了下次再恢復執行前面任務，它所對應的狀態必須存儲下來。這里我們定義了 FirstDoHTTP 和 SecondDoHTTP 兩個狀態。實現 poll 的時候，就是去做一個 loop，loop 里面會 match 當前狀態，去做狀態轉換。

//autogenerate
asyncfnsum()->i32{
do_http().await+dohttp().await+1
}

//manuallyimpl
fnsum()->SumFuture{SumFuture::FirstDoHTTP(DoHTTPFuture)}

enumSumFuture{
FirstDoHTTP(DOHTTPFuture),
SecondDoHTTP(DOHTTPFuture，i32),
}

implFutureforSumFuture{
typeOutput=i32;

fnpoll(self:Pin<&mut?Self>,cx:&mutContext<'?>)->Poll{
letthis=self.getmut();
loop{
matchthis{
SumFuture::FirstDoHTTP(f)=>{
letpinned=unsafe{Pin::new_unchecked(f)};
matchpinned.poll(cx){
Poll::Ready(r)=>{
*this=SumFuture::SecondDoHTTP(DOHTTPFuture，r);
}
Poll::Pending=>{
returnPol::Pending;
}
}
}
SumFuture::SecondDoHTTP(f，prev_sum)=>{
letpinned=unsafe{Pin::new_unchecked(f)};
returnmatchpinned.poll(cx){
Poll::Ready(r)=>Poll::Ready(*prev_sum+r+1),
Poll::Pending=>Pol::Pending,
};
}
}
}
}
}

Task, Future 和 Runtime 的關系

我們這里以 TcpStream 的 Read/Write 為例梳理整個機制和組件的關系。

首先當我們創建 TCP stream 的時候，這個組件內部就會把它注冊到一個 poller 上去，這個 poller 可以簡單地認為是一個 epoll 的封裝（具體使用什么 driver 是根據平臺而異的）。

按照順序來看，現在有一個 task ，要把這個 task spawn 出去執行。那么 spawn 本質上就是把 task 放到了 runtime 的任務隊列里，然后 runtime 內部會不停地從任務隊列里面取出任務并且執行——執行就是推動狀態機動一動，即調用它的 poll 方法，之后我們就來到了第2步。

我們執行它的 poll 方法，本質上這個 poll 方法是用戶實現的，然后用戶就會在這個 task 里面調用 TcpStream 的 read/write。這兩個函數內部最終是調用 syscall 來實現功能的，但在執行 syscall 之前需要滿足條件：這個 fd 可讀/可寫。如果它不滿足這個條件，那么即便我們執行了 syscall 也只是拿到了 WOULD_BLOCK 錯誤，白白付出性能。初始狀態下我們會設定新加入的 fd 本身就是可讀/可寫的，所以第一次 poll 會執行 syscall。當沒有數據可讀，或者內核的寫 buffer 滿了的時候，這個 syscall 會返回 WOULD_BLOCK 錯誤。在感知到這個錯誤后，我們會修改 readiness 記錄，設定這個 fd 相關的讀/寫為不可讀/不可寫狀態。這時我們只能對外返回 Pending。

之后來到第四步，當我們任務隊列里面任務執行完了，我們現在所有任務都卡在 IO 上了，所有的 IO 可能都沒有就緒，此時線程就會持續地阻塞在 poller 的 wait 方法里面，可以簡單地認為它是一個 epoll_wait 一樣的東西。當基于 io_uring 實現的時候，這可能對應另一個 syscall。

此時陷入 syscall 是合理的，因為沒有任務需要執行，我們也不需要輪詢 IO 狀態，陷入 syscall 可以讓出 CPU 時間片供同機的其他任務使用。如果有任何 IO 就緒，這時候我們就會從 syscall 返回，并且 kernel 會告訴我們哪些 fd 上的哪些事件已經就緒了。比如說我們關心的是某一個 FD 它的可讀，那么這時候他就會把我們關心的 fd 和可讀這件事告訴我們。

我們需要標記 fd 對應的 readiness 為可讀狀態，并把等在它上面的任務給叫醒。前面一步我們在做 read 的時候，有一個任務是等在這里的，它依賴 IO 可讀事件，現在條件滿足了，我們需要重新調度它。叫醒的本質就是把任務再次放到 task queue 里，實現上是通過 Waker 的 wake 相關方法做到的，wake 的處理行為是 runtime 實現的，最簡單的實現就是用一個 Deque 存放任務，wake 時 push 進去，復雜一點還會考慮任務竊取和分配等機制做跨線程的調度。

當該任務被 poll 時，它內部會再次做 TcpStream read，它會發現 IO 是可讀狀態，所以會執行 read syscall，而此時 syscall 就會正確執行，TcpStream read 對外會返回 Ready。

Waker

剛才提到了 Waker，接下來介紹 waker 是如何工作的。我們知道 Future 本質是狀態機，每次推它轉一轉，它會返回 Pending 或者 Ready ，當它遇到 io 阻塞返回 Pending 時，誰來感知 io 就緒? io 就緒后怎么重新驅動 Future 運轉？

pubtraitFuture{
typeOutput;
fnpoll(self:Pin<&mut?Self>,cx:&mutContext<'_>)->Poll;
}

pubstructContext<'a>{
//可以拿到用于喚醒Task的Waker
waker:&aWaker,
//標記字段,忽略即可
_marker:PhantomData&'a()>,
}

Future trait 里面除了有包含自身狀態機的可變以借用以外，還有一個很重要的是 Context，Context 內部當前只有一個 Waker 有意義，這個 waker 我們可以暫時認為它就是一個 trait object ，由 runtime 構造和實現。它實現的效果，就是當我們去 wake 這個 waker 的時候，會把任務重新加回任務隊列，這個任務可能立刻或者稍后被執行。

舉另一個例子來梳理整個流程。

用戶使用 listener.accept() 生成 AcceptFut 并等待:

fut.await 內部使用 cx 調用 Future 的 poll 方法

poll 內部執行 syscall

當前無連接撥入，kernel 返回 WOULD_BLOCK

將 cx 中的 waker clone 并暫存于 TcpListener 關聯結構內

本次 poll 對外返回 Pending

Runtime 當前無任務可做，控制權交給 Poller

Poller 執行 epoll_wait 陷入 syscall 等待 IO 就緒

查找并標記所有就緒 IO 狀態

如果有關聯 waker 則 wake 并清除

等待 accept 的 task 將再次加入執行隊列并被 poll

再次執行 syscall

12/13. kernel 返回 syscall 結果，poll 返回 Ready

Runtime

先從 executor 看起，它有一個執行器和一個任務隊列，它的工作是不停地取出任務，推動任務運行，之后在所有任務執行完畢必須等待時，把執行權交給 Reactor。

Reactor 拿到了執行權之后，會與 kernel 打交道，等待 IO 就緒，IO就緒好了之后，我們需要標記這個 IO 的就緒狀態，并且把這個 IO 所關聯的任務給喚醒。喚醒之后，我們的執行權又會重新交回給 executor 。在 executor 執行這個任務的時候，就會調用到 IO 組件所提供的一些能力。

IO 組件要能夠提供這些異步的接口，比如說當用戶想用 tcb stream 的時候，得用 runtime 提供的一個 TcpStream， 而不是直接用標準庫的。第二，能夠將自己的 fd 注冊到 Reactor 上。第三，在 IO 沒有就緒的時候，我們能把這個 waker 放到任務相關聯的區域里。

整個 Rust 的異步機制大概就是這樣。

03

Monoio 設計

以下將會分為四個部分介紹 Monoio Runtime 的設計要點：

基于 GAT(Generic associated types) 的異步 IO 接口；

上層無感知的 Driver 探測與切換；

如何兼顧性能與功能；

提供兼容 Tokio 的接口

基于 GAT 的純異步IO接口

首先介紹一下兩種通知機制。第一種是和 epoll 類似的，基于就緒狀態的一種通知。第二種是 io-uring 的模式，它是一個基于“完成通知”的模式。

在基于就緒狀態的模式下，任務會通過 epoll 等待并感知 IO 就緒，并在就緒時再執行 syscall。但在基于“完成通知”的模式下，Monoio 可以更懶：直接告訴 kernel 當前任務想做的事情就可以放手不管了。

io_uring 允許用戶和內核共享兩個無鎖隊列，submission queue 是用戶態程序寫，內核態消費；completion queue 是內核態寫，用戶態消費。通過 enter syscall 可以將隊列中放入的 SQE 提交給 kernel，并可選地陷入并等待 CQE。

在 syscall 密集的應用中，使用 io_uring 可以大大減少上下文切換次數，并且 io_uring 本身也可以減少內核中數據拷貝。

這兩種模式的差異會很大程度上影響 Runtime 的設計和 IO 接口。在第一種模式下，等待時是不需要持有 buffer 的，只有執行 syscall 的時候才需要 buffer，所以這種模式下可以允許用戶在真正調用 poll 的時候（如 poll_read）傳入 &mut Buffer；而在第二種模式下，在提交給 kernel 后，kernel 可以在任何時候訪問 buffer，Monoio 必須確保在該任務對應的 CQE 返回前 Buffer 的有效性。

如果使用現有異步 IO trait（如 tokio/async-std 等），用戶在 read/write 時傳入 buffer 的引用，可能會導致 UAF 等內存安全問題：如果在用戶調用 read 時將 buffer 指針推入 uring SQ，那么如果用戶使用 read(&mut buffer) 創建了 Future，但立刻 Drop 它，并 Drop buffer，這種行為不違背 Rust 借用檢查，但內核還將會訪問已經釋放的內存，就可能會踩踏到用戶程序后續分配的內存塊。

所以這時候一個解法，就是去捕獲它的所有權，當生成 Future 的時候，把所有權給 Runtime，這時候用戶無論如何都訪問不到這個 buffer 了，也就保證了在 kernel 返回 CQE 前指針的有效性。這個解法借鑒了 tokio-uring 的做法。

Monoio 定義了 AsyncReadRent 這個 trait。所謂的 Rent ，即租借，相當于是 Runtime 先把這個 buffer 從用戶手里拿過來，待會再還給用戶。這里的 type read future 是帶了生命周期泛型的，這個泛型其實是 GAT 提供了一個能力，現在 GAT 已經穩定了，已經可以在 stable 版本里面去使用它了。當要實現關聯的 Future 的時候，借助 TAIT 這個 trait 可以直接利用 async-await 形式來寫，相比手動定義 Future 要方便友好很多，這個 feature 目前還沒穩定（現在改名叫 impl trait in assoc type 了）。

當然，轉移所有權會引入新的問題。在基于就緒狀態的模式下，取消 IO 只需要 Drop Future 即可；這里如果 Drop Future 就可能導致連接上數據流錯誤（Drop Future 的瞬間有可能 syscall 剛好已經成功），并且一個更嚴重的問題是一定會丟失 Future 捕獲的 buffer。針對這兩個問題 Monoio 支持了帶取消能力的 IO trait，取消時會推入 CancelOp，用戶需要在取消后繼續等待原 Future 執行結束（由于它已經被取消了，所以會預期在較短的時間內返回），對應的 syscall 可能執行成功或失敗，并返還 buffer。

上層無感知的 Driver 探測和切換

第二個特性是支持上層無感知的 Driver 探測和切換。

traitOpAble{
fnuring_op(&mutself)->io_uring::Entry;
fnlegacy_interest(&self)->Option<(ready::Diirection,?usize)>;
fnlegacy_call(&mutself)->io::Result;
}

通過 Feature 或代碼指定 Driver，并有條件地做運行時探測

暴露統一的 IO 接口，即 AsyncReadRent 和 AsyncWriteRent

內部利用 OpAble 統一組件實現（對 Read、Write 等 Op 做抽象）

具體來說，比如想做 accept、connect 或者 read、write 之類的，這些 op 是實現了 OpAble 的，實際對應這三個 fn ：

uring_op：生成對應 uring SQE

legacy_interest：返回其關注的讀寫方向

legacy_call：直接執行syscall

整個流程會將一個實現了 opable 的結構 submit 到的 driver 上，然后會返回一個實現了 future 的東西，之后它 poll 的時候和 drop 的時候具體地會分發到兩個 driver 實現中的一個，就會用這三個函數里面的一個或者兩個。

性能

性能是 Monoio 的出發點和最大的優點。除了 io_uring 帶來的提升外，它設計上是一個 thread-per-core 模式的 Runtime。

所有 Task 均僅在固定線程運行，無任務竊取。

Task Queue 為 thread local 結構操作無鎖無競爭。

高性能其實主要源于兩個方面：

Runtime內部高性能：基本等價于裸對接syscall

用戶代碼高性能：結構盡量 thread local 不跨線程

任務竊取和 thread-per-core 兩種機制的對比：

如果用 tokio 的話，可能某一個線程上它的任務非常少，可能已經空了，但是另一個線程上任務非常多。那么這時候比較閑的線程就可以把任務從比較忙的任務上偷走，這一點和 Golang 非常像。這種機制可以較充分的利用 CPU，應對通用場景可以做到較好的性能。

但跨線程本身會有開銷，多線程操作數據結構時也會需要鎖或無鎖結構。但無鎖也不代表沒有額外開銷，相比純本線程操作，跨線程的無鎖結構會影響緩存性能，CAS 也會付出一些無效 loop。除此之外，更重要的是這種模式也會影響用戶代碼。

舉個例子，我們內部需要一個 SDK 去收集本程序的一些打點，并把這些打點聚合之后去上報。在基于 tokio 的實現下，要做到極致的性能就比較困難。如果在 thread-per-core 結構的 Runtime 上，我們完全可以將聚合的 Map 放在 thread-local 中，不需要任何鎖，也沒有任何競爭問題，只需要在每個線程上啟動一個任務，讓這個任務定期清空并上報 thread local 中的數據。而在任務可能跨線程的場景下，我們就只能用全局的結構來聚合打點，用一個全局的任務去上報數據。聚合用的數據結構就很難不使用鎖。

所以這兩種模式各有各的優點，thread-per-core 模式下對于可以較獨立處理的任務可以達到更好的性能。共享更少的東西可以做到更好的性能。但是 thread-per-core 的缺點是在任務本身不均勻的情況下不能充分利用 CPU。對于特定場景，如網關代理等，thread-per-core 更容易充分利用硬件性能，做到比較好的水平擴展性。當前廣泛使用 nginx 和 envoy 都是這種模式。

我們做了一些 benchmark，Monoio 的性能水平擴展性是非常好的。當 CPU 核數增加的時候，只需要增加對應的線程就可以了。

功能性

Thread-per-core 不代表沒有跨線程能力。用戶依舊可以使用一些跨線程共享的結構，這些和 Runtime 無關；Runtime 提供了跨線程等待的能力。

任務在本線程執行，但可以等待其他線程上的任務，這個是一個很重要的能力。舉例來說，用戶需要用單線程去拉取遠程配置，并下發到所有線程上。基于這個能力，用戶就可以非常輕松地實現這個功能。

跨線程等待的本質是在別的線程喚醒本線程的任務。實現上我們在 Waker 中標記任務的所屬權，如果當前線程并不是任務所屬線程，那么 Runtime 會通過無鎖隊列將任務發送到其所屬線程上；如果此時目標線程處于休眠狀態（陷入 syscall 等待 IO），則利用事先安插的 eventfd 將其喚醒。喚醒后，目標線程會處理跨線程 waker 隊列。

除了提供跨線程等待能力外，Monoio 也提供了 spawn_blocking 能力，供用戶執行較重的計算邏輯，以免影響到同線程的其他任務。

兼容接口

需要允許用戶以兼容方式使用，即便付出一些性能代價。由于目前很多組件（如 hyper 等）綁定了 tokio 的 IO trait，而前面講了由于地層 driver 的原因這兩種 IO trait 不可能統一，所以生態上會比較困難。對于一些非熱路徑的組件，需要允許用戶以兼容方式使用，即便付出一些性能代價。

//tokioway
lettcp=tokio:connect("1.1.1.1.1:80").await.unwrap();
//monoioway(withmonoio-compat)
lettcp=monoio_compat::new(monoio_tcp);
letmonoio_tcp=monoio::connect("1.1.1.1:80").await.unwrap();
//bothofthemimplementstokio::io::AsyncReaddandtokio::io:AsyncWrite

我們提供了一個 Wrapper，內置了一個 buffer，用戶使用時需要多付出一次內存拷貝開銷。通過這種方式，我們可以為 monoio 的組件包裝出 tokio 的兼容接口，使其可以使用兼容組件。

04

Runtime 對比 & 應用

這部分介紹 runtime 的一些對比選型和應用。

前面已經提到了關于均勻調度和 thread-per-core 的一些對比，這里主要說一下應用場景。對于較大量的輕任務，thread-per-core 模式是適合的。特別是代理、網關和文件 IO 密集的應用，使用 Monoio 就非常合適。
還有一點，Tokio 致力于一個通用跨平臺，但是 Monoio 設計之初就是為了極致性能，所以是期望以 io_uring 為主的。雖然也可以支持 epoll 和 kqueue，但僅作 fallback。比如 kqueue 其實就是為了讓用戶能夠在 Mac 上去開發的便利性，其實不期望用戶真的把它跑在這（未來將支持 Windows）。

生態部分，Tokio 的生態是比較全的，Monoio 的比較缺乏，即便有兼容層，兼容層本身是有開銷的。Tokio 有任務竊取，可以在較多的場景表現很好，但其水平擴展性不佳。Monoio 的水平擴展就比較好，但是對這個業務場景和編程模型其實是有限制的。所以 Monoio 比較適合的一些場景就是代理、網關還有緩存數據聚合等。以及還有一些會做文件 io 的，因為 io_uring 對文件 io 非常好。如果不用 io_uring 的話，在 Linux 下其實是沒有真異步的文件 io 可以用的，只有用 io_uring 才能做到這一點。還適用于這種文件 io 比較密集的，比如說像 DB 類型的組件。

Tokio-uring 其實是一個構建在 tokio 之上的一層，有點像是一層分發層，它的設計比較漂亮，我們也參考了它里面的很多設計，比如說像那個傳遞所有權的這種形式。但是它還是基于 tokio 做的，在 epoll 之上運行 uring，沒有做到用戶透明。當組件在實現時，只能在使用 epoll 和使用 uring 中二選一，如果選擇了 uring，那么編譯產物就無法在舊版本 linux 上運行。而 Monoio 很好的支持了這一點，支持動態探測 uring 的可用性。

Monoio 應用

Monoio Gateway: 基于 Monoio 生態的網關服務，我們優化版本 Benchmark 下來性能優于 Nginx；

Volo: CloudWeGo Team 開源的 RPC 框架，目前在集成中，PoC 版本性能相比基于 Tokio 提升 26%

我們也在內部做了一些業務業務試點，未來我們會從提升兼容性和組件建設上入手，就是讓它更好用。