在同步的 Rust 方法中調(diào)用異步代碼經(jīng)常會(huì)導(dǎo)致一些問(wèn)題，特別是對(duì)于不熟悉異步 Rust runtime 底層原理的初學(xué)者。在本文中，我們將討論我們遇到的一個(gè)特殊問(wèn)題，并分享我們采取的解決方法的經(jīng)驗(yàn)。

背景和問(wèn)題

在做GreptimeDB項(xiàng)目的時(shí)候，我們遇到一個(gè)關(guān)于在同步 Rust 方法中調(diào)用異步代碼的問(wèn)題。經(jīng)過(guò)一系列故障排查后，我們弄清了問(wèn)題的原委，這大大加深了對(duì)異步 Rust 的理解，因此在這篇文章中分享給大家，希望能給被相似問(wèn)題困擾的 Rust 開(kāi)發(fā)者一些啟發(fā)。

我們的整個(gè)項(xiàng)目是基于Tokio這個(gè)異步 Rust runtime 的，它將協(xié)作式的任務(wù)運(yùn)行和調(diào)度方便地封裝在.await調(diào)用中，非常簡(jiǎn)潔優(yōu)雅。但是這樣也讓不熟悉 Tokio 底層原理的用戶一不小心就掉入到坑里。

我們遇到的問(wèn)題是，需要在一個(gè)第三方庫(kù)的 trait 實(shí)現(xiàn)中執(zhí)行一些異步代碼，而這個(gè) trait 是同步的，我們無(wú)法修改這個(gè) trait 的定義。

traitSequencer{
fngenerate(&self)->Vec;
}

我們用一個(gè)PlainSequencer來(lái)實(shí)現(xiàn)這個(gè) trait ，而在實(shí)現(xiàn)generate方法的時(shí)候依賴一些異步的調(diào)用（比如這里的PlainSequencer::generate_async）：

implPlainSequencer{
asyncfngenerate_async(&self)->Vec{
letmutres=vec![];
foriin0..self.bound{
res.push(i);
tokio::sleep(Duration::from_millis(100)).await;
}
res
}
}

implSequencerforPlainSequencer{
fngenerate(&self)->Vec{
self.generate_async().await
}
}

這樣就會(huì)出現(xiàn)問(wèn)題，因?yàn)間enerate是一個(gè)同步方法，里面是不能直接 await 的。

error[E0728]:`await`isonlyallowedinside`async`functionsandblocks
-->src/common/tt.rs30
|
31|/fngenerate(&self)->Vec{
32||self.generate_async().await
||^^^^^^onlyallowedinside`async`functionsandblocks
33||}
||_____-thisisnot`async`

我們首先想到的是，Tokio 的 runtime 有一個(gè)Runtime::block_on方法，可以同步地等待一個(gè) future 完成。

implSequencerforPlainSequencer{
fngenerate(&self)->Vec{
RUNTIME.block_on(async{
self.generate_async().await
})
}
}

#[cfg(test)]
modtests{
#[tokio::test]
asyncfntest_sync_method(){
letsequencer=PlainSequencer{
bound:3
};
letvec=sequencer.generate();
println!("vec:{:?}",vec);
}
}

編譯可以通過(guò)，但是運(yùn)行時(shí)直接報(bào)錯(cuò)：

Cannotstartaruntimefromwithinaruntime.Thishappensbecauseafunction(like`block_on`)attemptedtoblockthecurrentthreadwhilethethreadisbeingusedtodriveasynchronoustasks.
thread'tests::test_sync_method'panickedat'Cannotstartaruntimefromwithinaruntime.Thishappensbecauseafunction(like`block_on`)attemptedtoblockthecurrentthreadwhilethethreadisbeingusedtodriveasynchronoustasks.',/Users/lei/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.17.0/src/runtime/enter.rs9

提示不能從一個(gè)執(zhí)行中的 runtime 直接啟動(dòng)另一個(gè)異步 runtime?？磥?lái) Tokio 為了避免這種情況特地在Runtime::block_on入口做了檢查。既然不行那我們就再看看其他的異步庫(kù)是否有類似的異步轉(zhuǎn)同步的方法。

果然找到一個(gè)futures::block_on。

implSequencerforPlainSequencer{
fngenerate(&self)->Vec{
futures::block_on(async{
self.generate_async().await
})
}
}

編譯同樣沒(méi)問(wèn)題，但是運(yùn)行時(shí)代碼直接直接 hang 住不返回了。

cargotest--color=always--packagetokio-demo

--bintttests::test_sync_method

