並列パターンライブラリに関するベストプラクティス

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

更新 : 2011 年 3 月

ここでは、並列パターンライブラリ (PPL) を効果的に使用する最適な方法について説明します。 PPL では、粒度の細かい並列化を実行するための汎用目的コンテナー、オブジェクト、およびアルゴリズムを提供します。

PPL の詳細については、「並列パターンライブラリ (PPL)」を参照してください。

セクション

このドキュメントは、次のセクションで構成されています。

小さなループ本体は並列化しない
可能な限り高いレベルで並列化を表現する
parallel_invoke を使用して分割と整理の問題を解決する
取り消し処理または例外処理を使用して並列ループから抜ける
取り消し処理および例外処理がオブジェクトの破棄に及ぼす影響について
並列ループではブロッキング操作を繰り返さない
並列処理を取り消すときにブロッキング操作を実行しない
並列ループでは共有データへの書き込みは行わない
可能な限り偽共有を避ける
変数がタスクの有効期間を通じて有効のまま維持されるようにする

小さなループ本体は並列化しない

比較的小さなループ本体を並列化すると、関連するスケジューリングオーバーヘッドが並列処理のメリットを上回る可能性があります。次の例について考えます。この例では、2 つの配列に各要素ペアを追加しています。

// small-loops.cpp
// compile with: /EHsc
#include <ppl.h>
#include <iostream>

using namespace Concurrency;
using namespace std;

int wmain()
{
   // Create three arrays that each have the same size.
   const size_t size = 100000;
   int a[size], b[size], c[size];

   // Initialize the arrays a and b.
   for (size_t i = 0; i < size; ++i)
   {
      a[i] = i;
      b[i] = i * 2;
   }

   // Add each pair of elements in arrays a and b in parallel 
   // and store the result in array c.
   parallel_for<size_t>(0, size, [&a,&b,&c](size_t i) {
      c[i] = a[i] + b[i];
   });

   // TODO: Do something with array c.
}

並列ループの各反復の作業負荷が小さすぎるため、並列処理のオーバーヘッドに勝る利益が得られません。ループ本体での処理量を増やすか、ループを逐次的に実行すると、ループのパフォーマンスが向上します。

[ページのトップへ]

可能な限り高いレベルで並列化を表現する

低いレベルでのみコードを並列化する場合は、プロセッサの数が増えても拡張されない fork-join コンストラクトを導入できます。 fork-join コンストラクトでは、タスクはその処理を複数の小さな並列サブタスクに分割し、それらのサブタスクが終了するまで待機します。サブタスクの場合も、それぞれの処理を再帰的に別のサブタスクに分割できます。

fork-join モデルはさまざまな問題を解決するのに役立ちますが、同期のオーバーヘッドに伴ってスケーラビリティが下がるような状況もあります。たとえば、イメージデータを逐次的に処理する次のコードについて考えます。

// Calls the provided function for each pixel in a Bitmap object.
void ProcessImage(Bitmap* bmp, const function<void (DWORD&)>& f)
{
   int width = bmp->GetWidth();
   int height = bmp->GetHeight();

   // Lock the bitmap.
   BitmapData bitmapData;
   Rect rect(0, 0, bmp->GetWidth(), bmp->GetHeight());
   bmp->LockBits(&rect, ImageLockModeWrite, PixelFormat32bppRGB, &bitmapData);

   // Get a pointer to the bitmap data.
   DWORD* image_bits = (DWORD*)bitmapData.Scan0;

   // Call the function for each pixel in the image.
   for (int y = 0; y < height; ++y)
   {      
      for (int x = 0; x < width; ++x)
      {
         // Get the current pixel value.
         DWORD* curr_pixel = image_bits + (y * width) + x;

         // Call the function.
         f(*curr_pixel);
      }
   }

   // Unlock the bitmap.
   bmp->UnlockBits(&bitmapData);
}

各ループ反復は独立しているため、次の例に示すように処理の大部分を並列化できます。この例では、Concurrency::parallel_for アルゴリズムを使用して、外側のループを並列化します。

