計算グラフを使用した並列プログラミング

メッセージングシステムとして適切に実装されたアプリケーションがあります。 広い意味では、メッセージは何でも構いません-「信号」などを制御するデータブロック ロジックは、メッセージを処理するノードと、それらの間の接続で構成されます。 そのような構造は、ノードで処理されるメッセージのエッジを「フロー」するグラフによって自然に表されます。 このようなモデルの最も有名な名前は、計算グラフです。



計算グラフを使用して、タスク間の依存関係を確立し、ある程度まで「データフローアーキテクチャ」をプログラムで実装できます。



この投稿では、 インテルスレッディングビルディングブロック （インテルTBB）ライブラリ、つまりtbb :: flow :: graphクラスを使用して、C ++でこのようなモデルを実装する方法について説明します。

インテルTBBとtbbとは::フロー::グラフクラス

Intel Threading Building Blocks-並列プログラミング用のC ++テンプレートライブラリ。 オープンソースの実装では無料で配布されますが、商用版もあります。 Windows *、Linux *、およびOS X *のバイナリ形式で利用可能。



TBBには、並列コンピューティングでの使用に合わせて調整された多くの既製のアルゴリズム、構成、およびデータ構造があります。 含めて、計算グラフの実装を可能にする構造があり、これについて説明します。



ご存知のように、グラフは頂点（ノード）とエッジで構成されています。 計算グラフtbb :: flow ::グラフも、ノード、エッジ、およびグラフ全体のオブジェクトで構成されます。







グラフのノードには、送信者と受信者のインターフェースがあり、メッセージを管理したり、いくつかの機能を実行したりします。 エッジはグラフのノードを接続し、メッセージパッシングの「チャネル」です。



各ノードの本体はTBBタスクで表され、それらの間に依存関係がない場合、他のノードと並行して実行できます。 TBBでは、多くの並列アルゴリズム（またはすべて）がタスク（ワークフローによって実行される小さな作業項目（命令））に基づいて構築されます。 タスク間に依存関係がある可能性があり、それらはスレッド間で動的に再分配できます。 タスクを使用することにより、CPUで最適な粒度とロードバランスを実現し、それらに基づいて高レベルの並列構造（tbb :: flow :: graphなど）を構築できます。

最も単純な依存関係グラフ

1つのエッジで接続された2つの頂点で構成されるグラフ。そのうちの1つは「Hello」、2つ目の「World」は、次のように概略的に表示できます。







そして、コードでは次のようになります。

#include <iostream> #include <tbb/flow_graph.h> int main(int argc, char *argv[]) { tbb::flow::graph g; tbb::flow::continue_node< tbb::flow::continue_msg > h( g, []( const tbb::flow::continue_msg & ) { std::cout << "Hello "; } ); tbb::flow::continue_node< tbb::flow::continue_msg > w( g, []( const tbb::flow::continue_msg & ) { std::cout << "World\n"; } ); tbb::flow::make_edge( h, w ); h.try_put(tbb::flow::continue_msg()); g.wait_for_all(); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これにより、グラフgのオブジェクトとタイプcontinue_nodeの2つのノード-hおよびwが作成されます。 これらのノードは、continue_msgタイプのメッセージ（内部制御メッセージ）を送受信します。 それらは、先行からメッセージを受信した後にのみノード本体が実行される場合に、依存関係グラフを構築するために使用されます。



continue_nodeのそれぞれは、条件付きで有用なコードを実行します-「Hello」と「World」を出力します。 ノードは、make_edgeメソッドを使用してエッジで結合されます。 すべて、計算グラフの構造の準備ができています-try_putメソッドでメッセージを送信することで、実行のために実行できます。 次に、グラフが機能し、すべてのタスクが完了したことを確認するために、wait_for_allメソッドを使用して待機します。

シンプルなメッセージンググラフ

プログラムがxに対して1から10までの式x ² + x ³を計算することを想像してください。はい、これは最も難しい計算タスクではありませんが、デモンストレーションには非常に適しています。



