問題タブ [tpl-dataflow]

にリンクするを使用しBroadcastBlockます。ただし、コードにバグが見つかりません。アイテムをビアに投稿し、アイテムを印刷すると、一部が欠落しており、不思議なことに重複していることがわかります。処理されたアイテムの順序を追跡しているため、アイテムが欠落しているときに、前のアイテムまたは後続のアイテムが重複していることがわかりました（経由で同じアイテムに供給されます）。何が起こっているのかわかりません。関連する私のコードスニペットのいくつかを次に示します。TPL DataflowActionBlocksBroadcastBlockSendAsyncActionBlocksActionblockLinkTo

ここでのインスタンス化BroadCastBlock:

アクションブロックにリンクする方法は次のとおりです。

これは私がアイテム（見積もり）をに提出する方法ですBroadCastBlock：

問題は、すべてを介して送信する直前に追跡情報を印刷すると、SendAsync完全に正常に見えることです。ActionBlockそのリンクに同じ情報を印刷すると、BroadCastBlockまれにエラーが発生します。アイテムが ( APIBroadCastBlockを介して) 非常に迅速にフィードされると、アイテムの欠落/重複が発生することに気付きました。EventHandler

私が間違っていることは何ですか？

バックグラウンド

定期的なデータ ダンプ (XML ファイル) を受け取り、それらを Entity Framework 5 (Code First) を使用して既存のデータベースにインポートするアプリケーションがあります。エンティティに既に存在するビジネス ルールを適用する必要があるため、インポートは BULK INSERT または BCP ではなく EF5 を介して行われます。

処理は、アプリケーション自体の CPU バウンドのようです (非常に高速で、書き込みキャッシュが有効なディスク IO サブシステムは、プロセス全体でほぼゼロのディスク待機時間を示し、SQL Server は 8% ～ 10% の CPU 時間しか示しません)。

効率を改善するために、TPL Dataflowとコンポーネントを使用してパイプラインを構築し、次のことを行いました。

これを行うことでパフォーマンスが大幅に向上しますが、CPU が約 60% を超えることはありません。

分析

ある種のリソース競合を疑って、VS2012 プロファイラーのリソース競合データ (同時実行)モードを使用してプロセスを実行しました。

プロファイラーは、 Handle 2というラベルの付いたリソースに対して 52% の競合を示しています。掘り下げてみると、 Handle 2で最も競合を引き起こしているメソッドは次のとおりであることがわかります。

2 位は、SaveChanges() の競合の約 40% です。

質問

• Handle 2が実際に何であるか(TPL の一部、EF の一部など) を知るにはどうすればよいですか?
• EF は、個別のスレッドから個別の DbContext インスタンスへの呼び出しを調整しますか? 彼らが争っている共有リソースがあるようです。
• この場合、並列処理を改善するためにできることはありますか?

アップデート

問題の実行では、SaveChanges を呼び出すタスクの最大並列度は 12 に設定されています (以前の実行で Unbounded を含むさまざまな値を試しました)。

更新 2

Microsoft の EF チームがフィードバックを提供しています。要約については、私の回答を参照してください。

TransformBlock<Uri, string>（それ自体がの実装であるIPropagatorBlock<Uri, string>）から始めて、（これはWebクローラーです）Uriのコンテンツを取得するとします。string

文字列のコンテンツを取得したら、リンクを解析します。ページには複数のリンクを含めることができるため、aを使用しTransformManyBlock<string, Uri>て単一の結果（コンテンツ）を多くのリンクにマップします。

パーサーの鍵は、空のシーケンスを返すことができることです。これは、解析する必要のあるアイテムがこれ以上ないことを示します。

ただし、これはツリーのブランチ（またはWebのセクション）の場合のみです。

次に、ダウンローダーをパーサーにリンクしてから、次のようにダウンローダーに戻します。

今、私はすべてをブロックの外側で停止させることができることを知っています（それらの1つを呼び出すことによって）が、ブロックの内側Completeからそれが完了したことをどのように知らせることができますか？

それとも、どういうわけか自分でこの状態を管理する必要がありますか？