// Calls the provided function for each pixel in a Bitmap object.
void ProcessImage(Bitmap* bmp, const function<void (DWORD&)>& f)
{
   int width = bmp->GetWidth();
   int height = bmp->GetHeight();

   // Lock the bitmap.
   BitmapData bitmapData;
   Rect rect(0, 0, bmp->GetWidth(), bmp->GetHeight());
   bmp->LockBits(&rect, ImageLockModeWrite, PixelFormat32bppRGB, &bitmapData);

   // Get a pointer to the bitmap data.
   DWORD* image_bits = (DWORD*)bitmapData.Scan0;

   // Call the function for each pixel in the image.
   parallel_for (0, height, [&, width](int y)
   {      
      for (int x = 0; x < width; ++x)
      {
         // Get the current pixel value.
         DWORD* curr_pixel = image_bits + (y * width) + x;

         // Call the function.
         f(*curr_pixel);
      }
   });

   // Unlock the bitmap.
   bmp->UnlockBits(&bitmapData);
}

次の例では、ループで ProcessImage 関数を呼び出す fork-join コンストラクトを示します。 ProcessImage を呼び出すたびに制御が戻されるのではなく、各サブタスクが終了してから戻されます。

// Processes each bitmap in the provided vector.
void ProcessImages(vector<Bitmap*> bitmaps, const function<void (DWORD&)>& f)
{
   for_each(bitmaps.begin(), bitmaps.end(), [&f](Bitmap* bmp) {
      ProcessImage(bmp, f);
   });
}

並列ループの各反復でほとんど処理が実行されないか、並列ループで実行される処理が不均衡の場合 (つまり、ループ反復ごとに所要時間が異なる場合)、処理のフォークと結合を頻繁に行うことによるスケジューリングオーバーヘッドが並列実行のメリットを上回ることがあります。プロセッサの数が増えると、このオーバーヘッドも大きくなります。

この例のスケジューリングオーバーヘッドの量を低減するには、内側のループを並列化する前に外側のループを並列化するか、パイプライン処理などの別の parallel コンストラクトを使用します。次の例では、ProcessImages 関数を変更して、Concurrency::parallel_for_each アルゴリズムによって外側のループを並列化するようにします。

// Processes each bitmap in the provided vector.
void ProcessImages(vector<Bitmap*> bitmaps, const function<void (DWORD&)>& f)
{
   parallel_for_each(bitmaps.begin(), bitmaps.end(), [&f](Bitmap* bmp) {
      ProcessImage(bmp, f);
   });
}

パイプラインを使用してイメージ処理を並列に実行する類似の例については、「チュートリアル: イメージ処理ネットワークの作成」を参照してください。

[ページのトップへ]

parallel_invoke を使用して分割と整理の問題を解決する

分割と整理 (divide-and-conquer) の問題は、再帰を使用してタスクをサブタスクに分割する fork-join コンストラクトの一種です。 Concurrency::task_group クラスや Concurrency::structured_task_group クラス以外に、Concurrency::parallel_invoke アルゴリズムを使用して分割と整理の問題を解決することもできます。 parallel_invoke アルゴリズムでは、タスクグループオブジェクトよりも簡単な構文を使用できます。このアルゴリズムは、並列タスクの数が固定されている場合に役立ちます。

次の例では、parallel_invoke アルゴリズムを使用して、バイトニックソートアルゴリズムを実装します。

// Sorts the given sequence in the specified order.
template <class T>
void parallel_bitonic_sort(T* items, int lo, int n, bool dir)
{   
   if (n > 1)
   {
      // Divide the array into two partitions and then sort 
      // the partitions in different directions.
      int m = n / 2;

      parallel_invoke(
         [&] { parallel_bitonic_sort(items, lo, m, INCREASING); },
         [&] { parallel_bitonic_sort(items, lo + m, m, DECREASING); }
      );

      // Merge the results.
      parallel_bitonic_merge(items, lo, n, dir);
   }
}

parallel_invoke アルゴリズムでは、オーバーヘッドを低減するために、一連のタスクの最後のタスクを呼び出し元のコンテキストで実行します。

この例の完全なバージョンについては、「方法: 並列呼び出しを使用して並列並べ替えルーチンを記述する」を参照してください。 parallel_invoke アルゴリズムの詳細については、「並列アルゴリズム」を参照してください。

[ページのトップへ]

取り消し処理または例外処理を使用して並列ループから抜ける

