Concurrency Runtime in Visual C++ 2010

PDC 2010 Hejlsberg的演讲中我们看到了VB.NET、C#新的简化异步编程的方式（可以下载新的Async CTP体验）。之前的TPL（Task Parallel Library）简化了并行编程。工业语言的飞速发展大大改进、简化了开发人员的编程方式。不仅是微软平台的托管语言，非托管语言也始终保持着一定程度的进化，我想通过两三篇文章来讲述Visual C++ 2010并行编程相关的内容。在介绍Visual C++ 2010的ppl之前，先简单介绍两个比较重要的新语法特性：

1、auto关键字

Visual C++ 2010中的auto关键字不再是以前简单的自动变量的概念，它被用于简化变量原型编译时的自动化推导。

vector numVec;

for (vector::const_iterator iter = numVec.begin(); iter != numVec.end(); ++iter) {

}

vector的遍历器写法稍微有些长了，代码过于烦琐。现在使用auto关键字编写：

vector numVec;

for (auto iter = numVec.begin(); iter != numVec.end(); ++iter) {

}

auto关键字的用法和C# 3.0新增的var关键字相同，用于简化那些编译时可以确定类型的声明。

Int32[] numArray = { 1, 2, 3, 4, 5 };
IOrderedEnumerable iter = numArray.Where(p => p > 0).OrderBy(p => p);
var iter2 = iter;

2、Lambda表达式

微软平台的托管语言都支持匿名函数，Lambda是为实现匿名函数提供了函数式的语法。

Int32[] numArray = { 1, 2, 3, 4, 5 };
var iter = numArray.Where(p => p > 0).Max();

旧有的Visual C++对于编写STL sort函数自定义排序需要这么编写：

bool less2(const int& lhs, const int& rhs) {
    return lhs < rhs;
}

vector numVec;

sort(numVec.begin(), numVec.end(), less2);
sort(numVec.begin(), numVec.end(), less<int>());

Visual C++ 2010：

vector numVec;

sort(numVec.begin(), numVec.end(), [&](int& lhs, int& rhs) { return lhs < rhs; });

对于一些不超过4、5行以上的代码，我个人更愿意使用匿名函数。Visual C++ 2010的ppl的定义被包含在ppl.h中。下面先看一段简单的代码：

#include <iostream>

using namespace std;

int sum = 0;

for (int i = 1; i < 101; ++i)
    sum += i;

wcout << “Result=” << sum << endl;

这组操作如果基于ppl该如何编写:

#include <iostream>
#include <ppl.h>

using namespace std;
using namespace Concurrency;

int sum = 0;

parallel_for(1, 101, 1, [&](int n) {
    sum += n;
});

wcout << “Result=” << sum << endl;

ppl提供了parallel_for，它的处理方式和for循环语句一样，下面是parallel_for在ppl.h中的定义：

template
void parallel_for(_Index_type _First, _Index_type _Last, _Index_type _Step, const _Function& _Func)

_First为起始索引、_Last为结束索引、_Step为步长、_Func为被调用函数指针。我以前的文章提到过Visual C++ 2008开始就提供了for each的支持，对于遍历vector之类的STL容器可以用for each实现（Visual C++的for each在for和each之间是有空格的，和Visual C#不同）：

vector numVec;

numVec.push_back(1);

for each (int i in numVec) {
    wcout << i << endl;
}

同样的ppl也提供了for each版本的并行函数parallel_for_each：

#include <algorithm>
#include <iostream>
#include <ppl.h>

using namespace std;
using namespace Concurrency;

int sum = 0;
const int NUMBER_ARRAY_SIZE = 10;
int numArray[NUMBER_ARRAY_SIZE];

generate(numArray, numArray + NUMBER_ARRAY_SIZE, rand);

parallel_for_each(numArray, numArray + NUMBER_ARRAY_SIZE, [&](int n) {
    sum += n;
});

wcout << “Result=” << sum << endl;

上面提到的parallel_for和parallel_for_each都是基于容器的遍历，ppl提供了parallel_invoke用于并行执行多个函数体。

#include “stdafx.h”
#include <algorithm>
#include <iostream>
#include <ppl.h>

using namespace std;
using namespace Concurrency;

void Func1() {
    int sum = 0;
    const int NUMBER_ARRAY_SIZE = 10;
    int numArray[NUMBER_ARRAY_SIZE];

    generate(numArray, numArray + NUMBER_ARRAY_SIZE, rand);

    for each (int n in numArray) {
        sum += n;
    }

    wcout << “Result=” << sum << endl;
}

parallel_invoke(&Func1, &Func1);

parallel_invoke最多支持10个function的函数重载。超过10以上的function完全可以用parallel_for和parallel_for_each来实现调用。

ppl提供了一些容器类型和对象用于线程安全的操作。先简单介绍两个容器类型concurrent_vector和concurrent_queue。这两个容器的使用方式和普通的STL的vector和queue相同，但有一个非常重要的区别在于它们的存储空间不是连续的，比如原来你操作vector时可以通过指针加减操作进行数据访问：

vector<int> vecArray;

int val = *(&vecArray[0] + 5);

但是对于concurrent_vector、concurrent_queue使用这种访问方式它的结果是未定义的，可能访问成功，可能是已经分配使用的内存空间或者访问违例。

1、concurrent_vector（使用方式和STL的vector相同）

#include <iostream>
#include <ppl.h>
#include <concurrent_vector.h>

concurrent_vector<int> vecArray;

parallel_invoke(
[&vecArray] {
    for (int i = 0; i < 50; ++i)
        vecArray.push_back(rand());
},
[&vecArray] {
    for (int i = 0; i < 100; ++i)
        vecArray.push_back(rand());
});

for each (int n in vecArray)
    wcout << n << endl;

2、concurrent_queue（STL的queue在访问元素时调用pop，而concurrent_queue则是try_pop）

#include <iostream>
#include <ppl.h>
#include <concurrent_queue.h>

concurrent_queue<int> queueArray;

parallel_invoke(
[&queueArray] {
    for (int i = 0; i < 50; ++i)
        queueArray.push(rand());
},
[&queueArray] {
    for (int i = 0; i < 100; ++i)
        queueArray.push(rand());
});

int val = 0;

while (!queueArray.empty() && queueArray.try_pop(val))
    wcout << val << endl;

ppl提供了一种可以合并操作的类型用于将不同线程间的本地存储值进行一定的操作，以往我们编写一个多线程或者并行操作计算总和的时候需要通过各种手段来确保操作值的更新，看一个最简单的例子：

volatile long sum2 = 0;

parallel_for(1, 101, [&sum2](int n) {
    InterlockedExchangeAdd(&sum2, n);
});

wcout << sum2 << endl;

再来看使用ppl新增的combinable类型完成同样的操作：

#include <iostream>
#include <ppl.h>

combinable sum3;

parallel_for(1, 101, [&sum3](int n) {
    int& local = sum3.local();

    local += n;
});

wcout << sum3.combine(plus()) << endl;

combinable的combine是一个模板函数用来定义函数指针，所以完全可以通过简单的lambda表达式定义自己的操作。默认提供了诸如：plus、multiplies、divides、minus操作。上面的例子最后一步的combine也可以这样写：

wcout << sum3.combine([&](int n1, int n2) { return n1 + n2; }) << endl;