インテルTBBから学ぶループの並列化

インテルTBBを通じて学ぶ並列処理（2）

　本連載では、C++プログラムで並列処理を行うアプローチを提供するテンプレートライブラリ「インテルスレッディング・ビルディング・ブロック」（インテルTBB）について解説します。今回はインテルTBBのアルゴリズムテンプレートとループの並列化を説明します。

はじめに

　この連載では、並列処理を高度に抽象化したインテルTBBを通じて、並列化の考え方を取得することを目的としています。今後、並列化は当たり前のものとなり、さまざまな形でサポートされるようになります。並列化処理の根底に流れる考え方を身につければ、その変化に対応できます。

　今回はインテルTBBのアルゴリズムテンプレートとループの並列化について解説します。この連載のサンプルはあくまでもインテルTBBの使い方を説明するものであり、実務を特別に意識したものではありません。その点をご理解下さい。

対象読者

　筆者が想定している読者はC++の基本的文法を理解し、並列化プログラミングに興味を持っている方です。高度なC++テクニックを極力さけ、基本的な文法さえ分かれば読めるように極力注意しますので、並列化に興味を持っている方はぜひこの連載に目を通して下さい。

必要な環境

　C++コンパイラが必要です。お持ちでない方は無償で提供されているマイクロソフト社の「Visual Studio C++ 2008 Express Edition」をダウンロードするなどして入手して下さい。

　この連載は基本的にWindows環境を想定して解説しますが、インテルTBBそのものは他のOS上でも動作しますので、適宜読み替えて参考にして下さい。

　その他に、インテルTBBを使うための準備が必要ですので次項で解説します。

インテルTBBの準備

　インテルTBBを使用する方法は2つあります。1つ目は「インテル Parallel Studio」を使用することです。インテル Parallel StudioはインテルTBBがあらかじめ用意されており、オプションを指定するだけでインテルTBBを使用することができます。詳しくは『インテル Parallel Studioを使って並列化プログラミングを試してみた』（CodeZine）を参照して下さい。

　2つ目の方法は、オープンソース版のインテルTBBを公式ホームページからダウンロードしてきて、ライブラリとインクルードファイルへのパスを指定することです。これからその手順を解説します。ただし、この記事は2009年10月に書かれたものであり、今後URLが変更される恐れがありますのでその点に注意して下さい。

　まずは公式ホームページにアクセスし、「Downloads」と書かれたタブを選択します。そして、「Stable Release」という見出しをクリックすると、インテルTBBのいくつかのバージョンが表示されます。

　ここでは「tbb22_20090809oss」を選択し、クリックするものとして話を進めます。画面下部の「tbb22_20090809oss_win.zip」と書かれた右横にある［Download］リンクをクリックすると、ダウンロードが始まります。ダウンロード後は、好きな場所に解凍して下さい。

　続いてVisual Studio（以下、VS）でパスの設定を行います。［ツール］-［オプション］-［プロジェクトおよびソリューション］-［VC++ディレクトリィ］を選択すると、「ディレクトリィを表示するプロジェクト」という項目があるので、そこでインクルードファイルを選択してから、先ほどダウンロードしたインテルTBBのパスを追加します。例えば、Cドライブに解凍したと仮定するとパスは「C:\tbb22_20090809oss\include」になります。このパスを追加したら、同様にして「ライブラリ」と「ソースコード」のパスを追加して下さい。

　以上で設定は完了です。これでインテルTBBを使う準備が整いました。

parallel_forについて

　インテルTBBで一番理解しやすく使いやすいのはparallel_forテンプレートです。まずはサンプルを掲載しますので、使い方を見て下さい。後から丁寧にparallel_forテンプレートについて説明を行います。

ParallelForSample

#include <time.h>   
#include <limits.h>
#include <iostream>
#include <vector>
#include <windows.h>
#include <winnt.h>
#include <tchar.h>
#include "tbb/parallel_for.h"
#include "tbb/blocked_range.h"
#include "tbb/task_scheduler_init.h"
#include "tbb/tick_count.h"

using namespace std;
using namespace tbb;

/*----------売上明細----------*/
struct SalesDetails {
private:
    int count; //数量
    double consumption_tax; //消費税額
    double amount; //合計金額
    double total; //売上＋税金
public:

    //コンストラクタ
    SalesDetails() 
        : count( 0 ), consumption_tax( 0 ), amount( 0 ), total( 0 ) {};
    
    //セッター
    void set_count( int count ) { this->count = count; };
    void set_consumption_tax( double tax ) { this->consumption_tax = tax; };
    void set_amount( double amount ) { this->amount = amount; };
    void set_total( double total ) { this->total = total; };
    
    //ゲッター
    int get_count() const { return this->count; };
    double get_consumption_tax() const { return this->consumption_tax; };
    double get_amount() const { return this->amount; };
    double get_total() const { return this->total; };
};