PPL では、2 とおりの方法を使用して、タスクグループまたは並列アルゴリズムによって実行される並列処理を取り消すことができます。 1 つは、Concurrency::task_group クラスと Concurrency::structured_task_group クラスが提供する取り消し機構を使用する方法です。もう 1 つは、タスクの処理関数の本体で例外をスローする方法です。並列処理ツリーを取り消す場合は、例外処理を行うよりも取り消し機構を使用する方が効率的です。並列処理ツリーは、関連するタスクグループのグループです。つまり、タスクグループの中に他のタスクグループが含まれます。取り消し機構では、タスクグループとその子タスクグループを上位から順に取り消します。反対に、例外処理では下位から順に処理されるため、例外が上位へ移動するたびに、子タスクグループを個別に取り消す必要があります。

タスクグループオブジェクトを直接操作する場合は、Concurrency::task_group::cancel メソッドまたは Concurrency::structured_task_group::cancel メソッドを使用して、そのタスクグループに属している処理を取り消します。たとえば、parallel_for では、並列アルゴリズムを取り消すために、親タスクグループを作成し、そのタスクグループを取り消します。たとえば、次の関数 parallel_find_any について考えます。この関数では、配列内の値を並列に検索します。

// Returns the position in the provided array that contains the given value, 
// or -1 if the value is not in the array.
template<typename T>
int parallel_find_any(const T a[], size_t count, const T& what)
{
   // The position of the element in the array. 
   // The default value, -1, indicates that the element is not in the array.
   int position = -1;

   // Use parallel_for to search for the element. 
   // The task group enables a work function to cancel the overall 
   // operation when it finds the result.

   structured_task_group tasks;
   tasks.run_and_wait([&]
   {
      parallel_for(std::size_t(0), count, [&](int n) {
         if (a[n] == what)
         {
            // Set the return value and cancel the remaining tasks. 
            position = n;            
            tasks.cancel();
         }
      });
   });

   return position;
}

並列アルゴリズムではタスクグループが使用されるため、並列反復処理のいずれかによって親タスクグループが取り消されると、タスク全体が取り消されます。この例の完全なバージョンについては、「方法: キャンセル処理を使用して並列ループを中断する」を参照してください。

例外処理を使用して並列処理を取り消す方法は、取り消し機構を使用する方法よりも効率の面で劣りますが、例外処理の方が適している状況もあります。たとえば、次のメソッド for_all では、tree 構造体の各ノードに対して処理関数を再帰的に実行します。この例では、_children データメンバーは、tree オブジェクトを含む std::list です。

// Performs the given work function on the data element of the tree and
// on each child.
template<class Function>
void tree::for_all(Function& action)
{
   // Perform the action on each child.
   parallel_for_each(_children.begin(), _children.end(), [&](tree& child) {
      child.for_all(action);
   });

   // Perform the action on this node.
   action(*this);
}

ツリーの各要素に対して処理関数を呼び出す必要がない場合、tree::for_all メソッドの呼び出し元は例外をスローすることがあります。次の例は search_for_value 関数で、提供された tree オブジェクト内で値を検索します。 search_for_value 関数で使用される処理関数は、ツリーの現在の要素と提供された値が一致する場合に例外をスローします。 search_for_value 関数は、try-catch ブロックを使用して例外をキャプチャし、結果をコンソールに出力します。

// Searches for a value in the provided tree object.
template <typename T>
void search_for_value(tree<T>& t, int value)
{
   try
   {
      // Call the for_all method to search for a value. The work function
      // throws an exception when it finds the value.
      t.for_all([value](const tree<T>& node) {
         if (node.get_data() == value)
         {
            throw &node;
         }
      });
   }
   catch (const tree<T>* node)
   {
      // A matching node was found. Print a message to the console.
      wstringstream ss;
      ss << L"Found a node with value " << value << L'.' << endl;
      wcout << ss.str();
      return;
   }

   // A matching node was not found. Print a message to the console.
   wstringstream ss;
   ss << L"Did not find node with value " << value << L'.' << endl;
   wcout << ss.str();   
}

この例の完全なバージョンについては、「方法: 例外処理を使用して並列ループを中断する」を参照してください。

PPL に用意されている取り消し機構と例外処理機構に関する一般的な情報については、「PPL における取り消し処理」および「同時実行ランタイムでの例外処理」を参照してください。

