問題タブ [backpressure]

私は (率直に言って素晴らしい)BlockingCollection<T>型を、高度にマルチスレッド化された高性能アプリに利用しています。

コレクション全体で多くのスループットがあり、ミクロレベルでは非常に高性能です。ただし、「バッチ」ごとに、キャンセルトークンにフラグを立てることで常に終了します。これにより、待機中のTake呼び出しで例外がスローされます。それは問題ありませんが、a) 例外には明らかなオーバーヘッドがあり、b) デバッグ時に、その特定の例外のブレークを手動でオフにしたくないため、戻り値または出力パラメーターでそれを通知することにしました。例外。

実装は複雑なようで、理論的には、例外を使用しない独自のバージョンを逆アセンブルして再作成できると思いますが、おそらくそれほど複雑ではない方法がありますか?

nullコレクションにオブジェクト (そうでない場合はプレースホルダー) を追加して、プロセスが終了する必要があることを示すこともできますが、適切に中止する手段も必要です。

では - 代替のコレクション型ですか? 私自身を再作成しますか？これを悪用する方法はありますか？

(いくつかのコンテキスト: を使用BlockingCollection<T>して手動でロックするよりも利点があるため、使用しましたQueue。スレッドプリミティブの使用が優れていることがわかります。私の場合、あちこちで数ミリ秒と最適なコアを使用することが重要です。 )

編集：私はちょうどこれのために報奨金を開きました. アナスタシアスヤルの答えが、私のコメントで提起した質問をカバーしているとは思いません。私はこれが難しい問題であることを知っています。誰でも支援できますか？

CSV ファイルから URL を読み取り、それらが使用可能かどうかを確認し、不適切な URL を別のファイルに書き込もうとしています。劇的に遅くなる前に（1秒あたり5回のチェックから10秒ごとのチェックまで）約数千を通過し、おそらく適用していないため、「致命的なエラー：CALL_AND_RETRY_2割り当てに失敗しました-プロセスがメモリ不足です」というメッセージが表示されます送信ストリームに正しくバックプレッシャします。

1 つはメッセージを生成し、もう 1 つは固定レートでメッセージを消費しています。

コンシューマー BoundedMailBox によってプロデューサを調整することは可能ですか? (背圧)

現在、プロデューサーは定期的にスケジュールされています (ティック メッセージを送信しています)。代わりに、コンシューマー メールボックスの可用性によってスケジュールする方法はありますか?

応答が必要ないため、ファイア アンド フォーゲット スタイル ( consumer.tell() ) を使用しています。別のメッセージ送信アプローチを使用する必要がありますか?

戻るボタンを押した後のログ:

11-07 22:48:08.376: D/AndroidRuntime(5325): VM をシャットダウンしています 11-07 22:48:08.376: W/dalvikvm(5325): threadid=1: キャッチされない例外で終了するスレッド (group=0x4162d700) 11 -07 22:48:08.384: E/AndroidRuntime(5325): 致命的な例外: メイン 11-07 22:48:08.384: E/AndroidRuntime(5325): java.lang.RuntimeException: アクティビティを破棄できません

{com.some.pack/com.some.packNY}: java.lang.NullPointerException 11-07 22:48:08.384: E/AndroidRuntime(5325): android.app.ActivityThread.performDestroyActivity で

(ActivityThread.java:3627) 11-07 22:48:08.384: E/AndroidRuntime(5325): android.app.ActivityThread.handleDestroyActivity(ActivityThread.java:3645) 11-07 22:48:08.384: E/AndroidRuntime (5325): android.app.ActivityThread.access$1200(ActivityThread.java:153) で 11-07 22:48:08.384: E/AndroidRuntime(5325): android.app.ActivityThread$H.handleMessage(ActivityThread.java で:1322) 11-07 22:48:08.384: E/AndroidRuntime(5325): android.os.Handler.dispatchMessage(Handler.java:99) 11-07 22:48:08.384: E/AndroidRuntime(5325): android.os.Looper.loop(Looper.java:137) 11-07 22:48:08.384: E/AndroidRuntime(5325): android.app.ActivityThread.main(ActivityThread.java:5289) 11-07 22 :48:08.384: E/AndroidRuntime(5325): java.lang.reflect.Method.invokeNative(ネイティブ メソッド) 11-07 22:48:08.384: E/AndroidRuntime(5325):java.lang.reflect.Method.invoke(Method.java:525) 11-07 22:48:08.384: E/AndroidRuntime(5325): com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run で

