並行性

理解

並行・並列処理は、メッセージパッシング（Message-passing）または状態共有（Shared-state）で実現できます。

スレッド

並行処理では次のに注意が必要です。

• データ競合

  スレッドがデータやリソースへ一貫しない順序でアクセスする競合状態。

• デッドロック

  2 つのスレッドが互いに相手の保持するリソースの使終了を待ち続け、どちらも進めなくなる状態。

• 現しにくいバグ

  特定の状況でだけ発生し、安定して現・修正するのが難しいバグ。

Green Thread はプログラミング言語レベルの thread であり、OS が提供する thread 機とは完全には同じではありません。

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {  // 注意这里的闭包如果需要入参，需要显式声明 move
        for i in 1..10 {
            println!("hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(1));  // 此时操作系统有机会切换执行其他线程
        }
    });

    handle.join().unwrap();  // 使用 join 等待所有线程执行结束（否则后续代码不执行）

    for i in 1..5 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }
}

メッセージパッシング

• Channel の基本的な書き方

  use std::sync::mpsc;  // multiple producer single consumer
  
  fn main() {
      let (tx, rx) = mpsc::channel();  // transmitter, receiver
  
  		thread::spawn(move || {
          let val = String::from("hi");
          tx.send(val).unwrap();  // 此时 val moved ownership
      });
  
  		let received = rx.recv().unwrap();  // 如果使用 try_recv 则不会阻塞
      println!("Got: {}", received);
  }
  

• 複数 producer/transmitter と複数メッセージの受信

  use std::sync::mpsc;
  use std::thread;
  use std::time::Duration;
  
  fn main() {
      let (tx, rx) = mpsc::channel();
  
  		// 因为后续闭包内会 move tx 的 ownership
      // 所以这里需要提前 clone 出 tx1 实现 multi producer
  		let tx1 = tx.clone();
  		thread::spawn(move || {
          let vals = vec![
              String::from("hi"),
              String::from("from"),
              String::from("the"),
              String::from("thread"),
          ];
  
          for val in vals {
              tx1.send(val).unwrap();
              thread::sleep(Duration::from_secs(1));
          }
      });
  
      thread::spawn(move || {
          let vals = vec![
              String::from("more"),
              String::from("messages"),
              String::from("for"),
              String::from("you"),
          ];
  
          for val in vals {
              tx.send(val).unwrap();
              thread::sleep(Duration::from_secs(1));
          }
      });
  
  		// 注意较于上例，使用了 for-in loop
      for received in rx {  
          println!("Got: {}", received);
      }
  }
  

状態共有

Mutex（Mutual Exclusion、相互排他）の一例は、同時に 1 スレッドだけが特定データへアクセスできることです。

スレッドは mutex のデータへアクセスする前に lock を取得する必要があります。lock は mutex の一部であり、誰がデータへ排他的にアクセスできるかを追跡します。そのため mutex は、ロック機構で保持データを保護すると説明されます。

Mutex は次の 2 つのルールがあるため扱いが難しいものです。

• データを使う前にロック取得を試みる必要があります。

  You must attempt to acquire the lock before using the data.

• mutex が保護するデータの使後は、他のスレッドがロックを取得できるよう解放しなければなりません。

  When you’re done with the data that the mutex guards, you must unlock the data so other threads can acquire the lock.

mutex の管理は難しいため、channel を好む人も多くいます。しかし Rust の型システムと有権ルールにより、ロックを誤って取得・解放することはありません。

• Mutex の基本的な書き方

  use std::sync::Mutex;
  
  fn main() {
      let m = Mutex::new(5);
  
      {
          let mut num = m.lock().unwrap();  // 调用 lock 获取锁
          *num = 6;
      }
  
      println!("m = {:?}", m);
  }
  

• 複数スレッドで Mutex を共有する

  use std::sync::{Arc, Mutex};
  use std::thread;
  
  fn main() {
      let counter = Arc::new(Mutex::new(0));
      let mut handles = vec![];
  
      for _ in 0..10 {
          let counter = Arc::clone(&counter);
          let handle = 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());
  }
  

Send と Sync trait

Send trait を実装する型は、ownership をスレッド間で移せます。

Sync trait を実装する型は、複数スレッドからアクセスできます。

Sync は Send の性質を含むことが分かります。

参照元

https://doc.rust-lang.org/book/ch16-00-concurrency.html

…

Arc<T> によるアトミック参照カウント

Fortunately, Arc<T> is a type like Rc<T> that is safe to use in concurrent situations. The a stands for atomic, meaning it’s an atomically reference counted type. Atomics are an additional kind of concurrency primitive that we won’t cover in detail here: see the standard library documentation for [std::sync::atomic](https://doc.rust-lang.org/std/sync/atomic/index.html) for more details. At this point, you just need to know that atomics work like primitive types but are safe to share across threads.

では、なぜすべてのプリミティブ型が atomic ではないのか疑問に思うかもしれません。 and why standard library types aren’t implemented to use Arc<T> by default. The reason is that thread safety comes with a performance penalty that you only want to pay when you really need to. If you’re just performing operations on values within a single thread, your code can run faster if it doesn’t have to enforce the guarantees atomics provide.

例に戻りましょう。 Arc<T> and Rc<T> have the same API, so we fix our program by changing the use line, the call to new, and the call to clone. The code in Listing 16-15 will finally compile and run:

…