非同期メッセージブロック

[アーティクル]
08/09/2011

更新 : 2010 年 8 月

エージェントライブラリには、アプリケーションコンポーネント間でメッセージをスレッドセーフに伝達できるメッセージブロックの型がいくつか用意されています。これらのメッセージブロックの型は、通常、Concurrency::send、Concurrency::asend、Concurrency::receive、Concurrency::try_receive などの各種メッセージパッシングルーチンと共に使用します。エージェントライブラリで定義されているメッセージパッシングルーチンの詳細については、「メッセージパッシング関数」を参照してください。

セクション

このトピックは、次のセクションで構成されています。

ソースとターゲット
メッセージの伝達
メッセージブロックの型の概要
unbounded_buffer クラス
overwrite_buffer クラス
single_assignment クラス
call クラス
transformer クラス
choice クラス
join クラスと multitype_join クラス
timer クラス
メッセージのフィルター処理
メッセージの予約

ソースとターゲット

ソースとターゲットは、メッセージパッシングの 2 つの重要な参加要素です。ソースとは、メッセージを送信する通信エンドポイントです。ターゲットとは、メッセージを受信する通信エンドポイントです。ソースは読み取るエンドポイントであり、ターゲットは書き込むエンドポイントであると考えることができます。アプリケーションでは、ソースとターゲットを接続することでメッセージングネットワークを形成します。

エージェントライブラリでは、Concurrency::ISource と Concurrency::ITarget という 2 つの抽象クラスを使用して、ソースとターゲットを表します。ソースとして機能するメッセージブロックの型は ISource から派生します。ターゲットとして機能するメッセージブロックの型は ITarget から派生します。ソースおよびターゲットとして機能するメッセージブロックの型は、それぞれ ISource および ITarget から派生します。

[ページのトップへ]

メッセージの伝達

メッセージの伝達とは、コンポーネント間でのメッセージの送信処理のことです。メッセージブロックにメッセージが提供されると、そのメッセージブロックはメッセージを受け入れるか、拒否するか、または延期します。各メッセージブロックの型は、さまざまな方法でメッセージを格納および送信します。たとえば、unbounded_buffer クラスはメッセージを無制限に格納し、overwrite_buffer クラスは一度に 1 つのメッセージを格納し、transformer クラスは各メッセージの変更後のバージョンを格納します。これらのメッセージブロックの型については、このドキュメントの後半で詳しく説明します。

メッセージブロックは、メッセージを受け入れるとき、必要に応じて処理を実行できます。メッセージブロックがソースである場合は、結果のメッセージをネットワークの別のメンバーに渡します。メッセージブロックは、フィルター関数を使用して、不要なメッセージを拒否できます。フィルターについては、このトピックの後半の「メッセージのフィルター処理」で詳しく説明します。メッセージを延期するメッセージブロックは、そのメッセージを予約しておいて後で処理できます。メッセージの予約については、このトピックの後半の「メッセージの予約」で詳しく説明します。

エージェントライブラリを使用すると、メッセージブロックによってメッセージを非同期的に渡すことも、同期的に渡すこともできます。 send 関数などを使用してメッセージを同期的にメッセージブロックに渡すと、ターゲットブロックがそのメッセージを受け入れるか拒否するまで、ランタイムは現在のコンテキストをブロックします。 asend 関数などを使用してメッセージを非同期的にメッセージブロックに渡すと、ランタイムはそのメッセージをターゲットに提供します。ターゲットがそのメッセージを受け入れると、ランタイムはメッセージを受信側に伝達する非同期タスクをスケジュールします。ランタイムは、軽量タスクを使用して、メッセージを協調的に伝達します。軽量タスクの詳細については、「タスクスケジューラ (同時実行ランタイム)」を参照してください。

