Executors.newWorkStealingPool（）を使用して再帰的なフォーク結合ソリューションを記述することはできますか？

Question

以下のコードでは、再帰的なfork join（find max）を簡単に使用することを目的としています.Java JITはこれをデモ用に単純なシングルスレッドループで高速化できます。

私は最初に、大規模な2倍の配列（1024 * 1024）でうまく動作するForkJoinフレームワークを使用してfind maxを実装しました。

をなしで、ForkJoinフレームワークを使用して、Executor.workStealingPool（）とCallables/Futuresのみを使用して達成できるはずです。

これは可能ですか？

下記の私の試み：

class MaxTask implements Callable<Double> { 

    private double[] array; 
    private ExecutorService executorService; 
    public MaxTask(double[] array, ExecutorService es){ 
     this.array = array; 
     this.executorService = es; 
    } 
    @Override 
    public Double call() throws Exception { 
     if (this.array.length!=2){ 
      double[] a = new double[(this.array.length/2)]; 
      double[] b = new double[(this.array.length/2)]; 
      for (int i=0;i<(this.array.length/2);i++){ 
       a[i] = array[i]; 
       b[i] = array[i+(this.array.length/2)]; 
      } 
      Future<Double> f1 = this.executorService.submit(new MaxTask(a,this.executorService)); 
      Future<Double> f2 = this.executorService.submit(new MaxTask(b,this.executorService)); 

      return Math.max(f1.get(), f2.get()); 
     } else { 
      return Math.max(this.array[0], this.array[1]); 
     } 
    } 

} 

ExecutorService es = Executors.newWorkStealingPool(); 

double[] x = new double[1024*1024]; 
for (int i=0;i<x.length;i++){ 
    x[i] = Math.random(); 
} 

MaxTask mt = new MaxTask(x,es); 

es.submit(mt).get(); 

Answer 1

ForkJoinフレームワークなしタイプの計算を「参加/フォーク」の書き込みにその可能かのように思え（以下コーラブルの使用を参照してください）。 ForkJoinフレームワーク自体はパフォーマンスの違いはないようですが、コード化するのが少し賢明かもしれません。

また、元の試行を修正しました。 元の試行ではしきい値が小さすぎて遅い理由に見えますが、少なくともコア数と同じにする必要があると思います。

ForkJoinPoolを使用する方が高速であれば、より多くの統計情報を収集する必要があります。長い間ブロックしていない操作はないと思います。

public class Main { 

static class FindMaxTask extends RecursiveTask<Double> { 

    private int threshold; 
    private double[] data; 
    private int startIndex; 
    private int endIndex; 

    public FindMaxTask(double[] data, int startIndex, int endIndex, int threshold) { 
     super(); 
     this.data = data; 
     this.startIndex = startIndex; 
     this.endIndex = endIndex; 
     this.threshold = threshold; 
    } 


    @Override 
    protected Double compute() { 
     int diff = (endIndex-startIndex+1); 
     if (diff!=(this.data.length/threshold)){ 
      int aStartIndex = startIndex; 
      int aEndIndex = startIndex + (diff/2) - 1; 
      int bStartIndex = startIndex + (diff/2); 
      int bEndIndex = endIndex; 

      FindMaxTask f1 = new FindMaxTask(this.data,aStartIndex,aEndIndex,threshold); 
      f1.fork(); 
      FindMaxTask f2 = new FindMaxTask(this.data,bStartIndex,bEndIndex,threshold); 
      return Math.max(f1.join(),f2.compute()); 
     } else { 
      double max = Double.MIN_VALUE; 
      for (int i = startIndex; i <= endIndex; i++) { 
       double n = data[i]; 
       if (n > max) { 
        max = n; 
       } 
      } 
      return max; 
     } 
    } 

} 

static class FindMax implements Callable<Double> { 

    private double[] data; 
    private int startIndex; 
    private int endIndex; 
    private int threshold; 

    private ExecutorService executorService; 

    public FindMax(double[] data, int startIndex, int endIndex, int threshold, ExecutorService executorService) { 
     super(); 
     this.data = data; 
     this.startIndex = startIndex; 
     this.endIndex = endIndex; 
     this.executorService = executorService; 
     this.threshold = threshold; 
    } 



    @Override 
    public Double call() throws Exception { 
     int diff = (endIndex-startIndex+1); 
     if (diff!=(this.data.length/this.threshold)){ 
      int aStartIndex = startIndex; 
      int aEndIndex = startIndex + (diff/2) - 1; 
      int bStartIndex = startIndex + (diff/2); 
      int bEndIndex = endIndex; 

      Future<Double> f1 = this.executorService.submit(new FindMax(this.data,aStartIndex,aEndIndex,this.threshold,this.executorService)); 
      Future<Double> f2 = this.executorService.submit(new FindMax(this.data,bStartIndex,bEndIndex,this.threshold,this.executorService)); 
      return Math.max(f1.get(), f2.get()); 
     } else { 
      double max = Double.MIN_VALUE; 
      for (int i = startIndex; i <= endIndex; i++) { 
       double n = data[i]; 
       if (n > max) { 
        max = n; 
       } 
      } 
      return max; 
     } 
    } 

} 

public static void main(String[] args) throws InterruptedException, ExecutionException { 

    double[] data = new double[1024*1024*64]; 
    for (int i=0;i<data.length;i++){ 
     data[i] = Math.random(); 
    } 

    int p = Runtime.getRuntime().availableProcessors(); 
    int threshold = p; 
    int threads = p; 
    Instant start = null; 
    Instant end = null; 

    ExecutorService es = null; 
    es = Executors.newFixedThreadPool(threads); 
    System.out.println("1. started.."); 
    start = Instant.now(); 
    System.out.println("max = "+es.submit(new FindMax(data,0,data.length-1,threshold,es)).get()); 
    end = Instant.now(); 
    System.out.println("Callable (recrusive), with fixed pool, Find Max took ms = "+ Duration.between(start, end).toMillis()); 

    es = new ForkJoinPool(); 
    System.out.println("2. started.."); 
    start = Instant.now(); 
    System.out.println("max = "+es.submit(new FindMax(data,0,data.length-1,threshold,es)).get()); 
    end = Instant.now(); 
    System.out.println("Callable (recursive), with fork join pool, Find Max took ms = "+ Duration.between(start, end).toMillis()); 

    ForkJoinPool fj = new ForkJoinPool(threads); 
    System.out.println("3. started.."); 
    start = Instant.now(); 
    System.out.println("max = "+fj.invoke(new FindMaxTask(data,0,data.length-1,threshold))); 
    end = Instant.now(); 
    System.out.println("RecursiveTask (fork/join framework),with fork join pool, Find Max took ms = "+ Duration.between(start, end).toMillis()); 
} 

}