--no-fail-fast----format=json

--exact-Zunstable-options--show-output

Compilingtokio-demov0.1.0(/Users/lei/Workspace/Rust/learning/tokio-demo) Finishedtest[unoptimized+debuginfo]target(s)in0.39s Runningunittestssrc/common/tt.rs(target/debug/deps/tt-adb10abca6625c07) {"type":"suite","event":"started","test_count":1} {"type":"test","event":"started","name":"tests::test_sync_method"} #theexecutionjusthangshere:(

明明generate_async方法里面只有一個(gè)簡(jiǎn)單的sleep()調(diào)用，但是為什么 future 一直沒(méi)完成呢？

并且吊詭的是，同樣的代碼，在tokio::test里面會(huì) hang 住，但是在tokio::main中則可以正常執(zhí)行完畢：

#[tokio::main]
pubasyncfnmain(){
letsequencer=PlainSequencer{
bound:3
};
letvec=sequencer.generate();
println!("vec:{:?}",vec);
}

執(zhí)行結(jié)果：

cargorun--color=always--packagetokio-demo--bintt
Finisheddev[unoptimized+debuginfo]target(s)in0.05s
Running`target/debug/tt`
vec:[0,1,2]

其實(shí)當(dāng)初真正遇到這個(gè)問(wèn)題的時(shí)候定位到具體在哪里 hang 住并沒(méi)有那么容易。真實(shí)代碼中 async 執(zhí)行的是一個(gè)遠(yuǎn)程的 gRPC 調(diào)用，當(dāng)初懷疑過(guò)是否是 gRPC server 的問(wèn)題，動(dòng)用了網(wǎng)絡(luò)抓包等等手段最終發(fā)現(xiàn)是 client 側(cè)的問(wèn)題。

這也提醒了我們在出現(xiàn) bug 的時(shí)候，抽象出問(wèn)題代碼的執(zhí)行模式并且做出一個(gè)最小可復(fù)現(xiàn)的樣例（Minimal Reproducible Example）是非常重要的。

Catchup

在 Rust 中，一個(gè)異步的代碼塊會(huì)被make_async_expr編譯為一個(gè)實(shí)現(xiàn)了std::Future的 generator。

#[tokio::test]
asyncfntest_future(){
letfuture=async{
println!("hello");
};

//theaboveasyncblockwon'tgetexecuteduntilweawaitit.
future.await;
}

而.await本質(zhì)上是一個(gè)語(yǔ)法糖，則會(huì)被lower_expr_await編譯成類似于下面的一個(gè)語(yǔ)法結(jié)構(gòu)：

//pseudo-rustcode
match::into_future(){
mut__awaitee=>loop{
matchunsafe{::poll(
<::Pin>::new_unchecked(&mut__awaitee),
::get_context(task_context),
)}{
::Ready(result)=>breakresult,
::Pending=>{}
}
task_context=yield();
}
}

在上面這個(gè)去掉了語(yǔ)法糖的偽代碼中，可以看到有一個(gè)循環(huán)不停地檢查 generator 的狀態(tài)是否為已完成（std::poll）。

自然地，必然存在一個(gè)組件來(lái)做這件事，這里就是 Tokio 和async-std這類異步運(yùn)行時(shí)發(fā)揮作用的地方了。Rust 在設(shè)計(jì)之初就特意將異步的語(yǔ)法（async/await）和異步運(yùn)行時(shí)的實(shí)現(xiàn)分開(kāi)，在上述的示例代碼中，poll 的操作是由 Tokio 的 executor 執(zhí)行的。

問(wèn)題分析

回顧完背景知識(shí)，我們?cè)倏匆谎鄯椒ǖ膶?shí)現(xiàn)：

fngenerate(&self)->Vec{
futures::block_on(async{
self.generate_async().await
})
}

調(diào)用generate方法的肯定是 Tokio 的 executor，那么 block_on 里面的self.generate_async().await這個(gè) future 又是誰(shuí)在 poll 呢？

一開(kāi)始我以為，futures::block_on會(huì)有一個(gè)內(nèi)部的 runtime 去負(fù)責(zé)generate_async的 poll。于是查看了代碼（主要是futures_executor::run_executor這個(gè)方法）：

fnrun_executor)->Poll>(mutf:F)->T{
let_enter=enter().expect(
"cannotexecute`LocalPool`executorfromwithin
anotherexecutor",
);

CURRENT_THREAD_NOTIFY.with(|thread_notify|{
letwaker=waker_ref(thread_notify);
letmutcx=Context::from_waker(&waker);
loop{
ifletPoll::Ready(t)=f(&mutcx){
returnt;
}
letunparked=thread_notify.unparked.swap(false,Ordering::Acquire);
if!unparked{
thread::park();
thread_notify.unparked.store(false,Ordering::Release);
}
}
})
}