アプリケーションでは、ソースとターゲットを接続することでメッセージングネットワークを形成します。通常、ネットワークをリンクし、send または asend を呼び出してデータをネットワークに渡します。ソースメッセージブロックをターゲットに接続するには、Concurrency::ISource::link_target メソッドを呼び出します。ソースブロックをターゲットから接続解除するには、Concurrency::ISource::unlink_target メソッドを呼び出します。ソースブロックをすべてのターゲットから接続解除するには、Concurrency::ISource::unlink_targets メソッドを呼び出します。定義済みのメッセージブロックの型のいずれかがスコープから外れるか、破棄されると、自動的にすべてのターゲットブロックから接続解除されます。メッセージブロックの型の中には、書き込み先として使用できるターゲットの最大数が制限されているものもあります。次のセクションでは、定義済みのメッセージブロックの型に適用される制限事項について説明します。

[ページのトップへ]

メッセージブロックの型の概要

次の表では、重要なメッセージブロックの型の役割について簡単に説明しています。

unbounded_buffer
メッセージのキューを格納します。
overwrite_buffer
複数回の書き込みと読み取りを行うことができるメッセージを 1 つ格納します。
single_assignment
1 回の書き込みと複数回の読み取りを行うことができるメッセージを 1 つ格納します。
call
メッセージを受信するときに処理を実行します。
transformer
データを受け取り、その処理の結果を別のターゲットブロックに送信するときに処理を実行します。 transformer クラスでは、異なる入力と出力の種類を操作できます。
choice
一連のソースから最初の使用可能なメッセージを選択します。
join と multitype join
一連のソースからすべてのメッセージを受信するまで待機し、別のメッセージブロックのために、複数のメッセージを結合して 1 つのメッセージにします。
timer
メッセージを定期的にターゲットブロックに送信します。

これらのメッセージブロックの型には、それぞれ異なる状況で役立つ特性があります。その特性のいくつかを次に示します。

伝達の種類: メッセージブロックがデータのソースとして機能するか、データの受信側として機能するか、またはその両方として機能するか。
メッセージの順序付け: メッセージブロックで、メッセージが送受信されたときの元の順序を保持するかどうか。定義済みのメッセージブロックの型では、メッセージが送受信されたときの元の順序が保持されます。
ソースカウント: メッセージブロックで読み取ることができるソースの最大数。
ターゲットカウント: メッセージブロックで書き込むことができるターゲットの最大数。

これらの特性とさまざまなメッセージブロックの型の関係を次の表に示します。

メッセージブロックの型	伝達の種類 (ソース、ターゲット、または両方)	メッセージの順序付け (順序ありまたは順序なし)	ソースカウント	ターゲットカウント
unbounded_buffer	両方	順序あり	制限なし	制限なし
overwrite_buffer	両方	順序あり	制限なし	制限なし
single_assignment	両方	順序あり	制限なし	制限なし
call	ターゲット	順序あり	制限なし	該当せず
transformer	両方	順序あり	制限なし	1
choice	両方	順序あり	10	1
join	両方	順序あり	制限なし	1
multitype_join	両方	順序あり	10	1
timer	ソース	該当せず	該当せず	1

以下のセクションでは、メッセージブロックの型について詳しく説明します。

[ページのトップへ]

unbounded_buffer クラス

Concurrency::unbounded_buffer クラスは、汎用的な非同期メッセージング構造体を表します。このクラスでは、複数のソースが書き込むことができるメッセージ、または複数のターゲットが読み取ることができるメッセージの先入れ先出し (FIFO) のキューを格納します。ターゲットが unbounded_buffer オブジェクトからメッセージを受信すると、そのメッセージはメッセージキューから削除されます。そのため、unbounded_buffer オブジェクトには複数のターゲットを設定できますが、各メッセージを受信するターゲットは 1 つだけです。 unbounded_buffer クラスは、複数のメッセージを別のコンポーネントに渡し、そのコンポーネントで各メッセージを受信する必要がある場合に便利です。

例

次の例では、unbounded_buffer クラスの使用方法の基本的な構造を示します。この例では、3 つの値を unbounded_buffer オブジェクトに送信し、同じオブジェクトからそれらの値を再び読み取ります。

// unbounded_buffer-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

