17 - 并发

• 系统开发编写并发代码的思路：
  • 一个后台线程（background thread）只负责一件事，周期性唤醒以执行任务。
  • 通用线程池（worker pool）通过队列与客户端通信。
  • 管道（pipeline）将数据从一个线程导入另一个线程，每个线程只做一小部分工作。
  • 数据并行（data parallelism）假设整个计算机主要用于一项大型计算作为主任务，主任务又拆分成 n 个小任务，在 n 个线程上执行，希望所有 n 个机器的核心同时工作。
  • 同步对象海（sea of synchronized object）中多个线程拥有同一数据权限，使用互斥量等低级原语的临时锁方案避免争用。
  • 原子整数操作（atomic integer operation）允许多核心通过以一个机器字大小的字段传递信息，进而实现通信。
• 使用 Rust 线程的 3 种方式：
  • 并行分叉 — 合并；
  • 通道；
  • 共享可修改状态。

17.1 - 并行交叉 - 合并

• 线程可以用于同时执行几个完全不相关的任务。

• 编写一个单线程的程序：

  fn process_files(filenames: Vec<String>) -> io::Result<()> {
      for document in filenames {
          let text = load(&document)?;  // 读取源文件
          let results = process(text);  // 计算统计值
          save(&document, results)?;    // 写入输出文件
      }
      Ok(())
  }
  1
  2
  3
  4
  5
  6
  7
  8

• 使用 “并行分叉 — 合并” 模式，可以实现多线程任务执行。

  • 实现思路是把数据资源分多个块，然后分别在独立的线程上处理每一块数据。
  • 分叉（fork）：启动一个新线程。
  • 合并（join）：等待线程完成。
  • 要求工作单元隔离。

17.1.1 - 产生及合并

• 使用 std::thread::spawn 函数可以产生一个新线程：

  spawn(|| {
      println!("hello from a child thread");
  })
  1
  2
  3

  • 接收一个参数，即 FnOnce 闭包或函数。
  • 所产生的新线程是一个真实的操作系统线程，有自己的栈。