すべてのコンテンツがダウンロードされて解析された後、ダウンローダーブロックが不足しているため、現在はハングしています。

これが完全に含まれているテストメソッドで、次の呼び出しでハングしますWait。

（予想どおり、ハングする前の）出力は次のとおりです。

複数のビデオ クリップをフレームごとに処理するアプリケーションを並列化したいと考えています。クリップごとの各フレームのシーケンスは重要です (明らかに)。TPL Dataflow を使用することにしたのは、これがデータフロー (ムービー フレームがデータである) の良い例であると信じているからです。

したがって、データベースからフレームをロードするプロセスが 1 つあります (たとえば、500 のバッチで、すべてまとめて)

それらを BufferBlock に投稿します。この BufferBlock に、ActionBlocks をフィルターにリンクして、MovieID ごとに 1 つの ActionBlock を持つようにしました。これにより、ある種のデータ パーティショニングが得られます。各 ActionBlock はシーケンシャルですが、理想的には、複数のムービーの複数の ActionBlock を並行して実行できます。

上記のネットワークは機能しており、並行して実行されていますが、私の計算では、同時に実行されている ActionBlock は 8 ～ 10 個にすぎません。各 ActionBlock の実行時間とその約 100 ～ 200 ミリ秒の時間を計りました。少なくとも 2 倍の同時実行性を実現するには、どのような手順を実行できますか?

アクション デリゲートを非同期メソッドに変換し、ActionBlock アクション デリゲート内でデータベース アクセスを非同期にしようとしましたが、役に立ちませんでした。

編集:追加レベルのデータ パーティショニングを実装しました。奇数 ID のムービーのフレームは ServerA で処理され、偶数ムービーのフレームは ServerB で処理されます。アプリケーションの両方のインスタンスが同じデータベースにヒットしました。問題が DB IO である場合、処理されたフレーム数の合計に改善は見られません (または 20% 未満の非常にわずかなもの)。しかし、私はそれが倍増していると見ています。したがって、これにより、Threadpool はより多くのフレームを並行して実行するために、より多くのスレッドを生成していないと結論付けることができます (両方のサーバーはクアッドコアであり、プロファイラーはアプリケーションごとに約 25 ～ 30 のスレッドを示します)。

次の問題に直面しています。

オブジェクトのデータ ストリームがありFoo、それらのオブジェクトをいくつかの同時インプロセス タスク/スレッドにストリーミングします。これらのタスク/スレッドは、オブジェクトを処理してオブジェクトを出力しFooResultます。それぞれのメンバーには、 の作成に使用されたものFooResultと同じものが含まれています。ただし、必ずしもすべてが.FooFooResultFooFooResult

私の問題は、このプロセス全体から、並行タスク内から作成されたFoo可能性のある元のオブジェクトと、もしあればすべてのオブジェクトを含むラッピングオブジェクトを渡したいということです。FooResultFoo

注: 現在、TPL データフローを使用していますが、各同時プロセスActionBlock<Foo>は からリンクされている内で発生しますBroadCastBlock<Foo>。SendAsync()ターゲット データフロー ブロックを使用して、潜在的に作成された を送信しますFooResult。明らかに、同時データ フロー ブロックFooResultは予測できない時間に生成されます。これは、私が現在苦労していることです。FooResultまとめて何個作成されたのか、ActionBlock<Foo>オリジナルと一緒にまとめてFooラッピングオブジェクトとして渡すことができるようには思えません。

擬似コードでは、現在次のようになっています。

ただし、現在のコードの問題は、アクション ブロックのいずれかでFooa がシングルを生成しなかった場合、targetBlock が何も受信しない可能性があることです。FooResultまた、FooResult各アクション ブロックがFooResult.

私が望むのは、targetBlock がそれぞれを含むラッピング オブジェクトを受け取り、オブジェクトが作成された場合は のコレクションもFoo受け取ることです。FooResultFooResult

説明されている方法でソリューションを機能させるために私ができることはありますか? TPL Dataflow を熟読する必要はありませんが、熟読できれば便利です。

