インテルTBB：Flow Graphによるデータフロープログラミング

もっと速いデータフロー

　さらにもっと速くできそうです。元々の計算式:

result += x*x + x*x:x;

を少しいじって:

item x2 = x*x;
result += x2 + x2*x;

　こうすればかけ算を1つ減らせます。シングルスレッド／10回で7000msだったのが5000msになりますね。

　この式に基づいてデータフローを構成します。ついでにsquarer,cuber,summerを汎用的なかけ算,たし算nodeに置き換えて:

　かなりややこしくなりましたが、実装してみました。

list07 TBB_tag_matching_complex.cpp

#include <atomic>
#include <cassert>
#include <tbb/tbb.h>
#include "utils.h"

using namespace std;
using namespace tbb::flow;

// 扱うデータの型とformat
typedef slow_item<int> item;
#define ITEM_FORMAT "%d"

// tag付けされた型 tagged<T>、および tagged<T> から タグ/値 を取り出す関数
template<typename T> using tagged = tuple<size_t,T>;
template<typename T> inline size_t tag(const tagged<T>& t) { return get<0>(t); }
template<typename T> inline T      val(const tagged<T>& t) { return get<1>(t); }

item calc(int lo, int hi) {

  item           result = 0;

  graph g;

  // x -> (t,x)
  atomic<size_t> tagval = 0U;
  auto set_tag = [&tagval](item v) {
    return tagged<item>(++tagval, v);
  };

  function_node<item,tagged<item>>
    node_0(g, unlimited, set_tag);

  broadcast_node<tagged<item>>
    node_1(g);

  // (t,x) -> ((t,x),(t,x))
  function_node<tagged<item>, tuple<tagged<item>,tagged<item>>>
    node_2(g, unlimited,
           [](const tagged<item>& t) {
             return tuple<tagged<item>,tagged<item>>(t,t);
           }
          );

  // ((t,x),(t,y)) -> (t,x*y)
  auto multiply = [](const tuple<tagged<item>,tagged<item>>& t) {
                    assert( tag(get<0>(t)) == tag(get<1>(t)) );
                    item v0 = val(get<0>(t));
                    item v1 = val(get<1>(t));
                    trace("multiply(" ITEM_FORMAT "," ITEM_FORMAT ")\n", value(v0), value(v1));
                    return tagged<item>(tag(get<0>(t)), v0 * v1); 
                  }; 

  function_node<tuple<tagged<item>,tagged<item>>,tagged<item>>
    node_3(g, unlimited, multiply);

  broadcast_node<tagged<item>>
    node_4(g);

  join_node<tuple<tagged<item>,tagged<item>>,tag_matching> 
    node_5(g, tag<item>, tag<item>);

  function_node<tuple<tagged<item>,tagged<item>>,tagged<item>>
    node_6(g, unlimited, multiply); 

  join_node<tuple<tagged<item>,tagged<item>>,tag_matching> 
    node_7(g, tag<item>, tag<item>);

  // ((t,x),(t,y)) -> (t,x+y)
  auto add = [](const tuple<tagged<item>, tagged<item>>& t) {
               assert( tag(get<0>(t)) == tag(get<1>(t)) );
               item v0 = val(get<0>(t));
               item v1 = val(get<1>(t));
               trace("add(" ITEM_FORMAT "," ITEM_FORMAT ")\n", value(v0), value(v1));
               return tagged<item>(tag(get<0>(t)), v0 + v1); 
             }; 

  function_node<tuple<tagged<item>,tagged<item>>,tagged<item>>  
    node_8(g, unlimited, add);

  // (t,x) -> Σx
  function_node<tagged<item>,item>  
    node_9(g, serial,
           [&result](const tagged<item>& t) {
             trace("sigma(" ITEM_FORMAT ")\n", value(val(t)));
             return result += val(t);
           } 
          );

  make_edge( node_0, node_1);
  make_edge( node_1, node_2);
  make_edge( node_2, node_3);
  make_edge( node_3, node_4);
  make_edge( node_1, get<0>(node_5.input_ports()) );
  make_edge( node_4, get<1>(node_5.input_ports()) );
  make_edge( node_5, node_6);
  make_edge( node_4, get<0>(node_7.input_ports()) );
  make_edge( node_6, get<1>(node_7.input_ports()) );
  make_edge( node_7, node_8);
  make_edge( node_8, node_9);

  for (int i = lo; i <= hi; ++i) {
      node_0.try_put(i);
  }
  g.wait_for_all();

  return result;
}

int main() {
  tbb::task_scheduler_init tbb_init(2); 

  item result;
  measure([&]() { result = calc(1,10);});
  cout << value(result) << endl;
}

　インテルTBBは数年前から僕の道具箱に入っていて、並列アルゴリズムや同期プリミティブなどはお気に入りの手駒として使っていたのですが、Flow Graphは今回初めて触りました。機能ブロックをパイプで繋いで生産ラインを組み上げるようなプログラミング・モデルに小さな驚きを感じています。