(ZygoteInit.java:739) 11-07 22:48:08.384: E/AndroidRuntime(5325): com.android.internal.os.ZygoteInit.main(ZygoteInit.java:555) 11-07 22:48:08.384 : E/AndroidRuntime(5325): dalvik.system.NativeStart.main(Native Method) 11-07 22:48:08.384: E/AndroidRuntime(5325): 原因: java.lang.NullPointerException 11-07 22:48 :08.384: E/AndroidRuntime(5325): com.some.pack.NY.stop(NY.java:100) 11-07 22:48:08.384: E/AndroidRuntime(5325): com.some.pack. onDestroy(NY.java:106) 11-07 22:48:08.384: E/AndroidRuntime(5325): android.app.Activity.performDestroy(Activity.java:5302) 11-07 22:48:08.384: E/ AndroidRuntime (5325): android.app.Instrumentation.callActivityOnDestroy で

(Instrumentation.java:1117) 11-07 22:48:08.384: E/AndroidRuntime(5325): android.app.ActivityThread.performDestroyActivity で

(ActivityThread.java:3614) 11-07 22:48:08.384: E/AndroidRuntime(5325): ... 11 11-07 22:48:08.392: W/ActivityManager(408): 強制終了アクティビティ com.some .pack/.MainActivity 11-07 22:48:09.079: W/ActivityManager(408): ActivityRecord のアクティビティ一時停止タイムアウト{41985990 u0

com.some.pack/.MainActivity} 11-07 22:48:09.095: W/Settings(4780): 設定 bugreport_in_power_menu は android.provider.Settings.Secure から移動しました

android.provider.Settings.Global. 11-07 22:48:09.095: E/Cryptfs(124): 暗号化を使用して実行されていません。11-07 22:48:09.923: I/Process(5325): シグナルを送信しています。PID: 5325 SIG: 9 11-07 22:48:09.946: W/InputDispatcher(408): チャネル '419e2ff0 com.some.pack/com.some.pack.MainActivity (サーバー)' ~ Consumer

入力チャネルが閉じられているか、エラーが発生しています。events=0x9 11-07 22:48:09.946: E/InputDispatcher(408): チャネル '419e2ff0 com.some.pack/com.some.pack.MainActivity (サーバー)' ~ チャネルは

取り返しのつかないほど壊れており、処分されます！

RTCDataChannel API は、いかなる種類のフロー/制御またはバックプレッシャーも提供しません。これは、理論的には、送信者が受信者のブラウザーをクラッシュさせる可能性があることを意味しますか? 私の意見では、ブラウザー (Chrome、Firefox などはすべて内部で SCTP を使用します) は、SCTP 接続から読み取り、パケットを消費する js-callback を実行するようにスケジュールします。イベント キューが送信側に追いつかない場合、ブラウザは基本的にパケットをバッファに格納しながら継続的にパケットを読み取りますが、バッファは無限に大きくなります。したがって、2 つのブラウザーを接続すると、TCP 受信ウィンドウなどのような障壁がないため、実際には送信側が常に他のブラウザーを圧倒する可能性があります。

これは websocket api にも当てはまります。

何か見逃しているだけですか、それともこれらの API が壊れているだけですか? 私が正しければ、認証されていないブラウザーと通信する場合 (たとえば、急流のシナリオで)、これは深刻なセキュリティ上の問題になります。

Play 2.3 アプリケーションに WebSocket-to-TCP プロキシを追加する必要があります。アクターベースの API は明らかにサポートされていませんが、James Roper は次のように述べています。

Iteratee は設計によってこれを処理します。それまでは参照がないため、返される最後の future が償還されるまで、新しい要素を iteratee にフィードすることはできません。

しかし、彼が何を指しているのかわかりません。Iteratee.foreach、例で使用されているように、単純すぎるようです。私が iteratee API で目にする先物は、計算結果を完成させるためのものだけです。メッセージごとに入力する必要がありFuture[Unit]ますか?

バックプレッシャーを適切に処理するストリームを実装する方法を理解することはできません。一時停止と再開を使用しないでください。

私は正しく動作するようにしようとしているこの実装を持っています:

出力は正しく送信されますが、背圧が正しく生成されません。バックプレッシャーを適切にサポートするように変更するにはどうすればよいですか?