Tokio ランタイム

Tokio は、タスクスケジューリング、非同期 I/O、タイマー、同期ツールを提供します。エグゼキュータは Future をポーリングし、ドライバーは OS のイベントを待機してタスクをウェイクアップします。ユーザーコードは、非同期環境に適した操作で積極的に実行権を譲渡する必要があります。

Tokio のタスクは軽量な非同期タスクであり、複数のスレッド間でスケジューリングできます。「本物の並行処理だからコルーチンではない」という書き方は避けてください。タスクが状態をどのように保存し、いくつの OS スレッド上で実行されるかは、異なる次元の問題です。

2つの主要なスケジューリング方式

current_thread は、block_on を呼び出したスレッド上でタスクを駆動しますが、大量の待機を並行して処理することも可能です。multi_thread は複数のワーカーを使用し、ワークステールなどのメカニズムを備えています。デフォルトのワーカー数は利用可能な CPU 数などの設定に関連しており、上書き可能です。これは、1つの物理コアが1つのワーカーに永久にバインドされることを保証するものではありません。

I/O ドライバーのプラットフォーム実装や具体的なキューアルゴリズムはバージョン固有の詳細であり、概念を理解する際に、各ソケット、ファイル、タイマーを同じシステムコールに無理やり当てはめる必要はありません。一部のファイル操作は、カーネルの readiness 通知に直接変化するのではなく、ブロッキングプールを介して行われます。Runtime

current_thread のランタイムは、block_on によって駆動されていない場合、関連するタスクが勝手に実行を続けることはありません。マルチスレッドランタイムの block_on 主 Future も、単に1つのワーカー上の通常の spawn タスクとは等しくありません。

完全な有界生産/消費プログラム

Cargo.toml：

[package]
name = "tokio-demo"
version = "0.1.0"
edition = "2024"

[dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] }

src/main.rs：

use tokio::sync::mpsc;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (tx, mut rx) = mpsc::channel(2);
    let producer = tokio::spawn(async move {
        for value in 1_u32..=4 {
            if tx.send(value).await.is_err() {
                return; // コンシューマが終了したため、生産しない
            }
        }
    });

    let mut total = 0;
    while let Some(value) = rx.recv().await {
        total += value;
    }
    producer.await?;
    assert_eq!(total, 10);
    println!("{total}");
    Ok(())
}

容量 2 はキュー内の項目数を制限し、キューがいっぱいの場合は send が待機します。生産者が終了して最後の sender を解放した後、コンシューマはバッファを空にしてから None を取得します。producer.await はタスク結果を明示的に回収し、パニックなどの失敗を「タスクの自然な終了」として扱わないようにします。

有界チャネルは、生産前に作成された無限のタスクを制限しません。設計時には、バックプレッシャーが実際にリソースを消費するエントリポイントにまで及ぶようにする必要があります。

キューが有界であっても、キュー外のメモリを見る必要がある

キュー容量が 100 で、1メッセージの最大サイズが 1 MiB であると仮定します。メッセージペイロードだけでも約 100 MiB に達する可能性があります。さらに、生産者が構築中の値、処理中の値、タスクの状態もあります。容量は項目数でカウントされ、各項目のバイト数や、上位で既に作成されたすべての Future を自動的に制限するものではありません。

無限に spawn してから、各タスク内部でセマフォを待機すると、リソース集中領域の並行性は制限されますが、待機中のタスクは無限に蓄積される可能性があります。タスク数を制御するには、タスクを作成する前に許可を取得するか、既存のタスクが完了してから次の作業を受け取る/作成する必要があります。結果も適時に回収する必要があります。Semaphore

spawn の Send とライフタイム

tokio::spawn は通常、Future とその出力が Send + 'static を満たすことを要求します。タスクは独立して存続し、ワーカー間で移動する可能性があるためです。current_thread スケジューラを使用している場合でも、この API の型要件は自動的に緩和されません。

!Send のローカルタスクが必要な場合は、unsafe で Rc を Send にラップするのではなく、Tokio のローカルタスクメカニズムを使用して LocalSet と spawn_local を使用してください。一時的な Rc や同期ロックガードは、await 点を超えて存続しない場合、Future 全体の Send を妨げない可能性があります。明確なレキシカルスコープが、これを表現するのが最も簡単です。

ローカルタスクを使用して Rc を処理する

以下の独立したプログラムは、本稿と同じ Cargo 依存関係を使用します。LocalSet は、そこで spawn_local されたタスクがローカル実行環境で実行されることを保証しますが、spawn_local には、タスクが所有するキャプチャが 'static である必要があります。これは、通常の借用スコープスレッドの代替手段ではありません。

use std::{cell::RefCell, rc::Rc};
use tokio::task::LocalSet;
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let local = LocalSet::new();
    local.run_until(async {
        let values = Rc::new(RefCell::new(vec![1]));
        let shared = Rc::clone(&values);
        let task = tokio::task::spawn_local(async move {
            tokio::task::yield_now().await;
            shared.borrow_mut().push(2);
        });
        task.await.unwrap();
        assert_eq!(&*values.borrow(), &[1, 2]);
    }).await;
}