[ページのトップへ]

取り消し処理および例外処理がオブジェクトの破棄に及ぼす影響について

並列処理ツリーでタスクを取り消すと、子タスクも実行されなくなります。そのため、アプリケーションで重要となる操作 (リソースの解放など) が子タスクのいずれかで実行されるような場合に問題となります。また、タスクを取り消すと、例外がオブジェクトのデストラクターを通じて反映され、アプリケーションで未定義の動作が発生する可能性があります。

次の例では、Resource クラスはリソースを表し、Container クラスはリソースを保持するコンテナーを表します。そのデストラクターで、Container クラスは Resource メンバーの 2 つに対して cleanup メソッドを並列に呼び出し、3 つ目の Resource メンバーに対して cleanup メソッドを呼び出します。

// parallel-resource-destruction.h
#pragma once
#include <ppl.h>
#include <sstream>
#include <iostream>

// Represents a resource.
class Resource
{
public:
   Resource(const std::wstring& name)
      : _name(name)
   {
   }

   // Frees the resource.
   void cleanup()
   {
      // Print a message as a placeholder.
      std::wstringstream ss;
      ss << _name << L": Freeing..." << std::endl;
      std::wcout << ss.str();
   }
private:
   // The name of the resource.
   std::wstring _name;
};

// Represents a container that holds resources.
class Container
{
public:
   Container(const std::wstring& name)
      : _name(name)
      , _resource1(L"Resource 1")
      , _resource2(L"Resource 2")
      , _resource3(L"Resource 3")
   {
   }

   ~Container()
   {
      std::wstringstream ss;
      ss << _name << L": Freeing resources..." << std::endl;
      std::wcout << ss.str();

      // For illustration, assume that cleanup for _resource1
      // and _resource2 can happen concurrently, and that 
      // _resource3 must be freed after _resource1 and _resource2.

      Concurrency::parallel_invoke(
         [this]() { _resource1.cleanup(); },
         [this]() { _resource2.cleanup(); }
      );

      _resource3.cleanup();
   }

private:
   // The name of the container.
   std::wstring _name;

   // Resources.
   Resource _resource1;
   Resource _resource2;
   Resource _resource3;
};

このパターン自体には問題はありませんが、2 つのタスクを並列に実行する次のコードについて考えてみてください。 1 つ目のタスクでは Container オブジェクトを作成し、2 つ目のタスクではタスク全体を取り消します。これを示すために、この例では、2 つの Concurrency::event オブジェクトを使用して、Container オブジェクトが作成された後に取り消し処理を行い、取り消し処理が終了したら Container オブジェクトを破棄するようにしています。

// parallel-resource-destruction.cpp
// compile with: /EHsc
#include "parallel-resource-destruction.h"

using namespace Concurrency;
using namespace std;

static_assert(false, "This example illustrates a non-recommended practice.");

int main()
{  
   // Create a task_group that will run two tasks.
   task_group tasks;

   // Used to synchronize the tasks.
   event e1, e2;

   // Run two tasks. The first task creates a Container object. The second task
   // cancels the overall task group. To illustrate the scenario where a child 
   // task is not run because its parent task is cancelled, the event objects 
   // ensure that the Container object is created before the overall task is 
   // cancelled and that the Container object is destroyed after the overall 
   // task is cancelled.

   tasks.run([&tasks,&e1,&e2] {
      // Create a Container object.
      Container c(L"Container 1");

      // Allow the second task to continue.
      e2.set();

      // Wait for the task to be cancelled.
      e1.wait();
   });

   tasks.run([&tasks,&e1,&e2] {
      // Wait for the first task to create the Container object.
      e2.wait();

      // Cancel the overall task.
      tasks.cancel();      

      // Allow the first task to continue.
      e1.set();
   });

   // Wait for the tasks to complete.
   tasks.wait();

   wcout << L"Exiting program..." << endl;
}

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

Container 1: Freeing resources...
Exiting program...

このコード例には次の問題が含まれているため、期待どおりの動作が得られない可能性があります。

