tokio-troubleshooting

Tokio のトラブルシューティング

このスキルでは、Tokio で構築された非同期アプリケーションをデバッグし、トラブルシューティングするための手法を紹介します。

tokio-console を使ったランタイムの検査

非同期ランタイムをリアルタイムで監視します。

// In Cargo.toml
[dependencies]
console-subscriber = "0.2"

// In main.rs
fn main() {
    console_subscriber::init();

    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            run_application().await
        });
}

別のターミナルでコンソールを実行します：

tokio-console

監視する主なメトリクス：

• タスクの生成率とタスクの合計数
• タスクごとのポーリング時間
• アイドル時間とビジー時間
• Waker の操作
• リソースの使用状況

問題の特定：

• ポーリング時間が長い：非同期コンテキストで CPU 負荷の高い処理が行われています
• Waker が多い：競合、または非効率なポーリングの可能性があります
• タスク数が増え続ける：タスクリーク、または無制限なタスク生成が発生しています
• アイドル時間が長い：処理が不足しているか、ブロッキング操作が発生しています

デッドロックとハングのデバッグ

デッドロックを検出して解決します。

よくあるデッドロックのパターン

// BAD: Potential deadlock
async fn deadlock_example() {
    let mutex1 = Arc::new(Mutex::new(()));
    let mutex2 = Arc::new(Mutex::new(()));

    let m1 = mutex1.clone();
    let m2 = mutex2.clone();
    tokio::spawn(async move {
        let _g1 = m1.lock().await;
        tokio::time::sleep(Duration::from_millis(10)).await;
        let _g2 = m2.lock().await; // May deadlock
    });

    let _g2 = mutex2.lock().await;
    tokio::time::sleep(Duration::from_millis(10)).await;
    let _g1 = mutex1.lock().await; // May deadlock
}

// GOOD: Consistent lock ordering
async fn no_deadlock_example() {
    let mutex1 = Arc::new(Mutex::new(()));
    let mutex2 = Arc::new(Mutex::new(()));

    // Always acquire locks in same order
    let _g1 = mutex1.lock().await;
    let _g2 = mutex2.lock().await;
}

// BETTER: Avoid nested locks
async fn best_example() {
    // Use message passing instead
    let (tx, mut rx) = mpsc::channel(10);

    tokio::spawn(async move {
        while let Some(msg) = rx.recv().await {
            process_message(msg).await;
        }
    });

    tx.send(message).await.unwrap();
}

タイムアウトを使ったハングの検出

use tokio::time::{timeout, Duration};

async fn detect_hang() {
    match timeout(Duration::from_secs(5), potentially_hanging_operation()).await {
        Ok(result) => println!("Completed: {:?}", result),
        Err(_) => {
            eprintln!("Operation timed out - potential hang detected");
            // Log stack traces, metrics, etc.
        }
    }
}

try_lock によるデッドロックの検出

use tokio::sync::Mutex;

async fn try_with_timeout(mutex: &Mutex<State>) -> Option<State> {
    for _ in 0..10 {
        if let Ok(guard) = mutex.try_lock() {
            return Some(guard.clone());
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    eprintln!("Failed to acquire lock - possible deadlock");
    None
}

メモリーリークの検出

メモリーリークを特定して修正します。

タスクリーク

// BAD: Tasks never complete
async fn leaking_tasks() {
    loop {
        tokio::spawn(async {
            loop {
                // Never exits
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        });
    }
}

// GOOD: Tasks have exit condition
async fn proper_tasks(shutdown: broadcast::Receiver<()>) {
    loop {
        let mut shutdown_rx = shutdown.resubscribe();
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => break,
                    _ = tokio::time::sleep(Duration::from_secs(1)) => {
                        // Work
                    }
                }
            }
        });
    }
}

Arc の循環参照

// BAD: Reference cycle
struct Node {
    next: Option<Arc<Mutex<Node>>>,
    prev: Option<Arc<Mutex<Node>>>, // Creates cycle!
}

