遇见C++ PPL：C++ 的并行和异步

Written by Allen Lee

You held it all, but you were careless to let it fall. You held it all, and I was by your side powerless.
– Linkin Park, Powerless

并行计算正弦值

      假设我们有一个数组，里面包含一组随机生成的浮点数，现在要计算每个浮点数对应的正弦值，如果你看过我的《遇见C++ Lambda》，你可能会想到用for_each函数，如代码1所示。为了可以把数组里的浮点数替换成对应的正弦值，我们需要把Lambda的参数声明为引用，如果你想保留那些浮点数，可以创建一个新的数组存放计算结果。

代码 1

      值得提醒的是，这里使用begin和end两个函数分别获取数组的起止位置，这是C++ 11的推荐写法。此前，我们使用STL容器的begin和end两个成员函数分别获取起止位置，但这种做法无法覆盖C风格数组；如今，C++ 11通过begin和end两个函数把获取C风格数组和STL容器的起止位置的写法统一起来，不难想象，遵循新的写法可以提高代码的一致性。

      STL提供的for_each函数是串行执行的，如果你想充分利用多核的优势，可以考虑换用PPL（Parallel Patterns Library）提供的parallel_for_each函数，整个改造过程只需三步：

1. #include <ppl.h>
2. using namespace concurrency;
3. 把for_each改为parallel_for_each，如代码2所示

代码 2

      需要说明的是，如果你在Visual C++ 2010上使用PPL，你需要引用Concurrency命名空间（首字母大写），这里引用的concurrency命名空间（全小写）是Visual C++ 2012的PPL为了和其他常见的全小写命名空间（如stl）保持一致而创建的命名空间别名。

      如果你不想影响那些浮点数，可以创建一个新的数组，然后通过parallel_for函数把计算结果对应地存到新的数组里，如代码3所示。这里选择parallel_for函数主要是为了借助索引管理两个数组的元素的对应关系，如果你要在多个数组之间周旋，比如说，你要为A、B、C和D四个集合实现对应元素的 (A + B) / (C - D) 操作，那么使用parallel_for函数就会非常直观。

代码 3

      对于我们这里的简单需求，如果你不想自己管理元素的对应关系，可以考虑parallel_transform函数，如代码4所示。parallel_transform函数的前两个参数指定输入容器的起止位置，第三个参数指定输出容器的开始位置，前两个参数指向的位置之间的元素个数必须小于或等于第三个参数指向的位置和输出容器的结束位置之间的元素个数，否则将会出错。

代码 4

并行数奇数个数

      在《遇见C++ Lambda》里，我们通过for_each函数数一下随机生成的整数里有多少个奇数，这个过程可以并行化吗？可以的，一般的做法是声明一个变量存放个数，在迭代的过程中一旦发现奇数就递增一下这个变量，由于涉及到多线程，可以通过系统提供的InterlockedIncrement函数确保递增操作的安全，如代码5所示。

代码 5

      上面的代码可以得到正确结果，但存在一个问题，每次发现奇数都要调用InterlockedIncrement函数，如果nums数组里的奇数占大多数，那么调用InterlockedIncrement函数带来的开销可能会抵消并行带来的好处，最终导致执行效率甚至比不上串行版本。为了避免这种影响，我们可以把volatile变量和InterlockedIncrement函数的组合写法替换成PPL提供的combinable对象，如代码6所示。

代码 6

      combinable对象是如何协助parallel_for_each函数提高执行效率的呢？这个需要稍微了解一下parallel_for_each函数的工作方式，简单的说，它会把我们传给它的数据分成N块，分别交给N个线程并行处理，但同一块数据会在对应的线程里串行处理，这意味着处理同一块数据的代码可以直接实现同步，combinable对象正是利用这点减少不必要的同步，从而提高parallel_for_each函数的执行效率。

      combinable对象会为每个线程提供一个线程局部存储（Thread-Local Storage），每个线程局部存储都会使用创建对象时提供的Lambda进行初始化。我们可以通过local成员函数访问当前线程的线程局部存储，因为combinable对象保证local成员函数返回的对象一定是当前线程的，所以我们可以放心的直接操作。当每个线程的操作都完成之后，我们就可以调用combine成员函数把每个线程局部存储的结果汇总起来，这个时候会产生线程之间的同步，但同步工作由combinable对象负责，无需我们费心，我们只需告诉它汇总的方法就行了，在我们的示例里，这个逻辑是STL提供的plus函数对象。

      parallel_for_each函数和combinable对象的组合写法本质上就是一个Reduce过程，PPL提供了一个parallel_reduce函数专门处理这类需求，如代码7所示，它非常直接地展示了parallel_for_each函数和combinable对象隐藏起来的二段处理过程。

代码 7

      第一个阶段，parallel_reduce函数会把我们传给它的数据分成N块，分别交给N个线程并行处理，每个线程执行的代码由第四个参数指定。在我们的示例里，这个参数是一个Lambda，parallel_reduce函数会通过Lambda的参数告诉我们每块数据的起止位置，以及计算的初始值，这个初始值其实来自parallel_reduce函数的第三个参数，而Lambda的函数体则是不折不扣的串行代码。所有线程执行完毕之后就会进入第二个阶段，汇总每个线程的执行结果，汇总的方法由第五个参数指定。

      parallel_reduce函数和前面提到的parallel_transform函数可以组合起来实现并行MapReduce操作，而STL提供的transform和accumulate两个函数则可以组合起来实现串行MapReduce操作。

同时执行不同任务

      假设我们现在的任务是计算一组随机整数里的所有奇数之和与第一个素数的商，一般的做法是按顺序执行以下步骤：

