インテルTBB 3.0によるパイプライン処理

lambda-friendlyなパイプライン

パイプラインの使い方（2）parallel_pipeline

　TBB 3.0で新たに追加された関数parallel_pipelineはパイプラインの各工程をlambda式で書き下すことを可能にしました。それにより、各工程ごとにfilterの導出クラスを定義せずに済むのでコードが非常にコンパクトになります。

　pipeline/filter版をparallel_pipeline+lamda式で書き換えた「おサシミ盛り合わせ製造ライン3.0」を以下に示します。

osashimi3.cpp

#include <iostream>
#include <vector>
#include <string>
#include <tbb/tbb.h>

using namespace std;

int main() {
  tbb::mutex mtx;

  const char* table[] = { "マグロ", "ハマチ", "サワラ", "ホタテ" };
  vector<string> fish(table, table+4);
  vector<string>::iterator first = fish.begin();
  vector<string>::iterator last  = fish.end();

  tbb::parallel_pipeline(1,
    // ---- 1st filter
    tbb::make_filter<void,string>(
      tbb::filter::serial_in_order, 
      [&first,last](tbb::flow_control& fc) -> string {
        if ( first == last ) {
          fc.stop();
          return ""; // dummy value
        }
        return *first++ + "のおサシミ";
      })
  &
    // ---- 2nd filter
    tbb::make_filter<string,string>(
      tbb::filter::serial_in_order, 
      [](const string& s) -> string {
        return s + "を皿に盛り";
      })
  &
    // ---- 3rd filter
    tbb::make_filter<string,string>(
      tbb::filter::serial_in_order, 
      [](const string& s) -> string {
        return s + "ツマを添えて";
      })
  &
    // ---- 4th filter
    tbb::make_filter<string,string>(
      tbb::filter::serial_in_order,
      [](const string& s) -> string {
        return s + "タンポポ乗せる\n";
      })
  &
    // ---- last filter
    tbb::make_filter<string,void>(
      tbb::filter::serial_in_order,
      [](const string& s) -> void {
        cout << s;
      })
  );

}

　実行するといずれの製造ラインも

マグロのおサシミを皿に盛りツマを添えてタンポポ乗せる
ハマチのおサシミを皿に盛りツマを添えてタンポポ乗せる
サワラのおサシミを皿に盛りツマを添えてタンポポ乗せる
ホタテのおサシミを皿に盛りツマを添えてタンポポ乗せる

　が出力されます…これじゃホントにパイプラインによる並列処理が行われたか分からないので面白くありませんね。そのためもう一つ、

• 整数1,2,...nを生成し
• その逆数を求めて
• 出力する

　の3つの工程それぞれに処理過程をプリントさせ、パイプライン実行します。

reciprocal.cpp

#include <iostream>
#include <utility>
#include <tbb/tbb.h>

using namespace std;
typedef tbb::mutex::scoped_lock lock_t;

int main() {
  tbb::mutex mtx;
  int count = 1;
  int limit = 5;
  tbb::parallel_pipeline(2, // ココを書き換えて遊んでみよう!

    // ---- 1st filter
    tbb::make_filter<void,int>(
      tbb::filter::serial_in_order, 
      [&](tbb::flow_control& fc) -> int {
        if ( count == limit ) {
          fc.stop();
          return -1; // dummy value
        }
lock_t lock(mtx);
cout << string(count,'\t') << "1st(" << count << ")\n";
        return count++;
      })
  &
    // ---- 2nd filter
    tbb::make_filter<int,pair<int,double>>(
      tbb::filter::serial_in_order, 
      [&](int n) -> pair<int,double> {
lock_t lock(mtx);
cout << string(n,'\t') << "2nd(" << n << ")\n";
        return pair<int,double>(n, 1.0/n);
      })
  &
    // ---- 3rd filter
    tbb::make_filter<pair<int,double>,void>(
      tbb::filter::serial_in_order,
      [&](pair<int,double> p) -> void {
lock_t lock(mtx);
cout << string(p.first,'\t') << "3rd(" << p.first << ") : " 
     << p.second << endl;
      })
  );

}

　関数parallel_pipelineの第一引数が同時実行される工程数です。これを1とした場合の実行結果は、

1st(1)
2nd(1)
3rd(1) : 1
        1st(2)
        2nd(2)
        3rd(2) : 0.5
                1st(3)
                2nd(3)
                3rd(3) : 0.333333
                        1st(4)
                        2nd(4)
                        3rd(4) : 0.25

　御覧の通り、オバちゃん（?）が一人でせっせと仕事をしているのが分かります。

　対して同時実行数を2にすると…

1st(1)
2nd(1)
        1st(2)
3rd(1) : 1
        2nd(2)
                1st(3)
        3rd(2) : 0.5
                2nd(3)
                        1st(4)
                3rd(3) : 0.333333
                        2nd(4)
                        3rd(4) : 0.25

　このように、工程の途中で別のラインの工程が割り込んでいます。僕の実行環境はdual-coreなのでこれ以上大きな値を指定しても大差ないのですが、quad-coreやhexa-core、HiperThreadingによる論理octa-coreな環境をお持ちの方なら振舞いの変化を観察できることでしょう。