// GOOD: Use weak references
use std::sync::Weak;

struct Node {
    next: Option<Arc<Mutex<Node>>>,
    prev: Option<Weak<Mutex<Node>>>, // Weak reference breaks cycle
}

メモリー使用量の監視

use sysinfo::{System, SystemExt};

pub async fn memory_monitor() {
    let mut system = System::new_all();
    let mut interval = tokio::time::interval(Duration::from_secs(60));

    loop {
        interval.tick().await;
        system.refresh_memory();

        let used = system.used_memory();
        let total = system.total_memory();
        let percent = (used as f64 / total as f64) * 100.0;

        tracing::info!(
            used_mb = used / 1024 / 1024,
            total_mb = total / 1024 / 1024,
            percent = %.2 percent,
            "Memory usage"
        );

        if percent > 80.0 {
            tracing::warn!("High memory usage detected");
        }
    }
}

tracing を使ったパフォーマンスのプロファイリング

パフォーマンス分析のためにコードを計測します。

use tracing::{info, instrument, span, Level};

#[instrument]
async fn process_request(id: u64) -> Result<Response, Error> {
    let span = span!(Level::INFO, "database_query");
    let _enter = span.enter();

    let data = fetch_from_database(id).await?;

    drop(_enter);

    let span = span!(Level::INFO, "transformation");
    let _enter = span.enter();

    let result = transform_data(data).await?;

    Ok(Response { result })
}

// Configure subscriber for flame graphs
use tracing_subscriber::layer::SubscriberExt;

fn init_tracing() {
    let fmt_layer = tracing_subscriber::fmt::layer();
    let filter_layer = tracing_subscriber::EnvFilter::from_default_env();

    tracing_subscriber::registry()
        .with(filter_layer)
        .with(fmt_layer)
        .init();
}

パニックメッセージの理解

よくある非同期処理のパニックパターン：

生成されたタスク内のパニック

// Panic is isolated to the task
tokio::spawn(async {
    panic!("This won't crash the program");
});

// To catch panics
let handle = tokio::spawn(async {
    // Work that might panic
});

match handle.await {
    Ok(result) => println!("Success: {:?}", result),
    Err(e) if e.is_panic() => {
        eprintln!("Task panicked: {:?}", e);
        // Handle panic
    }
    Err(e) => eprintln!("Task cancelled: {:?}", e),
}

Send + 'static エラー

// ERROR: future cannot be sent between threads
async fn bad_example() {
    let rc = Rc::new(5); // Rc is !Send
    tokio::spawn(async move {
        println!("{}", rc); // Error!
    });
}

// FIX: Use Arc instead
async fn good_example() {
    let rc = Arc::new(5); // Arc is Send
    tokio::spawn(async move {
        println!("{}", rc); // OK
    });
}

// ERROR: borrowed value does not live long enough
async fn lifetime_error() {
    let data = String::from("hello");
    tokio::spawn(async {
        println!("{}", data); // Error: data might not live long enough
    });
}

// FIX: Move ownership
async fn lifetime_fixed() {
    let data = String::from("hello");
    tokio::spawn(async move {
        println!("{}", data); // OK: data is moved
    });
}

よくあるエラーパターンと解決策

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

// PROBLEM: Detected with tokio-console (long poll time)
async fn blocking_example() {
    std::thread::sleep(Duration::from_secs(1)); // Blocks thread!
}

// SOLUTION
async fn non_blocking_example() {
    tokio::time::sleep(Duration::from_secs(1)).await; // Yields control
}

// For unavoidable blocking
async fn necessary_blocking() {
    tokio::task::spawn_blocking(|| {
        expensive_cpu_work()
    }).await.unwrap();
}

チャネルのクローズエラー