/*----------売上----------*/
struct Sales {
private:
    string name; //名前
    int price; //単価
    SalesDetails*  datas; //売上明細データ
    int count; //明細データの数
public:
    //コンストラクタ
    Sales( ) : name(), price( 0 ), datas(), count( 0 ) {};
    Sales( const string name, int price ) 
        : name( name ), price( price ), datas(), count( 0 )  {};
    
    //セッター
    void set_price( int price ) { this->price = price; };
    void set_name ( string name ) { this->name = name; };
    void set_details( SalesDetails datas[] ) { this->datas = datas; };
    void set_detailCount( int count ) { this->count = count; };
    
    //ゲッター
    int get_price() const { return this->price; };
    string get_name() const { return this->name; };

    /*
        売上明細の各種計算項目を算出します
    */
    void Calculate( SalesDetails* datas, int count ) 
    {
        for ( int i = 0; i < count; i++ ) {
            datas[ i ].set_amount( this->price * datas[ i ].get_count() );
            datas[ i ].set_consumption_tax( datas[ i ].get_amount() * 0.05 );
            datas[ i ].set_total( datas[ i ].get_amount() + datas[ i ].get_consumption_tax() );
        }
        this->datas = datas;
        this->count = count;
    }

    /*    
        並列処理用オペレーター
        売上明細の各種計算項目を算出します
    */
    void operator() ( const blocked_range<size_t>& range ) const 
    {
        SalesDetails* tmp = this->datas;
        for ( size_t i = range.begin(); i != range.end(); i++ ) {
            tmp[ i ].set_amount( this->price * tmp[ i ].get_count() );
            tmp[ i ].set_consumption_tax( tmp[ i ].get_amount() * 0.05 );
            tmp[ i ].set_total( tmp[ i ].get_amount() + tmp[ i ].get_consumption_tax() );
        }
    };
};


/*----------売上情報を解析するオブジェクト----------*/
class Analyzer {
public:
    /* 
        売上明細で算出するべき項目を【直列で】計算します。
    */
    void Calculate( Sales* target, SalesDetails* datas, int count ) 
    {
        target->Calculate( datas, count );
    };

    /* 
        売上明細で算出するべき項目を【並列で】計算します。
    */
    void ParallelCalculate( Sales* target, SalesDetails* datas, int count ) 
    {
        Sales obj( target->get_name(), target->get_price() );
        obj.set_details( datas );
        obj.set_detailCount( count );
        parallel_for( blocked_range<size_t>( 0, count, 10000 ), obj );
    };
};


int _tmain(void)
{
    //ロケールを設定してコンソールで日本語表示ができるようにする
    _tsetlocale( LC_ALL, _T("") ); 

    //ランダムに数を生成するための準備
    srand( static_cast<unsigned int> ( time( NULL ) ) );

    //売上分析の対象となる商品を設定
    Sales* target = new Sales( "yakitori", 100 );
    string name = target->get_name();
    cout << "これから";
    copy( name.begin(), name.end(), ostream_iterator<char>( cout ) );
    cout << "（" << target->get_price() << "円）の売上を分析します・・・" << endl;

    //売上明細情報を生成
    const int count = 1000000;
    SalesDetails* datas = new SalesDetails[ count ];
    for ( int i = 0; i < count; i++ ) {
        SalesDetails data;
        int number = static_cast<int>( ( rand() % 100 ) + 1 );
        data.set_count( number );
        datas[ i ] = data; 
    }

    //各種変数を初期化
    task_scheduler_init init;
    Analyzer* analyzer = new Analyzer();

    //直列で解析
    tick_count start = tick_count::now();
    analyzer->Calculate( target, datas, count );
    tick_count end = tick_count::now();
    double second = ( end - start ).seconds();
    cout << "直列で計算した場合" <<  second << "秒かかりました。" << endl;

    //後にエラーチェックするために直列処理の計算結果を退避&初期化
    SalesDetails* olds = new SalesDetails[ count ];
    for ( int i = 0; i < count; i++ ) {
        olds[ i ] = datas[ i ];
        datas[ i ].set_amount( 0 );
        datas[ i ].set_consumption_tax( 0 );
        datas[ i ].set_total( 0 );
    }

    //並列で解析
    start = tick_count::now();
    analyzer->ParallelCalculate( target, datas, count );
    end = tick_count::now();
    double paralleSecond = ( end - start ).seconds();
    cout << "並列で計算した場合" << paralleSecond << "秒かかりました。" << endl;

    //エラーチェック
    int error = 0;
    for ( int i = 0; i < count; i++ ) {
        if ( datas[ i ].get_count() != olds[ i ].get_count() ) error = 1;
        if ( datas[ i ].get_amount() != olds[ i ].get_amount() ) error = 1;
        if ( datas[ i ].get_consumption_tax() != olds[ i ].get_consumption_tax() ) error = 1;
        if ( datas[ i ].get_total() != olds[ i ].get_total() ) error = 1;
        if ( error == 1 ) {
            cerr << "直列処理と並列処理の値が一致しません。";
            cout << endl;
            return 1;
        }
    }

    //終了処理
    cout << "処理効率は" << ( second / paralleSecond ) << "倍です。" << endl;
    cout << endl << endl << endl;
    delete analyzer;
    delete target;
    return 0;
}