親タスクを取り消すと、子タスク (Concurrency::parallel_invoke の呼び出し) も取り消されます。したがって、これら 2 つのリソースは解放されません。
親タスクを取り消すと、子タスクが内部例外をスローします。 Container デストラクターではこの例外は処理されないため、例外が上位に反映され、3 つ目のリソースは解放されません。
子タスクがスローした例外は、Container デストラクターを通じて反映されます。デストラクターからスローされると、アプリケーションは未定義の状態になります。

タスクが絶対に取り消されない場合を除き、タスクでは重要な操作 (リソースの解放など) を実行しないことをお勧めします。また、採用している型のデストラクターと連携するようなランタイム機能は使用しないようにしてください。

[ページのトップへ]

並列ループではブロッキング操作を繰り返さない

Concurrency::parallel_for や Concurrency::parallel_for_each などの並列ループで何度もブロッキング操作を使用すると、ランタイムで多数のスレッドが短期間に作成される可能性があります。

同時実行ランタイムでは、タスクが終了するか、協調的にブロックまたは譲渡すると、追加の処理が実行されます。並列ループの 1 回の反復がブロックすると、別の反復が開始されることがあります。使用可能なアイドルスレッドが存在しない場合は、新しいスレッドが作成されます。

並列ループの本体がときどきブロックする程度であれば、この機構によりタスクの全体的なスループットが最大限に高まります。ただし、反復処理で頻繁にブロックが発生する場合は、追加の処理を実行するために多数のスレッドが作成される可能性があります。それに伴って、メモリ不足の状態に陥ったり、ハードウェアリソースが不適切に使用されたりする場合があります。

次の例について考えます。この例では、parallel_for ループの各反復で Concurrency::send 関数を呼び出します。 send は協調的にブロックするため、send を呼び出すたびに、追加の処理を実行するための新しいスレッドが作成されます。

// repeated-blocking.cpp
// compile with: /EHsc
#include <ppl.h>
#include <agents.h>

using namespace Concurrency;

static_assert(false, "This example illustrates a non-recommended practice.");

int main()
{
   // Create a message buffer.
   overwrite_buffer<int> buffer;

   // Repeatedly send data to the buffer in a parallel loop.
   parallel_for(0, 1000, [&buffer](int i) {

      // The send function blocks cooperatively. 
      // We discourage the use of repeated blocking in a parallel
      // loop because it can cause the runtime to create 
      // a large number of threads over a short period of time.
      send(buffer, i);
   });
}

このパターンを回避するようコードをリファクタリングすることをお勧めします。この例では、逐次的な for ループで send を呼び出すことによって、追加のスレッドの作成を回避できます。

[ページのトップへ]

並列処理を取り消すときにブロッキング操作を実行しない

可能であれば、Concurrency::task_group::cancel メソッドまたは Concurrency::structured_task_group::cancel メソッドを呼び出して並列処理を取り消す前に、ブロッキング操作を実行することは避けてください。

タスクがブロッキング操作を実行すると、ランタイムは最初のタスクがデータを待っている間に他の処理を実行できます。ユーザーモードスケジューリング (UMS: User-Mode Scheduling) を有効にした場合は、タスクが協調的ブロッキング操作を実行するか、カーネル遷移を伴うブロッキング操作を実行すると、ランタイムは他の処理を実行します。通常のスレッドスケジューリング (既定) を有効にした場合は、タスクが協調的ブロッキング操作を実行したときにのみ、ランタイムは他の処理を実行します。待機中のタスクがブロック解除すると、ランタイムはそのタスクを再スケジュールします。ランタイムは、通常、直近にブロック解除したタスクから順に再スケジュールします。そのため、ブロッキング操作中に不要な処理がスケジュールされ、パフォーマンスが低下します。したがって、並列処理を取り消す前にブロッキング操作を実行すると、そのブロッキング操作が原因で cancel の呼び出しが遅れる可能性があります。それにより、他のタスクで不要な処理が実行されるようになります。

次の例について考えます。この例では、提供された述語関数を満たす指定の配列の要素を検索する parallel_find_answer 関数を定義します。述語関数が true を返すと、並列処理関数は Answer オブジェクトを作成し、タスク全体を取り消します。

// blocking-cancel.cpp
// compile with: /c /EHsc
#include <windows.h>
#include <ppl.h>

using namespace Concurrency;

// Encapsulates the result of a search operation.
template<typename T>
class Answer
{
public:
   explicit Answer(const T& data)
      : _data(data)
   {
   }