Rc はこのタスク内で await を跨いで存続するため、tokio::spawn には適していませんが、明示的なローカルタスクモデルには適しています。RefCell の借用は適切な範囲内で終了させる必要があり、LocalSet はスレッド単一の再入競合を解消するものではありません。LocalSet

JoinHandle の Drop はタスクキャンセル不等于

Tokio JoinHandle をドロップしても、通常はタスクが待機から外れるだけで、タスクは引き続き実行されることがあります。abort はタスクのキャンセルを要求し、適切なスケジューリングの境界で有効になります。ハンドルを待機することで、キャンセルの終了を観察できます。決して実行権を譲渡しない長時間の poll は、別のタスクが abort を呼び出しても即座にプリエンプトされるわけではありません。JoinHandle

これは、まだ spawn されておらず、現在のタスクが直接所有する Future をドロップすることとは異なります。リソースのクリーンアップとバックグラウンド作業の終了には、明確な所有権と協調プロトコルが必要です。必要に応じて、シャットダウンシグナルを使用して完了を待機します。

JoinHandle のタイムアウト待機後に、明示的に終了を要求して回収する

use std::{sync::{Arc, atomic::{AtomicBool, Ordering}}, time::Duration};
use tokio::{sync::oneshot, time::timeout};
struct MarkDropped(Arc<AtomicBool>);
impl Drop for MarkDropped {
    fn drop(&mut self) { self.0.store(true, Ordering::SeqCst); }
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let dropped = Arc::new(AtomicBool::new(false));
    let flag = Arc::clone(&dropped);
    let (ready_tx, ready_rx) = oneshot::channel();
    let mut task = tokio::spawn(async move {
        let resource = MarkDropped(flag);
        ready_tx.send(()).unwrap();
        std::future::pending::<()>().await;
        drop(resource);
    });
    ready_rx.await.unwrap(); // リソースが構築されたことを確認し、タスクは pending で停止する。
    assert!(timeout(Duration::from_millis(1), &mut task).await.is_err());
    assert!(!dropped.load(Ordering::SeqCst));
    task.abort();
    assert!(task.await.unwrap_err().is_cancelled());
    assert!(dropped.load(Ordering::SeqCst));
}

timeout は &mut task を借用し、タイムアウト後も JoinHandle を保持するため、継続待機、協調シャットダウン、または abort の決定が可能です。ハンドルを値で timeout に渡すと、タイムアウト時に待機オブジェクト内のハンドルがドロップされますが、バックグラウンドタスクは通常存続し続けます。上記の例では Pending は永遠に完了せず、キャンセル結果は確定的です。通常のタスクの場合、abort が有効になる前に正常に完了している可能性があるため、処理コードはこの競合を許容する必要があります。

1 ミリ秒はタイムアウト分岐をトリガーするためだけであり、タイミング精度を保証するものではありません。timeout は、常に返さない poll をプリエンプトできません。時間のかかる計算は、期限を過ぎてからでも継続する可能性があります。これは、OS レベルの強制終了ではなく、待機プロトコルを制限するものです。timeout

待つ側を捨てるとタスクは止まるか

同じ drop 構文でも所有関係で効果が異なります。直接の Future は処理状態を所有し、JoinHandle は生成されたタスクを待機・制御する入口です。handle の破棄と、初期化後の abort を比べ、待機中止、取消要求、取消完了を区別します。

select! のキャンセル境界は演算子の所有権による

select! は現在のタスク内で複数のブランチをポーリングしますが、各ブランチを自動的に異なるスレッドに spawn するわけではありません。ブランチが選択されると、他のローカル Future はドロップされます。ブランチが外部の固定された Future を借用している場合、その基盤オブジェクトは外部によって保持され続ける可能性があります。

あるブランチが JoinHandle を所有している場合、それをドロップしてもタスクが外れるだけで、そのタスクは自動的に停止しません。部分的に入力を消費した read_exact Future をドロップした場合、進行状況が失われる可能性があります。繰り返される select ループでは、各操作のキャンセル安全性を確認する必要があります。select!

デフォルトのブランチポーリング順序と biased モデルでは、異なる公平性の考慮事項があります。biased もビジネスロジックの公平性を保証するものではありません。ブランチにブロッキングコードが含まれている場合、select の進捗全体がブロックされます。

JoinSet はタスクの作成と結果回収を同じループに配置する

