問題タブ [parallel.foreach]

TPLを使い始めたばかりで、Webサービスへの複数の呼び出しを並行して実行したいと思います。私が集めることができるものから、私はこれを行う2つの方法を見ます。

どちらかParallel.ForEach：

またはTask<T>：

構文が正しいかどうかを無視して、これらは同等ですか？

それらは同じ結果を生み出しますか？そうでない場合、なぜですか？そしてどちらが好ましいですか？

メトロアプリでは、いくつかのWCF呼び出しを実行する必要があります。かなりの数の呼び出しが行われる必要があるため、並列ループで呼び出す必要があります。問題は、WCF呼び出しがすべて完了する前に並列ループが終了することです。

期待どおりに機能するように、これをどのようにリファクタリングしますか？

内部に Linq クエリを含む foreach ループがあります。

foreach を Parallel.ForEach に変更するまでは、すべてうまくいきます。

そして、私が得る例外は次のとおりです。

System.AccessViolationException は処理されませんでした Message=保護されたメモリを読み書きしようとしました。これは多くの場合、他のメモリが破損していることを示しています。Source=Oracle.DataAccess StackTrace: at Oracle.DataAccess.Client.OpsCon.Open(IntPtr& opsConCtx, IntPtr& opsErrCtx, OpoConValCtx* pOpoConValCtx, OpoConRefCtx& pOpoConRefCtx) at Oracle.DataAccess.Client.ConnectionDispenser.Open(OpoConCtx opoConCtx) at Oracle.DataAccess. Client.OracleConnection.Open() で System.Data.EntityClient.EntityConnection.OpenStoreConnectionIf (ブール値の openCondition、DbConnection storeConnectionToOpen、DbConnection originalConnection、String exceptionCode、String attemptsOperation、Boolean& closeStoreConnectionOnFailure) で System.Data.EntityClient.EntityConnection.Open() で System.Data.EntityClient.EntityConnection.Open() 。データ。1.GetResults(Nullable1 forMergeOption) System.Data.Objects.ObjectQuery で1.System.Collections.Generic.IEnumerable<T>.GetEnumerator() at System.Collections.Generic.List1..ctor(IEnumerable 1 collection) at System.Linq.Enumerable.ToList[TSource](IEnumerable1 source) ChangeOrder.Program.<>c_ DisplayClass19.b _16(SF_CHANGE_ORDER_ITEM t) で C:\VS_apps\PMConsole\PMC Tools\ChangeOrderExecution\Program .cs: System.Threading.Tasks.Parallel.<>c_ DisplayClass2d の 220 行目2.<ForEachWorker>b__23(Int32 i) at System.Threading.Tasks.Parallel.<>c__DisplayClassfSystem.Threading.Tasks.Task.InnerInvoke() の System.Threading.Tasks.Task.InnerInvokeWithArg(Task childTask) の 1.b _c( ) System.Threading.Tasks.Task.<>c_ DisplayClass7.bでSystem.Threading.Tasks.Task.ExecuteSelfReplicating(タスク ルート) の _6(オブジェクト) System.Threading.Tasks.Task.Task.ExecutionContextCallback(オブジェクト obj) の System.Threading.ExecutionContext .Run(ExecutionContext executionContext、ContextCallback コールバック、オブジェクト状態、Boolean ignoreSyncCtx) System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot) at System.Threading.Tasks.Task.ExecuteEntry(Boolean bPreventDoubleExecution) at System.Threading.Tasks. ThreadPoolTask​​Scheduler.TryExecuteTaskInline (タスク タスク、ブール値の taskWasPreviouslyQueued) の System.Threading.Tasks.TaskScheduler.TryRunInline (タスク タスク、ブール値の taskWasPreviouslyQueued、オブジェクト threadStatics) の System.Threading.Tasks.Task.System.Threading.Tasks.Task.RunSynchronously(TaskScheduler スケジューラ) の InternalRunSynchronously(TaskScheduler スケジューラ) System.Threading.Tasks.Parallel.ForWorker[TLocal](Int32 fromInclusive、Int32 toExclusive、ParallelOptions parallelOptions、Action1 body, Action2 bodyWithState、Func 4 bodyWithLocal, Func1 localInit、Action 1 localFinally) at System.Threading.Tasks.Parallel.ForEachWorker[TSource,TLocal](IList1 list、ParallelOptions parallelOptions、Action 1 body, Action2 bodyWithState、Action 3 bodyWithStateAndIndex, Func4 bodyWithStateAndLocal、Func 5 bodyWithEverything, Func1 localInit、Action 1 localFinally) at System.Threading.Tasks.Parallel.ForEachWorker[TSource,TLocal](IEnumerable1 source、ParallelOptions parallelOptions、Action 1 body, Action2 bodyWithState、Action 3 bodyWithStateAndIndex, Func4 bodyWithStateAndLocal、Func 5 bodyWithEverything, Func1 localInit、Action1 localFinally) at System.Threading.Tasks.Parallel.ForEach[TSource](IEnumerableC:\VS_apps\PMConsole\PMC Tools\ChangeOrderExecution\Program.cs:line 216 の ChangeOrder.Program.Main( String[] args) in C:\VS_apps\PMConsole\PMC Tools\ChangeOrderExecution\Program.cs: System.AppDomain._nExecuteAssembly(RuntimeAssembly assembly, String[] args) at System.AppDomain.ExecuteAssembly(String assemblyFile, Evidence) の行 1373 assemblySecurity, String[] args) で Microsoft.VisualStudio.HostingProcess.HostProc.RunUsersAssembly() で System.Threading.ThreadHelper.ThreadStart_Context(オブジェクト状態) で System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback コールバック, オブジェクト状態,System.Threading.ThreadHelper.ThreadStart() で System.Threading.ExecutionContext.Run(ExecutionContext executionContext、ContextCallback コールバック、オブジェクト状態) で Boolean ignoreSyncCtx) InnerException:

