Rust未経験者がRust100-exercises をやってみた話 (非同期プログラミング)

よみとばしポイント

どうも限界派遣SESのnikawamikanです。
ひとりアドカレ19日目の今日もQiitan獲得のために頑張ります。
先ほど5分前ぐらいに18日目の記事を書いていましたが、続きを書いていきましょう。

先日の記事の続きです。

前回からの続き

前回はEnumとエラーハンドリングについて話しました。
いいですよね。

今回は私の苦手な非同期プログラミングについて話していきます。

非同期プログラミング

非同期プログラミングは、複数の処理を同時に行うことが出来るプログラミング手法です。
Rustでは非同期ランタイムを使った非同期プログラミングと、スレッドを使った非同期プログラミングがあります。

スレッドを使った非同期プログラミング

スレッドを使った非同期プログラミングは、スレッドを使って複数の処理を同時に行うことが出来ます。
Rustではstd::threadを使ってスレッドを生成することが出来ます。

use std::thread;
fn main() {
    // 新しいスレッドを生成し、そのスレッドのハンドルを取得する
    let handle = thread::spawn(|| {
        println!("Hello from a thread!");
    });

    // スレッドの終了を待つ
    handle.join().unwrap();
}

joinメソッドを使うことで、スレッドの終了を待つことが出来ます。
これはJavaのThreadクラスのjoinメソッドと同じような使い方ですね。

最近は非同期ランタイムを使った非同期プログラミングのほうが主流だと思うので使うことは少ないと思いますが、いざ使う時には「そんなのあったなー」と思い出せる程度には頭の片隅に残しておくと良さそうですね。

ライフタイム

Rustの非同期プログラミングではライフタイムが重要です。
ライフタイムは、参照が有効な期間を表すものです。

threadモジュールのspawnメソッドはクロージャを引数に取りますが、そのクロージャの中で参照を使う場合、ライフタイムを指定する必要があります。

これは先に述べた通り、非同期プログラミングは複数の処理を同時に行うため、参照が有効な期間を明示的に示す必要があるためです。

これは非同期プログラミングというか、Rustにおいて本当に難しい概念だと思います。

以下の例を見てみましょう。

use once_cell::sync::Lazy;
use std::thread;

static X: Lazy<Vec<i32>> = Lazy::new(|| vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);

fn main() {
    println!("The answer is: {}", sum(&X));
}

fn sum(slice: &'static [i32]) -> i32 {
    let mid = slice.len() / 2;
    let (left, right) = slice.split_at(mid);

    let left_handle = thread::spawn(|| -> i32 { left.iter().sum() });
    let right_handle = thread::spawn(|| -> i32 { right.iter().sum() });

    left_handle.join().unwrap() + right_handle.join().unwrap()
}

最初のXの定義でstaticのライフタイムの参照を持つ変数を定義しています。

sum関数では、slice引数に'staticライフタイムの参照を取るようにしています。

これは呼び出し元でXのライフタイムがスレッド終了時まで有効であるということを示さなければならないためです。

要するにスレッド内で参照を使う場合は、ライフタイムを明示的に示す必要があるということです。

スコープ内でのライフタイム保証

前述の例ではstaticなライフタイム保証を行うことにより、スレッド内で参照を使うことが出来ました。
ただし、staticというのはメインスレッドが終了するまで有効なライフタイムであるため、それ以降その変数を使わないものであったとしてもメモリが開放されません。

そのため、スコープ内でライフタイムを保証する方法があります。

use std::thread;

fn main() {
    let v = vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
    println!("Sum: {}", sum(v));
}

fn sum(v: Vec<i32>) -> i32 {
    let mid = v.len() / 2;
    // スコープ内でライフタイムを保証する
    thread::scope(|scope| {
        let h1 = scope.spawn(|| v[..mid].into_iter().sum::<i32>());
        let h2 = scope.spawn(|| v[mid..].into_iter().sum::<i32>());

        h1.join().unwrap() + h2.join().unwrap()
    })
}

thread::scopeを使うことで、スコープ内でライフタイムを保証することが出来ます。

