Rust并发编程

留言

姊妹篇: Future和Rust异步编程


并发程序是指运行多个任务的程序（或看上去是多任务），即两个及以上的任务在重叠的时间跨度内交替运行。这些任务由线程——最小的处理单元执行。在其背后，并不完全是多任务（并行）处理，而是线程之间以普通人无法感知的速度进行上下文快速切换。很多现代应用程序都依赖于这种错觉，比如服务器可以在处理请求的同时等待其他请求。当线程间共享数据时可能会出很多问题，最常见的两种是：竞态条件和死锁

作者：RustCoder
链接：https://juejin.cn/post/7210747150815707193


Rust 同时提供了 async/await 和 多线程两种并发模型，要熟练使用多线程并发模型首先要掌握 Rust 的线程等相关知识，比如线程创建，线程同步，线程安全等。线程间消息传递的通道 channel，线程间共享状态的智能指针 Mutex 和 Arc。类型系统和借用检查器会确保这些场景中的代码，不会出现数据竞争和无效的引用。一旦代码可以编译了，我们就可以坚信这些代码可以正确的运行于多线程环境，而不会出现其他语言中经常出现的那些难以追踪的 bug。Sync 和 Send trait 的为多线程中数据传递和共享提供安全保证。

作者：RustCoder
链接：https://juejin.cn/post/7210747150815707193
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。



Rust并发系列1-Thread

spawn: 类似C的fork，java的Runnable，Go的go func，传入function在另一个线程中执行。

Rust并发系列2-Lock

Rust并发系列3-channel




异步编程和并发编程的区别


我的理解: 异步是有A,B,C三件事,可不用串行去做,可以做A的同时去做B

而并发,是可以有1000个线程/协程去调用某个func


异步编程和并发编程是两个相关但不同的概念。以下是它们的主要区别：

  1. 定义:

    • 异步编程：一种编程范式，允许程序在等待某些操作完成时继续执行其他任务。
    • 并发编程：同时处理多个任务的能力，可能涉及真正的并行执行。

  2. 目的:

    • 异步编程：主要用于提高程序的响应性和效率，特别是在I/O密集型任务中。
    • 并发编程：旨在提高程序的整体吞吐量，尤其是在多核处理器上。

  3. 执行方式:

    • 异步编程：通常在单个线程上执行，使用事件循环来管理任务。
    • 并发编程：可能涉及多个线程或进程的真正并行执行。

  4. 复杂性:

    • 异步编程：通常较简单，不需要处理复杂的同步问题。
    • 并发编程：可能更复杂，需要处理race conditions、死锁等问题。

  5. 适用场景:

    • 异步编程：适合I/O密集型任务，如网络请求、文件操作等。
    • 并发编程：适合CPU密集型任务，如大规模数据处理、科学计算等。

  6. 实现机制:

    • 异步编程：常用回调函数、Promise、async/await等机制。
    • 并发编程：使用线程、进程、协程等机制。

需要注意的是，这两种编程方式并不互斥，实际上它们经常结合使用以获得最佳性能。

您是否需要我进一步解释这些概念或提供一些具体的编程示例？


rust异步编程和并发编程的区别

Rust中的异步编程和并发编程虽然都旨在提高程序的效率,但它们有一些关键区别:

  1. 执行模型:

    • 并发编程: 通常使用多线程,每个任务在自己的线程中执行。
    • 异步编程: 使用单线程或有限的线程池,通过任务切换来处理多个任务。

  2. 资源使用:

    • 并发编程: 每个线程都需要单独的栈空间,资源消耗较大。
    • 异步编程: 通常资源消耗更少,因为它在较少的线程上管理多个任务。

  3. 适用场景:

    • 并发编程: 适合CPU密集型任务。
    • 异步编程: 更适合I/O密集型任务,如网络操作。

  4. 编程模型:

    • 并发编程: 使用标准的Rust线程API。
    • 异步编程: 使用async/await语法和Future特性。

  5. 切换开销:

    • 并发编程: 线程切换开销较大。
    • 异步编程: 任务切换开销通常较小。

  6. 同步原语:

    • 并发编程: 使用标准的同步原语如Mutex、RwLock等。
    • 异步编程: 需要特殊的异步版本的同步原语。

  7. 错误处理:

    • 并发编程: 使用标准的错误处理机制。
    • 异步编程: 错误处理可能更复杂,尤其是在跨越await点时。

需要注意的是,这两种方法并不互斥,在实际应用中常常结合使用以获得最佳性能。

如果您想了解更多细节或有具体的问题,我很乐意进一步解释。



rust中有了thread::spawn，为什么还要有async，await这些？？

这是一个很好的问题。Rust中的thread::spawn和async/await解决了不同的问题，并且在不同的场景下有各自的优势。让我解释一下它们的区别和各自的用途：

  1. 线程 (thread::spawn)：

    • 适用于CPU密集型任务。
    • 每个线程都有自己的栈，占用较多内存。
    • 线程数量受限于系统资源。
    • 线程切换开销较大。
    • 适合并行执行长时间运行的任务。

  2. 异步编程 (async/await)：

    • 适用于I/O密集型任务。
    • 轻量级，可以创建大量任务而不会显著增加内存使用。
    • 可以处理大量并发任务。
    • 任务切换开销小。
    • 在单线程上也能实现并发。

异步编程的主要优势：

  1. 更高的并发性：可以同时处理数千甚至数十万个任务。

  2. 资源效率：异步任务共享同一线程，减少了线程创建和切换的开销。

  3. 更好的I/O处理：特别适合处理网络I/O等需要等待的操作。

  4. 细粒度控制：可以更精细地控制任务的执行和调度。

  5. 更容易实现复杂的并发模式：如同时等待多个异步操作完成。

例如，使用async/await的代码可能看起来像这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
use tokio::time::{sleep, Duration};

async fn fn1() {
    sleep(Duration::from_secs(3)).await;
}

async fn fn2() {
    sleep(Duration::from_secs(6)).await;
}

async fn fn3() {
    sleep(Duration::from_secs(3)).await;
}

#[tokio::main]
async fn main() {
    let start = std::time::Instant::now();

    tokio::join!(
        async {
            println!("开始执行 fn1");
            fn1().await;
            println!("fn1 执行完毕");
        },
        async {
            println!("开始执行 fn2");
            fn2().await;
            println!("fn2 执行完毕");
        },
        async {
            println!("开始执行 fn3");
            fn3().await;
            println!("fn3 执行完毕");
        }
    );

    let duration = start.elapsed();
    println!("总耗时: {:?}", duration);
}

这个异步版本在功能上与线程版本相似，但它可以更有效地处理大量并发任务，特别是在I/O密集型场景中。

总的来说，thread::spawn和async/await都有其适用的场景，选择使用哪一个取决于你的具体需求和应用场景。


Rust多线程编程和异步编程







https://space.bilibili.com/35891473/search/video?keyword=%E5%B9%B6%E5%8F%91

并发、并行和异步之间的区别是什么


并发、并行和异步之间的区别是什么?

** 简单的说：并行是多线程的一种形式，多线程是并发的一种形式。异步也是并发的一种形式**

通过这两篇

爽哥这样总结:

并发和并行确实可以明确区分出来,因为cpu的速度非常快,在执行一个任务时经常要等其他组件,比如网络,磁盘等,如果一直串行等待这样就会造成很大的浪费. (就类似于烧水的同时,可以切菜,不用等烧水完成了才去切菜,我可以烧一会水,火生起来了水壶放上了,随后这段时间就能去切菜,切着切着菜发现水烧开了,就又可以切换到烧水,并发就是高频高速做这样的切换) .

即并发就是做这样事情的, 不用一直等着,可以去做别的事,因为cpu一直切换,单线程,或者说单核cpu也可以不停切换. 以Go来说, go func1, go func2 ... go func100 和 for循环里启动100个协程 for i:=0;i<100;i++ { go func666 }, 其实都叫并发, 区别是 go func1,go func2 需要是互相不依赖的才会这样写,而for循环里启多个func666, func666相互之间更不存在依赖关系了(其实和外面go func666, go func666 …写100个等效)

即当go func1,go func2..go func100 无相互关系时, 和循环里启100个go func666 两者本质一样(其实效果也一样,只不过for里写方便点,代码量少一点).

(面试要求中常说的高并发,其实等同于高QPS,一定程度上模糊了其概念和作用)

如果是单核情况下的并发,本质上还是同一时间,只能做一个事,不可能既烧水又做饭 (cpu高频,看起来是能同时做), 即现在这100个协程,由GMP机制不停调度, 但在任何一个时刻,能得到执行的,还是最多只有1个协程 (但这种情况还是可以导致map等有并发问题…). 即func1,func2..func100同一时间点只能有一个在运行,或者for循环里启100个协程,同时也只有一个在执行(go淡化了协程id的概念,不给程序员暴露,但其实是有协程id的~)

而并行,必须得是多核cpu. 即无多核,不并行…因为是有多核,所以同一时刻,可以做多件事情.多核cpu情况下的并发, 就算是并行了(并行不一定都在执行并发相关的代码,大概率可能在做别的; 比如多核cpu,一个核在处理并发相关的逻辑块,另外的核在处理其他程序比如qq,或者哪怕在处理同样说的上面这个100个协程的go程序,也可能在处理这个程序并发无关的模块,所以有后面这句话的最多), 还是以上面100个协程的case来说, 比如有4核心,那同一时刻,确实可以有最多4个协程得到执行…

即1号协程的func1,2号协程的func2 .. 是可以同一时间点都运行的


而异步, 异步帮助实现并发,在多线程环境中是实现并行的一种方法。



所以说,和串行相对的概念不是并行,而是并发.. 串行是从上到下func1完了做func2,再func3….而并发是这些没有前后关联的func可以同时做, 在某个func陷入停滞时,去做其他func 而不会阻塞在那干等着(如何才能判断某个func陷入停滞? go里面是每个协程都高速切换,最多有10ms执行机会)



~

这个问题在StackOverflow上很受欢迎,已有13.8万次浏览。虽然提问时间已接近13年,但8天前仍有修改,说明这个话题持续受到关注。

让我们来看看高赞回答是怎么说的:

  1. 并发和并行:
    并发和并行都与同时执行任务有关,但有细微差别:

    • 并行是真正的多任务处理,任务在时间点上真的同时执行。
    • 并发可能意味着任务共享执行线程,看起来像并行执行,但实际上在时间点上并不同时。

  2. 异步:
    异步与并发和并行没有直接关系。异步方法给人一种并发和并行任务的印象,但实际上:

    • 异步通常用于需要离开当前应用程序进行工作的进程。
    • 我们不想等待并阻塞应用程序,而是继续执行其他任务。

例如,从数据库获取数据可能需要时间,但我们不想阻塞UI。使用异步方法,我们可以:

  1. 向远程系统发出请求
  2. 继续响应用户
  3. 当数据返回时,通过回调方法更新UI

从用户角度看,这像是多任务处理,但实际可能不是。

补充说明:在许多实现中,异步方法调用会启动新线程,但这不是必须的。它取决于正在执行的操作和如何将响应通知回系统。 (go 就不是, 起一个go func,这就算是一个异步了,但并不一定新增线程)

另一个高赞回答用图解释了并发、并行和异步:

  1. 并发:
    多个任务在单个处理单元上交错执行,通过分割时间片方式处理。

  2. 并行:
    多个处理单元同时处理不同任务。

  3. 异步:
    异步是一个独立概念,指一个事件可能在另一个事件的不同时间发生(不同步)。

异步执行示例:

  • X1告诉X2执行任务A
  • X2将A加入日程安排
  • X1继续做其他事情
  • X2完成手头工作后执行A
  • A完成后,X2通知X1

相比之下,同步执行时X2会立即执行A,完成后立即通知X1。

关于异步的其他观点:

  • 异步意味着程序执行非阻塞操作,如发起HTTP请求后继续执行其他任务。
  • 异步属于线程执行,允许在一个任务执行时切换到另一个任务,无需等待前一个完成。
  • 异步帮助实现并发,在多线程环境中是实现并行的一种方法。
  • 异步是同步的反义词,描述线程是否与其他线程同步。

总的来说,异步、同步是编程模型,而并发、并行是任务执行的方式。异步为并行铺平了道路,通常涉及:

  1. 定义任务的初始化(何时开始,参数,完成后做什么)
  2. 确定代码在任务期间应该做什么
  3. 使用回调方法表示

一个简单的类比:

  • 异步:告诉某人洗衣服和洗碗,他们完成后向你汇报。
  • 同步:告诉某人洗碗,你等着他们完成,然后才能做下一件事。

总结:异步的目的是将一件事合理地拆分成多个小步骤,然后可以并行执行这些步骤。



谷歌Rust教程之并发（1）


谷歌Rust教程之并发(1)

https://www.bilibili.com/video/BV1Ly4y1A73J/

让我们来看一下Google的Rust教程中讲到的并发内容的前一部分。因为它有点长,并且并发涉及到的知识相对来讲比较多一些。与《Atomic and Locks》那本书一样,它也是从线程开始讲起,因为并发基本上是基于线程的。

这是它的链接,有兴趣可以去看。Rust的线程的工作方式与其他语言的线程类似。线程都是守护线程,守护线程很好理解,基本上是在后台运行的。就是相当于大哥都走了,小弟也会跟着自动离开。也就是说在主线程退出后,它们也会自动退出,这是守护线程的一个特征。

上面可以看到守护线程的特点,主线程不会等待它们退出。我们直接运行这个例子的小程序,就能看到子线程是打印1到10,而主线程是从1打到5。我们可以看到在主线程打印到4或5的时候就退出了,它不会等到子线程打印到10。很明显看到主线程是不会等待子线程完成再退出的,它不会等着守护线程结束。这个其实也跟那本书的内容如出一辙。

