Rust Message Passing – DECODE X NET

以前Go言語のチャンネルによる並行処理間でのメッセージ通信を試したことがありました。

Go-routine Channel

ここでは最近力を入れているRust言語でも同様の通信をしたいと思います。
参考) https://doc.rust-jp.rs/book-ja/ch16-02-message-passing.html

導入などは以下参照

Rust Programming Language

今回こだわったのは、送信スレッド2、受信スレッド2のたすき掛けの通信です。

use std::thread;
use std::sync::mpsc;
use std::time::Duration;

fn main() {
    let (tx11, rx1) = mpsc::channel();
    let (tx21, rx2) = mpsc::channel();

    let tx12 = mpsc::Sender::clone(&tx11);
    let tx22 = mpsc::Sender::clone(&tx21);
    let handle1 = thread::spawn(move || {
        let vals = vec!["A1", "A2", "A3"];
        for val in vals {
            tx11.send(val).unwrap();
            tx21.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });
    let handle2 = thread::spawn(move || {
        let vals = vec!["B1", "B2", "B3"];
        for val in vals {
            tx12.send(val).unwrap();
            tx22.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });
    
    thread::spawn(move || {
        for received in rx1 {
            println!("Got A: {}", received);
        }
    });
    thread::spawn(move || {
        for received in rx2 {
            println!("Got B: {}", received);
        }
    });
    handle1.join().unwrap();
    handle2.join().unwrap();
}

use std::thread;

use std::sync::mpsc;

use std::time::Duration;

fn main() {

let (tx11, rx1) = mpsc::channel();

let (tx21, rx2) = mpsc::channel();

let tx12 = mpsc::Sender::clone(&tx11);

let tx22 = mpsc::Sender::clone(&tx21);

let handle1 = thread::spawn(move || {

let vals = vec!["A1", "A2", "A3"];

for val in vals {

tx11.send(val).unwrap();

tx21.send(val).unwrap();

thread::sleep(Duration::from_secs(1));

}

});

let handle2 = thread::spawn(move || {

let vals = vec!["B1", "B2", "B3"];

for val in vals {

tx12.send(val).unwrap();

tx22.send(val).unwrap();

thread::sleep(Duration::from_secs(1));

}

});

thread::spawn(move || {

for received in rx1 {

println!("Got A: {}", received);

}

});

thread::spawn(move || {

for received in rx2 {

println!("Got B: {}", received);

}

});

handle1.join().unwrap();

handle2.join().unwrap();

}

実行結果

Got A: A1
Got A: B1
Got B: A1
Got B: B1
Got A: B2
Got B: A2
Got A: A2
Got B: B2
Got A: A3
Got B: B3
Got A: B3
Got B: A3

一秒おきに4行ずつ表示されます。4行の中の順番は同じにならないかもしれません。

メッセージを以下のように作成すると2回目送信のval変数がコンパイルエラーとなります。

let vals = vec![
String::from(“A1”),
String::from(“A2”),
String::from(“A3”),
];

これは、String型がムーブセマンティクスでsendを実行するとval変数の所有権の移動が生じるためです。
整数リテラルなどコピーセマンティクスは、所有権がコピーされます。
ムーブセマンティクスもCopyトレイとを実装すればコピーできるようですが、このあたり簡単に変数の再利用ができないようになっています。
これが安全性を高めるRustの特徴的のところです。

上記参考サイト冒頭に、

人気度を増してきている安全な並行性を保証する一つのアプローチがメッセージ受け渡しで、スレッドやアクターがデータを含むメッセージを相互に送り合うことでやり取りします。こちらが、Go言語のドキュメンテーションのスローガンにある考えです: 「メモリを共有することでやり取りするな; 代わりにやり取りすることでメモリを共有しろ」

とありますが、Channelという名前を使ったりするところからも、Go言語を意識した形跡もみられます。