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

lambda-friendlyなパイプライン

パイプラインの使い方（1）class pipeline/filter

　ではTBB3.0を使ったパイプラインのお試しコードです。まずはTBB2.xからサポートしているclass pipelineとfilterを使った実装から。

　filterはパイプラインの各工程を表現し、ユーザーは各工程に応じたクラスをclass filterから導出します。void* operator()(void*) を定義、つまり前工程から引き渡されたvoid*を引数（材料）として何らかの処理を行い、その結果をvoid*でreturnすることで次工程に引き渡します。各工程（filterの導出クラス）ができたら、それらのインスタンスを用意し、pipelineのメンバ関数add_filterで順に登録し、run()すればパイプラインが動き始めます。run()に与える引数は同時実行される最大工程（filter）数。これを大きくしておけば複数のオバちゃんが同時に仕事をしてくれるという次第。とはいえ大きくすればするほど速くなるわけもなく、CPU論理コア数で頭打ちになるでしょうけど。

　pipeline/filter版「おサシミ盛り合わせ製造ライン2.2」の実装を以下に示します。

osashimi2.cpp

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

using namespace std;
tbb::mutex mutex;

typedef vector<string>::iterator iterator_t;

// 1st-stage : おサシミを引く
class osashimi : public tbb::filter {
  iterator_t first_;
  iterator_t last_;
public:
  osashimi(iterator_t f, iterator_t l) 
    : tbb::filter(tbb::filter::serial_in_order), first_(f), last_(l) {}
  void* operator()(void*) { // 初段なので引数は無視
    if ( first_ == last_ ) { // おサカナがなくなったら終了
      return 0;
    }
    string* result = new string(*first_ + "のおサシミ");
    ++first_;
    return result;
  }
};

// 2nd-stage : 皿に盛る
class moritsuke : public tbb::filter {
public:
  moritsuke() : tbb::filter(tbb::filter::serial_in_order) {}
  void* operator()(void* p) {
    string* sp = static_cast<string*>(p);
    sp->append("を皿に盛り");
    return sp;
  }
};

// 3rd-stage : ツマを添える
class tsuma : public tbb::filter {
public:
  tsuma() : tbb::filter(tbb::filter::serial_in_order) {}
  void* operator()(void* p) {
    string* sp = static_cast<string*>(p);
    sp->append("ツマを添えて");
    return sp;
  }
};

// 4th-stage : タンポポを乗せる
class tampopo : public tbb::filter {
public:
  tampopo() : tbb::filter(tbb::filter::serial_in_order) {}
  void* operator()(void* p) {
    string* sp = static_cast<string*>(p);
    sp->append("タンポポ乗せる");
    return sp;
  }
};

// last-stage : ショーケースに並べる
class showcase : public tbb::filter {
public:
  showcase() : tbb::filter(tbb::filter::serial_in_order) {}
  void* operator()(void* p) {
    string* sp = static_cast<string*>(p);
    cout << *sp << endl;
    delete sp;
    return 0;
  }
};

int main() {

  // 入力系列とパイプラインを生成
  const char* table[] = { "マグロ", "ハマチ", "サワラ", "ホタテ" };
  vector<string> fish(table, table+4);
  tbb::pipeline pipeline;

  // 各stageをパイプラインに接続
  osashimi  first(fish.begin(), fish.end());
  moritsuke second;
  tsuma     third;
  tampopo   fourth;
  showcase  last;
  
  pipeline.add_filter(first);
  pipeline.add_filter(second);
  pipeline.add_filter(third);
  pipeline.add_filter(fourth);
  pipeline.add_filter(last);
  
  pipeline.run(2);
  pipeline.clear();
}