线程panic是彼此独立的,就是说一个线程的panic不会影响到其他的线程。因为这种情况下没有使用join,所以线程之间是相互没有影响的,性质上是相互隔离的。panic可以携带有效的payload,也就是说panic信息中可以包含数据。可以使用downcast_ref函数来解析panic信息中的具体内容。

其实这里讲的就是线程默认有这种特性。要注意的是,在上面那个例子中,子线程在达到打印10之前就被停止了,因为主线程没有等待它。使用thread::spawn和handle.join()可以等待线程完成。使用join函数可以在线程中引发异常,注意这对main是没有影响的。可以使用handle.join()的result返回值来访问异常的具体信息。

他说这是讨论Any的好时机。为什么呢?我们来看一下。在Rust中可以使用spawn函数来启动一个新的线程。这个函数会返回一个JoinHandle,它可以用于等待线程完成。如果要等待线程完成,可以使用handle.join()函数。这个函数会阻塞当前线程,直到子线程完成。如果线程中引发了异常,那么handle.join()函数会返回一个包含异常信息的错误结果。

讲到Any的好时机是因为Any类型可以用于存储任何类型的Rust值,所以它通常用于在线程之间传递数据。这就是Any的好处。

再来看作用域线程。其实在讲到闭包的时候,我们就很容易想到闭包这个特性是有作用域的说法的。普通线程不能从环境中借用。我们可以看到,从环境中借用基本上要用到闭包。虽然闭包可以读取环境中的信息,但是能不能借用是另外一回事。

比如说在这个例子中,我们可以看到变量s。s能够被看到,但是它能不能被借用呢?这是一个问题。我们可以看到它报错了:”may outlive borrowed value”。这意味着线程可能比main活得更久,这是不允许的。因为这是一个普通线程。

但是可以使用作用域线程。使用thread::scope,然后在里面再起一个新的线程,这时候就可以让线程借用环境中的变量。而且它会等待线程运行完成。你可以看到它打印出了字符串的长度。这是因为当作用域函数完成时,所有线程会被自动join。因此它们可以安全地借用数据。Rust的借用规则仍然适用:你可以有一个线程进行可变借用,或者有任意数量的线程进行不可变借用。

再看一下通道。通道其实是用得比较多、也比较安全好用的一种方式。Rust通道由两个部分组成:一个是发送端(Sender)和一个接收端(Receiver)。通过通道进行连接,但我们只能看到它们的接口(endpoints)。

他给了一个例子,展示了通道的发送和接收。可以看到发送的是1、2、3、10,接收到的也是1、2、3、10。这很容易理解。MPSC就是”Multi-producer, single-consumer”的缩写,意味着多生产者和单消费者。也就是说,一端可以产生很多数据,而另一端只有一个接收者。Sender实现了Clone trait,所以你可以创建多个生产者,而Receiver没有实现Clone。

send和recv都会返回Result。如果它们返回错误,就表示对应的Sender或Receiver已经被丢弃,也就是通道已经关闭。

我们先看一个无界通道。mpsc::channel()函数可以创建一个无界的异步通道。无界通道是一种可以发送和接收任意数量数据的通道。就是说,这个通道里面的发送和接收不限制长度,可以很长很长,没有限制。所以它叫无界通道。

在这个例子中,我们特意用循环做了一个限制。如果不限制的话,它就是无限的。所以这里从1到10发送信息。发送和接收很好理解:一个线程去发送,另一边接收。可以看到从1到10发送,然后接收。这里的限制是通过循环做的,如果不限制的话,那就是不停地发送和接收。

再看有界通道就很好理解了。理解无界就能理解有界。对于有界或同步通道,send可能会阻塞当前线程。为什么这样说呢?我们可以这样理解:有界通道是一种只能容纳一定数量数据的Rust通道。就像一个容器,它只能装这么多,再装就装不下了。这意味着容器是有限的。

如果通道已经满了,那么发送线程就会被阻塞,因为发送不了了。同步通道意味着发送数据会阻塞当前线程,直到数据被接收端接收。就是直到接收端把数据消耗掉,通道才会空出来。

他给了一个例子,我们给通道限定只允许3个输入。然后我们运行,可以看到发送的时候是1、2、3,然后接收。虽然有点乱序,可以看到接收是4、5、6,发送是5、6、7、8、9。就是这样三个三个操作,中间虽然会有点乱序,但基本上是按照三个来的,不会超过三个,不会变成四个五个。

这里解释了一下:调用send会阻塞当前线程,直到通道中有足够的空间容纳新的数据。如果没有线程从这个通道中读取数据,这个线程可能会一直被阻塞。比如说在这个例子里,我们发送了8个。这是因为前面已经剩了一个,然后再消耗掉三个,再有三个进去,所以你可以看到它连着发了四个。

如果通道关闭了,那么send调用将终止并返回错误。这就是为什么它会返回错误。当Receiver被丢弃时,通道也会关闭。

大小为零的有界通道被称为”rendezvous channel”(约会通道)。每次发送都将阻塞当前线程,直到另一个线程调用recv。为什么叫约会通道呢?因为这相当于在达成协议的时候,通道容量是零,就相当于接收者和发送者之间对话,把约定好了,所以叫约会通道或约定通道。就像两个人之间做好约定,然后再去传输这些信息。

这就是并发里面的通道和线程的一个解释,比较浅显易懂。后面又讲到了Send和Sync,因为这个相对来讲也可以独立地做一个主题来讲,所以我们把它放在下一篇再去讲Send和Sync。


谷歌Rust教程之并发（2）

谷歌Rust教程之并发（2）
https://www.bilibili.com/video/BV1F94y1L71F/

接着来看谷歌的Rust教程当中并发的第2部分,Send和Sync。这两个特性其实在并发中使用得比较多。

让我们看看Rust是怎么禁止跨线程共享访问的。答案是两个trait:Send和Sync。

如果一个类型的值可以安全地从一个线程移动到另一个线程,那么这个类型就具有Send trait。如果一个类型的引用可以安全地从一个线程移动到另一个线程,那么这个类型就具有Sync trait。

Send和Sync是unsafe traits,也就是不安全的trait。但是,如果你的类型只包含Send和Sync类型,编译器就会自动为它们推导出这两个trait。如果你知道它们是有效的,也可以手动去实现它们。

这两个trait实际上是标记,表明这个类型具有线程安全的属性。它们也可以像普通的trait一样用于泛型约束。

我们先看Send。如果将T类型的变量移动到另一个线程是安全的,那么T类型就具有Send特性。将一个值的所有权移动到另一个线程会导致析构函数在原线程中运行。这意味着我们需要考虑什么时候可以从一个线程中分配值,并在另一个线程中释放它。这涉及到生命周期,你一定要考虑好在什么时间点可以进行Send操作。

举个例子,像tokio这个库,连接就必须是线程安全的,只能在一个线程中使用。

再看Sync。如果可以同时从多个线程访问T类型的值,那么这个T类型就属于Sync。要注意强调的是”同时”和”多个线程”。更准确的定义是:T具有Sync特性当且仅当T的引用具有Send特性。

我们可以这样解读:Sync特性表示一个类型可以安全地跨线程共享。这意味着如果多个线程同时访问该类型的值,不会导致任何数据损坏或其他问题。从字面意思理解,Sync就是同步的,不管哪些线程访问,它都是同步的,大家都能得到一致的值。

Send特性表示一个类型可以安全地跨线程移动。这意味着如果这类型的值从一个线程移动到另一个线程,不会导致任何数据破坏或其他问题。

实际上,这句话是在说:如果一个类型对于共享使用是线程安全的,那么在线程间传递它的引用也是线程安全的。因为如果一个类型是Sync的,就意味着它可以被多个线程共享,没有数据竞争或其他同步问题的风险。所以可以安全地将它移动到另一个线程。

我们再看一些例子:

  1. 既是Send的,也是Sync的:
    这是比较常见的情况,大多数我们遇到的类型都是Send和Sync的。比如说像i8、f32、bool、String、引用、元组、结构体、String、Option、Vector、Box、Arc、Mutex、原子类型等。泛型通常也是Send和Sync的,只要类型参数是Send和Sync的。

  2. 是Send的,但不是Sync的:
    这种类型可以跨线程移动,但不是线程安全的,不能多线程共享。通常它们是内部可变的。比如说MPSC通道的发送端和接收端,以及Cell或RefCell。

  3. 不是Send的,但是Sync的:
    这种类型是线程安全的,但不能移动到另一个线程上去。比如说MutexGuard。这些类型通常使用操作系统级原语,必须在创建它们的线程上释放。这是因为它们用到了操作系统提供的底层机制来实现同步和互斥。如果移动了,可能会出现死锁。这种情况在并发中不是特别常见,因为我们使用操作系统原语的场景不是特别多。

  4. 既不是Send也不是Sync的:
    这种类型既不是线程安全的,也不能移动到另一个线程。比如说Rc。Rc使用非原子的引用计数,可能导致数据损坏或其他问题。原始指针也可能具有特殊的并发考虑,因为它们允许在不同的线程之间访问同一块内存。

这是并发中很容易见到的Send和Sync特性。后面还有共享状态等内容,我们会在另一个视频中讨论。这个视频主要是为了单纯地看一下什么是Send和Sync,以及我们常见的一些类型和场景。



并行与并发图解


并行与并发图解

并行和并发这两个概念已经出现很久了，研究至少有十几年了。那么并行和并发，它们到底有什么不同呢？我们可以通过画图来更清晰地理解这两个概念。

首先，并行的概念大家都比较熟悉，就是两个CPU或进程同时进行工作，这很容易理解。

那么，没有并行的并发执行会是什么样子呢？并发是这样的：假设我们有一个进程或单个处理单元，这个处理单元要处理三个任务（Task 1、Task 2、Task 3）。当它在处理Task 1时，可能需要读取IO等待，这时它就会转而处理Task 2。所以我们可以看到，在并发中，单个线程或进程会交替处理多个任务。每个任务之间没有重叠的部分。这就是并发，一个处理单元同时处理多个任务。从人的感觉来看，就像一个人同时在做三件事。对于单个处理单元来说，这种方式是非常饱满的，能够同时处理多个任务。

接下来，我们来看有并行的并发会是什么样子。假设我们还是要完成那三个任务，但这次我们至少有两个进程或处理单元同时工作。在这种情况下，我们可以看到，当Task 1在等待时，我们可以让另一个进程来处理Task 2的一部分。比如，当进程一在处理Task 2.1时，进程二可以同时处理Task 2.3。这样，两个进程同时工作，显然可以更早地完成所有任务。这就是既有并发又有并行的情况，明显提高了效率。

然而，在现实中，情况可能会更复杂。假设Task 1.3依赖于Task 1.2的结果（比如需要等Task 1.2读取文件后才能继续），那么即使有多个进程，Task 1.3也必须等到Task 1.2完成才能开始。这就导致了一些空白时间，难以完全填满所有的处理时间。

因此，能否满足并行条件，要看任务之间是否有依赖关系。如果有依赖关系，就无法完全并行，只能并发处理。如果没有依赖关系，才能truly并行执行。

此外，如果我们想充分利用硬件性能，实现完美的并行和并发，是非常困难的。这需要有足够多的任务，同时这些任务之间的依赖性不能太强。

有些极端情况下，比如IO特别慢，即使你的并行并发做得再好，也要一直等待IO完成。在这种特殊场景下，IO的缓慢可能决定了整体性能无法提升。

从某种意义上讲，并行并发其实是为了解决IO处理时间过长的问题。如果IO性能非常好，无论是从内存读取还是从磁盘读取，并行和并发的意义可能就没那么大了。并行并发的本质是为了解决计算和IO之间速度差异的矛盾。


什么是并发


这篇文章取自《Rust Service and App》这本书,作者是Pro。作者主要想讲解Rust中的异步概念,开头部分专门讲述了异步编程,内容比较浅显。

从”async programming concepts”这部分开始讲解异步的概念。在计算机科学中,并发性是指程序的不同部分能够乱序执行或同时执行,而不影响最终结果的能力。这里强调的是不影响最终结果,中间过程如何并不重要。并发的目的是要保证最终结果正确,同时提高执行速度和效率。

严格来说,并发性指乱序执行程序部分,而并行性是指同时执行多个任务。但在实践中,并发性和并行性经常结合使用。文章中使用”并发性”这个术语来广泛指代这两个方面。

文章配有图片来说明并行和并发的区别。并行处理时,多个CPU各自处理不同的任务,互不干扰。而并发处理时,一个CPU可能在不同任务间切换,充分利用CPU时间,但增加了复杂性。

促使进行并发编程的主要因素有两个:需求方面和供给方面。

在需求方面,用户希望程序运行得更快、效率更高。为满足这一需求,软件开发人员需要考虑并发编程技术。

在供给方面,随着多核CPU的普及,软件开发人员有机会编写可以利用多个可用内核或处理器的程序,从而提高整体执行速度和效率。

设计和编写并发程序是一项复杂的任务。首先要确定哪些任务可以并发执行。开发人员需要考虑如何确定代码的哪些部分可以并发执行。

程序执行的处理可以大致分为两类:CPU密集型任务和IO密集型任务。虽然现实世界中的大多数代码都是二者兼有的,但我们将两种任务分开进行讨论。

CPU密集型任务的例子包括基因测序、视频编码、图形处理和在区块链中计算加密证明等。这类任务主要涉及访问内存中的数据,将程序指令和数据加载到堆栈中并执行,对CPU和内存都有较高要求。

IO密集型任务的例子包括访问文件系统或数据库中的数据,以及处理网络TCP/HTTP请求等。这类任务在IO操作上花费的时间比在CPU计算上花的时间更长。
文章举例说明了在CPU密集型和IO密集型任务中如何应用并发编程技术。例如,对于一个计算数字列表中每个数字平方根的程序,可以将其结构化以便使用多个CPU核心同时处理独立的数字。这个例子很容易联想到SIMD或使用GPU的方式,可能会有很大优势。