式のカウントをグラフ形式で表示してみましょう。 最初のノードは、着信データストリームからx値を取得し、立方体および二乗されたノードに送信します。 べき乗演算は互いに独立しており、並行して実行できます。 起こりうる不均衡を解消するために、結果をバッファノードに転送します。 次に、統合ノードがあります。これは、集計ノードの累乗の結果を提供し、計算が終了します。







そのようなグラフのコード：

 #include <tbb/flow_graph.h> #include <windows.h> using namespace tbb::flow; struct square { int operator()(int v) { printf("squaring %d\n", v); Sleep(1000); return v*v; } }; struct cube { int operator()(int v) { printf("cubing %d\n", v); Sleep(1000); return v*v*v; } }; class sum { int &my_sum; public: sum( int &s ) : my_sum(s) {} int operator()( std::tuple<int,int> v ) { printf("adding %d and %d to %d\n", std::get<0>(v), std::get<1>(v), my_sum); my_sum += std::get<0>(v) + std::get<1>(v); return my_sum; } }; int main(int argc, char *argv[]) { int result = 0; graph g; broadcast_node<int> input (g); function_node<int,int> squarer( g, unlimited, square() ); function_node<int,int> cuber( g, unlimited, cube() ); buffer_node<int> square_buffer(g); buffer_node<int> cube_buffer(g); join_node< std::tuple<int,int>, queueing > join(g); function_node<std::tuple<int,int>,int> summer( g, serial, sum(result) ); make_edge( input, squarer ); make_edge( input, cuber ); make_edge( squarer, square_buffer ); make_edge( squarer, input_port<0>(join) ); make_edge( cuber, cube_buffer ); make_edge( cuber, input_port<1>(join) ); make_edge( join, summer ); for (int i = 1; i <= 10; ++i) input.try_put(i); g.wait_for_all(); printf("Final result is %d\n", result); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

プロセスを視覚化するためにSleep（1000）関数が追加されました（例はWindowsでコンパイルされ、他のプラットフォームで同等の呼び出しを使用します）。 その後、すべてが最初の例のようになります-ノードを作成し、エッジと組み合わせて実行するために実行します。 function_nodeの2番目のパラメーター（無制限またはシリアル）は、並列実行できるノード本体のインスタンスの数を決定します。 タイプjoin_nodeのノードは、各入力での入力データ/メッセージの準備を決定し、両方の準備ができたら、それらをstd :: tupleの形式で次のノードに渡します。

TBB ::フロー::グラフを使用したディナー哲学者の問題の解決

ウィキペディアから ：

「食事する哲学者の問題」は、コンピューターサイエンスで使用される古典的な例であり、並列アルゴリズムの設計における同期の問題と、これらの問題を解決する手法を示しています。



タスクでは、数人の哲学者がテーブルに座っており、食べることも考えることもできますが、同時にはできません。 私たちのバージョンでは、哲学者は箸で麺を食べます-食べるには2本の棒が必要ですが、それぞれが利用可能です：







このような状況では、たとえば、各哲学者が杖を左につかむとデッドロック（デッドロック）が発生する可能性があるため、ダイナー間のアクションの同期が必要です。



tbb :: flow :: graphの形式で哲学者のテーブルを提示してみましょう。 各哲学者は、スティックをキャプチャするためのjoin_nodeと、「食べる」および「考える」タスクを達成するためのfunction_nodeの2つのノードで表されます。 テーブル上のスティックの配置は、queue_nodeを介して実装されます。 queue_nodeキューには複数のワンドを含めることはできません。ワンドがある場合は、キャプチャに使用できます。 グラフは次のようになります。







