タスクの並列化。「完全な並列化」の場合。パート2

.NET 4の推奨ソリューション

これは、理想的なサイクルの並列化に関する記事の第2部です。最初の部分では、この場合に発生する問題と、その解決策への一般的なアプローチを検討しました。ここでは、これらのタスクをサポートするために.NET 4.0が提供する特定のライブラリコンポーネントについて説明します。

「理想的な」ループを並列化するために、次のオプションが提供されています。

For（）、ForEach（）メソッドを持つSystem.Threading.Tasks.Parallelクラス
AsParallel（）拡張メソッドを使用した並列LINQ。

Parallel.ForおよびParallel.ForEachメソッド

まず、Parallelクラスとそのループ編成メソッドを見てみましょう。

私たちが検討している最初のメソッドにはシグネチャがあります（これは基本的なもので、多くのオーバーロードの1つです）：

public static ParallelLoopResult For ( int fromInclusive, int toExclusive, Action < int > body ) ;

戻り値の型を除いて、多くの点で記事の最初の部分の実験メソッドのシグネチャに似ています。

2番目の「基本構成」は次のようになります。

public static ParallelLoopResult ForEach < TSource > （ IEnumerable < TSource > source、Action < TSource > body ） ;

Parallel.ForおよびParallel.ForEachメソッドができることを検討してください。

例外処理。 いずれかの反復で例外が発生すると、新しい反復は開始されなくなります。それにもかかわらず、すでに開始している人は作業を終了でき、すでに完了すると、すべての例外（最終反復での初期および可能な後続の例外）がAggregateExceptionタイプの1つの例外に集約され、スローされます。

サイクルの早期中断。 Forメソッドの対応するオーバーロードは、コンテキストオブジェクトでStopまたはBreakメソッドを呼び出すことで実行を中断する機能を提供します（各反復に渡されます）。

StopとBreakの違いは、Stopは一般に新しい反復の開始を停止する必要があることを通知することと、Break-Breakが呼び出されたものに続いて（順番に）新しい反復の開始を停止することを示します。つまり、5回目の繰り返しでBreakを呼び出すと、並列性にもかかわらず、1回目から4回目の繰り返しが実行されることが保証されます。また、Stopが5回の繰り返しで呼び出されたときに、繰り返し4がまだ開始されていない場合、開始されません。

StopまたはBreakの呼び出し時に既に実行されていた現在の反復は、サイクルの中断のステータスを確認し、サイクル全体の中断について学習する場合は事前に終了できます。これを行うために、彼らはコンテキストオブジェクトの対応するプロパティをチェックすることができます：IsStoppedとLowestBreakIteration。

ストリームレベルのデータストレージのサポート 。メソッドのオーバーロードによっては、各ストリームに対してローカルなデータウェアハウスに中間計算結果を保存できます。これにより、たとえば、結果を集計するときに、最初に個別のスレッドの結果を集計し、次にすべての作業スレッド間で結果を集計できます。これにより、計算プロセスでの不要な同期コストが排除されます。

並列処理のレベルを構成する機能 。実行に使用されるスレッドの最大数を指定できます。

ネスト呼び出しのサポート 。スレッドプールを使用するため、ForメソッドとForEachメソッドのネストされた呼び出し、およびそれらの並列実行の両方で、過度のマルチスレッドは発生しません。

スレッド数の動的な変更 。 Parallel.Forは、読み込み時間を変更し、作業の一部の要素が他の要素よりも多くの計算を必要とする可能性があるという事実を考慮して設計されました。したがって、使用されるスレッドの数は、メソッドの操作中に変化する可能性があります。

洗練されたスレッド読み込み管理 。この方法では、多数のさまざまな要因を考慮したフローの負荷分散のロジックを実装しました。また、ブロックのサイズ（処理用の「データ部分」）は操作中に増加します。これにより、少数の反復でより良い負荷分散を得ることができます（同時に「より重い」と仮定されます）。多くの反復。

外部からループの実行をキャンセルするためのサポート。 これを行うには、クラスCancellationTokenSourceを使用します。ループを開始するときに、Tokenプロパティをループに渡す必要があります。その後、外部からループのキャンセルを要求するには、CancellationTokenSourceオブジェクトでCancel（）メソッドを呼び出すだけです。ところで、現在の反復では、キャンセルステータスを確認できるため、サイクル全体がキャンセルされたことがわかった場合、事前に「自発的に」終了することができます。

並列並列

.NET 3.5で拡張機能として利用可能なParallel LINQは、.NET 4.0でSystem.Coreですぐに利用できます。それを使用するアイデアは非常に簡単です。.AsParallel（）呼び出しをLINQクエリチェーンに追加し、チェーン内の後続のすべての呼び出しは並行して実行されます。例：

var doubled = new [ ] { 1、2、3、4 } 。 AsParallel （）。選択（ i => i * 2 ） ;

並列処理されるため、結果のコレクション内の要素の順序が任意であることは明らかではない場合があります。コレクション内の要素の順序を保持するには、最初のフラグメントを次のように変更します。