　サンプルコード一式は記事にも添付しているので、適宜ダウンロードしてParallelForSampleプロジェクトを確認してみて下さい。

　このサンプルの内容は、トランザクションデータである売上明細内の未計算項目を算出する単純なものです。主な処理の流れは次の通りです。

1. 売上分析の対象となる商品（Sales：売上オブジェクト）を設定します。
2. 売上明細情報（SalesDetails：売上明細オブジェクト）を生成します。
3. インテルTBBを使用するためにtask_scheduler_initオブジェクトを初期化します。
4. 直列（並列化しない）で各明細の売上金額（数量＊単価）と税金額（売上金額＊0.05）およびその合計を算出して設定します。またその処理時間を計測します。
5. 後でエラーチェックをするために直列で算出したデータを退避して、その後計算結果を初期化します。
6. 今度は並列化して各明細の売上金額（数量＊単価）と税金額（売上金額＊0.05）およびその合計を算出して設定します。またその処理時間を計測します。
7. 直列処理の結果と並列処理の結果を比較してエラーチェックをします。
8. 処理効率を表示して終了します。

　次項以降で並列処理についての解説を少しずつ行います。

インテルTBBの決まりごと

　インテルTBBを使用するプログラムには、いくつかの決まりごとがありますので、それを最初に解説します。まず1つ目に、必ずインテルTBB使用前にtask_scheduler_init initオブジェクトを初期化せねばなりません。このオブジェクトを初期化しないと、インテルTBBは正常に動作しません。

　次にtask_scheduler_init initオブジェクトが必要ですので、それに伴いインテルTBBを使用するプログラムは#include "tbb/task_scheduler_init.h"が必要となります。

　最後に標準テンプレート・ライブラリー（STL）のコンテナはインテルTBBと一緒に使用しないで下さい。STLのコンテナは並列化を意識したものではないので、インテルTBBと一緒に使うと壊れてしまいます。vectorなどのSTLコンテナを使用せずに、今後紹介するインテルTBB用のコンテナを使用して下さい。

ParallelForSampleの並列処理について

　このサンプルでは、処理を並列化するためにインテルTBBのparallel_forテンプレートを使用しています。parallel_forテンプレートはループを並列化するためのもので、並列化する処理が互いに独立している場合に使用します。ループ内の処理が互いに干渉しあう場合に使用すると正しい結果を得られませんので注意して下さい。

　parallel_forを正常に動作させるためには3つの要件を満たさねばなりません。1つ目は、コピー・コンストラクターを用意する必要があるということです。2つ目にデストラクタがなければなりません。3つ目に、operator()「関数呼び出し演算子」を多重定義しなくてはなりません。コピー・コンストラクターとデストラクターは、コンパイラが自動で用意してくれますが、これは自分で定義しなくてはなりません。

　関数呼び出し演算子にあまり馴染みがない方は、サンプルFunctionCallのコードを読んで実行して見て下さい。そうすれば、関数呼び出し演算子の定義方法と大まかな動作が分かると思います。

FunctionCall：関数呼び出し演算子のサンプル

#include <iostream>
#include <tchar.h>

using namespace std;


class FunctionCall {
    int value;
public:
    //関数呼び出し演算子
    void operator()( int param ) {
        cout << "関数呼び出し演算子が呼ばれました。" << 
            "値は" << param << "です。" << endl;
    };

    //constを指定した関数呼び出し演算子
    void operator() () const {
        //下記プログラムのコメントを外すとエラー
        //value = 0;
    };
};

int _tmain(int argc, _TCHAR* argv[])
{
    FunctionCall obj;
    obj(1);
    cout << endl << endl;
    return 0;
}

　インテルTBBでは関数呼び出し演算子の多重定義を多用します。慣れていない方は、関数呼び出し演算子の多重定義方法を身につけておくことをお勧めします。

　本題に戻ります。今回のParallelForSampleでは関数呼び出し演算子を多重定義して、直列のものとほぼ同じ計算プログラムをコーディングしているだけです。比較して見ると定義方法以外はプログラムがほぼ同じです。

直列計算プログラムと並列計算プログラムをSalesオブジェクトから抜粋

//直列処理
void Calculate( SalesDetails* datas, int count ) 
{
    for ( int i = 0; i < count; i++ ) {
        datas[ i ].set_amount( this->price * datas[ i ].get_count() );
        datas[ i ].set_consumption_tax( datas[ i ].get_amount() * 0.05 );
        datas[ i ].set_total( datas[ i ].get_amount() + datas[ i ].get_consumption_tax() );
    }
    this->datas = datas;
    this->count = count;
}