いくつかの定数とヘッダーファイルを持つメイン関数：

 #include <windows.h> #include <tbb/flow_graph.h> #include <tbb/task_scheduler_init.h> using namespace tbb::flow; const char *names[] = { "Archimedes", "Aristotle", "Democritus", "Epicurus", "Euclid", "Heraclitus", "Plato", "Pythagoras", "Socrates", "Thales" }; …. int main(int argc, char *argv[]) { int num_threads = 0; int num_philosophers = 10; if ( argc > 1 ) num_threads = atoi(argv[1]); if ( argc > 2 ) num_philosophers = atoi(argv[2]); if ( num_threads < 1 || num_philosophers < 1 || num_philosophers > 10 ) exit(1); tbb::task_scheduler_init init(num_threads); graph g; printf("\n%d philosophers with %d threads\n\n", num_philosophers, num_threads); std::vector< queue_node<chopstick> * > places; for ( int i = 0; i < num_philosophers; ++i ) { queue_node<chopstick> *qn_ptr = new queue_node<chopstick>(g); qn_ptr->try_put(chopstick()); places.push_back( qn_ptr ); } std::vector< philosopher > philosophers; for ( int i = 0; i < num_philosophers; ++i ) { philosophers.push_back( philosopher( names[i], g, places[i], places[(i+1)%num_philosophers] ) ); g.run( philosophers[i] ); } g.wait_for_all(); for ( int i = 0; i < num_philosophers; ++i ) philosophers[i].check(); return 0; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

コマンドラインパラメータを処理した後、タイプtbb :: task_scheduler_initのオブジェクトを作成することにより、ライブラリが初期化されます。 これにより、初期化の時間を制御し、スレッドハンドラーの数を手動で設定できます。 これがないと、初期化は自動的に行われます。 次に、グラフgのオブジェクトが作成されます。 「杖の場所」queue_nodeはstd :: vectorに配置され、各杖は杖に配置されます。



他の哲学者も同様の方法で作成されます-std :: vectorに配置されます。 各哲学者のオブジェクトは、グラフオブジェクトのrun関数に渡されます。 哲学者クラスには演算子（）が含まれ、run関数を使用すると、グラフgオブジェクトのルートタスクの子であるタスクでこのファンクターを実行できます。 したがって、g.wait_for_all（）の呼び出し中にこれらのタスクが完了するまで待つことができます。



哲学者クラス：

 const int think_time = 1000; const int eat_time = 1000; const int num_times = 10; class chopstick {}; class philosopher { public: typedef queue_node< chopstick > chopstick_buffer; typedef join_node< std::tuple<chopstick,chopstick> > join_type; philosopher( const char *name, graph &the_graph, chopstick_buffer *left, chopstick_buffer *right ) : my_name(name), my_graph(&the_graph), my_left_chopstick(left), my_right_chopstick(right), my_join(new join_type(the_graph)), my_function_node(NULL), my_count(new int(num_times)) {} void operator()(); void check(); private: const char *my_name; graph *my_graph; chopstick_buffer *my_left_chopstick; chopstick_buffer *my_right_chopstick; join_type *my_join; function_node< join_type::output_type, continue_msg > *my_function_node; int *my_count; friend class node_body; void eat_and_think( ); void eat( ); void think( ); void make_my_node(); };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

各哲学者には、名前、グラフオブジェクトへのポインタ、左右スティックへのポインタ、join_nodeノード、function_node関数ノード、および哲学者が考えて食べた回数をカウントするmy_countカウンターがあります。



