async/await の内部メカニズム

async fn を呼び出すと Future が構築され、その関数本体はポーリングされることで進行します。Future は完了していない計算を記述するものであり、独立したスレッドを内包しません。実行時（Executor）がいつポーリングするかを決定し、非同期 I/O などの待機ソースは、処理を継続できる場合に実行時に通知を行います。

.await は必ずしも一度のスレッド切り替えを意味しません。子 Future が即座に Ready を返す場合、現在のポーリング処理はそのまま継続でき、1回のポーリング内で複数の await を完了させることも可能です。

Future の構築と関数本体の実行

use std::{cell::Cell, future::Future, task::{Context, Poll, Waker}};
async fn calculate(started: &Cell<u32>) -> u32 {
    started.set(started.get() + 1);
    let left = std::future::ready(20).await;
    let right = std::future::ready(22).await;
    left + right
}
fn main() {
    let started = Cell::new(0);
    let future = calculate(&started);
    assert_eq!(started.get(), 0);
    let mut future = std::pin::pin!(future);
    let mut context = Context::from_waker(Waker::noop());
    assert_eq!(future.as_mut().poll(&mut context), Poll::Ready(42));
    assert_eq!(started.get(), 1);
}

関数本体は初回の poll 呼び出し時に実行されます。2つの ready Future は即座に完了するため、関数全体が1回の poll で 42 を返します。ここでは noop Waker を使用していますが、これは待機を必要としないサンプルであるためです。これはタスクの実行をスケジュールせず、実際の非同期 I/O を駆動するために使用することはできません。Waker::noop

呼び出し式の引数評価は、Future の構築時に行われます。例えば calculate(make_counter()) の場合、make_counter の呼び出しは自動的に遅延されません。async 関数本体の実行のみがポーリングによって駆動されます。async move を使用した場合でも、キャプチャされた値の所有権はブロック作成時に取得され、その後一度も poll されなかったとしても同様です。

コンパイラはどのような状態を保存するか

コンパイラは async 関数を、一時停止と再開が可能な匿名の状態機械として表現します。これには、現在待機している子 Future と、後続の処理で使用する必要があるローカル値が含まれます。read_file を待機している場合、通常、すでに取得済みだが実際にはまだ生成されていないファイル内容ではなく、読み込み操作の状態が保存されます。

状態機械の具体的なレイアウトは実装の詳細であり、「各 await に対応する同名の列挙型バリアントが必ず存在する」ということを ABI として扱うことはできません。大きな値が一時停止点をまたぐと Future のサイズが大きくなる可能性があるため、ライフサイクルを早期に終了させることで状態のサイズを削減できる場合があります。

poll と Waker の契約

Future::poll は Pin<&mut Self> と Context を受け取り、Poll<Output> を返します。Pending は結果がまだ得られないことを示し、関連する実装はタスクが継続できる場合に備えてウェイクアップ経路を確立する必要があります。この作業はポーリングされている子 Future に委譲でき、各層で Waker を保存する必要はありません。

wake はスケジューリングへのヒントであり、マージされる場合もあれば、poll が Pending を返す前に発生することもあります。独自の待機プリミティブを実装する際は、状態のチェックと通知の登録の間の競合状態を正しく処理する必要があります。「まず結果がないか確認し、その後でコールバックを登録する」という順序では、その間に完了したイベントを見逃す可能性があります。

Ready を返した後、呼び出し側は完了した Future を任意にポーリングしてはいけません。具体的な型ではパニックが発生したり、インターフェースで定義された他の動作が発生したりする可能性があり、これを Iterator の next を継続するプロトコルのように扱うべきではありません。Future

Ready は完了を意味するが、ビジネスロジック上の成功を意味するわけではない

Future の Output は Result<T, E> になり得ます。したがって、Poll::Ready(Err(error)) は計算が完了し、エラーを返したことを意味します。Pending は最終的な結果がまだ得られないことを意味します。実行時は Err を「準備ができていない」と見なしてポーリングを続けるべきではありません。

use std::{future::Future, task::{Context, Poll, Waker}};
fn main() {
    let mut future = std::pin::pin!(async {
        "bad".parse::<u32>()
    });
    let mut context = Context::from_waker(Waker::noop());
    assert!(matches!(future.as_mut().poll(&mut context), Poll::Ready(Err(_))));
}

なぜ最初の Waker だけを保存すれば十分でないのか