//並列処理
void operator() ( const blocked_range<size_t>& range ) const 
{
    SalesDetails* tmp = this->datas;
    for ( size_t i = range.begin(); i != range.end(); i++ ) {
        tmp[ i ].set_amount( this->price * tmp[ i ].get_count() );
        tmp[ i ].set_consumption_tax( tmp[ i ].get_amount() * 0.05 );
        tmp[ i ].set_total( tmp[ i ].get_amount() + tmp[ i ].get_consumption_tax() );
    }
};

　operator()にconstが指定されている点に注意して下さい。parallel_forは内部で、処理範囲を分割し、その分割した範囲内に指定されたオブジェクトのコピーを作成して、operator()を呼び出します。この時にもし変更を許してしまうと、複数のコピーを矛盾なく並列処理させることが困難になります。そういった理由がありますので、対象となるオブジェクトのデータメンバーを変更することができません。そこで変更できないようにconstを必ず指定せねばなりません。この点に注意して下さい。

　もしconstを指定すると、どのような動きをするのか確認したい場合は、FunctionCallプロジェクト内のコメントを取り除いてみて下さい。そうすればコンパイラエラーが発生します。つまり、parallel_forは誤りを事前に防いでくれるわけです。

　今回のサンプルのoperator()内でSalesDetails* tmp = this->datas;のプログラムがあるのは、constが定義がされているからです。constが定義されている時に直接メンバーを変更することはできませんが、間接的に変更することはできます。それでこのプログラムで間接的に変更しているのです。間接的に変更するだけならば、複数のコピーの値を同じにすることができるので安全です。

　これで呼ばれるオブジェクトの準備は終わりです。後はparallel_forでそのオブジェクトを呼ぶだけです。呼び出し方は簡単で、parallel_for( blocked_range( 0, count, 1000 ), obj );のようにparallel_forテンプレートで、blocked_rangeオブジェクトを初期化したものと、並列処理をするオブジェクト（今回はSales）のインスタンスを指定するだけです。

parallel_forテンプレート呼び出し部分をAnalyzerオブジェクトより抜粋

/* 
    売上明細で算出するべき項目を【並列で】計算します。
*/
void ParallelCalculate( Sales* target, SalesDetails* datas, int count ) 
{
    Sales obj( target->get_name(), target->get_price() );
    obj.set_details( datas );
    obj.set_detailCount( count );
    parallel_for( blocked_range<size_t>( 0, count, 10000 ), obj ); //ここに注目
};

　blocked_rangeオブジェクトは、範囲を表すオブジェクトです。開始地点・繰り返し回数・粒度を指定します。粒度を簡単にいうと、処理を分割する基準となる値です。この値を変更することにより並列処理のパフォーマンスが変化します。粒度については高度な概念なので今回は10000にするとよいと考えて下さい。

　これでParallelForSampleの説明は終わりです。早速実行してみてください。そうすると、直列処理よりも並列処理の方がかなり効率が良いことを体験できます。

　ただし、一つ注意するべき点があります。それはスピードは一定ではなくある程度の幅がある点です。実務でインテルTBBを使用する際には、何度か実行して処理効率を確認して下さい。

　次項では少し複雑な並列化処理を採り上げます。

全体的な並列ループ処理

　先ほど提示したParallelReduceSampleは、限られた範囲内で計算をするものでした。データの全体を加味しないとならないないような処理、総和の計算・最小値の検出・最大値の検出などをするには、parallel_reduceテンプレートを使用します。

　ParallelReduceSampleプロジェクトでparallel_reduceを使用しているので見て下さい。

ParallelReduceSample

#include <time.h>   
#include <limits.h>
#include <iostream>
#include<iomanip>
#include <vector>
#include <windows.h>
#include <winnt.h>
#include <tchar.h>
#include "tbb/parallel_reduce.h"
#include "tbb/blocked_range.h"
#include "tbb/task_scheduler_init.h"
#include "tbb/tick_count.h"

using namespace std;
using namespace tbb;

/*----------売上明細----------*/
struct SalesDetails {
private:
    int count; //数量
    double consumption_tax; //消費税額
    double amount; //合計金額
    double total; //売上＋税金
public:
    //コンストラクタ
    SalesDetails() 
        : count( 0 ), consumption_tax( 0 ), amount( 0 ), total( 0 ) {};

    //セッター
    void set_count( int count ) { this->count = count; };
    void set_consumption_tax( double tax ) { this->consumption_tax = tax; };
    void set_amount( double amount ) { this->amount = amount; };
    void set_total( double total ) { this->total = total; };

    //ゲッター
    int get_count() const { return this->count; };
    double get_consumption_tax() const { return this->consumption_tax; };
    double get_amount() const { return this->amount; };
    double get_total() const { return this->total; };
};