情報を取得しているだけなので、何をロックする必要があるのか​​ わかりません（ROだけですよね？）。

「AsParallel」を追加することを考えましたが、それはクエリをそれ自体に対して並列に実行させるだけの PLINQ ディレクティブであると理解しています。

Parallel.ForEach ループ内で Linq クエリを実行している例を見つけることができないため、自分が何をしているのかさえわかりません。

これを、機能するいくつかのことをどのように理解するかについての基本的な説明から始め、tldr ですべてを締めくくります。人々が単に私がここで抱えている実際の質問に到達したい場合. ここでの私の理解が間違っている場合は、訂正してください。

TPL は Task Parallel Library の略で、開発者が使いやすいようにスレッド化をさらに簡素化しようとする .NET 4.0 の答えです。慣れていない場合は (非常に基本的なレベルで)、新しい Task オブジェクトを開始し、それにデリゲートを渡します。デリゲートは、スレッド プールから取得したバックグラウンド スレッドで実行されます (真にタスクを作成するのではなく、スレッド プールを使用することにより)。新しいスレッドを作成して破棄する代わりに、これらの既存のスレッドを使用することで、時間とリソースを節約できます)。

私が理解していることから、C# の Parallel.ForEach コマンドは、実行することになっているデリゲートごとに (おそらくスレッド プールから) 新しいスレッドを生成します。コンパイラーが、より効率的になるのに十分な速さで反復が行われると判断した場合。

私の目標に関する最も関連性の高い背景情報:

タスクから開始して、残りのプログラムと同時に実行する簡単なプログラムを作成しようとしています。このタスクでは、Parallel.ForEach が 3 回の「反復」で実行されます。合計で、プログラムは合計 5 つのスレッド (最大) を実行すると予想されます。メイン スレッドに 1 つ、実際のタスクに 1 つ、Parallel.ForEach に最大 3 つです。各スレッドには、達成する独自の目標があります (ただし、Parallel.ForEach はすべて、計算する関連する itemNumber に異なる値を持つ同じ目標を持っています。メイン スレッドがすべての目標を完了すると、Task.Wait() を使用して待機します)。 Parallel.ForEach も終了するのを待つ終了タスクで、値が使用され、検証されます。

tldr; 実際の質問:

前述のアイデアが実行されると、Parallel.ForEach は、予想される数の 2 倍の SynchronizationContexts (本質的には別のスレッドである TPL オブジェクト) を初期化し、それらすべてを実行しているように見えますが、予想される数だけ待機しています。Parallel.ForEach().Wait() コマンドは実行中のスレッドの予想数で終了するため、Task もすべてが完了したと見なして終了します。その後、メイン プログラムは Task が終了したことを検出し、現在バックグラウンド スレッドが実行されていないことを確認すると、残りの Parallel.ForEach() がまだ終了していないためにエラーがスローされることがあります。

各 SynchronizationContext の post 呼び出し (Async メソッドのキッカー) でデバッグ ウィンドウに出力することによって、スレッドの量が私が述べたものと一致することが確認されました。各スレッドは、タスクの終了時に破棄される予定のメイン スレッド オブジェクトによっても参照されますが、実際には作成されるとは想定されていなかった未終了のスレッドが原因で参照がまだ残っているため、破棄は適切に行われません。

上記の例では、メイン スレッド、backgroundTask タスク、および 8 つの Parallel.ForEach スレッドが存在することになり、そのうちの最後の 9 つは SynchronizationContexts で作成されます。

私のカスタムの SynchronizationContext でオーバーライドされる唯一のメソッドは post で、次のとおりです。

私が信じている TaskScheduler は、必要な基本的なことだけを行っています。

タスクファクトリー:

なぜこれが起こっているのかについてのアイデアはありますか?

今日は、LinqクエリとPLinqクエリのパフォーマンスへの影響をテストしました。したがって、msdnの方法に関する記事を使用しました：PLINQクエリのパフォーマンスを測定します。

