問題タブ [concurrent-collections]

WPF UI でデータグリッドにバインドされたコレクションがあります。

私の要件は、コレクション内のすべてのアイテムに対して、プロパティの値を毎秒 10 回更新する必要があるようなものです。

だから私は ConcurrentBag タイプのコレクションを取りました。すべてのアイテムの値を更新した後。RaisePropertyChange を明示的に起動しています。ただし、UI は変更されません。

重複
の可能性: ConcurrentBag でメモリリークの可能性がありますか?

アプリで重大なメモリ リークが発生しています。メソッドの1つでローカルのconcurrentBagコレクションに追加したすべてのデータは収集されませんでした。

この単純なコードは、私がそれをどのように使用するかを示しています:

私が期待すること: バッグが作成され、メソッド「Func」で破棄されます。

私が見ているもの: バッグは破棄されず、Parallel.ForEach で作成されたすべてのスレッドを保持し、追加したすべてのデータを保持します。=(

さて、リストに追加するときに、「TryTake」を使用してバッグからアイテムを削除できます。しかし、空のバッグはまだメモリに保持されています。

現在、ConcurrentBag の代わりに List を使用して問題を解決しています。しかし、これを記憶プロファイラーで見たので、よく眠れません。私の英語でごめんなさい=）

アップデート

メソッド「Func」を変更します。

次に、VS でプロジェクトを作成し、プログラムをコンパイルして実行します。このインスタンス グラフは、プログラムがすべての作業を完了してから 10 分後に収集したメモリ スナップショットから ".Net Memory Profiler 4.0" によって作成されました。

http://xmages.net/storage/10/1/0/f/d/upload/4c67f305.jpg (申し訳ありませんが、画像を投稿できません)

一度に 1 つずつではなく、一度に複数のアイテムを ConcurrentBag に追加する方法はありますか? ConcurrentBag には AddRange() メソッドはありませんが、Concat() はあります。しかし、それは私にとってはうまくいきません：

このコードは以前 Parallel.ForEach() に含まれていましたが、トラブルシューティングできるように上記のように変更しました。変数 newList には確かにオブジェクトがありますが、objectList.Concat<> 行の後では、objectList には常に 0 オブジェクトが含まれています。Concat<> はそのように機能しませんか? Add() メソッドを使用して、アイテムを一度に 1 つずつ ConcurrentBag に追加する必要がありますか?

Microsoft の Parallel Programming ホワイトペーパーでは、さまざまな FLOPS しきい値の下で最適な状況が説明されており、FLOPS レートは、特定の実装をいつ使用する必要があるかを決定するポイントであると説明されています。

アプリケーションで FLOPS を測定するにはどうすればよいですか?

次のような動作をするコレクションを実装するフレームワークがあるかどうかに興味があります。

最初に次のものが含まれているとします：[1、2、3]

• 私はそれを（イテレーターを使用して）繰り返し、要素2に到達します。ここで、最後に4を追加します（コレクションは[1、2、3、4]になります）。
• ここで、新しいイテレータを作成してコレクションを反復すると、[1、2、3、4]になります。
• 私は最初のイテレータで反復を続けます、そしてそれは私にちょうど3を与えて戻ってきます
• ここで、最初のイテレータをリセットすると、[1、2、3、4]が得られます（新しいイテレータを作成するのと同様です）。

同じことが要素の除去にも当てはまります。追加する代わりに3を削除すると、2番目のイテレータは[1、2]を返しますが、最初のイテレータは3と最後を返します。

したがって、イテレータを取得して、イテレータを作成したときに持っていたコレクションを取得したいのですが（後でイテレータを繰り返しても、少し繰り返して後で続行します）、イテレータをリセットすると、ガベージコレクションが行われます。は最新バージョンに更新され、イテレータが作成されたときの配列の内容に応じて異なるバージョンを提供する複数のイテレータインスタンスを異なる時間に作成できるはずです。

複数のスレッドでうまく機能するためにそれが必要であり、効率的な実装が望ましいです。

誰かがそのようなコレクションの実装を知っていますか、それとも私はそれを自分で実装する必要がありますか？

次のコードがあります。

私の質問は、新しいタスクを設定するときに、Take または Try Take メソッドを使用する必要がありますか、それとも上記で問題ありませんか? つまり、アイテムをコレクションから安全に削除し、タスクが安全に完了したときに再度追加する必要はありません。私の質問に答えたい人に提供できる唯一の保証は、2 つのタスクが「Cache.TreeNodeItemsCollection」の同じノードで動作しないということです。

タスクを開始するためのコードをまだ追加していませんが、現在のセットアップが適切かどうかがわかったら追加されますか?

Windows サービスでホストされている WCF サービスがあります。

WCF サービスにはConcurrentQueue<SomeClass>.

Windows サービス (WCF サービスをホストする) にもConcurrentQueue<SomeClass>.

Windows サービスは WCF サービスを開始し、ワーカー スレッドも生成します。

Enqueueワーカー スレッドは、 の ( ) オブジェクトをにプッシュSomeClassしConccurrentQueueます。

オブジェクトがキューに入れられたときに WCF サービスがトリガーされ、ConcurrentQueueそれ (およびキューにある可能性のある他のオブジェクト) をデキューし、接続されているすべてのクライアントにメッセージを送信できるようにしたいと考えています。

.

2 つの質問:

1. 注入された ConccurentQueue の「Enqueue」メソッドで WCF サービスを「フック」して、オブジェクトがキューにプッシュされたときに WCF サービスが何かを行うようにするにはどうすればよいですか?

2. なんとかして「エンキュー」イベントをフックし、Windows サービスのワーカー スレッドが2 つのオブジェクトをキューにエンキューした場合、つまり、WCF サービスの「エンキュー フック」も 2 回「トリガー」することになります。トリガーされた 2 つのイベントが異なるスレッドで発生しますか? どういうわけか、WCF サービスがキューからオブジェクトを順番にプルするようにする必要があります。エンキュー操作ごとに突然それを見たくありません-WCFサービスは別のスレッドでトリガーされ、オブジェクトを引き出します...

私の質問が十分に明確であることを願っています...

.

[アップデート]

Chris と話し合った結果、WCF サービスからワーカー スレッドを分離し、ワーカー スレッドがクライアントとして WCF サービスを呼び出せるようにするのが最善の方法であるという結論に達しました。これは、キューをいじる必要がないことを意味します。

非常に簡単です: ConcurrentDictionary (必要に応じて使用しますが、実際には正しい概念ではありません) 以外に、アイテムまたは述語の単純な等価性に基づいて特定のアイテムの削除をサポートするコンカレント コレクション (IProducerConsumer 実装) はありますか?削除の条件を定義しますか?

説明: DB からオブジェクトをプルして「開始」キューに入れる、マルチスレッド、マルチステージのワークフロー アルゴリズムがあります。そこから、それらは次のステージに取り込まれ、さらに作業が行われ、他のキューに詰め込まれます。このプロセスは、さらにいくつかの段階を経て続きます。一方、最初のステージはそのスーパーバイザーによって再度呼び出され、オブジェクトを DB から引き出します。これらのオブジェクトには、まだ処理中のオブジェクトが含まれる可能性があります (処理が完了しておらず、次のように設定されたフラグで再永続化されていないため)。彼らは終わった）。

私が設計しているソリューションは、「作業中」のマスター コレクションです。オブジェクトは、最初のステージで処理のために取得されるとそのキューに入り、必要な処理を完了したワークフローのステージによって「処理済み」として DB に再保存された後に削除されます。オブジェクトがそのリストにある間、最初のステージで再取得された場合、オブジェクトは無視されます。

ConcurrentBag を使用する予定でしたが、唯一の削除方法 (TryTake) は、指定されたアイテムではなく、任意のアイテムをバッグから削除します (そして、.NET 4 ではConcurrentBag は低速で​​す)。ConcurrentQueue と ConcurrentStack では、次に提供されるアイテム以外のアイテムを削除することもできず、ConcurrentDictionary を残します。これは機能しますが、必要以上のものです (本当に必要なのは、処理中のレコードの ID を格納することだけです。ワークフロー中に変更されません)。

私は現在 IRC ボットを書いています。過剰なフラッドを避けたいので、X ミリ秒ごとに次のメッセージを送信するメッセージ キューを作成することにしましたが、失敗しました。43行目:

OutOfMemory 例外をスローします。私は何が間違っているのか全くわかりません。

おそらく、そのような (おそらく複雑な) キューイング方法の背後にある一般的な考え方についても説明する必要があります。

まず、メッセージのターゲットがキーとして機能するメインHashtable queuehtストアタイプです。ConcurrentQueue<string>ボットがハッシュテーブルを反復処理し、各キューから 1 つのメッセージを送信するようにします (キューが空になった場合はキーを削除します)。ハッシュテーブル自体を処理する適切な方法が思いつかなかったのでConcurrentQueue<string> queue、キューを空にするときにキーとその使用順序を格納する別のキュー を作成することにしました。

キューに数百のアイテムがあるという仮説的な状況を想定すると (これは可能かもしれません)、新しいリクエストはどれくらいの時間 (メッセージ間の組み込みの遅延と遅延) を知っているかによって遅延されるため、メソッド Add() を再構築しqueueます。のディープ コピーを作成しqueueht(そう願っています)、queueこの使い捨てコピーに基づいて新しいコピーを生成し、その過程でそれを取り除きます。

私はスレッド化、単純な配列よりも複雑なコレクション、および OOP の習慣/慣習をまったく経験していないため、私の一連の思考やコードはひどく間違っていると思います。説明で私の問題を解決していただければ幸いです。前もって感謝します！

編集: クラス全体を投稿します。

EDIT2:while (ht.Count > 0)無期限に実行されることを認識しています。それは、次のように見えた以前のバージョンからの残りの部分です。

しかし、評価時にコレクションを変更することはできません (そして、私はそれが難しい方法であることがわかりました)。の条件を変更するのを忘れていましたwhile。

私は自由にTheThingのソリューションを試しました。目的を果たしているように見えますが、メッセージは送信されません...最終的な形式は次のとおりです。

に切り替えてみましたがConcurrentQueue、これも同様に機能するはずです（ただし、よりスレッドセーフな方法で、スレッドセーフについて何も知らないというわけではありません）。に切り替えてみましたSystem.Threading.Timerが、それも役に立ちません。私はずっと前にアイデアを使い果たしました。

編集: 完全でまったくのばかであるため、タイマーの開始時間を設定しませんでした。bool 部分を、タイマーの dueTime と interval を変更する Start() メソッドに変更すると、機能するようになりました。問題が解決しました。

頻繁に読み取られるが、ほとんど書き込まれないマップがあります。一部の操作 (読み取りまたは書き込み) には、アトミックに操作する必要がある複数のオブジェクトが含まれるため、パフォーマンスを向上させるために ReadWriteLock を使用しました。

現在、並行マップを通常のハッシュ マップにダウングレードするオプションがありますが、一部の遅い反復コードが懸念されます。

マップをダウングレードすると、同時アクセス例外を回避するために、長い反復子が読み取りロックを保持する必要があります。これにより、書き込みスレッドが長時間ブロックされると思います。

一部の反復子は一貫性のないデータに影響されないため、反復子を並行書き込みで使用できるように並行マップを保持できます。ただし、これにより、ロックを適切に使用している操作に (並行マップからの) 不必要なオーバーヘッドが追加されます。

または、Read-on-write マップのようなものを実装することもできます。このマップでは、(非同時実行の) マップ全体が書き込み操作用に複製されるため、既存のイテレーターは読み取りロックの外でも動作し続けます。

明らかに、これらのメソッドはすべて有効であり、パフォーマンスは実際のコードとセットアップに依存します。しかし、これに関する研究はあるのでしょうか (自分で実験を行う必要はありません)。