Posted inプログラミング

TokioによるRustのハイパフォーマンス非同期プログラミング：高負荷下でのタスク、チャネル、並行処理の制御

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

現場のトラブル：午前2時15分のPagerDutyアラートと沈黙するAPIのフリーズ

火曜日の午前2時15分、私たちのインジェスチョン（データ取り込み）サービスが突然応答を停止しました。レイテンシは通常時の軽快な12ミリ秒から45秒以上へと跳ね上がりました。メモリ使用量は4GBのコンテナ制限値に向かって一途に増加し、ロードバランサーではTCP接続のタイムアウトが頻発しました。しかし、プロセス自体は生存していました。パニックは起きておらず、クラッシュログもゼロ。セグメンテーション違反すらありません。Rustがメモリ安全性を完全に保証していたにもかかわらず、サービスは完全に凍りついていたのです。

このマイクロサービスは、Webhookイベントを受信し、暗号署名を検証し、JSONペイロードをパースしてKafkaに通知を送信する役割を担っています。8コアのCPU環境下で、日中の標準負荷である毎秒3,200リクエストを処理している間は、Tokioは余裕で動作していました。しかし、決済ゲートウェイが自動リトライの嵐を引き起こし、毎秒25,000リクエストが突如押し寄せました。90秒も経たないうちにワーカースレッドが停止。キューは詰まり、非同期パイプライン全体が完全にロックしてしまいました。

根本原因の分析：イベントループの飢餓状態と無制限バッファ

インシデント後のランタイムメトリクスとスレッドダンプの分析により、一見綺麗に書かれた非同期コードの裏に潜んでいた2つの致命的な欠陥が明らかになりました。

  1. 非同期ランタイムスレッドのブロッキング： 非同期ハンドラの中で、あるチームメンバーが同期的なEd25519署名検証ルーチンと、標準の std::fs::write による監査ログ出力を呼び出していました。Tokioは協調的マルチタスク（cooperative scheduling）に依存しています。各OSワーカースレッドは数千ものタスクを切り替えながら処理します。タスクが .await ポイントで制御を譲渡（yield）することなく計算負荷の高い処理を実行すると、そのタスクを抱えるOSスレッド全体がブロックされます。8コアマシン上で8つのワーカースレッドが動いている場合、わずか8つの暗号検証が同時に走るだけでランタイム全体が停止してしまいます。待機中のFutureはポーリングされず、ソケットはトラフィックの受け付けを停止しました。
  2. 無制限チャネルによるバックプレッシャー制御の破綻： パース済みのWebhookをバッチ送信タスクに渡すために、tokio::sync::mpsc::unbounded_channel() を使用していました。トラフィックスパイクの間、コンシューマーの処理が追いつかなくなりました。その結果、240万個以上のWebhook構造体がヒープ上に蓄積。メモリ消費量は数分で180MBから3.8GBへと急増しました。メモリアロケーションのオーバーヘッドが跳ね上がり、極度のアロケータスラッシングによってTokioの協調的タスクバジェットが窒息したのです。

Rustはコンパイル時にデータ競合を排除してくれます。しかし、スレッドの飢餓状態（starvation）、デッドロック、あるいは暴走するメモリリークから守ってくれるわけではありません。こうした障害モードを理解することこそが、単なるホビー用途の非同期コードと本番レベルの堅牢なインフラを分ける境界線です。

解決策の比較：高負荷下における並行処理戦略

高スループットなTokioアプリケーションをスケールさせるには、意図的なトレードオフの選択が不可欠です。負荷がかかった状況で不適切な並行処理プリミティブを選択すると、サービスは容易に崩壊します。

1. ネイティブOSスレッド vs. Tokioの軽量タスク

std::thread::spawn によるOSスレッドの生成は実行を完全に分離するため、1つのスレッドがブロックされても別のスレッドが凍結することはありません。しかし、OSスレッドは重量級です。Linuxでは、デフォルトの各スレッドが8MBのスタック領域を予約し（即座に20〜64KBの物理メモリを消費することが多い）、高コストなコンテキストスイッチが発生します。20,000個のソケットが同時接続している場合、カーネルはネットワークパケットを処理するよりもレジスタの退避・復元に多くのCPUサイクルを費やすことになります。

一方、Tokioのタスク（tokio::spawn）は、ユーザー空間のみで多重化されるグリーンスレッドです。各タスクはヒープメモリ上でわずか300〜500バイト程度から開始できます。Tokioのワークスティーリングスケジューラは、物理CPUコア数に応じたワーカースレッド全体に、これら数万ものタスクを効率的に分散します。

2. 同期ロック vs. 非同期Mutex

std::sync::MutexGuard を .await を跨いで保持することは絶対に避けてください。ロックを保持したままワーカースレッドが処理を中断（yield）すると、他のワーカースレッド上でそのロックを待つタスクがOSスレッドごとブロックされてしまいます。

逆に、すべての用途で tokio::sync::Mutex を安易に選ぶのも適切ではありません。非同期Mutexはヒープへのボックス化やアトミック同期ステートマシンのオーバーヘッドを伴います。カウンターのインクリメントや小さな VecDeque へのプッシュなど、.await を含まない短時間のインメモリ更新であれば、標準の std::sync::Mutex を使い、.await を呼び出す前にガードを明示的にドロップするのが適切です。