テスト環境：Win7 Enterprise上のLinqPad4、64ビット、8 GB RAM、I7-2600（8コア）

1つのコア（測定1）でのクエリが、並列クエリよりも高速である理由を理解しましたが、説明できません。並列タスクのメリットを享受するには、選択したデリゲートをさらに追加する必要がありますか？

しかし今、結果は：

1.実行：60000の列挙可能な範囲で：

2.実行：600000000の列挙可能な範囲で：

興味深い事実：テストスクリプトを実行する前にガベージコレクターを使用すると、測定4の時間が大幅に増加します。

3.実行：600000000の列挙可能な範囲とガベージコレクター（LinqPadから）：

結論として、方法1は、選択デリゲートが増えるときに、小さな選択クエリと方法5を実行するのに最も適したオプションであると言えますか？

タスク並列ライブラリを使用Parallel.ForEach()して、いくつかのバックグラウンド タスクを実行しています。Phil Haack の WebBackGrounder アプリケーションに出会いました。私はそれを実装したかったのですが、DoWorkメソッドはJobHostサポートのみThreading.Taskでアクションはサポートしていません。コンテキストでそれを使用する方法についてのアイデアはParallel.ForEach非常に役立ちます。または、アクションを処理する独自のクラスを作成する必要があります。

My Parallel 各メソッドは次のとおりです。

私はスレッドを持っています、それを「解析スレッド」と呼びます。

私はこのスレッドでいくつかの計算を実行しますが、最後の計算にはより多くの並列計算が含まれます。

問題は、これらの非常に長く計算コストの高い非同期foreach操作が終了すると（約30分）、「スレッドの解析」が再開されないことです。私のGUIはまだ応答しますが、「解析スレッド」で実行し続けることになっているRaiseDirectoryParsed関数が呼び出されることはありません。私はこの時点までプログラムをデバッグしましたが、この状況で何をすべきかについてかなり困惑しています。

たくさんのファイルをpdfに変換する小さなプログラムを書きました。

プログラムは次のことを行います。

• フォルダからオブジェクトの配列を取得する FileInfo（10'000ドキュメント）
• それぞれについてFileInfo
  • FileInfo.CopyTo()、を使用してバックアップコピーを作成します
  • いくつかのAsposeライブラリを使用してドキュメントをPDFに変換します
  • 変換後、PDFを新しい宛先にコピーします
  • foreach内で、イベントが発生し、WinForm UIによって処理されて、進行状況が表示されます

ドキュメントのサイズによっては、ドキュメントの変換に0〜3秒かかる場合があります。それが完璧な候補になると思ったParallel.ForEachので、プログラムを変更しました。

ただし、変換には、従来のforeachでは1時間ではなく、1.5時間かかりましたParallel.Foreach（私が試したサーバーには、2つのIntel Xeon Procsがあります）。

何を間違えたのですか、それともパフォーマンスを向上させるために何を考慮する必要がありますか？

私は少し混乱していParallel.ForEachます。
それは何でParallel.ForEachあり、それは正確に何をしますか？
MSDNリンクを参照しないでください。

簡単な例を次に示します。

この例をで書き直すにはどうすればよいParallel.ForEachですか？

32 スレッドで並列 foreach ループを実行しようとしています。

この foreach ループは次を呼び出しています。

並列 foreach ループは、約 65000 項目のリストで構成されています。

ある時点で、約 6000 ～ 6700 項目で、WorkflowInvoker.Invoke(new MyWorkflow(), myInputDictionary) の呼び出しにより、次のメッセージとともに AccessViolationException がスローされます。 ."

WorkflowInvoker のインスタンスを作成し、MyWorkflow のインスタンスを新しく作成した WorkflowInvoker に直接渡し、クラスで Invoke メソッドを呼び出すなど、さまざまなことを試しました。また、WorkflowInvoker.Invoke を静的メソッドとして呼び出すだけですが、違いはありません。アプリケーションは毎回同じ時点でクラッシュします。スレッドがあるため、問題の原因を正確にデバッグして特定することは困難です。

もちろん、通常の foreach ループで呼び出しを行っている場合、問題は発生しません。

最初は、仮想マシンのメモリが不足しているのではないかと考えました。これは、アプリケーションが使用可能なすべてのメモリを消費し始め、合計メモリの 85% (4GB) の制限に達し、その後クラッシュしたように見えたためです。そのため、メモリ サイズを 1GB に変更して、ループの早い段階で例外を誘発できるかどうかを確認しましたが、代わりに Windows がガベージ コレクションを開始したため、最終的には 6000 ～ 6700 アイテム付近の同じポイントで例外が発生しました。

私の次の動きはどうあるべきかについての提案は、大歓迎です;-)

-- 編集 - 完全なスタック トレースを追加 --

-- 余分な編集 - 問題のメソッドを追加 - ビジネスロジック名を削除するために少し難読化 --