var doubled = new [ ] { 1、2、3、4 } 。 AsParallel （）。 AsOrdered （）。選択（ i => i * 2 ） ;

処理は引き続き並行して行われますが、「最初の」要素をカウントするスレッドを最初に待機し、結果を取得してから、「最後の」を読み取るスレッドから結果を取得する必要があるため、集約ステージ（各スレッドからの結果の収集）に時間がかかる場合があります要素。

PLINQを使用するときは、追加のオーバーヘッドがあることに注意することが重要です。

フローによる元の列挙の着信要素の分布
共通のコレクションへの計算された要素の集約。

さらに、フローによる作業の分散から逃れることはできませんが、タスクが新しいコレクションを取得するのではなく、既存のコレクションの要素を単純に処理する場合は、集約を遅らせることもできます。

AsParallel（）メソッドはIEnumerableではなくParallelEnumerableを返します。その後、他のすべてのLINQメソッド（Select、Whereなど）も同じ型を返します。 ParallelEnumarableがチェーンで送信される限り、各フローの計算結果は不必要に集約されず、パイプラインが構築されます。

パイプラインが中断すると、並列処理が停止し、データが単一の結果コレクションに集約されることを理解することが重要です。たとえば、次のコードは誤って想定される可能性があるため機能しません。

リスト< InputData > inputData = ... ;
foreach （ inputDataのvar o AsParallel （）。 Select （ i => new OutputData （ i ）））
{
ProcessOutput （ o ） ;
}

並行して、ここではOutputDataオブジェクトの作成のみが実行されます。その後、すべてのフローがこれらのオブジェクトを収集し、結果のコレクションを形成します。これにより、各要素のProcessOutput（）の呼び出しが順次不要になります

ここで不要な集計手順を回避するには、ParallelEnumarable.ForAll（）メソッドを使用できます。

リスト< InputData > inputData = ... ;
inputData。 AsParallel （）。（ i => new OutputData （ i ））を選択します。 ForAll （ o =>
{
ProcessOutput （ o ） ;
} ） ;

この場合、「新しいOutputData（i）」ステージの後、「ProcessOutput（o）」ステージも、それらの間の集約ステージなしで並行して実行されます。

「inputData.AsParallel（）.Select（i => new OutputData（i））」に対してParallel.ForEach（）を呼び出すと、通常のforeach：in Parallel.ForEach（）での最初の例と同じ欠点があることに注意してください。 IEnumerableは、ParallelEnumerableではなく渡されます-したがって、コレクションをParallel.ForEach（）に渡す前に、集計されます。これを回避するには、ParallelEnumerable.ForAll（）メソッドを使用します。このメソッドはこの場合に使用する必要があります。

一般的な問題とエラー

これらのコンポーネントを使用するときは、考えられる問題を考慮してください。

ネイティブスレッドセーフティ

まず、Parallel.ForEachを単独で使用してもコードがスレッドセーフにならないことを理解する必要があります-反復が互いに独立していることを確認するか、依存している場合は、共有リソースでスレッドセーフな操作を明確に提供する必要があります。

また、Parallel.Forを使用する場合、トップダウンサイクルおよびカウンターの非標準（1に等しくない）インクリメントのサイクルは直接サポートされません。元のアルゴリズムがこの方法で記述されている場合、多くの場合、非標準のループは反復間の依存関係（前の段階で計算された配列の前の要素など）のために記述されるため、慎重に分析する必要があります。

自転車本体寸法

Parallelクラスの使用には、少なくともオーバーヘッドが伴います

デリゲートを呼び出してループの本体を実行し、
タスクを配布するときにスレッド間で同期するため。

サイクルの本体が十分に長い場合、これらの追加コストは小さな役割を果たします。ただし、「i = i * i」のような非常に単純なものを並列化すると、この場合のオーバーヘッドは有用な作業を上回ります。この欠点を取り除くには、サイクルの本体を「拡大」する必要があります。 1つではなく、多数の反復を含めることでこれを行うのは非常に簡単です。

入力シーケンスをブロックのセットに明示的に分割して手動でこれを実行し、このセットで並列サイクルを開始します。この並列サイクルの本体では、各ブロックはすでに順次処理されています。ただし、ここでこれらのブロックの数を決定する必要があります。これをライブラリに割り当て、特別に作成されたクラスを使用して、入力シーケンスSystem.Concurrent.Collections.Partitionerからサブセットを作成できます。

それを使用してサイクル

for （ int i = 0 ; i <長さ; i ++ ）
result [ i ] = i * i ;

素朴なバージョンの代わりに：

平行（ from、to、i =>
{
result [ i ] = i * i ;
} ） ;

このように効果的に並列化できます：

平行 ForEach （パーティショナー。作成（ from、to ）、範囲=>
{
for （ int i = range。Item1 ; i < range。Item2 ; i ++ ）
{
result [ i ] = i * i ;
}
} ） ;

Partitioner.Create（from、to）は、ループ本体内の各ブロックを順番に処理して、並行して通過するブロックの同じセットを作成します。したがって、長い本体に並列サイクルを提供し、オーバーヘッドをより多くの有用な作業に分配します。

