問題タブ [fork-join]

私は ExecutorService クラスを学び始めています。ドキュメント (およびオンラインのチュートリアル) には、常に ExecutorService.shutDown() を呼び出してリソースを再利用するように記載されています。ただし、ドキュメントには、shutDown() を呼び出した後は、新しいタスクは受け入れられないとも記載されています。それで、私の質問は、データ処理を並列化する必要があるときはいつでも、新しい ExecutorService をインスタンス化する必要があるのですか?

現在、Callable オブジェクトのリストがあり、次のことを行います。

ただし、私の OuterCallable もデータを分割したり、InnerCallable を使用して並列にデータ処理を実行したりします。

それが ExecutorService のためだったのか、Fork/Join アプローチのためだったのか思い出せませんが、データを操作するための実際の並列手順には I/O 操作を含めるべきではなく、すべてをメモリ内で実行する必要があると述べているドキュメントとチュートリアルを覚えています。ただし、私の InnerCallable では、実際に JDBC 呼び出しを行っています (ここには示されていません)。

最終的に、私が ExecutorService を使用している方法は機能しますが、まだ懸念が残っています。

1. 私のアプローチは、ExecutorService を使用した優れたプログラミング プラクティスを超えていますか?
2. ExecutorService のシングルトン インスタンスを使用する必要がありますか?
3. 並列メソッド内の I/O 操作だけでなく、JDBC 呼び出しも回避する必要がありますか?

最後の懸念事項として、Fork/Join と ExecutorService について少し調べてみました。Fork/Join API/クラスを完全に爆破した記事に出くわしました。Fork/Join を学ぶ価値はありますか? Stackoverflow などに関する記事をいくつか見ました。ここでは、テストを使用して Fork/Join と ExecutorService を比較し、Fork/Join と ExecutorService の CPU 使用率が向上していることを示すグラフがあります (Windows タスク マネージャー経由)。ただし、ExecutorService (JDK 1.7.x) を使用すると、CPU 使用率が最大になります。ExecutorService は最新の JDK で改善されましたか?

ヘルプ/ガイダンスをいただければ幸いです。

書き込みロックを保持しているスレッドから ForkJoinPool または ParallelArray にタスクを送信したいと考えています。ドメイン モデルへのアクセスは、現在のスレッドが関連するロックを保持していることを確認することで保護されます。FJ ワーカーがタスク (読み取り専用、クエリなど) を実行できるようにするには、それらを生成したスレッドにアクセス チェックを委任する必要があります。

生成スレッドへの参照を使用して、ForkJoinWorkerThread をサブクラス化しました。次に、ReentrantReadWriteLock をサブクラス化し、isWriteLockedByCurrentThread をオーバーライドして通常のチェックを実行し、スレッドがデリゲート FJWorker のインスタンスである場合は、ReentrantReadWriteLock#getOwnerを使用して、デリゲート スレッド (親) がロックの所有者であることをチェックします。 () :

ただし、getOwner() のドキュメントには次のように記載されています。

このメソッドが所有者ではないスレッドによって呼び出された場合、戻り値は現在のロック ステータスのベスト エフォート近似を反映します。たとえば、ロックを取得しようとしているがまだ取得していないスレッドがある場合でも、所有者が一時的に null になることがあります。

これは、既にアクセスが許可されているスレッド内でタスクを送信した場合、このメソッドがそのスレッドへの参照を正しく返すことを意味することを理解したいと思います。残念ながら、これは暗示されていません。

この方法を使用できない場合、この種の委任には他にどのような方法がありますか?

ありがとうございました。

テストプログラムの2つのバリエーションを比較しています。どちらもForkJoinPool、4コアのマシンで4スレッドで動作しています。

「モード1」では、エグゼキュータサービスのようにプールを使用します。たくさんのタスクをに投げ込みExecutorService.invokeAllます。通常の固定スレッドエグゼキュータサービスよりもパフォーマンスが向上します（Luceneへの呼び出しがあり、そこでI / Oが実行されます）。

ここには分割統治法はありません。文字通り、私はします