   T get_data() const
   {
      return _data;
   }

   // TODO: Add other methods as needed.

private:
   T _data;

   // TODO: Add other data members as needed.
};

// Searches for an element of the provided array that satisfies the provided
// predicate function.
template<typename T, class Predicate>
Answer<T>* parallel_find_answer(const T a[], size_t count, const Predicate& pred)
{
   // The result of the search.
   Answer<T>* answer = nullptr;
   // Ensures that only one task produces an answer.
   volatile long first_result = 0;

   // Use parallel_for and a task group to search for the element.
   structured_task_group tasks;
   tasks.run_and_wait([&]
   {
      // Declare the type alias for use in the inner lambda function.
      typedef T T;

      parallel_for<size_t>(0, count, [&](const T& n) {
         if (pred(a[n]) && InterlockedExchange(&first_result, 1) == 0)
         {
            // Create an object that holds the answer.
            answer = new Answer<T>(a[n]);
            // Cancel the overall task.
            tasks.cancel();
         }
      });
   });

   return answer;
}

new 演算子は、ブロッキングの対象となる可能性のあるヒープ割り当てを実行します。ユーザーモードスケジューリング (UMS) を有効にした場合は、ブロッキング操作中、ランタイムは他の処理を実行します。通常のスレッドスケジューリングを有効にした場合は、タスクが協調的ブロッキング呼び出し (Concurrency::critical_section::lock の呼び出しなど) を実行したときにのみ、ランタイムは他の処理を実行します。

次の例では、不要な処理を回避し、それによってパフォーマンスを向上させる方法を示します。この例では、Answer オブジェクト用のストレージを割り当てる前に、タスクグループを取り消します。

// Searches for an element of the provided array that satisfies the provided
// predicate function.
template<typename T, class Predicate>
Answer<T>* parallel_find_answer(const T a[], size_t count, const Predicate& pred)
{
   // The result of the search.
   Answer<T>* answer = nullptr;
   // Ensures that only one task produces an answer.
   volatile long first_result = 0;

   // Use parallel_for and a task group to search for the element.
   structured_task_group tasks;
   tasks.run_and_wait([&]
   {
      // Declare the type alias for use in the inner lambda function.
      typedef T T;

      parallel_for<size_t>(0, count, [&](const T& n) {
         if (pred(a[n]) && InterlockedExchange(&first_result, 1) == 0)
         {
            // Cancel the overall task.
            tasks.cancel();
            // Create an object that holds the answer.
            answer = new Answer<T>(a[n]);            
         }
      });
   });

   return answer;
}

[ページのトップへ]

並列ループでは共有データへの書き込みは行わない

同時実行ランタイムには、共有データへの同時アクセスを同期するいくつかのデータ構造体 (Concurrency::critical_section など) が用意されています。これらのデータ構造体は、複数のタスクでリソースへの共有アクセスがまれに必要になる場合など、さまざまな状況で役立ちます。

次の例について考えます。この例では、Concurrency::parallel_for_each アルゴリズムと critical_section オブジェクトを使用して、std::array オブジェクト内の素数の数を計算します。この例では、各スレッドが共有変数 prime_sum にアクセスするのを待つ必要があるため、効率は改善されません。

critical_section cs;
prime_sum = 0;
parallel_for_each(a.begin(), a.end(), [&](int i) {
   cs.lock();
   prime_sum += (is_prime(i) ? i : 0);
   cs.unlock();
});

また、頻繁なロック操作によってループが事実上シリアル化されるため、パフォーマンスが低下する可能性もあります。さらに、同時実行ランタイムオブジェクトがブロック操作を実行すると、スケジューラによって追加のスレッドが作成され、最初のスレッドがデータを待っている間に他の処理が実行される可能性があります。共有データを待つタスクが多数存在し、それに対処するためにランタイムで多数のスレッドが作成されると、アプリケーションのパフォーマンス低下やリソース不足状態が発生することがあります。

PPL には、ロック制御なしで共有リソースへのアクセスを提供することによって共有状態を解消する Concurrency::combinable クラスが定義されています。 combinable クラスにはスレッドローカルストレージが用意されており、詳細な計算を実行した後、その計算を最終結果にマージできます。 combinable オブジェクトは減少変数と考えることができます。