3. 無制限チャネル vs. バッファ制限付きバックプレッシャーチャネル

無制限チャネル（mpsc::unbounded_channel）は送信側をブロックしないため、手軽に見えます。しかし、これにはバックプレッシャー（背圧）の制御がありません。下流のコンシューマーが詰まった途端、メモリが爆発します。制限付きチャネル（mpsc::channel(capacity)）はバックプレッシャーを強制します。キューが一杯になると tx.send().await が送信側をサスペンドし、自然と上流へとバックプレッシャーを伝播させてTCPソケットの読み込みを減速させます。

本番向けアーキテクチャ：オフロード、流量制御、キャンセル安全性

サービスを抜本的に改善するため、私たちは取り込みパイプラインを3つのパターンに基づいて再設計しました。同期処理のオフロード、制限付きチャネルの徹底、そして構造化されたキャンセル安全性の確保です。

1. CPU集約的処理およびブロッキング処理の分離

暗号署名の検証、画像の圧縮、ブロッキングを伴うファイルシステム呼び出しは、Tokioのプライマリワーカープールから排除します。tokio::task::spawn_blocking を使用して同期計算をTokioのエラスティックなセカンダリプールにルーティングします。このプールはデフォルトで最大512スレッドまでスケールします。

use tokio::task;

// Bad: 呼び出しごとに協調的ランタイムスレッドを約1.8msブロックしてしまう
async fn process_payload_bad(data: Vec<u8>) -> Result<Vec<u8>, String> {
    let verified = heavy_crypto_verification(&data);
    Ok(verified)
}

// Good: ブロッキング処理をTokioの専用スレッドプールへオフロードする
async fn process_payload_good(data: Vec<u8>) -> Result<Vec<u8>, String> {
    let result = task::spawn_blocking(move || {
        heavy_crypto_verification(&data)
    })
    .await
    .map_err(|e| format!("タスクの結合エラー: {}", e))?;

    Ok(result)
}

fn heavy_crypto_verification(data: &[u8]) -> Vec<u8> {
    // 負荷の高いハッシュ計算や暗号検証のシミュレーション
    data.to_vec()
}

2. 制限付きチャネルによる並行性とバックプレッシャーの管理

キューバッファの上限を1,024イベントに設定しました。下流のワーカーが遅延した場合、tx.send(event).await がプロデューサータスクを一時停止させるため、メモリの無制限な浪費を防止できます。

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[derive(Debug)]
struct WebhookEvent {
    id: u64,
    payload: String,
}

#[tokio::main]
async fn main() {
    // 制限付きバッファ: 最大1,024件のイベントを保持（約1.2 MBのRAM）
    let (tx, mut rx) = mpsc::channel::<WebhookEvent>(1024);

    // コンシューマータスク: イベントを取り出してバッチ処理
    let consumer_handle = tokio::spawn(async move {
        while let Some(event) = rx.recv().await {
            handle_event(event).await;
        }
    });

    // プロデューサータスク: ネットワークからWebhookを受信
    let producer_handle = tokio::spawn(async move {
        for i in 0..5000 {
            let event = WebhookEvent {
                id: i,
                payload: format!("event_payload_{}", i),
            };

            // 1024件のバッファが満杯の場合は実行を一時中断する
            if let Err(e) = tx.send(event).await {
                eprintln!("コンシューマーが予期せず終了しました: {}", e);
                break;
            }
        }
    });

    let _ = tokio::join!(producer_handle, consumer_handle);
}

async fn handle_event(event: WebhookEvent) {
    sleep(Duration::from_millis(2)).await;
    println!("処理済みイベント: {}", event.id);
}

3. selectマクロによる複数並行ストリームの協調制御

本番環境のデーモンプロセスには、クリーンなシャットダウンシグナルとタイムアウト処理が必須です。tokio::select! マクロは複数の分岐を同時に監視し、いずれか1つが完了すると未完了のブランチを直ちにキャンセルします。未完了のFutureが破棄（drop）される際のキャンセル安全性には常に注意を払ってください。

use tokio::sync::watch;
use tokio::time::{sleep, Duration};

async fn worker_loop(mut shutdown_rx: watch::Receiver<bool>) {
    loop {
        tokio::select! {
            // 外部からのシャットダウンシグナルを監視
            _ = shutdown_rx.changed() => {
                if *shutdown_rx.borrow() {
                    println!("シャットダウンシグナルを受信しました。正常終了します...");
                    break;
                }
            }
            // 定期的なハートビートのフラッシュ
            _ = sleep(Duration::from_millis(500)) => {
                println!("テレメトリバッファをフラッシュ中...");
            }
        }
    }
}

Tokio並行処理の本番チェックリスト

  • 隠れたブロッキングI/Oの監査： 非同期コードパス内で std::fs や非同期非対応のDBドライバ（r2d2など）、重量級のシリアライズ処理がないか検索し、task::spawn_blocking 経由で実行するように分離します。
  • チャネル上限の設定： 高トラフィック経路にあるすべての mpsc::unbounded_channel を、メモリ許容量に応じたサイズの mpsc::channel(n) に置き換えます。
  • tokio-consoleによるタスクレイテンシのプロファイリング： ステージング環境で console-subscriber を導入します。ポーリング時間を計測し、1タスクあたり10〜50マイクロ秒以上ポーリングしている場合は、ランタイムの飢餓状態を疑ってください。
Share:
Tags: