分析: ForkJoinPool的性能

Question

问题

由于福克斯连接似乎是当前的炒作，并在许多答案中被推荐，我想:为什么不对它的实际速度做一些研究呢？

为了衡量这一点，我编写了一个小程序(见下面的代码)，它做一些数字的添加，并使用各种参数(包括线程数、分叉深度和分叉扩展)分叉出来，然后测量执行时间，特别是实际计算和分叉所花费的时间。

摘要答案

虽然实现得很好，但ForkJoin是并行任务的一种效率极低的方法，因为每个分叉的成本都很高。一个简单的问题优化实现可以轻松地归档99%的线程执行时间(这比使用Fork度量的所有内容都要好)，因此这样的实现总是比Fork连接实现更快。此外，如果每个叉的实际任务是次要的，那么fork实现甚至可能比单线程线性实现慢得多。

因此，Fork更多地是一个问题，它是否有助于您的代码体系结构，因为它与其他实现相比没有任何性能上的好处。因此，只有在下列情况下才应使用叉-联接：

• 性能并不重要，任务经常需要等待其他任务的结果才能继续。因此，基本上，如果Fork结构大大简化了任务，而不是简单的实现。
• 实际的任务大大减轻了分叉的成本，因此损失可以忽略不计。在我的测试中，一个添加了2个值的循环必须在每个叉子上循环至少10000次才能获得合理的性能。

编辑:有关我所提到的更深入的分析，请参见这里。

测试设置

在我的程序中，我有一个RecursiveTask计算给定N的Fibonacci级数，这将实际计算减少到3个赋值和1个加法。对于任何给定的CPU，这应该是一个次要的任务。

在测试过程中，我改变了线程的数量、每个任务的分叉数量以及Fibonacci循环的长度。此外，我还使用异步参数进行了一些测试，但是将这个参数设置为false只显示计算时间略有减少，所以我跳过了。扩展参数(叉叉)也大多被跳过，结果没有显著差异。

一般来说，计算时间是非常稳定的，实际花费在任务上的时间百分比通常变化不到1%，因此每个测试集在有4个核(+4个超核)的空闲系统上运行了大约5次(或者更多)，然后选择了中间执行时间。

已通过各种测试变量验证了正确的执行，特别是实际使用的线程数已被验证为与最初给定的并行性参数没有任何不同。

详细测试结果

其中：

• Time total是从主线程的角度计算整个计算的总时间。
• Time task是实际计算所有叉的Fibonacci级数的时间总和。
• Time task percentage是线程(时间、任务/时间总数)的相对增益。
• spread->depth是(set)扩展(每叉叉)和(计算)分叉深度.
• threads是实际使用的线程数量。
• task-time/thread是每个线程实际花费在计算斐波纳契级数上的时间。

扩散->深度测试：

Time total: 8766.670 ms, time task: 1717.418 ms ( 19.59%), spread->depth:  2->26, thread#: 1, task-time/thread: 19.59%
Time total: 7872.244 ms, time task: 1421.478 ms ( 18.06%), spread->depth: 10-> 8, thread#: 1, task-time/thread: 18.06%
Time total: 7336.052 ms, time task: 1280.036 ms ( 17.45%), spread->depth: 100-> 4, thread#: 1, task-time/thread: 17.45%

结论:叉子的数量只具有较小的效果(更少的叉子=更好的叉子)，实现似乎相当复杂。类似的结果是收集与其他设置，所以我跳过这些在这里。

Fib(0) (几乎所有的时间都花在叉子上)

Time total: 7866.777 ms, time task: 1421.488 ms ( 18.07%), spread->depth: 10-> 8, thread#: 1, task-time/thread: 18.07%
Time total: 7085.142 ms, time task: 1349.207 ms ( 19.04%), spread->depth: 10-> 8, thread#: 2, task-time/thread:  9.52%
Time total: 6580.609 ms, time task: 1476.467 ms ( 22.44%), spread->depth: 10-> 8, thread#: 4, task-time/thread:  5.61%

结论:对于一个非常小的任务，大部分时间花在分叉上，使得单线程实现比任何Fork连接安装都快5倍。即使使用多个线程，使用Fork也不可能获得任何性能的提高。