对于IO密集型任务,如Web服务器处理HTTP请求,并发编程可以让处理器在等待IO操作完成时执行其他任务,从而提高效率。文章详细讲解了HTTP请求处理的过程,以及如何在处理多个同时到达的请求时应用并发编程。

文章还介绍了同步处理、多线程处理和异步处理这三种不同的编程模式,并通过图示和示例说明了它们的异同。同步处理按顺序执行任务,遇到阻塞就等待。多线程处理可以将含有阻塞任务的部分分配到单独的操作系统线程上。异步处理则使用运行时(如Tokio)来管理任务调度,在遇到阻塞操作时切换到其他任务。

文章强调,选择哪种方法最适合要根据特定的运行环境和设计的考虑来决定,并非所有情况都适合并发处理。

最后,文章提到了使用操作系统原生线程实现并发的方法,但具体内容将在下一部分讲解。


文章介绍了同步处理、多线程处理和异步处理这三种不同的编程模式，并通过图示和示例说明了它们的异同。

  • 同步处理：

在同步处理的情况下，我们可以看到thread1处理task1。如果task1包含一个IO操作，可能会出现阻塞（blocking）。在阻塞期间，线程会等待，比如等待网络数据传输完成。阻塞结束后，线程继续执行。这种方式是按顺序执行的，不会出现数据不一致的现象，但效率可能较低。

  • 多线程处理：

在多线程模式下，可以将包含阻塞任务的任务1分配到一个单独的操作系统线程上处理。处理器可以继续在另一个线程上执行其他任务。例如，thread1处理task1，thread2处理task2。这种方式比同步处理效率更高，某种程度上也算是异步的。

  • 异步处理：

在异步处理模式下，运行时（如Tokio）会管理处理器上的任务调度。它会执行task1直到遇到阻塞操作。当遇到阻塞时，异步运行时会安排执行第二个任务。当第一个任务的阻塞操作完成时，它会被重新安排执行。这种方式能更高效地利用CPU时间。

文章通过一个示例来说明这三种处理方式的区别。假设task1包含三个部分：处理输入数据、执行阻塞操作、打包并返回数据。阻塞操作意味着当前执行线程正在等待某些外部操作完成，比如从大文件或数据库读取数据。

在同步处理中，处理器先完成第一部分，然后等待阻塞操作完成，最后执行第三部分。

在多线程模式下，可以将包含阻塞任务的task1分配到一个单独的操作系统线程上，处理器可以继续在另一个线程上执行其他任务。

在异步处理中，运行时（如Tokio）会管理任务调度。它执行task1直到遇到阻塞操作，然后安排执行第二个任务。当第一个任务的阻塞操作完成时，它会被重新安排执行。

文章强调，从高层次上看，同步处理与两种并发处理模式（多线程和异步）有明显区别。选择哪种方法最适合要根据特定的运行环境和设计考虑来决定，并非所有情况都适合并发处理。

文章还提到了第二个实例，即Web服务器接收多个同时发生的网络请求，并解释如何使用这两种类型的并发处理技术。第一种是使用操作系统原生线程实现并发，但具体内容将在下一部分讲解。



深度理解Rust并发（一）



好的,我会根据您的要求,原原本本地整理内容,不会遗漏或总结。以下是整理后的内容:

深度理解Rust并发（一）

不知不觉之间已经来到了这本书的第19章。回想从第二章开始,逐渐到19章,居然也坚持到了现在。这本书剩下的章节并不多了,但是后面的章节会越来越显得非常重要。这本书总共有23章,现在后面还有四章内容,但这四章内容其实每一个章节都可以扩展成几乎一本书。所以它是非常重要的。

那这第19章是讲的并发,我们来看它的具体内容。它的引言是这样讲的:从长远来看,使用允许无限使用存储位置及其地址的机器语言,编写大型并发程序并不是明智的。即使借助复杂的硬件机制,我们也无法使此类程序变得可靠。其实这种并发和在一台机器上去无限制地用这个存储和这个内存的方式,我们通常把它叫做scale up。其实我们在做很多,不管是做存储也好,做这个数据库也好,scale up往往是能够满足一部分的需求,那更多的需求是需要scale out才能完成的。其实这里面有讲到,就是说如果在scale out这种情况下,并发怎么样去处理。就在这个引言来讲的话,其实跟很多相关的一些技术上的东西都是非常相似的。

那还有引言,就是说通信模式是并行模式。好我们来看它的正文。贯穿职业生涯,就是我们在职业生涯当中,他说可能作者年纪是不是比较大,我没注意,他想到贯穿这生涯,你对并发编程的态度也许会发生改变。这并不奇怪,这种情况很普遍,就是对并发这种很容易会有变化,就是你在不断地去用它的时候,你会改变自己的这种想法。

起初,编写并发代码既轻松又有趣。线程、锁、队列等工具非常容易上手和使用。成长中并发编程存在很多陷阱,但幸运的是你都知道这些陷阱,并且会小心地避免犯错。在某个时候,你不得不调试别人写的多线程代码,你会被迫承认有限,真的不应该使用这些工具。就有限在用这个东西,不是很擅长,就是没有理解透就用它。

然后在接下来的某个时刻,你将不得不调试自己编写的多线程代码。经验会让人对所有多线程代码持怀疑态度,甚至是愤世嫉俗。偶尔会出现一些文章,详细解释为什么一些看似正确的多线程惯用法根本不起作用。这与内存模型有关。但最终你会找到一种你认为可以现实使用而不会经常犯错的并发方法。你可以将几乎所有东西都塞进那种惯用法中,并且如果你真的厉害,你会学会拒绝多余的复杂性。就是多余的复杂性,你会知道最终不要把它舍弃掉。

那当然并发编程的惯用法有很多,系统程序员常用的方法包括以下几种:

  1. 后台线程:这种线程是拥有单个任务,会周期性地唤醒并执行这个任务。

  2. 通用工作线程:是通过任务队列与客户端进行通信的。

  3. 管道:数据从一个线程流到另一个线程,每个线程执行比较少量的工作。有点像流水线一样的。

  4. 数据并行:这是一种假设正确或错误,整个计算机主要用于执行一个大型计算的方式。该计算会被分成N个部分,并在N个线程上运行,期望同时利用机器的所有N个内核。

  5. 同步对象海洋:就是一堆的同步对象允许多个线程访问相同数据,并使用基于低级原语,比如说互斥锁那个自有临时锁定方案,来避免竞态条件。Java内置了对这种模型的支持。这个模型在90年代和2000年代非常流行。

  6. 原子整数操作:允许多个内核通过大小为一个机器字的字段传递信息来进行通信。这比其他所有方法都更难正确实现,除非交换的数据确实是整数值。在实践中它通常是指针,就是Arc。

随着时间的推移,你可能会熟练使用其中几种方法并安全地组合它们。你会成为这项艺术的大师。如果没人修改系统,那么一切都会非常好。使用线程良好的程序充满了不成文的规则。

他一直在讲就是在业界通用的一些并发的方式。他并没有一个非常明文的规定并发应该是什么样式的。有这么多方式都可以去做。这些不成文的规则,包括很多语言,他没有去再从编译器角度去给他做一些东西。

这里面他讲到了Rust,提供了一种更好的方式来使用并发编程。你看他这句话,这些后面讲的东西非常关键。就他并不强制所有程序采用单一风格,也就是说在并发的时候,他不强制你单一风格。这对于系统程序员来说根本不是解决方案,而是通过编译器安全地支持多种风格。不成文的规则被写在了代码中,并由编译器强制执行。这一点就是比其他的一些语言,比如说C++要好得多的地方。

你听说过Rust可以让你编写安全快速并发的程序。本章会向你展示如何做到这一点。我们将介绍三种使用Rust线程的方法:Fork-join、并发通道、共享可变状态。在这个过程当中,你将会用到你迄今为止所学到的所有关于Rust语言的知识。Rust在引用、可变性和生命周期方面所下的功夫,在单线程程序中已经足够有价值,但是在并发编程中,这些规则的真正意义才显得显而易见。它们使你可以扩展你的工具箱,快速正确地编写各种风格的多线程代码,没有怀疑,没有愤世嫉俗,没有恐惧。

使用线程并行

现存最简单的并行之一,就是处理多个完全独立的任务。比如说我们正在对一个大型语料库进行自然语言处理。这个其实就可以看到跟机器学习有点类似了,就是有些地方可能要用到这种地方。就是大型语料库进行自然语言处理,我们可以写一个循环来处理每个文档。海量的数据文档去对它进行一个处理。

你可以看到处理是什么样式的。程序会按照图19-1所设置的方式运行。它是什么的,你可以看到就是在它的程序里面显示load,就是把这个原始的文件读出来。然后就compute,那就是计算它的这个统计的一些数据。然后write output file,然后就是保存。就是读,然后计算保存。处理读,那就load process save,也不停地去这样做。像这样一个process file,process file这个单线程的process file,就是对这个文件处理。就做了这么个单线程的就这么做。就是串行地这么一步步往下来。非常easy好理解。

所以看到这边,就是这个你我们要讲这个并发的话,有很多的书或者文章都会讲到并发,但是我认为他这个章节,他的图示做的是非常非常棒的。

我们接着往下看,由于每个文档都是单独处理的,因此可以通过将语料库分割成块,并在单独的线程上处理每一个块,来相对容易地加快此任务的速度。像图19-2所示,现在它分成几个,别说这里面几个,三个处理三个,load process save load process,然后这里面也再来几个,它也是三个,然后再处理几个,两个load process save也是两个。这样子处理1234,4个,这个一起同时处理。

这种模式称为fork-join并行。你看它单独拆开,它其实是像分成两步似的。fork是什么,是指启动一个新的线程,分叉嘛。为什么是叉子呀,就是分叉嘛,到一定的地方它会分叉,所以叫fork。那就是分开分开了一下,就是启动一个新的线程。那join呢是指等待线程完成。大家一起。先分叉,启动多个线程,然后把它作用在一块。我们已经看过这种技术了,因为在第二章中用它来加速mandelbrot这个程序。还记得吗?然后没关系,后面会讲到它。

fork-join并行之所以广受欢迎,有以下几个原因。这几个原因非常值得一读:

  1. 简单易用:fork-join是易于实现的。Rust的语言本身也提供了很多特性,使得并行编程更加轻松。就已经有很多这个特性库里面可以去用它。就很容易去用。

  2. 避免瓶颈:fork-join并行中不存在共享资源,所以没有线程之间彼此等待的时机,只会在执行结束的时候才会有这个时机。在此期间每个线程都可以独立运行,互不影响。这有助于降低任务切换的开销。就任务之间互相切换的话,它会有很大的开销的。进程之间如果做这种切换也好,做这种信息沟通也好,都是很耗费资源的。

  3. 性能可预测:理论上,理论上一定要注意,他说理论上如果启动四个线程,我们可以将工作时间缩短至原来的1/4。原来你只有一个线程,现在四个线程,那么所有的1/4理论上来讲。然而图19-2也展示了无法实现理想加速的原因之一。是什么呢?就任务可能无法均匀分配给所有线程。你看这里面,它是它进行两个进程,这两个这是两个处理,这三个这三个,但处理的不会很均匀。就很难均匀地分成另一个数。要注意的另一个方面是,有时即便线程执行完毕,已经到join的地方了,fork-join程序也需要额外的时间来整合各个线程的计算结果。就是在这个地方大家都完成了,它需要去把整合所有人的结果计算结果综合。也就是说完全隔离任务可能会带来一些额外的处理开销。不过除此之外,任何依赖CPU且任务之间互不影响的程序,都可以通过fork-join并行获得显著的性能提升。

这里面我为什么把它变成红色,任务之间互不影响,也就是说,如果我们想用fork-join去做并发的时候,一定要注意它的任务之间都有满足什么特征,就任务之间互不影响,才比较适合用fork-join。

  1. 易于推理程序正确性:只要线程真正彼此隔离,他又强调一遍,就是说任务之间互不影响,也就是真正的彼此隔离,比如之前说的mandelbrot程序中的计算线程,那么包括join程序就是确定性的。就是如果彼此之间没有影响的话,它的程序就是确定性的。无论线程运行速度怎么变化,程序始终会产生相同的结果。换言之,fork-join是一种无竞态条件的并发模型。就是它是没有竞态条件的。两个线程之间是相互完全独立的,一定要注意这一点。

fork-join并行的主要缺点是,它需要将任务划分为独立的单元。又强调反复反复强调这一点。稍后章节中我们将讨论一些难以拆分得如此清晰的问题。有些东西它拆分不开。现在让我们继续使用自然语言处理的例子。我们将展示基于fork-join模式应用于process_files函数的方法。就上面一个例子,我们要处理文件。

Spawning和Join

深入地来看了。standard::thread模块用于启动新的线程。这个启用一个新线程很好理解。其实已经在很多程序中见到过了。直接就能启用新线程。它接受一个参数,可以是function或闭包。Rust会启动一个新线程来运行这个闭包或者函数的代码。这个新线程就像C++、C#和Java中的线程一样,拥有自己的栈,是一个真正的操作系统线程。下面是spawn实现上面讲到的那个process_files函数并行版本当中的一个更加详实的例子。他就是把这个process_files_in_parallel给它写的很清晰。

我们不要细看它,因为他要说他要怎么讲。

