嵌套的Java8并行forEach循环性能很差。这种行为在意料之中吗？

Question

注意：我已经在另一篇文章中解决了这个问题，所以在嵌套的Java8并行流操作中使用信号量可能会造成死锁。这是窃听器吗？-，但这篇文章的标题暗示问题与一个信号量的使用有关--这多少分散了讨论的注意力。我创建这个例子是为了强调嵌套循环可能存在性能问题--尽管这两个问题可能有一个共同的原因（也许是因为我花了很多时间来解决这个问题）。（我不认为它是一个重复，因为它强调了另一个症状--但如果你真的这么做了，就把它删掉吧）。

问题：如果嵌套两个Java8stream.parallel（）.foreach循环，并且所有任务都是独立的、无状态的等等--除了提交到公共的FJ池之外--那么在并行循环中嵌套并行循环的性能要比在并行循环中嵌套顺序循环的性能差得多。更糟糕的是：如果包含内部循环的操作是同步的，您将会得到死锁。

绩效问题的论证

没有“synchronized”，您仍然可以观察到性能问题。您可以在http://svn.finmath.net/finmath%20experiments/trunk/src/net/finmath/experiments/concurrency/nestedParallelForeachTest.java找到一个演示代码（更详细的描述请参阅此处的JavaDoc）。

我们在这里的设置如下：我们有一个嵌套的stream.parallel（）.foreach（）。

• 内部循环是独立的（无状态、无干扰等--除了使用公共池之外），在最坏的情况下（即顺序处理时）总共消耗1秒。
• 外部循环的一半任务在该循环之前消耗10秒。
• 循环后10秒消耗一半。
• 因此每个线程总共消耗11秒（最坏情况）。 *我们有一个布尔值，它允许将内部循环从parallel（）切换到sequential（）。

现在：将24个外循环任务提交到并行度为8的池中，我们最多期望24/8*11=33秒（在8核或更好的机器上）。

• 内部顺序循环：33秒。
• 内部并行循环:>80秒（我有92秒）。

问题：你能证实这一行为吗？这是一个可以从框架中得到的东西吗？（我现在更小心了一点，声称这是一个bug，但我个人认为这是由于ForkJointAsk实现中的一个bug造成的。备注：我已经将此发布到了concurrency-interest（参见http://cs.oswego.edu/pipermail/concurrency-interest/2014-may/012652.html)，但到目前为止我还没有从那里得到证实）。

僵局演示

以下代码将死锁

    // Outer loop
    IntStream.range(0,numberOfTasksInOuterLoop).parallel().forEach(i -> {
        doWork();
        synchronized(this) {
            // Inner loop
            IntStream.range(0,numberOfTasksInInnerLoop).parallel().forEach(j -> {
                doWork();
            });
        }
    });

其中NumberOfTaskSinouterLoop=24,NumberOfTaskSininnerLoop=240,OuterLoopOverHeadFactor=10000和DoWork是某个无状态CPU刻录机。

您可以在http://svn.finmath.net/finMath%20experiments/trunk/src/net/finMath/experiments/concurrency/nestedParallelForeachandsynchronization.java找到完整的演示代码（更详细的描述请参阅此处的JavaDoc）。

这种行为在意料之中吗？请注意，关于Java并行流的文档没有提到嵌套或同步的任何问题。另外，没有提到两者都使用一个公共的fork-join-pool。

更新

看起来这个问题和信号量更严重的死锁在Java8 U40中已经解决了。

Answer 1

问题是，您配置的相当有限的并行性被外部流处理吃掉了：如果您说您需要8个线程，并使用parallel（）处理一个超过8个项目的流，它将创建8个工作线程，并让它们处理项目。

然后，在您的使用者中，您正在使用parallel（）处理另一个流，但是没有工作线程了。由于工作线程在等待内部流处理结束时被阻塞，因此forkjoinpool必须创建新的工作线程，这违反了您配置的并行性。在我看来，它并没有回收这些扩展线程，而是让它们在处理完后立即死亡。因此，在您的内部处理中，创建并释放新线程，这是一个昂贵的操作。

您可能会将其视为一个缺陷，即启动线程不对并行流处理的计算做出贡献，而只是等待结果，但即使修复了该问题，您仍然有一个很难修复的一般性问题：