Rustがメモリ管理を行うために、私達はよしよししてあげる必要があるということですね。

MutexとArc

非同期プログラミングでは複数のスレッドが同時にアクセスするため、データ競合が発生する可能性もあります。

そのため、RustではMutexやArcを使ってデータ競合を防ぐことが出来ます。

Mutexはスレッド間でデータを共有するための構造体で、これはスレッドがデータにアクセスする際、ロックを取得することでデータの競合を防ぎます。

Arcは複数のスレッドが同時にデータにアクセスするための構造体で、これはデータを共有するための構造体です。
またatomic reference countingの略で、データの参照カウントを増減させ、参照が0になった際データを解放することが出来ます。

これは循環参照などが行われた場合はメモリリークを引き起こす可能性があるため、注意が必要です。

以下の例を見てみましょう。

use std::sync::{Arc, Mutex};

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = std::thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            *num += 1;
        });
        handles.push(handle);
    }

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

    println!("Result: {}", *counter.lock().unwrap());
}

ArcとMutexを使ってスレッド間でデータを共有しています。

Arc::cloneでArcの参照カウントを増やし、Mutex::lockでロックを取得してデータにアクセスしてnumをインクリメントしています。

最後にcounterの値を出力しています。

RwLock

RwLockはMutexと同じくデータ競合を防ぐための構造体ですが、Mutexと異なり複数のスレッドが同時にデータに読み込みを行うことが出来ます。

use std::sync::{Arc, RwLock};

fn main() {
    let counter = Arc::new(RwLock::new(0));

    let mut handles = vec![];
    for _ in 0..10 {
        let counter = counter.clone();
        let handle = std::thread::spawn(move || {
            let num = counter.read().unwrap();
            println!("Read: {}", *num);
        });
        handles.push(handle);
    }

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

    let mut num = counter.write().unwrap();
    *num += 1;
    println!("Write: {}", *num);
}

RwLockはreadメソッドで読み込みロックを取得し、writeメソッドで書き込みロックを取得します。

RwLockはMutexと異なり、複数のスレッドが同時にデータに読み込みを行うことが出来るため、読み込みが多い場合に有効です。
ただし、RwLockはMutexよりもオーバーヘッドが大きいらしいです。

非同期ランタイムを使った非同期プログラミング

Rustでは標準の非同期ランタイムは提供されていません。
そのため、非同期ランタイムを使った非同期プログラミングを行う場合は、tokioやasync-stdなどの非同期ランタイムを使う必要があります。

この辺りは正直詳しく無いのですが、プロジェクトの規模や目的によって使い分けると良いようです。

ここではtokioを使ってMutexを使った非同期プログラミングを行ってみましょう。

use std::sync::Arc;
use tokio::{sync::Mutex, task};

#[tokio::main]
async fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = task::spawn(async move {
            let mut num = counter.lock().await;
            *num += 1;
        });
        handles.push(handle);
    }

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

    println!("Result: {}", *counter.lock().await);
}

tokio::mainのマクロで非同期関数を実行することが出来ます。

tokio::sync::Mutexを使って先ほど同様にスレッド間でデータを共有しています。
lockメソッドは非同期関数であるため、awaitを使ってロックを取得しています。

先ほどの例とあまり変わりませんが、非同期ランタイムでも同様のことが出来るということですね。

まとめ

Rustにおいての非同期プログラミングはその厳密なライフタイム管理やデータ競合を防ぐための構造体など、まるでデータベースを設計するような感覚でプログラミングを行う必要があります。

確かにこれらのことを意識させる設計になっていることで、バグの少ないコードが書けるという点においてとても良いと思います。

今回は非同期プログラミングの基本的な部分のみを紹介しましたが、恐らく実際にプロジェクトで使う際にはフレームワークなどを利用して処理を記述していくかと思います。
そんな時に基礎的な部分としてこんな感じの処理があるんだなぁーって覚えておけば、フレームワークを使う際にもスムーズに理解できるのかなーと思っています。

今度RustでCMSなど作ってみたいなーと思っているので、その時はまた紹介したいと思います。

それでは。