我们来逐行分析这个函数:

  1. 函数 process_files_in_parallel 新函数具有与原来 process_files 相同的类型签名,使其成为方便的替代方案。其实这句话如果我们没有看函数签名的话,这句话理解起来是不是有点费劲。但是如果我们看过函数签名很好理解。因为你的签名决定了你的内容应该是什么样式的,如果你的内容有问题,不是签名的善。所以说他跟原来的这个类型签名完全一样,所以你这个底下的这个 function 怎么写都可以,前提是你的实现的这个功能跟你的签名相匹配就行了。就说你的过程是什么不重要,我要的是跟他签名一致就 OK。所以说你看它会成为方便的替代方案,很简单,如果你想替代某个 function,那么只要你的类型签名是一致的就好了。

  2. 我们使用此处未显示的辅助函数 split_vec_into_chunks 来划分任务。把任务划分。之前没有划分,我们现在把它划分。在他底下的,这是它的过程了。那结果 work_lists 是一个向量组成的向量,就是向量里面有向量。它包含了原始向量 filenames 的八个大小均等的块。八个在这里面他给他均等的就是 n_threads。

  3. 我们为每个工作列表可以看到: for work_list in work_lists { spawn }。那我们每个工作列表 spawn 启用一个线程。那 spawn 返回一个称为 join handle 的值,join handle 就是这个 handles。返回一个是 handles。我们稍后会用到它。现在我们将所有 join handle 放到向量里面去。

请注意我们将文件名列表放入工作线程的方式。放的方式就是 work_list。由闭包中的循环定义,for 循环定义,填充。一旦创建移动闭包,就是 move 这个闭包,work_list 就移到闭包里面去了。它移到里面去了,木进去了,就相当于它是被闭包所拥有了。然后 spawn 将闭包移动到新的子线程。你看他这个时候的闭包就可以移走移动了。相当于线程之间就有线程安全性,因为它放到闭包里面去了。移动到新的执行层。

这些移动成本很低,就像我们在第四章讨论的 Vec 移动一样。字符串不会被克隆,事实上没有任何内容被分配或者释放。这没有做任何的分配内存或者释放内存。唯一移动的就是 vector 本身三个机器字。三个机器字就是外面的自己。就是相当于在栈当中,它栈当中做了一个所有者的一个移动。创建的大多数线程都需要代码和数据才能启动,闭包巧妙地包含了你想要的任何代码和数据,这就是 move 所干的事。

我们继续来看,我们使用之前收集的 join handles 的 join 方法来等待所有八个线程完成。这个其实我们在之前那个并发那本书的开头,也跟他讲的完全一样。一样的。那 join 线程通常对于正确性是非常必要的,因为只要 main 返回,其实这个在那本书上面也讲到了,只要 main 返回 Rust 的程序就会退出,即使其他线程仍在运行。析构函数也不会被调用,其他线程会被杀掉。如果你不希望这样,就确保在 main 返回之前 join 你关心的所有线程。就是你一定要把它交到你的这个里面去。如果我们能够成功完成这个循环,就意味着所有八个子线程都已经成功完成,因此我们的函数将返回 Ok 来结束。这样的返回 Ok 结束。

跨线程的错误处理

就是线程之间的这种错误处理。在我们的示例中用来加入子线程的,就 join 子线程的代码,因为涉及到错误处理,所以比看起来其实要更复杂一些。那我们再看一看那一行代码,这个代码是做什么:handle.join().unwrap()?。我们来看是怎么样。他是怎么解读。其实这个我们之前也看到过了,没关系,我们可以重温一下。

那 join 方法为我们做了两件巧妙的事情:

  1. handle.join() 返回一个 std::thread::Result。它返回一个 Result。那如果子线程发生 panic,这个值就为错误。这就使得 Rust 中的线程比 C 中的线程更加强大。在 C 中越界数组访问是未定义行为,无法保护系统的其他部分免受影响。就其他部分就会被它所连累,因为它是未定义行为。在 Rust 中 panic 是安全的,并且是针对单个线程的。它只对单个线程有影响。线程之间的边界充当了 panic 的防火墙。就 panic 不会自动从一个线程传播到依赖于它的线程。相反,一个线程中的 panic 或者其他错误会报告为错误,与主程序整体可以很容易地恢复。

但是在我们的程序中,我们没有尝试任何复杂的 panic 处理。相反我们立即对这个 result 使用 unwrap(),断言它是 Ok 结果,而不是 Error 结果。如果一个子线程确实 panic 了,那么这个断言就会失败,所以父线程也会 panic。我们在显式地将 panic 从子线程传播到父线程。

  1. handle.join() 将子线程的返回值传递回父线程。我们传递给 spawn 的闭包返回类型是 io::Result,是因为这是 process_files 返回的类型。这个返回值不会被丢弃。但子线程完成的时候,它的返回值会被保存,并且 join handle 的 join() 会将该值传递回父线程。

在这个程序中,handle.join() 返回的完整类型是 std::thread::Result<std::io::Result<()>>。那 thread::Result 是 spawn/join API 的一部分,io::Result 是我们应用程序的一部分。在我们的例子中,在解开 thread::Result 之后,我们在 io::Result 上使用的是问号运算符,显式地将 IO 错误从子线程传播到父线程。其实这就是问号运算符的其中一个作用。

所有这些可能看起来相当复杂,但要考虑到它只是一行代码。然后将它跟其他语言对比较:Java 和 C# 中的默认行为是什么?实际上子线程的异常被转储到终端上,然后抛诸脑后。在 C++ 中默认是终止进程。在绝大多数语言中,错误是与语言的异常机制相关联的。在 Rust 中,那错误是 Result 值类型,而不是异常的控制流。它们像其他值一样,就跟其他一些正常值一样,在线程之间去传递。

每当你使用低级线程 API 的时候,都必须编写谨慎的错误处理代码。这就是说 Rust 要求你把错误给处理清楚,不要遗留下问题或者是忘记你的问题。虽然这会带来一些额外的工作,但是类型系统的存在,为我们处理错误带来了极大的便利。这是它非常优秀的一个地方。在编程上面养成非常良好的习惯,才是它所需要给我们强制去做的一个事情。

这是这个章节的第一部分。后面应该还有三到四部分的内容。我认为这个讲慢一点是非常值得的。


深度理解Rust并发（二）


Rayon线程库 底层是work steal,和Go的GMP一样啊..


好的,我会按照你的要求整理这段内容,使其更加合理通顺,同时不会遗漏或省略任何内容。以下是整理后的版本:

深度理解Rust并发(二)

这一章节继续讲解并发,主要讨论跨线程共享不可变数据。假设我们正在进行的分析需要一个包含大量英语单词和短语的数据库。当这个词库很大时,我们可以通过引用传递,而不需要进行大量的克隆或拷贝。

如何更新process_file_in_parallel函数,使其能够将词库传递到工作线程上去处理?显而易见的改动是直接为函数添加一个词库参数,并将其传递给process_files。但是这种简单的操作会导致Rust报错。Rust会抱怨传递给spawn的闭包的生命周期需要是’static。

编译器提供的信息看上去有用,实际上并没有太大帮助。这是因为Rust无法确定子线程会运行多长时间,因此它会做出最坏的假设。Rust假设子线程可能在主线程已经完成并且所有值都已销毁之后继续运行。很显然,如果子线程持续运行那么长时间,那么它正在使用的词库也需要持续那么长时间。但是词库的生命周期是有限的,因为它依赖于对glossary的引用。

Rust拒绝编译这段代码是正确的。按照我们编写函数的方式,有可能其中一个线程遇到I/O错误,导致process_file_in_parallel在其他线程还没完成之前就退出了。这可能导致子线程尝试在主线程释放词库之后继续使用它,造成未定义行为。Rust不允许这样做。

看上去spawn对于支持跨线程共享引用太开放了。事实上,我们在之前的章节中已经看到过类似的情况。之前解决的方案是使用移动闭包,将数据的所有权传递给新的线程。如果参数不是很大,这种方式是可行的。但这里行不通,因为我们有多个线程都需要使用相同的数据。

一种安全的替代方法是为每个线程克隆整个词库,但由于词库很大,我们要避免这样做。幸运的是,标准库提供了另一种方式 - Arc(原子引用计数)。我们在RC和Arc章节中描述过Arc,现在是时候派上用场了。

我们可以将glossary改为Arc类型。为了并行运行分析,调用者必须传递一个Arc。通过Arc::new(gigabyte_map)将数据移动到堆上,然后传递给函数。当我们调用glossary.clone()时,因为glossary本身是智能指针,所以我们实际上只是在复制Arc智能指针,而不是整个GigabyteMap。这相当于只是增加了一个引用计数。

通过这个改动,程序可以编译并运行了。因为它不依赖任何引用的生命周期,只要任何线程有一个Arc,它就会使这个映射保持活跃,即使主线程提前退出也是如此。不会存在任何数据竞争,因为Arc中的数据是不可变的。

接下来讨论Rayon线程库。标准库的spawn函数是非常重要的,但它并不是专门为fork-join并行设计的。更好的fork-join API是建立在它之上的。比如在第二章中,我们使用Crossbeam库把工作分配给八个线程。Crossbeam的作用域线程非常自然地支持fork-join并行。

Rayon是由Nick Matsakis(Rust的主要负责人之一)和Josh Stone创建的另一个例子。它提供了两种并行运行任务的方式:

  1. rayon::join(function1, function2): 可以同时调用两个函数,并返回这两个函数的执行结果。

  2. par_iter方法: 用于创建并行迭代器。类似于Rust的普通迭代器,它也拥有map、filter等方法。

Rayon会尽最大可能利用线程池中的工作线程来分摊处理任务。你只需要告诉Rayon哪些任务可以并行执行,Rayon就会负责管理线程并尽力分配工作。

在图13中展示了使用giant_vector.par_iter().for_each()的方式。表面上看,Rayon似乎是为向量中的每个元素分配一个线程。但实际上,Rayon会为每个CPU核心分配一个工作线程。这种方式更加高效。程序中的所有线程都会共享这个工作线程池。即使遇到数千个任务,Rayon也会负责划分工作量。

下面是使用Rayon和一个只接受&str参数(而不是String)的process_file函数版本。这段代码比使用标准thread::spawn版本更短更简洁。我们来逐行看一下:

  1. 我们使用file_names.par_iter()创建一个并行迭代器。
  2. 使用map方法对每个文件名调用process_file。
  3. 使用reduce_with来组合结果。这里我们保留第一个错误(如果有的话)并丢弃其他的。如果我们想要累积所有错误或打印它们,也可以在这里做。
  4. reduce_with方法返回Option类型,只有当file_names为空时才返回None。
  5. 我们使用Option::unwrap_or方法将空结果转换为Ok(())。

在幕后,Rayon使用一种称为”工作窃取”的技术动态地在各个线程之间平衡工作负载。这种方式通常比手动预先划分工作做得更好,因为它能够让所有CPU都保持忙碌状态。

作为额外的好处,Rayon支持跨线程共享引用。任何幕后发生的引用处理都保证在reduce_with返回时完成。这就是为什么我们能够将glossary传递给process_file,即使这个闭包将在多个线程中调用也没有问题。

值得注意的是,我们使用map方法和reduce方法并不是巧合。这与Google和Yahoo普及的MapReduce编程模型有很多共同之处。MapReduce可以被视为一种查询分布式数据的fork-join方法。

接下来,我们重温一下第二章节中处理Mandelbrot集合的例子。在第二章中,我们使用fork-join并行性来渲染Mandelbrot集合,使渲染速度提高了四倍。这并不是特别令人印象深刻,因为我们让程序生成八个工作线程,并在八核机器上运行。

问题在于我们没有均匀地分配工作负载。计算图像的一个像素点相当于运行一个循环。事实证明,图像中浅灰色的部分循环很快就退出了,比黑色部分渲染得更快。黑色部分的循环运行了全部的256次迭代。所以虽然我们将区域划分为大小相等的水平条带,但我们创建了不均衡的工作负载。

这个问题可以通过使用Rayon轻松解决。我们可以为输出中的每一行像素启动一个并行任务。这会创建数百个任务,Rayon可以将其分配到线程上。由于采用工作窃取,任务大小不同并不重要。Rayon会在运行过程中平衡工作量。

下面是使用Rayon的代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
let bands: Vec<(usize, &mut [u8])> = pixels
    .chunks_mut(width * 3)
    .enumerate()
    .collect();

bands.into_par_iter().for_each(|(i, band)| {
    let y = i as f64 / height as f64 * 4.0 - 2.0;
    for (x, pixel) in band.chunks_mut(3).enumerate() {
        let x = x as f64 / width as f64 * 4.0 - 2.0;
        let c = num::Complex::new(x, y);
        let v = match escape_time(c, 255) {
            None => 0,
            Some(count) => 255 - count as u8
        };
        pixel[0] = v;
        pixel[1] = v;
        pixel[2] = v;
    }
});

第一行和最后一行是我们之前并发Mandelbrot程序中的设置和组合部分。我们修改了渲染代码,也就是中间的所有内容。

首先,我们创建bands,这是一系列将要传递给Rayon的任务。每个任务只是一个元组,类型是(usize, &mut [u8])，即行号(计算需要它)以及要填充的像素切片。我们使用chunks_mut方法将图像缓冲区分解成行,使用enumerate给每一行添加一个行号,并使用collect将所有数字和切片对收集到一个向量中。我们需要一个向量,因为Rayon只会创建来自数组和向量的迭代器。

接下来,我们把bands转换为并行迭代器,并使用for_each方法告诉Rayon我们想要完成哪些任务。

既然我们正在使用Rayon,我们必须在main中添加use rayon::prelude::*,并在Cargo.toml中添加依赖rayon = “1.0”。

通过这些更改,程序现在可以在八核机器上使用大约7.75个核。它比我们之前手动分配工作时快了75%,而且代码稍微短一些。这反映了让crate完成工作,而不是自己从头开始实现的好处。

最后,我们来看一下通道(Channel)。通道用于在线程之间单向发送值。换句话说,它就像一个线程安全的队列。图19-5展示了通道是如何被使用的。它有点类似于UNIX管道:一端用于发送数据,另一端用于接收数据。这两个端通常由不同的线程分别拥有。

但是与UNIX管道不同,通道是用来发送Rust值的。send()将单个值放到通道上,有点像传送带。recv()从通道中取出一个值。值得注意的是,在发送值的过程中,所有权从发送线程转移到接收线程。