int wmain()
{
   // Create an unbounded_buffer object that works with
   // int data.
   unbounded_buffer<int> items;

   // Send a few items to the unbounded_buffer object.
   send(items, 33);
   send(items, 44);
   send(items, 55);

   // Read the items from the unbounded_buffer object and print
   // them to the console.
   wcout << receive(items) << endl;
   wcout << receive(items) << endl;
   wcout << receive(items) << endl;
}

この例を実行すると、次の出力が生成されます。

33
44
55

unbounded_buffer クラスの使用方法を示す完全な例については、「方法: さまざまなプロデューサー/コンシューマーパターンを実装する」を参照してください。

[ページのトップへ]

overwrite_buffer クラス

Concurrency::overwrite_buffer クラスは unbounded_buffer クラスに似ていますが、overwrite_buffer オブジェクトに格納されるメッセージが 1 つだけである点が異なります。また、ターゲットが overwrite_buffer オブジェクトからメッセージを受信しても、そのメッセージはバッファーから削除されません。そのため、複数のターゲットがメッセージのコピーを受信します。

overwrite_buffer クラスは、複数のメッセージを別のコンポーネントに渡すときに、そのコンポーネントで必要になるのが最新の値のみである場合に便利です。また、このクラスは、メッセージを複数のコンポーネントにブロードキャストする場合にも便利です。

例

次の例では、overwrite_buffer クラスの使用方法の基本的な構造を示します。この例では、3 つの値を overwrite _buffer オブジェクトに送信し、同じオブジェクトから現在の値を 3 回読み取ります。この例は unbounded_buffer クラスの例に似ています。ただし、overwrite_buffer クラスに格納されるメッセージは 1 つだけです。また、メッセージが読み取られた後も、overwrite_buffer オブジェクトからそのメッセージは削除されません。

// overwrite_buffer-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

int wmain()
{
   // Create an overwrite_buffer object that works with
   // int data.
   overwrite_buffer<int> item;

   // Send a few items to the overwrite_buffer object.
   send(item, 33);
   send(item, 44);
   send(item, 55);

   // Read the current item from the overwrite_buffer object and print
   // it to the console three times.
   wcout << receive(item) << endl;
   wcout << receive(item) << endl;
   wcout << receive(item) << endl;
}

この例を実行すると、次の出力が生成されます。

55
55
55

overwrite_buffer クラスの使用方法を示す完全な例については、「方法: さまざまなプロデューサー/コンシューマーパターンを実装する」を参照してください。

[ページのトップへ]

single_assignment クラス

Concurrency::single_assignment クラスは overwrite_buffer クラスに似ていますが、single_assignment オブジェクトに 1 回しか書き込むことができない点が異なります。 overwrite_buffer クラスと同様に、ターゲットが single_assignment オブジェクトからメッセージを受信しても、そのメッセージはそのオブジェクトから削除されません。そのため、複数のターゲットがメッセージのコピーを受信します。また、single_assignment クラスは、1 つのメッセージを複数のコンポーネントにブロードキャストする場合にも便利です。

例

次の例では、single_assignment クラスの使用方法の基本的な構造を示します。この例では、3 つの値を single_assignment オブジェクトに送信し、同じオブジェクトから現在の値を 3 回読み取ります。この例は overwrite_buffer クラスの例に似ています。 overwrite_buffer クラスおよび single_assignment クラスに格納されるメッセージはどちらの場合も 1 つだけですが、single_assignment クラスには 1 回しか書き込むことができません。

// single_assignment-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

int wmain()
{
   // Create an single_assignment object that works with
   // int data.
   single_assignment<int> item;

   // Send a few items to the single_assignment object.
   send(item, 33);
   send(item, 44);
   send(item, 55);

   // Read the current item from the single_assignment object and print
   // it to the console three times.
   wcout << receive(item) << endl;
   wcout << receive(item) << endl;
   wcout << receive(item) << endl;
}

この例を実行すると、次の出力が生成されます。

33
33
33

single_assignment クラスの使用方法を示す完全な例については、「チュートリアル: フューチャの実装」を参照してください。

[ページのトップへ]

call クラス

