使用並行的方法計算斐波那契數列 (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

從此文也可以看出，用循環的方式實現斐波那契數列是一種更好的方式。