グラフ実行関数によって呼び出されるoperator（）（）は、哲学者が最初に考えてからグラフにアタッチするように実装されます。

 void philosopher::operator()() { think(); make_my_node(); }  think  eat    : void philosopher::think() { printf("%s thinking\n", my_name ); Sleep(think_time); printf("%s done thinking\n", my_name ); } void philosopher::eat() { printf("%s eating\n", my_name ); Sleep(eat_time); printf("%s done eating\n", my_name ); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

make_my_nodeメソッドは関数ノードを作成し、それとjoin_nodeの両方をグラフの残りの部分に関連付けます。

 void philosopher::make_my_node() { my_left_chopstick->register_successor( input_port<0>(*my_join) ); my_right_chopstick->register_successor( input_port<1>(*my_join) ); my_function_node = new function_node< join_type::output_type, continue_msg >( *my_graph, serial, node_body( *this ) ); make_edge( *my_join, *my_function_node ); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

グラフは動的に作成されることに注意してください-エッジはregister_successorメソッドによって形成されます。 最初にグラフ構造を完全に作成してから実行する必要はありません。 TBBには、グラフが既に実行されている場合でも、新しいノードを削除および追加する場合でも、この構造をその場で変更する機能があります。 これにより、計算グラフの概念がさらに柔軟になります。



node_bodyクラスは、哲学者:: eat_and_think（）メソッドを呼び出す単純なファンクターです。

 class node_body { philosopher &my_philosopher; public: node_body( philosopher &p ) : my_philosopher(p) { } void operator()( philosopher::join_type::output_type ) { my_philosopher.eat_and_think(); } };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

eat_and_thinkメソッドはeat（）関数を呼び出し、カウンターをデクリメントします。 その後、哲学者は杖をテーブルに置いて考えます。 そして、彼が食べて正しい回数を考えた場合、彼はテーブルから立ち上がる-remove_successorメソッドを使用して、join_nodeとグラフとの接続を切断します。 ここでも、グラフの動的構造が表示されます。ノードの一部は削除され、残りは機能し続けます。

 void philosopher::eat_and_think( ) { eat(); --(*my_count); if (*my_count > 0) { my_left_chopstick->try_put( chopstick() ); my_right_chopstick->try_put( chopstick() ); think(); } else { my_left_chopstick->remove_successor( input_port<0>(*my_join) ); my_right_chopstick->remove_successor( input_port<1>(*my_join) ); my_left_chopstick->try_put( chopstick() ); my_right_chopstick->try_put( chopstick() ); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このグラフでは、queue_node（スティックの場所）から哲学者、より正確にはjoin_nodeへのエッジがあります。 しかし、反対の方向では、ありません。 ただし、eat_and_thinkメソッドは、杖をキューに戻すためにtry_putを呼び出すことができます。



main（）関数の最後に、すべての哲学者に対してcheckメソッドが呼び出され、哲学者が正しい回数を食べて考え、必要な「クリーニング」を行うことを確認します。

 void philosopher::check() { if ( *my_count != 0 ) { printf("ERROR: philosopher %s still had to run %d more times\n", my_name, *my_count); exit(1); } else { printf("%s done.\n", my_name); } delete my_function_node; delete my_join; delete my_count; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

この例のデッドロックは、join_nodeの使用により発生しません。 このタイプのノードは、両方の入力から受け取ったオブジェクトからstd :: tupleを作成します。 この場合、入力データは受信直後に消費されません。 join_nodeは、最初に両方の入力にデータが表示されるまで待機してから、順番に予約を試みます。 この操作が成功した場合のみ-それらは「消費」され、std :: tupleがそれらから作成されます。 少なくとも1つの入力「チャネル」の予約がうまくいかない場合、すでに予約されているものは解放されます。 つまり 哲学者が1本の杖を捕まえることができるが、2本目が忙しい場合、彼は最初の手を放して待機し、隣人を無駄にブロックしません。



この食事哲学者の例は、TBBグラフのいくつかの機能を示しています。

• join_nodeを使用してリソースアクセスの同期を確保する
• 動的なグラフ構築-操作中にノードを追加および削除できます
• 単一の入口および出口点がないため、グラフにループがある場合があります
• グラフ実行機能を使用する

ノードの種類

tbb ::フロー::グラフは、かなり幅広いノードオプションを提供します。 これらは、機能、バッファリング、結合と分離、その他の4つのグループに分けることができます。 凡例付きのノードタイプのリスト：

おわりに

インテルTBBで実装されたグラフを使用して、「非構造化並列処理」とも呼ばれる並列プログラムの複雑で興味深いロジックを作成できます。 計算グラフを使用すると、タスク間の依存関係を整理し、メッセージとイベントの送信に基づいてアプリケーションを構築できます。



グラフ構造は静的または動的のいずれかです。ノードとエッジはその場で追加および削除できます。 個々のサブグラフを大きなグラフに接続できます。



ほとんどの資料は、海外の同僚の英語の出版物に基づいています。



興味がある人のために、試してみてください：



インテルスレッディングビルディングブロックライブラリ（オープンソースバージョン）をダウンロードします。

http://threadingbuildingblocks.org



Intel TBBの商用バージョン（機能的に違いはありません）：

http://software.intel.com/en-us/intel-tbb



TBBに関する英語のブログ::フロー::グラフ：

http://software.intel.com/en-us/tags/17218

http://software.intel.com/en-us/tags/17455