Concurrency::call クラスは、データを受信するときに処理関数を実行するメッセージの受信側として機能します。この処理関数には、ラムダ式、関数オブジェクト、または関数ポインターを使用できます。 call オブジェクトは、メッセージを送信する他のコンポーネントに対して並列に動作するため、通常の関数呼び出しとは動作が異なります。 call オブジェクトがメッセージを受信したときに処理を実行していると、そのメッセージはキューに追加されます。各 call オブジェクトでは、キューに配置されたメッセージを受信した順序で処理します。

例

次の例では、call クラスの使用方法の基本的な構造を示します。この例では、受信した各値をコンソールに出力する call オブジェクトを作成します。その後、3 つの値を call オブジェクトに送信します。また、call オブジェクトはメッセージを別個のスレッドで処理するため、この例ではカウンター変数と event オブジェクトを使用して、wmain 関数が制御を戻す前に call オブジェクトがすべてのメッセージを処理するようにしています。

// call-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

int wmain()
{
   // An event that is set when the call object receives all values.
   event received_all;

   // Counts the 
   long receive_count = 0L;
   long max_receive_count = 3L;

   // Create an call object that works with int data.
   call<int> target([&received_all,&receive_count,max_receive_count](int n) {
      // Print the value that the call object receives to the console.
      wcout << n << endl;

      // Set the event when all messages have been processed.
      if (++receive_count == max_receive_count)
         received_all.set();
   });

   // Send a few items to the call object.
   send(target, 33);
   send(target, 44);
   send(target, 55);

   // Wait for the call object to process all items.
   received_all.wait();
}

この例を実行すると、次の出力が生成されます。

33
44
55

call クラスの使用方法を示す完全な例については、「方法: call クラスおよび transformer クラスに処理関数を提供する」を参照してください。

[ページのトップへ]

transformer クラス

Concurrency::transformer クラスは、メッセージの受信側および送信側の両方として機能します。 transformer クラスは、データを受信するとユーザー定義の処理関数を実行するため、call クラスに似ています。ただし、transformer クラスも処理関数の結果を受信側のオブジェクトに送信します。 call オブジェクトと同様に、transformer オブジェクトはメッセージを送信する他のコンポーネントに対して並列に動作します。 transformer オブジェクトがメッセージを受信したときに処理を実行していると、そのメッセージはキューに追加されます。各 transformer オブジェクトでは、キューに配置されたメッセージを受信した順序で処理します。

transformer クラスでは、メッセージを 1 つのターゲットに送信します。コンストラクターの _PTarget パラメーターを NULL に設定した場合は、Concurrency::link_target メソッドを呼び出すことで、後からターゲットを指定できます。

エージェントライブラリに用意されている他のすべての非同期メッセージブロックの型とは異なり、transformer クラスでは異なる入力と出力の種類を操作できます。データをある型から別の型に変換するこの機能により、transformer クラスは多くの同時実行ネットワークで重要なコンポーネントとなっています。また、transformer オブジェクトの処理関数により詳細な並列機能を追加できます。

例

次の例では、transformer クラスの使用方法の基本的な構造を示します。この例では、double 値の出力を生成するために、int の各入力値に 0.33 を乗算する transformer オブジェクトを作成します。その後、変換後の値を同じ transformer オブジェクトから受け取り、それらをコンソールに出力します。

// transformer-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

int wmain()
{
   // Create an transformer object that receives int data and 
   // sends double data.
   transformer<int, double> third([](int n) {
      // Return one-third of the input value.
      return n * 0.33;
   });

   // Send a few items to the transformer object.
   send(third, 33);
   send(third, 44);
   send(third, 55);

   // Read the processed items from the transformer object and print
   // them to the console.
   wcout << receive(third) << endl;
   wcout << receive(third) << endl;
   wcout << receive(third) << endl;
}

この例を実行すると、次の出力が生成されます。

10.89
14.52
18.15

transformer クラスの使用方法を示す完全な例については、「方法: データパイプラインでトランスフォーマーを使用する」を参照してください。

[ページのトップへ]

choice クラス

Concurrency::choice クラスでは、一連のソースから最初の使用可能なメッセージを選択します。 choice クラスは、データフロー機構ではなく、制御フロー機構を表します (データフローと制御フローの違いについては、「非同期エージェントライブラリ」を参照してください)。