次の例では、前の例を変更し、合計を計算するときに critical_section オブジェクトの代わりに combinable オブジェクトを使用するようにしています。この例では、各スレッドが合計のローカルコピーをそれぞれ保持するため、効率が改善されます。この例では、Concurrency::combinable::combine メソッドを使用して、ローカルの計算を最終結果にマージします。

combinable<int> sum;
parallel_for_each(a.begin(), a.end(), [&](int i) {
   sum.local() += (is_prime(i) ? i : 0);
});
prime_sum = sum.combine(plus<int>());

この例の完全なバージョンについては、「方法: combinable を使用してパフォーマンスを向上させる」を参照してください。 combinable クラスの詳細については、「並列コンテナーと並列オブジェクト」を参照してください。

[ページのトップへ]

可能な限り偽共有を避ける

別々のプロセッサで実行されている複数の同時実行タスクが、同じキャッシュラインに配置されている変数に書き込みを行うと、偽共有が発生します。 1 つのタスクが変数のいずれかに書き込みを行うと、両方の変数のキャッシュラインが無効化されます。キャッシュラインが無効化されるたびに、各プロセッサでキャッシュラインの再読み込みを行う必要があります。したがって、偽共有が発生すると、アプリケーションでパフォーマンスが低下する可能性があります。

次の基本的な例では、2 つの同時実行タスクがそれぞれ共有カウンター変数をインクリメントする場合を示します。

volatile long count = 0L;
Concurrency::parallel_invoke(
   [&count] {
      for(int i = 0; i < 100000000; ++i)
         InterlockedIncrement(&count);
   },
   [&count] {
      for(int i = 0; i < 100000000; ++i)
         InterlockedIncrement(&count);
   }
);

2 つのタスク間でデータが共有されないようにするには、2 つのカウンター変数を使用するように例を変更します。この例では、タスクの終了後に最終的なカウンター値を計算します。ただし、count1 変数と count2 変数が同じキャッシュラインに配置される可能性があるため、偽共有となります。

long count1 = 0L;
long count2 = 0L;
Concurrency::parallel_invoke(
   [&count1] {
      for(int i = 0; i < 100000000; ++i)
         ++count1;
   },
   [&count2] {
      for(int i = 0; i < 100000000; ++i)
         ++count2;
   }
);
long count = count1 + count2;

偽共有をなくす方法の 1 つは、カウンター変数を必ず別々のキャッシュラインに配置することです。次の例では、64 バイト境界上に count1 変数と count2 変数を配置します。

__declspec(align(64)) long count1 = 0L;      
__declspec(align(64)) long count2 = 0L;      
Concurrency::parallel_invoke(
   [&count1] {
      for(int i = 0; i < 100000000; ++i)
         ++count1;
   },
   [&count2] {
      for(int i = 0; i < 100000000; ++i)
         ++count2;
   }
);
long count = count1 + count2;

この例では、メモリキャッシュのサイズが 64 バイト以下であると仮定しています。

タスク間でデータを共有する必要がある場合は、Concurrency::combinable クラスを使用することをお勧めします。 combinable クラスは、偽共有が発生する可能性を減らすようにスレッドローカル変数を作成します。 combinable クラスの詳細については、「並列コンテナーと並列オブジェクト」を参照してください。

[ページのトップへ]

変数がタスクの有効期間を通じて有効のまま維持されるようにする

タスクグループまたは並列アルゴリズムにラムダ式を渡す場合、ラムダ式の本体で外側のスコープ内の変数を値でアクセスするのか参照でアクセスするのかは、capture 句で指定します。ラムダ式に変数を渡すときに参照渡しを使用する場合、タスクが終了するまでその変数の有効期間が続くようにする必要があります。

次の例について考えます。この例では、object クラスと perform_action 関数を定義します。 perform_action 関数は、object 変数を作成し、その変数に対していくつかのアクションを非同期的に実行します。 perform_action 関数から制御が戻される前にタスクが終了するとは限らないため、タスクの実行中に object 変数が破棄されると、プログラムのクラッシュや未定義の動作が発生します。

// lambda-lifetime.cpp
// compile with: /c /EHsc
#include <ppl.h>

using namespace Concurrency;