更新: 以下は、svick によって提案された JoinBlock の実装を通じて取得したものです。私はそれを使用するつもりはありません (パフォーマンスの観点から微調整できない限り)。実行が非常に遅いため、1 秒あたり約 89000 個のアイテム (およびそれは int 値型のみ) に達します。

提案を反映するためのコードの更新:

両方のトランスフォームブロックが完了したときにコードが完了するコードをどのように書き直すことができますか？完了とは、完了とマークされ、「アウトキュー」が空であることを意味すると思いましたか？

コードを編集し、各変換ブロックの入力バッファー数を追加しました。明らかに、100個すべてのアイテムが各変換ブロックにストリーミングされます。ただし、トランスフォームブロックの1つが終了するとすぐに、プロセッサブロックはそれ以上のアイテムを受け入れず、代わりに不完全なトランスフォームブロックの入力バッファが入力バッファをフラッシュするだけです。

私は、Microsoft チームの一部、または新しい TPL データフロー ライブラリ、async/await 同時実行フレームワーク、および TPL の機能を詳述している他の作成者による、かなりの数の技術文書を読みました。ただし、どちらをいつ使用するかを明確に説明しているものに実際には遭遇していません。それぞれに独自の場所と適用性があることは認識していますが、具体的には次の状況に関して疑問に思います。

完全にインプロセスで実行されるデータ フロー モデルがあります。一番上にはデータ生成コンポーネント (A) があり、データを生成し、データ フロー ブロック リンケージを介して、または処理コンポーネント (B) にイベントを発生させることによってデータを渡します。(B) 内の一部の部分は同期的に実行する必要がありますが、(A) プロセスのほとんどは I/O または CPU バウンド (ディスクからバイナリ データを読み取り、それらを逆シリアル化して並べ替える) であるため、並列処理から大きなメリットがあります。最後に、処理コンポーネント (B) は変換された結果を (C) に渡し、さらに使用できるようにします。

具体的には、タスク、async/await、および TPL データ フロー ブロックをいつ使用するか、次の点で疑問に思います。

• データ生成コンポーネント (A) を開始します。GUI/ダッシュボードをロックしたくないのは明らかなので、このプロセスは別のスレッド/タスクで実行する必要があります。

• (A)、(B)、および (C) 内のメソッドを呼び出して、データの生成および処理プロセスに直接関与しないが、返されるまでに数百ミリ秒/秒かかる可能性のある構成作業を実行する方法。ここで async/await が活躍するというのが私の推測です。

• 私が最も苦労しているのは、あるコンポーネントから次のコンポーネントに渡されるメッセージを最適に設計する方法です。TPL Dataflow は非常に興味深いように見えますが、私の目的には遅すぎる場合があります。(パフォーマンスの問題に関しては最後に注意してください)。TPL Dataflow を使用しない場合、インプロセスのタスク間/同時データ受け渡しによって応答性と同時実行性を実現するにはどうすればよいですか? たとえば、タスク内でイベントを発生させた場合、サブスクライブされたイベント ハンドラーは別のタスクに渡されるのではなく、同じタスクで実行されます。要約すると、コンポーネント (B) がデータを取得して処理に集中している間に、コンポーネント (B) にデータを渡した後、コンポーネント (A) はどのように業務を遂行できるのでしょうか? ここで使用するのに最適な同時実行モデルはどれですか? ここでデータ フロー ブロックを実装しましたが、それが本当に最善の方法でしょうか?

• 上記の要約ポイントは、標準的な手法を使用して API タイプのコンポーネントを設計および実装する方法に苦労していると思いますか? メソッドを非同期に設計し、データ入力をデータ フロー ブロックとして設計し、データ出力をデータ フロー ブロックまたはイベントとして設計する必要がありますか? 一般的に最善のアプローチは何ですか？上記のコンポーネントのほとんどは独立して動作するはずなので、本質的にスワップアウトしたり、アクセサーや出力を書き直さなくても内部で個別に変更できるため、質問しています。

