4.6.并發 · Rust Book中文翻譯

并發與并行是計算機科學中異常重要的兩個主題，并且在當今生產環境中也十分熱門。計算機正擁有越來越多的核心，然而很多程序員還沒有準備好去完全的利用它們。 Rust的內存安全功能也適用于并發環境。甚至并發的Rust程序也會是內存安全的，并且沒有數據競爭。Rust的類型系統也能勝任，并給你強大的能力在編譯時推論出并發代碼。在我們討論Rust提供的并發功能之前，理解一些問題是很重要的：Rust非常底層以至于所有這些都是由標準庫，而不是由語言提供的。這意味著如果你不喜歡一些Rust處理并發的方面，你可以自己實現一個。[mio](https://github.com/carllerche/mio)是現實生活中這個原則的實踐。 ## 背景：`Send`和`Sync` 并發難以推理。在Rust中，我們有一個強大，靜態類型系統來幫助我們推理我們的代碼。Rust自身提供了兩個特性來幫助我們理解可能是并發的代碼的意思。 ### `Send` 第一個我們要談到的特性是[Send](http://doc.rust-lang.org/std/marker/trait.Send.html)。當一個`T`類型實現了`Send`，它向編譯器表明這個類型的所有權可以在進程間安全的轉移。強制這個限制是很重要的。例如，我們有一個連接兩個線程的通道，我們想要能夠向通道發送些數據到另一個線程。因此，我們要確保這個類型實現了`Send`。想反，如果我們通過FFI封裝了一個不是線程安全的庫，我們并不想實現`Send`，并且因此編譯器會幫助我們強制它不會離開當前線程。 ### `Sync` 第二個特性是`Sync`。當一個類型`T`實現了`Sync`，它向編譯器表明這個類型在多線程并發時沒有導致內存不安全的可能性。例如，使用一個原子引用來共享可變數據是線程安全的。Rust提供了一個這樣的類型，`Arc`，并且它實現了`Sync`，所以它可以安全的在線程間共享。這兩個特性允許你使用類型系統來確保你代碼在并發環境的特性。在我們演示為什么之前，我們需要先學會如何創建一個并發Rust程序！ ## 線程 Rust標準庫提供了一個“線程”庫，它允許你并行的執行Rust代碼。這是一個使用`std::thread`的基本例子： ~~~ use std::thread; fn main() { thread::scoped(|| { println!("Hello from a thread!"); }); } ~~~ `Thread::scoped()`方法接受一個閉包，它將會在一個新線程中執行。它叫做`scoped`因為這個線程返回一個聯合守護（join guard）： ~~~ use std::thread; fn main() { let guard = thread::scoped(|| { println!("Hello from a thread!"); }); // guard goes out of scope here } ~~~ 當`guard`離開作用域，它會阻塞執行直到線程結束。如果我們不想要這個行為，我們可以使用`thread::spawn()`： ~~~ use std::thread; use std::old_io::timer; use std::time::Duration; fn main() { thread::spawn(|| { println!("Hello from a thread!"); }); timer::sleep(Duration::milliseconds(50)); } ~~~ 我們需要在這里`sleep`是因為當`main()`結束，它會殺死所有正在執行的線程。 [scoped](http://doc.rust-lang.org/std/thread/fn.scoped.html)有一個有意思的類型標記： ~~~ fn scoped<'a, T, F>(self, f: F) -> JoinGuard<'a, T> where T: Send + 'a, F: FnOnce() -> T, F: Send + 'a ~~~ 特別的，`F`，那個我們傳遞到新線程執行的閉包。它有兩個限制：它必須是一個從`()`到`T`的`FnOnce`。`FnOnce`允許這個閉包從父線程獲取任何它提到的數據。另一個限制是`F`必須實現`Send`。我們不能傳遞這個所有權除非這個類型認為這是可以的。很多語言有多線程執行的能力，不過是很不安全的。這里有完整的書籍關于如何避免在共享可變狀態下出現錯誤。Rust也用它的類型系統幫助我們，通過在編譯時避免數據競爭。讓我們具體討論下如何在線程間共享數據。 ## 安全的共享可變狀態（Safe Shared Mutable State）根據Rust的類型系統，我們有個聽起來很假的概念叫做：“安全的共享可變狀態”。很多程序猿都同意共享可變狀態是灰常，灰常壞的。有人曾說道： > 共享可變狀態是一切罪惡的根源。大部分語言嘗試解決這個問題的“可變”部分，而Rust則嘗試解決“共享”部分。同樣[所有權系統](http://kaisery.gitbooks.io/rust-book-chinese/content/content/5.8.Ownership%20%E6%89%80%E6%9C%89%E6%9D%83.md)也通過防止不當的使用指針來幫助我們排除數據競爭，最糟糕的并發bug之一。作為一個例子，這是一個在很多語言中會產生數據競爭的Rust版本程序。它不能編譯： ~~~ use std::thread; fn main() { let mut data = vec![1u32, 2, 3]; for i in 0..3 { thread::spawn(move || { data[i] += 1; }); } thread::sleep_ms(50); } ~~~ 這會給我們一個錯誤： ~~~ 8:17 error: capture of moved value: `data` data[i] += 1; ^~~~ ~~~ 在這個例子中：我們知道我們的代碼_應該_是安全的，不過Rust并不確定。并且它確實不安全：如果每個線程中都有一個`data`的引用，并且這些線程獲取了引用的所有權，我們就有了3個所有者！這是很糟糕的。我們可以用`Arc<T>`類型來修復它，它是一個原子引用計數的指針。“原子”部分是指它可以安全的跨線程共享。 `Arc<T>`假設它的內容有另一個屬性來確保它可以跨線程共享：它假設它的內容實現了`Sync`。不過在我們的例子中，我們想要可以改變它的值。我們需要一個同一時間只有一個人可以修改它的值的類型。為此，我們可以使用`Mutex<T>`類型。下面是我們代碼的第二版。它還是不能工作，不過是因為另外的原因： ~~~ use std::thread; use std::sync::Mutex; fn main() { let mut data = Mutex::new(vec![1u32, 2, 3]); for i in 0..3 { let data = data.lock().unwrap(); thread::spawn(move || { data[i] += 1; }); } thread::sleep_ms(50); } ~~~ 下面是錯誤： ~~~ :9:9: 9:22 error: the trait `core::marker::Send` is not implemented for the type `std::sync::mutex::MutexGuard>` [E0277] :11 thread::spawn(move || { ^~~~~~~~~~~~~ :9:9: 9:22 note: `std::sync::mutex::MutexGuard>` cannot be sent between threads safely :11 thread::spawn(move || { ^~~~~~~~~~~~~ ~~~ 你看，[Mutex](http://doc.rust-lang.org/std/sync/struct.Mutex.html)有一個有如下標記的[lock](http://doc.rust-lang.org/std/sync/struct.Mutex.html#method.lock)方法： ~~~ fn lock(&self) -> LockResult> ~~~ 因為`MutexGuard<T>`沒有實現`Send`，我們不能傳送守護穿過線程邊界，它給了我們這個錯誤。我們可以用`Arc<T>`修復這個問題。下面是一個可以工作的版本： ~~~ use std::sync::{Arc, Mutex}; use std::thread; fn main() { let data = Arc::new(Mutex::new(vec![1u32, 2, 3])); for i in 0..3 { let data = data.clone(); thread::spawn(move || { let mut data = data.lock().unwrap(); data[i] += 1; }); } thread::sleep_ms(50); } ~~~ 我們現在在`Arc`上調用`clone()`，它增加了其內部計數。這個句柄接著被移動到新線程。讓我們更仔細的檢查一個線程代碼： ~~~ thread::spawn(move || { let mut data = data.lock().unwrap(); data[i] += 1; }); ~~~ 首先，我們調用`lock()`，它獲取了互斥鎖。因為這可能失敗，它返回一個`Result<T, E>`，并且因為這僅僅是一個例子，我們`unwrap()`結果來獲得一個數據的引用。現實代碼在這里應該有更健壯的錯誤處理。下面我們可以隨意修改它，因為我們持有鎖。然而，定時器部分顯得有點奇怪。我們選擇等待了一個合理的時間，不過這也完全有可能我們等了太久，我們可以等更少的時間。也有可能我們等了太短，這樣我們并沒有實際完成計算。 Rust的標準庫提供了更多同步兩個線程的機制。讓我們看看其中一個：通道。 ## 通道（Channels）下面是我們代碼使用通道同步的版本，而不是等待特定時間： ~~~ use std::sync::{Arc, Mutex}; use std::thread; use std::sync::mpsc; fn main() { let data = Arc::new(Mutex::new(0u32)); let (tx, rx) = mpsc::channel(); for _ in 0..10 { let (data, tx) = (data.clone(), tx.clone()); thread::spawn(move || { let mut data = data.lock().unwrap(); *data += 1; tx.send(()); }); } for _ in 0..10 { rx.recv(); } } ~~~ 我們使用`mpsc::channel()`方法創建了一個新的通道。我們僅僅向通道中`send`了一個簡單的`()`，然后等待它們10個都返回。因為這個通道只是發送了一個通用信號，我們也可以通過通道發送任何實現了`Send`的數據！ ~~~ use std::thread; use std::sync::mpsc; fn main() { let (tx, rx) = mpsc::channel(); for _ in 0..10 { let tx = tx.clone(); thread::spawn(move || { let answer = 42u32; tx.send(answer); }); } rx.recv().ok().expect("Could not receive answer"); } ~~~ `u32`實現了`Send`因為我們可以拷貝它。所以我們創建了一個線程，讓它計算結果，然后通過通道`send()`給我們結果。 ## 恐慌（Panics） `panic!`會使當前執行線程崩潰。你可以使用Rust的線程來作為一個簡單的隔離機制： ~~~ use std::thread; let result = thread::spawn(move || { panic!("oops!"); }).join(); assert!(result.is_err()); ~~~ 我們的`Thread`返回一個`Result`,它允許我們檢查我們的線程是否發生了恐慌。