Можно ли написать решение для рекурсивного объединения вил, используя Executors.newWorkStealingPool()?

Question

Приведенный ниже код предназначен для демонстрации простого использования рекурсивного объединения вил (найти макс.). Я знаю, что Java JIT может добиться этого быстрее в простом однопоточном цикле, но это только для демонстрации.

Первоначально я реализовал find max с использованием фреймворка ForkJoin, который отлично работал для больших массивов двойных чисел (1024*1024).

Я чувствую, что смогу добиться того же самого, не используя инфраструктуру 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

Кажется, что можно написать вычисление типа "fork/join" без фреймворка ForkJoin (см. Использование Callable ниже). Кажется, сама среда ForkJoin не оказывает никакого влияния на производительность, но, возможно, она немного более аккуратна для кода, я предпочитаю просто использовать Callables.

Я также исправил первоначальную попытку. Похоже, что порог был слишком мал при первоначальной попытке, поэтому он был медленным, я думаю, он должен быть как минимум таким же большим, как количество ядер.

Я не уверен, будет ли использование 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());
}

}