Java ForkJoinPool 未利用所有 CPU 核心

Question

我正在运行跟踪虚拟墙节点上的代码。该节点有 32 个 Intel Xeon E7/E5 内核和 128GB RAM。监控 CPU 使用情况表明该节点远未满负荷运行。由于节点大小，此问题不同于大多数 fork-join 问题。有时节点在多个内核上有 20%+ CPU 负载，显示出并行的迹象，但我似乎无法让它使用更多资源。

提供一些背景信息；该问题是 111 个节点 (Parcs/parken) 图中的最大化问题。每个公园内都隐藏着许多鸡蛋。每过一秒，这个数字就会呈指数下降。目标是在时间到期之前获得尽可能多的鸡蛋。 'opl' 是我使用贪婪算法找到的解决方案，因此为了缩小我们的递归树，我只允许递归，当我们发现最多 5 个鸡蛋比我们的贪婪算法同时发现的鸡蛋少时。

我熟悉（多）线程，但远非专家。我以前没有使用过很多 ForkJoinPools。我也尝试将 ForkJoinPool 参数设置为 16/32，但没有成功。

Example of current core-load

主要：

Algoritmes.AlgoritmeRecursive.run(new AlgoritmeRecursive(parken, tabel, opl, 22, 1000, 0, 0)));

Class:

public static class AlgoritmeRecursive extends RecursiveTask<Double> {
    private ArrayList<Park> parken = new ArrayList<Park>();
    private double[][] afstandenTabel;
    private double[][] oplossing;
    private int startpark;
    private double duur;
    private double eieren;
    private int time;

    AlgoritmeRecursive(ArrayList<Park> parken, double[][] afstandenTabel, double[][] oplossing, int startpark, double duur, double eieren, int time) {
        for (Park p : parken) {
            this.parken.add(new Park(p));
        }
        this.afstandenTabel = afstandenTabel;
        this.oplossing = oplossing;
        this.startpark = startpark;
        this.duur = duur;
        this.eieren = eieren;
        this.time = time;
    }

    public static double run(AlgoritmeRecursive ar) {
        ForkJoinPool pool= new ForkJoinPool();
        return pool.invoke(ar);
    }

    protected Double compute() {
        if (duur < 1.0) return eieren;

        double gevonden = 0;

        /* startpark zoeken adhv gegeven naam */
        for (Park p : parken) {
            if (p.getId() == startpark) {
                gevonden = p.verwachtAantalEieren(40, 0);
                p.updateEggs(p.getEggs() * exp((-1.0/10800.0) * ((p.getStartEggs()/20.0) + 40.0)));
            }
            else {
                p.updateEggs(p.getEggs() * exp((-1.0/10800.0) * (p.getStartEggs()/20.0)));
            }
        }
        double score = eieren;
        for (Park p : parken) {
            if (p.getId() == startpark && eieren >= (oplossing[1000-(int)duur][1] - 5)) {
                AlgoritmeRecursive ar = new AlgoritmeRecursive(parken, afstandenTabel, oplossing, startpark, duur-1, eieren + gevonden, time+1);
                ar.fork();
                double res = ar.join();
                if(res > score) score = res;
            }
            else if (duur-afstandenTabel[startpark][p.getId()] > 60.0 && time > 120.0 && eieren >= oplossing[1000-(int)duur][1] && gevonden < p.verwachtAantalEieren(40,afstandenTabel[startpark][p.getId()])){
                AlgoritmeRecursive ar = new AlgoritmeRecursive(parken, afstandenTabel, oplossing, p.getId(), duur-afstandenTabel[startpark][p.getId()], eieren, 0);
                for (Park p2 : ar.parken) {
                    p2.updateEggs(p2.getEggs() * exp((-1.0/10800.0) * (p2.getStartEggs()/20.0) * (afstandenTabel[startpark][p.getId()]-1)));
                }
                ar.fork();
                double res = ar.join();
                if(res > score) score = res;
            }
        }
        return score;
    }
    public double exp(double x) {
          x = 1d + x / 256d;
          x *= x; x *= x; x *= x; x *= x;
          x *= x; x *= x; x *= x; x *= x;
          return x;
    }
}

Answer 1

我自己对此不是很熟悉，但是对 ar.join() 的调用是否会使您的 RecursiveTask 等待子任务完成？如果是这种情况，您的其他任务将不会在前一个任务完成之前开始运行.

您可以尝试将运行任务存储在一个列表中，然后再加入它们。这将有希望确保您所有的子任务在您等待它们之前开始运行。

像这样（修改 compute 中的第二个循环）：

List<AlgoritmeRecursive> tasks = new ArrayList<>();

for (Park p : parken) {
    if (p.getId() == startpark && eieren >= (oplossing[1000-(int)duur][1] - 5)) {

        AlgoritmeRecursive ar = new AlgoritmeRecursive(parken, afstandenTabel, oplossing, startpark, duur-1, eieren + gevonden, time+1);
        ar.fork();
        tasks.add(ar); // Adding the running task to the list.

    } else if (duur-afstandenTabel[startpark][p.getId()] > 60.0 && time > 120.0 && eieren >= oplossing[1000-(int)duur][1] && gevonden < p.verwachtAantalEieren(40,afstandenTabel[startpark][p.getId()])){

        AlgoritmeRecursive ar = new AlgoritmeRecursive(parken, afstandenTabel, oplossing, p.getId(), duur-afstandenTabel[startpark][p.getId()], eieren, 0);
        for (Park p2 : ar.parken) {
            p2.updateEggs(p2.getEggs() * exp((-1.0/10800.0) * (p2.getStartEggs()/20.0) * (afstandenTabel[startpark][p.getId()]-1)));
        }
        ar.fork();
        tasks.add(ar); // Adding the running task to the list.

    }
}

double score = eieren;
for(AlgoritmeRecursive task : tasks) {
    double res = ar.join();
    if(res > score) score = res;
}

return score;

Answer 2

我认为问题在于你的算法的递归部分是这样的：

    for (...) {
        // ar <- create sub-problem
        ar.fork();
        double res = ar.join();
        // Use result
    }

问题是，当您分叉然后立即加入时，没有两个或多个子问题并行运行的范围。这与使用经典线程执行此操作一样：

    Thread t = new Thread(someRunnable);
    t.start();
    t.join();

即启动一个新线程，并立即阻塞当前线程，直到新线程结束；即它是有效单线程。这样做效率更高：

    someRunnable.run();

尝试在一个循环中分叉，在另一个循环中加入。

Java ForkJoinPool 未利用所有 CPU 核心

Java ForkJoinPool not utilising all CPU cores

java

fork-join

forkjoinpool