// PROBLEM: SendError because receiver dropped
async fn send_error_example() {
    let (tx, rx) = mpsc::channel(10);
    drop(rx); // Receiver dropped

    match tx.send(42).await {
        Ok(_) => println!("Sent"),
        Err(e) => eprintln!("Send failed: {}", e), // Channel closed
    }
}

// SOLUTION: Check if receiver exists
async fn handle_closed_channel() {
    let (tx, rx) = mpsc::channel(10);

    tokio::spawn(async move {
        // Receiver keeps channel open
        while let Some(msg) = rx.recv().await {
            process(msg).await;
        }
    });

    // Or handle the error
    if let Err(e) = tx.send(42).await {
        tracing::warn!("Channel closed: {}", e);
        // Cleanup or alternative action
    }
}

タスクのキャンセル

// PROBLEM: Task cancelled unexpectedly
let handle = tokio::spawn(async {
    // Long-running work
});

handle.abort(); // Cancels task

// SOLUTION: Handle cancellation gracefully
let handle = tokio::spawn(async {
    let result = tokio::select! {
        result = do_work() => result,
        _ = tokio::signal::ctrl_c() => {
            cleanup().await;
            return Err(Error::Cancelled);
        }
    };
    result
});

非同期コードを効果的にテストする

信頼性の高い非同期テストを作成します。

#[tokio::test]
async fn test_with_timeout() {
    tokio::time::timeout(
        Duration::from_secs(5),
        async {
            let result = my_async_function().await;
            assert!(result.is_ok());
        }
    )
    .await
    .expect("Test timed out");
}

#[tokio::test]
async fn test_concurrent_access() {
    let shared = Arc::new(Mutex::new(0));

    let handles: Vec<_> = (0..10)
        .map(|_| {
            let shared = shared.clone();
            tokio::spawn(async move {
                let mut lock = shared.lock().await;
                *lock += 1;
            })
        })
        .collect();

    for handle in handles {
        handle.await.unwrap();
    }

    assert_eq!(*shared.lock().await, 10);
}

// Test with mocked time
#[tokio::test(start_paused = true)]
async fn test_with_time_control() {
    let start = tokio::time::Instant::now();

    tokio::time::sleep(Duration::from_secs(100)).await;

    // Time is mocked, so this completes instantly
    assert!(start.elapsed() < Duration::from_secs(1));
}

デバッグチェックリスト

非同期処理の問題をトラブルシューティングするときは、次の項目を確認します。

• tokio-console を使ってランタイムの動作を監視する
• tracing を使ってブロッキング操作を確認する
• すべてのロックが適切に解放されていることを確認する
• タスクリーク（タスク数の増加）がないか確認する
• メモリー使用量の経時変化を監視する
• ハングを検出するためにタイムアウトを追加する
• チャネルのクローズエラーを確認する
• Send + 'static の境界条件が満たされていることを確認する
• try_lock を使ってデッドロックの可能性を検出する
• tracing を使ってパフォーマンスのボトルネックをプロファイリングする
• 時間に依存するコードを tokio-test でテストする
• Weak 参照を使って Arc の循環参照がないか確認する

便利なツール

• tokio-console：非同期ランタイムをリアルタイムで監視
• tracing：構造化ログとプロファイリング
• cargo-flamegraph：フレームグラフを生成
• valgrind/heaptrack：メモリーのプロファイリング
• perf：Linux での CPU プロファイリング
• Instruments：macOS でのプロファイリング

ベストプラクティス

1. 開発では必ず tokio-console を使います
2. 重要なコードパスにtracing のスパンを追加します
3. ハングを検出するためにタイムアウトを積極的に使います
4. リークを検出するためにタスク数を監視します
5. 最適化の前にプロファイリングします — まず測定します
6. 実際の並行処理でテストします — 正常系だけをテストしないでください
7. すべてのタスクでキャンセルを適切に処理します
8. デバッグには構造化ログを使います
9. ロックのネストを避けます — メッセージパッシングを優先します
10. 必要に応じてロックの順序を文書化します