Rust Mutex構造体

Rustのstd::sync::Mutexのメモ。

概要

Mutex構造体は、排他制御を行う構造体。

use std::sync::Mutex;

Mutexには、排他して使用したい値を指定して使う。

    let mutex: Mutex<Vec<i32>> = Mutex::new(Vec::new());

    {
        let mut vec = mutex.lock().unwrap();
        vec.push(123);
    }
    {
        let vec = mutex.lock().unwrap();
        println!("{:?}", vec);
    }

Mutexのロック操作は不変

Mutexのlockメソッドやtry_lockメソッドは不変メソッドである。（ 第1引数が&selfであって&mut selfではない）[2024-10-07]
そして、不変のくせに、獲得したロックインスタンスに対して変更する操作が出来る！

つまり、Mutex内部の値を書き換えるメソッドであっても、不変メソッドにすることが出来る。

struct MyStruct {
    value: Mutex<i32>,
}

impl MyStruct {
    fn new() -> MyStruct {
        MyStruct {
            value: Mutex::new(0),
        }
    }

    fn add(&self, v: i32) {	// addメソッドの中でselfフィールドの値を変更しているにも関わらず、addメソッド自体は不変メソッド（引数の型が&self）である
        let mut value = self.value.lock().unwrap();	// 変更対象の値の方はmutにする
        *value += v;
    }

    fn get(&self) -> i32 {
        let value = self.value.lock().unwrap();
        *value
    }
}

この場合、MyStructの各メソッドできちんと排他制御されており、呼び出すのは不変メソッドのみということになるので、Arcで直接MyStructを管理できる。
MyStructを複数スレッドに渡すときも、MyStruct自身をMutexで囲む必要は無い。

fn main() {
    let s = Arc::new(MyStruct::new());	// Arc::new(Mutex::new(MyStruct::new()))にする必要は無い

    let mut handles = vec![];

    for _ in 0..10 {
        let s = Arc::clone(&s);
        let handle = thread::spawn(move || {
            for i in 1..=100 {
                s.add(i);
            }
        });
        handles.push(handle);
    }

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

    println!("{}", s.get());
}

Mutexのメソッド

Mutexのメソッド（抜粋）。

Rustの標準ライブラリー（stdクレート）のMutexには、指定時間待つtry_lock()は無い。

parking_lotという外部クレートには、指定された時間待つtry_lock_for()や指定時刻まで待つtry_lock_until()といったメソッドがあるらしい。

Mutexの例

構造体のフィールドでMutexを持つ例。
フィールドにアクセスするメソッドがマルチスレッドで呼ばれる想定。

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

struct MyCounter {
    counter: Mutex<i32>,
}

impl MyCounter {
    fn new() -> MyCounter {
        MyCounter {
            counter: Mutex::new(0),
        }
    }

    fn increment(&self) {	// このメソッドが複数スレッドから呼ばれる
        let mut counter = self.counter.lock().unwrap();
        *counter += 1;
    }

    fn get(&self) -> i32 {
        let counter = self.counter.lock().unwrap();
        *counter
    }
}

この例では、フィールドのcounterがMutexになっている。
（→カウンターをAtomicI32で実装する例）

counterフィールドにアクセスするincrementメソッドやgetメソッドでは、Mutexのlock()でロックを取っている。

lock()はロックが獲得できるまで待つ。
lock()の返り値はResultで、ロックが獲得できたらOkが返る。
Errが返るのは異常事態なので、アプリケーション側では対処不能である。
したがって、Resultから値を取得するのにunwrap()を使っている。（失敗していたらアプリケーションがパニック（異常終了）する）

lock().unwrap()で、MutexGuardという型のロックインスタンスが取得できる。
MutexGuardはDerefトレイトを実装しているので、MutexGuard内部の型として扱うコーディングが出来る。

ロックインスタンス（MutexGuard）がスコープから外れると、ロックが解除される。

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

    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            for _ in 0..1000 {
                counter.increment();
            }
        });
        handles.push(handle);
    }

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

    println!("{}", counter.get());
}

この例のmain関数では、スレッドを10個起こし、各スレッドでは1000回Mycounterのincrementメソッドを呼んでいる。
このカウンターはちゃんとMutexにより排他されているので、最終的にカウントは10000になる。

try_lockの例

上記のMyCounver.increment()をMutexのtry_lock()に置き換えてみた。

    fn increment(&self) {
        loop {
            let counter = self.counter.try_lock();
            if let Ok(mut counter) = counter {
                *counter += 1;
                break;
            }

            println!("try_lock fail");
        }
    }

Mutexのtry_lock()は、ロックが獲得できたらResultのOkを返し、獲得できなかったら即座にErrを返す。

10スレッドで各スレッドが1000回increment()を呼ぶ程度では、10回前後しかfailしなかった。

