进入多核时代的C++示例

示例
用TBB写并行transform代替原std::transform,分别使用parallel_for和pipeline实现

 

 

#include <iostream>
#include <algorithm>
#include <vector>
#include <list>
#include <math.h>
#include <tbb/task_scheduler_init.h>
#include <tbb/blocked_range.h>
#include <tbb/parallel_for.h>
#include <tbb/pipeline.h>
#include <tbb/tick_count.h>
 
//-----------------随机存取迭代器版本-------------------------------------
//both_random_helper，作用:测试两个模板是否都是random_access_iterator_tag
//是则Iter_cat返回random_access_iterator_tag
//否则返回forward_iterator_tag
template<class _Cat1, class _Cat2>
struct both_random_helper{
    typedef std::forward_iterator_tag Iter_cat;
};
 
template<>
struct both_random_helper<
    std::random_access_iterator_tag,
    std::random_access_iterator_tag>{
    typedef std::random_access_iterator_tag Iter_cat;
};
 
// 用于存放一对迭代器
template<class _InIt, class _OutIt>
struct Iter_pair{
    _InIt m_in;
    _OutIt m_out;
    Iter_pair(_InIt _in, _OutIt _out)
        :m_in(_in),m_out(_out){}
    //支持blocked_range拆分
    Iter_pair operator+(size_t off) const
    {
        return Iter_pair(m_in+off, m_out+off);
    }
 
    size_t operator-(Iter_pair rhs) const
    {
        return m_in-rhs.m_in;
    }
 
    bool operator<(Iter_pair rhs) const
    {
        return m_in<rhs.m_in;
    }
};
 
// 随机存取迭代器版本
template<class _InIt, class _OutIt, class _Fn1>
struct op_parallel_transform{
    op_parallel_transform(_Fn1 _Func)
        :m_Func(_Func){}
    void operator()(const tbb::blocked_range<Iter_pair<_InIt,_OutIt> > &r) const
    {
        std::transform(r.begin().m_in, r.end().m_in, r.begin().m_out, m_Func);
    }
private:
    _Fn1 m_Func;
};
 
template<class _InIt, class _OutIt, class _Fn1>
_OutIt _parallel_transform(_InIt _First, _InIt _Last, _OutIt _Dest,
    _Fn1 _Func, std::random_access_iterator_tag)
{
    // 使用parallel_for来处理
    typedef typename Iter_pair<_InIt,_OutIt> iter_pair_type;
    _OutIt LastDest = _Dest + (_Last - _First);
    iter_pair_type begin(_First, _Dest);
    iter_pair_type end(_Last, LastDest);
    tbb::blocked_range<iter_pair_type> x(begin, end);
    tbb::parallel_for(x, op_parallel_transform<_InIt,_OutIt,_Fn1>(_Func), tbb::auto_partitioner());
    return LastDest;
}
 
//-----------------顺序存取迭代器版本-------------------------------------
template<class _InIt>
struct filter_in : tbb::filter{
    filter_in(_InIt _First, _InIt _Last)
        :tbb::filter(true),m_First(_First), m_Last(_Last){}
    void* operator()(void*)
    {
        if(m_First==m_Last) return NULL;
 
        void* p = &(*m_First);
        ++m_First;
        return p;
    }
private:
    _InIt m_First, m_Last;
};
 
template<class _Fn1>
struct filter_process : tbb::filter{
    typedef typename _Fn1::result_type r_type;
    typedef typename _Fn1::argument_type a_type;
 
    filter_process(_Fn1 _Func)
        :tbb::filter(false),m_Func(_Func){}
    void* operator()(void* data)
    {
        a_type &at = *(a_type*)data;
        m_r = m_Func( at );
        return &m_r;
    }
private:
    _Fn1 m_Func;
    r_type m_r;
};
 
template<class _OutIt, class _DataType>
struct filter_out : tbb::filter{
    filter_out(_OutIt _Dest)
        :tbb::filter(true),m_Dest(_Dest){}
    void* operator()(void* data)
    {
        _DataType *p = (_DataType*) data;
        *m_Dest = *p;
        ++m_Dest;
        return NULL;
    }
private:
    _OutIt m_Dest;
};
 