同じ Future が後続の poll で異なるタスクの Waker を受信する可能性があります。例えば、異なる外側の組み合わせによって駆動されている場合などです。待機プリミティブはプロトコルに従って現在の通知先を更新する必要があり、初回の登録後はずっと変わらないと仮定してはいけません。wake 自体は結果を伴わないため、結果は適切な同期された共有状態に保存され、その後 poll によって読み取られる必要があります。Future::poll

推論が容易な待機プリミティブでは、「結果が存在するか」と「現在の Waker」を同じロックで保護します。消費者はロック内で結果をチェックし、未完成であれば Waker を登録します。生産者はロック内で結果を書き込み、Waker を取得してロックを解放し、wake を呼び出します。これにより、完了イベントがチェックと登録の間に発生して誰も通知しないという事態を防ぐことができます。実際のライブラリでは、クローズ、重複する完了、キャンセル、公平性の処理も必要です。この順序の説明を、チャネルが完全に実装されたものとして扱わないでください。

wake は実行そのものではない

外部の準備完了、wake、poll を分けると、await を独立して動く関数だと誤解しにくくなります。Pending にしてから I/O だけを完了すると、再実行可能でも Future はまだ結果を返していません。wake の回数と poll の回数も一致する必要はありません。

実行時不要の手動ポーリングの例

以下の Future は、最初の要求で再スケジューリングを要求し、2回目で結果を返します。この例では安全な Wake インターフェースを使用しており、RawWaker を手書きする必要はありません。

use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
use std::task::{Context, Poll, Wake, Waker};