// A type that performs an action.
class object
{
public:
   void action() const
   {
      // TODO: Details omitted for brevity.
   }
};

// Performs an action asynchronously.
void perform_action(task_group& tasks)
{
   // Create an object variable and perform some action on 
   // that variable asynchronously.
   object obj;
   tasks.run([&obj] {
      obj.action();
   });

   // NOTE: The object variable is destroyed here. The program
   // will crash or exhibit unspecified behavior if the task
   // is still running when this function returns.
}

アプリケーションの要件に応じて、次の手法のいずれかを使用して、変数が各タスクの有効期間を通じて有効のまま維持されるようにすることができます。

次の例では、タスクに対して object 変数の値渡しを行います。したがって、タスクは変数の個別コピーに対して作用します。

// Performs an action asynchronously.
void perform_action(task_group& tasks)
{
   // Create an object variable and perform some action on 
   // that variable asynchronously.
   object obj;
   tasks.run([obj] {
      obj.action();
   });
}

object 変数は値渡しで渡されるため、この変数の状態変化が発生しても、元のコピーには反映されません。

次の例では、Concurrency::task_group::wait メソッドを使用して、perform_action 関数から制御が戻される前にタスクが確実に終了するようにします。

// Performs an action.
void perform_action(task_group& tasks)
{
   // Create an object variable and perform some action on 
   // that variable.
   object obj;
   tasks.run([&obj] {
      obj.action();
   });

   // Wait for the task to finish. 
   tasks.wait();
}

今回は、関数から制御が戻される前にタスクが終了するため、perform_action 関数が非同期に動作することはありません。

次の例では、perform_action 関数を変更して、object 変数の参照が使用されるようにします。呼び出し元は、タスクが終了するまで object 変数の有効期間が続くようにする必要があります。

// Performs an action asynchronously.
void perform_action(object& obj, task_group& tasks)
{
   // Perform some action on the object variable.
   tasks.run([&obj] {
      obj.action();
   });
}

また、タスクグループまたは並列アルゴリズムに渡すオブジェクトの有効期間をポインターで制御することもできます。

ラムダ式の詳細については、「Lambda Expressions in C++」を参照してください。

[ページのトップへ]

参照

履歴の変更

日付	履歴	理由
2011 年 3 月	並列ループでブロッキング操作が繰り返されないようにする方法、および取り消し処理と例外処理がオブジェクトの破棄に及ぼす影響についての情報を追加。	情報の拡充
2010 年 5 月	ガイドラインを拡張。	情報の拡充

2011 年 3 月

並列ループでブロッキング操作が繰り返されないようにする方法、および取り消し処理と例外処理がオブジェクトの破棄に及ぼす影響についての情報を追加。

情報の拡充

2010 年 5 月

ガイドラインを拡張。

情報の拡充

次の方法で共有

並列パターンライブラリに関するベストプラクティス

セクション

小さなループ本体は並列化しない

可能な限り高いレベルで並列化を表現する

parallel_invoke を使用して分割と整理の問題を解決する

取り消し処理または例外処理を使用して並列ループから抜ける

取り消し処理および例外処理がオブジェクトの破棄に及ぼす影響について

並列ループではブロッキング操作を繰り返さない

並列処理を取り消すときにブロッキング操作を実行しない

並列ループでは共有データへの書き込みは行わない

可能な限り偽共有を避ける

変数がタスクの有効期間を通じて有効のまま維持されるようにする

参照

処理手順

概念

その他の技術情報

履歴の変更

その他のリソース

次の方法で共有

並列パターン ライブラリに関するベスト プラクティス

セクション

小さなループ本体は並列化しない

可能な限り高いレベルで並列化を表現する

parallel_invoke を使用して分割と整理の問題を解決する

取り消し処理または例外処理を使用して並列ループから抜ける

取り消し処理および例外処理がオブジェクトの破棄に及ぼす影響について

並列ループではブロッキング操作を繰り返さない

並列処理を取り消すときにブロッキング操作を実行しない

並列ループでは共有データへの書き込みは行わない

可能な限り偽共有を避ける

変数がタスクの有効期間を通じて有効のまま維持されるようにする

参照

処理手順

概念

その他の技術情報

履歴の変更

その他のリソース

並列パターンライブラリに関するベストプラクティス