【译】Rust，无畏并发
- 2020 年 11 月 3 日
- 筆記
原文链接：//dev.to/imaculate3/fearless-concurrency-5fk8 >
原文标题：That’s so Rusty! Fearless concurrency
公众号：Rust 碎碎念
翻译： Praying
并发程序是指运行多个任务的程序（或看上去是多任务），即两个及以上的任务在重叠的时间跨度内运行。这些任务由线程——最小的处理单元执行。在其背后，并不完全是多任务（并行）处理，而是线程之间以普通人无法感知的速度进行上下文快速切换。很多现代应用程序都依赖于这种错觉，比如服务器可以在处理请求的同时等待其他请求。当线程间共享数据时可能会出很多问题，最常见的两种 bug 是：竞态条件和死锁。
-
竞态条件常发生于多个线程以不一致的顺序访问/修改共享数据。这对于那些必须以原子性执行的事务会造成很严重的影响。每次只能有一个线程访问共享数据是必要的，并且无论线程以何种顺序运行，程序都应该工作良好。
-
死锁发生于两个及以上的线程互相等待对方采取行动，由于资源的锁定，没有线程能够获取所需资源，从而导致无限挂起。比如，线程 1 需要两个资源，它已经获取了资源 1 的锁并在等待线程 2 释放资源 2，但是此时，线程 2 也在等待线程 1 释放资源 1，这个时候就会发生死锁。这种情况被称作循环等待。如果这些线程可以在等待获取锁的同时并持有资源，且不支持抢占，死锁就无法恢复。
在任何一门编程语言中要避免这些问题都需要大量的练习。当这些问题真的渗透到项目中，它们很难被检测出来。最好的情况下，这些 bug 表现为崩溃和挂起，最坏的情况下，程序将以不可预测的方式运行。换句话说，无畏并发的保证不可轻视。恰好，Rust 宣称可以提供这种保证，不是么？
Rust 并发（Rust Concurrency）
Rust 并发的主要构成是线程和闭包。
1. 闭包（Closures）
闭包是指能够访问在其所被定义的作用域内的变量的匿名函数。它们是 Rust 的函数式特性之一。它们可以被赋予变量，作为参数传递以及从函数中返回。它们的作用域仅限于局部变量，因此，不能暴露在 crate 之外。在语法上，除了没有名字之外，它们和函数非常相似，参数在竖线括号（||）中传递，类型标注是可选的。和函数不同的是，闭包能够从其被定义的作用域内访问变量。这些被捕获的变量可以以借用（borrow）或移动（move)的方式进入闭包，具体取决于变量的类型以及该变量如何在闭包中被使用。下面是一个闭包（被赋于print_greeting）的例子，该闭包接收一个参数，并且捕获变量generic_greeting。
fn main() {
let generic_greeting = String::from("Good day,");
let print_greeting = |name| println!("{} {}!", generic_greeting, name);
let person = String::from("Crab");
print_greeting(person);
// println!("Can I use generic greeting? {}", generic_greeting);
// println!("Can I use person {}", person);
}
在上面的例子中，变量person和generic_greeting都被移动（move）到闭包当中，因此，在调用闭包之后，它们就不能再被使用了。取消最后两行打印语句的注释，程序将无法编译。
2. 线程（Threads）
Rust 并发是通过生成（spawn）多个线程来实现的，这些线程运行处于无参数闭包中的不同任务。当一个线程被生成（spawn）时，返回类型时JoinHandle类型，除非JoinHandle被 joined，否则主线程不会等待它完成。因此，传递给线程的闭包必须拿到被捕获变量的所有权以确保在线程最终执行的时候，这些被捕获变量依然时有效的。这一点在下面的例子中有所体现。
use std::thread;
use std::time::Duration;
fn main() {
let t1 = thread::spawn(|| {
for i in 1..10 {
println!("Greeting {} from other thread!", i);
thread::sleep(Duration::from_millis(1));
}
});
for i in 1..5 {
println!("Greeting {} from main thread!", i);
thread::sleep(Duration::from_millis(1));
}
t1.join().unwrap();
}
我们得到了来自两个线程的完整输出，来自生成（spawn）线程的 9 个 greetings 和来自主线程的 4 个 greetings。
当注释掉t1.join().unwrap()时，程序会在主线程执行完成后退出。你可以在Rust playground[1]上进行尝试。
需要注意的是，多次运行程序得到的结果可能有所不同。这样的不确定性也是并发的特点之一，也正如我们将要看到的那样，这也是很多 bug 的根源。
就多任务处理而言，线程之间共享信息非常关键。在标准库中，Rust 支持了两种通信方式：消息传递（Message Passing）和共享状态（Shared-State）。
1. 消息传递（Message Passing）
Rust 支持 channel，线程可以通过 channel 来发送和接收消息。一个例子就是多生产者单消费者（缩写为mpsc）channel。这种 channel 允许多个发送方和单个接收方通信，这些发送方和接收方可能处于不同的线程中。channel 在发送方结尾处获取变量的所有权，并在接收方结尾处将其丢弃。下面的例子展示了消息是如何在两个发送方和一个接收方之间的 channel 传递的。
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx1, rx) = mpsc::channel();
let tx2 = mpsc::Sender::clone(&tx1);
thread::spawn(move || {
println!("Sending message from thread 1");
tx1.send(String::from("Greeting from thread 1")).unwrap();
});
thread::spawn(move || {
println!("Sending message from thread 2");
tx2.send(String::from("Greeting from thread 2")).unwrap();
});
for recvd in rx {
println!("Received: {}", recvd);
}
}
第二个发送方通过对第一个发送方克隆（clone）产生。这里不需要 join 这些线程，因为接收方会阻塞直到收到消息，在那里等待线程执行完成。
2. 共享状态（Shared state）
另一种通信方式是共享内存。消息传递固然很好，但是它受单个所有权限制。对象必须被移动（move）/克隆（clone）才能传送到另一个线程，如果对象被移动（move），它就变为不可用，如果它被克隆（clone)，在该对象上的任何更新都必须要通过消息传递来通信。解决这个问题的方案是多所有权（ multiple ownership），你可以回顾smartpointers post[2]这篇文章（译注： smartpointers 这篇文章已翻译，译文为Rust与智能指针），多所有权（ multiple ownership）通过引用计数智能指针Rc<T>来实现。我们看到，通过和RefCell<T>组合使用，我们可以创建可变的共享指针。为什么不在一个多线程程序中使用它们呢？下面是一个尝试：
fn main() {
let counter = Rc::new(RefCell::new(0));
let mut threads = vec![];
for _ in 0..5 {
let counter = Rc::clone(&counter);
let handle = thread::spawn(move || {
let mut num = (*counter).borrow_mut();
*num += 1;
});
threads.push(handle);
}
for thread in threads {
thread.join().unwrap();
}
println!("Result: {}", *counter.borrow());
}
counter的共享指针（ shared pointer ）被发送到 5 个线程，每个线程都将其增加 1，在线程结束后打印出这个数字。上面这段代码能运行么？
这段代码无法编译，但是错误信息给出了提示。Rc<RefCell<T>>不能在线程间安全地被传递。（上图中）再往下就是原因了：它没有实现Send trait。看起来我们的意图已经被正确地表达了，但是我们需要这些指针的线程安全版本。安全版本的替代选项是否存在？确实，Rc<T>的替代选项是原子引用计数类型Arc<T>。除了Rc<T>的属性（共享所有权）外，它还可以通过实现了Send trait 在线程间安全地共享。对上面的代码进行替换，我们应该可以能够更接近正确的运行状态。下面的代码可以编译嘛？
use std::cell::RefCell;
use std::thread;
use std::sync::Arc;
fn main() {
let counter = Arc::new(RefCell::new(0));
let mut threads = vec![];
for _ in 0..5 {
let counter = Arc::clone(&counter);
let handle = thread::spawn(move || {
let mut num = (*counter).borrow_mut();
*num += 1;
});
threads.push(handle);
}
for thread in threads {
thread.join().unwrap();
}
println!("Result: {}", *counter.borrow());
}
还是不行，现在我们有了一个和之前类似但是略微不同的错误。RefCell<T>不能在线程间共享，因为它没有实现Send和Sync trait。
如果RefCell<T>不适用于并发，那就必须要有一个线程安全的替代选项。事实上，Mutex（Mutex<T>）就是这个替代选项。除了提供内部可变性之外，mutex 还可以在线程间共享。线程在访问 mutex 对象之前必须先获取一个锁，以确保一次只有一个线程访问。锁定一个 mutex 会返回一个LockResult<MutexGuard<T>>类型的智能指针。LockResult是一个枚举（enum），可以是Ok<T>或Error。简单起见，我们通过调用unwrap()将其导出，如果是Ok，unwrap()会返回其内部对象（这里是MutexGuard），如果是Error则会 panic。MutexGuard是另一个智能指针，它可以被解引用以获取其内部对象，当LockResult离开作用域时，锁会被释放。在我们的代码中，我们将RefCell<T>替换为Mutex<T>并更新其操作方式，更新后的代码如下：
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
let counter = Arc::new(Mutex::new(0));
let mut threads = vec![];
for _ in 0..5 {
let counter = Arc::clone(&counter);
let handle = thread::spawn(move || {
let mut num_guard = (*counter).lock().unwrap();
*num_guard += 1;
});
threads.push(handle);
}
for thread in threads {
thread.join().unwrap();
}
println!("Result: {}", *counter.lock().unwrap());
}
让我们高兴的是，它奏效了！
值得注意的是，Arc<T>和Mutex<T>的原子性是有性能开销的。在单线程程序中也可以使用它们但这并非明智之举。
了解了上述情况后，Rust 并发是否如其所声称的那样无须畏惧（fearless）呢？答案就在于它处理大多数常见并发陷阱的方式，其中的一些上面已经介绍过了。
1. 竞态条件（Race conditions）
从Mutable[3]这篇文章我们知道，数据竞争/不一致发生于两个或更多的指针同时访问相同的数据，其中至少有一个指针被用于写入数据并且对数据的访问没有得到同步。通过保证一个 mutex 总是和一个对象关联，进而保证对对象的访问总是同步的（synchronized）。这一点和 C++不同，在 C++中，mutex 是一个单独分开的实体，程序员必须人为地保证在访问一个资源之前要获取一个锁。下面是一个关于不正确使用 mutex 如何引发不一致性的例子：
#include <iostream>
#include <thread>
#include <mutex>
#include <chrono>
using namespace std;
mutex mtx;
int counter = 0;
void increment_to_5(int id)
{
cout << "Executing from thread: " << id << endl;
if (counter < 5)
{
lock_guard<mutex> lck(mtx);
counter++;
}
}
int main()
{
thread threads[10];
for (int i = 0; i < 10; ++i)
threads[i] = thread(increment_to_5, i);
for (auto &thread : threads)
thread.join();
cout << "Final result: " << counter << endl;
return 0;
}
函数increment_to_5()在不同的线程中被调用，这段代码的期望是增加counter变量直到它达到 5。
尽管只在临界区（增加 counter）之前锁定 mutex 是合理的，但是这个锁应该在将counter和 5 进行比较之前就应该获取。否则，多线程可能会读取一个过期的值，并且将其增加从而导致超过预期的 5。在获取锁之前添加一个小的延迟，这个结果就能很容易地复现出来，如下所示：
cout << "Executing from thread: " << id << endl;
if (counter < 5)
{
this_thread::sleep_for(chrono::milliseconds(1));
lock_guard<mutex> lck(mtx);
counter++;
}
}
counter最终的值在每次运行之后都会不同。如果把这段代码用 Rust 来写就不会出现这种问题，因为counter是一个 mutex 类型，并且在任何访问之前都要先获取锁。下面的代码运行多次之后将会产生相同的结果：
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
fn main() {
let counter = Arc::new(Mutex::new(0));
let mut threads = vec![];
for i in 0..10 {
let counter = Arc::clone(&counter);
let handle = thread::spawn(move || {
println!("Executing from thread: {}", i);
let mut num_guard = (*counter).lock().unwrap();
if *num_guard < 5 {
thread::sleep(Duration::from_millis(1));
*num_guard += 1;
}
});
threads.push(handle);
}
for thread in threads {
thread.join().unwrap();
}
println!("Result: {}", *counter.lock().unwrap());
}
2. 死锁
虽然 Rust 减轻了一些并发风险，但是它还不能够在编译期检测死锁。下面的例子展示了两个线程如何在两个 mutex 上发生死锁：
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
fn main() {
let v1 = Arc::new(Mutex::new(0));
let v2 = Arc::new(Mutex::new(0));
let v11 = Arc::clone(&v1);
let v21 = Arc::clone(&v2);
let t1 = thread::spawn(move ||
{
println!("t1 attempting to lock v1");
let v1_guard = (*v11).lock().unwrap();
println!("t1 acquired v1");
println!("t1 waiting ...");
thread::sleep(Duration::from_secs(5));
println!("t1 attempting to lock v2");
let v2_guard = (*v21).lock().unwrap();
println!("t1 acquired both locks");
}
);
let v12 = Arc::clone(&v1);
let v22 = Arc::clone(&v2);
let t2 = thread::spawn(move ||
{
println!("t2 attempting to lock v2");
let v1_guard = (*v22).lock().unwrap();
println!("t2 acquired v2");
println!("t2 waiting ...");
thread::sleep(Duration::from_secs(5));
println!("t2 attempting to lock v1");
let v2_guard = (*v12).lock().unwrap();
println!("t2 acquired both locks");
}
);
t1.join().unwrap();
t2.join().unwrap();
}
从输出结果中可以看出，在每个线程各自获取第一个锁后，程序被挂起了。
类似的方式，在消息传递时也会发生死锁，如下面代码所示：
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx1, rx1) = mpsc::channel();
let (tx2, rx2) = mpsc::channel();
thread::spawn(move || {
println!("Waiting for item 1");
rx1.recv().unwrap();
println!("Sending item 2");
tx2.send(()).unwrap();
});
println!("Waiting for item 2");
rx2.recv().unwrap();
println!("Sending item 1");
tx1.send(()).unwrap();
}
因为每个 channel 在发送之前都要等待接收一个 item，所以两个 channel 都会一直等待下去。
3. 循环引用（Reference cycles）
通过smart pointers post[4]这篇文章，我们知道，Rc指针可能引发循环引用。Arc指针对此也不能幸免，但是类似的，它可以通过 Atomic Weak 指针减少这种情况。
通过上面的观察，可以说 Rust 的并发并非 100%的完美无暇。它在编译期避免了竞态条件（race conditions），但是无法避免死锁和循环引用。 回过头来看，这些问题并不是 Rust 独有的，与大多数语言相比，Rust 的表现要好得多。Rust 中的并发不一定是无须畏惧（fearless）的，但它不那么可怕。
致谢
感谢太阳快递员、惰性气体、没得三位同学对fearless一词翻译提供的建议。
参考资料
Rust playground: //play.rust-lang.org/?version=stable&mode=debug&edition=2018&gist=0879538b56c60c89c4d083308424c701
[2]
smartpointers post: //dev.to/imaculate3/that-s-so-rusty-smart-pointers-245l
[3]
Mutable: //dev.to/imaculate3/that-s-so-rusty-mutables-5b40
[4]
smart pointers post: //dev.to/imaculate3/that-s-so-rusty-smart-pointers-245l