choice オブジェクトからの読み取りは、bWaitAll パラメーターが FALSE に設定されている場合の Windows API 関数 WaitForMultipleObjects の呼び出しに似ています。ただし、choice クラスでは、外部同期オブジェクトではなく、イベント自体にデータをバインドします。

通常、アプリケーションで制御フローを操作するためには、Concurrency::receive 関数と共に choice クラスを使用します。さまざまな型のメッセージバッファーの中から選択する必要がある場合は、choice クラスを使用します。同じ型のメッセージバッファーの中から選択する必要がある場合は、single_assignment クラスを使用します。

ソースを choice オブジェクトにリンクさせる順序によって選択するメッセージを決定できるため、その順序は重要です。たとえば、既にメッセージが格納されている複数のメッセージバッファーを choice オブジェクトにリンクさせるとします。 choice オブジェクトによって、リンク先の最初のソースからメッセージが選択されます。すべてのソースをリンクさせると、choice オブジェクトでは各ソースがメッセージを受信した順序を保持します。

例

次の例では、choice クラスの使用方法の基本的な構造を示します。この例では、Concurrency::make_choice 関数を使用して、3 つのメッセージブロックのいずれかを選択する choice オブジェクトを作成します。次に、さまざまなフィボナッチ数列を計算し、それぞれの結果を別個のメッセージブロックに格納します。その後、最初に終了した操作に基づくメッセージをコンソールに出力します。

// choice-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <ppl.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

// Computes the nth Fibonacci number.
// This function illustrates a lengthy operation and is therefore
// not optimized for performance.
int fibonacci(int n)
{
   if (n < 2)
      return n;
   return fibonacci(n-1) + fibonacci(n-2);
}

int wmain()
{
   // Although the following thee message blocks are written to one time only, 
   // this example illustrates the fact that the choice class works with 
   // different message block types.

   // Holds the 35th Fibonacci number.
   single_assignment<int> fib35;
   // Holds the 37th Fibonacci number.
   overwrite_buffer<int> fib37;
   // Holds half of the 42nd Fibonacci number.
   unbounded_buffer<double> half_of_fib42;   

   // Create a choice object that selects the first single_assignment 
   // object that receives a value.
   auto select_one = make_choice(&fib35, &fib37, &half_of_fib42);

   // Execute a few lengthy operations in parallel. Each operation sends its 
   // result to one of the single_assignment objects.
   parallel_invoke(
      [&fib35] { send(fib35, fibonacci(35)); },
      [&fib37] { send(fib37, fibonacci(37)); },
      [&half_of_fib42] { send(half_of_fib42, fibonacci(42) * 0.5); }
   );

   // Print a message that is based on the operation that finished first.
   switch (receive(select_one))
   {
   case 0:
      wcout << L"fib35 received its value first. Result = " 
            << receive(fib35) << endl;
      break;
   case 1:
      wcout << L"fib37 received its value first. Result = " 
            << receive(fib37) << endl;
      break;
   case 2:
      wcout << L"half_of_fib42 received its value first. Result = " 
            << receive(half_of_fib42) << endl;
      break;
   default:
      wcout << L"Unexpected." << endl;
      break;
   }
}

この例では、次のサンプル出力が生成されます。

fib35 received its value first. Result = 9227465

35 番目のフィボナッチ数を計算するタスクは最初に終了するとは限らないため、この例の出力は変わることがあります。

この例では、Concurrency::parallel_invoke アルゴリズムを使用して、フィボナッチ数列を並列に計算します。 parallel_invoke の詳細については、「並列アルゴリズム」を参照してください。

choice クラスの使用方法を示す完全な例については、「方法: 完了したタスクから選択する」を参照してください。

[ページのトップへ]

join クラスと multitype_join クラス

Concurrency::join クラスおよび Concurrency::multitype_join クラスを使用すると、一連のソースの各メンバーがメッセージを受信するまで待機できます。 join クラスでは、共通のメッセージ型を持つソースオブジェクトを処理します。 multitype_join クラスでは、異なるメッセージ型を持つことができるソースオブジェクトを処理します。