パフォーマンスに関する注意: TPL Dataflow ブロックが遅い場合があると述べました。私は高スループット、低レイテンシのタイプのアプリケーションとターゲット ディスク I/O 制限を扱っているため、tpl データフロー ブロックは、同期処理ユニットなどよりもはるかに低速で実行されることがよくあります。問題は、プロセスを独自のタスクまたは並行モデルに埋め込んで、tpl データフロー ブロックが既に処理しているものと同様のものを実現する方法がわからないことですが、tpl df に伴うオーバーヘッドはありません。

複数のアイテムを保持でき、複数のターゲット ブロックにリンクできるが、フィルター/述語を渡す特定のターゲット ブロックにのみアイテムを転送できる TPL データ フロー ブロック ソリューションを探しています。アイテムが同時に複数のターゲット ブロックに配信されることはありません。常に、フィルターに一致するブロックにのみ配信されるか、アイテムを破棄できます。私は BroadCastBlock が好きではありません。私の理解が正しければ、配信は保証されず (または保証されますか?)、フィルタリングはターゲット ブロック側で行われるため、BroadCastBlock は基本的に各項目のコピーをすべての linkedTo ターゲット ブロックに送信します。また、私の理解が正しければ、いつでも複数のアイテムを保持することはありません。Post/Async を使用したくありませんが、LinkTo チェーンを維持します。

完全なカスタム データ フロー ブロックを回避する方法はありますか? または、BroadCastBlock の仕組みを誤解していますか? 残念ながら、詳細を説明し、ユースケースをカバーするドキュメントはあまりありません。どんなアイデアでも大歓迎です。

Post() または SendAsync() を介してアイテムを送信することの違いについて混乱しています。私の理解では、すべての場合において、アイテムがデータ ブロックの入力バッファーに到達すると、制御は呼び出し元のコンテキストに返されますよね? では、なぜ SendAsync が必要になるのでしょうか? 私の仮定が間違っているとしたら、反対に、データ ブロックを使用する全体的なアイデアが同時実行および非同期環境を確立することである場合、なぜ誰かが Post() を使用するのでしょうか。

もちろん、Post() が bool を返すのに対し、SendAsync は bool の待機可能な Task を返すという点で、技術的に違いを理解しています。しかし、それにはどのような意味があるのでしょうか? bool の戻り値 (項目がデータ ブロックのキューに配置されたかどうかの確認であると理解しています) はいつ遅延しますか? async/await 同時実行フレームワークの一般的な考え方は理解していますが、ここではあまり意味がありません。ブール値以外に、渡されたアイテムに対して行われた結果は呼び出し元に返されることはなく、代わりに「out-queue」で、リンクされたデータ ブロックに転送されるか、破棄されます。

また、アイテムを送信する際に、2 つの方法にパフォーマンスの違いはありますか?

次の場合：

はtargetBlock完了していないようです。その理由は、を他のDataflowブロックにTransformBlock targetBlockリンクしていないため、のすべてのアイテムが出力キューで待機しているためだと思います。targetBlockしかし、私が実際に達成したいのは、（A）targetBlock完了が通知され、（B）入力キューが空になったときの通知です。アイテムがまだの出力キューにあるかどうかは気にしたくありませんTransformBlock。どうすればそれについて行くことができますか？sourceBlockANDの完了ステータスを照会してInputCount、のtargetBlockがゼロであることを確認したいものを取得する唯一の方法はありますか？これが非常に安定しているかどうかはわかりません（の最後のアイテムがに渡されたsourceBlock場合にのみ、本当に完了とマークされますsourceBlocktargetBlock？）。同じ目標を達成するためのよりエレガントで効率的な方法はありますか？

sourceBlock編集： ANDがゼロInputCountであることの完了をチェックする「汚い」方法でさえ、実装するのは簡単ではないことに気づきました。targetBlockそのブロックはどこにありますか？targetBlock上記の2つの条件が満たされると、メッセージは明らかに処理されなくなるため、内に含めることはできませtargetBlockん。また、導入の完了状況を確認するとsourceBlock、多くの非効率性が生じます。