async_msg テンプレート・クラス

概要

async_msg は、インテル® スレッディング・ビルディング・ブロック (インテル® TBB) のフローグラフと (スレッドやデバイスのような) 外部並列処理の間の相互通信を実装するために使用される特別なメッセージクラスです。

構文

        template < typename T >
        class async_msg;
      

ヘッダー

        #define TBB_PREVIEW_FLOW_GRAPH_FEATURES 1
        #include "tbb/flow_graph.h"
      

説明

async_msg または async_msg から派生したユーザー型は、フローグラフのノードにより特別に制御されます。async_msg<T> は、結果の準備ができる前にグラフで処理できる T 型の非同期ジョブ結果の表現です。結果はいつでも async_msg<T>::set( T result ) メソッドを使用した非同期処理により提供することができます。ほかのクラスは、追加データを含む高度な非同期メッセージを作成する async_msg<T> から継承することができます。T 型および非同期メッセージを受け付けるポートとノードは、どちらも非同期メッセージを受け取ることができ、make_edge インターフェイスに接続できますが、制御方法は異なります。ノードが非同期メッセージを受け付けた場合、(例えば、メッセージをボディーに直接渡して) 着信非同期メッセージに直ちに反応します。しかし、ノードが T 結果型を受け付けた場合、async_msg<T>::finalize() 仮想メソッドが呼び出され、ノードは結果が利用可能になるのを待ちます。async_msg<T>::finalize() メソッドは、(例えば、結果が必要であると非同期処理に通知したり、コールバックを設定する) 派生した非同期メッセージ型でオーバーライドすることができます。

async_msg の最も簡単な使用シナリオを下記の図に示します。フローグラフの左のノード (n1) はボディで非同期処理を開始し、結果が設定されるのを待つことなく async_msg オブジェクトを返します。async_msg のコピーを介して非同期処理により結果が設定されると、次のノード (n2) は自動的に結果を受け取ります。

最も簡単なシナリオのステップを次に示します。

• function_node n1 が async_msg<T> のインスタンスを作成します。
• n1 がワークと async_msg のコピーを外部の非同期処理に送ります。
• n1 が async_msg<T> をサクセサーにプットします。
• 外部の非同期処理がステップ 2 で取得した async_msg インスタンスの set(result) メソッドを呼び出し、結果をフローグラフに戻します。
• サクセサーノードが T 型のデータを取得します。

このシナリオでは、async_msg と function_node はともに async_node クラスに非常に似た動作を提供します。

async_node に対する async_msg の主な利点は、次の使用シナリオのように、未変更または更新されたフローグラフを介してこれらのメッセージを渡せることです。

• function_node n1 が async_msg<T> のインスタンスを作成します。
• n1 がワークと async_msg のコピーを外部の非同期処理に送ります。
• n1 が async_msg<T> を次のノード n2 にプットします。
• n2 がワークの 2 番目の部分を外部の非同期処理に送ります。必要な場合は async_msg メッセージのインスタンスを更新します。
• async_msg<T> がグラフの (T 型を受け取る) 最後のノードに移ると、仮想メソッド finalize() がライブラリーにより自動的に呼び出されます。このメソッドはカスタム非同期メッセージでオーバーライドされます。その用途は、結果をフローグラフに返すように非同期処理にコールバックを設定することです。

  注

  async_msg<T>::finalize() はライブラリーにより数回呼び出されますが、set() メソッドが async_msg で呼び出された後は呼び出されません。

• 次のノード n3 が UserDataType データを取得します。

async_msg から派生したユーザークラスは、上記のユースケースで使用できます。例えば、データ処理チェーンの動作中に追加の状態やフラグを格納または更新する派生した非同期メッセージクラスを実装できます。

サンプル

次のサンプルで source_node は、ユーザースレッドによって処理されるように、ワークを AsyncActivity に渡します。async_msg は、ダウンストリームの function_node に結果を伝えるために使用されます。

#define TBB_PREVIEW_FLOW_GRAPH_FEATURES 1

#include "tbb/flow_graph.h"
#include "tbb/concurrent_queue.h"
#include <thread>

using namespace tbb::flow;
typedef int input_type;
typedef int output_type;
typedef tbb::flow::async_msg<output_type> async_msg_type;


class AsyncActivity {
public:
    struct work_type {
        input_type input;
        async_msg_type msg;
    };

    AsyncActivity(tbb::flow::graph& g) : my_graph(g), my_the_end(false), service_thread( [this]() {
        while( ! my_the_end ) {
            work_type w;
            while( my_work_queue.try_pop(w) ) {
                output_type result = do_work( w.input );
                // 結果をグラフに返す
                w.msg.set(result);
                // ワーク完了を通知する (グラフをアンブロック)
                my_graph.decrement_wait_count();
            }
        }
    } ) {}

    ~AsyncActivity() {
        my_the_end = true; // スレッドを終了すべきであることを示す
        service_thread.join();
    }

    void submit( input_type i, const async_msg_type& msg ) {
        work_type w = {i, msg};
        my_graph.increment_wait_count();
        my_work_queue.push(w);
    }

private:
    output_type do_work(input_type& v) {
        // 入力を処理して出力に変換する
    }

    tbb::flow::graph& my_graph;
    tbb::concurrent_queue<work_type> my_work_queue;
    bool my_the_end;

    std::thread service_thread;
};

int main() {
    tbb::flow::graph g;
    AsyncActivity async_activity(g);

   tbb::flow::source_node<async_msg_type> s(g, [&](async_msg_type& v)->bool {
       /* 非同期処理用のデータを生成する */
       if ( /* ソースが終了していない */ ) {
           async_msg_type msg;
           /* ここに "入力" を生成する */
           async_activity.submit(input, msg);
           v = msg;
           return true;
       }
   
       return false;
   });

    tbb::flow::function_node<output_type> f( g, unlimited, [](const output_type& v) 
    { /* 非同期処理からデータを取得する */ });

    tbb::flow::make_edge( s, f );
    g.wait_for_all();
    return 0;
}

メンバー

  
namespace tbb {
    namespace flow {

        template <typename T>
        class async_msg {
        public:
            typedef T async_msg_data_type;

            async_msg();
            virtual ~async_msg() {}

            void set(const T& t);
            void set(T&& t);

        protected:
            // 派生クラスでオーバーライドして非同期計算チェーンが
            // 終了したことを通知する
            virtual void finalize() const {}
        };

    }
}