join オブジェクトまたは multitype_join オブジェクトからの読み取りは、bWaitAll パラメーターが TRUE に設定されている場合の Windows API 関数 WaitForMultipleObjects の呼び出しに似ています。ただし、choice オブジェクトと同様に、join オブジェクトおよび multitype_join オブジェクトでは、外部同期オブジェクトではなく、イベント自体にデータをバインドするイベント機構を使用します。

join オブジェクトから読み取ると、std::vector オブジェクトが生成されます。 multitype_join オブジェクトから読み取ると、std::tuple オブジェクトが生成されます。対応するソースバッファーと同じ順序でこれらのオブジェクトに出現する要素は、join オブジェクトまたは multitype_join オブジェクトにリンクされます。ソースバッファーを join オブジェクトまたは multitype_join オブジェクトにリンクする順序は、結果の vector オブジェクトまたは tuple オブジェクトの要素の順序に関連するため、既存のソースバッファーを結合からリンク解除しないことをお勧めします。そのようにすると、未定義の動作が発生する可能性があります。

最長一致の結合と最短一致の結合

join クラスおよび multitype_join クラスでは、最長一致の結合と最短一致の結合の概念をサポートしています。最長一致の結合では、各ソースからの使用可能になったメッセージの受け入れを、すべてのメッセージが使用可能になるまで続けます。最短一致の結合では、2 つの段階でメッセージを受信します。まず、各ソースからメッセージが提供されるまで待機します。次に、すべてのソースのメッセージが使用可能になったら、各メッセージの予約を試みます。各メッセージを予約できる場合、すべてのメッセージを処理し、ターゲットに伝達します。それ以外の場合、メッセージの予約を解除し (取り消し)、再び各ソースがメッセージを受信するまで待機します。

最長一致の結合はメッセージを即座に受け入れるため、最短一致の結合よりもパフォーマンスが優れています。ただし、最長一致の結合では、デッドロックが発生する可能性がまれにあります。 1 つ以上の共有されたソースオブジェクトを含む複数の結合がある場合は、最短一致の結合を使用します。

例

次の例では、join クラスの使用方法の基本的な構造を示します。この例では、Concurrency::make_join 関数を使用して、3 つの single_assignment オブジェクトから受信する join オブジェクトを作成します。この例では、さまざまなフィボナッチ数列を計算し、それぞれの結果を別個の single_assignment オブジェクトに格納します。その後、join オブジェクトが保持している各結果をコンソールに出力します。この例は choice クラスの例に似ています。ただし、join クラスでは、すべてのソースメッセージブロックがメッセージを受信するまで待機します。

// join-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <ppl.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

// Computes the nth Fibonacci number.
// This function illustrates a lengthy operation and is therefore
// not optimized for performance.
int fibonacci(int n)
{
   if (n < 2)
      return n;
   return fibonacci(n-1) + fibonacci(n-2);
}

int wmain()
{
   // Holds the 35th Fibonacci number.
   single_assignment<int> fib35;
   // Holds the 37th Fibonacci number.
   single_assignment<int> fib37;
   // Holds half of the 42nd Fibonacci number.
   single_assignment<double> half_of_fib42;   

   // Create a join object that selects the values from each of the
   // single_assignment objects.
   auto join_all = make_join(&fib35, &fib37, &half_of_fib42);

   // Execute a few lengthy operations in parallel. Each operation sends its 
   // result to one of the single_assignment objects.
   parallel_invoke(
      [&fib35] { send(fib35, fibonacci(35)); },
      [&fib37] { send(fib37, fibonacci(37)); },
      [&half_of_fib42] { send(half_of_fib42, fibonacci(42) * 0.5); }
   );

   auto result = receive(join_all);
   wcout << L"fib35 = " << get<0>(result) << endl;
   wcout << L"fib37 = " << get<1>(result) << endl;
   wcout << L"half_of_fib42 = " << get<2>(result) << endl;
}

この例を実行すると、次の出力が生成されます。

fib35 = 9227465
fib37 = 24157817
half_of_fib42 = 1.33957e+008