use tokio::task::{JoinError, JoinSet};
fn accept(result: Result<Result<u32, &'static str>, JoinError>,
          values: &mut Vec<u32>, rejected: &mut usize) {
    match result {
        Ok(Ok(value)) => values.push(value),
        Ok(Err("invalid input")) => *rejected += 1,
        other => panic!("予期しないタスク結果：{other:?}"),
    }
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let mut tasks = JoinSet::new();
    let mut values = Vec::new();
    let mut rejected = 0;
    for value in 0u32..5 {
        if tasks.len() >= 2 {
            accept(tasks.join_next().await.unwrap(), &mut values, &mut rejected);
        }
        tasks.spawn(async move {
            tokio::task::yield_now().await;
            if value == 3 { Err("invalid input") } else { Ok(value * 2) }
        });
        assert!(tasks.len() <= 2);
    }
    while let Some(result) = tasks.join_next().await {
        accept(result, &mut values, &mut rejected);
    }
    values.sort_unstable();
    assert_eq!(values, [0, 2, 4, 8]);
    assert_eq!(rejected, 1);
}

ここでは、未回収のタスクを最大2つまで保持します。JoinSet は完了順に結果を渡すため、順序は保証されません。そのため、例では検証時にソートしています。ビジネスロジックで元の順序が必要な場合は、シーケンス番号を付与して再ソートする必要がありますが、ソートバッファの上限も考慮してください。出力の Vec はこの有限の例では安全かつ制御可能ですが、無限の入力に対しては、結果を逐次配信するように変更する必要があります。

戻り値には2層の失敗があります。外側の JoinError はタスクのキャンセルまたはパニックを表し、内側の Err はビジネスロジックが明示的に返した失敗です。.await? は外側のエラーのみを伝播するため、これに基づいてビジネスロジックの成功を断定することはできません。JoinSet

JoinSet をドロップすると、そのタスクの中止が要求されます。これは単一の JoinHandle をドロップすることとは異なります。すべてのデストラクションの完了を確認する必要がある場合は、対応するシャットダウン/回収プロセスを待機する必要があります。中止要求を発行するだけでは、リソース回収が完了したことを意味しません。

ブロッキング作業と CPU 作業

std::thread::sleep、ブロッキングライブラリ呼び出し、長時間の計算は、非同期ワーカーを長時間占有すべきではありません。有限のブロッキングタスクには spawn_blocking を使用できますが、すでに実行を開始したクロージャは通常、abort でキャンセルできません。ランタイムのシャットダウン時も、その終了動作を考慮する必要があります。spawn_blocking

ブロッキングプールには独自のスレッドとキュー戦略があり、無限の CPU 作業に対して適切な並行上限を自動的に提供するものではありません。大量の CPU 計算は、明示的なレートリミッティングまたは適切な専用実行プールを使用し、目標負荷で検証する必要があります。

ロックは「async 内では非同期ロックが必要」という理由だけで選択しない

短く、競合が少なく、await を跨がないクリティカルセクションでは、通常の Mutex を使用できます。ロックを await を跨いで保持すると、他のタスクをブロックしたり、デッドロックを引き起こしたりする可能性があります。また、std の MutexGuard は Future が Send を満たさない原因になることもあります。

Tokio の非同期 Mutex は、ロックを非同期で待機することを許可し、必要に応じて await を跨いでガードを保持できます。しかし、これはビジネスロジックのデッドロックや長時間の独占問題を免除するものではありません。共有 I/O リソースは、場合によっては1つのタスクが所有し、メッセージパッシングでリクエストを送信する方が適していることもあります。Tokio 共有状態

レイテンシを診断する際は、キュー待機、スケジュールによる実行取得の遅延、CPU 実行、I/O 待機、ロック待機を区別してください。リクエストが遅いことだけを見ていても、ワーカー数が少なすぎるのか、下流の能力不足なのかを特定することはできません。

シャットダウン順序と観測可能性

有界コンシューマがすべての sender が消失するまで待ってから終了する場合、各 sender のクローンが誰によって保持されているかを知る必要があります。長寿命の構造内に残っている不要なクローンにより、recv が永遠に None を取得できない場合があります。Receiver::close を介して新しい送信を明示的に停止した後でも、バッファリングされたメッセージを引き続き受信して処理する必要があります。以前に取得した許可は、チャネルの契約に従って処理する必要があり、close を即座にすべてクリアすることと解釈してはいけません。mpsc Receiver::close

サービスのシャットダウンは、エントリの停止、協調シグナルの送信、受信済み作業の処理、タスクの回収、そしてランタイムの終了という順序で行うことができます。各ステップで、保留中の数量、最古のタスクの年齢、残りの予算を記録することで、生産者が停止していないのか、あるタスクが実行権を譲渡していないのか、下流が待機中なのかを区別できます。

ランタイムを手動で構築する場合は、必要な時間または I/O ドライバーを有効にする必要があります。「作成に成功した」からといって、ドライバーに依存するすべての Future が実行可能になるわけではありません。ランタイムハンドルへのコンテキストの入り込みと、ランタイムを継続的に駆動する実際の動作は異なります。特に current_thread 環境では、明確な block_on 所有権を維持する必要があります。