Back to index...

	/*
	* 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
	* 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 java.util.stream;

	import java.util.Objects;
	import java.util.Spliterator;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.CountedCompleter;
	import java.util.function.Consumer;
	import java.util.function.DoubleConsumer;
	import java.util.function.IntConsumer;
	import java.util.function.IntFunction;
	import java.util.function.LongConsumer;

	/**
	* Factory for creating instances of {@code TerminalOp} that perform an
	* action for every element of a stream. Supported variants include unordered
	* traversal (elements are provided to the {@code Consumer} as soon as they are
	* available), and ordered traversal (elements are provided to the
	* {@code Consumer} in encounter order.)
	*
	* <p>Elements are provided to the {@code Consumer} on whatever thread and
	* whatever order they become available. For ordered traversals, it is
	* guaranteed that processing an element <em>happens-before</em> processing
	* subsequent elements in the encounter order.
	*
	* <p>Exceptions occurring as a result of sending an element to the
	* {@code Consumer} will be relayed to the caller and traversal will be
	* prematurely terminated.
	*
	* @since 1.8
	*/
	final class ForEachOps {

	private ForEachOps() { }

	/**
	* Constructs a {@code TerminalOp} that perform an action for every element
	* of a stream.
	*
	* @param action the {@code Consumer} that receives all elements of a
	* stream
	* @param ordered whether an ordered traversal is requested
	* @param <T> the type of the stream elements
	* @return the {@code TerminalOp} instance
	*/
	public static <T> TerminalOp<T, Void> makeRef(Consumer<? super T> action,
	boolean ordered) {
	Objects.requireNonNull(action);
	return new ForEachOp.OfRef<>(action, ordered);
	}

	/**
	* Constructs a {@code TerminalOp} that perform an action for every element
	* of an {@code IntStream}.
	*
	* @param action the {@code IntConsumer} that receives all elements of a
	* stream
	* @param ordered whether an ordered traversal is requested
	* @return the {@code TerminalOp} instance
	*/
	public static TerminalOp<Integer, Void> makeInt(IntConsumer action,
	boolean ordered) {
	Objects.requireNonNull(action);
	return new ForEachOp.OfInt(action, ordered);
	}

	/**
	* Constructs a {@code TerminalOp} that perform an action for every element
	* of a {@code LongStream}.
	*
	* @param action the {@code LongConsumer} that receives all elements of a
	* stream
	* @param ordered whether an ordered traversal is requested
	* @return the {@code TerminalOp} instance
	*/
	public static TerminalOp<Long, Void> makeLong(LongConsumer action,
	boolean ordered) {
	Objects.requireNonNull(action);
	return new ForEachOp.OfLong(action, ordered);
	}

	/**
	* Constructs a {@code TerminalOp} that perform an action for every element
	* of a {@code DoubleStream}.
	*
	* @param action the {@code DoubleConsumer} that receives all elements of
	* a stream
	* @param ordered whether an ordered traversal is requested
	* @return the {@code TerminalOp} instance
	*/
	public static TerminalOp<Double, Void> makeDouble(DoubleConsumer action,
	boolean ordered) {
	Objects.requireNonNull(action);
	return new ForEachOp.OfDouble(action, ordered);
	}

	/**
	* A {@code TerminalOp} that evaluates a stream pipeline and sends the
	* output to itself as a {@code TerminalSink}. Elements will be sent in
	* whatever thread they become available. If the traversal is unordered,
	* they will be sent independent of the stream's encounter order.
	*
	* <p>This terminal operation is stateless. For parallel evaluation, each
	* leaf instance of a {@code ForEachTask} will send elements to the same
	* {@code TerminalSink} reference that is an instance of this class.
	*
	* @param <T> the output type of the stream pipeline
	*/
	abstract static class ForEachOp<T>
	implements TerminalOp<T, Void>, TerminalSink<T, Void> {
	private final boolean ordered;

	protected ForEachOp(boolean ordered) {
	this.ordered = ordered;
	}

	// TerminalOp

	@Override
	public int getOpFlags() {
	return ordered ? 0 : StreamOpFlag.NOT_ORDERED;
	}

	@Override
	public <S> Void evaluateSequential(PipelineHelper<T> helper,
	Spliterator<S> spliterator) {
	return helper.wrapAndCopyInto(this, spliterator).get();
	}

	@Override
	public <S> Void evaluateParallel(PipelineHelper<T> helper,
	Spliterator<S> spliterator) {
	if (ordered)
	new ForEachOrderedTask<>(helper, spliterator, this).invoke();
	else
	new ForEachTask<>(helper, spliterator, helper.wrapSink(this)).invoke();
	return null;
	}

	// TerminalSink

	@Override
	public Void get() {
	return null;
	}

	// Implementations

	/** Implementation class for reference streams */
	static final class OfRef<T> extends ForEachOp<T> {
	final Consumer<? super T> consumer;

	OfRef(Consumer<? super T> consumer, boolean ordered) {
	super(ordered);
	this.consumer = consumer;
	}

	@Override
	public void accept(T t) {
	consumer.accept(t);
	}
	}

	/** Implementation class for {@code IntStream} */
	static final class OfInt extends ForEachOp<Integer>
	implements Sink.OfInt {
	final IntConsumer consumer;

	OfInt(IntConsumer consumer, boolean ordered) {
	super(ordered);
	this.consumer = consumer;
	}

	@Override
	public StreamShape inputShape() {
	return StreamShape.INT_VALUE;
	}

	@Override
	public void accept(int t) {
	consumer.accept(t);
	}
	}

	/** Implementation class for {@code LongStream} */
	static final class OfLong extends ForEachOp<Long>
	implements Sink.OfLong {
	final LongConsumer consumer;

	OfLong(LongConsumer consumer, boolean ordered) {
	super(ordered);
	this.consumer = consumer;
	}

	@Override
	public StreamShape inputShape() {
	return StreamShape.LONG_VALUE;
	}

	@Override
	public void accept(long t) {
	consumer.accept(t);
	}
	}

	/** Implementation class for {@code DoubleStream} */
	static final class OfDouble extends ForEachOp<Double>
	implements Sink.OfDouble {
	final DoubleConsumer consumer;

	OfDouble(DoubleConsumer consumer, boolean ordered) {
	super(ordered);
	this.consumer = consumer;
	}

	@Override
	public StreamShape inputShape() {
	return StreamShape.DOUBLE_VALUE;
	}

	@Override
	public void accept(double t) {
	consumer.accept(t);
	}
	}
	}

	/** A {@code ForkJoinTask} for performing a parallel for-each operation */
	@SuppressWarnings("serial")
	static final class ForEachTask<S, T> extends CountedCompleter<Void> {
	private Spliterator<S> spliterator;
	private final Sink<S> sink;
	private final PipelineHelper<T> helper;
	private long targetSize;

	ForEachTask(PipelineHelper<T> helper,
	Spliterator<S> spliterator,
	Sink<S> sink) {
	super(null);
	this.sink = sink;
	this.helper = helper;
	this.spliterator = spliterator;
	this.targetSize = 0L;
	}

	ForEachTask(ForEachTask<S, T> parent, Spliterator<S> spliterator) {
	super(parent);
	this.spliterator = spliterator;
	this.sink = parent.sink;
	this.targetSize = parent.targetSize;
	this.helper = parent.helper;
	}

	// Similar to AbstractTask but doesn't need to track child tasks
	public void compute() {
	Spliterator<S> rightSplit = spliterator, leftSplit;
	long sizeEstimate = rightSplit.estimateSize(), sizeThreshold;
	if ((sizeThreshold = targetSize) == 0L)
	targetSize = sizeThreshold = AbstractTask.suggestTargetSize(sizeEstimate);
	boolean isShortCircuit = StreamOpFlag.SHORT_CIRCUIT.isKnown(helper.getStreamAndOpFlags());
	boolean forkRight = false;
	Sink<S> taskSink = sink;
	ForEachTask<S, T> task = this;
	while (!isShortCircuit \|\| !taskSink.cancellationRequested()) {
	if (sizeEstimate <= sizeThreshold \|\|
	(leftSplit = rightSplit.trySplit()) == null) {
	task.helper.copyInto(taskSink, rightSplit);
	break;
	}
	ForEachTask<S, T> leftTask = new ForEachTask<>(task, leftSplit);
	task.addToPendingCount(1);
	ForEachTask<S, T> taskToFork;
	if (forkRight) {
	forkRight = false;
	rightSplit = leftSplit;
	taskToFork = task;
	task = leftTask;
	}
	else {
	forkRight = true;
	taskToFork = leftTask;
	}
	taskToFork.fork();
	sizeEstimate = rightSplit.estimateSize();
	}
	task.spliterator = null;
	task.propagateCompletion();
	}
	}

	/**
	* A {@code ForkJoinTask} for performing a parallel for-each operation
	* which visits the elements in encounter order
	*/
	@SuppressWarnings("serial")
	static final class ForEachOrderedTask<S, T> extends CountedCompleter<Void> {
	/*
	* Our goal is to ensure that the elements associated with a task are
	* processed according to an in-order traversal of the computation tree.
	* We use completion counts for representing these dependencies, so that
	* a task does not complete until all the tasks preceding it in this
	* order complete. We use the "completion map" to associate the next
	* task in this order for any left child. We increase the pending count
	* of any node on the right side of such a mapping by one to indicate
	* its dependency, and when a node on the left side of such a mapping
	* completes, it decrements the pending count of its corresponding right
	* side. As the computation tree is expanded by splitting, we must
	* atomically update the mappings to maintain the invariant that the
	* completion map maps left children to the next node in the in-order
	* traversal.
	*
	* Take, for example, the following computation tree of tasks:
	*
	* a
	* / \
	* b c
	* / \ / \
	* d e f g
	*
	* The complete map will contain (not necessarily all at the same time)
	* the following associations:
	*
	* d -> e
	* b -> f
	* f -> g
	*
	* Tasks e, f, g will have their pending counts increased by 1.
	*
	* The following relationships hold:
	*
	* - completion of d "happens-before" e;
	* - completion of d and e "happens-before b;
	* - completion of b "happens-before" f; and
	* - completion of f "happens-before" g
	*
	* Thus overall the "happens-before" relationship holds for the
	* reporting of elements, covered by tasks d, e, f and g, as specified
	* by the forEachOrdered operation.
	*/

	private final PipelineHelper<T> helper;
	private Spliterator<S> spliterator;
	private final long targetSize;
	private final ConcurrentHashMap<ForEachOrderedTask<S, T>, ForEachOrderedTask<S, T>> completionMap;
	private final Sink<T> action;
	private final ForEachOrderedTask<S, T> leftPredecessor;
	private Node<T> node;

	protected ForEachOrderedTask(PipelineHelper<T> helper,
	Spliterator<S> spliterator,
	Sink<T> action) {
	super(null);
	this.helper = helper;
	this.spliterator = spliterator;
	this.targetSize = AbstractTask.suggestTargetSize(spliterator.estimateSize());
	// Size map to avoid concurrent re-sizes
	this.completionMap = new ConcurrentHashMap<>(Math.max(16, AbstractTask.getLeafTarget() << 1));
	this.action = action;
	this.leftPredecessor = null;
	}

	ForEachOrderedTask(ForEachOrderedTask<S, T> parent,
	Spliterator<S> spliterator,
	ForEachOrderedTask<S, T> leftPredecessor) {
	super(parent);
	this.helper = parent.helper;
	this.spliterator = spliterator;
	this.targetSize = parent.targetSize;
	this.completionMap = parent.completionMap;
	this.action = parent.action;
	this.leftPredecessor = leftPredecessor;
	}

	@Override
	public final void compute() {
	doCompute(this);
	}

	private static <S, T> void doCompute(ForEachOrderedTask<S, T> task) {
	Spliterator<S> rightSplit = task.spliterator, leftSplit;
	long sizeThreshold = task.targetSize;
	boolean forkRight = false;
	while (rightSplit.estimateSize() > sizeThreshold &&
	(leftSplit = rightSplit.trySplit()) != null) {
	ForEachOrderedTask<S, T> leftChild =
	new ForEachOrderedTask<>(task, leftSplit, task.leftPredecessor);
	ForEachOrderedTask<S, T> rightChild =
	new ForEachOrderedTask<>(task, rightSplit, leftChild);

	// Fork the parent task
	// Completion of the left and right children "happens-before"
	// completion of the parent
	task.addToPendingCount(1);
	// Completion of the left child "happens-before" completion of
	// the right child
	rightChild.addToPendingCount(1);
	task.completionMap.put(leftChild, rightChild);

	// If task is not on the left spine
	if (task.leftPredecessor != null) {
	/*
	* Completion of left-predecessor, or left subtree,
	* "happens-before" completion of left-most leaf node of
	* right subtree.
	* The left child's pending count needs to be updated before
	* it is associated in the completion map, otherwise the
	* left child can complete prematurely and violate the
	* "happens-before" constraint.
	*/
	leftChild.addToPendingCount(1);
	// Update association of left-predecessor to left-most
	// leaf node of right subtree
	if (task.completionMap.replace(task.leftPredecessor, task, leftChild)) {
	// If replaced, adjust the pending count of the parent
	// to complete when its children complete
	task.addToPendingCount(-1);
	} else {
	// Left-predecessor has already completed, parent's
	// pending count is adjusted by left-predecessor;
	// left child is ready to complete
	leftChild.addToPendingCount(-1);
	}
	}

	ForEachOrderedTask<S, T> taskToFork;
	if (forkRight) {
	forkRight = false;
	rightSplit = leftSplit;
	task = leftChild;
	taskToFork = rightChild;
	}
	else {
	forkRight = true;
	task = rightChild;
	taskToFork = leftChild;
	}
	taskToFork.fork();
	}

	/*
	* Task's pending count is either 0 or 1. If 1 then the completion
	* map will contain a value that is task, and two calls to
	* tryComplete are required for completion, one below and one
	* triggered by the completion of task's left-predecessor in
	* onCompletion. Therefore there is no data race within the if
	* block.
	*/
	if (task.getPendingCount() > 0) {
	// Cannot complete just yet so buffer elements into a Node
	// for use when completion occurs
	@SuppressWarnings("unchecked")
	IntFunction<T[]> generator = size -> (T[]) new Object[size];
	Node.Builder<T> nb = task.helper.makeNodeBuilder(
	task.helper.exactOutputSizeIfKnown(rightSplit),
	generator);
	task.node = task.helper.wrapAndCopyInto(nb, rightSplit).build();
	task.spliterator = null;
	}
	task.tryComplete();
	}

	@Override
	public void onCompletion(CountedCompleter<?> caller) {
	if (node != null) {
	// Dump buffered elements from this leaf into the sink
	node.forEach(action);
	node = null;
	}
	else if (spliterator != null) {
	// Dump elements output from this leaf's pipeline into the sink
	helper.wrapAndCopyInto(action, spliterator);
	spliterator = null;
	}

	// The completion of this task and the dumping of elements
	// "happens-before" completion of the associated left-most leaf task
	// of right subtree (if any, which can be this task's right sibling)
	//
	ForEachOrderedTask<S, T> leftDescendant = completionMap.remove(this);
	if (leftDescendant != null)
	leftDescendant.tryComplete();
	}
	}
	}

Back to index...