每当工作线程数量与外部流项之间的比率较低时，实现就会将它们全部用于外部流，因为它不知道该流是外部流。因此并行执行内部流请求的工作线程比可用线程多。使用调用方线程对计算做出贡献可以使计算的性能与串行计算相等，但在这里获得并行执行的优势与固定数量的工作线程的概念不能很好地工作。

请注意，您在这里只是触及了问题的表面，因为您对项目的处理时间相当平衡。如果两者的处理，内部项目和外部项目，发散（与同一级别的项目相比），问题将会更糟。

更新：通过分析和查看代码，forkjoinpool似乎尝试使用等待线程进行“工作窃取”，但根据thread是工作线程还是其他线程使用不同的代码。因此，一个工作线程实际上是等待大约80%的时间，并做很少或没有工作，而其他线程真正贡献了计算…

更新2：为了完整起见，这里使用注释中描述的简单并行执行方法。由于它将每个项目都排入队列，当单个项目的执行时间相当小时，它将会有很大的开销。所以这不是一个复杂的解决方案，而是一个演示，它可以处理长时间运行的任务，而不需要太多的魔法…

import java.lang.reflect.UndeclaredThrowableException;
import java.util.concurrent.*;
import java.util.function.IntConsumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class NestedParallelForEachTest1 {
    static final boolean isInnerStreamParallel = true;

    // Setup: Inner loop task 0.01 sec in worse case. Outer loop task: 10 sec + inner loop. This setup: (100 * 0.01 sec + 10 sec) * 24/8 = 33 sec.
    static final int numberOfTasksInOuterLoop = 24;  // In real applications this can be a large number (e.g. > 1000).
    static final int numberOfTasksInInnerLoop = 100; // In real applications this can be a large number (e.g. > 1000).
    static final int concurrentExecutionsLimitForStreams = 8;

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        System.out.println(System.getProperty("java.version")+" "+System.getProperty("java.home"));
        new NestedParallelForEachTest1().testNestedLoops();
        E.shutdown();
    }

    final static ThreadPoolExecutor E = new ThreadPoolExecutor(
        concurrentExecutionsLimitForStreams, concurrentExecutionsLimitForStreams,
        2, TimeUnit.MINUTES, new SynchronousQueue<>(), (r,e)->r.run() );

    public static void parallelForEach(IntStream s, IntConsumer c) {
        s.mapToObj(i->E.submit(()->c.accept(i))).collect(Collectors.toList())
         .forEach(NestedParallelForEachTest1::waitOrHelp);
    }
    static void waitOrHelp(Future f) {
        while(!f.isDone()) {
            Runnable r=E.getQueue().poll();
            if(r!=null) r.run();
        }
        try { f.get(); }
        catch(InterruptedException ex) { throw new RuntimeException(ex); }
        catch(ExecutionException eex) {
            Throwable t=eex.getCause();
            if(t instanceof RuntimeException) throw (RuntimeException)t;
            if(t instanceof Error) throw (Error)t;
            throw new UndeclaredThrowableException(t);
        }
    }
    public void testNestedLoops(NestedParallelForEachTest1 this) {
        long start = System.nanoTime();
        // Outer loop
        parallelForEach(IntStream.range(0,numberOfTasksInOuterLoop), i -> {
            if(i < 10) sleep(10 * 1000);
            if(isInnerStreamParallel) {
                // Inner loop as parallel: worst case (sequential) it takes 10 * numberOfTasksInInnerLoop millis
                parallelForEach(IntStream.range(0,numberOfTasksInInnerLoop), j -> sleep(10));
            }
            else {
                // Inner loop as sequential
                IntStream.range(0,numberOfTasksInInnerLoop).sequential().forEach(j -> sleep(10));
            }
            if(i >= 10) sleep(10 * 1000);
        });
        long end = System.nanoTime();
        System.out.println("Done in "+TimeUnit.NANOSECONDS.toSeconds(end-start)+" sec.");
    }
    static void sleep(int milli) {
        try {
            Thread.sleep(milli);
        } catch (InterruptedException ex) {
            throw new AssertionError(ex);
        }
    }
}