template<class _InIt, class _OutIt, class _Fn1>
_OutIt _parallel_transform(_InIt _First, _InIt _Last, _OutIt _Dest,
    _Fn1 _Func, std::forward_iterator_tag)
{
    // 使用管线pipeline来处理
    tbb::pipeline pipeline;
    filter_in<_InIt> f1(_First, _Last);
    filter_process<_Fn1> f2(_Func);
    filter_out<_OutIt, _Fn1::result_type> f3(_Dest);
    pipeline.add_filter(f1);
    pipeline.add_filter(f2);
    pipeline.add_filter(f3);
    pipeline.run(3);
    return _Dest;
}
 
//----------------------parallel_transform----------------------------
template<class _InIt, class _OutIt, class _Fn1>
inline
_OutIt parallel_transform(_InIt _First, _InIt _Last, _OutIt _Dest, _Fn1 _Func)
{
    typedef typename std::iterator_traits<_InIt>::iterator_category cat1;
    typedef typename std::iterator_traits<_OutIt>::iterator_category cat2;
    return _parallel_transform(_First, _Last, _Dest,
        _Func, both_random_helper<cat1,cat2>::Iter_cat());
}
 
// 测试代码
struct SinFunctor
    :std::unary_function<double, double>{
    double operator()(double &d) const
    {
        // 高强度运算
        double sum = 0;
        for(int i=0; i<10000; i++)  sum += sin(i*d);
        return sum;
    }
};
 
int main() 
{
    // 随机存取迭代
    std::vector<double> a(10000,1.5);
    // 顺序存取迭代
    std::list<double>   b(10000,1.5);
 
    tbb::task_scheduler_init init;
 
    tbb::tick_count t0,t1;
 
    t0 = tbb::tick_count::now();
    parallel_transform(a.begin(), a.end(), a.begin(), SinFunctor());
    t1 = tbb::tick_count::now();
    std::cout << "并行（随机迭代）" << (t1 - t0).seconds() << std::endl;
    
    t0 = tbb::tick_count::now();
    std::transform(a.begin(), a.end(), a.begin(), SinFunctor());
    t1 = tbb::tick_count::now();
    std::cout << "原版（随机迭代）" << (t1 - t0).seconds() << std::endl;
 
    t0 = tbb::tick_count::now();
    parallel_transform(b.begin(), b.end(), b.begin(), SinFunctor());
    t1 = tbb::tick_count::now();
    std::cout << "并行（顺序迭代）" << (t1 - t0).seconds() << std::endl;
 
    t0 = tbb::tick_count::now();
    std::transform(b.begin(), b.end(), b.begin(), SinFunctor());
    t1 = tbb::tick_count::now();
    std::cout << "原版（顺序迭代）"<< (t1 - t0).seconds() << std::endl;
    
    std::cin.get();
    return 0;
}

 在我的双核Centrino电脑上测试结果如下：

并行（随机迭代）3.17299
原版（随机迭代）5.41531
并行（顺序迭代）3.13054
原版（顺序迭代）5.33182

     在顺序存取迭代器版本的_parallel_transform中，因为迭代器不能随意跳转，所以使用了tbb::pipeline加三个filter分 别执行顺序读取，处理和写入。其中的处理是可以并行处理的。从上面的结果可以看出，pipeline的性能甚至不亚于parallel_for循环。关于 pipeline的使用说明，请参考文章：TBB流水线

    这里写的parallel_transform可以直接替换大家原有代码中的std::transform，当然如果有兴趣的话完全可以把标准库中的算法 全用TBB改写成并行算法。不过要注意的一点是并不是任何地方都适合使用并行运算的，象这个例子中测试用的处理算子是“for(int i=0; i<10000; i++)  sum += sin(i*d);”这样的需要耗时较长的运算，如果把它改成“return sin(d);”。那么考虑到线程调度以及TBB的任务调度，并行算法的效率可能还不如串行算法。

    TBB库可以从这里下载：http://www.threadingbuildingblocks.org/download.php

    另外再推荐几篇TBB入门文章：
    Intel Threading Building Blocks 之 并行循环
    Intel Threading Building Blocks 之 并发容器
    Intel Threading Building Blocks 基于任务编程

相信度过了自己的20岁生日之后的C++，在多核时候将再创辉煌！