Mutexの使用上の注意点

lock()は連続して呼び出せない

Mutexのロックを獲得した後、アンロックせずに再び同じMutexのlock()を呼び出すことは出来ない。
というか、呼び出すと永遠に待ち状態になる。

    let mutex = Mutex::new(0);
    let lock1 = mutex.lock().unwrap();	// ロック獲得成功
    let lock2 = mutex.lock().unwrap();	// ロック獲得待ち（lock1が解放されることが無いので、永遠に待つ）

（JavaのsynchronizedはMutexでなくMonitorによるロックなので、ロック済みのロックオブジェクトに対して同一スレッド内で再度ロックしようとしても、待ち状態になることは無い）

例えば、あるメソッドでロックを獲得した後、サブルーチンのメソッドを呼び出して、その中でもロックを獲得しようとすると、永遠に待ち状態になってしまうということだ。

struct LockExample {
    mutex: Mutex<i32>,
}

impl LockExample {
    fn lock1(&self) {
        println!("lock1 start");

        let _lock = self.mutex.lock().unwrap();	// ロック獲得
        self.lock2();

        println!("lock1 end");
    }

    fn lock2(&self) {
        println!("lock2 start");

        let _lock = self.mutex.lock().unwrap();	// lock1()でロック獲得済みなので、永遠に待つ

        println!("lock2 end");
    }
}

fn main() {
    let example = LockExample {
        mutex: Mutex::new(0),
    };
    example.lock1();
}

Mutex::new()に渡す値について

Mutex::new()に同じ値を渡しても、異なるMutexとして扱われる。

    let mutex1 = Mutex::new(0);
    let mutex2 = Mutex::new(0);
    let lock1 = mutex1.lock().unwrap();	// mutex1のロックを獲得成功
    let lock2 = mutex2.lock().unwrap();	// mutex1とmutex2は異なるロックなので、mutex2のロックを獲得できる

Mutex::new()の内部では、受け取った値をUnsafeCellという構造体でラップして管理しているようだ。
ロック対象インスタンスはUnsafeCellで識別されるようなので、元が同じ値でもロック対象としては別物ということになる。

別の言い方をすると、初期値で与えた値を変更しても、Mutexのロック対象は変わらない。

Mutex内のi32を変更する例

    let mutex = Mutex::new(0);

    {
        let mut n = mutex.lock().unwrap();
        *n += 1;				// 中の値を変更
        assert_eq!(1, *n);
    }

    let opt = mutex.lock().unwrap();	// 中の値は変わっているが、ちゃんとロックできる
    assert_eq!(1, *opt);

→AtomicI32

Mutex内のOptionを変更する例

    let mutex = Mutex::new(Some(123));

    {
        let mut opt = mutex.lock().unwrap();
        let v = opt.take();	// 中身を取り出してNoneに変える
        assert_eq!(Some(123), v);
        assert_eq!(None, *opt);
    }
    {
        let mut opt = mutex.lock().unwrap();	// 中の値は変わっているが、ちゃんとロックできる
        let v = opt.insert(456);
        assert_eq!(Some(456), *opt);
    }
    {
        let mut opt = mutex.lock().unwrap();	// 中の値は変わっているが、ちゃんとロックできる
        *opt = Some(789);
        assert_eq!(Some(789), *opt);
    }

    let opt = mutex.lock().unwrap();	// 中の値は変わっているが、ちゃんとロックできる
    assert_eq!(Some(789), *opt);

Mutex内のVecを変更する例

    let mutex = Mutex::new(vec![1, 2, 3]);

    {
        let mut vec = mutex.lock().unwrap();
        vec.push(4);				// 中身に値を追加
        assert_eq!(vec![1, 2, 3, 4], *vec);
    }

    let vec = mutex.lock().unwrap();	// 中の値は変わっているが、ちゃんとロックできる
    assert_eq!(vec![1, 2, 3, 4], *vec);

Mutex内のVecをマルチスレッドで変更する例

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

struct MyVec {
    vec: Mutex<Vec<i32>>,
}

impl MyVec {
    fn new() -> MyVec {
        MyVec {
            vec: Mutex::new(Vec::new()),
        }
    }

    fn push(&self, value: i32) {
        let mut vec = self.vec.lock().unwrap();	// 中の値は変わっていくが、ちゃんとロックできる
        vec.push(value);
    }

    fn print(&self) {
        let vec = self.vec.lock().unwrap();
        println!("{:?}", *vec);
    }
}

fn main() {
    let my_vec = Arc::new(MyVec::new());

    let mut handles = vec![];

    for _ in 0..10 {
        let my_vec = Arc::clone(&my_vec);
        let handle = thread::spawn(move || {
            for i in 0..1000 {
                my_vec.push(i);
            }
        });
        handles.push(handle);
    }

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

    my_vec.print();
}