From b7456c038292a2cb9c036fd50dc6fe8e9830a538 Mon Sep 17 00:00:00 2001 From: psandoz Date: Wed, 8 Nov 2017 10:27:10 -0800 Subject: [PATCH] 8190974: Parallel stream execution within a custom ForkJoinPool should obey the parallelism Reviewed-by: martin, tvaleev --- .../java/util/stream/AbstractTask.java | 29 ++-- .../classes/java/util/stream/ForEachOps.java | 5 +- .../java/util/stream/StreamSpliterators.java | 6 +- .../java/util/stream/CustomFJPoolTest.java | 152 ++++++++++++++++++ 4 files changed, 177 insertions(+), 15 deletions(-) create mode 100644 test/java/util/stream/test/org/openjdk/tests/java/util/stream/CustomFJPoolTest.java diff --git a/src/share/classes/java/util/stream/AbstractTask.java b/src/share/classes/java/util/stream/AbstractTask.java index 33de7d5c5..8c5be49af 100644 --- a/src/share/classes/java/util/stream/AbstractTask.java +++ b/src/share/classes/java/util/stream/AbstractTask.java @@ -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 @@ package java.util.stream; 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 @@ abstract class AbstractTask> extends CountedCompleter { - /** - * 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 helper; @@ -156,6 +151,22 @@ abstract class AbstractTask 0L ? est : 1L; } diff --git a/src/share/classes/java/util/stream/ForEachOps.java b/src/share/classes/java/util/stream/ForEachOps.java index b527f054f..38d63e929 100644 --- a/src/share/classes/java/util/stream/ForEachOps.java +++ b/src/share/classes/java/util/stream/ForEachOps.java @@ -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.Objects; 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 @@ final class ForEachOps { 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; } diff --git a/src/share/classes/java/util/stream/StreamSpliterators.java b/src/share/classes/java/util/stream/StreamSpliterators.java index 4f2d2a2c5..42ad027f7 100644 --- a/src/share/classes/java/util/stream/StreamSpliterators.java +++ b/src/share/classes/java/util/stream/StreamSpliterators.java @@ -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 @@ -898,7 +898,7 @@ class StreamSpliterators { * 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. */ static abstract class UnorderedSliceSpliterator> { static final int CHUNK_SIZE = 1 << 7; @@ -915,7 +915,7 @@ class StreamSpliterators { 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); } diff --git a/test/java/util/stream/test/org/openjdk/tests/java/util/stream/CustomFJPoolTest.java b/test/java/util/stream/test/org/openjdk/tests/java/util/stream/CustomFJPoolTest.java new file mode 100644 index 000000000..e2846bf88 --- /dev/null +++ b/test/java/util/stream/test/org/openjdk/tests/java/util/stream/CustomFJPoolTest.java @@ -0,0 +1,152 @@ +/* + * 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. + */ + +package org.openjdk.tests.java.util.stream; + +import org.testng.annotations.Test; + +import java.util.Comparator; +import java.util.Spliterator; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.ForkJoinTask; +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; + +/** + * Tests stream execution in a custom ForkJoinPool. See JDK-8190974. + */ +@Test +public class CustomFJPoolTest { + + // A Spliterator that counts the number of spliterators created + // including itself, thus the count starts at 1 + static class SplitCountingSpliterator implements Spliterator { + final Spliterator s; + final AtomicInteger nsplits; + + // Top-level constructor + public SplitCountingSpliterator(Spliterator s) { + this.s = s; + nsplits = new AtomicInteger(1); + } + + // Splitting constructor + SplitCountingSpliterator(Spliterator s, AtomicInteger nsplits) { + this.s = s; + this.nsplits = nsplits; + } + + int splits() { + return nsplits.get(); + } + + @Override + + public boolean tryAdvance(Consumer action) { + return s.tryAdvance(action); + } + + @Override + public void forEachRemaining(Consumer action) { + s.forEachRemaining(action); + } + + @Override + public Spliterator trySplit() { + Spliterator 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 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 + ForkJoinTask fInteger = fjp.submit(() -> { + Spliterator s = IntStream.range(0, 1024).boxed().parallel().spliterator(); + SplitCountingSpliterator 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)); + } +} -- GitLab