async-rust

非同期 Rust

非同期 Rust は、エグゼキューターのスレッドがブロックされる、ガードを await ポイントをまたいで保持する、フューチャーが処理の途中でドロップされる、という 3 つの箇所で失敗します。このスキルが扱うのは、並行実行とキャンセル時の正しさです。そもそもコードを非同期にすべきかどうかは判断しません。

非同期処理は無料で並行性をもたらすわけではありません

async fn は、ポーリングされるまで何もしないフューチャーを返します。並行性をもたらすのはランタイムと、フューチャーを組み合わせる仕組み — join!、select!、spawn — であり、async キーワードではありません：

// Serial: the second fetch starts only after the first completes.
let a = fetch_user(id).await;
let b = fetch_orders(id).await;

// Concurrent: both futures are polled at the same time.
let (a, b) = tokio::join!(fetch_user(id), fetch_orders(id));

関数内の各 .await は、生成されるフューチャーのサイズを大きくします。また、大きなフューチャーは移動のたびにコピーされます — Box::pin に渡すときや、JoinSet に渡すときなどです。await が多く、ローカル変数が大きいと、サイズがキロバイト単位になることもあります。プロファイルで移動が問題だとわかったら、内部のフューチャーをボックス化するか、関数を分割してください。

ランタイムは最初に一度だけ選択する

大多数のケースでは tokio を、意図的に選択してください。ライブラリでは、可能であればランタイムに依存しない設計にし、それができなければ統合機能をフィーチャーゲートで制御してください。また、ライブラリ関数内でランタイムを起動してはいけません（#[tokio::main] はバイナリまたはテストで使います。そこで Runtime::block_on を呼ぶと、利用者に 2 つ目の、競合するエグゼキューターを渡すことになります）。

非同期コンテキストでのブロッキング処理

async fn 内での CPU 負荷の高いループ、同期ファイル／ネットワーク呼び出し、または .await を含まない計算処理は、エグゼキューターのスレッドを停止させ、そのスレッド上のほかのタスクをすべて飢餓状態にします。対処法は、ブロッキング I/O と短時間の CPU 処理には tokio::task::spawn_blocking、継続的な CPU 処理には専用スレッドプール（rayon）を使い、タスク内で std::thread::sleep を呼ばないことです。計算ループで定期的に yield する方法は、より弱い対処法です。分割した処理が本当に小さい場合にのみ効果があります。

Send、Sync、spawn の境界条件

tokio::spawn は Send + 'static を要求します。.await をまたいで保持されるものはすべて Send でなければなりません。Rc と RefCell はコンパイルに失敗し、std::sync::MutexGuard を .await をまたいで保持するとデッドロックが発生します：

// Deadlock waiting to happen: the std guard is held across the await.
let mut guard = state.lock().unwrap();
guard.value = fetch().await;

// Either drop the guard before awaiting…
let value = fetch().await;
state.lock().unwrap().value = value;

// …or use the async-aware mutex when the lock genuinely must span the await.
let mut guard = state.lock().await; // tokio::sync::Mutex
guard.value = fetch().await;

ロックを .await をまたいで保持する必要がある場合に限り tokio::sync::Mutex を使い、それ以外では std::sync::Mutex を優先してください。clippy::await_holding_lock はよくあるケースを検出します。

キャンセル安全性

どのフューチャーも、任意の .await ポイントでドロップされる可能性があります — select! とタイムアウトが行うのはこれです。入力を消費したものの、まだ確定していないフューチャーは、処理の途中でドロップされるとデータを失います。原則として、取り消せない処理をキャンセルされない 1 つのまとまりで実行するか、処理を再開可能にしてください。具体例は CANCELLATION.md にあります。

タスク間での共有状態

共有状態の変更には Arc<Mutex<T>>、タスク間の作業の受け渡しにはチャネル、状態の論理的な所有者が 1 つの場合はメッセージパッシングを使います。状態を所有し、指示を受けるタスクのほうが、共有メモリにアクセスするタスクよりも優れています。

チャネル	用途
mpsc	複数のプロデューサーから 1 つのコンシューマーへ
oneshot	1 回限りの応答
watch	中間の値が抜けても問題ない、最新の値
broadcast	バッファーサイズに上限のある、すべてのメッセージをすべての受信側に届ける場合

選択基準はパフォーマンスではなく意味です。watch を使うべき場面で broadcast を選ぶと、予期しないラグ受信エラーが発生します。

動的なタスクグループとその中止の意味については、CANCELLATION.md を参照してください。

公開 API での非同期処理

追加の制御が必要な場合を除き、-> impl Future より async fn を優先してください。デシュガーされた形式は、呼び出し側が依存する形で Send 性（またはその欠如）も追加で保証します。トレイトでは async fn にまだ Send 境界条件を指定する簡潔な構文がなく、オブジェクト安全な非同期トレイトでは #[async_trait] が引き続き実用的な選択肢です。

詳細の参照先

タイムアウトとキャンセルのエラー型については /rust-errors、共有所有権については /ownership-not-clone、公開 API のコミットメントについては /rust-api-design、手動で実装する Future の Pin 不変条件については /unsafe-rust を参照してください。

検証

cargo clippy --all-targets   # add -- -D warnings only if the repo has no lint config
cargo test

非同期処理に関連する lint（await_holding_lock、await_holding_refcell_ref、 async_yields_async）は、明示的に有効化しなくても出力されます。非同期処理のバグはタイミングに依存するため、cargo test が通っても十分な証拠にはなりません。リポジトリで #[tokio::test] を使用している場合は、変更したパスを #[tokio::test(flavor = "multi_thread", worker_threads = 2)] でカバーしてください。