「モード2」では、単一のタスクをプールに送信し、そのタスクでForkJoinTask.invokeAllを呼び出してサブタスクを送信します。したがって、から継承するオブジェクトがRecursiveActionあり、それがプールに送信されます。そのクラスのcomputeメソッドで、も継承する別invokeAllのクラスのオブジェクトのコレクションでを呼び出します。テストの目的で、最初のオブジェクトを一度に1つずつ送信します。スレッドの呼び出しは、ただ座ってブロックするのではなく、サブタスクの1つを自分自身で取得するため、4つのスレッドすべてがビジーであることがわかります。それがそのように機能しないかもしれないいくつかの理由を考えることができます。RecursiveActioninvokeAll

VisualVMで、モード2で監視している場合、ほとんどの場合、1つのスレッドが待機しています。私が期待しているのは、invokeAllを呼び出すスレッドが、ただじっと座っているだけでなく、呼び出されたタスクの1つですぐに機能することです。これは、通常のスレッドプールでこのスキームを試した結果として生じるデッドロックよりも確かに優れていますが、それでもどうでしょうか。他の何かが提出された場合に備えて、1つのスレッドを保留していますか？そして、もしそうなら、なぜモード1で同じ問題を起こさないのですか？

これまで、Java1.6のブートクラスパスに追加されたjsr166jarを使用してこれを実行してきました。

両方のjavadocは次のとおりです。

1. get（）：必要に応じて計算が完了するのを待ってから、その結果を取得します。
2. invoke（）：このタスクの実行を開始し、必要に応じて完了を待ち、その結果を返すか、基礎となる計算で実行された場合は（チェックされていない）RuntimeExceptionまたはErrorをスローします。

Doug Lea の論文「A Java Fork/Join Framework」では:

http://gee.cs.oswego.edu/dl/papers/fj.pdf

2.1 Work-Stealing で、彼は次のように述べています。

ワーカー スレッドが結合操作に遭遇すると、他のタスクが利用可能であれば、ターゲット タスクが (isDone を介して) 完了したことが通知されるまで処理します。それ以外の場合、すべてのタスクはブロックされることなく完了するまで実行されます。

では、これらの「他のタスク」がどこから来たのか、具体的に教えてもらえますか? それらは他のワーカースレッドのタスクキューからのものですか? ワーカースレッドが結合呼び出しに遭遇するたびに、「独自のキュー内の他のタスクにスキップする」のではなく、「他のスレッドからタスクを盗む」ことに進むということですか?

私が理解している限りForkJoinPool、そのプールは固定数のスレッド（デフォルト：コアの数）を作成し、それ以上のスレッドを作成することはありません（アプリケーションがを使用してスレッドの必要性を示さない限りmanagedBlock）。

しかし、使用するForkJoinPool.getPoolSize()と、30,000個のタスク（RecursiveAction）を作成するプログラムで、ForkJoinPoolそれらのタスクを実行すると、平均で700個のスレッドが使用されることがわかりました（タスクが作成されるたびにスレッドがカウントされます）。タスクはI/Oを実行しませんが、純粋な計算を実行します。タスク間の同期は、の呼び出しForkJoinTask.join()とアクセスのみAtomicBooleanです。つまり、スレッドブロッキング操作はありません。

私が理解しているように、呼び出し元のスレッドをブロックしないのでjoin()、プール内のスレッドがブロックされる理由はありません。したがって、（私が想定していた）それ以上のスレッドを作成する理由はないはずです（それでも明らかに発生しています） 。

では、なぜForkJoinPoolこれほど多くのスレッドを作成するのでしょうか。作成されるスレッドの数を決定する要因は何ですか？

コードを投稿せずにこの質問に答えられることを望んでいましたが、ここではリクエストに応じて提供されます。このコードは、4倍のサイズのプログラムからの抜粋であり、重要な部分に縮小されています。そのままではコンパイルされません。もちろん、必要に応じて、プログラム全体を投稿することもできます。

プログラムは、深さ優先探索を使用して、特定の始点から特定の終点までのパスを迷路で検索します。解決策が存在することが保証されています。主なロジックは次のcompute()方法にありSolverTaskます：ARecursiveActionある特定のポイントで開始し、現在のポイントから到達可能なすべての隣接ポイントで継続します。新しいものを作成するのではなくSolverTask各分岐点（非常に多くのタスクを作成する）で、1つを除くすべてのネイバーをバックトラッキングスタックにプッシュして後で処理し、スタックにプッシュされていない1つのネイバーのみを続行します。そのように行き止まりに達すると、バックトラッキングスタックに最後にプッシュされたポイントがポップされ、そこから検索が続行されます（それに応じて、タックの開始点から構築されたパスが削減されます）。タスクが特定のしきい値よりも大きいバックトラッキングスタックを検出すると、新しいタスクが作成されます。その時から、タスクはバックトラックスタックからポップし続け、それが使い果たされるまで、分岐ポイントに到達したときにスタックにそれ以上のポイントをプッシュしませんが、そのようなポイントごとに新しいタスクを作成します。したがって、タスクのサイズは、スタック制限しきい値を使用して調整できます。