join クラスの使用方法を示す完全な例については、「方法: 完了したタスクから選択する」および「チュートリアル: join を使用したデッドロックの防止」を参照してください。

[ページのトップへ]

timer クラス

Concurrency::timer クラスは、メッセージソースとして機能します。 timer オブジェクトでは、指定された時間が経過するとメッセージをターゲットに送信します。 timer クラスは、メッセージの送信を遅延させる必要がある場合や、定期的にメッセージを送信する場合に便利です。

timer クラスでは、メッセージを 1 つのターゲットにのみ送信します。コンストラクターの _PTarget パラメーターを NULL に設定した場合は、Concurrency::ISource::link_target メソッドを呼び出すことで、後からターゲットを指定できます。

timer オブジェクトには、繰り返しと非繰り返しがあります。繰り返しタイマーを作成するには、コンストラクターを呼び出すときに _Repeating パラメーターに true を渡します。また、非繰り返しタイマーを作成するには、_Repeating パラメーターに false を渡します。タイマーが繰り返しの場合、一定の間隔で同じメッセージをターゲットに送信します。

エージェントライブラリによって、開始されていない状態の timer オブジェクトが作成されます。タイマーオブジェクトを開始するには、Concurrency::timer::start メソッドを呼び出します。 timer オブジェクトを停止するには、そのオブジェクトを破棄するか、Concurrency::timer::stop メソッドを呼び出します。繰り返しタイマーを一時停止するには、Concurrency::timer::pause メソッドを呼び出します。

例

次の例では、timer クラスの使用方法の基本的な構造を示します。この例では、timer オブジェクトと call オブジェクトを使用して、時間のかかる操作の進行状況を報告します。

// timer-structure.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

// Computes the nth Fibonacci number.
// This function illustrates a lengthy operation and is therefore
// not optimized for performance.
int fibonacci(int n)
{
   if (n < 2)
      return n;
   return fibonacci(n-1) + fibonacci(n-2);
}

int wmain()
{
   // Create a call object that prints characters that it receives 
   // to the console.
   call<wchar_t> print_character([](wchar_t c) {
      wcout << c;
   });

   // Create a timer object that sends the period (.) character to 
   // the call object every 100 milliseconds.
   timer<wchar_t> progress_timer(100u, L'.', &print_character, true);

   // Start the timer.
   wcout << L"Computing fib(42)";
   progress_timer.start();

   // Compute the 42nd Fibonacci number.
   int fib42 = fibonacci(42);

   // Stop the timer and print the result.
   progress_timer.stop();
   wcout << endl << L"result is " << fib42 << endl;
}

この例では、次のサンプル出力が生成されます。

Computing fib(42)..................................................
result is 267914296

timer クラスの使用方法を示す完全な例については、「方法: メッセージを定期的に送信する」を参照してください。

[ページのトップへ]

メッセージのフィルター処理

メッセージブロックオブジェクトを作成する場合、メッセージブロックがメッセージを受け入れるか拒否するかを決めるフィルター関数を指定できます。フィルター関数は、メッセージブロックが特定の値のみを受信するようにする便利な方法です。

次の例では、フィルター関数を使用して偶数のみを受け入れる unbounded_buffer オブジェクトを作成する方法を示します。 unbounded_buffer オブジェクトは奇数を拒否するため、奇数はターゲットブロックに伝達されません。

// filter-function.cpp
// compile with: /EHsc
#include <agents.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

int wmain()
{
   // Create an unbounded_buffer object that uses a filter
   // function to accept only even numbers.
   unbounded_buffer<int> accept_evens(
      [](int n) {
         return (n%2) == 0;
      });

   // Send a few values to the unbounded_buffer object.
   unsigned int accept_count = 0;
   for (int i = 0; i < 10; ++i)
   {
      // The asend function returns true only if the target
      // accepts the message. This enables us to determine
      // how many elements are stored in the unbounded_buffer
      // object.
      if (asend(accept_evens, i))
      {
         ++accept_count;
      }
   }

   // Print to the console each value that is stored in the 
   // unbounded_buffer object. The unbounded_buffer object should
   // contain only even numbers.
   while (accept_count > 0)
   {
      wcout << receive(accept_evens) << L' ';
      --accept_count;
   }
}