/*----------売上----------*/
struct Sales {
private:
    string name; //名前
    int price; //単価
    SalesDetails*  datas; //売上明細データ
    int count; //明細データの数
    double sumAmount; //売上合計
    double sumTax; //税金合計
    int sumCount; //売上数合計
    int maxCount; //最大売上数
    int maxIndex; //最大売上数をもつ明細のインデックス
    int minCount; //最小売上数
    int minIndex; //最小売上数をもつ明細のインデックス
    double aveCount; //平均売上数
    double aveAmount; //平均売上金額
public:
    //コンストラクタ
    Sales( ) 
        : name(), price( 0 ), 
        datas(), count( 0 ), 
        sumAmount( 0 ), sumTax( 0 ), sumCount( 0 ),
        maxCount( INT_MIN ), maxIndex( -1 ), minCount( INT_MAX ), minIndex( -1 ),
        aveCount( 0 ), aveAmount( 0 ) {};
    Sales( const string name, int price ) 
        : name( name ), price( price ), 
        datas(), count( 0 ), 
        sumAmount( 0 ), sumTax( 0 ), sumCount( 0 ),
        maxCount( INT_MIN ), maxIndex( -1 ), minCount( INT_MAX ), minIndex( -1 ),
        aveCount( 0 ), aveAmount( 0 ) {};
    
    //セッター
    void set_price( int price ) { this->price = price; };
    void set_name ( string name ) { this->name = name; };
    void set_details( SalesDetails datas[] ) { this->datas = datas; };
    void set_detailCount( int count ) { this->count = count; };
    void set_sum( double value ) { this->sumAmount = value; };
    void set_sumTax( double value ) { this->sumTax = value; };
    void set_countSum( int value ) { this->sumCount = value; };
    void set_maxCount( int count ) { this->maxCount = count; };
    void set_maxIndex( int index ) { this->maxIndex = index; };
    void set_minCount( int count ) { this->minCount = count; };
    void set_minIndex( int index ) { this->minIndex = index; };
    void set_aveCount( double value ) { this->aveCount = value; };
    void set_aveAmount( double value ) { this->aveAmount = value; };
    
    //ゲッター
    int get_price() const { return this->price; };
    string get_name() const { return this->name; };
    SalesDetails* get_details() { return this->datas; };
    int get_detailCount() { return this->count; };
    double get_sum() const { return this->sumAmount; };
    double get_sumTax() const { return this->sumTax; };
    int get_countSum() const { return this->sumCount; };
    int get_maxCount() const { return this->maxCount; };
    int get_maxIndex() const { return this->maxIndex; };
    int get_minCount() const { return this->minCount; };
    int get_minIndex() const { return this->minIndex; };
    double get_aveCount() const { return this->aveCount; };
    double get_aveAmount() const { return this->aveAmount; };

    /*
        売上明細の各種計算項目を算出します
    */
    void Calculate( SalesDetails* datas, int count ) 
    {
        for ( int i = 0; i < count; i++ ) {
            datas[ i ].set_amount( this->price * datas[ i ].get_count() );
            datas[ i ].set_consumption_tax( static_cast<int>( datas[ i ].get_amount() * 0.05 ) );
            datas[ i ].set_total( datas[ i ].get_amount() + datas[ i ].get_consumption_tax() );
        }
        this->datas = datas;
        this->count = count;
    }

    /*
        各種数値の平均値を算出します
    */
    void Average() 
    {
        this->aveAmount = static_cast<float>(this->sumAmount) / this->count;
        this->aveCount = static_cast<float>(this->sumCount) / this->count;
    }

    /*
        分析値を算出します
    */
    void Analysis( SalesDetails* datas, int count ) 
    {
        for ( int i = 0; i < count; i++ ) {
            //合計を算出
            this->sumAmount += datas[ i ].get_amount();
            this->sumTax += datas[ i ].get_consumption_tax();
            this->sumCount += datas[ i ].get_count();
            //最大値もしくは最小値を決定
            if ( this->maxCount < datas[ i ].get_count() ) {
                this->maxCount = datas[ i ].get_count();
                this->maxIndex = i;
            } else if ( this->minCount > datas[ i ].get_count() ) {
                this->minCount = datas[ i ].get_count();
                this->minIndex = i;
            }
        }
        this->datas = datas;
        this->count = count;
        Average();
    }