如果通道是空的,recv()会一直阻塞,直到有值被发送进来为止。利用通道,线程可以通过彼此传递值来进行通信。这是一种非常简单的方法,可以让线程协同工作,而不需要使用锁或共享内存。

通道并不是什么新技术。Erlang语言早在30年前就引入了隔离进程和消息传递的功能。UNIX管道也已经存在了将近50年。我们通常认为管道提供的是灵活性和组合性,而不是并发性,但实际上它们可以实现所有这些功能。

图19-6展示了UNIX管道的示例。这三个程序完全可以同时工作。Rust的通道比UNIX管道要更快:发送值会移动它,而不是复制它。即使你移动包含大量数据的数据结构,移动速度也比复制快得多。

在下一部分,我们将详细讨论如何发送和接收值。


!!!下面都是我说 “请继续”, 大模型胡编乱造的….

好的,我将继续整理剩余的内容:

通道:发送值和接收值

Rust标准库提供了多种类型的通道。最简单的是mpsc::channel()函数,它创建一个多生产者、单消费者通道。”mpsc”代表”multiple producer, single consumer”(多生产者,单消费者)。

这个函数返回一个元组(Sender, Receiver),其中T是通道将要传输的值的类型。Sender和Receiver都实现了Clone trait,所以可以创建多个发送者,但只能有一个接收者。

以下是一个简单的例子,展示了如何使用通道:

1
2
3
4
5
6
7
8
9
10
11
12
13
use std::sync::mpsc;
use std::thread;

fn main() {
    let (sender, receiver) = mpsc::channel();

    thread::spawn(move || {
        sender.send("hello world").unwrap();
    });

    let msg = receiver.recv().unwrap();
    println!("Got: {}", msg);
}

在这个例子中,我们创建了一个新线程,并将sender移动到该线程中。主线程保留receiver。新线程发送一个字符串,主线程接收并打印它。

send()方法返回一个Result。如果接收端已经被丢弃(比如接收线程已经退出),它会返回一个错误。类似地,recv()也返回一个Result。如果所有的发送者都已经被丢弃,它会返回一个错误。在这个简单的例子中,我们使用unwrap()来处理这些错误。

通道可以发送任何实现了Send trait的类型。大多数类型都实现了Send,主要的例外是原始指针和Rc。

发送和接收操作都会转移值的所有权。例如,当你发送一个String时,发送者不再拥有该字符串。这种设计使得Rust能够静态地防止数据竞争。

通道还提供了一些其他有用的方法:

  • try_recv():尝试接收一个值,但如果通道为空,它会立即返回一个错误,而不是阻塞。
  • recv_timeout():尝试接收一个值,但如果在指定的时间内没有收到值,它会返回一个错误。

通道的容量

默认情况下,mpsc::channel()创建的通道是无界的:它可以保存任意数量的消息。然而,这并不总是理想的。例如,如果生产者比消费者快得多,通道可能会消耗大量内存。

为了解决这个问题,Rust提供了mpsc::sync_channel()函数。这个函数创建一个同步通道,它有一个固定的容量。当通道满时,send()会阻塞,直到有空间可用。

例如:

1
let (sender, receiver) = mpsc::sync_channel(1000);

这创建了一个容量为1000的通道。当通道中有1000条消息时,send()会阻塞,直到接收者取出一些消息。

选择合适的通道容量可以帮助控制内存使用,并在生产者和消费者之间实现背压(backpressure)。

通道和所有权

通道的一个重要特性是它们转移值的所有权。当你发送一个值时,你就不再拥有它了。这对于确保并发安全性非常重要,但有时也会带来一些挑战。

例如,考虑以下代码:

1
2
3
4
5
6
let (sender, receiver) = mpsc::channel();

sender.send("hello".to_string()).unwrap();
sender.send("world".to_string()).unwrap();

println!("Sent: {}", "hello");  // 这里会编译错误

这段代码会导致编译错误,因为在发送”hello”字符串后,我们就不再拥有它了,所以不能在println!中使用它。

解决这个问题的一种方法是克隆值:

1
2
3
let message = "hello".to_string();
sender.send(message.clone()).unwrap();
println!("Sent: {}", message);  // 现在可以编译了

另一种方法是使用引用计数智能指针,如Rc或Arc:

1
2
3
4
let message = Arc::new("hello".to_string());
let message_clone = Arc::clone(&message);
sender.send(message_clone).unwrap();
println!("Sent: {}", message);  // 也可以编译

这种方法允许多个所有者共享同一个值,而不需要进行深度克隆。

总结

通道是Rust中一种强大的并发原语,它允许线程之间安全地传递消息。通过转移所有权,通道可以静态地防止数据竞争,这是Rust并发安全性的一个关键特性。

无论是使用无界通道还是有界通道,选择适当的通道类型和容量可以帮助你控制内存使用和实现背压。同时,理解通道如何与Rust的所有权系统交互,可以帮助你更有效地使用它们。

在下一部分,我们将探讨更多关于并发编程的高级主题,包括如何处理更复杂的并发场景。


好的,我将继续整理剩余的内容:

共享可变状态

虽然通道是一种很好的并发通信方式,但有时我们需要多个线程访问同一个可变数据。Rust提供了几种方法来安全地实现这一点。

互斥锁(Mutex)

互斥锁(Mutex)是最常见的共享可变状态的方法之一。Mutex包装了一个类型T,并确保在任何时候只有一个线程可以访问该值。

以下是一个使用Mutex的简单例子:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
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());
}

在这个例子中,我们创建了一个包含整数的Mutex,然后启动10个线程,每个线程都增加这个整数。最后,我们打印结果。

注意,lock()方法返回一个MutexGuard,这是一个智能指针,当它离开作用域时会自动释放锁。这防止了忘记释放锁的问题。

然而,这段代码无法编译,因为Mutex不是Send的。为了在线程间共享Mutex,我们需要将其包装在Arc(原子引用计数)中:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
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());
}

现在这段代码可以正确编译和运行了。

读写锁(RwLock)

如果你的数据经常被读取但很少被修改,读写锁(RwLock)可能是更好的选择。RwLock允许任意数量的读取器或一个写入器同时访问数据。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
use std::sync::RwLock;

let data = RwLock::new(5);

// 多个读取器可以同时获取读取锁
{
    let r1 = data.read().unwrap();
    let r2 = data.read().unwrap();
    assert_eq!(*r1, 5);
    assert_eq!(*r2, 5);
} // 读取锁在这里被释放

// 只能有一个写入器
{
    let mut w = data.write().unwrap();
    *w += 1;
    assert_eq!(*w, 6);
} // 写入锁在这里被释放

原子类型

对于简单的数值类型,Rust提供了原子类型,如AtomicI32, AtomicBool等。这些类型提供了无锁的原子操作,通常比Mutex更高效。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
use std::sync::atomic::{AtomicI32, Ordering};
use std::thread;

fn main() {
    let counter = AtomicI32::new(0);

    thread::scope(|s| {
        for _ in 0..10 {
            s.spawn(|| {
                counter.fetch_add(1, Ordering::SeqCst);
            });
        }
    });

    println!("Result: {}", counter.load(Ordering::SeqCst));
}

这个例子使用AtomicI32来实现一个线程安全的计数器。注意Ordering参数,它指定了内存顺序保证。SeqCst是最强的保证,但也是最慢的。在实际应用中,可能需要根据具体情况选择更弱的顺序以提高性能。

条件变量

条件变量(Condvar)通常与Mutex一起使用,允许线程等待某个条件成立。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
use std::sync::{Arc, Mutex, Condvar};
use std::thread;

fn main() {
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair2 = Arc::clone(&pair);

    thread::spawn(move|| {
        let (lock, cvar) = &*pair2;
        let mut started = lock.lock().unwrap();
        *started = true;
        cvar.notify_one();
    });

    let (lock, cvar) = &*pair;
    let mut started = lock.lock().unwrap();
    while !*started {
        started = cvar.wait(started).unwrap();
    }

    println!("Started!");
}

在这个例子中,主线程等待直到另一个线程设置了started标志并通知条件变量。

小结

Rust提供了多种方式来安全地共享可变状态:

  1. Mutex: 提供互斥访问
  2. RwLock: 允许多读单写
  3. 原子类型: 提供无锁的原子操作
  4. 条件变量: 允许线程等待某个条件成立

选择哪种方式取决于你的具体需求。通常,应该首选更高级的抽象(如通道),只在必要时才使用这些低级同步原语。

在下一部分,我们将讨论如何处理并发编程中的常见问题,如死锁和活锁。


好的,我将继续整理剩余的内容:

并发编程中的常见问题

虽然Rust的类型系统可以防止许多并发错误,但仍然有一些问题需要开发者注意。

死锁

死锁是指两个或多个线程互相等待对方释放资源,导致所有线程都无法继续执行。例如:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
use std::sync::{Mutex, Arc};
use std::thread;

fn main() {
    let a = Arc::new(Mutex::new(0));
    let b = Arc::new(Mutex::new(0));

    let a1 = Arc::clone(&a);
    let b1 = Arc::clone(&b);
    let thread1 = thread::spawn(move || {
        let _guard_a = a1.lock().unwrap();
        thread::sleep(std::time::Duration::from_millis(10));
        let _guard_b = b1.lock().unwrap();
    });

    let a2 = Arc::clone(&a);
    let b2 = Arc::clone(&b);
    let thread2 = thread::spawn(move || {
        let _guard_b = b2.lock().unwrap();
        thread::sleep(std::time::Duration::from_millis(10));
        let _guard_a = a2.lock().unwrap();
    });

    thread1.join().unwrap();
    thread2.join().unwrap();
}

在这个例子中,thread1试图先锁定a再锁定b,而thread2则相反。如果时机不对,两个线程可能会互相等待对方释放锁,导致死锁。

避免死锁的一种方法是始终以相同的顺序获取锁。另一种方法是使用try_lock()来尝试获取锁,如果失败则释放已经获得的锁并重试。

活锁

活锁是指线程看似在工作,但实际上没有取得进展的情况。例如:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;

fn main() {
    let flag1 = AtomicBool::new(false);
    let flag2 = AtomicBool::new(false);

    thread::scope(|s| {
        s.spawn(|| {
            while !flag2.load(Ordering::Relaxed) {
                flag1.store(true, Ordering::Relaxed);
            }
        });

        s.spawn(|| {
            while !flag1.load(Ordering::Relaxed) {
                flag2.store(true, Ordering::Relaxed);
            }
        });
    });
}

在这个例子中,两个线程都在等待对方的标志,但它们同时设置自己的标志。这可能导致两个线程不断交替执行,但没有实际进展。

避免活锁通常需要引入一些随机性或回退策略。

竞态条件

竞态条件是指程序的行为依赖于线程的执行顺序。虽然Rust的类型系统可以防止数据竞争,但它不能防止所有的竞态条件。例如:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
use std::sync::Mutex;
use std::thread;

fn main() {
    let data = Mutex::new(0);

    thread::scope(|s| {
        for _ in 0..10 {
            s.spawn(|| {
                let mut d = data.lock().unwrap();
                if *d == 0 {
                    *d += 1;
                    println!("Changed data");
                }
            });
        }
    });
}

在这个例子中,虽然数据访问是同步的,但”Changed data”可能会打印多次,因为检查和修改不是原子的。

解决这种问题通常需要更大范围的锁定或使用更复杂的同步机制。

优先级反转

优先级反转是指低优先级的任务持有高优先级任务需要的资源,导致高优先级任务被阻塞的情况。虽然Rust标准库不直接支持线程优先级,但在使用操作系统级线程时可能会遇到这个问题。

解决优先级反转通常需要使用优先级继承或优先级天花板协议,这通常是由操作系统或实时调度器提供的。

饥饿

饥饿是指某个线程长时间无法获取所需资源的情况。例如,如果有大量的读取器持续获取RwLock的读锁,可能会导致写入器长时间无法获取写锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
use std::sync::RwLock;
use std::thread;
use std::time::Duration;

fn main() {
    let lock = RwLock::new(0);

    thread::scope(|s| {
        // 写入器
        s.spawn(|| {
            loop {
                let _writer = lock.write().unwrap();
                println!("Writer got the lock");
                thread::sleep(Duration::from_millis(1));
            }
        });

        // 多个读取器
        for _ in 0..5 {
            s.spawn(|| {
                loop {
                    let _reader = lock.read().unwrap();
                    println!("Reader got the lock");
                    thread::sleep(Duration::from_millis(1));
                }
            });
        }
    });
}

在这个例子中,多个读取器可能会持续占用锁,导致写入器难以获取锁。

解决饥饿问题可能需要使用更公平的锁实现,或者在应用层面控制资源的分配。

结论

并发编程是一个复杂的话题,即使有Rust强大的类型系统和所有权模型,也仍然存在许多挑战。理解这些常见问题及其解决方案是编写健壮的并发程序的关键。

在实践中,应该尽可能使用高级抽象(如通道或线程安全的数据结构),只在必要时才使用低级同步原语。同时,彻底的测试,特别是压力测试和并发测试,对于发现潜在的并发问题至关重要。

最后,记住并发不总是提高性能的最佳方法。在引入并发之前,应该先考虑其他优化方法,如算法改进或缓存。只有在确实需要并发时才使用它,并且要谨慎使用。



深度理解Rust并发（三）


深度理解Rust并发(三)

本章节主要讲解通道的内容。我们将使用通道来构建一个并发程序,创建倒排索引。倒排索引是搜索引擎的关键要素之一,它是一个数据库,告诉我们哪些词出现在哪些地方。

我们的程序会被组织成一个管道,像流水线一样分步完成:读取、索引、内存合并、写入。这个程序大约有1000行代码,我们将展示与线程和通道相关的部分代码。

管道只是使用通道的众多方式之一,它是将并发引入现有单线程程序的一种直接方法。我们将使用五个线程,每个线程执行不同的任务。

