フローグラフの基本: ノード

ノードは tbb::flow::graph_node から継承するクラスで、通常は tbb::flow::sender<T>、tbb::flow::receiver<T>、あるいは両方から継承します。ノードは通常、着信メッセージで操作を行い、0 以上の出力メッセージを生成します。2 つ以上の入力メッセージを要求したり、2 つ以上の出力メッセージを生成するノードもあります。

graph_node、sender、および receiver から継承して独自のノードタイプを定義することは可能ですが、一般的には事前定義ノードタイプを使用してグラフを構築します。事前定義ノードのリストは、下記の関連情報セクションを参照してください。

function_node は flow_graph.h に含まれている事前定義タイプで、1 つの入力と 1 つの出力の単純な関数を表します。function_node のコンストラクターには、3 つの引数を指定します。

template< typename Body> function_node(graph &g, size_t concurrency, Body body)

次のコードは、1 つの function_node を含む単純なグラフを作成します。この例では、グラフ g に属するノード n を構築しています。第 2 引数は 1 であるため、ノードの同時実行数は 1 です。ボディーは、受け取った値 v を出力し、v 秒間スピンして、値を再度出力した後、v を変更しないで返すラムダ式です。spin_for 関数のコードは提供されていません。

    graph g;
    function_node< int, int > n( g, 1, []( int v ) -> int { 
        cout << v;
        spin_for( v );
        cout << v;
        return v;
    } );

上記の例でノードを構築した後、エッジを介してほかのノードに接続するか、try_put 関数を呼び出して、ノードにメッセージを渡すことができます。エッジの使用方法は、次のセクションで説明します。

    n.try_put( 1 );
    n.try_put( 2 );
    n.try_put( 3 );

次に、グラフ・オブジェクトで wait_for_all を呼び出し、メッセージが処理されるのを待ちます。

    g.wait_for_all(); 

上記のコード例で、function_node n は同時実行数を 1 に制限して作成されています。メッセージシーケンス 1、2、3 を受け取ると、ノード n は最初に入力 1 のボディーに適用するタスクをスポーンします。そのタスクが完了すると、ノードは入力 2 のボディーに適用する別のタスクをスポーンします。同様に、ノードは、2 番目のタスクが完了するのを待機してから、入力 3 のボディーに適用する 3 番目のタスクをスポーンします。タスクがスポーンされるまで、try_put の呼び出しはブロックされません。ノードがメッセージを処理するタスクを直ちにスポーンできない場合、メッセージはノードにバッファーされます。有効な場合、同時実行数に基づいて、次にバッファーされたメッセージを処理するタスクがスポーンされます。

上記のグラフでは、各メッセージが順番に処理されます。しかし、異なる同時実行数でノードを構築して、並列処理を行うこともできます。

    function_node< int, int > n( g, tbb::flow::unlimited, []( int v ) -> int { 
        cout << v;
        spin_for( v );
        cout << v;
        return v;
    } );

同時実行数として unlimited を使用すると、ほかのタスクがスポーンした数に関係なく、メッセージを受け取るとすぐにタスクをスポーンするようにライブラリーに指示できます。同時実行数に 4 や 8 などの特定の値を使用して、同時に実行できる数を指定することもできます。タスクのスポーンはスレッドの作成を意味しないことに注意してください。そのため、グラフが多くのタスクをスポーンする場合でも、ライブラリーのスレッドプールで利用可能なスレッド数のみ、これらのタスクを実行するのに使用されます。

function_node コンストラクターで unlimited を使用し、ノードで try_put を呼び出す例を考えてみます。

    n.try_put( 1 );
    n.try_put( 2 );
    n.try_put( 3 );
    g.wait_for_all(); 

ライブラリーは 3 つのタスクをスポーンし、各タスクは n のラムダ式をメッセージの 1 つに適用します。システムで十分な数のスレッドを利用可能な場合、3 つのボディーの呼び出しは並列に行われます。しかし、システムで 1 つのスレッドしか利用できない場合、順番に実行されます。