    /*-----------------------------------------------------------------
        並列的に分析値を算出するための設定です
    ----------------------------------------------------------------*/
    //分割コンストラクター
    Sales( Sales& obj, split ) : 
        name( obj.get_name() ), price( obj.get_price() ), 
        datas( obj.get_details() ), count( obj.get_detailCount() ), 
        sumAmount( 0 ), sumTax( 0 ), sumCount( 0 ),
        maxCount( INT_MIN ), maxIndex( -1 ), minCount( INT_MAX ), minIndex( -1 ), 
        aveCount( 0 ), aveAmount( 0 ){};
    //計算結果を結合するためのメソッド
    void join( const Sales& obj ) { 
        //合計を算出
        this->sumAmount += obj.get_sum(); 
        this->sumTax += obj.get_sumTax();
        this->sumCount += obj.get_countSum();
        //最大値と最小値を決定
        if ( this->maxCount < obj.get_maxCount() ) {
            this->maxCount = obj.get_maxCount();
            this->maxIndex = obj.get_maxIndex();
        }
        if ( this->minCount > obj.get_minCount() ) {
            this->minCount = obj.get_minCount();
            this->minIndex = obj.get_minIndex();
        }
    };
    //範囲内で各種計算を行います
    void operator() ( const blocked_range<size_t>& range ) 
    {
        SalesDetails* tmp = this->datas;
        for ( size_t i = range.begin(); i != range.end(); i++ ) {
            //合計を算出
            this->sumAmount += tmp[ i ].get_amount();
            this->sumTax += tmp[ i ].get_consumption_tax();
            this->sumCount += tmp[ i ].get_count();
            //最大値もしくは最小値を決定
            if ( this->maxCount < tmp[ i ].get_count() ) {
                this->maxCount = tmp[ i ].get_count();
                this->maxIndex = i;
            } else if ( this->minCount > tmp[ i ].get_count() ) {
                this->minCount = tmp[ i ].get_count();
                this->minIndex = i;
            }
        }
    };
};


/*----------売上情報を解析するオブジェクト----------*/
class Analyzer {
public:

    //売上明細で算出するべき項目を直列で計算します。
    void Calculate( Sales* target, SalesDetails* datas, int count ) 
    {
        target->Calculate( datas, count );
    };

    /* 
        【直列で】分析値を算出します
    */
    void Analysis( Sales* target, SalesDetails* datas, int count ) 
    {
        target->Analysis( datas, count );
    }

    /* 
        【並列に】分析値を算出します
    */
    Sales ParallelSum( Sales* target, SalesDetails* datas, int count ) 
    {
        Sales obj( target->get_name(), target->get_price() );
        obj.set_details( datas );
        obj.set_detailCount( count );
        parallel_reduce( blocked_range<size_t>( 0, count, 100000 ), obj );
        obj.Average(); //並列計算では誤差が出るので直列に計算
        return obj;
    };
};

int _tmain(void)
{
    //ロケールを設定してコンソールで日本語表示ができるようにする
    _tsetlocale( LC_ALL, _T("") ); 

    //ランダムに数を生成するための準備
    srand( static_cast<unsigned int> ( time( NULL ) ) );

    //売上分析の対象となる商品を設定
    string name = "yakitori";
    int price = 100;
    Sales* target = new Sales( name, price );
    cout << "これから";
    copy( name.begin(), name.end(), ostream_iterator<char>( cout ) );
    cout << "（" << target->get_price() << "円）の売上を分析します・・・" << endl;

    //売上明細情報を生成
    const int count = 1000000;
    SalesDetails* datas = new SalesDetails[ count ];
    for ( int i = 0; i < count; i++ ) {
        SalesDetails data;
        int number = static_cast<int>( ( rand() % 100 ) + 1 );
        data.set_count( number );
        datas[ i ] = data; 
    }

    //各種変数を初期化
    task_scheduler_init init;
    Analyzer* analyzer = new Analyzer();
    analyzer->Calculate( target, datas, count ); 

    //直列で解析
    tick_count start = tick_count::now();
    analyzer->Analysis( target, datas, count );
    tick_count end = tick_count::now();
    double second = ( end - start ).seconds();
    cout << "直列で計算した場合" <<  second << "秒かかりました。" << endl;

    //後にエラーチェックするために直列処理の計算結果を退避
    Sales* old = new Sales( name, price );
    old->set_details( datas );
    old->set_countSum( target->get_countSum() );
    old->set_sum( target->get_sum() );
    old->set_sumTax( target->get_sumTax() );
    old->set_maxCount( target->get_maxCount() );
    old->set_maxIndex( target->get_maxIndex() );
    old->set_minCount( target->get_minCount() );
    old->set_minIndex( target->get_minIndex() );
    old->set_aveCount( target->get_aveCount() );
    old->set_aveAmount( target->get_aveAmount() );

    //並列で解析
    start = tick_count::now();
    Sales result = analyzer->ParallelSum( target, datas, count );
    end = tick_count::now();
    double paralleSecond = (end - start ).seconds();
    cout << "並列で計算した場合" << paralleSecond << "秒かかりました。" << endl;
    
    //エラーチェック
    int error = 0;
    if ( result.get_countSum() != old->get_countSum() ) error = 1;
    if ( result.get_sum() != old->get_sum() ) error = 1;
    if ( result.get_sumTax() != old->get_sumTax() ) error = 1;
    if ( result.get_maxCount() != old->get_maxCount() ) error = 1;
    if ( result.get_maxIndex() != old->get_maxIndex() ) error = 1;
    if ( result.get_minCount() != old->get_minCount() ) error = 1;
    if ( result.get_minIndex() != old->get_minIndex() ) error = 1;
    if ( result.get_aveCount() != old->get_aveCount() ) error = 1;
    if ( result.get_aveAmount() != old->get_aveAmount() ) error = 1;
    if ( error == 1 ) {
        cerr << "直列処理と並列処理の値が一致しません。";
        cout << endl;
        return 1;
    }

    //終了処理
    cout << setprecision(12);
    cout << "処理効率は" << ( second / paralleSecond ) << "倍です。" << endl;
    cout << "総売上数：" << result.get_countSum() << endl;
    cout << "売上合計：" << result.get_sum() << endl;
    cout << "税金合計：" << result.get_sumTax() << endl;
    cout << "最大売上数：" << result.get_maxCount() << endl;
    cout << "最大売上数をもつ明細：" << result.get_maxIndex() << endl;
    cout << "最小売上数：" << result.get_minCount() << endl;
    cout << "最大売上数をもつ明細：" << result.get_minIndex() << endl;
    cout << "明細データの数：" << result.get_detailCount() << "件" << endl;
    cout << "平均売上数：" << result.get_aveCount() << endl;
    cout << "平均売上金額：" << result.get_aveAmount() <<endl;
    cout << endl << endl << endl;
    delete analyzer;
    delete target;
    return 0;
}