Fib(100)

Time total: 12487.634 ms, time task: 5707.720 ms ( 45.71%), spread->depth: 10-> 8, thread#: 1, task-time/thread: 45.71%
Time total:  8386.855 ms, time task: 5768.881 ms ( 68.78%), spread->depth: 10-> 8, thread#: 2, task-time/thread: 34.39%
Time total:  7078.769 ms, time task: 6086.997 ms ( 85.99%), spread->depth: 10-> 8, thread#: 4, task-time/thread: 21.50%

结论:似乎接近单线程执行的盈亏平衡点，而多线程开始产生影响。不过，单线程实现要比任何Fork安装程序都要快。

Fib(1000)

Time total:  5941.344 ms, time task:  5228.258 ms ( 88.00%), spread->depth: 10-> 7, thread#: 1, task-time/thread: 88.00%
Time total:  3160.818 ms, time task:  5244.241 ms (165.91%), spread->depth: 10-> 7, thread#: 2, task-time/thread: 82.96%
Time total: 16301.697 ms, time task: 53351.694 ms (327.28%), spread->depth: 10-> 8, thread#: 4, task-time/thread: 81.82%

结论:多线程执行的次数开始稳定，接近线性增益，而每线程计算时间的20%仍用于分叉。虽然此时分叉可以通过线程来提高性能，但是简单的实现仍然会显着地更快。

Fib(10000)

Time total:  5204.786 ms, time task:  5119.133 ms ( 98.35%), spread->depth: 10-> 6, thread#: 1, task-time/thread: 98.35%
Time total: 26033.889 ms, time task: 51084.118 ms (196.22%), spread->depth: 10-> 7, thread#: 2, task-time/thread: 98.11%
Time total: 13183.573 ms, time task: 51637.471 ms (391.68%), spread->depth: 10-> 7, thread#: 4, task-time/thread: 97.92%

结论:在此数字下，计算出了分叉的成本.虽然简单的实现仍会稍微快一些，但如果任务以另一种方式更难实现，分叉所造成的损失是可以忽略不计的。

码

public class Test {

  static final int NUM_THREADS = 4;
  static final int SPREAD = 10;
  static final int LOOPS = 4000000;
  static final int CALCULATION_N = 10000;
  static final boolean DO_ASYNC = true;
  //---
  static final long MAX_DEPTH = Math.round(Math.log(LOOPS) / Math.log(SPREAD)); // try to have the execution take about the same time

  private static class Task extends RecursiveTask<Integer> {

    final static AtomicLong timeExecute = new AtomicLong(0);
    final static AtomicLong totalLoops = new AtomicLong(0);
    final long depth;

    public Task(final long depth) {
      this.depth = depth;
    }

    @Override
    protected Integer compute() {
      if (depth < MAX_DEPTH) {
        final Task[] subTasks = new Task[SPREAD];
        for (int i = 0; i < subTasks.length; ++i) {
          subTasks[i] = new Task(depth + 1);
        }
        try {
          invokeAll(subTasks);
          final long startTime = System.nanoTime();
          int result = 0;
          for (final Task task : subTasks) {
            if (task.isCompletedNormally()) {
              result += task.get();
            }
          }
          timeExecute.addAndGet(System.nanoTime() - startTime);
          return result;
        } catch (Exception e) {
          this.completeExceptionally(e);
          return null;
        }
      } else {
        totalLoops.incrementAndGet();
        final long startTime = System.nanoTime();
        int a = 0, b = 1, h;
        for (int n = 0; n < CALCULATION_N; ++n) {
          h = b;
          b = a + b;
          a = h;
        }
        timeExecute.addAndGet(System.nanoTime() - startTime);
        return b;
      }
    }
  }

  public static void main(String[] args) {
    final AtomicInteger threadCount = new AtomicInteger(0);
    final ForkJoinPool pool = new ForkJoinPool(NUM_THREADS, new ForkJoinPool.ForkJoinWorkerThreadFactory() {
      @Override
      public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
        threadCount.getAndIncrement();
        final ForkJoinWorkerThread result = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
        result.setPriority(Thread.MIN_PRIORITY);
        return result;
      }
    }, null, DO_ASYNC);
    final long startTime = System.nanoTime();
    final Integer result = pool.invoke(new Task(0));
    final double duration = ((double) (System.nanoTime() - startTime)) / 1000000.0;
    final double executionDuration = ((double) Task.timeExecute.get()) / 1000000.0;
    final double executionPercent = executionDuration / duration * 100.0;
    final double executionPercentPerThread = executionPercent / ((double) NUM_THREADS);

