Rust並行処理：非同期ランタイムの適切な使い方

Posted at 2025-04-18

Rust の非同期ランタイムはさまざまな場面で非常に有用であり、特に高い並行性と高性能が求められる I/O 集約型のアプリケーションにおいて顕著です。以下は、Rust の非同期ランタイムがよく使用されるシナリオのいくつかです。

ネットワークプログラミング

Web サーバーおよびクライアント：多数の同時接続を処理することは Web サーバーの主要な要件です。非同期ランタイムを使用することで、各接続ごとにスレッドを生成することなく、数千から数万の同時リクエストを効率的に処理できます。Tokio や async-std のようなランタイムは、高性能な Web アプリケーションの構築に必要なツール、たとえば TCP/UDP ソケット、HTTP クライアントおよびサーバーなどを提供します。
リアルタイム通信アプリ：チャットサーバーやオンラインゲームサーバーなどは、大量の同時接続と低遅延のメッセージ配信を処理する必要があります。非同期ランタイムはこれらの接続を効率よく管理し、高速なメッセージルーティングおよび処理を提供します。
プロキシサーバーおよびロードバランサー：これらのアプリケーションは、大量の同時接続とデータ転送を処理する必要があります。非同期ランタイムは、高性能な接続管理およびデータ転送を提供できます。

I/O 集約型アプリケーション

データベースドライバおよびクライアント：データベース操作は通常、大量の I/O 待機を伴います。非同期ランタイムを使用することで、アプリケーションはデータベースの応答を待っている間にも他のタスクを実行でき、全体的な性能が向上します。
ファイル I/O 操作：大規模なファイルを処理したり、複数のファイルに同時アクセスする必要があるアプリケーションでは、非同期 I/O を活用することで効率が向上します。たとえば、画像処理やログ解析などです。
マイクロサービスアーキテクチャ：マイクロサービス間の通信は通常ネットワーク経由で行われます。非同期ランタイムを利用することで、サービス間の通信効率と並行処理能力を向上させることができます。

並行および並列処理

並列計算：Rust の std::thread モジュールは並列計算に使用できますが、I/O 集約型の場面では非同期ランタイムの方が一般に効果的です。たとえば、非同期タスクを用いて複数のファイルを並列にダウンロードしたり、複数のネットワークリクエストを同時に処理したりできます。
タスクキューおよびバックグラウンド処理：非同期ランタイムはタスクキューの実装にも適しています。時間のかかるタスクをキューに入れて非同期に実行することで、メインスレッドのブロックを回避できます。

CPU 集約型のアプリケーションには、非同期ランタイムの使用は一般的に推奨されません。Rust の非同期ランタイムは I/O 集約型アプリケーションにおいて非常に優れた性能を発揮しますが、CPU 集約型の場面では大きな性能向上が得られないどころか、むしろオーバーヘッドを招く可能性があります。

なぜ CPU 集約型アプリケーションに非同期ランタイムは推奨されないのか？

非同期ランタイムの主な利点は、I/O 待機の効率的な処理にあります。プログラムがネットワークリクエストやファイルの読み書きなど、I/O 操作の完了を待つ必要がある場合、非同期ランタイムはその間に CPU を他のタスクに割り当てることで CPU 使用率を高めます。

しかし、CPU 集約型アプリケーションでは、CPU は常にビジー状態であり、他のタスクを実行する余地がないため、非同期ランタイムの利点は発揮できません。

さらに、非同期プログラミングには追加のオーバーヘッドが伴います。非同期コードはタスクの状態管理やコンテキストスイッチなどを行う必要があり、これらはパフォーマンスコストとなります。CPU 集約型のアプリケーションでは、これらのオーバーヘッドが潜在的な利益を相殺したり、場合によっては逆効果になることもあります。

CPU 集約型タスクにはスレッドが依然として最適な選択肢です。CPU に負荷のかかる処理には、タスクを複数に分割し、マルチコア CPU 上で並行に実行するマルチスレッド処理の方が効果的です。Rust の std::thread モジュールを用いることで、スレッドの生成と管理が簡単に行えます。また、スレッドプールを使用してリソース消費を最適化することも可能です。

CPU 集約型アプリケーションで非同期ランタイムを使用すべきタイミングとは？

一般的には純粋な CPU 集約型アプリケーションで非同期ランタイムを使用することは推奨されませんが、CPU 集約型の処理の中に一部 I/O 操作（たとえば設定ファイルの読み込みやログの書き込みなど）が含まれている場合、それらの I/O 処理の効率を上げる目的で非同期ランタイムの利用を検討できます。ただし、I/O 操作の比率が大きすぎないことが前提です。比率が高すぎると、逆に効率が落ちることがあります。

また、CPU 集約型のタスクが他の非同期タスクと統合されており、ネットワークリクエストなどを同時に処理する必要があるような場面では、非同期ランタイムを使用することで、統一されたプログラミングモデルが実現でき、コードの構成や保守が容易になります。

Tokio は I/O 集約型のタスク向けに最適化されていますが、CPU 集約型のタスクを処理するための手段も提供しています。これにより、Tokio のランタイムをブロックして他のタスクの実行に悪影響を与えるのを防ぐことができます。以下に、Tokio で CPU 集約型のタスクを処理する代表的な方法を示します。

tokio::task::spawn_blocking の使用

これは、Tokio 内で CPU 集約型のタスクを実行する際の第一選択肢です。spawn_blocking 関数は、指定されたクロージャを専用のスレッドプールに移して実行します。このスレッドプールは Tokio ランタイムとは分離されているため、CPU 集約型のタスクが Tokio のランタイムスレッドをブロックすることはありません。その結果、他の I/O 集約型のタスクにも悪影響を与えずに実行できます。

use tokio::task;

#[tokio::main]
async fn main() {
    // 非同期 I/O 処理の開始
    println!("I/O 処理開始");
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    println!("I/O 処理完了");

    // CPU 集約型タスクの実行
    let result = task::spawn_blocking(move || {
        println!("CPU 集約型タスク開始");
        // 重い計算処理
        let mut sum = 0u64;
        for i in 0..100000000 {
            sum += i;
        }
        println!("CPU 集約型タスク完了");
        sum
    }).await.unwrap();

    println!("計算結果：{}", result);
}

この例では、CPU を多く消費する計算処理を spawn_blocking 内のクロージャに委ねています。処理に時間がかかっても、Tokio の非同期ランタイムはブロックされず、I/O 操作は正常に動作し続けます。

独立スレッドプールの使用

もう一つの方法は、CPU 集約型のタスクを独立したスレッドプールで実行することです。これには std::thread や rayon のようなライブラリを用いることができます。非同期ランタイムである Tokio との間でデータをやり取りするためには、std::sync::mpsc や tokio::sync::mpsc などのチャネルを使用します。

use std::thread;
use std::sync::mpsc;
use tokio::runtime::Runtime;

fn main() {
    let rt = Runtime::new().unwrap();
    let (tx, rx) = mpsc::channel();

    rt.block_on(async {
        // 非同期 I/O 処理の開始
        println!("I/O 処理開始");
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        println!("I/O 処理完了");

        // CPU 集約型タスクをスレッドプールに送信
        let tx_clone = tx.clone();
        thread::spawn(move || {
            println!("CPU 集約型タスク開始");
            // 重い計算処理
            let mut sum = 0u64;
            for i in 0..100000000 {
                sum += i;
            }
            println!("CPU 集約型タスク完了");
            tx_clone.send(sum).unwrap();
        });

        // チャネルから結果を受信
        let result = rx.recv().unwrap();
        println!("計算結果：{}", result);
    });
}