使用并行的方法计算斐波那契数列 (Fibonacci)

更新：我的同事Terry告诉我有一种矩阵运算的方式计算斐波那契数列，更适于并行。他还提供了利用TBB的parallel_reduce模板计算斐波那契数列的代码（在TBB示例代码的基础上修改得来，比原始代码更加简洁易懂）。实验结果表明，这种方法在计算的斐波那契数列足够长时，可以提高性能。

矩阵方式计算斐波那契数列的原理：

代码：

#include <tbb/task_scheduler_init.h>
#include <tbb/blocked_range.h>
#include <tbb/parallel_reduce.h>
#include <tbb/tick_count.h>
#include <stdio.h>

using namespace std;
using namespace tbb;

//! Matrix 2x2 class
struct Matrix2x2
{
    //! Array of unsigned ints
    unsigned int v[2][2];
    Matrix2x2() {}
    Matrix2x2(unsigned int v00, unsigned int v01, unsigned int v10, unsigned int v11) {
        v[0][0] = v00; v[0][1] = v01; v[1][0] = v10; v[1][1] = v11;
    }
    Matrix2x2 operator * (const Matrix2x2 &to) const; //< Multiply two Matrices
};
//! matrix to be multiplied
static const Matrix2x2 Matrix1110(1, 1, 1, 0);

//! Identify matrix to be served as the base of the product
static const Matrix2x2 Matrix1001(1, 0, 0, 1);

//! Raw arrays matrices multiply
void Matrix2x2Multiply(const unsigned int a[2][2], const unsigned int b[2][2], unsigned int c[2][2])
{
    for( int i = 0; i <= 1; i++)
        for( int j = 0; j <= 1; j++)
            c[i][j] = a[i][0]*b[0][j] + a[i][1]*b[1][j];
}

Matrix2x2 Matrix2x2::operator *(const Matrix2x2 &to) const
{
    Matrix2x2 result;
    Matrix2x2Multiply(v, to.v, result.v);
    return result;
}

unsigned int serialFibo(unsigned int n)
{
    if (n < 2) {
        return n;
    }
    else
    {
        Matrix2x2 product = Matrix1110;
        for (int i = 2; i < n; i++)
        {
            product = Matrix1110*product;
        }
        return product.v[0][0];
    }
}

//! Functor for parallel_reduce
struct parallel_reduceFibBody {
    Matrix2x2 product;

    //! Constructor fills product with initial matrix
    parallel_reduceFibBody() : product( Matrix1001 ) { }
    //! Splitting constructor
    parallel_reduceFibBody( parallel_reduceFibBody& other, split ) : product( Matrix1001 ){}
    //! Join point
    void join( parallel_reduceFibBody &s ) {
        product = product * s.product;
    }
    //! Process multiplications
    void operator()( const blocked_range<int> &r ) {
        for( int k = r.begin(); k < r.end(); ++k )
            product = product * Matrix1110;
    }
};


unsigned int parallelFibo(unsigned int n)
{
    parallel_reduceFibBody b;
    parallel_reduce(blocked_range<int>(1, n, 3), b);
    return b.product.v[0][0];
}

int main(int argc, const char * argv[])
{
    task_scheduler_init init;
    
    unsigned int a = 40000;
    
    // serial Fibo
    //
    tick_count start1 = tick_count::now();
    unsigned int serialResult = serialFibo(a);
    tick_count end1 = tick_count::now();
    tick_count::interval_t time1 = end1 - start1;
    printf("serial fibo(%u) = %u, calculated in %f seconds\n", a, serialResult, time1.seconds());

    // parallel Fibo
    //
    tick_count start2 = tick_count::now();
    unsigned int parallelResult = parallelFibo(a);
    tick_count end2 = tick_count::now();
    tick_count::interval_t time2 = end2 - start2;
    printf("parallel fibo(%u) = %u, calculated in %f seconds\n", a, parallelResult, time2.seconds());

    return 0;
}

实验结果：

serial fibo(40000) = 160266427, calculated in 0.001070 seconds

parallel fibo(40000) = 160266427, calculated in 0.000568 seconds

=====================以下为老的博文==========================

今天给公司同事做关于并行编程的内部培训，大家对是否能用并行的方法计算斐波那契数列（Fibonacci），以及使用并行的方法能否提高其性能进行了一些讨论。

我的结论是：

1.可以使用并行的方法计算。

2. 至于能否提高性能，与计算斐波那契数列的实现方式有关。若采用最基础的递归方式，则对于“类似递归实现的计算斐波那契数列问题”，如果问题的计算量够大，则可能提高性能。如果用循环的方式实现斐波那契数列，则并行无论如何无法提高性能。

下面利用苹果的并行程序库GCD(Grand Central Dispatch)，写了一段代码来验证我的结论。为了仿真“大计算量的用递归实现的类斐波那契数列问题”，我在计算斐波那契数列的函数里加入了sleep 两秒的语句。实验结果显示，仅在这种方式下并行可以提高性能。

