8190974: Parallel stream execution within a custom ForkJoinPool should obey the parallelism
Reviewed-by: martin, tvaleev
--- a/src/java.base/share/classes/java/util/stream/AbstractTask.java Wed Nov 15 11:50:55 2017 -0800
+++ b/src/java.base/share/classes/java/util/stream/AbstractTask.java Wed Nov 08 10:27:10 2017 -0800
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012, 2017, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
@@ -27,6 +27,7 @@
import java.util.Spliterator;
import java.util.concurrent.CountedCompleter;
import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
/**
* Abstract base class for most fork-join tasks used to implement stream ops.
@@ -88,13 +89,7 @@
K extends AbstractTask<P_IN, P_OUT, R, K>>
extends CountedCompleter<R> {
- /**
- * Default target factor of leaf tasks for parallel decomposition.
- * To allow load balancing, we over-partition, currently to approximately
- * four tasks per processor, which enables others to help out
- * if leaf tasks are uneven or some processors are otherwise busy.
- */
- static final int LEAF_TARGET = ForkJoinPool.getCommonPoolParallelism() << 2;
+ private static final int LEAF_TARGET = ForkJoinPool.getCommonPoolParallelism() << 2;
/** The pipeline helper, common to all tasks in a computation */
protected final PipelineHelper<P_OUT> helper;
@@ -157,6 +152,22 @@
}
/**
+ * Default target of leaf tasks for parallel decomposition.
+ * To allow load balancing, we over-partition, currently to approximately
+ * four tasks per processor, which enables others to help out
+ * if leaf tasks are uneven or some processors are otherwise busy.
+ */
+ public static int getLeafTarget() {
+ Thread t = Thread.currentThread();
+ if (t instanceof ForkJoinWorkerThread) {
+ return ((ForkJoinWorkerThread) t).getPool().getParallelism() << 2;
+ }
+ else {
+ return LEAF_TARGET;
+ }
+ }
+
+ /**
* Constructs a new node of type T whose parent is the receiver; must call
* the AbstractTask(T, Spliterator) constructor with the receiver and the
* provided Spliterator.
@@ -181,7 +192,7 @@
* @return suggested target leaf size
*/
public static long suggestTargetSize(long sizeEstimate) {
- long est = sizeEstimate / LEAF_TARGET;
+ long est = sizeEstimate / getLeafTarget();
return est > 0L ? est : 1L;
}
--- a/src/java.base/share/classes/java/util/stream/ForEachOps.java Wed Nov 15 11:50:55 2017 -0800
+++ b/src/java.base/share/classes/java/util/stream/ForEachOps.java Wed Nov 08 10:27:10 2017 -0800
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012, 2017, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
@@ -28,7 +28,6 @@
import java.util.Spliterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountedCompleter;
-import java.util.concurrent.ForkJoinTask;
import java.util.function.Consumer;
import java.util.function.DoubleConsumer;
import java.util.function.IntConsumer;
@@ -378,7 +377,7 @@
this.spliterator = spliterator;
this.targetSize = AbstractTask.suggestTargetSize(spliterator.estimateSize());
// Size map to avoid concurrent re-sizes
- this.completionMap = new ConcurrentHashMap<>(Math.max(16, AbstractTask.LEAF_TARGET << 1));
+ this.completionMap = new ConcurrentHashMap<>(Math.max(16, AbstractTask.getLeafTarget() << 1));
this.action = action;
this.leftPredecessor = null;
}
--- a/src/java.base/share/classes/java/util/stream/StreamSpliterators.java Wed Nov 15 11:50:55 2017 -0800
+++ b/src/java.base/share/classes/java/util/stream/StreamSpliterators.java Wed Nov 08 10:27:10 2017 -0800
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2012, 2016, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012, 2017, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
@@ -897,7 +897,7 @@
* Note: The source spliterator may report {@code ORDERED} since that
* spliterator be the result of a previous pipeline stage that was
* collected to a {@code Node}. It is the order of the pipeline stage
- * that governs whether the this slice spliterator is to be used or not.
+ * that governs whether this slice spliterator is to be used or not.
*/
abstract static class UnorderedSliceSpliterator<T, T_SPLITR extends Spliterator<T>> {
static final int CHUNK_SIZE = 1 << 7;
@@ -914,7 +914,7 @@
this.unlimited = limit < 0;
this.skipThreshold = limit >= 0 ? limit : 0;
this.chunkSize = limit >= 0 ? (int)Math.min(CHUNK_SIZE,
- ((skip + limit) / AbstractTask.LEAF_TARGET) + 1) : CHUNK_SIZE;
+ ((skip + limit) / AbstractTask.getLeafTarget()) + 1) : CHUNK_SIZE;
this.permits = new AtomicLong(limit >= 0 ? skip + limit : skip);
}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/test/jdk/java/util/stream/CustomFJPoolTest.java Wed Nov 08 10:27:10 2017 -0800
@@ -0,0 +1,154 @@
+/*
+ * Copyright (c) 2017, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation. Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+/*
+ * @test
+ * @summary Tests stream execution in a custom ForkJoinPool
+ * @bug 8190974
+ * @run testng/othervm CustomFJPoolTest
+ * @run testng/othervm -Djava.util.concurrent.ForkJoinPool.common.parallelism=0 CustomFJPoolTest
+ */
+
+import org.testng.annotations.Test;
+
+import java.util.Comparator;
+import java.util.Spliterator;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.stream.IntStream;
+import java.util.stream.StreamSupport;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+@Test
+public class CustomFJPoolTest {
+
+ // A Spliterator that counts the number of spliterators created
+ // including itself, thus the count starts at 1
+ static class SplitCountingSpliterator<T> implements Spliterator<T> {
+ final Spliterator<T> s;
+ final AtomicInteger nsplits;
+
+ // Top-level constructor
+ public SplitCountingSpliterator(Spliterator<T> s) {
+ this.s = s;
+ nsplits = new AtomicInteger(1);
+ }
+
+ // Splitting constructor
+ SplitCountingSpliterator(Spliterator<T> s, AtomicInteger nsplits) {
+ this.s = s;
+ this.nsplits = nsplits;
+ }
+
+ int splits() {
+ return nsplits.get();
+ }
+
+ @Override
+
+ public boolean tryAdvance(Consumer<? super T> action) {
+ return s.tryAdvance(action);
+ }
+
+ @Override
+ public void forEachRemaining(Consumer<? super T> action) {
+ s.forEachRemaining(action);
+ }
+
+ @Override
+ public Spliterator<T> trySplit() {
+ var split = s.trySplit();
+ if (split != null) {
+ nsplits.incrementAndGet();
+ return new SplitCountingSpliterator<>(split, nsplits);
+ }
+ else {
+ return null;
+ }
+ }
+
+ @Override
+ public long estimateSize() {
+ return s.estimateSize();
+ }
+
+ @Override
+ public long getExactSizeIfKnown() {
+ return s.getExactSizeIfKnown();
+ }
+
+ @Override
+ public int characteristics() {
+ return s.characteristics();
+ }
+
+ @Override
+ public boolean hasCharacteristics(int characteristics) {
+ return s.hasCharacteristics(characteristics);
+ }
+
+ @Override
+ public Comparator<? super T> getComparator() {
+ return s.getComparator();
+ }
+ }
+
+ public void testCustomPools() throws Exception {
+ int splitsForP1 = countSplits(new ForkJoinPool(1));
+ int splitsForP2 = countSplits(new ForkJoinPool(2));
+ assertEquals(splitsForP2, splitsForP1 * 2);
+
+ int commonParallelism = ForkJoinPool.getCommonPoolParallelism();
+ if (commonParallelism > 1 && commonParallelism < 128) {
+ int splitsForPHalfC = countSplits(new ForkJoinPool(commonParallelism / 2));
+ int splitsForPC = countSplits(ForkJoinPool.commonPool());
+
+ assertTrue(splitsForPHalfC < splitsForPC);
+ assertEquals(splitsForPC / splitsForPHalfC,
+ nearestPowerOfTwo(commonParallelism) / nearestPowerOfTwo(commonParallelism / 2));
+ }
+ }
+
+ static int countSplits(ForkJoinPool fjp) throws Exception {
+ // The number of splits will be equivalent to the number of leaf nodes
+ // and will be a power of 2
+ var fInteger = fjp.submit(() -> {
+ var s = IntStream.range(0, 1024).boxed().parallel().spliterator();
+ var cs = new SplitCountingSpliterator<>(s);
+ StreamSupport.stream(cs, true).forEach(e -> {});
+ return cs.splits();
+ });
+ return fInteger.get();
+ }
+
+ static int nearestPowerOfTwo(int i) {
+ return (i & (i - 1)) == 0
+ ? i
+ : 1 << (32 - Integer.numberOfLeadingZeros(i));
+ }
+}