　主な処理の流れは次の通りです。

1. 売上分析の対象となる商品（Sales：売上オブジェクト）を設定します。
2. 売上明細情報（SalesDetails：売上明細オブジェクト）を生成します。
3. インテルTBBを使用するためにtask_scheduler_initオブジェクトを初期化します。
4. 直列（並列化しない）で売上明細データを分析します。またその処理時間を計測します。
5. 後でエラーチェックをするために直列で算出したデータを退避します。
6. 今度は並列化して売上明細データを分析します。またその処理時間を計測します。
7. 直列処理の結果と並列処理の結果を比較してエラーチェックをします。
8. 処理効率と分析結果を表示して終了します。

　次項で並列処理部分について詳しく解説します。

ParallelReduceSampleの並列処理について

　ParallelReduceSampleはparallel_reduceテンプレートを使用しています。parallel_reduceテンプレートを使用するために、並列処理を行うオブジェクトは4つの条件を満たさねばなりません。

　1つ目は、専用のコピー・コンストラクター（インテルTBBでは「分割コンストラクター」と呼びます）を実装することです。普通のコピー・コンストラクタと区別するためにtbb::splitクラスを指定しなくてはなりません。

　2つ目は、デストラクターを用意することです。

　3つ目は、operator()を多重定義することです。

　4つ目は、joinというメソッドを定義しなくてはなりません。詳細は後でサンプルの解説と共に行います。

　ParallelReduceSampleの並列処理部分は下記の通りです。

ParallelReduceSampleの並列処理（Salesオブジェクトより抜粋）

/*-----------------------------------------------------------------
    並列的に分析値を算出するための設定です
----------------------------------------------------------------*/
//分割コンストラクター
Sales( Sales& obj, split ) : 
    name( obj.get_name() ), price( obj.get_price() ), 
    datas( obj.get_details() ), count( obj.get_detailCount() ), 
    sumAmount( 0 ), sumTax( 0 ), sumCount( 0 ),
    maxCount( INT_MIN ), maxIndex( -1 ), minCount( INT_MAX ), minIndex( -1 ), 
    aveCount( 0 ), aveAmount( 0 ){};
//計算結果を結合するためのメソッド
void join( const Sales& obj ) { 
    //合計を算出
    this->sumAmount += obj.get_sum(); 
    this->sumTax += obj.get_sumTax();
    this->sumCount += obj.get_countSum();
    //最大値と最小値を決定
    if ( this->maxCount < obj.get_maxCount() ) {
        this->maxCount = obj.get_maxCount();
        this->maxIndex = obj.get_maxIndex();
    }
    if ( this->minCount > obj.get_minCount() ) {
        this->minCount = obj.get_minCount();
        this->minIndex = obj.get_minIndex();
    }
};
//範囲内で各種計算を行います
void operator() ( const blocked_range<size_t>& range ) 
{
    SalesDetails* tmp = this->datas;
    for ( size_t i = range.begin(); i != range.end(); i++ ) {
        //合計を算出
        this->sumAmount += tmp[ i ].get_amount();
        this->sumTax += tmp[ i ].get_consumption_tax();
        this->sumCount += tmp[ i ].get_count();
        //最大値もしくは最小値を決定
        if ( this->maxCount < tmp[ i ].get_count() ) {
            this->maxCount = tmp[ i ].get_count();
            this->maxIndex = i;
        } else if ( this->minCount > tmp[ i ].get_count() ) {
            this->minCount = tmp[ i ].get_count();
            this->minIndex = i;
        }
    }
};

　分割コンストラクターでは、初期化リスト内で渡されたオブジェクトの任意の値をコピーし、分析によって導出する値（最大値など）については初期化しています。このコンストラクターは、parallel_reduceテンプレートがオブジェクトを分割する時に呼び出します。この時点ではまだ分析値（総売上、最大金額など）は導出されていませんので初期値に設定しています。