立刻嗅到了一絲不對(duì)的味道，雖然這個(gè)方法名為run_executor，但是整個(gè)方法里面貌似沒(méi)有任何 spawn 的操作，只是在當(dāng)前線程不停的循環(huán)判斷用戶提交的 future 的狀態(tài)是否為 ready ??！

這意味著，當(dāng) Tokio 的 runtime 線程執(zhí)行到這里的時(shí)候，會(huì)立刻進(jìn)入一個(gè)循環(huán)，在循環(huán)中不停地判斷用戶的的 future 是否 ready。如果還是 pending 狀態(tài)，則將當(dāng)前線程 park 住。

假設(shè)，用戶 future 的異步任務(wù)也是交給了當(dāng)前線程去執(zhí)行，futures::block_on等待用戶的 future ready，而用戶 future 等待futures::block_on釋放當(dāng)前的線程資源，那么不就死鎖了？

這個(gè)推論聽(tīng)起來(lái)很有道理，讓我們來(lái)驗(yàn)證一下。既然不能在當(dāng)前 runtime 線程 block，那就重新開(kāi)一個(gè) runtime block：

implSequencerforPlainSequencer{
fngenerate(&self)->Vec{
letbound=self.bound;
futures::block_on(asyncmove{
RUNTIME.spawn(asyncmove{
letmutres=vec![];
foriin0..bound{
res.push(i);
tokio::sleep(Duration::from_millis(100)).await;
}
res
}).await.unwrap()
})
}
}

果然可以了。

cargotest--color=always--packagetokio-demo

--bintttests::test_sync_method

--no-fail-fast----format=json

--exact-Zunstable-options--show-output

Finishedtest[unoptimized+debuginfo]target(s)in0.04s Runningunittestssrc/common/tt.rs(target/debug/deps/tt-adb10abca6625c07) vec:[0,1,2]

值得注意的是，在futures::block_on里面，額外使用了一個(gè)RUNTIME來(lái) spawn 我們的異步代碼。其原因還是剛剛所說(shuō)的，這個(gè)異步任務(wù)需要一個(gè) runtime 來(lái)驅(qū)動(dòng)狀態(tài)的變化。

如果我們刪除 RUNTIME，而為 futures::block_on 生成一個(gè)新的線程，雖然死鎖問(wèn)題得到了解決，但tokio::sleep 方法的調(diào)用會(huì)報(bào)錯(cuò)"no reactor is running"，這是因?yàn)?Tokio 的功能運(yùn)作需要一個(gè) runtime：

called`Result::unwrap()`onan`Err`value:Any{..}
thread''panickedat'thereisnoreactorrunning,mustbecalledfromthecontextofaTokio1.xruntime',
...

tokio::main和tokio::test

在分析完上面的原因之后，“為什么tokio::main中不會(huì) hang 住而tokio::test會(huì) hang ??？“ 這個(gè)問(wèn)題也很清楚了，他們兩者所使用的的 runtime 并不一樣。tokio::main使用的是多線程的 runtime，而tokio::test使用的是單線程的 runtime，而在單線程的 runtime 下，當(dāng)前線程被futures::block_on卡死，那么用戶提交的異步代碼是一定沒(méi)機(jī)會(huì)執(zhí)行的，從而必然形成上面所說(shuō)的死鎖。

Best practice

經(jīng)過(guò)上面的分析，結(jié)合 Rust 基于 generator 的協(xié)作式異步特性，我們可以總結(jié)出 Rust 下橋接異步代碼和同步代碼的一些注意事項(xiàng)：

?將異步代碼與同步代碼結(jié)合使用可能會(huì)導(dǎo)致阻塞，因此不是一個(gè)明智的選擇。

?在同步的上下文中調(diào)用異步代碼時(shí)，請(qǐng)使用 futures::block_on 并將異步代碼 spawn 到另一個(gè)專用的 runtime 中執(zhí)行 ，因?yàn)榍罢邥?huì)阻塞當(dāng)前線程。

?如果必須從異步的上下文中調(diào)用有可能阻塞的同步代碼（比如文件 IO 等），則建議使用 tokio::spawn_blocking 在專門處理阻塞操作的 executor 上執(zhí)行相應(yīng)的代碼。

審核編輯：劉清