jdk-sandbox: comparison jdk/src/share/classes/java/util/stream/ForEachOps.java

equal deleted inserted replaced

-:81efc55fac99
+:bb9c71b84919
 * 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 Object lock;
 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());
-this.completionMap = new ConcurrentHashMap<>();
+// Size map to avoid concurrent re-sizes
+this.completionMap = new ConcurrentHashMap<>(Math.max(16, AbstractTask.LEAF_TARGET << 1));
 this.action = action;
-this.lock = new Object();
 this.leftPredecessor = null;
 }
 ForEachOrderedTask(ForEachOrderedTask<S, T> parent,
 Spliterator<S> spliterator,
 this.helper = parent.helper;
 this.spliterator = spliterator;
 this.targetSize = parent.targetSize;
 this.completionMap = parent.completionMap;
 this.action = parent.action;
-this.lock = parent.lock;
 this.leftPredecessor = leftPredecessor;
 }
 @Override
 public final void compute() {
 (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);
-task.addToPendingCount(1); // forking
-rightChild.addToPendingCount(1); // right pending on left child
+// If task is not on the left spine
 if (task.leftPredecessor != null) {
-leftChild.addToPendingCount(1); // left pending on previous subtree, except left spine
+/*
-if (task.completionMap.replace(task.leftPredecessor, task, leftChild))
+* Completion of left-predecessor, or left subtree,
-task.addToPendingCount(-1); // transfer my "right child" count to my left child
+* "happens-before" completion of left-most leaf node of
-else
+* right subtree.
-leftChild.addToPendingCount(-1); // left child is ready to go when ready
+* 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;
 task = rightChild;
 taskToFork = leftChild;
 }
 taskToFork.fork();
 }
-if (task.getPendingCount() == 0) {
-task.helper.wrapAndCopyInto(task.action, rightSplit);
+/*
-}
+* Task's pending count is either 0 or 1.  If 1 then the completion
-else {
+* 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
 Node.Builder<T> nb = task.helper.makeNodeBuilder(
 task.helper.exactOutputSizeIfKnown(rightSplit),
 size -> (T[]) new Object[size]);
 task.node = task.helper.wrapAndCopyInto(nb, rightSplit).build();
+task.spliterator = null;
 }
 task.tryComplete();
 }
 @Override
 public void onCompletion(CountedCompleter<?> caller) {
-spliterator = null;
 if (node != null) {
-// Dump any data from this leaf into the sink
+// Dump buffered elements from this leaf into the sink
-synchronized (lock) {
+node.forEach(action);
-node.forEach(action);
-}
 node = null;
 }
-ForEachOrderedTask<S, T> victim = completionMap.remove(this);
+else if (spliterator != null) {
-if (victim != null)
+// Dump elements output from this leaf's pipeline into the sink
-victim.tryComplete();
+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();
 }
 }
 }

changeset 22289	bb9c71b84919
parent 19218	8e7212b90b81
child 22297	1c62c67d9dd2