　operator()の多重定義では、売上金額・税金額・商品数の各項目を加算し、それと同時に最小値と最大値を割り出しています。この時点では、分割されたオブジェクト内での計算課程ですので、全体としての合計値や最大／最小値が割り出されているわけではないことと、今回はconstが指定されていないことに注意して下さい。このテンプレートでは変更が予想されますのでconstを定義してはなりません。ここは間違えやすい点ですので注意して下さい。

　全体としての各種値を出すのは、joinメソッドの仕事です。parallel_reduceテンプレートは最初に、対象となるオブジェクト（このサンプルではSales）を分割していきオブジェクトのコピーを作ります。分割が終わったらその範囲内で処理を行います。そして最後に、各コピーしたオブジェクトのjoinメソッドを呼び出して、2つのオブジェクトを結合していき、最終的に1つのオブジェクトになるまでそれを繰り返します。joinメソッドの引数が自分と同じオブジェクト（このサンプルではSales）なのはこれが理由です。

　ParallelReduceSampleを実行して下さい。並列処理の処理効率が良いことが確認できます。

　しかし、parallel_reduceテンプレートを使う上で注意するべき点があります。それは、演算の丸め誤差です。並列処理では処理の実行順序がバラバラで、割り算の計算結果が毎回違います。

　例えば、100/4/3/2を計算するとします。直列では毎回(((100/4)/3)/2)の順序で計算されます。しかし、並列では((100/4)/(3/2))の様に計算されます。従って、丸め誤差により計算結果が同じではなくなります。並列プログラミングを初めて行った際によく起こる過ちですので十分に注意して下さい。

　また、parallel_reduceテンプレートは結果を結合する処理がありますので、どうしてもparallel_forテンプレートよりも低くなります。ですから極力parallel_forテンプレートを利用するとよいでしょう。

並列ループの考え方

　今回2つのテンプレートを通じて、ループ処理を並列化する方法について解説しました。並列化に慣れていない人はあまりピンと来ないと思いますが、考え方そのものは非常に単純です。ループの処理は分解して考えると、次のように処理をしていることになります。

ループ処理の分解

//処理1
tmp[ 0 ].set_amount( this->price * tmp[ 0 ].get_count() );
tmp[ 0 ].set_consumption_tax( tmp[ 0 ].get_amount() * 0.05 );
tmp[ 0 ].set_total( tmp[ 0 ].get_amount() + tmp[ 0 ].get_consumption_tax() );

//処理2
tmp[ 1 ].set_amount( this->price * tmp[ 1 ].get_count() );
tmp[ 1 ].set_consumption_tax( tmp[ 1 ].get_amount() * 0.05 );
tmp[ 1 ].set_total( tmp[ 1 ].get_amount() + tmp[ 1 ].get_consumption_tax() );

//以下省略

　この処理をよく見ると、各々の処理が他の処理に依存せずに、限られた範囲内で処理をしています。ということは、マルチコアプロセッサならば、処理1と処理2が同時に行えると考えるのは自然です。これがparallel_forテンプレートの考え方です。

　次に、ParallelReduceSampleの様に要素全体で処理をする場合、タスクを分割してその範囲内で最大値などを求め、後でその結果を突き合わせて判断をするようにすれば、マルチコアプロセッサで同時に複数のタスクが処理できることになります。これが、parallel_reduceテンプレートの考え方です。

　この2つの考え方を改めて考えると、普段我々人間がしている仕事と同じです。私たちは多くの場合、1つのプロジェクトの作業を分割して複数の人間で作業しますし（parallel_forテンプレートと同じ）、各プログラマーが複数のオブジェクトをプログラミングしてからそれを統合して1つのシステムを実装します（parallel_reduceテンプレートと同じ）

　こうしてよく考えてみれば、並列プログラミングもそれ程特殊な考え方でないことが分かって頂けると思います。

まとめ

　今回は、インテルTBBのparallel_forテンプレートとparallel_reduceテンプレートの使い方を解説しました。また、その根底に流れる「ループ処理を並列化する」という考え方も解説しました。今まで並列プログラミングに対して苦手意識をもっている方は多いと思います。

　しかし、今回の記事を読めば、並列プログラミングの考え方がそれ程特殊なものではなく、普段我々人間がしていることと大差がないことが分かって頂けると思います。

　並列プログラミングは今後当たり前のものとなります。少しでもその準備のお手伝いができれば幸いです。次回は、より高度な並列プログラミングの概念を噛み砕いて解説します。お楽しみに。

参考資料

書籍

• 『インテル スレッディング・ビルディング・ブロック』 James Reinderss著、菅原清文・エクセルソフト 訳、オライリー・ジャパン、2008年2月26日

Webサイト

• インテルTBB 公式ホームページ
• CodeZine 『インテル Parallel Studioを使って並列化プログラミングを試してみた』
• CodeZine 『並列プログラミングの効率的なデバッグを実現する「Parallel Inspector」』