下面是代码和实验结果：

代码： 

#import <Foundation/Foundation.h>

unsigned int recursiveFibo(unsigned int n)
{
    [NSThread sleepForTimeInterval:2];
    
    if (n < 2)
        return n;
    else
        return recursiveFibo(n-1) + recursiveFibo(n-2);
}

unsigned int loopFibo(unsigned int n)
{    
    if (n < 2)
    {
        [NSThread sleepForTimeInterval:2];
        return n;
    }
    else
    {
        int fisub1 = 1;
        int fisub2 = 0;
        
        int fi = 0;
        for (unsigned int i = 2; i < n+1; i++)
        {
            [NSThread sleepForTimeInterval:2];
            fi = fisub1 + fisub2;
            fisub2 = fisub1;
            fisub1 = fi;
        }
        
        return fi;
    }
}

int main(int argc, const char* argv[])
{
    unsigned int a = 6;
    
    NSLog(@"serial caculating Fibonacci(%u)...", a);
    double start1 = [[NSDate date] timeIntervalSince1970];
    unsigned int fibo1 = loopFibo(a);
    double end1 = [[NSDate date] timeIntervalSince1970];
    NSLog(@"fibonacci(%u) = %u, cost %f seconds", a, fibo1, end1-start1);
    
    NSLog(@"parallel calculateing Fibonacci(%u)...", a);
    double start2 = [[NSDate date] timeIntervalSince1970];
    __block unsigned int fibo2sub1 = 0;
    __block unsigned int fibo2sub2 = 0;
    dispatch_group_t group = dispatch_group_create();
    dispatch_group_async(group, dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{fibo2sub1 = loopFibo(a-1);});
    dispatch_group_async(group, dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{fibo2sub2 = loopFibo(a-2);});
    dispatch_group_wait(group, DISPATCH_TIME_FOREVER);
    
    [NSThread sleepForTimeInterval:2];
    unsigned int fibo2 = fibo2sub1 + fibo2sub2;
    double end2 = [[NSDate date] timeIntervalSince1970];
    NSLog(@"fibonacci(%u) = %u, cost %f seconds", a, fibo2, end2 - start2);
    
    dispatch_release(group);

    return (0);
} //main

实验结果：

如果用递归实现（主程序中调用recursiveFibo），

对于“大计算量的类斐波那契数列问题”，并行可以提高性能。

2013-07-18 17:42:04.613 recursiveFibo[34666:303] serial caculating Fibonacci(6)...

2013-07-18 17:42:54.636 recursiveFibo[34666:303] fibonacci(6) = 8, cost 50.021813 seconds

2013-07-18 17:42:54.637 recursiveFibo[34666:303] parallel calculateing Fibonacci(6)...

2013-07-18 17:43:26.651 recursiveFibo[34666:303] fibonacci(6) = 8, cost 32.013492 seconds

如果把代码中的sleep语句全部注释掉，单纯计算斐波那契数列，则并行无法提高性能：

2013-07-18 18:04:39.885 recursiveFibo[35617:303] serial caculating Fibonacci(6)...

2013-07-18 18:04:39.888 recursiveFibo[35617:303] fibonacci(6) = 8, cost 0.000017 seconds

2013-07-18 18:04:39.889 recursiveFibo[35617:303] parallel calculateing Fibonacci(6)...

2013-07-18 18:04:39.890 recursiveFibo[35617:303] fibonacci(6) = 8, cost 0.000246 seconds

 

如果用循环实现（主程序中调用loopFibo），则无论计算量的大小，并行均不可能提高性能。

对于“大计算量的类斐波那契数列问题”,实验结果是：

2013-07-19 07:20:32.382 loopFibo[37029:303] serial caculating Fibonacci(6)...

2013-07-19 07:20:34.385 loopFibo[37029:303] fibonacci(6) = 8, cost 10.004636 seconds

2013-07-19 07:20:34.386 loopFibo[37029:303] parallel calculateing Fibonacci(6)...

2013-07-19 07:20:38.389 loopFibo[37029:303] fibonacci(6) = 8, cost 10.005218 seconds

对于单纯计算斐波那契数列，实验结果是：

2013-07-19 07:31:59.530 loopFibo[37404:303] serial caculating Fibonacci(6)...

2013-07-19 07:31:59.532 loopFibo[37404:303] fibonacci(6) = 8, cost 0.000013 seconds

2013-07-19 07:31:59.532 loopFibo[37404:303] parallel calculateing Fibonacci(6)...

2013-07-19 07:31:59.533 loopFibo[37404:303] fibonacci(6) = 8, cost 0.000042 seconds

从此文也可以看出，用循环的方式实现斐波那契数列是一种更好的方式。