struct YieldOnce(bool);
impl Future for YieldOnce {
    type Output = u32;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
        if self.0 {
            Poll::Ready(42)
        } else {
            self.0 = true;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

struct WakeCount(AtomicUsize);
impl Wake for WakeCount {
    fn wake(self: Arc<Self>) { self.0.fetch_add(1, Ordering::Relaxed); }
    fn wake_by_ref(self: &Arc<Self>) { self.0.fetch_add(1, Ordering::Relaxed); }
}

fn main() {
    let count = Arc::new(WakeCount(AtomicUsize::new(0)));
    let waker = Waker::from(count.clone());
    let mut context = Context::from_waker(&waker);
    let mut future = std::pin::pin!(YieldOnce(false));
    assert!(future.as_mut().poll(&mut context).is_pending());
    assert_eq!(count.0.load(Ordering::Relaxed), 1);
    assert_eq!(future.as_mut().poll(&mut context), Poll::Ready(42));
}

これは決定論的な2回のポーリングのデモであり、汎用の Executor ではありません。実際の Executor はタスクキューと通知の競合を管理する必要があり、すべての Pending を無条件の park に対応させることはできません。

Pin が固定するのは指し示される値

状態機械には内部データへの参照が含まれる可能性があるため、適切な pinning 契約に基づいてアドレスの安定性が確立されると、!Unpin の固定された値を正しくデストラクトするまで、勝手にムーブすることはできません。

Pin はコンパイラがすべての値に課す魔法のアドレスロックではなく、安全な API の制限付き操作と unsafe 実装の義務によって維持される契約です。Pin<Box<F>> というポインタラッパー自体はムーブ可能ですが、それが指す F の位置が保たれている限りです。F: Unpin の場合、Pin は同様のムーブ制限を課しません。Pin

初回の固定前、Future は通常ムーブ可能です。すべての async Future が実際に自己参照を持つわけではありませんし、ローカル変数に String が1つあるからといって、必ずしも String オブジェクトのムーブによってヒープバッファが移動するわけではありません。オブジェクトとそれが指すストレージを区別する必要があります。

ラッパの移動と固定された値の移動は違う

所有ハンドルだけを動かすと、アドレス依存の F はヒープの A に残ります。F 自体を動かす反実仮想で、内部アドレスが自動修正されない理由を確認します。Pin が関係を自動構築するわけではなく、初期化・投影・破棄にも契約が必要です。Unpin 型はこの移動制限から外れます。

await、spawn、join はそれぞれ何をするのか

await は現在のタスク内で子 Future を進行させます。spawn は Future を独立したタスクとして登録します。join 系コンビネータは、OS スレッドを作成せずに、同じタスク内で複数の Future を進行させることができます。並行性はライフサイクルの重複を意味し、並列実行かどうかは実行環境に依存します。

1回の poll で長時間の CPU ループやブロッキング呼び出しを行うと、現在の Executor スレッドを占有してしまいます。関数を単に async とマークしただけでは、其中的な std::thread::sleep、ブロッキング I/O、または重い計算が自動的に実行権を手放すことはありません。

Drop はキャンセルメカニズムの一部

未完成の Future をドロップすることは、Rust の async における一般的なキャンセル方法です。言語はこの動作を許可しており、Future が「自由にドロップできない」と言うべきではありません。これにより保持されている状態がデストラクトされ、そのオブジェクトが後続でポーリングされることはなくなります。ただし、独立したタスク、デバイスに渡されたリクエスト、およびリモートトランザクションは継続する可能性があります。

キャンセル安全性の議論は、ドロップ後に進捗が失われたり、ビジネスプロトコルが破綻したりしないかどうかに関わります。例えば、read_exact が一部のバイトを読み取った後にキャンセルした場合、操作を再作成しても元の位置から始まらない可能性があります。リソースが解放されるかどうかと、ビジネス上の効果が取り消されるかどうかは別の問題です。

キャンセルがローカル値をデストラクトすること、かつ一時停止点後のクリーンアップを実行しないことを検証

use std::{cell::Cell, future::Future, rc::Rc, task::{Context, Waker}};
struct Resource(Rc<Cell<u32>>);
impl Drop for Resource {
    fn drop(&mut self) { self.0.set(self.0.get() + 1); }
}
async fn waiting(drops: Rc<Cell<u32>>, after: Rc<Cell<bool>>) {
    let resource = Resource(drops);
    std::future::pending::<()>().await;
    after.set(true);
    drop(resource);
}
fn main() {
    let drops = Rc::new(Cell::new(0));
    let after = Rc::new(Cell::new(false));
    let mut future = Box::pin(waiting(Rc::clone(&drops), Rc::clone(&after)));
    let mut context = Context::from_waker(Waker::noop());
    assert!(future.as_mut().poll(&mut context).is_pending());
    assert_eq!(drops.get(), 0);
    drop(future);
    assert_eq!(drops.get(), 1);
    assert!(!after.get());
}

pending は決して自ら Ready になることはありません。ここでは意図的に1回だけ poll してからキャンセルするため、ウェイクアップを待つ必要はありません。Resource は構築され、一時停止点をまたいで生存しているため、キャンセル時に Drop が呼び出されます。after.set(true) は await の後にあり、実行されることはありません。重要なクリーンアップ処理を関数の末尾にのみ記述した場合、キャンセルパスがそこに到達することを保証できません。

通常の Drop は同期デストラクションであり、リモートでの解放確認を自動的に await することはありません。プロトコルレベルでのクローズが必要な場合は、明示的に close/shutdown などの非同期ステップを提供し、キャンセル、タイムアウト、またはクラッシュ後の補償方法を決定する必要があります。ローカルのガードがすでに解放されていても、リモートトランザクションが取り消されたことを推論してはいけません。

Pin 参照のドロップと、所有する Future のドロップ

上記の例では Box::pin を使用して、drop(future) がヒープ上の Future を明示的に破棄できるようにしています。もし pin! が Pin<&mut F> を返す場合、この参照ラッパーをドロップしても、それが指す F が即座に破棄されるわけではありません。実際のオブジェクトは、そのスコープのライフサイクルに従って解放されます。キャンセルを推論する際は、まず Future を所有するコンテナを見つけ、借用を「停止する」ことを「デストラクションがすべて実行された」と誤認しないようにしてください。

キャンセル安全性では進捗が誰に保存されているかを確認する

Future 自身が「何バイト送信したか」という進捗を保存している場合、キャンセルによってこの進捗が失われ、基盤となるストリームがすでに一部を受信している可能性があります。最初から再送信するとデータが重複する可能性があります。進捗が外側の永続オブジェクトに保存されている場合、次の操作で継続できる可能性がありますが、メッセージの境界、冪等性、および障害復旧を定義する必要があります。

したがって、同時に以下の質問を行う必要があります。どのオブジェクトをドロップするか、そのオブジェクトがどの進捗を所有しているか、外部アクションがどこまで進行したか、そして操作を再作成した場合、どこから始まるか。キャンセル安全性は具体的な操作の契約であり、すべての async fn が自動的に備える性質ではありません。

タイムアウト、select、およびバックグラウンドタスクの具体的な境界については、Tokio ランタイム および Tower を参照してください。