この例を実行すると、次の出力が生成されます。

0 2 4 6 8

ラムダ関数、関数ポインター、または関数オブジェクトをフィルター関数として使用できます。各フィルター関数の形式は次のいずれかになります。

bool (_Type)
bool (_Type const &)

データの不必要なコピーをなくすには、値で伝達される集約型を扱うときに 2 番目の形式を使用します。

メッセージのフィルター処理では、データフロープログラミングモデルがサポートされています。このモデルでは、コンポーネントがデータの受信時に計算を実行します。フィルター関数を使用してメッセージパッシングネットワークでデータのフローを制御する方法の例については、「方法: メッセージブロックフィルターを使用する」、「チュートリアル: データフローエージェントの作成」、および「チュートリアル: イメージ処理ネットワークの作成」を参照してください。

[ページのトップへ]

メッセージの予約

メッセージの予約により、メッセージブロックはメッセージを予約して後で使用できるようにすることができます。通常、メッセージの予約を直接使用することはありません。ただし、メッセージの予約について理解しておくと、定義済みのいくつかのメッセージブロックの型の動作を理解しやすくなります。

最短一致の結合と最長一致の結合について考えてみてください。どちらの場合も、メッセージの予約によって、後で使用できるようにメッセージを予約します。前に述べたとおり、最短一致の結合では、2 つの段階でメッセージを受信します。最初の段階で、最短一致の join オブジェクトは各ソースがメッセージを受信するまで待機します。次に、各メッセージの予約を試みます。各メッセージを予約できる場合、すべてのメッセージを処理し、ターゲットに伝達します。それ以外の場合、メッセージの予約を解除し (取り消し)、再び各ソースがメッセージを受信するまで待機します。

最長一致の結合でも、複数のソースから入力メッセージを読み取ります。この場合、各ソースからメッセージを受信するのを待っている間、メッセージの予約を使用して追加のメッセージを読み取ります。たとえば、メッセージブロック A および B からメッセージを受信する最長一致の結合を考えてみます。 B から 2 つのメッセージを受信し、A からはまだメッセージを受信していない場合、最長一致の結合は、B からの 2 番目のメッセージに対応する一意のメッセージ識別子を保存します。 A からメッセージを受信し、これらのメッセージを伝達した後、保存済みのメッセージ識別子を使用して、B からの 2 番目のメッセージがまだ有効かどうかを確認します。

独自のカスタムメッセージブロックの型を実装する場合は、メッセージの予約を使用できます。カスタムメッセージブロックの型を作成する方法の例については、「チュートリアル: カスタムメッセージブロックの作成」を参照してください。

[ページのトップへ]

参照

概念

非同期エージェントライブラリ

履歴の変更

日付	履歴	理由
2010 年 8 月	基本的な例を追加。	情報の拡充

2010 年 8 月

基本的な例を追加。

情報の拡充

次の方法で共有

非同期メッセージブロック

セクション

ソースとターゲット

メッセージの伝達

メッセージブロックの型の概要

unbounded_buffer クラス

例

overwrite_buffer クラス

例

single_assignment クラス

例

call クラス

例

transformer クラス

例

choice クラス

例

join クラスと multitype_join クラス

最長一致の結合と最短一致の結合

例

timer クラス

例

メッセージのフィルター処理

メッセージの予約

参照

概念

履歴の変更

その他のリソース

次の方法で共有

非同期メッセージ ブロック

セクション

ソースとターゲット

メッセージの伝達

メッセージ ブロックの型の概要

unbounded_buffer クラス

例

overwrite_buffer クラス

例

single_assignment クラス

例

call クラス

例

transformer クラス

例

choice クラス

例

join クラスと multitype_join クラス

最長一致の結合と最短一致の結合

例

timer クラス

例

メッセージのフィルター処理

メッセージの予約

参照

概念

履歴の変更

その他のリソース

非同期メッセージブロック

メッセージブロックの型の概要