入れ子ループ処理

ネストされたループを処理するとき、どれだけ深く並列化する価値があるのかという疑問が生じます。長方形の画像を処理する例を見てみましょう。

for （ int y = 0 ; y < screenHeight ; y ++ ）
{
int stride = y * screenWidth ;
for （ int x = 0 ; x < screenWidth ; x ++ ）
{
rgb [ x + stride ] = calcColor （ x、y ） ; //色を計算します
}
} ;

外部ループのみをParalllel.Forの呼び出し、またはその両方で置き換えることができます。両方のサイクルを平行にすると、内側のボディが小さくなりすぎますか？内部の一貫性を維持する場合、すべてのコアに作業をロードすることになりますか？

これらの質問に対する答えは、パフォーマンステストのみを提供できます。色の計算があまり複雑でない場合は、内部ループの一貫性を保つ必要があります。しかし、幅は広いが低いイメージで作業する場合、外部ループを並列化するだけではすべてのプロセッサコアがロードされない可能性があります。

別の方法として、2つのネストされたループを1つに展開し、すでに並列化することもできます。

int totalPixels = screenHeight * screenWidth ;
平行（ 0 、totalPixelsの場合、i =>
{
int y = i / screenWidth、x = i ％ screenWidth ;
rgb [ i ] = calcColor （ x、y ） ;
} ） ;

calcColor（）が単純すぎる場合、ループの本体を拡大するには、前の例のようにPartitionerクラスを使用できます。

int totalPixels = screenHeight * screenWidth ;
平行 ForEach （パーティショナー。Create （ 0 、totalPixels ）、範囲=>
{
for （ int i = range。Item1 ; i < range。Item2 ; i ++ ）
{
int y = i / screenWidth、x = i ％ screenWidth ;
rgb [ i ] = calcColor （ x、y ） ;
}
} ） ;

安全でないIList実装

Parallel.ForEachとPLINQはどちらも、入力でIEnumerableを受け入れますが、同時に、渡されたコレクションで動作するための最速のインターフェイスを見つけようとします。特に、並列化の場合、IListインターフェイスは、IEnumerableよりも優れています。これは、任意の要素へのランダムアクセスのためのインデクサーがあるためです。したがって、転送されたコレクションに対してIListが定義されている場合、このインターフェイスを介して作業が行われます。これにより同期のコストは下がりますが、ライブラリコードはインデクサーのスレッドセーフな実装に依存しています 。

使用するコレクションがスレッドセーフなインデクサーを提供しない場合（要素がインデックス作成のために複雑な形式で格納されている場合や遅延ロードされている場合もよくあります）、IListを使用する必要はなく、IEnumerableに制限することをシステムに明示的に伝える必要があります。

これを行うには、2つのオプションが適しています。 1つ目は、コレクションのSystem.Collections.Concurrent.Partitionerを作成することです。

// IList <T>は、コレクションでサポートされている場合、ここで使用できます
IEnumerable < T > source = ... ;
平行 ForEach （ source、item => { /*...*/ } ） ;
//そして、ここで使用されることが保証されているIEnumerable <T>
IEnumerable < T > source = ... ;
平行 ForEach （パーティショナー。作成（ソース）、アイテム=> { /*...*/ } ） ;

2番目の方法は明らかであり、通常のLINQでよく知られています。コレクション ".Select（item => item）"を呼び出すだけです。 Parallel.ForとPLINQの両方に適しています。

//これは、IEnumerable <T>の使用も保証されます
ソース。（ i => i ）を選択します。 AsParallel （）。 Select （ i => { /*...*/ } ） ;

ソースコレクションのスレッドアフィニティ

Parallel.ForEachおよびPLINQを実行すると、各ワークフロー自体がソースコレクションでMoveNext（）を呼び出します。たとえば、Windows FormsまたはWPFでUIコンポーネントを操作する場合など、特定の1つのストリームからのみアクセスできるコレクション（「ストリームとの親和性」）である場合、これらのメカニズムを直接使用することはできません。

このようなコレクションの並列処理を保証するには、.NET 4.0のBlockingCollectionクラスがメインの「ブリック」であるProducer-Consumerテンプレートを使用する必要があります。これについては後で別途説明します。元の文書の53ページにあるProducer-Consumerセクションで、英語で詳細を読むことができます。

PSこの記事は、「パラレルプログラミングのパターン：.NET FRAMEWORK 4およびVISUAL C＃を使用したパラレルパターンの理解と適用」という書籍の影響を受けて書かれたもので、処理を伴う無料翻訳と見なすことができます。

おわりに

このツアーが、コードを並列化する際のミスを減らすのに役立つことを願っています。しかし、とにかく、成功したデバッグ、そして幸せな休日:)

未来のために

興味があれば、4フレームワークで並列処理のトピックを続けることができます。次のようになります。

依存関係のある並列コンピューティングのタスククラスとインフラストラクチャ、
新しいスレッドセーフコレクション
怠inessと並行性
共有データと同期。

これのどれが面白いか書いてください:)

All Articles

タスクの並列化。 「完全な並列化」の場合。 パート2