問題タブ [blockingcollection]

BlockingCollection を使用して Producer/Consumer パターンを実装しましたが、期待どおりにブロックしていないようです。

Web カメラからフレームを受信し、BlockingCollection に追加するスレッドが 1 つあります。

そして別のスレッドでは、コレクションへの参照があり、使用してフレームを処理します

ただし、以下に示すように、InvalidOperationException をスローして、プルしているフレームが他の場所で使用されていることを通知する傾向があります。

img http://i17.photobucket.com/albums/b52/orubap/2012-03-24_020858.png

常にすぐに発生するとは限りませんが、キューが空またはほぼ空である場合にのみ発生することに気付きました (つまり、コンシューマーがプロデューサーよりも高速です)。最初に追加された画像または最後に撮った画像。なぜこれが起こっているのでしょうか？

単純なパイプラインを作成しようとしているときにBlockingCollection<T>、Task Parallel Library を使用して、定量化および再現可能な問題が発生しました。ConcurrentQueue<T>GetConsumingEnumerable

簡単に言えば、あるスレッドからデフォルトBlockingCollection<T>(フードの下では a に依存している) にエントリを追加しても、メソッドを呼び出している別のスレッドからエントリConcurrentQueue<T>がポップされることは保証されません。BlockingCollection<T>GetConsumingEnumerable()

これを再現/シミュレートするための非常に単純なWinformsアプリケーションを作成しました。これは、整数を画面に出力するだけです。

• Timer1作業項目をキューに入れる責任があります... と呼ばれる並行辞書を使用し_trackerて、ブロッキングコレクションに既に追加されているものを認識します。
• Timer2BlockingCollectionの& の両方のカウント状態をログに記録しているだけです_tracker
• START ボタンはParalell.ForEach、ブロッキング コレクションを単純に反復しGetConsumingEnumerable()、2 番目のリスト ボックスへの出力を開始する を開始します。
• STOP ボタンはTimer1、ブロッキング コレクションにエントリが追加されるのを防ぎます。

イベントのシーケンスは次のとおりです。

• スタートを押します
• Timer1 がティックし、ListBox1 が 3 つのメッセージで即座に更新されます (0、1、2 を追加)
• ListBox2 はその後、1 秒間隔で 3 つのメッセージで更新されます
  • 処理中 0
  • 処理1
  • 加工2
• Timer1 がティックし、ListBox1 が 3 つのメッセージで即座に更新されます (3、4、5 を追加)
• ListBox2 は、1 秒間隔で 2 つのメッセージで更新されます。
  • 処理3
  • 処理 4
  • 処理 5が印刷されていない... 「行方不明」になったように見える
• STOP を押して、タイマー 1 によってそれ以上メッセージが追加されないようにします
• 待って…「処理中5」がまだ表示されない

コンカレント ディクショナリは、1 つのアイテムがまだ処理されておらず、その後削除されたことを追跡していることがわかります。_tracker

もう一度 Start を押すと、timer1 がさらに 3 つのエントリを追加し始め、Parallel ループが復活して 5、6、7、8 を出力します。

なぜこれが起こるのか、私は完全に途方に暮れています。start を再度呼び出すと、明らかに Parallell foreach を呼び出す newtask が呼び出され、不足しているエントリを魔法のように見つける GetConsumingEnumerable() が再実行されます...

BlockingCollection.GetConsumingEnumerable()コレクションに追加されたすべてのアイテムを反復処理することが保証されていないのはなぜですか。

その後、さらにエントリを追加すると、「スタックが解除」されて処理が続行されるのはなぜですか?

このような例を書いて、非同期実行の BlockingCollection の速さを測定しました

結果は残念です:

平均して約 50 マイクロ秒を費やしましたが、最大で 600 マイクロ秒を費やしたこともあります。

遅い Pentium U5400 を使用しても、数マイクロ秒、最大 10 マイクロ秒、決して最大 10 マイクロ秒の定数になるはずです。

.NET には、非同期実行のためのより高速な方法がありますか? async exec がスケジュールされた後、できるだけ早く開始する必要があります。これは、金融の時間に敏感な計算です。

コレクションをブロックすると、順序が保証され、アイテムが 1 つずつ処理されることが保証されるため、この質問には実際には 2 つの質問が含まれます

