Verschachtelte Java 8 parallele forEach-Schleife durchführen Armen. Ist dieses Verhalten zu erwarten?

Das problem ist, dass die eher begrenzte Parallelität, den Sie konfiguriert haben, wird aufgefressen durch die äußere stream-Verarbeitung: wenn Sie sagen, dass Sie wollen, acht threads und verarbeiten eines Stroms von mehr als acht Elemente mit parallel() wird es schaffen, acht worker-threads und lassen Sie Sie verarbeiten Elemente.

Dann in Ihren Verbraucher Sie sind der Verarbeitung ein anderer stream mit parallel() aber es gibt keine worker-threads zu linken. Da die worker-threads werden blockiert, wartet auf das Ende des der innere stream-Verarbeitung, die ForkJoinPool hat, neues zu schaffen worker-threads, die gegen Ihre Parallelität konfiguriert. Es scheint mir, dass es nicht wieder diese erstrecken sich Fäden, aber Sie sterben zu lassen, rechts nach der Bearbeitung. Also in Ihrem inneren Verarbeitung, neue threads erstellt werden und entsorgt, das ist ein teurer Vorgang.

Könnten Sie sehen es als Makel, dass die Einleitung von threads, nicht zur Berechnung einer parallelen stream-Verarbeitung aber nur auf das Ergebnis zu warten aber selbst wenn das behoben wurde haben Sie immer noch ein Allgemeines problem, das ist schwer (wenn überhaupt) zu beheben ist:

Wenn das Verhältnis zwischen der Anzahl der worker-threads, um äußere stream-Elemente ist gering, die Umsetzung verwenden Sie für den äußeren Strom, da Sie nicht wissen, dass der stream ist eine äußere stream. Also die Ausführung einer inneren Datenstrom in parallele Anfragen mehr worker-threads zur Verfügung. Mit den caller-thread für den Beitrag zur Berechnung beheben könnte es in einer Weise, dass die Leistung entspricht dem seriellen Berechnung aber immer einen Vorteil der parallelen Ausführung hier funktioniert nicht gut mit dem Konzept einer festen Anzahl von worker-threads.

Beachten Sie, dass Sie Kratzer auf der Oberfläche dieses problem hier, da haben Sie Recht ausgewogen Bearbeitungszeiten für die Elemente. Wenn die Verarbeitung der beiden, die inneren Elemente und die äußeren Elemente, abweichen (im Vergleich zu Elementen, die auf der gleichen Ebene), das problem wird sogar noch schlimmer.

Update: durch profiling und den code anzuschauen, es scheint, dass die ForkJoinPool hat versucht, die wartenden thread für "work stealing", sondern mit unterschiedlichen code abhängig von der Tatsache, ob die Thread ist ein worker-thread oder einem anderen thread. Als ein Ergebnis, ein worker-thread ist eigentlich warten über 80% der Zeit, und tun sehr wenig, um nicht zu arbeiten, während andere threads wirklich dazu beitragen, um die Berechnung...

Update 2: der Vollständigkeit halber, hier die einfache parallel-execution-Ansatz wie beschrieben in die Kommentare. Da es reiht jedes item es ist zu erwarten, dass zu viel overhead, wenn die Ausführungszeit für einen einzelnen Punkt ist eher klein. So ist es nicht eine anspruchsvolle Lösung, sondern eher eine demonstration, dass es möglich ist, mit langen ausführen-Aufgaben ohne viel Magie...

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);
        }
    }
}

Nach dem Aufräumen der code ein wenig. Ich sehe nicht die gleichen Ergebnisse mit Java 8 update 45. Es ist zweifellos ein Aufwand, aber es ist sehr klein im Vergleich zu den Zeitspannen, die Sie sprechen.

Das Potenzial für einen deadlock zu rechnen, als Sie verbrauchen alle verfügbaren threads im pool mit der äußeren Schleife, so dass Sie keine threads Links zur Ausführung der inneren Schleife.

Folgende Programm druckt

isInnerStreamParallel: false, isCPUTimeBurned: false
java.util.concurrent.ForkJoinPool.common.parallelism = 8
Done in 33.1 seconds.
isInnerStreamParallel: false, isCPUTimeBurned: true
java.util.concurrent.ForkJoinPool.common.parallelism = 8
Done in 33.0 seconds.
isInnerStreamParallel: true, isCPUTimeBurned: false
java.util.concurrent.ForkJoinPool.common.parallelism = 8
Done in 32.5 seconds.
isInnerStreamParallel: true, isCPUTimeBurned: true
java.util.concurrent.ForkJoinPool.common.parallelism = 8
Done in 32.6 seconds.

Den code

import java.util.stream.IntStream;

public class NestedParallelForEachTest {
    //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;    //java.util.concurrent.ForkJoinPool.common.parallelism

    public static void main(String[] args) {
        testNestedLoops(false, false);
        testNestedLoops(false, true);
        testNestedLoops(true, false);
        testNestedLoops(true, true);
    }

    public static void testNestedLoops(boolean isInnerStreamParallel, boolean isCPUTimeBurned) {
        System.out.println("isInnerStreamParallel: " + isInnerStreamParallel + ", isCPUTimeBurned: " + isCPUTimeBurned);
        long start = System.nanoTime();

        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism",Integer.toString(concurrentExecutionsLimitForStreams));
        System.out.println("java.util.concurrent.ForkJoinPool.common.parallelism = " + System.getProperty("java.util.concurrent.ForkJoinPool.common.parallelism"));

        //Outer loop
        IntStream.range(0, numberOfTasksInOuterLoop).parallel().forEach(i -> {
//           System.out.println(i + "\t" + Thread.currentThread());
            if(i < 10) burnTime(10 * 1000, isCPUTimeBurned);

            IntStream range = IntStream.range(0, numberOfTasksInInnerLoop);
            if (isInnerStreamParallel) {
                //Inner loop as parallel: worst case (sequential) it takes 10 * numberOfTasksInInnerLoop millis
                range = range.parallel();
            } else {
                //Inner loop as sequential
            }
            range.forEach(j -> burnTime(10, isCPUTimeBurned));

            if(i >= 10) burnTime(10 * 1000, isCPUTimeBurned);
        });

        long end = System.nanoTime();

        System.out.printf("Done in %.1f seconds.%n", (end - start) / 1e9);
    }

    static void burnTime(long millis, boolean isCPUTimeBurned) {
        if (isCPUTimeBurned) {
            long end = System.nanoTime() + millis * 1000000;
            while (System.nanoTime() < end)
                ;

        } else {
            try {
                Thread.sleep(millis);
            } catch (InterruptedException e) {
                throw new AssertionError(e);
            }
        }
    }
}