• 使用 spawn 实现前面 process_files 函数的并行版本：

  use std::thead::spawn;
  
  fn process_files_in_parallel(filenames: Vec<String>) -> io::Result<()> {
      // 将工作分成几块
      const NTHREADS: usize = 8;
      let worklists = split_vec_into_chunks(filenames, NTHREADS);
  
      // 分叉：每个块产生一个线程来处理
      let mut thread_handles = vec![];
      for worklist in worklists {
          thread_handles.push(spawn(move || process_files(worklist)));
      }
  
      // 合并：等待所有线程完成
      for handle in thread_handles {
          handle.join.unwrap()?;
      }
      Ok(())
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19

• 把文件名列表转换为工作线程的过程：

  • 在父线程中，通过 for 循环定义并转移 worklist；
  • 创建 move 闭包时，wordlist 被转移到闭包中；
  • 然后 spawn 将闭包（以及 wordlist 向量）转移到新的子线程中。

17.1.2 - 跨线程错误处理

• handle.join()
  1

  方法：

  • 返回一个 std::thread::Result，如果子线程诧异则是一个错误。
  • 该方法把子线程返回的值传给父线程。

• 诧异不会自动从一个线程传播到依赖它的其他线程。

• 一个线程的诧异在其他线程中会体现为包含错误的 Result。

• handle.join().unwrap()：.unwrap() 执行断言操作，实现诧异的传播。如果子线程诧异后，那么会返回 Ok 结果。那么其调用的父线程也会诧异。相当于显示地将诧异从子线程传播到父线程。

17.1.3 - 跨线程共享不可修改数据

• 在向函数中传引用的时候，如果一个线程触发了 IO 错误，就可能导致调用函数在其他线程完成前退出。那么其他子线程就有可能在主线程被释放之后还继续使用传入的参数，这样就会导致数据发生争用。

• 在 Rust 中是不允许这种情况发生的。只要有任何线程拥有 Arc<GigabyteMap>，映射就不会释放，即使父线程已经退出了。因为 Arc 中的数据是不可修改的，不会出现任何数据争用。

  use std::sync::Arc;
  
  fn process_files_in_parallel(filenames: Vec<String>, glossary: Arc<GigabyteMap>)
      -> io::Result<()>
  {
      ...
      for worklist in worklists {
          // 调用.clone()，克隆Arc并触发引用计数。不会克隆GigabyteMap
          let glossary_for_child = glossary.clone();
          thread_handles.push(
              spawn(move || process_files(worklist, &glossary_for_child))
          );
      }
      ...
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15

17.1.4-Rayon

• Rayon 库：专门为 “并行分叉 — 合并” 模式设计，提供两种运行并发任务的方式：

  extern crate rayon;
  use rayon::prelude::*;
  
  // 并行实现两个任务
  let (v1, v2) = rayon::join(fn1, fn2);
  
  // 并行实现N个任务
  giant_vector.par_iter().for_each(|value| {  // 创建一个ParallelIterator，类似迭代器。
      do_thing_with_value(value);
  });
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

• 使用 Rayon 重写 process_files_in_parallel：

  extern crate rayon;
  
  use rayon::prelude::*;
  
  fn process_files_in_parallel(filenames: Vec<String>, glossary: &GigabyteMap)
      -> io::Result<()>
  {
      filenames.par_iter()    // 创建并行迭代器
          // 调用文件名，得到一个ParallelIterator
          .map(|filename| process_file(filename, glossary))
  
          // 组合结果，返回一个Option，只有filename为空是才是None
          // 任何后台发生的并行处理，都可保证reduce_with返回时才完成。
          .reduce_with(|r1, r2| {    
              if r1.is_err() {
                  r1
              } else {
                  r2
              }
          })
          .unwrap_or(Ok(()))          // 让结果Ok(())
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22

• Rayon 也支持在线程间共享引用。

• 要使用 Rayon 需要添加以下代码：

  • main.rs 中：

    extern crate rayon;
    use rayon::prelude::*;
    1
    2

  • Cargo.toml 中：

    [dependencies]
    rayon = "0.4"
    1
    2

17.2 - 通道

• 通道（channel）：把值从一个线程发送到另一个线程的单向管道，本质是一个线程安全的队列。

  • Unix 管道发送的是字节数据；
  • Rust 通道发送的是值。是 std::sync::mps 模块的一部分。

• sender.send(item)
  1

  把一个值放进通道，

  receiver.recv()
  1

  则移除一个值。

  • 值的所有权从发送线程转移到接收线程。
  • 如果通道是空的，那么 receiver.recv() 会一直阻塞直到有值发送。

• Rust 通道比管道快：

  • 管道是一种提供灵活性、复杂性、但不并发的特性。
  • 使用通道，线程可以通过传值实现通信，无须使用锁或者共享内存。
  • 发送值是转移而不是复制，而且转移的值不局限于数据的大小。

17.2.1 - 发送值

• 倒排索引（inverted index）：一种数据库，可以查询哪个关键词在哪里出现过。是实现搜索引擎的关键之一。

• 启动读取文件线程的代码：

  use std::fs::File;
  use std::io::prelude::*;  // 使用Read::read_to_string
  use std::thread::spawn;
  use std::sync::mpsc::channel;
  
  let (sender, receiver) = channel(); // 队列数据结构，返回一对值：发送者和接收者。
  
  let handle = spawn(  // 启动线程std::thread::spawn。
      // 发送者的所有权会通过move闭包返回转移给新线程。
      move || {   // Rust执行类型推断，判断通道的类型
          for filename in documents {
              let mut f = File::open(filename)?; // 从磁盘读取文件
              let mut text = String::new();
              f.read_to_string(&mut text)?;
  
              if sender.send(text).is_err() {  // 读取文件后，把其文本内容text发送给通道
                  break;
              }
          }
          Ok(())   // 线程读取完所有文档后，程序返回Ok(())
      }
  );
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22

17.2.2 - 接收值

• 创建第二个线程循环调用 receiver.recv()。

  // while循环实现
  while let Ok(text) = receiver.recv() {
      do_something_with(text);
  }
  
  // for循环实现
  for text in receiver {
      do_something_with(text);
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9

• 接收者线程示例：

  fn start_file_indexing_thread(texts: Receiver<>)
      -> (Receiver<InMemoryIndex>, JoinHandle<()>)
  {
      let (sender, receiver) = channel();
      let handle = spawn(
          move || {
              for (doc_id, text) in texts.into_iter().enumerate() {
                  let index = InMemoryIndex::from_single_document(doc_id, text);
                  if sender.send(index).is_err() {
                      break;
                  }
              }
          }
      );
      (receiver, handle)
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

• 接收时，如果线程发生了 IO 错误，会立即退出，而错误会存储在线程的 JoinHandle 中。包装组合接收者、返回者和新线程的 JoinHandle 代码如下：

  fn start_file_reader_thread(documents: Vec<PathBuf>)
      -> (Receiver<Strig>, JoinHandle<io::Result<()>>)
  {
      let (sender, receiver) = channel();
  
      let handle = spawn(
          move || { ... }
      );
  
      (receiver, handle)
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11

17.2.3 - 运行管道

• 在内存中合并索引，直至足够大：

  fn start_in_memory_merge_thread(file_indexes: Receiver<InMemoryIndex>)
      -> (Receiver<InMemoryIndex>, JoinHandle<()>)
  1
  2

• 把大索引写入磁盘：

  fn start_index_write_thread(big_indexes: Receiver<InMemoryIndex>, output_dir: &Path)
      -> (Receiver<PathBuf>, JoinHandle<io::Result<()>>)
  1
  2

• 如果有多个大文件，则使用基于文件的合并算法，将它们合并起来：

  fn merge_index_files(files: Receiver<PathBuf>, output_dir: &Path)
      -> io::Result<()>
  1
  2

• 启动线程，并检查错误：

  fn run_pipeline(documents: Vec<PathBuf>, output_dir: PathBuf)
      -> io::Result<()>
  {
      // 启动管道的全部5个阶段
      let (texts,   h1) = start_file_reader_thread(documents);
      let (pints,   h2) = start_file_indexing_thread(texts);
      let (gallons, h3) = start_in_memory_merge_thread(pints);
      let (files,   h4) = start_index_writer_thread(gallons, &output_dir);
      let result = merge_index_files(files, &output_dir);
  
      // 等待线程完成，保存任何错误
      let r1 = h1.join().unwrap();
      h2.join().unwrap();
      h3.join().unwrap();
      let r4 = h4.join().unwrap();
  
      // 返回遇到的第一个错误
      // h2和h3不会失败，因为它们是纯内存数据处理
      r1?;
      r4?;
      result
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22

• 管道实现流水线作业，整体性能受限于最慢阶段的生成能力。

17.2.4 - 通道特型与性能

• std::sync::mps 特型：mpsc（multi-producer, single-consumer）是多生产者，单消费者。

• Sender<T> 实现了 Clone 特型：实现创建一个常规通道，然后再克隆发送者。可以把每个 Sender 值转移到不同线程。

• Receiver<T> 无法克隆，如果需要多个线程从同一个通道接收值，就需要使用 Mutex。

• 反压力（backpressure）：如果发送值的速度超过接收和处理值的速度，就会导致通道内部的值越积越多。

• 同步通道（synchronous channel）：Unix 的每个管道都有固定大小，如果一个进程尝试向随时可能满的管道写入数据，系统就会直接阻塞该进程，直至管道中有了空间。

  use std::sync::mpsc::sync_channel;
  let (sender, receiver) = sync_channel(1000);
  1
  2

  • 同步通道与常规通道一样，只是在创建时需要指定它可以保存多少值。
  • sender.send(value) 是一个潜在的阻塞操作。

17.2.5 - 线程安全

• std::marker::Send 特型：实现 Send 的类型，可以安全地把值传到另一个线程，即实现线程间转移值。
• std::marker::Sync 特型：实现 Sync 的类型，可以安全地把不可修改引用传到另一个线程，即他们可以在线程间共享值。
• Rust 通过上述特型实现线程安全：没有数据争用和其他未定义行为。
• 结构体或枚举的字段如果支持上述特型，那么它们本身也是支持的。
• Rust 会为自定义类型自动实现上述特型，不需要通过#[derive] 派生。
• 少数没有实现 Send 和 Sync 的类型主要用于在非线程安全的条件下提供可修改能力。例如引用计数指针类型 std::rc::Rc<T>。Rust 要求在通过 spawn 创建线程时，传入的闭包必须是 Send。

17.2.6 - 将所有迭代器都接到通道上

下述实现适用于所有迭代器。

use std::thread::spawn;

impl<T> OffThreadExt for T where T: Iterator + Send + 'static, T::Item: Send + 'static {
    fn off_thread(self) -> mpsc::IntoIter<Self::Item> {
        // 创建一个通道并将工作线程中的项传出来
        let (sender, receiver) = mpsc::sync_channel(1024);
        
        // 把这个迭代器转移到一个新线程中并在那里运行它
        spawn(
            move || {
                for item in self {
                    if sender.send(item).is_err() {
                        break;
                    }
                }
            }
        );
        
        // 返回一个从通道提取值的迭代器
        receiver.into_iter()
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

---

详见《Rust 程序设计》（吉姆 - 布兰迪、贾森 - 奥伦多夫著，李松峰译）第十九章
原文地址