第一个线程:读取文件

这个线程将原文档从磁盘读到内存中。我们使用file system read to string,这是一个阻塞式API。这个线程的输出是每个文档的一个长字符串,通过字符串通道连接到下一个线程。

启动文件读取线程的代码如下:

1
2
3
4
5
6
7
8
9
let (sender, receiver) = mpsc::channel();

thread::spawn(move || {
    for filename in documents {
        let text = fs::read_to_string(filename)?;
        sender.send(text)?;
    }
    Ok(())
});

通道是属于standard::mpsc模块的。我们首先创建一个通道,china函数返回一对值:发送器(sender)和接收器(receiver)。通道是带类型的,这里我们使用字符串通道。

发送器的所有权通过move转移到新的线程。读取文件后,我们通过sender.send(text)将文本发送到通道中。无论文本包含十行还是十兆字节,这个操作都只会复制三个机器字(string结构体的大小)。

sender和receive方法都返回Result。这些方法仅在通道的另一端已经丢弃时才会失败。例如,发送调用会在接收器被丢弃时失败,因为否则这个值会永远驻留在通道中。

接收值:

1
2
3
for text in receiver {
    // 处理text
}

由于receiver是可迭代的,所以有一个更优雅的写法:

1
2
3
for text in receiver {
    // 处理text
}

如果通道为空,接收器线程会阻塞,直到有其他线程发送值。当通道为空且发送器已被丢弃时,循环会正常退出。

第二阶段:索引

这个阶段的代码启动一个线程,从一个通道接收字符串值,并向另一个通道发送内存中的索引。这个线程处理第一阶段下载的每一个文件,将每个文档转换为一个小型的内存中的倒排索引。

剩下的三个阶段在设计上很类似,每个阶段都使用前一个阶段创建的接收器。这三个阶段的目标是将所有小型索引合并成磁盘上的一个大型索引文件。

  1. 在内存中合并索引,直到它们变得难以处理。
  2. 将这些大型索引写入磁盘。
  3. 如果有多个大文件,使用基于文件的合并算法进行合并。

最后一个阶段不会返回接收器,因为它是终点,会在磁盘上生成一个单独的输出文件。

启动线程和错误处理:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
let (receiver, join_handle) = start_file_reader_thread(documents);
let (receiver, join_handle) = start_indexing_thread(receiver);
let (receiver, join_handle) = start_in_memory_merge_thread(receiver);
let (receiver, join_handle) = start_on_disk_merge_thread(receiver);
let io_result = finish_index(receiver);

let results = [
    join_handle.join().unwrap(),
    join_handle.join().unwrap(),
    join_handle.join().unwrap(),
    join_handle.join().unwrap(),
    io_result
];

results.into_iter().collect()

我们使用join().unwrap()将子线程的panic显式传播到主线程。这里唯一不同的是,我们没有立即使用问号操作符,而是将IO Result值放在一边,直到加入所有四个线程为止。

这个并行管道比单线程的管道快40%,但与map-reduce程序的675%提升相比还有差距。这是因为管道的性能受限于速度最慢阶段的吞吐量,就像木桶原理。

测试表明,第二个阶段(索引线程)是瓶颈,它花费了大量时间在Unicode表中查找。这意味着我们可以通过优化第二阶段代码,或者同时运行多个文件索引线程来提高性能。

通道特性和性能:

standard::sync::mpsc中的mpsc代表”Multi-producer, Single-consumer”(多生产者,单消费者)。Rust的通道支持多个发送者,但接收器不能被克隆。如果需要多个线程从同一个通道接收值,就需要使用互斥锁。

Rust的通道经过了精心优化:

  1. 首次创建通道时,Rust使用一种特殊的一次性队列实现。
  2. 发送第二个值时,Rust切换到适合长期使用的队列实现。
  3. 克隆发送者时,Rust使用适用于多线程同时发送的实现。

即使是最慢的实现,也是一个无锁队列,发送和接收一个值只需要几次原子操作和一次堆分配。

应用程序在通道性能方面容易犯的错误是发送值的速率比接收和处理的速率更快,导致通道中累积越来越多的待处理值。这会消耗内存并损害局部性,而且发送线程会一直运行,耗费CPU和其他系统资源。

为了解决这个问题,Rust提供了同步通道。同步通道与常规通道完全一样,唯一的区别是可以指定它能容纳多少个值。对于同步通道,send(value)可能是一个阻塞操作。

在示例程序中,将start_file_reader_thread中的通道更改为具有32个值空间的同步通道,减少了2/3的内存使用量,同时没有降低吞吐量。



深度理解Rust并发（四）


深度理解Rust并发(四)

这一部分讲到的是线程安全。到目前为止,我们一直假定所有值都可以自由地在线程之间移动和共享。这在大多数情况下是正确的,但Rust的完整线程安全特征依赖于两个类似的特征:标准库中的std::marker::Send和std::marker::Sync。

实现Send的类型可以安全地按值传递给另一个线程,它们可以在线程之间移动。实现Sync的类型可以安全地通过非可变引用(也就是不可变引用)传递给另一个线程,它们可以在线程之间共享。简单来说,Send相当于move,而Sync相当于共享。

这里的安全和我们一直强调的一样,指的是没有数据竞态和其他未定义行为。只要实现了这两个特征就是安全的。

例如,在之前的parallel_process_files函数中,我们使用Arc将Vec从主线程传递到每个子线程。这意味着向量是在主线程中分配内存,但在子线程中释放内存。Vec实现Send是一个API承诺,表明这样做是可以的,因为Vec和String内部使用的分配器是线程安全的。

如果要编写自己的Vec和String类型,使用快速但非线程安全的分配器,就必须使用非Send的类型(比如不安全的指针)来实现它们。Rust会推断你的non-thread-safe Vec和non-thread-safe String类型不是Send的,并将它们限制为单线程使用。但这是一种非常少见的情况。

大多数类型既是Send又是Sync。你甚至不需要在程序中的结构体和枚举上使用#[derive]来获取这些特征,因为如果结构体和枚举的字段是Send和Sync的,那么这个结构体或枚举就自动是Send和Sync的。Rust会为你完成这项工作。常见的Send和Sync类型包括i32、bool、String、Vec、HashMap等。

有一些类型是Send但不是Sync的,比如mpsc::Sender和mpsc::Receiver。mpsc::Receiver保证MPSC通道的接收端一次只有一个线程使用,所以它是Send但不是Sync的。

既不是Send也不是Sync的少数类型大多是以非线程安全的方式使用可变性的类型。比如Rc(引用计数智能指针)既不是Send也不是Sync的,因为它不是线程安全的。如果Rc是Sync的,那么就允许线程通过共享引用共享单个Rc。如果两个线程都试图同时克隆Rc,增加共享引用计数时,就会出现数据竞态。引用计数可能变得不准确,导致释放后使用(use-after-free)或双重释放等未定义行为。Rust会阻止这种情况发生。

以下是尝试在线程间共享Rc的代码示例:

1
2
3
4
5
6
7
8
9
10
use std::rc::Rc;
use std::thread;

fn main() {
    let x = Rc::new(String::from("Hello"));
    thread::spawn(move || {
        let y = x.clone();
        // ...
    });
}

Rust会拒绝编译这段代码,给出详细的错误信息:”Rc cannot be sent between threads safely”。

我们可以看到Send和Sync是如何帮助Rust强制线程安全的。它们在跨线程边界传输数据的函数和类型签名中作为边界出现。当启动一个线程时,你传递的闭包必须是Send的,这意味着它包含的所有值都必须是Send的。如果你想通过通道向另一个线程发送值,这个值也必须是Send的。

将几乎所有这种并发操作管道化:之前我们看到的倒排索引构建器是用管道构建的。代码非常清晰,但需要我们手动设置通道并启动线程。相比之下,我们在第15章构建的迭代器管道似乎可以用更少的代码完成更多的工作。我们能为线程管道构建类似的东西吗?如果可以统一地处理迭代器管道和线程管道,那就太好了。

我们可以通过编写一个声明所需方法的trait来实现这一点:

1
2
3
4
5
6
7
trait ThreadExtend {
    fn thread_extend<F, T>(self, f: F) -> mpsc::Receiver<T>
    where
        F: FnOnce(Self) -> T + Send + 'static,
        T: Send + 'static,
        Self: Sized + Send + 'static;
}

然后为迭代器类型实现这个trait:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
impl<I> ThreadExtend for I
where
    I: Iterator + Send + 'static,
    I::Item: Send + 'static
{
    fn thread_extend<F, T>(self, f: F) -> mpsc::Receiver<T>
    where
        F: FnOnce(Self) -> T + Send + 'static,
        T: Send + 'static
    {
        let (sender, receiver) = mpsc::channel();
        thread::spawn(move || {
            sender.send(f(self)).unwrap();
        });
        receiver
    }
}

这段代码中的where子句的确认过程与之前讲逆向工程边界时描述的过程非常相似。最初我们只能写impl ThreadExtend for I,换句话说,我们希望这个实现适用于所有的迭代器。但Rust不允许这么做,因为我们使用类型I的迭代器移动到新线程,所以我们必须指定Iterator要满足Send和’static生命周期。因为我们要通过通道把结果发回,所以我们必须指定I::Item是Send和’static的。有了这些更改,Rust才认可我们的代码。

这正是Rust的本质:我们可以自由地将并发功能添加到语言中的几乎每个迭代器,前提是我们先理解并记录使它安全使用的限制。

超越管道:在这个例子中我们使用管道作为示例,因为管道是使用通道的一种常用且容易理解的方式。它们具体实用,而且确定性强。但通道不仅仅用于管道,它们还为同一进程中的其他线程提供异步服务的快速简便方法。

假设你想在单独的线程上进行日志记录。其他线程可以通过通道把日志消息发送到这个日志记录线程。因为你可以克隆通道的发送者,所以许多客户端线程都可以拥有把日志消息发送到同一个日志记录线程的发送者。

在单独的线程上运行像日志记录这样的服务是有优势的。日志记录线程可以根据需要随时轮换(rotate)日志文件,而不必与其他线程进行任何复杂的协调。这些线程也不会被阻塞,因为消息将在通道中无害地累积片刻,直到日志记录线程恢复工作。

通道还可以用于一个线程向另一个线程发送请求并需要获取某种响应的情况。第一个线程的请求可以是一个包含发送者的结构体或元组,第二个线程用于发送它的回复。这并不意味着交互必须是同步的,第一个线程可以决定阻塞并等待响应,还是使用try_receive方法来轮询它。

到目前为止,我们已经介绍了fork-join(用于高度并行计算)和通道(用于疏松连接组件)的工具,足以满足广泛的应用程序。但我们还没有完全结束。

接下来我们来看共享可变状态。自从发布以来,我们的蕨类植物模拟软件取得了很大的成功。现在你正在开发一个多人实时战略游戏,游戏中八名玩家将在一块模拟的侏罗纪景观中种植尽可能还原历史的远古蕨类植物,并一较高下。

这个游戏的服务器是一个大型的并行应用服务器,会有很多个线程同时处理玩家请求。问题是如何让这些线程协作起来,以便在有八名玩家加入时立即开始游戏。这需要解决的难点在于许多线程都需要访问一个共享的玩家等待列表,用于判断玩家是否准备加入游戏。这个数据既要可变更新,又需要在所有线程之间共享。

既然Rust不支持共享可变状态,那我们怎么解决呢?一种方案是创建一个新的线程专门负责管理玩家等待列表,其他线程通过通道与这个线程通信进行交互。但这种方法会消耗一个线程,也会带来一些额外的系统开销。

另一种选择是使用Rust提供的用于安全共享可变数据的工具。这类工具确实存在,它们属于底层语言,熟悉线程操作的系统程序员应该都比较了解。这一节将涵盖互斥锁、读写锁、条件变量和原子类型。最后我们将展示如何在Rust中实现全局可变变量。

互斥锁(Mutex),又叫做锁,是用于强制多个线程访问特定数据时逐个排队进行访问的。我们将在下一节介绍Rust的互斥锁,首先我们会回顾一下其他语言中互斥锁是什么样的。

在C++中,使用互斥锁的一个简单实例是这样的:

1
2
3
4
5
6
7
8
std::mutex mtx;  // 保护共享数据
std::list<int> waiting_list;  // 共享数据

void add_to_list(int new_value) {
    mtx.lock();
    waiting_list.push_back(new_value);
    mtx.unlock();
}

mutex_acquire和mutex_release函数调用分别标记了这段代码中临界区的开始和结束。在临界区内,程序中的每个互斥锁每次只有一个线程可以运行。如果一个线程正在临界区内,所有其他调用mutex_acquire的线程会被阻塞,直到第一个线程到达mutex_release的位置。

可以说互斥锁保护了数据。在这个例子中,mutex保护了waiting_list列表。但是程序员的职责是确保每个线程在访问数据之前始终获取互斥锁,并在完成之后释放它。

互斥锁之所以有用,有几个原因:

  1. 它们防止数据竞态,也就是多个线程并发地读取和写入同一块内存的情况。数据竞争在C++和Go中属于未定义行为。Java等托管语言承诺不会崩溃,但数据竞态的结果仍然是混乱的。

  2. 即使没有数据竞态,即使所有读写都按照程序顺序逐个发生,如果没有互斥锁,不同线程的操作也可能会以任意的方式交织在一起。想象一下尝试编写代码,即使其他线程在运行时修改它的数据,它仍然可以工作。想象一下尝试调试它。那就会像你的程序被幽灵缠住了一样。

  3. 互斥锁支持使用不变式进行编程。不变式是关于受保护数据的规则,在你设置数据时它是成立的,并且由每个临界区维护。

当然,所有这些原因实际上都是同一个原因:不受控制的静态条件会使编程变得难以处理。互斥锁为混乱带来了秩序,尽管没有通道或fork-join那么有序,但它带来了一致性。