1. 注文が必要で、アイテムが表示された順序で処理される必要がある場合、より速いものはありますか? つまり、FIFO クエリが必要です。
2. 順序を気にせず、アイテムが 1 つずつ処理されるか、並行して処理されるかを気にしない場合、より速いものはありますか?

答えは次のとおりだと思います。

1. いいえ。その参照には BlockingCollection を使用する必要があります。BlockingCollection<T> をシングル プロデューサー、シングル コンシューマーの FIFO クエリとして使用するのは適切ですか?

2. おそらくデリゲートを試すことができますか？http://msdn.microsoft.com/en-us/library/2e08f6yc.aspx

MyDict<Type, BlockingCollection<"differentTypes">キャストを回避するインターフェイスを介してディクショナリをセットアップして、オブジェクトまたは動的タイプの使用を防ぐことはできますか?

私がそのようなものを使用したい理由は、実行時にさまざまなタイプのブロッキング コレクションを初期化したいからです。着信オブジェクトはさまざまなタイプであり、そのような着信オブジェクトをタイプに一致するブロッキング コレクションに追加したいからです。また、後でキャストせずに BlockingCollection 内のオブジェクトにアクセスしたいと考えています。Blocking コレクションの必要な型は、ラッピング クラスのインスタンス化時に認識されます。着信オブジェクトのタイプは、実行時にのみ認識されます。着信オブジェクト タイプは、クラスのインスタンス化時に認識されているタイプと異なることはありません。一致する BlockingCollection から取得するオブジェクトの型も、実行時にのみわかります。キャストせずにこれを設定する方法はありますか?

関連する多くの質問を閲覧しましたが、ここに記載されている仕様とまったく同じものは見つかりませんでした。

現在追加されている関数を実行したくありません。後でビジネスロジックを使用して手動で行います。!!!

タスクを永続的に保持し、WCF サービスから無期限に実行するのに苦労しています。私はこれを間違った方法で行っている可能性があり、提案を喜んで受け入れます。

BlockingCollection にドロップされた着信要求の処理を開始するタスクがあります。私が理解していることから、 GetConsumingEnumerable() メソッドを使用すると、データが到着したときに永続的にプルできるはずです。それ自体は問題なく動作します。Windowsフォームを使用してリクエストに記入し、送信することで、エラーや欠陥を1つも発生させることなく、数十のリクエストを処理することができました. このプロセスに自信が持てたら、asmx Web サービスを介して自分のサイトに接続し、jQuery ajax 呼び出しを使用してリクエストを送信しました。

サイトは送信された URL に基づいてリクエストを送信し、Web サービスは URL から html コンテンツをダウンロードし、コンテンツ内の他の URL を探します。次に、見つかった URL ごとにリクエストを作成し、BlockingCollection に送信します。WCF サービス内で、アプリケーションがオンラインの場合 (つまり、タスクが開始された場合)、Parallel.ForEach を介して GetConsumingEnumerable を使用して要求を取得し、要求を処理します。

これは最初のいくつかの送信では機能しますが、タスクは予期せず停止します。もちろん、これはテストでシミュレートできるよりも 10 倍多くのリクエストを実行していますが、スロットルするだけだと思っていました。問題は、タスクを開始する私の方法にあると思います：

これを WF4 サービスに移動し、ワークフローに接続してワークフローの永続性を使用することを考えましたが、必要でない限り WF4 を学習するつもりはありません。さらに情報が必要な場合はお知らせください。

TCP ポートからストリーミング データを取得し、リアルタイムで計算を行うアプリケーションを作成しようとしています。これまでのところすべて問題ありませんが、プロデューサー スレッドがデータ チャンクのパッケージ化を開始するときに、ユーザー インターフェイスとマーキー プログレス バーが (不規則な短い期間) 動かなくなります (コードを参照)。

多くのテストを行ったところ、BlockingCollection bcPort へのアクセスが問題のようであることがわかりました。bcPort は、UI スレッドに影響を与えないはずの別のデータ加算スレッドから常にチャンクを取得します。したがって、次のことがわかりません。

1.) チャンクの追加とパッケージ化に別のスレッドを使用すると、GUI が動かなくなるのはなぜですか?

2.) データの保存に BC を使用しているのに、なぜこのようなことが起こるのですか? これらの特定の目的のためのスレッド セーフ コレクションではありませんか?

