🚊 📷 🍞 Parallel.ForEachを使用する場合とPLINQを使用する場合 ✊🏻 🥥 📕

はじめに

通常、マルチコアコンピューター用にプログラムを最適化する場合、最初のステップは、アルゴリズムを並列実行する部分に分割する可能性を見つけることです。問題を解決するために、大きなデータセットから個別の要素を並列処理する必要がある場合、最初の候補は、.NET Framework 4： Parallel.ForEachおよびParallel LINQ（ PLINQ ）の新しい並列処理機能です。

Parallel.ForEach

Parallelクラスには、 ForEachメソッドが含まれています。これは、C＃の通常のforeachループのマルチスレッドバージョンです。通常のforeachと同様に、Parallel.ForEachは列挙可能なデータを繰り返し処理しますが、複数のスレッドを使用します。より一般的に使用されるParallel.ForEachオーバーロードの1つは次のとおりです。

public static ParallelLoopResult ForEach<TSource>( IEnumerable<TSource> source, Action<TSource> body)

Ienumerableは反復されるシーケンスを示し、Action本体は各要素に対して呼び出すデリゲートを設定します。 Parallel.ForEachオーバーロードの完全なリストは、ここにあります。

Plinq

Parallel.ForEachに関連するPLINQは、並列データ操作のプログラミングモデルです。ユーザーは、プロジェクション、フィルター、集計などを含む標準のオペレーターセットから操作を定義します。 Parallel.ForEachと同様に、 PLINQは入力シーケンスを異なるスレッドのパーツと処理要素に分割することにより、並列処理を実現します。

この記事では、並列処理に対するこれら2つのアプローチの違いを強調しています。 PLINQの代わりにParallel.ForEachを使用するのが最適な使用シナリオを理解し、その逆も同様です。

独立した操作

シーケンスの要素で長時間の計算を実行する必要があり、結果が独立している場合、Parallel.ForEachを使用することをお勧めします。 PLinqは、そのような操作には重すぎます。さらに、 Parallel.ForEachの最大スレッド数が示されます。つまり、 ThreadPoolのリソースが少なく、 ParallelOptions.MaxDegreeOfParallelismで指定されているよりも少ないスレッドが使用可能な場合、最適なスレッド数が使用され、実行時に増加します。 PLINQの場合、実行するスレッドの数は厳密に指定されています。

データ順序の保存を伴う並列操作

順序を維持するPLINQ