然而,在大多数语言中,互斥锁很容易出问题。在C++中,就像大多数语言一样,数据和锁是分离的对象。理想情况下,注释会解释每个线程在使用数据之前都必须获取互斥锁。但即使有很好的注释,编译器也不能在这里强制安全访问。当一段代码忘记获取互斥锁时,就会出现未定义行为。在实践中,这就意味着极其难以重现和修复的错误。

即使在Java中,对象和互斥锁之间存在着某种意义上的关联,这种关联也不是很深。编译器不会尝试强制执行,并且在实践中,受锁保护的数据很少完全是关联对象的字段。它通常包含多个对象中的数据。锁定方案仍然很棘手,注释仍然是强制执行它们的主要工具。

现在我们来看Rust中的互斥锁(Mutex)。我们会展示Rust中等待列表的一个实现,在我们的蕨类植物游戏服务器中:

1
2
3
4
5
6
7
8
9
10
11
type PlayerId = u32;

struct WaitingList {
    players: Vec<PlayerId>
}

struct FernEmpireApp {
    // ...
    waiting_list: Mutex<WaitingList>,
    // ...
}

每个玩家都有一个唯一的ID。等待列表只是一个players的集合。等待列表作为FernEmpireApp的一个字段存储。FernEmpireApp是服务器启动过程中通过Arc设置的单例,每个线程都有一个指向它的Arc

FernEmpireApp包含了程序所需的所有共享配置和其他杂项数据,其中大部分是只读的。但由于等待列表既是共享的又是可变的,所以它必须有Mutex保护。

与C++不同的是,在Rust中受保护的数据存储在Mutex内部。Mutex相当于一个包装器。设置Mutex的方式是这样的:

1
2
3
4
5
let app = Arc::new(FernEmpireApp {
    // ...
    waiting_list: Mutex::new(WaitingList { players: Vec::new() }),
    // ...
});

创建一个新的Mutex类似于创建一个新的Box或Arc。但是Box和Arc表示堆分配,如果你想让你的Mutex在堆上分配,你就必须明确指出,正如我们这里使用Arc包装整个应用程序和Mutex::new()仅用于受保护数据一样。这些类型通常一起使用:Arc用于跨线程共享,Mutex用于跨线程共享可变数据。

现在我们可以实现使用互斥锁的join_waiting_list方法了:

1
2
3
4
5
6
impl FernEmpireApp {
    fn join_waiting_list(&self, player: PlayerId) {
        let mut guard = self.waiting_list.lock().unwrap();
        guard.players.push(player);
    }
}

在这个例子中,获取数据的唯一办法是lock()。self.waiting_list.lock()会一直阻塞到可以获取锁为止。这个方法调用返回的MutexGuard值是Mutex的一层薄包装。

感谢我们之前讨论过的隐式强制转换,我们可以直接在这个Mutex防护器上调用waiting_list的方法,比如guard.players.push(player)。防护器甚至允许我们直接借用底层数据的引用。Rust的生命周期系统确保这些引用不会比防护器本身更长寿。

没有持有锁,就无法访问Mutex的数据。但防护器离开作用域时,锁就会被释放。通常情况下这会发生在代码块的末尾,但你也可以手动释放它。

这里的释放使用的是Rust的生命周期机制。

互斥锁和Mutex:

  • Mutex是一个智能指针。
  • lock方法获取互斥锁并返回一个叫做MutexGuard的智能指针。
  • MutexGuard实现了Deref和DerefMut,所以我们可以把它当作一个指向T的指针。
  • 当MutexGuard超出作用域时,它会自动释放锁。

Rust的Mutex比其他语言中的互斥锁更安全、更容易使用,原因如下:

  1. 没有忘记锁定的风险。要访问受保护的数据,你必须调用lock()。
  2. 没有忘记解锁的风险。MutexGuard在超出作用域时自动解锁。
  3. 没有在解锁后使用受保护数据的风险。一旦MutexGuard超出作用域,你就无法再访问数据。
  4. 没有在锁定时抛出异常的风险。Rust没有异常。
  5. 没有死锁的风险,至少在只使用一个锁时没有。

当然,Mutex也不是没有缺点:

  • 与未加锁的数据相比,它们会增加一些运行时开销。
  • 它们可能会导致争用,一个线程必须等待另一个线程完成对共享数据的操作。
  • 不正确使用时可能导致死锁。

读写锁:
有时我们需要比Mutex提供的更细粒度的控制。例如,如果有多个读取器和一个写入器,我们可能希望允许多个读取器同时访问数据,但写入器需要独占访问。

对于这种情况,标准库提供了std::sync::RwLock。它的工作方式与Mutex类似,但有两种锁定方法:

  • read()返回一个RwLockReadGuard
  • write()返回一个RwLockWriteGuard

任意数量的读取器可以同时持有读锁,但写入器必须等待所有读取器完成才能获取写锁。同样,当写入器持有锁时,读取器必须等待。

条件变量:
条件变量是另一种同步原语,它允许线程等待某个条件变为真。在Rust中,我们可以使用std::sync::Condvar。

原子类型:
对于简单的数值类型,我们可以使用原子操作而不是互斥锁。原子操作保证在多线程环境中是安全的,并且通常比互斥锁更高效。Rust在std::sync::atomic模块中提供了几种原子类型。

全局变量:
Rust支持全局变量,但它们必须是不可变的,除非被标记为unsafe。对于可变全局状态,我们可以使用lazy_static宏和Mutex或RwLock。

总结:
Rust提供了多种工具来处理共享可变状态,包括Mutex、RwLock、Condvar和原子类型。这些工具允许我们安全地在多个线程之间共享和修改数据,同时避免了许多常见的并发错误。通过仔细选择和正确使用这些工具,我们可以编写既安全又高效的并发程序。




深度理解Rust并发（五）


深度理解Rust并发(五)

这一小节主要讲的是互斥锁(Mutex)。乍一看可能会觉得奇怪,join_waiting_list方法并没有通过可变引用接收self,它的签名是引用self。但底层的集合是Vector,在调用push方法时确实需要一个可变引用。这段代码可以编译并正常运行,这是怎么回事呢?

在Rust中,可变引用表示独占访问,而单引用表示共享访问。我们习惯了类型层次传递,可变引用从父级传递到子集,从容器传递到内容。比如,只有在拥有starships的可变引用时,才能调用starships[id].engine的可变引用self方法,或者你拥有所有权。这是Rust的默认行为,因为如果没有对父级元素的独占访问权,Rust通常无法保证你对子集元素也具有独占访问权。

但是Mutex提供了一种实现独占访问的方法,用锁来实现。Mutex的核心作用就是提供这种机制,即使没有线程可以共享Mutex本身的非独占访问,它也可以保证对内部数据的独占访问权。Mutex实际上是一个包装器,类似于父级一样把数据包裹起来。Rust类型系统通过Mutex的定义阐明了它的作用,Mutex动态地强制执行独占访问,这在Rust中通常是静态的(在编译时完成)。你可能还记得std::cell::RefCell也具有类似的功能,但它不支持多线程。Mutex和RefCell都属于一种称为”内部可变性”的模式。

互斥锁并不总是好主意,只是在某些场景比较好用。在学习Mutex之前,我们讲了一些并发编程的方法。如果你熟悉C++,可能会觉得这些方法用起来特别简单。这些方法只在针对并发编程中最让人困惑的方面提供强有力的保证:

  1. 完全使用fork-join并行的程序是确定性的,也不会死锁。
  2. 使用通道的程序表现也很好,就像我们的索引构建器一样。仅将通道用于管道处理程序,它也是有确定性的。消息传递的时机可能有变化,有时候传递消息的时间不一样,导致后面有点变化,但不会影响最终输出的结果。

总的来说,多线程程序的安全性保障很重要。Rust Mutex设计几乎可以肯定会让你比以往任何时候都更加系统、合理地使用互斥锁。但值得思考一下,Rust的安全保障能做到什么,又做不到什么。

经过Rust编译的安全代码不会触发数据竞争。数据竞争是一种特殊的错误,指的是多个线程同时读写相同内存,从而导致毫无意义的结果。这是件好事,因为数据竞争在实际的多线程程序中很常见,并且始终都会产生错误。但是使用互斥锁的线程仍然可能会遇到一些其他问题,而Rust不会帮你解决这些问题。

有效的Rust程序不会发生数据竞争,但仍然会出现其他的竞态条件。竞态条件指程序的行为取决于线程之间的执行时机。执行时间不同,导致程序的行为不同,这就会出现不确定性。因此可能会导致每次运行的结果不同。有些竞态条件是良性的,有些则会表现为程序的不稳定性和难以修复的错误。无序地使用互斥锁容易引入竞态条件,需要确定它们的行为是好的。

共享可变状态也会影响程序设计。通道在你的代码中充当抽象层的边界,方便分离独立的组件进行测试。但互斥锁会鼓励一种逐步添加方法的工作方式,这可能会导致代码变成一个由相互关联的代码块组成的庞大程序。

正如接下来两节内容要展示的,互斥锁的使用并不像看上去那么简单。这些问题都与这些工具本身的特性有关。能够使用更结构化的方法时,就尽量使用结构化的方法,只有在必须使用互斥锁时才采用互斥锁。

死锁:
线程可以通过尝试获取已经被持有的锁来使得自己死锁。假设对self.waiting_list.lock()的第一次调用成功获得了锁,第二次调用看到锁已经被持有,所以就会被阻塞。等待的线程正是持有锁的线程。换句话说,Mutex中的锁不是递归的。

这里的错误很明显,在实际中这两个lock调用可能在两个不同的方法中,其中一个方法调用另一个方法。每个方法的代码单独来看好像都没有问题,但事实上它们之间已经形成死锁。还有其他方法可以导致死锁,涉及多个线程一次获取多个互斥锁的时候。Rust的借用系统没法保护你免于死锁,最好的保护措施是保持临界区很小:进入、完成你的工作,然后退出。如果你在临界区停留太久,容易形成死锁。

通道也可能导致死锁。比如,两个线程可能会阻塞,每个线程都等待从另一个线程接收消息。但是良好的程序设计可以让你高度确信在实践中不会发生这种情况。在管道(比如我们的倒排索引生成器)中数据流是非循环的,这样的程序发生死锁的可能性和Unix shell管道一样小,因为它不是循环的。如果通道之间是循环的,就容易形成死锁。

中毒的互斥锁:
Mutex::lock方法和JoinHandle的join方法一样,都返回Result类型的值。这背后的原因是为了在另一个线程发生panic时能够比较缓和地失败,而不是一下子全部崩溃。当我们写handle.join().unwrap()时,是在告诉Rust来自一个线程的panic信息传递到另一个线程。类似地,mutex.lock().unwrap()也是一样的道理。

如果一个线程在持有Mutex锁的时候发生了panic,Rust会把这个Mutex标记为中毒状态。任何后续尝试锁定中毒的Mutex都会得到一个错误结果。我们使用的unwrap()方法告诉Rust,在发生这种情况时也应该触发panic,把来自另一个线程的panic信息传递到当前线程中。

中毒的Mutex到底有多严重呢?虽然”中毒”听起来很致命,但这种情况并不一定是致命的。就像我们在第7章讲的,panic是一种安全的错误处理机制,不会把很多错误的东西往下传递,因为它终止了线程。所以一个线程的panic并不会危及程序其他部分的安全性。

Mutex在发生panic时会中毒,并不是因为担心出现未定义行为,更确切地说,只是因为程序可能依赖某些不变式来运行。不变式指的是程序运行过程中始终保持为真的某些条件,它一直是真的,有点像断言。由于程序panic并退出而没有完成正在做的事情(比如只更新受保护数据的一部分字段而没有更新其他部分),那么这些不变式现在可能已经被破坏了。Rust将Mutex设为中毒,是为了防止其他线程无意中进入这种破坏的场景并使得情况更加恶化。

不过,你仍然可以通过PoisonError::into_inner()方法来锁定中毒的Mutex并访问它里面的数据。互斥访问的机制仍然会得到完全的执行,但是你不会意外地做到这一点,因为unwrap()会让你显式地意识到潜在的危险。

使用互斥锁的多消费者通道:
我们之前提到过,Rust通道是单生产者多消费者(Multiple Producer, Single Consumer, MPSC)的。确切地说,一个通道只能有一个接收者(receiver)。这就意味着我们没有办法使用线程池让多个线程共享一个接收者,当作任务列表。

不过幸运的是,使用标准库的工具,我们可以实现一个非常简单的变通方案。我们可以用Mutex来包装接收者,就变成了多个接收者被Mutex包起来。看上去好像只有一个,所以就变成了多消费者。可以是Multiple Producer和Multi Consumer,那个Multi Consumer被Mutex包装,就变成了Single Consumer。

我们正在使用Arc<Mutex<Receiver>>这个类型。这个嵌套确实有点多,在Rust中比C++中更常见。这可能会让人觉得比较困惑,但在通常情况下,只需要读懂类型名称就可以帮助理解正在发生的事情:

  • Arc这部分表示对类型T的原子引用计数,它允许我们在多个地方共享对于通道数据的访问,同时确保没有人意外地释放它。
  • Mutex这部分表示互斥锁,它保证一次只能有一个线程修改内部数据。
  • Receiver这部分表示从通道接收类型T数据的接收者。

所以整个类型Arc<Mutex<Receiver>>表示对一个互斥锁的原子引用计数,这个互斥锁保护着用于从通道接收类型T数据的接收者。对于下面来说,透明的一点是它也是多个consumer。

读写锁:
Rust标准库线程同步工具包(std::sync)中除了互斥锁(Mutex)之外还有其他工具。因为涵盖所有工具的细节超出了这本书的范围,所以这里只做概述。