ちなみに、Windows 7 の ResourceManager は、ストリーミング中に 100% の CPU 使用率を示します。チャンクにはそれぞれ約 2000 の浮動小数点値が含まれており、そのうちの 4 つまたは 5 つが毎秒急いでいます。ロガーも無効にしましたが、効果はありません。コンシューマー スレッドと評価スレッドは無効になっています。

UI スレッドのほかに、"ReceiveAndSave" と呼ばれるスレッドだけがあり、受信する float 値からチャンクを作成します (コード、メソッド "Add" を参照)。スレッド「プロデューサー」はさらにパッケージ化を行っており、コンシューマー (非アクティブ化) のチャンクをキューに入れています。

プロデューサーは 'myThreads[0] = new Thread(Produce); で開始されます。myThreads[0].Name = "プロデュース"; myThreads[0].Start();'

コンシューマーがツリーの各レベルでいくつかの作業を行ってから、ツリーを再帰的に下って次のレベルで同じ作業を実行する必要があるという再帰的な問題があります。

ConcurrentBag/BlockingCollectionなどを使用してこれを並行して実行したいと思います。このシナリオでは、キューのコンシューマーはキューのプロデューサーでもあります。

私の問題はこれです：BlockingCollectionを使用すると、アイテムをデキューし、新しいアイテムをキューに入れるための非常に単純なforeachロジックを記述できます-キューが空の場合、ブロッキングコレクションは正しくブロックされ、他のいずれかによって新しい作業が生成されるのを待ちます消費者。

しかし、すべての消費者がブロックしているかどうかをどうやって知ることができますか？！

CompleteAdding（）については知っていますが、実際に完了するのは、すべてのプロデューサーがプロデュースを完了し、キューが空になったときだけであるため、それは役に立たないようです。 「無料」でCompleteAdding（）を設定します。これを検出する方法はありますか？（おそらく、ブロックすると発火し、ブロックを解除すると再び発火する可能性のあるイベントですか？）

foreachを使用せずに、手動でwhile（！complete）ループを作成し、TryTakeを使用することで、これを手動で処理できますが、手動でスリープする必要があります。これは非効率的です（ブロッキングコレクションを使用する理由とそもそも同時収集！）ループを通過するたびに、TryTakeがfalseの場合、アイドルフラグを設定し、キューが空で、すべてのスレッドがアイドル状態であるかどうかをマスターにチェックさせ、完全フラグを設定します。しかし、繰り返しになりますが、これは厄介なようです。

直感的に、ブロッキングコレクションを使用してこれを行う方法があると言われていますが、私はそこにたどり着くことができません。

とにかく、誰もが消費者がプロデューサーであるときの良いパターンを持っており、すべてのブロックをいつリリースするかを検出できるのは素晴らしいことです

私は持ってSystem.Collections.Generic.List _myListおり、多くのスレッドがそれから読み取ったり、アイテムを同時に追加したりできます。私が読んだことから、「BlockingCollection」を使用する必要があるため、これは機能します。ReaderWriterLockSlimとについても読みましたが、lockの代わりにそれらを使用する方法がわからないBlockingCollectionので、私の質問は次のように同じことができますか?

1. ReaderWriterLockSlim
2. ロック

「BlockingCollection」を使用する代わりに。はいの場合、簡単な例と、、、を使用することの長所と短所を教えBlockingCollectionてReaderWriterLockSlimくださいlock。

更新され た読者は、作家以上のものになります!

byte[20] を持つオブジェクトが 1 つのスレッドで BlockingCollection に渡され、別のスレッドが BlockingCollection.Take() を使用して byte[0] を持つオブジェクトを返すという問題があります。これはスレッドの問題だと思いますが、BlockingCollection が並行コレクションであることを考えると、どこで、またはなぜこれが起こっているのかわかりません。

スレッド 2 では、myclass2.mybytes が byte[0] と等しい場合があります。これを修正する方法に関する情報は大歓迎です。

[編集] 元のコード。問題なく動作するように見えた上記のコードを削除したので、時間をかけて元のコードを調べて投稿しました。

MessageBuffer.cs

Listener() と ReceivedMessageHandler(object messageProcessor) を持つクラスで

スレッド 1 で

Message.ReadMessage(NetworkStream ストリーム、iTcpConnectClient クライアント)

スレッド 2 で

PlayerStateMessage.cs

Message.GetVector2FromBuffer(int bufferlocation) これは例外がスローされる場所です。これは、this.Buffer が byte[20] である必要があるのに byte[0] であるためです。