上で引用した数値（「30,000タスク、平均700スレッド」）は、5000x5000セルの迷路を検索したものです。したがって、ここに重要なコードがあります：

Scala 2.10に移行するということですか、それともjsr166yが別途リリースされるということですか？...または、他の何か？

Java7で導入されたFork/Joinフレームワークの実装について読みましたが、魔法がどのように機能するかを理解していることを確認したかっただけです。

私が理解しているように、スレッドがフォークすると、キューにサブタスクが作成されます（他のスレッドが盗む場合と盗まない場合があります）。スレッドが「参加」しようとすると、実際にキューで既存のタスクをチェックしてから再帰的に実行します。つまり、「参加」操作では、2つのフレームがスレッド呼び出しスタックに追加されます（1つは参加用、もう1つは1つ）。新しく実行されたタスクの呼び出し用）。

JVMは末尾呼び出しの最適化（この状況でjoinメソッドのスタックフレームを削除するのに役立つ可能性があります）をサポートしていないことを知っているので、多くのフォークと結合で複雑な操作を実行すると、スレッドがスローされる可能性があると思いますStackOverflowError。

私は正しいですか、それとも彼らはそれを防ぐためのいくつかのクールな方法を見つけましたか？

編集

質問を明確にするのに役立つシナリオを次に示します。（簡単にするために）フォークジョインプールにはスレッドが1つしかないことを伝えます。ある時点で、スレッドはフォークしてからjoinを呼び出します。joinメソッドを使用している間、スレッドは（キューで見つかったように）フォークされたタスクを実行できることを検出し、次のタスクを呼び出します。このタスクは次にフォークしてからjoinを呼び出します。そのため、joinメソッドを実行している間、スレッドは（前と同様に）キュー内でフォークされたタスクを見つけて呼び出します。その段階では、コールスタックには少なくとも2つの結合と2つのタスクのフレームが含まれます。

ご覧のとおり、フォーク結合フレームワークは単純な再帰に変換されています。Javaは末尾呼び出しの最適化をサポートしていないため、Javaでのすべての再帰は、StackOverflowError十分に深くなると発生する可能性があります。

私の質問は-fork/joinフレームワークの実装者は、この状況を防ぐためのクールな方法を見つけましたか？

現在、ビジネス上の問題を解決するために、さまざまな同時実行ソリューションを評価しています。ユースケースは、「恥ずかしい並列」アルゴリズムに似ています。

基本的に、1 つのユーザー リクエストに対して、レスポンスを計算する前に、複数の異なるデータ ソースからデータを取得する必要があります。現在、3 つの DAO 呼び出しはすべてシリアルに行われますが、相互依存関係がないため、並行して行うことができます。

これまでに実装されたソリューション:

• Callable と Future を使用した java.util.concurrent.ExecutorService
• org.springframework.scheduling.annotation.Async を使用して、Spring がスレッド プールを管理できるようにしますが、aysnchronous 呼び出しを行うこともできます。
• 比較的単純なユースケースの Akka (オーバーキルと見なされる)

私が評価したかった最後のフレームワークは Java ForkJoin フレームワークでした。RecursiveTasks の使用例を複数見ることができますが、私のユース ケースは本質的に再帰的ではないため、モデルに適合しません。タスクが十分に小さい場合は、タスクを分割して同じものを再帰的に呼び出します。メソッド (つまり、分割統治)

私のユースケースは、タスクを 3 つのタスクに分割することです。3 つすべてをフォークして、再度参加します。これは ForkJoin 実装の有効な使用例ですか? または、一般的な ExecutorService 実装に固執する必要があります。

作業の盗用は、たとえば、JavaプラットフォームのFork/Joinフレームワークで利用できます。（fork / joinフレームワークはスレッドプールよりも優れていますか？を参照してください）-OmniThreadLibraryで同様のことが可能ですか？

作業の盗用：実行することが不足しているワーカースレッドは、まだビジー状態にある他のスレッドからタスクを盗むことができます。