2.4在运行时选择线程数量

在运行时选择线程数量

C++标准库中对此有所帮助的特性是std::thread::hardware_currency()。这个函数返回一个对于给定程序执行时能够真正并发运行的线程数量的指示。例如，在多核系统上它可能是CPU 核心的数量。它仅仅是一个提示，如果该信息不可用则函数可能会返回0，但它对于在线程间分割任务是一个有用的指南。

清单2.8展示了std::accumulate 的一个简单的并行版本实现。它在线程之间划分所做的工作，使得每个线程具有最小数目的元素以避免过多线程的开销。请注意，该实现假定所有的操作都不引发异常，即便异常可能会发生。例如，std::thread构造函数如果不能启动一个新的执行线程那么它将引发异常。在这样的算法中处理异常超出了这个简单示例的范围。

//std::accumulate的简单的并行版本
#include <thread>
#include <numeric>
#include <algorithm>
#include <functional>
#include <vector>
#include <iostream>template<typename Iterator,typename T>
struct accumulate_block
{void operator()(Iterator first,Iterator last,T& result){result=std::accumulate(first,last,result);}
};template<typename Iterator,typename T>
T parallel_accumulate(Iterator first,Iterator last,T init)
{unsigned long const length=std::distance(first,last);if(!length)return init; //❶unsigned long const min_per_thread=25;unsigned long const max_threads=(length+min_per_thread-1)/min_per_thread; //❷unsigned long const hardware_threads= //❸std::thread::hardware_concurrency();unsigned long const num_threads=std::min(hardware_threads!=0?hardware_threads:2,max_threads);unsigned long const block_size=length/num_threads; //❹std::vector<T> results(num_threads);std::vector<std::thread>  threads(num_threads-1); //❺Iterator block_start=first;for(unsigned long i=0;i<(num_threads-1);++i){Iterator block_end=block_start;std::advance(block_end,block_size); //❻threads[i]=std::thread( //❼accumulate_block<Iterator,T>(),block_start,block_end,std::ref(results[i]));block_start=block_end; //❽}accumulate_block<Iterator,T>()(block_start,last,results[num_threads-1]); //❾std::for_each(threads.begin(),threads.end(),std::mem_fn(&std::thread::join)); //❿return std::accumulate(results.begin(),results.end(),init); //⓫
} int main()
{std::vector<int> vi;for(int i=0;i<10;++i){vi.push_back(10);}int sum=parallel_accumulate(vi.begin(),vi.end(),5);std::cout<<"sum="<<sum<<std::endl;
}

虽然这是一个相当长的函数，但它实际上是很直观的。如果输入范围为空❶，只返回初始值init。否则，此范围内至少有一个元素，于是你将要处理的元素数量除以最小的块大小，以获取线程的最大数量❷。这是为了避免当范围中只有五个值时，在一个32核的机器上创建32个线程。

要运行的线程数是你计算出的最大值和硬件线程数量❸的较小值。你不会想要运行比硬件所能支持的更多的线程（超额订阅，oversubscription)，因为上下文切换将意味着更多的线程会降低性能。如果对std::thread::hardware_concurrency()的调用返回0，你只需简单地替换上你所选择的数量，在这个例子中我选择了2。你不会想要运行过多的线程，因为在单核的机器上这会使事情变慢，但同样地你也不希望运行的过少，因为那样的话，你就会错过可用的并发。

每个待处理的线程的条目数量是范围的长度除以线程的数量❹。如果你担心数量不能整除，没必要——稍后再来处理。

既然你知道有多少个线程，你可以为中间结果创建一个 std::vector<T>，同时为线程创建一个 std::vector<std::thread>❺。请注意，你需要启动比
num_threads 少一个的线程，因为已经有一个了。

启动线程是个简单的循环：递进block_end迭代器到当前块的结尾❻，并启动一个新的线程来累计此块的结果❼。下一个块的开始是这一个的结束❽。

当你启动了所有的线程后，这个线程就可以处理最后的块❾。这就是你处理所有未被整除的地方。你知道最后一块的结尾只能是last，无论在那个块里有多少元素。一旦累计出最后一个块的结果，你可以等待所有使用std::for_each 生成的线程❿，如清单2.7中所示，接着通过最后调用std::accumulate将结果累加起来⓫。

在你离开这个例子前，值得指出的是在类型T的加法运算符不满足结合律的地方(如float和 double)，这个parallel_accumulate的结果可能会跟std::accumulate的有所出入，这是将范围分组成块导致的。此外，对迭代器的需求要更严格一些，它们必须至少是前向迭代器(forward iterators)，然而std::accumulate可以和单通输入迭代器(input iterators）一起工作，同时T必须是可默认构造的(default constructible)以使得你能够创建results向量。这些需求的各种变化是并行算法很常见的：就其本质而言，它们以某种方式的不同是为了使其并行，并且在结果和需求上产生影响。另外值得一提的是，因为你不能直接从一个线程中返回值，所以你必须将相关项的引用传入results向量中。从线程中返回结果的替代方法，会通过使用future来实现。

在这种情况下，每个线程所需的所有信息在线程开始时传入，包括存储其计算结果的位置。实际情况并非总是如此。有时，作为进程的一部分有必要能够以某种方式标识线程。你可以传入一个标识数，如同在清单2.7中 i 的值，但是如果需要此标识符的函数在调用栈中深达数个层次，并且可能从任意线程中被调用，那样做就很不方便。当我们设计C++线程库时就预见到了这方面的需求，所以每个线程都有一个唯一的标识符。