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

データフローによるプログラミング

　データフローの構成要素はnode（ノード:節）とedge（エッジ:辺）、複数のnodeをedgeで繋ぎ合わせてデータフローを組み上げます。

　nodeは入力ポートと出力ポートを持ち、入力ポートにはFIFOバッファ（キュー）が仕込まれています。nodeは入力ポートにたまったデータを取り出し、何らかの処理を施して出力ポートに送り出します。前段nodeの出力ポートと後段nodeの入力ポートはedgeで接続されます。nodeが機能ブロックで、edgeが機能ブロック間をつなぐパイプというわけ。

　TBBには目的と用途に応じてさまざまなnodeが用意されています。今回のサンプルで使うnodeを紹介しておきましょう。

function_node (1-in/1-out)

　関数f(x)を実現するnodeで、関数オブジェクトをコンストラクタに与えます。例えば

auto square = [](int x) { return x*x; };

をfunction_nodeに与えておくと、入力ポートから取り出したint値を二乗して出力ポートに送ります。

　引数を複数個与える関数はtupleでまとめます。

auto add = [](const tuple<int,int>& in) { return get<0>(in) + get<1>(in); }

のように。

broadcast_node (1-in,n-out)

　入力ポートに入ってきたデータをそのまま複数の出力ポートにバラ撒きます。（単方向の）USB-hubみたいなものですね。

join_node (n-in/1-out)

　複数の入力ポートすべてにデータが揃うのを待ち、各ポートから得られたデータを1つのtupleにまとめて出力ポートに送ります。

　以上3種のnodeを使って f(x) = x^2 + x^3 を計算するデータフローを構成したのがコレ。

　左端のbroadcast_nodeに投げ込まれたデータxは2つのfunction_nodeに分配され、それぞれx^2,x^3を求めてjoin_nodeへ、join_nodeは双方のデータが揃ったところで(x^2,x^3)を後段へ、そこで両者を加えてresultに積算するというダンドリです。

　コードをお見せしましょう。各nodeを生成し、make_edge(前段node, 後段node) でnodeを繋ぎ、先頭nodeに1,2,……,10を投げ込んで処理が完了する（全nodeが処理を終える）のを待ちます。

list05 TBB.cpp

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

using namespace std;
using namespace tbb::flow;

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

item calc(int lo, int hi) {
  item result = 0;

  // 1. graph（nodeを貼る台紙）
  graph g;

  // 2. graph内にnodeを置く
  //  - broadcast_node<inout-type>
  //  - function_node<in-type,out-type>
  //  - join_node<out-type>

  broadcast_node<item>
    input(   g);

  function_node<item,item>
    squarer( g, unlimited, 
             [](item v) {
               trace("squarer(" ITEM_FORMAT ")\n", value(v));
               return v * v;
             }
           );

  function_node<item,item>
    cuber(   g, unlimited, 
             [](item v) { 
               trace("cuber(" ITEM_FORMAT ")\n", value(v)); 
               return v * v * v;
             }
         );

  join_node<tuple<item,item>> 
    join(    g );

  function_node<tuple<item,item>,item>  
    summer(  g, serial,
             [&result](const tuple<item,item>& v) {
               trace("summer(" ITEM_FORMAT "," ITEM_FORMAT ")\n",
                     value(get<0>(v)), value(get<1>(v))); 
               return result += get<0>(v) + get<1>(v);
             } 
          );

  // 3. node間をedgeで繋ぐ
  make_edge( input,   squarer                    );
  make_edge( input,   cuber                      );
  make_edge( squarer, get<0>(join.input_ports()) );
  make_edge( cuber,   get<1>(join.input_ports()) );
  make_edge( join,    summer                     );

  // 最上流nodeにデータを流し込む
  for (int i = lo; i <= hi; ++i) {
    input.try_put(i);
  }

  // graph内の全nodeが処理を終えるのを待つ
  g.wait_for_all();

  return result;
}

int main() {
  // 生成するスレッド数を設定
  tbb::task_scheduler_init tbb_init(2); 

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

　function_nodeコンストラクタの第二引数について補足しておかねばなりません。TBBの舞台裏でタスク・スケジューラが動いています。タスク・スケジューラは仕事をしたくてウズウズしているタスク（処理単位）をスレッドに割り当てるのがお仕事です。どのタスクを／どの順番で／どのスレッドに割り当てるかをスケジュールしているのですね。

　function_nodeにおけるタスクはコンストラクト時に与える関数オブジェクト（上記コードではラムダ式）です。コンストラクタの第二引数はタスク・スケジューラに投入するタスク数の上限を表しています。unlimited（=0）とあるのは、タスク投入数に上限のないことを意味し、入力ポートにデータがあるならお構いなしにタスクを投入します。投入されたタスクの実行順序や、どのスレッドで実行するかはタスク・スケジューラまかせ。

　x^2とx^3のたし算を行い、結果をresultに積み上げるsummerでは第二引数が serial（=1） となっています。タスク・スケジューラに投入するタスク数を1、つまり投入されている現タスクの処理が終わらぬ限り次のタスクを投入しません。これにはワケがありまして、複数のタスクを投入するとそれらが同時に動き出すかもしれません。summerのタスクはその処理内に結果の積み上げ: result += … が含まれますから、同時に動くとデータ競合（data race）を起こしかねませんからね。

　そんなわけでこのコードはsquarerとcuberがそれぞれのタスクを目一杯投入し、タスク・スケジューラはスレッドの使用率を100%に近づけるべくせっせとタスクを割り当てます。

　スレッド数を2つにしたときの実行結果がコレ。

　スレッド1つで7000msかかるなら、スレッド2つがフル稼働して3500msの限界値に対して約3600ms、上等です。ちなみにスレッド4つだと約1900msでした。いい値が出てますねぇ。