服务器程序通常会加载一些配置信息,在启动时加载一次之后就很少改动。大多数线程只需要查询配置信息就行了,但因为配置可以被修改(比如说服务器可能允许重新从磁盘加载配置),所以无论如何都需要使用某种机制来保护它。在这种情况下,虽然可以使用互斥锁来实现,但它会成为不必要的瓶颈。如果配置没有频繁更改,那么线程就没有必要为查询信息而排队。

读写锁(RwLock)更加适合这种场景。与互斥锁只有一个锁定方法不同,读写锁有两种锁定方法:

  1. RwLock::write()方法类似于Mutex::lock(),它会等待获取对受保护数据的独占修改访问权。
  2. RwLock::read()方法用于获取非修改访问权,它的优点是可以减少等待时间,因为它允许多个线程同时安全地读取数据。

使用互斥锁时,受保护数据在任何时刻只能有一个线程进行读操作或写操作。而使用读写锁时,它可以同时拥有一个写线程和多个读线程。这就类似于Rust的引用机制。

ForumApp可能具有一个用于存储配置信息的结构体,结构体由读写锁进行保护。读取配置的方法使用RwLock::read(),重新加载配置的方法用RwLock::write()。这样就能保证一致性。

Rust特别适合强制执行读写锁数据的安全规则。”单写者或多读者”的概念正好是Rust借用检查系统的核心,与读写锁完全契合。self.config.read()方法返回的是guard,它提供了对app_config非修改(即不可变)的共享访问权限。这个guard有点类似于概念上的锁,它确保在读取配置信息时不会发生数据竞争。self.config.write()方法返回的是不同类型的guard,它提供对app_config的修改(即独占)访问权限。类似地,写操作的guard则保证了写入配置信息时的互斥访问。

这种方法特别好理解,与Rust的借用检查规则特别相似。实际上,它们几乎是一样的:同一时刻只能有一个可变引用,或者多个不可变引用指向同一数据,从而保证了内存安全。



深度理解Rust并发（六）


好的,我会按照您的要求整理内容,保持原有信息不遗漏,同时使文本更加合理通顺。以下是整理后的内容:

深度理解Rust并发（六）

这是并发章节的最后一部分,讲解条件变量。条件变量是线程通常需要等待某个条件变成真才能继续执行的机制。这在实际应用中很常见,例如:

  1. 关闭服务器时,主线程需要等到其他线程都完成退出后才能继续。这类似于笔记本电脑软关机的过程。

  2. 当工作线程没有任务时,需要等待有新的数据可以处理为止。

  3. 实现分布式一致性协议的线程需要等待足够数量的决策节点(投票节点或仲裁节点)做出响应后才能继续。这类似于集群协议中,需要满足一定数量的节点同意后才能进行下一步。

有时会存在针对我们想要等待的确切条件的便捷API,比如服务器关闭的例子中的join。在其他情况下,没有内置的阻塞API,可以使用条件变量来构建自己的阻塞逻辑。

在Rust中,std::sync::Condition类型用于实现条件变量。Condition具有wait和notify_all方法:

  • wait方法会一直阻塞,直到其他线程调用notify_all。
  • notify_all用于通知所有等待的线程。

条件变量的使用比较复杂,因为它始终与某个受特定互斥锁保护的数据的真假条件相关。互斥锁和条件变量是相关的。完整解释会超出本书篇幅,为方便理解:

  • 当所需的条件为真时,调用Condition::notify_all或notify_one来唤醒等待的线程。
  • 为了进入睡眠并等待条件为真,使用Condition::wait。

使用条件变量的标准用法是在while循环中使用。wait的签名有点特殊:它通过值获取一个MutexGuard对象,消耗它,并在成功后返回一个新的MutexGuard。这反映了wait方法的直觉:释放互斥锁,等待一会儿,然后在返回之前重新获取它。通过值传递MutexGuard相当于授予wait方法释放互斥锁的独占权限。

原子类型

std::sync::atomic模块包含用于无锁并发编程的原子类型。这些类型基本上与标准C++原子类型一样,并有一些额外功能:

  • AtomicIsize和AtomicUsize是共享的整数类型,分别对应单线程的isize和usize类型。
  • AtomicI8、AtomicI16、AtomicI32、AtomicI64以及对应的无符号版本是共享的整数类型,分别对应单线程的i8、i16等类型。
  • AtomicBool是共享的布尔值。
  • AtomicPtr是指向类型为T的可变、不安全指针的共享值。

简单来说,多个线程可以同时读写原子值而不会引起数据竞争,因为它们是原子操作。原子是不可分的,必须把操作一次性完成。

与通常的算术逻辑运算符不同,原子类型提供的方法可以执行原子操作,包括单个的读、存储、交换,以及以原子单元方式进行的顺序运算。即使其他线程也正在执行涉及相同内存位置的操作,这些操作仍然是安全的。

例如,对名为x的AtomicIsize进行递增操作:

1
x.fetch_add(1, Ordering::SeqCst);

编译器可能会将原子操作的方法转换为特殊的机器码指令。例如,在x86-64架构上,fetch_add调用可能会变成lock inc指令,而普通的数值累加操作(如n = n + 1)可能会变为简单的inc指令或类似指令的形式。

与普通的读写操作不同,原子操作可能会立即影响或被其他线程影响。所以Rust编译器在处理原子操作时需要放弃一些优化手段,这会导致性能上稍微差一些。

参数Ordering::SeqCst指定了一种内存排序。内存排序有点类似于数据库中的事务隔离级别,它告诉系统在程序的正确性和性能之间做出取舍。程序员可以选择关心诸如可观察性、因果关系和时间不会出现循环等理论概念,但这样可能会降低程序性能。

内存排序对程序正确性至关重要,但理解和推理起来比较复杂。幸运的是,选择最严格的内存排序Ordering::SeqCst(顺序一致性)通常带来的性能损耗很低,不像数据库中使用serializable模式那样会带来明显性能下降。因此,当不知道使用哪种内存排序方式时,建议使用Ordering::SeqCst。

Rust还继承了来自标准C++原子操作的几种其他内存排序选项。这些选项对数据的一致性和因果关系的保证比较弱,这里暂不讨论。

原子操作的一个简洁用法是用于线程取消。假设我们有一个执行长时间计算的线程(如视频渲染),并希望能够异步地取消它。难点在于如何向线程传达我们希望它停止运行的意愿。我们可以通过共享的AtomicBool来实现这一点:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
use std::sync::Arc;
use std::sync::atomic::AtomicBool;

let cancel_flag = Arc::new(AtomicBool::new(false));
let worker_cancel_flag = cancel_flag.clone();

// 在worker线程中:
while !worker_cancel_flag.load(Ordering::SeqCst) {
    // 渲染一个像素
}

// 在主线程中取消worker:
cancel_flag.store(true, Ordering::SeqCst);
// 等待线程退出

这段代码创建了两个指向同一个堆分配的AtomicBool的Arc智能指针,初始值为false。第一个名为cancel_flag的指针保留在主线程中,第二个名为worker_cancel_flag的指针会移动到worker线程里。

在渲染每个像素之后,线程通过调用load方法检查标志的值。如果我们在主线程决定取消worker线程,那么就在AtomicBool中存储true,然后等待线程退出。

实现线程取消还有其他方法,这里使用的AtomicBool可以替换成Mutex或通道。主要的区别在于原子操作开销很低,永远不会用到系统调用。一次加载或存储操作通常会变成单个的CPU指令。

原子操作类似于互斥锁或读写锁,它是一种内部可变性。所以它的函数都采用&self的共享引用(不可变引用),这就使得我们可以用它作为简单的全局变量。

全局变量

假设我们正在编写网络代码,我们希望有一个全局变量作为计数器,每处理一个数据包就递增一下。我们可能会这样写:

1
2
3
4
5
static PACKETS_SERVED: usize = 0;

fn process_packet() {
    PACKETS_SERVED += 1;
}

这段代码编译可以通过,但存在一个问题:PACKETS_SERVED被声明为不可变,所以我们无法对它进行修改。

Rust语言尽一切合理的努力去阻止使用全局可变状态。使用const关键字声明的常量肯定是不可变的,默认情况下静态变量也是不可变的。这就表示你无法获得它的可变引用。

虽然可以把静态变量声明为mut(可变类型),但这样会使得访问这个变量变得不安全。Rust的严格线程安全性正是这些规则背后的主要原因。全局可变状态还有一些不好的软件工程方面的影响:它会使程序不同部分耦合更加紧密,更难测试,而且后期也很难修改。

但在某些情况下,确实没有更好或更合理的替代方案,所以我们必须找到一种安全的方式来声明可变的静态变量。为了支持PACKETS_SERVED进行递增操作,同时还保持线程安全性,最简单的方法是把它声明为原子整数:

1
2
3
4
5
6
7
use std::sync::atomic::AtomicUsize;

static PACKETS_SERVED: AtomicUsize = AtomicUsize::new(0);

fn process_packet() {
    PACKETS_SERVED.fetch_add(1, Ordering::SeqCst);
}

一旦声明了这个静态变量,更新数据包的计数就比较简单了,因为原子操作不会被打断。

原子全局变量仅限于简单的整数和布尔值。创建任何其他类型的全局变量就相当于要解决两个问题:

  1. 变量必须以某种方式变成线程安全的,否则它就不可能成为全局变量。为了安全起见,静态变量必须同时是Sync的(可以安全地在线程间共享)和Send的(可以安全地在线程间移动)。幸运的是,我们已经看到了解决这个问题的方法:Rust具有用于安全共享可变值的类型,比如Mutex、RwLock和原子类型。即使声明为不可变,这些类型也可以被修改,这正是它们的作用。

  2. 静态初始化器只能调用专门标记为const的函数,编译器可以在编译时对它进行评估。换句话说,它们的输出是确定性的,只取决于它们的参数,而不依赖任何其他状态或I/O。这样,编译器可以把计算结果嵌入为编译时常量,类似于C++的constexpr。

原子类型如AtomicUsize、AtomicBool等的构造函数都是const函数,这就允许我们创建静态的AtomicUsize。其他类型如String、Ipv4Addr、Ipv6Addr也具有简单的构造函数,它们也是const的。

你也可以通过在函数签名之前加上const关键字来定义你自己的常量函数。Rust限制了常量函数可以执行的操作,只允许一小部分操作,既能满足使用需求,又能避免非确定性的结果。常量函数不能使用泛型参数作为类型参数,只能使用生命周期参数,并且即使在unsafe块中,也无法分配内存或操作原始指针。但是我们可以使用算术运算(包括溢出和饱和运算)和不会短路的逻辑运算,以及其他的常量函数。

结合这些技术,我们可能会想写成这样:

1
2
3
4
use std::sync::Mutex;
use std::collections::HashMap;

static STATISTICS: Mutex<HashMap<String, usize>> = Mutex::new(HashMap::new());

不幸的是,虽然AtomicUsize::new和String::new是常量函数,但Mutex::new不是。为了克服这些限制,我们需要使用lazy_static crate。我们在”延迟构建正则表达式”一节中介绍过lazy_static。使用lazy_static!宏定义变量可以让你使用任何喜欢的表达式来初始化它,它会在变量首次被引用时运行,并且这个值会保存下来供所有后续使用。我们可以这样写:

1
2
3
4
5
6
7
use lazy_static::lazy_static;
use std::sync::Mutex;
use std::collections::HashMap;

lazy_static! {
    static ref STATISTICS: Mutex<HashMap<String, usize>> = Mutex::new(HashMap::new());
}

这种技术也适用于其他复杂数据结构,比如哈希表和双端队列。对于那些完全不可变但初始化过程比较复杂的静态变量也很实用。

使用lazy_static!宏来定义惰性静态变量会带来轻微的性能开销,这是因为每次访问静态变量时都需要进行检查。内部实现使用了std::sync::Once,这是一种底层的同步原语,专门用于一次性初始化。简单来讲,每次访问惰性静态变量时,程序都会执行一个原子加载指令,目的是确认初始化是否已经完成。

由于Once具有特殊用途,因此这里不会详细讲解。通常使用lazy_static!宏会更方便,但对于初始化非Rust库来说,Once是有用的。在”安全地使用libgit2”一节中就提到了这个例子。

使用Rust编写并发代码的体验

本章介绍了在Rust中使用线程的三种技术:join并行、通道,以及带锁的共享可变状态。我们的目标是让你熟悉Rust提供的多线程工具,并且重点介绍如何将它们集成到实际程序中去。

Rust语言始终把安全性放在首位,所以从你决定编写多线程程序的那一刻起,就需要注重构建安全、结构化的通信机制。让线程保持相对独立是向Rust证明你的代码安全的一种有效方法。幸运的是,线程间的隔离也有助于确保程序的正确性和可维护性。换句话说,Rust通过语言特性引导你编写良好的程序。

更重要的是,Rust允许你组合多种技术并进行试验。你可以快速迭代代码并与编译器交互,这可以帮助你更快地编写正确的代码,比调试数据竞争等问题要高效得多。

这是并发章节的所有内容了。这个章节讲解得很有逻辑性,而且有详细的分类,很清晰。当然,并发还有很多内容可以讲解,再多都不嫌多。你可以看到这个章节总共有六个部分,已经不算少了,但还有更多内容。下一个部分将讲解异步编程,那里会有更多值得学习的内容。

总结:

  1. 本章介绍了Rust中的条件变量,它允许线程等待某个条件变为真。

  2. 讲解了原子类型,它们用于无锁并发编程,可以安全地在多线程间共享和修改。

  3. 探讨了全局变量的使用,包括静态变量和如何安全地创建可变全局状态。

  4. 介绍了lazy_static宏,用于创建复杂的静态变量。

  5. 总结了Rust并发编程的特点,强调了安全性和组合多种技术的能力。

  6. 指出这只是并发编程的一部分,还有更多内容值得学习,如异步编程。

这个章节为读者提供了Rust并发编程的基础知识,同时也为进一步学习更高级的并发技术铺平了道路。





一次性理解：Java、Go、Rust都是如何应对并发挑战的？