devbox
代码艺术创作者
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

现代C++学习笔记——第7章 并行与并发_results.emplace_back( pool.enqueue([i] { std::cout

作者：代码艺术创作者 | 2024-01-31 18:45:33

赞

踩

results.emplace_back( pool.enqueue([i] { std::cout << "hello " << i << std::

7.1 并行基础
std::thread 用于创建一个执行的线程实例，所以它是一切并发编程的基础，使用时需要包含
<thread> 头文件，它提供了很多基本的线程操作，例如 get_id() 来获取所创建线程的线程 ID，使用join() 来加入一个线程等等，例如：

  1. #include <iostream>
  2. #include <thread>
  3.  
  4. int main()
  5. {
  6.     std::thread t([](){std::cout << "hello world." << std::endl;});
  7.     t.join();
  8.  
  9.     return 0;
  10. }

运行结果：

 7.2 互斥量与临界区
我们在操作系统、亦或是数据库的相关知识中已经了解过了有关并发技术的基本知识， mutex 就是其中的核心之一。 C++11 引入了 mutex 相关的类，其所有相关的函数都放在 <mutex> 头文件中。
std::mutex 是 C++11 中最基本的 mutex 类，通过实例化 std::mutex 可以创建互斥量，而通过
其成员函数 lock() 可以进行上锁， unlock() 可以进行解锁。但是在在实际编写代码的过程中，最好不去直接调用成员函数，因为调用成员函数就需要在每个临界区的出口处调用 unlock()，当然，还包括异常。这时候 C++11 还为互斥量提供了一个 RAII 语法的模板类 std::lock_guard。 RAII 在不失代码简洁性的同时，很好的保证了代码的异常安全性。
在 RAII 用法下，对于临界区的互斥量的创建只需要在作用域的开始部分，例如：
 

  1. #include <iostream>
  2. #include <thread>
  3. #include <mutex>
  4.  
  5. int v = 1;
  6.  
  7. void critical_section(int change_v)
  8. {
  9.     static std::mutex mtx;
  10.     std::lock_guard<std::mutex> lock(mtx);
  11.     //执行竞争操作
  12.     std::cout << "do contention operations=====change_v==" << change_v << std::endl;
  13.     v = change_v;
  14.     //离开此作用域后mtx会被释放
  15. }
  16.  
  17. int main()
  18. {
  19.     std::thread t1(critical_section, 2), t2(critical_section, 3);
  20.     std::thread t3(critical_section, 4), t4(critical_section, 5);
  21.     t1.join();
  22.     t2.join();
  23.     t3.join();
  24.     t4.join();
  25.  
  26.     std::cout << "v==" << v << std::endl;
  27.     return 0;
  28. }

运行结果：

 由于 C++ 保证了所有栈对象在声明周期结束时会被销毁，所以这样的代码也是异常安全的。无论critical_section() 正常返回、还是在中途抛出异常，都会引发堆栈回退，也就自动调用了unlock()。
而 std::unique_lock 则相对于 std::lock_guard 出现的， std::unique_lock 更加灵活，std::unique_lock 的对象会以独占所有权（没有其他的 unique_lock 对象同时拥有某个 mutex对象的所有权）的方式管理 mutex 对象上的上锁和解锁的操作。所以在并发编程中，推荐使用std::unique_lock。
std::lock_guard 不能显式的调用 lock 和 unlock，而 std::unique_lock 可以在声明后的任意位置调用，可以缩小锁的作用范围，提供更高的并发度。如果你用到了条件变量 std::condition_variable::wait 则必须使用 std::unique_lock 作为参数。
例如：
 

  1. #include <iostream>
  2. #include <thread>
  3. #include <mutex>
  4.  
  5. int v = 1;
  6. void critical_section(int change_v)
  7. {
  8.     static std::mutex mtx;
  9.     std::unique_lock<std::mutex> lock(mtx);
  10.     // 执行竞争操作
  11.     v = change_v;
  12.     std::cout << "critical_section v = " << v << std::endl;
  13.     // 将锁进行释放
  14.     lock.unlock();
  15.     // 在此期间，任何人都可以抢夺 v 的持有权
  16.  
  17.     // 开始另一组竞争操作，再次加锁
  18.     lock.lock();
  19.     v+=1;
  20.     std::cout << "critical_section v = " << v << std::endl;
  21. }
  22.  
  23. int main()
  24. {
  25.     std::thread t1(critical_section, 2), t2(critical_section, 3);
  26.     t1.join();
  27.     t2.join();
  28.  
  29.     std::cout << "v = " << v << std::endl;
  30.  
  31.     return 0;
  32. }

运行结果：

 7.3 期物
期物（Future）表现为 std::future，它提供了一个访问异步操作结果的途径，这句话很不好理解。
为了理解这个特性，我们需要先理解一下在 C++11 之前的多线程行为。试想，如果我们的主线程 A 希望新开辟一个线程 B 去执行某个我们预期的任务，并返回我一个结果。而这时候，线程 A 可能正在忙其他的事情，无暇顾及 B 的结果，所以我们会很自然的希望能够在某个特定的时间获得线程 B 的结果。
在 C++11 的 std::future 被引入之前，通常的做法是：创建一个线程 A，在线程 A 里启动任务
B，当准备完毕后发送一个事件，并将结果保存在全局变量中。而主函数线程 A 里正在做其他的事情，当需要结果的时候，调用一个线程等待函数来获得执行的结果。
而 C++11 提供的 std::future 简化了这个流程，可以用来获取异步任务的结果。自然地，我们很
容易能够想象到把它作为一种简单的线程同步手段，即屏障（barrier）。
为了看一个例子，我们这里额外使用 std::packaged_task，它可以用来封装任何可以调用的目标，
从而用于实现异步的调用。举例来说：
 

  1. #include <iostream>
  2. #include <future>
  3. #include <thread>
  4.  
  5. int main()
  6. {
  7.     // 将一个返回值为 7 的 lambda 表达式封装到 task 中
  8.     // std::packaged_task 的模板参数为要封装函数的类型
  9.     std::packaged_task<int()> task([](){return 7;});
  10.     // 获得 task 的期物
  11.     std::future<int> result = task.get_future();// 在一个线程中执行 task
  12.     std::thread(std::move(task)).detach();
  13.  
  14.     std::cout << "waiting...";
  15.     result.wait();//在此设置屏障，阻塞到期物的完成
  16.     // 输出执行结果
  17.     std::cout << "done!" << std::endl << "future result is " << result.get() << std::endl;
  18.     return 0;
  19. }

 在封装好要调用的目标后，可以使用 get_future() 来获得一个 std::future 对象，以便之后实施
线程同步。
7.4 条件变量
条件变量 std::condition_variable 是为了解决死锁而生，当互斥操作不够用而引入的。比如，线程
可能需要等待某个条件为真才能继续执行，而一个忙等待循环中可能会导致所有其他线程都无法进入临界区使得条件为真时，就会发生死锁。所以， condition_variable 实例被创建出现主要就是用于唤醒等待线程从而避免死锁。 std::condition_variable 的 notify_one() 用于唤醒一个线程； notify_all()则是通知所有线程。下面是一个生产者和消费者模型的例子：
 

  1. #include <queue>
  2. #include <chrono>
  3. #include <mutex>
  4. #include <thread>
  5. #include <iostream>
  6. #include <condition_variable>
  7.  
  8. int main()
  9. {
  10.     std::queue<int> produced_nums;
  11.     std::mutex mtx;
  12.     std::condition_variable cv;
  13.     bool notified = false; // 通知信号
  14.     // 生产者
  15.     auto producer = [&](){
  16.         for(int i = 0; ; ++i)
  17.         {
  18.             std::this_thread::sleep_for(std::chrono::milliseconds(900));
  19.             std::unique_lock<std::mutex> lock(mtx);
  20.             std::cout << "producing " << i << std::endl;
  21.             produced_nums.push(i);
  22.             notified = true;
  23.             cv.notify_all();//此处也可以使用notify_one
  24.         }
  25.     };
  26.  
  27.     //消费者
  28.     auto consumer = [&](){
  29.         while(true){
  30.             std::unique_lock<std::mutex> lock(mtx);
  31.             while(!notified){// 避免虚假唤醒
  32.                 cv.wait(lock);
  33.             }
  34.             // 短暂取消锁，使得生产者有机会在消费者消费空前继续生产
  35.             lock.unlock();
  36.             std::this_thread::sleep_for(std::chrono::milliseconds(1000));// 消费者慢于生产者
  37.             lock.lock();
  38.             while(!produced_nums.empty())
  39.             {
  40.                 std::cout << "consuming " << produced_nums.front() << std::endl;
  41.                 produced_nums.pop();
  42.             }
  43.             notified = false;
  44.         }
  45.     };
  46.     // 分别在不同的线程中运行
  47.     std::thread p(producer);
  48.     std::thread cs[2];
  49.     for(int i = 0; i < 2; ++i)
  50.         cs[i] = std::thread(consumer);
  51.  
  52.     p.join();
  53.     for(int i = 0; i < 2; ++i)
  54.         cs[i].join();
  55.  
  56.     return 0;
  57. }

运行结果：

 值得一提的是，在生产者中我们虽然可以使用 notify_one()，但实际上并不建议在此处使用，因为
在多消费者的情况下，我们的消费者实现中简单放弃了锁的持有，这使得可能让其他消费者争夺此锁，从而更好的利用多个消费者之间的并发。话虽如此，但实际上因为 std::mutex 的排他性，我们根本无法期待多个消费者能真正意义上的并行消费队列的中生产的内容，我们仍需要粒度更细的手段。
7.5 原子操作与内存模型
细心的读者可能会对前一小节中生产者消费者模型的例子可能存在编译器优化导致程序出错的情况
产生疑惑。例如，布尔值 notified 没有被 volatile 修饰，编译器可能对此变量存在优化，例如将其作为一个寄存器的值，从而导致消费者线程永远无法观察到此值的变化。这是一个好问题，为了解释清楚这个问题，我们需要进一步讨论从 C++ 11 起引入的内存模型这一概念。我们首先来看一个问题，下面这段代码输出结果是多少？
 

  1. #include <thread>
  2. #include <iostream>
  3.  
  4. int main()
  5. {
  6.     int a = 0;
  7.     volatile int flag = 0;
  8.     std::thread t1([&](){
  9.         while(flag != 1)
  10.             ;
  11.         int b = a;
  12.         std::cout << "b = " << b << " flag = " << flag << std::endl;
  13.     });
  14.  
  15.     std::thread t2([&](){
  16.        a = 5;
  17.        std::cout << "a = " << a << std::endl;
  18.        flag = 1;
  19.        std::cout << "flag = " << flag << std::endl;
  20.     });
  21.  
  22.     t1.join();
  23.     t2.join();
  24.  
  25.     return 0;
  26. }

运行结果：

从直观上看， t2 中 a = 5; 这一条语句似乎总在 flag = 1; 之前得到执行，而 t1 中 while (flag
!= 1) 似乎保证了 std::cout << "b = " << b << std::endl; 不会再标记被改变前执行。从逻辑上
看，似乎 b 的值应该等于 5。但实际情况远比此复杂得多，或者说这段代码本身属于未定义的行为，因为对于 a 和 flag 而言，他们在两个并行的线程中被读写，出现了竞争。除此之外，即便我们忽略竞争读写，仍然可能受 CPU 的乱序执行，编译器对指令的重排的影响，导致 a = 5 发生在 flag = 1 之后。从而 b 可能输出 0。
原子操作
std::mutex 可以解决上面出现的并发读写的问题，但互斥锁是操作系统级的功能，这是因为一个互
斥锁的实现通常包含两条基本原理：
1. 提供线程间自动的状态转换，即『锁住』这个状态
2. 保障在互斥锁操作期间，所操作变量的内存与临界区外进行隔离
这是一组非常强的同步条件，换句话说当最终编译为 CPU 指令时会表现为非常多的指令（我们之
后再来看如何实现一个简单的互斥锁）。这对于一个仅需原子级操作（没有中间态）的变量，似乎太苛刻了。关于同步条件的研究有着非常久远的历史，我们在这里不进行赘述。读者应该明白，在现代 CPU 体系结构下提供了 CPU 指令级的原子操作，因此在 C++11 中多线程下共享变量的读写这一问题上，还引入了 std::atomic 模板，使得我们实例化一个原子类型，将一个原子类型读写操作从一组指令，最小化到单个 CPU 指令。例如：
std::atomic<int> counter;
并为整数或浮点数的原子类型提供了基本的数值成员函数，举例来说，包括 fetch_add, fetch_sub
等，同时通过重载方便的提供了对应的 +， - 版本。比如下面的例子：
 

  1. #include <atomic>
  2. #include <thread>
  3. #include <iostream>
  4.  
  5. std::atomic<int> count = {0};
  6.  
  7. int main()
  8. {
  9.     std::thread t1([](){
  10.         count.fetch_add(1);
  11.     });
  12.  
  13.     std::thread t2([](){
  14.         count++;        // 等价于 fetch_add
  15.         count += 1;     // 等价于 fetch_add
  16.     });
  17.  
  18.     t1.join();
  19.     t2.join();
  20.     std::cout << "count = " << count << std::endl;
  21.  
  22.     return 0;
  23. }

 运行结果：

 当然，并非所有的类型都能提供原子操作，这是因为原子操作的可行性取决于 CPU 的架构以及所实例化的类型结构是否满足该架构对内存对齐条件的要求，因而我们总是可以通过 std::atomic<T>::is_lock_free来检查该原子类型是否需支持原子操作，例如：
 

  1. #include <atomic>
  2. #include <iostream>
  3.  
  4. struct A{
  5.     float x;
  6.     int y;
  7.     long long z;
  8. };
  9.  
  10. int main()
  11. {
  12.     std::atomic<A> a;
  13.     std::cout << std::boolalpha << a.is_lock_free() << std::endl;
  14.  
  15.     return 0;
  16. }

输出结果： false

一致性模型
并行执行的多个线程，从某种宏观层面上讨论，可以粗略的视为一种分布式系统。在分布式系统中，任何通信乃至本地操作都需要消耗一定时间，甚至出现不可靠的通信。
如果我们强行将一个变量 v 在多个线程之间的操作设为原子操作，即任何一个线程在操作完 v 后，
其他线程均能同步感知到 v 的变化，则对于变量 v 而言，表现为顺序执行的程序，它并没有由于引入多线程而得到任何效率上的收益。对此有什么办法能够适当的加速呢？答案便是削弱原子操作的在进程间的同步条件。从原理上看，每个线程可以对应为一个集群节点，而线程间的通信也几乎等价于集群节点间的通信。削弱进程间的同步条件，通常我们会考虑四种不同的一致性模型：
1. 线性一致性：又称强一致性或原子一致性。它要求任何一次读操作都能读到某个数据的最近一次写的数据，并且所有线程的操作顺序与全局时钟下的顺序是一致的。

 在这种情况下线程 T1, T2 对 x 的两次写操作是原子的，且 x.store(1) 是严格的发生在 x.store(2)之前， x.store(2) 严格的发生在 x.load() 之前。值得一提的是，线性一致性对全局时钟的要求是难以实现的，这也是人们不断研究比这个一致性更弱条件下其他一致性的算法的原因。
2. 顺序一致性：同样要求任何一次读操作都能读到数据最近一次写入的数据，但未要求与全局时钟的顺序一致。
         x.store(1) x.store(3) x.load()
T1 ---------+-----------+----------+----->
T2 ---------------+------------------------>
                 x.store(2)
或者
          x.store(1) x.store(3) x.load()
T1 ---------+-----------+----------+----->
T2 ------+--------------------------------->
        x.store(2)
在顺序一致性的要求下， x.load() 必须读到最近一次写入的数据，因此 x.store(2) 与 x.store(1)并无任何先后保障，即只要 T2 的 x.store(2) 发生在 x.store(3) 之前即可。
3. 因果一致性：它的要求进一步降低，只需要有因果关系的操作顺序得到保障，而非因果关系的操作顺序则不做要求。
           a = 1             b = 2
T1 --------+---------------+-------------------------------->
T2 ----------+------------------------+------------+-------->
          x.store(3)                c = a + b      y.load()
或者
            a = 1            b = 2
T1 --------+---------------+-------------------------------->
T2 ----------+------------------------+------------+-------->
          x.store(3)                  y.load()    c = a + b
亦或者
            b = 2           a = 1
T1 --------+---------------+-------------------------------->
T2 ----------+------------------------+------------+-------->
            y.load()                   c = a + b     x.store(3)
上面给出的三种例子都是属于因果一致的，因为整个过程中，只有 c 对 a 和 b 产生依赖，而 x 和y 在此例子中表现为没有关系（但实际情况中我们需要更详细的信息才能确定 x 与 y 确实无关）
4. 最终一致性：是最弱的一致性要求，它只保障某个操作在未来的某个时间节点上会被观察到，但并未要求被观察到的时间。因此我们甚至可以对此条件稍作加强，例如规定某个操作被观察到的时间总是有界的。当然这已经不在我们的讨论范围之内了。
         x.store(3)     x.store(4)
T1 --------+---------------+----------------------------------------------------->
T2 -------------+----------------+------------------------+------------+-------->
                x.read()          x.read()                   x.read()     x.read()
在上面的情况中，如果我们假设 x 的初始值为 0，则 T2 中四次 x.read() 结果可能但不限于以下
情况：
3 4 4 4 // x 的写操作被很快观察到
0 3 3 4 // x 的写操作被观察到的时间存在一定延迟
0 0 0 4 // 最后一次读操作读到了 x 的最终值，但此前的变化并未观察到
0 0 0 0 // 在当前时间段内 x 的写操作均未被观察到，但未来某个时间点上一定能观察到 x 为 4 的情况
内存顺序
为了追求极致的性能，实现各种强度要求的一致性， C++11 为原子操作定义了六种不同的内存顺序std::memory_order 的选项，表达了四种多线程间的同步模型：
1. 宽松模型：在此模型下，单个线程内的原子操作都是顺序执行的，不允许指令重排，但不同线程间原子操作的顺序是任意的。类型通过 std::memory_order_relaxed 指定。我们来看一个例子：
 

  1. std::atomic<int> counter = {0};
  2. std::vector<std::thread> vt;
  3. for (int i = 0; i < 100; ++i) {
  4.     vt.emplace_back([&](){
  5.     counter.fetch_add(1, std::memory_order_relaxed);
  6. });
  7. }
  8. for (auto& t : vt) {
  9.     t.join();
  10. }
  11. std::cout << "current counter:" << counter << std::endl;

2. 释放/消费模型：在此模型中，我们开始限制进程间的操作顺序，如果某个线程需要修改某个值，但另一个线程会对该值的某次操作产生依赖，即后者依赖前者。具体而言，线程 A 完成了三次对 x 的写操作，线程 B 仅依赖其中第三次 x 的写操作，与 x 的前两次写行为无关，则当 A 主动 x.release()时候（即使用 std::memory_order_release），选项 std::memory_order_consume 能够确保 B 在调用 x.load() 时候观察到 A 中第三次对 x 的写操作。我们来看一个例子：
 

  1. // 初始化为 nullptr 防止 consumer 线程从野指针进行读取
  2. std::atomic<int*> ptr(nullptr);
  3. int v;
  4. std::thread producer([&]() {
  5.     int* p = new int(42);
  6.     v = 1024;
  7.     ptr.store(p, std::memory_order_release);
  8. });
  9. std::thread consumer([&]() {
  10.     int* p;
  11.     while(!(p = ptr.load(std::memory_order_consume)));
  12.     std::cout << "p: " << *p << std::endl;
  13.     std::cout << "v: " << v << std::endl;
  14. });
  15. producer.join();
  16. consumer.join();

3. 释放/获取模型：在此模型下，我们可以进一步加紧对不同线程间原子操作的顺序的限制，在释放std::memory_order_release 和获取 std::memory_order_acquire 之间规定时序，即发生在释放
操作之前的所有写操作，对其他线程的任何获取操作都是可见的，亦即发生顺序（happens-before）。可以看到， std::memory_order_release 确保了它之后的写行为不会发生在释放操作之前，是一个向前的屏障，而 std::memory_order_acquire 确保了它之前的写行为，不会发生在该获取操作之后，是一个向后的屏障。对于选项 std::memory_order_acq_rel 而言，则结合了这两者的特点，唯一确定了一个内存屏障，使得当前线程对内存的读写不会被重排到此操作的前后。我们来看一个例子：
 

  1. std::vector<int> v;
  2. std::atomic<int> flag = {0};
  3. std::thread release([&]() {
  4.     v.push_back(42);
  5.     flag.store(1, std::memory_order_release);
  6. });
  7. std::thread acqrel([&]() {
  8.     int expected = 1; // must before compare_exchange_strong
  9.     while(!flag.compare_exchange_strong(expected, 2, std::memory_order_acq_rel)) {
  10.         expected = 1; // must after compare_exchange_strong
  11. }
  12. // flag has changed to 2
  13. });
  14. std::thread acquire([&]() {
  15.     while(flag.load(std::memory_order_acquire) < 2);
  16.         std::cout << v.at(0) << std::endl; // must be 42
  17. });
  18. release.join();
  19. acqrel.join();
  20. acquire.join();

在此例中我们使用了 compare_exchange_strong，它便是比较交换原语（Compare-and-swap primitive），它有一个更弱的版本，即 compare_exchange_weak，它允许即便交换成功，也仍然返回false 失败。其原因是因为在某些平台上虚假故障导致的，具体而言，当 CPU 进行上下文切换时，另一线程加载同一地址产生的不一致。除此之外， compare_exchange_strong 的性能可能稍差于compare_exchange_weak，但大部分情况下， compare_exchange_strong 应该被有限考虑。
4. 顺序一致模型：在此模型下，原子操作满足顺序一致性，进而可能对性能产生损耗。可显式的通过std::memory_order_seq_cst 进行指定。最后来看一个例子：
 

  1. std::atomic<int> counter = {0};
  2. std::vector<std::thread> vt;
  3. for (int i = 0; i < 100; ++i) {
  4.     vt.emplace_back([&](){
  5.     counter.fetch_add(1, std::memory_order_seq_cst);
  6. });
  7. }
  8. for (auto& t : vt) {
  9.     t.join();
  10. }
  11. std::cout << "current counter:" << counter << std::endl;

这个例子与第一个宽松模型的例子本质上没有区别，仅仅只是将原子操作的内存顺序修改为了
memory_order_seq_cst，有兴趣的读者可以自行编写程序测量这两种不同内存顺序导致的性能差
异。

完整理示例

  1. #include <atomic>
  2. #include <thread>
  3. #include <vector>
  4. #include <iostream>
  5.  
  6. using namespace std;
  7. using namespace std::chrono;
  8.  
  9. const int N = 10000;
  10.  
  11. void relaxed_order()
  12. {
  13.     cout << "relaxed_order:" << endl;
  14.     atomic<int> counter = {0};
  15.     vector<thread> vt;
  16.     for(int i = 0; i < N; ++i)
  17.     {
  18.         vt.emplace_back([&](){
  19.             counter.fetch_add(1, memory_order_relaxed);
  20.         });
  21.     }
  22.  
  23.     auto t1 = high_resolution_clock::now();
  24.     for(auto& t: vt){
  25.         t.join();
  26.     }
  27.     auto t2 = high_resolution_clock::now();
  28.     auto duration = (t2 - t1).count();
  29.     cout << "relaxed order speed: " << duration / N << "ns" << endl;
  30. }
  31.  
  32. void release_consume_order()
  33. {
  34.     cout << "release_consume_order:" << endl;
  35.     //初始化为 nullptr 防止 consumer 线程从野指针进行读取
  36.     atomic<int*> ptr(nullptr);
  37.     int v;
  38.     std::thread producer([&](){
  39.         int* p = new int(42);
  40.         v = 1024;
  41.         ptr.store(p, std::memory_order_release);
  42.     });
  43.     std::thread consumer([&](){
  44.         int *p;
  45.         while(!(p = ptr.load(std::memory_order_consume)));
  46.         std::cout << "p: " << *p << std::endl;
  47.         std::cout << "v: " << v << std::endl;
  48.     });
  49.     producer.join();
  50.     consumer.join();
  51. }
  52.  
  53. void release_acquire_order()
  54. {
  55.     cout << "release_acquire_order: " << endl;
  56.     std::vector<int> v;
  57.     std::atomic<int> flag = {0};
  58.     std::thread release([&](){
  59.         v.push_back(42);
  60.         flag.store(1, std::memory_order_release);
  61.     });
  62.  
  63.     std::thread acqrel([&](){
  64.         int expected = 1;// must before compare_exchange_strong
  65.         while(!flag.compare_exchange_strong(expected, 2, std::memory_order_acq_rel))
  66.         {
  67.             expected = 1;// must before compare_exchange_strong
  68.         }
  69.         // flag has changed to 2
  70.     });
  71.     std::thread acquire([&](){
  72.         while(flag.load(std::memory_order_acquire) < 2);
  73.         std::cout << v.at(0) << std::endl;//must be 42
  74.     });
  75.     release.join();
  76.     acqrel.join();
  77.     acquire.join();
  78. }
  79.  
  80. void sequential_consistent_order()
  81. {
  82.     cout << "sequential_consistent_order: " << endl;
  83.     std::atomic<int> counter = {0};
  84.     std::vector<std::thread> vt;
  85.     for(int i = 0; i < 100; ++i)
  86.     {
  87.         vt.emplace_back([&](){
  88.             counter.fetch_add(1, std::memory_order_seq_cst);
  89.         });
  90.     }
  91.     for(auto& t: vt){
  92.         t.join();
  93.     }
  94.     std::cout << "current counter: " << counter << std::endl;
  95. }
  96.  
  97. int main()
  98. {
  99.     relaxed_order();
  100.     release_consume_order();
  101.     release_acquire_order();
  102.     sequential_consistent_order();
  103.  
  104.     return 0;
  105. }

运行结果：

 总结
C++11 语言层提供了并发编程的相关支持，本节简单的介绍了 std::thread, std::mutex,std::future 这些并发编程中不可回避的重要工具。除此之外，我们还介绍了 C++11 最重要的几个特性之一的『内存模型』，它们为 C++ 在标准化高性能计算中提供了重要的基础。
1. 请编写一个简单的线程池，提供如下功能：
ThreadPool p(4); // 指定四个工作线程
// 将任务在池中入队，并返回一个 std::future
auto f = pool.enqueue([](int life) {
return meaning;
}, 42);
// 从 future 中获得执行结果
std::cout << f.get() << std::endl;
 

  1. #ifndef XITI7_1_THREAD_POOL_H
  2. #define XITI7_1_THREAD_POOL_H
  3.  
  4. #include <vector>               // std::vector
  5. #include <queue>                // std::queue
  6. #include <memory>               // std::make_shared
  7. #include <thread>               // std::thread
  8. #include <mutex>                // std::mutex, std::unique_lock
  9. #include <condition_variable>   // std::condition_variable
  10. #include <future>               // std::future, std::packaged_task
  11. #include <functional>           // std::function, std::bind
  12. #include <stdexcept>            // std::runtime_error
  13. #include <utility>              // std::move, std::forward
  14.  
  15. class ThreadPool
  16. {
  17. public:
  18.     // initialize the number of concurrency threads``
  19.     ThreadPool(size_t);
  20.     //入队新线程任务
  21.     template<class F, class... Args>
  22.     decltype(auto) enqueue(F&& f, Args&&... args);
  23.     //销毁线程池和所有创建的线程
  24.     ~ThreadPool();
  25.  
  26. private:
  27.     // thread list, stores all threads
  28.     std::vector<std::thread> workers;
  29.     // queue task, the type of queue elements are functions with void return type
  30.     std::queue<std::function<void()>> tasks;
  31.     // for synchonization
  32.     std::mutex queue_mutex;
  33.     // std::condition_variable is a new feature from c++11,
  34.     // it's a synchronization primitives. it can be used
  35.     // to block a thread or threads at the same time until
  36.     // all of them modified condition_variable.
  37.     std::condition_variable condition;
  38.     bool stop;
  39. };
  40.  
  41. // constructor initialize a fixed size of worker
  42. ThreadPool::ThreadPool(size_t threads):stop(false)
  43. {
  44.     // initialize worker
  45.     for(size_t i = 0; i < threads; ++i)
  46.         // std::vector::emplace_back :
  47.         //    append to the end of vector container
  48.         //    this element will be constructed at the end of container, without copy and move behavior
  49.         workers.emplace_back([this]{// the lambda express capture this, i.e. the instance of thread pool
  50.             // avoid fake awake
  51.             while(true){
  52.                 // define function task container, return type is void
  53.                 std::function<void()> task;
  54.                 // critical section
  55.                 {
  56.                     //get mutex
  57.                     std::unique_lock<std::mutex> lock(this->queue_mutex);
  58.                     //block current thread
  59.                     this->condition.wait(lock, [this]{return this->stop || !this->tasks.empty();});
  60.                     // return if queue empty and task finished
  61.                     if(this->stop && this->tasks.empty())
  62.                         return ;
  63.                     // otherwise execute the first element of queue
  64.                     task = std::move(this->tasks.front());
  65.                     this->tasks.pop();
  66.                 }
  67.                 //execution
  68.                 task();
  69.             }
  70.         });
  71. }
  72.  
  73. // Enqueue a new thread
  74. // use variadic templates and tail return type
  75. template<class F, class... Args>
  76. decltype (auto) ThreadPool::enqueue(F&& f, Args&&... args)
  77. {
  78.     // deduce return type
  79.     using return_type = typename std::result_of<F(Args...)>::type;
  80.     //fetch task
  81.     auto task = std::make_shared<std::packaged_task<return_type()>>(
  82.                                                                        std::bind(std::forward<F>(f), std::forward<Args>(args)...));
  83.     std::future<return_type> res = task->get_future();
  84.     //critical section
  85.     {
  86.         std::unique_lock<std::mutex> lock(queue_mutex);
  87.         // avoid add new thread if theadpool is destroyed
  88.         if(stop)
  89.             throw std::runtime_error("enqueue on stopped ThreadPool");
  90.  
  91.         //add thread to queue
  92.         tasks.emplace([task]{(*task)();});
  93.     }
  94.     // notify a wait thread
  95.     condition.notify_one();
  96.     return res;
  97. }
  98.  
  99. inline ThreadPool::~ThreadPool()
  100. {
  101.     // critical section
  102.     {
  103.         std::unique_lock<std::mutex> lock(queue_mutex);
  104.         stop = true;
  105.     }
  106.  
  107.     // wake up all threads
  108.     condition.notify_all();
  109.     // let all processes into synchronous execution, use c++11 new for-loop: for(value:values)
  110.     for(std::thread &worker: workers)
  111.         worker.join();
  112. }
  113.  
  114. #endif // XITI7_1_THREAD_POOL_H
  1. #include <iostream> // std::cout, std::endl
  2.  
  3. #include <vector>   // std::vector
  4. #include <string>   // std::string
  5. #include <future>   // std::future
  6. #include <thread>   // std::this_thread::sleep_for
  7. #include <chrono>   // std::chrono::seconds
  8.  
  9. #include "xiti7_1_thread_pool.h"
  10.  
  11.  
  12. int main()
  13. {
  14.     // create a thread pool with max. 4 concurrency threads
  15.     ThreadPool pool(4);
  16.     // create execution results list
  17.     std::vector<std::future<std::string>> results;
  18.     // start eight thread task
  19.     for(int i = 0; i < 8; ++i)
  20.     {
  21.         // add all task to result list
  22.         results.emplace_back(
  23.                     // ass print task to thread pool
  24.                     pool.enqueue([i]{
  25.             std::cout << "hello " << i << std::endl;
  26.             // wait a sec when the previous line is out
  27.             std::this_thread::sleep_for(std::chrono::seconds(1));
  28.             // keep output and return the status of execution
  29.             std::cout << "world " << i << std::endl;
  30.             return std::string("---thread ") + std::to_string(i) + std::string(" finished.---\n") ;
  31.         })
  32.                     );
  33.     }
  34.  
  35.     //outputs
  36.     for(auto && result: results)
  37.         std::cout << result.get() << ' ';
  38.     std::cout << std::endl;
  39.  
  40.     return 0;
  41. }

运行结果：

 2. 请使用 std::atomic<bool> 实现一个互斥锁。
 

  1. #include <atomic>
  2. #include <thread>
  3. #include <iostream>
  4.  
  5. class mutex{
  6.     std::atomic<bool> flag{false};
  7.  
  8. public:
  9.     void lock()
  10.     {
  11.         while(flag.exchange(true, std::memory_order_relaxed))
  12.             ;
  13.         std::atomic_thread_fence(std::memory_order_acquire);
  14.     }
  15.  
  16.     void unlock()
  17.     {
  18.         std::atomic_thread_fence(std::memory_order_release);
  19.         flag.store(false, std::memory_order_relaxed);
  20.     }
  21. };
  22.  
  23. int a = 0;
  24.  
  25. int main()
  26. {
  27.     mutex mtx_a;
  28.     std::thread t1([&](){
  29.         std::cout << "t1 lock ==== a = " << a << std::endl;
  30.         mtx_a.lock();
  31.         a += 1;
  32.         mtx_a.unlock();
  33.         std::cout << "t1 unlock == a = " << a << std::endl;
  34.     });
  35.  
  36.     std::thread t2([&](){
  37.         std::cout << "t2 lock ==== a = " << a << std::endl;
  38.         mtx_a.lock();
  39.         a += 2;
  40.         mtx_a.unlock();
  41.         std::cout << "t2 unlock == a = " << a << std::endl;
  42.     });
  43.  
  44.     t1.join();
  45.     t2.join();
  46.  
  47.     std::cout << "a = " << a << std::endl;
  48.  
  49.     return 0;
  50. }

运行结果：

 

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号