リダクション

問題

データセットにまたがって結合則を満たすリダクション操作を行う。

コンテキスト

多くのシリアル・アルゴリズムは、アイテムのセットを走査してサマリー情報を収集します。

影響

サマリーはデータセットの結合則を満たす操作として表現されます。または、少なくとも再結合が重要でないほど結合則を満たしたものになります。

ソリューション

インテル® スレッディング・ビルディング・ブロック (インテル® TBB) には 2 つのソリューションが含まれています。どちらのソリューションを使用するかは、次の条件に依存します。

• 操作は可換で結合則を満たす操作か?

• リダクション型のインスタンスの構築と破棄にコストがかかるか? 例えば、浮動小数点数の構築にはコストはかかりませんが、浮動小数点数の疎行列の構築にはコストが非常にかかるかもしれません。

オブジェクトを構築するコストがあまりかからない場合は、tbb::parallel_reduce を使用します。これは、リダクション操作が可換でない場合も動作します。インテル® TBB のチュートリアルでは、基本的なリダクション tbb::parallel_reduce を使用する方法を説明しています。

リダクション操作が可換で、型のインスタンスにコストがかかる場合、tbb::parallel_for と tbb::combinable を使用します。

操作が厳密には結合則を満たしていないが厳密な決定性による結果が必要な場合は、再帰的なリダクションを使用して、tbb::parallel_invoke で並列化します。

サンプル

次のサンプルは、さまざまなソリューションといくつかのトレードオフを示しています。

最初のサンプルは、tbb::parallel_reduce を使用して T 型のシーケンスに a + リダクションを行っています。シーケンスは、半開区間 [first,last) で定義されます。

T AssocReduce( const T* first, const T* last, T identity ) {
   return tbb::parallel_reduce(
       // リダクションのインデックス範囲を指定
       tbb::blocked_range<const T*>(first,last),
       // 単位元要素
       identity,
       // 部分範囲と部分和をまとめる
       [&]( tbb::blocked_range<const T*> r, T partial_sum )->float {
           return std::accumulate( r.begin(), r.end(), partial_sum );
       },
       // 2 つの部分和をまとめる
       std::plus<T>()
   );
} 

この形式の parallel_reduce の第 3 および第 4 引数は、擬集化パターンの形式で構築されます。リダクションの前に要素単位動作を行わなければならない場合、第 3 引数 (部分範囲のリダクション) に動作を追加すると、参照の局所性が向上し、パフォーマンスが改善されます。

2 つめの例は、+ が T で可換であると仮定しています。T オブジェクトの構築にコストがかかる場合には良いソリューションです。

T CombineReduce( const T* first, const T* last, T identity ) {
   tbb::combinable<T> sum(identity);
   tbb::parallel_for(
       tbb::blocked_range<const T*>(first,last),
       [&]( tbb::blocked_range<const T*> r ) {
           sum.local() += std::accumulate(r.begin(), r.end(), identity);
       }
   );
   return sum.combine( []( const T& x, const T& y ) {return x+y;} );
}

最終的な結果を生成するために部分的な結果を使用することが望ましい場合があります。例えば、部分的な結果がリストの場合、リストを結合して最終的な結果を生成することができます。その場合、combinable の代わりに tbb::enumerable_thread_specific クラスを使用します。「分割統治」のサンプル ParallelFindCollisions は、このテクニックを示しています。

浮動小数点の加算と乗算はほぼ結合則を満たしています。結合を行うと、丸めの影響により結果が変更されることがあります。次に示す手法では、アイテムを非決定論的に再結合します。全く結合則を満たさない操作に対して完全に決定性のある並列リダクションを行うには、決定性のある再結合を使用する必要があります。次のコードは、T 型の値のシーケンスに a + リダクションを行うテンプレートで、この処理を示しています。

template<typename T>
T RepeatableReduce( const T* first, const T* last, T identity ) {
   if( last-first<=1000 ) {
       // シリアル・リダクションを使用
       return std::accumulate( first, last, identity );
   } else {
       // 並列の分割統治リダクションを行う
       const T* mid = first+(last-first)/2;
       T left, right;
       tbb::parallel_invoke(
           [&]{left=RepeatableReduce(first,mid,identity);},
           [&]{right=RepeatableReduce(mid,last,identity);} 
       );
       return left+right;
   }
}

if-else 文は、再帰計算の擬集化パターンのインスタンスです。リダクション・グラフは厳密な二分木ではありませんが、完全な決定性があります。このため、すべてのスレッドで浮動小数点の丸めが同一であれば、指定された入力シーケンスで結果は常に同じになります。

最後のサンプルは、通常はリダクションとは見なされない問題をリダクションとして並列化する方法を示しています。この問題は、データセットにまたがった計算で浮動小数点例外フラグを取得しています。シリアルコードは次のようになります。

   feclearexcept(FE_ALL_EXCEPT); 
   for( int i=0; i<N; ++i )
       C[i]=A[i]*B[i];
   int flags = fetestexcept(FE_ALL_EXCEPT);
   if (flags & FE_DIVBYZERO) ...;
   if (flags & FE_OVERFLOW) ...;
   ...

このコードは、ループのチャンクを別々に計算して各チャンクの浮動小数点フラグをマージすることで並列化できます。この処理を tbb:parallel_reduce で行うには、次に示すように、最初に「ボディー」の型を定義します。

struct ComputeChunk {
   int flags;          // これまでに見つかった浮動小数点例外を保持
   void reset_fpe() {
       flags=0;
       feclearexcept(FE_ALL_EXCEPT);
   }
   ComputeChunk () {
       reset_fpe();
   }
   // 範囲を部分範囲に分割するとき parallel_reduce に呼び出される "分割コンストラクター"
   ComputeChunk ( const ComputeChunk&, tbb::split ) {
       reset_fpe();
   }
   // チャンクを操作し、浮動小数点例外の状態を収集して flags メンバーに設定
   void operator()( tbb::blocked_range<int> r ) {
       int end=r.end();
       for( int i=r.begin(); i!=end; ++i )
           C[i] = A[i]/B[i];
       // ここで = ではなく |= を使用することが重要
       // そうしないと同一スレッドにおいて先に見つかっている例外が失われる
       flags |= fetestexcept(FE_ALL_EXCEPT);
   }
   // 2 つの部分範囲の結果を結合するとき parallel_reduce に呼び出される
   void join( Body& other ) {
       flags |= other.flags;
   }
};

その後、次のように呼び出します。

// cc の構築によって FP 例外状態が暗黙的にリセットされる
   ComputeChunk cc;
   tbb::parallel_reduce( tbb::blocked_range<int>(0,N), cc );
   if (cc.flags & FE_DIVBYZERO) ...;
   if (cc.flags & FE_OVERFLOW) ...;
   ...