1. 生成一组随机整数
2. 计算所有奇数的和
3. 找出第一个素数
4. 计算最终结果

由于第二、三步是相互独立的，它们只依赖于第一步的结果，我们可以同时执行这两步提高程序的整体执行效率。那么，如何同时执行两个不同的代码呢？可以使用parallel_invoke函数，如代码8所示。

代码 8

      parallel_invoke函数最多可以接受十个参数，换句话说，它可以同时执行最多十个不同的代码，如果我们需要同时执行超过十个代码呢？这个时候我们可以考虑创建一个Lambda数组，然后交给parallel_for_each/parallel_for函数去执行，如代码9所示。

代码 9

      这些代码都能得到正确的结果，但它们都有一个缺点——阻塞当前线程。想想看，一般需要动用并行编程的地方都是计算量比较大的，如果要等它们算好才能继续，恐怕会把用户惹毛，但是，如果不等它们算好，后面的步骤可能没法正常运作，怎么办呢？

async + continuation

      我们可以通过task对象异步执行第一步，然后通过continuation把后续步骤按照既定的顺序连结起来，这样既可避免阻塞当前线程，又能确保正确的执行顺序。

      首先，把各个步骤需要共享的变量挪到前面，如代码10所示，这些变量将被对应的步骤捕获并使用。

代码 10

      然后，通过create_task函数创建一个task对象，异步执行第一步，如代码11所示。create_task函数负责用我们传给它的Lambda创建task对象，这个Lambda可以有返回值，但不能接受任何参数，否则将会编译出错。当我们需要从外部获取输入时，可以借助闭包或者调用其他函数。

代码 11

      接着，在create_task函数返回的task对象上调用then函数创建一个continuation，如代码12所示。这个continuation会在前一个task结束之后才开始，从而确保执行第二、三步所需的数据在执行之前准备好。

代码 12

      最后，在then函数返回的task对象上调用then函数创建一个continuation，执行第四步，如代码13所示。理论上，你可以通过then函数创建任意数目的continuation。值得提醒的是，在Metro风格的应用程序里，continuation默认是在UI线程里执行的，因此可以在continuation里直接更新UI控件而不必使用Dispatcher对象，但是，如果你想在后台执行continuation，你需要把task_continuation_context::use_arbitrary传给then函数的_ContinuationContext参数。

代码 13

      如果你把这些代码组合起来放在main函数里执行，并且在最后放置一个cin.get()等待结果，那么一切都会运作正常。但是，如果你把它们放在一个work函数里，然后在main函数里调用这个work函数，你可能会碰到异常，大概是说我们读了不该读的地方。这是因为我们的task是异步执行的，执行的过程中work函数可能已经返回了，连带那些分配在栈上的变量也一并销毁了，如果此时访问那些变量就会出错。怎么解决这个问题？

      前面曾经说过，我们传给create_task函数的Lambda可以有返回值，这个返回值将会通过参数传给后续的continuation，我们可以通过这个机制把那些变量内化到Lambda里，如代码14所示。

代码 14

      值得提醒的是，我们通过tuple对象把第二、三步的计算结果传给第四步，然后通过tie函数把tuple对象里的数据提取到两个变量里，这种写法类似于F#的"let sum_of_odds, first_prime = operands"。

      另外，如果你担心在task之间传递vector<int>会带来性能问题，可以通过智能指针单独处理，如代码15所示。智能指针本身是一个对象，会随着work函数的返回而销毁，因此需要通过按值传递的方式捕获它。

代码 15

      到目前为止，我们还没有任何异常处理的代码，如果其中一个task抛出异常怎么处理？我们可以在任务链的末端加上一个特殊的continuation，如代码16所示，它的参数是一个task对象，任务链上的任何一个task抛出来的异常都会传到这里，这个异常可以通过调用get函数重新抛出，因此我们用一个try…catch语句把get成员函数的调用包围起来，然后处理它抛出来的异常。

代码 16

你可能会问的问题

      1. 使用PPL需要什么条件？

      parellel_for、parellel_for_each和parallel_invoke等函数可以在Visual Studio 2010上使用，使用时需要包含ppl.h头文件并引用Concurrency命名空间，而parellel_transform和parallel_reduce函数，以及和task相关的部分则需要Visual Studio 2012，使用时需要分别包含ppl.h和ppltask.h头文件。

      2. 能否推荐一些PPL的参考资料？

      关于本文提到的PPL函数和类型，可以参考MSDN的concurrency类库。另外，MSDN的Parallel Patterns Library (PPL)和Parallel Programming with Microsoft Visual C++: Design Patterns for Decomposition and Coordination on Multicore Architectures也是很好的学习资料。

      3. STL是否提供task的替代品？

      C++ 11的STL提供了std::future类，结合std::async函数可以实现task的异步效果，如代码17所示，但std::future类目前不支持contiuation，只能通过get成员函数获取结果，调用get成员函数的时候，如果相关代码还在执行，则会阻塞当前线程。

代码 17

      4. PPL能否在Windows以外的平台上使用？

      PPL目前只能在Windows上使用，如果你想在其他平台上进行类似的并行编程，可以考虑Intel Threading Building Blocks，它同时支持Windows、Mac OS X和Linux，提供的API和PPL的类似。TBB是开源的，Intel为它提供商业和GPLv2两种许可协议。

      5. 能否推荐一些TBB的参考资料？

      Intel Threading Building Blocks: Outfitting C++ for Multi-Core Processor Parallelism是一本不错的学习资料，另外，Intel也提供了丰富的示例代码。

*声明：本文已经首发于InfoQ中文站，版权所有，《遇见PPL：C++ 的并行和异步》，如需转载，请务必附带本声明，谢谢。