変換で入力順序を保持する必要がある場合は、 Parallel.ForEachよりもPLINQを使用する方が簡単であることがほとんどです。たとえば、出力時にRGBカラービデオフレームを白黒に変換する場合、フレームの順序は自然に保持されます。この場合、 PLINQとAsOrdered（）関数を使用することをお勧めします。AsOrdered（）関数は、PLINQの深さで入力シーケンスを分割し、変換を実行してから、正しい順序で結果を配置します。

 public static void GrayscaleTransformation(IEnumerable<Frame> Movie) { var ProcessedMovie = Movie .AsParallel() .AsOrdered() .Select(frame => ConvertToGrayscale(frame)); foreach (var grayscaleFrame in ProcessedMovie) { // Movie frames will be evaluated lazily } }

ここでParallel.ForEachを使用しないのはなぜですか？

些細な場合を除き、 Parallel.ForEachを使用してシリアルデータに並列操作を実装するには、大量のコードが必要です。この場合、 Foreach関数のオーバーロードを使用して、AsOrdered（）演算子の効果を繰り返すことができます。

 public static ParallelLoopResult ForEach<TSource >( IEnumerable<TSource> source, Action<TSource, ParallelLoopState,Int64>body)

Foreachのオーバーロードバージョンでは、現在の要素のインデックスパラメーターがデータアクションデリゲートに追加されています。これで、結果を同じインデックスの出力コレクションに書き込み、コストのかかる計算を並行して行い、最終的に正しい順序で出力シーケンスを取得できます。次の例は、 Parallel.ForEachで順序を維持する1つの方法を示しています。

 public static double [] PairwiseMultiply( double[] v1, double[] v2) { var length = Math.Min(v1.Length, v2.Lenth); double[] result = new double[length]; Parallel.ForEach(v1, (element, loopstate, elementIndex) => result[elementIndex] = element * v2[elementIndex]); return result; }

ただし、このアプローチの欠点はすぐに発見されます。入力シーケンスが配列ではなくIEnumerable型である場合、順序の保存を実装するには4つの方法があります。

最初のオプションは、IEnumerable.Count（）を呼び出すことで、O（n）かかります。要素の数がわかっている場合は、出力配列を作成して、指定したインデックスに結果を保存できます
2番目のオプションは、コレクションを具体化することです（たとえば、コレクションを配列に変換することにより）。大量のデータがある場合、この方法はあまり適していません。
3番目のオプションは、出力コレクションについて慎重に検討することです。出力コレクションはハッシュである場合があり、その場合、出力値を格納するために必要なメモリの量は、ハッシュの衝突を避けるために入力メモリの少なくとも2倍になります。大量のデータがある場合、ハッシュのデータ構造が非常に大きくなり、さらに、誤った共有とガベージコレクターのためにパフォーマンスが低下する可能性があります。
最後のオプションは、元のインデックスで結果を保存し、出力コレクションをソートするための独自のアルゴリズムを適用することです。

PLINQでは、ユーザーは単に順序の保存を要求し、クエリエンジンは結果の正しい順序を確保するためのすべてのルーチンの詳細を管理します。 PLINQ フレームワークを使用すると、 AsOrdered（）演算子でストリーミングデータを処理できます。つまり、PLINQは遅延マテリアライゼーションをサポートします。 PLINQでは、シーケンス全体を具体化することが最悪のソリューションです。上記の問題を簡単に回避し、 AsOrdered（）演算子を使用してデータの並列操作を実行できます。

並列ストリーミング

PLINQを使用してストリームを処理する

PLINQは、要求をストリーム上の要求として処理する機能を提供します。この機能は、次の理由で非常に価値があります。

1.結果はアレイで具体化されないため、メモリにデータを保存する際の冗長性はありません。
2.新しいデータを受信すると、計算の単一ストリームで結果を列挙できます。

証券の分析の例を続けて、証券のポートフォリオから各ペーパーのリスクを計算し、リスク分析の基準を満たす証券のみを提供し、フィルタリング結果に対していくつかの計算を実行するとします。 PLINQでは、コードは次のようになります。

 public static void AnalyzeStocks(IEnumerable<Stock> Stocks) { var StockRiskPortfolio = Stocks .AsParallel() .AsOrdered() .Select(stock => new { Stock = stock, Risk = ComputeRisk(stock)}) .Where(stockRisk => ExpensiveRiskAnalysis(stockRisk.Risk)); foreach (var stockRisk in StockRiskPortfolio) { SomeStockComputation(stockRisk.Risk); // StockRiskPortfolio will be a stream of results } }

この例では、要素はパーツ（パーティション）に分散され、複数のスレッドで処理されてから並べ替えられます。これらの手順は並行して実行されることを理解することが重要です。フィルタリング結果が表示されると、 foreachループのシングルスレッドコンシューマが計算を実行できます。 PLINQは、待ち時間ではなくパフォーマンス用に最適化されており、内部でバッファーを使用します。部分的な結果がすでに取得されていても、出力バッファが完全に飽和し、それ以上の処理が許可されなくなるまで出力バッファに残ることがあります。この状況は、出力バッファリングを指定できるPLINQ WithMergeOptions拡張メソッドを使用して修正できます。 WithMergeOptionsメソッドは、 ParallelMergeOptions列挙をパラメーターとして受け取り、単一のストリームで使用される最終結果をクエリが返す方法を指定できます。次のオプションが提供されます。

ParallelMergeOptions.NotBuffered-処理された各アイテムは、処理されるとすぐに各スレッドから返されることを示します。
ParallelMergeOptions.AutoBuffered-要素がバッファに収集されることを示します。バッファは定期的にコンシューマストリームに返されます
ParallelMergeOptions.FullyBuffered-出力シーケンスが完全にバッファリングされていることを示します。これにより、他のオプションを使用するよりも高速に結果を取得できますが、コンシューマスレッドは処理のために最初の要素を受信するまで長い時間待たなければなりません。

MSDNで入手可能なUsingMergeOptionsの例

Parallel.ForEachを選ばない理由

Parallel.ForEachの欠点を脇に置いて、シーケンスの順序を保持します。 Parallel.ForEachを使用したストリームでの順序なし計算の場合、コードは次のようになります。

 public static void AnalyzeStocks(IEnumerable<Stock> Stocks) { Parallel.ForEach(Stocks, stock => { var risk = ComputeRisk(stock); if(ExpensiveRiskAnalysis(risk) { // stream processing lock(myLock) { SomeStockComputation(risk) }; // store results } }

このコードはPLINQの例とほぼ同じですが、明示的なブロックとあまり洗練されていないコードを除きます。この状況では、 Parallel.ForeEachは結果をスレッドセーフスタイルで保存することを意味しますが、PLINQは結果を保存します。

結果を保存するには、3つの方法があります。最初の方法は、ストリームセーフでないコレクションに値を保存し、各レコードにロックを要求することです。 2つ目は、スレッドセーフなコレクションに保存することです。幸いなことに、.NET Framework 4は、 System.Collections.Concurrent名前空間にそのようなコレクションのセットを提供するため、自分で実装する必要はありません。 3番目の方法は、 スレッドローカルストレージでParallel.ForEachを使用することです。これについては後で説明します。これらの各メソッドでは、コレクションへの書き込みのサードパーティの影響を明示的に制御する必要がありますが、PLINQではこれらの操作から抽象化することができます。

2つのコレクションの操作

2つのコレクションの操作にPLINQを使用する

PLINQ ZIP演算子は、特に2つの異なるコレクションで並列計算を実行します。他のクエリと組み合わせることができるため、2つのコレクションを組み合わせる前に、各コレクションで複雑な操作を同時に実行できます。例：

 public static IEnumerable<T> Zipping<T>(IEnumerable<T> a, IEnumerable<T> b) { return a .AsParallel() .AsOrdered() .Select(element => ExpensiveComputation(element)) .Zip( b .AsParallel() .AsOrdered() .Select(element => DifferentExpensiveComputation(element)), (a_element, b_element) => Combine(a_element,b_element)); }

上記の例は、各データソースが異なる操作によって並列に処理され、両方のソースからの結果がZip演算子によって結合される方法を示しています。

Parallel.ForEachを選ばない理由

同様の操作を、インデックスを使用してParallel.ForEachオーバーロードを使用して実行できます。次に例を示します。

 public static IEnumerable<T> Zipping<T>(IEnumerable<T> a, IEnumerable<T> b) { var numElements = Math.Min(a.Count(), b.Count()); var result = new T[numElements]; Parallel.ForEach(a, (element, loopstate, index) => { var a_element = ExpensiveComputation(element); var b_element = DifferentExpensiveComputation(b.ElementAt(index)); result[index] = Combine(a_element, b_element); }); return result; }

ただし、Parallel.ForEachのアプリケーションには、データの順序を保持することで説明されている潜在的なトラップと欠点があります。欠点の1つは、コレクション全体を最後まで表示し、明示的なインデックス管理を行うことです。

スレッドローカル状態

Parallel.ForEachを使用してストリームのローカル状態にアクセスする

PLINQはデータの並列操作に対してより簡潔な手段を提供しますが、一部の処理シナリオはParallel.ForEachの使用により適しています。たとえば、ストリームのローカル状態をサポートする操作などです。対応するParallel.ForEachメソッドのシグネチャは次のようになります。

 public static ParallelLoopResult ForEach<TSource,TLocal>( IEnumerable<TSource> source, Func<TLocal> localInit, Func<TSource, ParallelLoopState, TLocal, TLocal> body, Action<TLocal> localFinally)

Aggregate演算子のオーバーロードがあり、ストリームのローカル状態へのアクセスを許可し、データ処理テンプレートが次元の減少として記述できる場合に使用できることに注意してください。次の例は、シーケンスから素数でない数を除外する方法を示しています。

 public static List<R> Filtering<T,R>(IEnumerable<T> source) { var results = new List<R>(); using (SemaphoreSlim sem = new SemaphoreSlim(1)) { Parallel.ForEach(source, () => new List<R>(), (element, loopstate, localStorage) => { bool filter = filterFunction(element); if (filter) localStorage.Add(element); return localStorage; }, (finalStorage) => { lock(myLock) { results.AddRange(finalStorage) }; }); } return results; }

このような機能はPLINQを使用するとはるかに簡単に実現できます。この例の目的は、 Parallel.ForEachとストリームのローカル状態を使用すると、同期コストを大幅に削減できることを示すことです。ただし、他のシナリオでは、ローカルフロー状態が絶対に必要になります;次の例はそのようなシナリオを示しています。

優秀なコンピューター科学者および数学者として、証券リスクを分析するための統計モデルを開発したと想像してください。このモデルは、他のすべてのリスクモデルを9に分割すると考えています。これを証明するには、株式市場に関する情報を持つサイトのデータが必要です。ただし、データシーケンスのロードは非常に長くなり、8コアコンピューターのボトルネックになります。 Parallel.ForEachを使用すると、 WebClientを使用してデータを並列にロードする簡単な方法ですが、ダウンロードされるたびに各スレッドがブロックされ、非同期I / Oの使用を改善できます。詳細についてはこちらをご覧ください。パフォーマンス上の理由から、 Parallel.ForEachを使用してURLのコレクションを反復処理し、データを並行してアップロードすることにしました。コードは次のようになります。

 public static void UnsafeDownloadUrls () { WebClient webclient = new WebClient(); Parallel.ForEach(urls, (url,loopstate,index) => { webclient.DownloadFile(url, filenames[index] + ".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); }

驚いたことに、実行時に例外が発生します。 「System.NotSupportedException-> WebClientは同時I / O操作をサポートしていません。」多くのスレッドが同じWebClientに同時にアクセスできないことに気付いたら、 WebClientを作成することにします。ダウンロードごとに。

 public static void BAD_DownloadUrls () { Parallel.ForEach(urls, (url,loopstate,index) => { WebClient webclient = new WebClient(); webclient.DownloadFile(url, filenames[index] + ".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); }

このコードにより、プログラムは100を超えるWebクライアントを作成できます;プログラムはWebClientでタイムアウト例外をスローします。コンピューターでサーバーオペレーティングシステムが実行されていないため、接続の最大数が制限されていることがわかります。次に、ストリームのローカル状態でParallel.ForEachを使用すると問題が解決すると推測できます。

 public static void downloadUrlsSafe() { Parallel.ForEach(urls, () => new WebClient(), (url, loopstate, index, webclient) => { webclient.DownloadFile(url, filenames[index]+".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); return webclient; }, (webclient) => { }); } }

この実装では、各データアクセス操作は互いに独立しています。同時に、アクセスポイントは独立しておらず、スレッドセーフでもありません。ローカルストリームストレージを使用すると、作成されるWebClientインスタンスの数が必要な数になり、各WebClientインスタンスがそれを作成したストリームに属していることを確認できます。

ここでPLINQが悪いのはなぜですか？

ThreadLocalおよびPLINQオブジェクトを使用して前の例を実装する場合、コードは次のとおりです。

 public static void downloadUrl() { var webclient = new ThreadLocal<WebClient>(()=> new WebClient ()); var res = urls .AsParallel() .ForAll( url => { webclient.Value.DownloadFile(url, host[url] +".dat")); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); }

実装は同じ目標を達成しますが、どのシナリオでも、 ThreadLocal <>を使用することは、対応するParallel.ForEachオーバーロードよりもかなり高価であることを理解することが重要です。このシナリオでは、インターネットからファイルをダウンロードするのにかかった時間と比較して、 ThreadLocal <>インスタンスを作成するコストは無視できることに注意してください。

終了操作

Parallel.ForEachを使用して操作を終了する

操作の実行の制御が不可欠な状況では、 Parallel.ForEachサイクルを終了すると、サイクルの本体内で計算を継続する必要があるかどうかの条件をチェックするのと同じ効果が得られることを理解することが重要です。 ParallelLoopStateを追跡できるParallel.ForEachオーバーロードの1つは次のようになります。

 public static ParallelLoopResult ForEach<TSource >( IEnumerable<TSource> source, Action<TSource, ParallelLoopState> body)

ParallelLoopStateは、以下に説明する2つの異なる方法でループ実行の中断をサポートします。

ParallelLoopState.Stop（）

Stop（）は、反復を停止する必要性についてループに通知します。 ParallelLoopState.IsStoppedプロパティを使用すると、各反復で他の反復がStop（）メソッドを呼び出したかどうかを判断できます。通常、 Stop（）メソッドは、ループが順序付けられていない検索を実行し、アイテムが見つかったらすぐに終了する必要がある場合に役立ちます。たとえば、コレクションにオブジェクトが存在するかどうかを確認する場合、コードは次のようになります。

 public static boolean FindAny<T,T>(IEnumerable<T> TSpace, T match) where T: IEqualityComparer<T> { var matchFound = false; Parallel.ForEach(TSpace, (curValue, loopstate) => { if (curValue.Equals(match) ) { matchFound = true; loopstate.Stop(); } }); return matchFound; }

PLINQを使用して機能を実現することもできます。この例では、ParallelLoopState.Stop（）を使用して実行フローを制御する方法を示します。

ParallelLoopState.Break（）

Break（）は、現在の要素の前の要素を処理する必要があることをループに通知しますが、反復の後続の要素については停止する必要があります。低い反復値は、 ParallelLoopState.LowestBreakIterationプロパティから取得できます。 Break（）は通常、順序付けされたデータを検索する場合に役立ちます。つまり、データ処理の必要性には一定の基準があります。たとえば、一致するオブジェクトの下位インデックスを見つける必要がある非一意の要素を含むシーケンスの場合、コードは次のようになります。

 public static int FindLowestIndex<T,T>(IEnumerable<T> TSpace, T match) where T: IEqualityComparer<T> { var loopResult = Parallel.ForEach(source, (curValue, loopState, curIndex) => { if (curValue.Equals(match)) { loopState.Break(); } }); var matchedIndex = loopResult.LowestBreakIteration; return matchedIndex.HasValue ? matchedIndex : -1; }

この例では、オブジェクトが見つかるまでループが実行されます。Break（）シグナルは、見つかったオブジェクトよりも低いインデックスを持つ要素のみを処理することを意味します。別の一致するインスタンスが見つかった場合、Break（）信号が再び受信されます。要素が見つかるまで繰り返されます。オブジェクトが見つかった場合、LowestBreakIterationフィールドは一致するオブジェクトの最初のインデックスを指します。

PLINQを使用しない理由

PLINQはクエリ実行の終了をサポートしていますが、PLINQとParallel.ForEachの終了メカニズムの違いは重要です。 PLINQリクエストを終了するには、ここで説明するように、リクエストにキャンセルトークンを提供する必要があります。 C Parallel.ForEach終了フラグは、各反復でポーリングされます。 PLINQの場合、キャンセルされたリクエストに依存してすぐに停止することはできません。

おわりに

Parallel.ForEachとPLINQは、作業のメカニズムに深く浸る必要なく、アプリケーションに並行性をすばやく導入するための強力なツールです。ただし、特定の問題を解決するための適切なツールを選択するには、この記事で説明する違いとヒントを覚えておいてください。

便利なリンク：

C＃のスレッド化

RSDN：C＃でストリームを操作します。並行プログラミング

.NET Frameworkを使用した並列プログラミングのMicrosoftサンプル

Parallel.ForEachを使用する場合とPLINQを使用する場合

はじめに