| /* |
| * Copyright (c) 2012, 2013, 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.Spliterator; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| /** |
| * Abstract class for fork-join tasks used to implement short-circuiting |
| * stream ops, which can produce a result without processing all elements of the |
| * stream. |
| * |
| * @param <P_IN> type of input elements to the pipeline |
| * @param <P_OUT> type of output elements from the pipeline |
| * @param <R> type of intermediate result, may be different from operation |
| * result type |
| * @param <K> type of child and sibling tasks |
| * @since 1.8 |
| */ |
| @SuppressWarnings("serial") |
| abstract class AbstractShortCircuitTask<P_IN, P_OUT, R, |
| K extends AbstractShortCircuitTask<P_IN, P_OUT, R, K>> |
| extends AbstractTask<P_IN, P_OUT, R, K> { |
| /** |
| * The result for this computation; this is shared among all tasks and set |
| * exactly once |
| */ |
| protected final AtomicReference<R> sharedResult; |
| |
| /** |
| * Indicates whether this task has been canceled. Tasks may cancel other |
| * tasks in the computation under various conditions, such as in a |
| * find-first operation, a task that finds a value will cancel all tasks |
| * that are later in the encounter order. |
| */ |
| protected volatile boolean canceled; |
| |
| /** |
| * Constructor for root tasks. |
| * |
| * @param helper the {@code PipelineHelper} describing the stream pipeline |
| * up to this operation |
| * @param spliterator the {@code Spliterator} describing the source for this |
| * pipeline |
| */ |
| protected AbstractShortCircuitTask(PipelineHelper<P_OUT> helper, |
| Spliterator<P_IN> spliterator) { |
| super(helper, spliterator); |
| sharedResult = new AtomicReference<>(null); |
| } |
| |
| /** |
| * Constructor for non-root nodes. |
| * |
| * @param parent parent task in the computation tree |
| * @param spliterator the {@code Spliterator} for the portion of the |
| * computation tree described by this task |
| */ |
| protected AbstractShortCircuitTask(K parent, |
| Spliterator<P_IN> spliterator) { |
| super(parent, spliterator); |
| sharedResult = parent.sharedResult; |
| } |
| |
| /** |
| * Returns the value indicating the computation completed with no task |
| * finding a short-circuitable result. For example, for a "find" operation, |
| * this might be null or an empty {@code Optional}. |
| * |
| * @return the result to return when no task finds a result |
| */ |
| protected abstract R getEmptyResult(); |
| |
| /** |
| * Overrides AbstractTask version to include checks for early |
| * exits while splitting or computing. |
| */ |
| @Override |
| public void compute() { |
| Spliterator<P_IN> rs = spliterator, ls; |
| long sizeEstimate = rs.estimateSize(); |
| long sizeThreshold = getTargetSize(sizeEstimate); |
| boolean forkRight = false; |
| @SuppressWarnings("unchecked") K task = (K) this; |
| AtomicReference<R> sr = sharedResult; |
| R result; |
| while ((result = sr.get()) == null) { |
| if (task.taskCanceled()) { |
| result = task.getEmptyResult(); |
| break; |
| } |
| if (sizeEstimate <= sizeThreshold || (ls = rs.trySplit()) == null) { |
| result = task.doLeaf(); |
| break; |
| } |
| K leftChild, rightChild, taskToFork; |
| task.leftChild = leftChild = task.makeChild(ls); |
| task.rightChild = rightChild = task.makeChild(rs); |
| task.setPendingCount(1); |
| if (forkRight) { |
| forkRight = false; |
| rs = ls; |
| task = leftChild; |
| taskToFork = rightChild; |
| } |
| else { |
| forkRight = true; |
| task = rightChild; |
| taskToFork = leftChild; |
| } |
| taskToFork.fork(); |
| sizeEstimate = rs.estimateSize(); |
| } |
| task.setLocalResult(result); |
| task.tryComplete(); |
| } |
| |
| |
| /** |
| * Declares that a globally valid result has been found. If another task has |
| * not already found the answer, the result is installed in |
| * {@code sharedResult}. The {@code compute()} method will check |
| * {@code sharedResult} before proceeding with computation, so this causes |
| * the computation to terminate early. |
| * |
| * @param result the result found |
| */ |
| protected void shortCircuit(R result) { |
| if (result != null) |
| sharedResult.compareAndSet(null, result); |
| } |
| |
| /** |
| * Sets a local result for this task. If this task is the root, set the |
| * shared result instead (if not already set). |
| * |
| * @param localResult The result to set for this task |
| */ |
| @Override |
| protected void setLocalResult(R localResult) { |
| if (isRoot()) { |
| if (localResult != null) |
| sharedResult.compareAndSet(null, localResult); |
| } |
| else |
| super.setLocalResult(localResult); |
| } |
| |
| /** |
| * Retrieves the local result for this task |
| */ |
| @Override |
| public R getRawResult() { |
| return getLocalResult(); |
| } |
| |
| /** |
| * Retrieves the local result for this task. If this task is the root, |
| * retrieves the shared result instead. |
| */ |
| @Override |
| public R getLocalResult() { |
| if (isRoot()) { |
| R answer = sharedResult.get(); |
| return (answer == null) ? getEmptyResult() : answer; |
| } |
| else |
| return super.getLocalResult(); |
| } |
| |
| /** |
| * Mark this task as canceled |
| */ |
| protected void cancel() { |
| canceled = true; |
| } |
| |
| /** |
| * Queries whether this task is canceled. A task is considered canceled if |
| * it or any of its parents have been canceled. |
| * |
| * @return {@code true} if this task or any parent is canceled. |
| */ |
| protected boolean taskCanceled() { |
| boolean cancel = canceled; |
| if (!cancel) { |
| for (K parent = getParent(); !cancel && parent != null; parent = parent.getParent()) |
| cancel = parent.canceled; |
| } |
| |
| return cancel; |
| } |
| |
| /** |
| * Cancels all tasks which succeed this one in the encounter order. This |
| * includes canceling all the current task's right sibling, as well as the |
| * later right siblings of all its parents. |
| */ |
| protected void cancelLaterNodes() { |
| // Go up the tree, cancel right siblings of this node and all parents |
| for (@SuppressWarnings("unchecked") K parent = getParent(), node = (K) this; |
| parent != null; |
| node = parent, parent = parent.getParent()) { |
| // If node is a left child of parent, then has a right sibling |
| if (parent.leftChild == node) { |
| K rightSibling = parent.rightChild; |
| if (!rightSibling.canceled) |
| rightSibling.cancel(); |
| } |
| } |
| } |
| } |
