c++多线程编程：实现标准库accumulate函数的并行计算版本

Posted 2020-06-13

      今天使用c++实现了标准库头文件<numeric>中的accumulate函数的并行计算版本,代码如下,注释写的比较详细,仅对其中几点进行描述:

      ①该实现假定不发生任何异常,故没有对可能产生的异常进行处理

      ②第42行的语句：

      const unsigned int num_thread = std::min((hardware_thread != 0 ? hardware_thread : 2), max_thread);

        要运行的线程数是计算出的最大线程数和硬件线程数量的较小值。这是因为若运行的线程数超出了硬件支持的范围,CPU的上下文切换会降低性能。又因为hardware_thread的值可能为0,在这种情况下用户需要自行替换线程的数量,在代码中为2,因为在单核的机器上运行过多的线程会导致性能降低,但过少的线程也会使用户错过可用的并发。

      ③第44行计算每个线程操作的元素个数时算式可能无法整除，但无须担心，因为最后一个线程将会处理剩下的所有元素，如第68行语句所示：

      Accum<Iterator, T>()(block_begin, last, results[num_thread - 1]);

      ④注意第64行的语句：

      threads[i] = thread(Accum<Iterator, T>(), block_begin, block_end, std::ref(results[i]));

      传给线程执行的函数的第三个参数增加了std::ref()，该函数包含在头文件<functional>中。在一般情况下，thread对象的构造函数只是简单地拷贝用户提供的参数，然后传递给线程关联的可调用对象。也就是说，该可调用对象接收的是该参数的副本，对其所作的修改无法影响到最初用户所传递的参数。若需要接收参数的引用版本，则需要使用std::ref()函数。

 1 //实现标准库头文件<numeric>中accumulate函数的并行版本
 2 #include <iostream>
 3 #include <thread>
 4 #include <numeric>
 5 #include <algorithm>
 6 #include <vector>
 7 #include <functional>
 8 #include <utility>
 9 
10 using std::thread;
11 using std::vector;
12 using std::accumulate;
13 using std::cout;
14 using std::endl;
15 
16 template <typename Iterator, typename T> class Accum
17 {
18 public:
19     void operator() (Iterator first, Iterator last, T &sum)
20     {
21         sum = std::accumulate(first, last, sum);
22     }
23 };
24 
25 template <typename Iterator, typename T>
26 T ParallelAccum(Iterator first, Iterator last, T &sum)
27 {
28     //计算迭代器中包含的元素数量
29     const unsigned int len = std::distance(first, last);
30     //若迭代器中没有元素则直接返回
31     if (!len)
32     {
33         return sum;
34     }
35     //每个线程处理的元素的最小数量
36     const unsigned int min_per_thread = 25;
37     //获取线程的最大数量,向上取整
38     const unsigned int max_thread = (len - 1 + min_per_thread) / min_per_thread;
39     //获取机器支持的并发线程数
40     const unsigned int hardware_thread = thread::hardware_concurrency();
41     //取上述两者的较小值,同时避免线程数过少
42     const unsigned int num_thread = std::min((hardware_thread != 0 ? hardware_thread : 2), max_thread);
43     //最终实际上每个线程处理的元素个数
44     const unsigned int block_size = len / num_thread;
45     //保存每个线程累加的结果
46     vector<T> results(num_thread);
47     //启动比num_thread - 1个线程，因为main函数本身已开启一个线程
48     vector<thread> threads(num_thread - 1);
49     //
50     cout << "Number of elements: " << len << endl;
51     cout << "Hardware concurrency: " << hardware_thread << endl;
52     cout << "Maximum number of threads: " << max_thread << endl;
53     cout << "Number of threads: " << num_thread << endl;
54     cout << "Block size: " << block_size << endl;
55     cout << "Started parallel calculating..." << endl;
56     //开始并行计算
57     Iterator block_begin = first;
58     for (unsigned int i = 0; i < (num_thread - 1); ++i)
59     {
60         Iterator block_end = block_begin;
61         //将迭代器向前推进一个块,到达当前块的末尾位置
62         std::advance(block_end, block_size);
63         //传递参数,通常情况下thread的构造函数将复制所提供的参数,需要将模板参数转为引用
64         threads[i] = thread(Accum<Iterator, T>(), block_begin, block_end, std::ref(results[i]));
65         block_begin = block_end;
66     }
67     //处理最后一个线程,由于block_size = len / num_thread得到的结果不一定为整数,该线程处理剩余的所有元素
68     Accum<Iterator, T>()(block_begin, last, results[num_thread - 1]);
69     //对threads中所有线程调用join()
70     std::for_each(threads.begin(), threads.end(), std::mem_fn(&thread::join));
71     //
72     return accumulate(results.begin(), results.end(), sum);
73 }
74 
75 int main()
76 {
77     vector<int> i_vec;
78     int sum = 0;
79     for (int i = 1; i != 501; ++i)
80     {
81         i_vec.push_back(i);
82     }
83     sum = ParallelAccum(i_vec.cbegin(), i_vec.cend(), sum);
84     cout << "sum = " << sum << endl;
85     system("pause");
86     return 0;
87 }