<style> /* basic design */ .reveal h1, .reveal h2, .reveal h3, .reveal h4, .reveal h5, .reveal h6, .reveal section, .reveal table, .reveal li, .reveal blockquote, .reveal th, .reveal td, .reveal p { font-family: 'Meiryo UI', 'Source Sans Pro', Helvetica, sans-serif, 'Helvetica Neue', 'Helvetica', 'Arial', 'Hiragino Sans', 'ヒラギノ角ゴシック', YuGothic, 'Yu Gothic'; text-align: left; line-height: 1.6; letter-spacing: normal; text-shadow: none; word-wrap: break-word; color: #444; } .reveal h1, .reveal h2, .reveal h3, .reveal h4, .reveal h5, .reveal h6 {font-weight: bold;} .reveal h1, .reveal h2, .reveal h3 {color: #2980b9;} .reveal th {background: #DDD;} .reveal section img {background:none; border:none; box-shadow:none; max-width: 95%; max-height: 95%;} .reveal blockquote {width: 90%; padding: 0.5vw 3.0vw;} .reveal table {margin: 1.0vw auto;} .reveal code {line-height: 1.2;} .reveal p, .reveal li {padding: 0vw; margin: 0vw;} .reveal .box {margin: -0.5vw 1.5vw 2.0vw -1.5vw; padding: 0.5vw 1.5vw 0.5vw 1.5vw; background: #EEE; border-radius: 1.5vw;} /* table design */ .reveal table {background: #f5f5f5;} .reveal th {background: #444; color: #fff;} .reveal td {position: relative; transition: all 300ms;} .reveal tbody:hover td { color: transparent; text-shadow: 0 0 3px #aaa;} .reveal tbody:hover tr:hover td {color: #444; text-shadow: 0 1px 0 #fff;} /* blockquote design */ .reveal blockquote { width: 90%; padding: 0.5vw 0 0.5vw 6.0vw; font-style: italic; background: #f5f5f5; } .reveal blockquote:before{ position: absolute; top: 0.1vw; left: 1vw; content: "\f10d"; font-family: FontAwesome; color: #2980b9; font-size: 3.0vw; } /* font size */ .reveal h1 {font-size: 5.0vw;} .reveal h2 {font-size: 4.0vw;} .reveal h3 {font-size: 2.8vw;} .reveal h4 {font-size: 2.6vw;} .reveal h5 {font-size: 2.4vw;} .reveal h6 {font-size: 2.2vw;} .reveal section, .reveal table, .reveal li, .reveal blockquote, .reveal th, .reveal td, .reveal p {font-size: 2.2vw;} .reveal code {font-size: 1.6vw;} /* new color */ .red {color: #EE6557;} .blue {color: #16A6B6;} /* split slide */ #right {left: -18.33%; text-align: left; float: left; width: 50%; z-index: -10;} #left {left: 31.25%; text-align: left; float: left; width: 50%; z-index: -10;} </style> <style> /* specific design */ .reveal h2 { padding: 0 1.5vw; margin: 0.0vw 0 2.0vw -2.0vw; border-left: solid 1.2vw #2980b9; border-bottom: solid 0.8vw #d7d7d7; } </style> # 第16回　 ## 第16章 恐れるな！並行性 ### 2020/11/25 原山和之 --- ## 本日やること - スレッドを生成して、複数のコードを同時に走らせる方法 - チャンネルがスレッド間でメッセージを送るメッセージ受け渡し並行性 - 複数のスレッドが何らかのデータにアクセスする状態共有並行性 - 標準ライブラリが提供する型だけでなく、ユーザが定義した型に対してもRustの並行性の安全保証を拡張するSyncとSendトレイト --- ## スレッドを使用してコードを同時に走らせる - スレッドがデータやリソースに矛盾した順番でアクセスする競合状態 - 2つのスレッドがお互いにもう一方が持っているリソースを使用し終わるのを待ち、両者が継続するのを防ぐデッドロック - 特定の状況でのみ起き、確実な再現や修正が困難なバグ --- ## 1つのOSスレッドに対して1つの言語スレッドは プログラミング言語によってスレッドはいくつかの方法で実装されています。多くのOSで、新規スレッドを生成するAPIが提供されています。 言語がOSのAPIを呼び出してスレッドを生成するこのモデルを時に1:1と呼びます。 --- ## グリーンスレッド グリーンスレッドを使用する言語は、それを異なる数のOSスレッドの文脈で実行します。 このため、グリーンスレッドのモデルはM:Nモデルと呼ばれます。 M個のグリーンスレッドに対して、 N個のOSスレッドがあり、MとNは必ずしも同じ数字ではありません。 --- ## ランタイムの意味 - 言語によって全てのバイナリに含まれるコードのことを意味します。 - 口語的に誰かが「ノーランタイム」と言ったら、「小さいランタイム」のことを意味することがしばしばあります。 ランタイムが小さいと機能も少ないですが、バイナリのサイズも小さくなるという利点があります。 - 多くの言語では、 より多くの機能と引き換えにランタイムのサイズが膨れ上がるのは、受け入れられることですが、 Rustにはほとんどゼロのランタイムが必要でパフォーマンスを維持するためにCコードを呼び出せることを妥協できないのです。 --- ## Rustでは - Rustの標準ライブラリは、1:1スレッドの実装のみを提供しています。 - M:Nのグリーンスレッドモデルは、スレッドを管理するのにより大きな言語ランタイムが必要です。 - M:Nのグリーンスレッドモデルを実装したクレートがある。 --- ## 新規スレッドを生成例 ```rust= use std::thread; use std::time::Duration; fn main() { thread::spawn(|| { for i in 1..10 { // やあ！立ち上げたスレッドから数字{}だよ！ println!("hi number {} from the spawned thread!", i); thread::sleep(Duration::from_millis(1)); } }); for i in 1..5 { // メインスレッドから数字{}だよ！ println!("hi number {} from the main thread!", i); thread::sleep(Duration::from_millis(1)); } } ``` --- ## 生成結果 ```bash= hi number 1 from the main thread! hi number 1 from the spawned thread! hi number 2 from the main thread! hi number 2 from the spawned thread! hi number 3 from the main thread! hi number 3 from the spawned thread! hi number 4 from the main thread! hi number 4 from the spawned thread! hi number 5 from the spawned thread! ``` <!--- thread::sleepを呼び出すと、少々の間、スレッドの実行を止め、違うスレッドを走らせることができます。 スレッドはおそらく切り替わるでしょうが、保証はありません: OSがスレッドのスケジュールを行う方法によります。 この実行では、コード上では立ち上げられたスレッドのprint文が先に現れているのに、メインスレッドが先に出力しています。また、 立ち上げたスレッドにはiが9になるまで出力するよう指示しているのに、メインスレッドが終了する前の5までしか到達していません。 --> --- ## joinハンドルで全スレッドの終了を待つ <!-- コードは、メインスレッドが終了するためにほとんどの場合、立ち上げたスレッドがすべて実行されないだけでなく、 立ち上げたスレッドが実行されるかどうかも保証できません。原因は、スレッドの実行順に保証がないからです。 --> ```rust= use std::thread; use std::time::Duration; fn main() { let handle = thread::spawn(|| { for i in 1..10 { println!("hi number {} from the spawned thread!", i); thread::sleep(Duration::from_millis(1)); } }); for i in 1..5 { println!("hi number {} from the main thread!", i); thread::sleep(Duration::from_millis(1)); } handle.join().unwrap(); } ``` <!-- thread:spawnの戻り値の型はJoinHandleです。 JoinHandleは、そのjoinメソッドを呼び出したときにスレッドの終了を待つ所有された値です。立ち上げたスレッドが確実に完了する方法を示しています。 --> --- ## Joinの結果 ```bash= hi number 1 from the spawned thread! hi number 2 from the spawned thread! hi number 3 from the spawned thread! hi number 4 from the spawned thread! hi number 5 from the spawned thread! hi number 6 from the spawned thread! hi number 7 from the spawned thread! hi number 8 from the spawned thread! hi number 9 from the spawned thread! hi number 1 from the main thread! hi number 2 from the main thread! hi number 3 from the main thread! hi number 4 from the main thread! ``` --- ## スレッドでmoveクロージャを使用する - moveクロージャは、thread::spawnとともによく使用されます。 あるスレッドのデータを別のスレッドで使用できるようになるからです。 <!-- クロージャに使用している値の所有権を強制的に奪わせます。 --> ```rust= use std::thread; fn main() { let v = vec![1, 2, 3]; let handle = thread::spawn(move || { println!("Here's a vector: {:?}", v); }); handle.join().unwrap(); } ``` --- ## メッセージ受け渡しを使ってスレッド間でデータを転送する - プログラミングにおけるチャンネルは、2分割できます: 転送機と受信機です。転送機はアヒルのおもちゃを川に置く上流になり、 受信機は、アヒルのおもちゃが行き着く下流になります... ```rust= use std::sync::mpsc; fn main() { let (tx, rx) = mpsc::channel(); } ``` --- ## mpsc::channel関数 - mpscはmultiple producer, single consumerを表しています。 - 1つのチャンネルが値を生成する複数の送信側と、 その値を消費するたった1つの受信側を持つことができるということを意味します。 - mpsc::channel関数はタプルを返し、1つ目の要素は、送信側、2つ目の要素は受信側になります。 - txとrxという略称は、多くの分野で伝統的に転送機と受信機 - let文を使うと、 mp - mpsc::channelで返ってくるタプルの部品を抽出するのが便利になります。 --- ## やりとり例 <!-- --> ```rust= use std::thread; use std::sync::mpsc; fn main() { let (tx, rx) = mpsc::channel(); thread::spawn(move || { let val = String::from("hi"); tx.send(val).unwrap(); }); let received = rx.recv().unwrap(); // 値は{}です println!("Got: {}", received); } ``` --- ## チャンネルと所有権の転送 <!-- --> ```rust= use std::thread; use std::sync::mpsc; fn main() { let (tx, rx) = mpsc::channel(); thread::spawn(move || { let val = String::from("hi"); tx.send(val).unwrap(); // valは{} println!("val is {}", val); }); let received = rx.recv().unwrap(); println!("Got: {}", received); } ``` --- ## 複数の値を送信し、受信側が待機するのを確かめる <!-- --> ```rust= use std::thread; use std::sync::mpsc; use std::time::Duration; fn main() { let (tx, rx) = mpsc::channel(); thread::spawn(move || { // スレッドからやあ(hi from the thread) let vals = vec![ String::from("hi"), String::from("from"), String::from("the"), String::from("thread"), ]; for val in vals { tx.send(val).unwrap(); thread::sleep(Duration::from_secs(1)); } }); for received in rx { println!("Got: {}", received); } } ``` --- ## 転送機をクローンして複数の生成器を作成する <!-- --> ```rust= let (tx, rx) = mpsc::channel(); let tx1 = mpsc::Sender::clone(&tx); 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 || { // 君のためにもっとメッセージを(more messages for you) 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 received in rx { println!("Got: {}", received); } ``` --- ## 状態共有並行性 - メモリ共有並行性は、複数の所有権に似ています: 複数のスレッドが同時に同じメモリ位置にアクセスできるのです。 --- ## ミューテックス - どんな時も1つのスレッドにしかなんらかのデータへのアクセスを許可しないというように、 "mutual exclusion"(相互排他)の省略形です。 - ミューテックスにあるデータにアクセスする = ミューテックスのロック - ミューテックスのロック = データを死守する(guarding) --- ## ミューテックスは、2つの規則 - データを使用する前にロックの獲得を試みなければならない。 - ミューテックスが死守しているデータの使用が終わったら、他のスレッドがロックを獲得できるように、 データをアンロックしなければならない。 <!-- ミューテックスを現実世界の物で例えるなら、マイクが1つしかない会議のパネルディスカッションを思い浮かべてください。 パネリストが発言できる前に、マイクを使用したいと申し出たり、通知しなければなりません。マイクを受け取ったら、 話したいだけ話し、それから次に発言を申し出たパネリストにマイクを手渡します。 --> --- ## Mutex<T\>のAPI - まずはシングルスレッド ```rust= use std::sync::Mutex; fn main() { let m = Mutex::new(5); { let mut num = m.lock().unwrap(); *num = 6; } println!("m = {:?}", m); } ``` <!-- Mutex<T>APIオブジェクトを獲得して、lockメソッドでロックを獲得する。 これは現在のスレッドをブロックする。可変参照numを逆参照で値自体にアクセスして6という値を入れます。lock呼び出しが、MutexGuardというスマートポインタを返却しているようです。スコープを外れたときにロックの解除が自動的に行われるようです。 --> --- ## 複数のスレッド間でMutex<T\>を共有する - 10個のスレッド...エラーになる... ```rust= use std::sync::Mutex; use std::thread; fn main() { let counter = Mutex::new(0); let mut handles = vec![]; for _ in 0..10 { 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()); } ``` --- ## 修正 - 2つのスレッドを作成してやってみる...エラー... ```rust= use std::sync::Mutex; use std::thread; fn main() { let counter = Mutex::new(0); let mut handles = vec![]; let handle = thread::spawn(move || { let mut num = counter.lock().unwrap(); *num += 1; }); handles.push(handle); let handle2 = thread::spawn(move || { let mut num2 = counter.lock().unwrap(); *num2 += 1; }); handles.push(handle2); for handle in handles { handle.join().unwrap(); } println!("Result: {}", *counter.lock().unwrap()); } ``` --- ## さらに修正 ... Rc<T\>使う - 複数のスレッドで複数の所有権 ```rust= use std::rc::Rc; use std::sync::Mutex; use std::thread; fn main() { let counter = Rc::new(Mutex::new(0)); let mut handles = vec![]; for _ in 0..10 { let counter = Rc::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()); } ``` <!-- Rc<T>はスレッド間で共有するには安全ではないのです。Rc<T>が参照カウントを管理する際、 cloneが呼び出されるたびにカウントを追加し、クローンがドロップされるたびにカウントを差し引きます。 しかし、並行基本型を使用してカウントの変更が別のスレッドに妨害されないことを確認していないのです。 これは間違ったカウントにつながる可能性があり、今度はメモリリークや、使用し終わる前に値がドロップされることにつながる可能性のある潜在的なバグです。 --> --- ## Arc<T\>で原子的な参照カウント ```rust= use std::sync::{Mutex, Arc}; 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()); } ``` --- ## RefCell<T\>/Rc<T\>とMutex<T\>/Arc<T\>の類似性 counterは不変なのに、その内部にある値への可変参照を得ることができたことに気付いたでしょうか; つまり、Mutex<T\>は、Cell系のように内部可変性を提供するわけです。 第15章でRefCell<T\>を使用してRc<T\>の内容を可変化できるようにしたのと同様に、 Mutex<T\>を使用してArc<T\>の内容を可変化しているのです。 <!-- 気付いておくべき別の詳細は、Mutex<T>を使用する際にあらゆる種類のロジックエラーからは、 コンパイラは保護してくれないということです。第15章でRc<T>は、循環参照を生成してしまうリスクを伴い、 そうすると、2つのRc<T>の値がお互いを参照し合い、メモリリークを引き起こしてしまうことを思い出してください。 同様に、Mutex<T>はデッドロックを生成するリスクを伴っています。これは、処理が2つのリソースをロックする必要があり、 2つのスレッドがそれぞれにロックを1つ獲得して永久にお互いを待ちあってしまうときに起こります。 デッドロックに興味があるのなら、デッドロックのあるRustプログラムを組んでみてください; --> --- ## SyncとSendトレイトで拡張可能な並行性 - Rust言語には、寡少な並行性機能があります。 - Sendでスレッド間の所有権の転送を許可する。 - Syncで複数のスレッドからのアクセスを許可する。 - SendとSyncを手動で実装するのは非安全である。 ---