RustのRayon入門：データ並列処理をかんたんに実現する

Rayonとは？

Rustで並列処理を行うとき、スレッドを自前で管理するのは意外と大変です。std::threadを使えば実装できますが、チャネルやMutexなどを駆使する必要があります。

Rayonはそのような苦労を大幅に軽減してくれるクレートです。通常のイテレータを**並列イテレータ（par_iter）**に切り替えるだけで、自動的に複数のスレッドへ処理を分散してくれます。データを並列で処理したい場面に非常に相性が良く、Rustコミュニティでも広く利用されています。

セットアップ

まずCargo.tomlにRayonを追加します。

[dependencies]
rayon = "1.10"

その後、cargo buildを実行して依存関係を取得してください。

par_iter()の基本

Rayonの最大の特徴は、通常の.iter()を.par_iter()に変えるだけで並列処理が有効になる点です。

use rayon::prelude::*;

fn main() {
    let numbers: Vec<i64> = (1..=10).collect();

    // 通常のイテレータ（シングルスレッド）
    let sum_single: i64 = numbers.iter().map(|&x| x * x).sum();
    println!("通常の合計: {}", sum_single);

    // 並列イテレータ（マルチスレッド）
    let sum_parallel: i64 = numbers.par_iter().map(|&x| x * x).sum();
    println!("並列の合計: {}", sum_parallel);
}

rayon::prelude::*をインポートするだけでpar_iter()が使えるようになります。.iter()との切り替えはたった4文字の変更です。

filter・map・collectとの組み合わせ

RayonはIteratorトレイトと同じような感覚でメソッドチェーンを使えます。

use rayon::prelude::*;

fn main() {
    let data: Vec<u64> = (1..=1_000_000).collect();

    // 偶数のみ取り出し、2乗して集める
    let result: Vec<u64> = data
        .par_iter()
        .filter(|&&x| x % 2 == 0)
        .map(|&x| x * x)
        .collect();

    println!("結果の件数: {}", result.len());
    println!("先頭5件: {:?}", &result[..5]);
}

filter・map・collectなどの操作はすべて並列で実行されます。要素数が多くなるほど、並列処理の恩恵を受けやすくなります。

par_iter_mut()で可変参照を使う

コレクションの中身を書き換えたい場合はpar_iter_mut()が使えます。

use rayon::prelude::*;

fn main() {
    let mut values: Vec<f64> = (1..=10).map(|x| x as f64).collect();

    // 各要素を平方根に更新する
    values.par_iter_mut().for_each(|x| {
        *x = x.sqrt();
    });

    println!("{:?}", values);
}

for_eachと組み合わせることで、各要素への副作用のある操作も簡単に並列化できます。

par_sort()で並列ソート

Rayonはソートも並列化できます。

use rayon::prelude::*;

fn main() {
    let mut data: Vec<i32> = (0..1_000_000).rev().collect();

    // 並列ソート
    data.par_sort();

    println!("先頭: {:?}", &data[..5]);
    println!("末尾: {:?}", &data[data.len()-5..]);
}

標準のsort()と同じインターフェースで並列ソートが実現できます。大量データの整列が必要な場面で効果的です。

パフォーマンス比較の例

実際にどれくらい差が出るか、計算量の多いタスクで確認してみましょう。

use rayon::prelude::*;
use std::time::Instant;

fn heavy_calc(n: u64) -> u64 {
    // 素数判定（重い処理の代わり）
    if n < 2 {
        return 0;
    }
    for i in 2..=(n as f64).sqrt() as u64 {
        if n % i == 0 {
            return 0;
        }
    }
    1
}

fn main() {
    let data: Vec<u64> = (2..=100_000).collect();

    // シングルスレッド
    let start = Instant::now();
    let single: u64 = data.iter().map(|&x| heavy_calc(x)).sum();
    println!("シングル: {}ms, 素数の数: {}", start.elapsed().as_millis(), single);

    // マルチスレッド（Rayon）
    let start = Instant::now();
    let parallel: u64 = data.par_iter().map(|&x| heavy_calc(*x)).sum();
    println!("並列:     {}ms, 素数の数: {}", start.elapsed().as_millis(), parallel);
}

実行環境によって差は異なりますが、CPUコアが多い環境ほど並列版が大幅に高速になります。

注意点とベストプラクティス

オーバーヘッドに注意

要素数が少ない場合はスレッドを立ち上げるオーバーヘッドが大きく、並列版のほうが遅くなることもあります。目安として数万件以上の大量データや1件あたりの処理が重いタスクに向いています。

スレッドセーフな型を使う

Rayonは内部でスレッドプールを管理しています。クロージャ内でキャプチャする変数はSendトレイトを実装している必要があります。RcやCellなどスレッドセーフでない型は使えないため、共有が必要な場合はArcやMutexを使いましょう。

スレッドプールのカスタマイズ

デフォルトではCPUコア数に合わせてスレッドが起動しますが、スレッド数を変更したい場合は以下のように設定できます。

rayon::ThreadPoolBuilder::new()
    .num_threads(4)
    .build_global()
    .unwrap();

まとめ

Rayonを使うと、.iter()を.par_iter()に変えるだけで手軽にデータ並列処理を実現できます。

スレッド管理を意識せず並列処理を書けるRayonは、パフォーマンスが求められるRustアプリケーションの強い味方です。ぜひ大量データを扱う処理に取り入れてみてください。