    System.out.println("Completed: " + result + " in " + Task.totalLoops.get() + " loops.");
    System.out.println(String.format("Time total: %8.3f ms, time task: %8.3f ms (%6.2f%%), spread->depth: %2d->%2d, thread#: %1d, task-time/thread: %5.2f%%", duration, executionDuration, executionPercent, SPREAD, MAX_DEPTH, threadCount.get(), executionPercentPerThread));
  }
}

请随时指出错误或提出改进建议。我将接受一些加分的最有价值的答案。

Answer 1

建议：

• 打印已制作的叉子数量+所做工作的成本估算(即，如果切换到分叉，则添加的数量或BigIntegers的长度之和)。这个比例将显示出你的分叉策略有多有效，并让你了解什么是最低的工作规模，这是有意义的。
• 检查您的算法- Fibonacci的指数增长，您的任务返回整数，所以您应该得到溢出很快。

因此，我们的目标是选择一个门槛，这个阈值可以说是分叉，也可以不是分叉：

使用叉/连接并行实现算法时要考虑的主要问题之一是选择阈值，以决定任务是否将执行顺序计算，而不是分叉并行子任务。 如果阈值太大，那么程序可能无法创建足够的任务来充分利用可用的处理器/核心。 如果阈值太小，那么任务创建和管理的开销就会变得很大。 一般来说，需要进行一些实验才能找到合适的阈值。来源

这也可能是有用的：如何确定分叉连接任务的正确分工阈值。

Answer 2

我还没有尝试过您的测试，但是对于任何分治或排队方法，您必须权衡划分工作、队列和作业处理以及聚合作业结果的代价。因此，与单线程版本相比，总CPU周期永远不会有100%的效率。我有另一个基于fibonacci的测试，在这里我尝试设置一个递归限制，以便在同一个线程中递归地计算fib(极限)，而不会为下一个递归级别生成新的作业。因此，用于此递归级别的时间是每个ForkJoinTask所花费的时间。我在实际基准测试之前测量了这段时间，以找出一个任务在最小开销和最大核心利用率之间的最佳平衡应该持续多长时间。对于我测试过的硬件，对于4路机器来说，单套接字x86大约是10到1ms。

Answer 3

你的“测量”有很大的观察效果..。

您可能希望用AtomicLongs代替LongAdder，以减少测量的影响.考虑更多地减少它们..。

使用像JMH这样的框架来减轻基准问题.

你的测量不是任何人都能做出任何非天真结论的东西.

FJP是一个非常好的线程池实现，它是您在JDK中利用cpu内核的最佳选择。

在我的基准测试中(使用JMH)，将FJP与"Legacy“JDK执行器进行比较：

https://github.com/zolyfarkas/spf4j/blob/master/spf4j-benchmarks/src/test/java/org/spf4j/concurrent/ThreadPoolBenchmarkFjp.java

和

https://github.com/zolyfarkas/spf4j/blob/master/spf4j-benchmarks/src/test/java/org/spf4j/concurrent/ThreadPoolBenchmarkStdJdk.java

在jdk 1.7 FJP上运行大约快2倍：

Benchmark                                   Mode  Cnt     Score     Error  Units
ThreadPoolBenchmarkFjp.fjpBenchmark        thrpt   10  6873.926 ± 334.733  ops/s
ThreadPoolBenchmarkStdJdk.stdJdkBenchmark  thrpt   10  3210.170 ± 170.883  ops/s

Jdk 1.8 FJP比Jdk 1.8快3倍：

Benchmark                                   Mode  Cnt     Score      Error  Units
ThreadPoolBenchmarkFjp.fjpBenchmark        thrpt   10  9679.502 ± 1160.887  ops/s
ThreadPoolBenchmarkStdJdk.stdJdkBenchmark  thrpt   10  3466.997 ±   81.594  ops/s