# HG changeset patch
# User chegar
# Date 1365524867 -3600
# Node ID fcce5e09e23b1e50866fd26f3d15b76f555345a4
# Parent b3ca7ed8e44f155bf7d48498c44ee92e5d4f3b72
8005696: Add CompletableFuture
Reviewed-by: chegar, martin
diff -r b3ca7ed8e44f -r fcce5e09e23b jdk/make/java/java/FILES_java.gmk
--- a/jdk/make/java/java/FILES_java.gmk Tue Apr 09 15:51:50 2013 +0100
+++ b/jdk/make/java/java/FILES_java.gmk Tue Apr 09 17:27:47 2013 +0100
@@ -316,6 +316,8 @@
java/util/concurrent/BrokenBarrierException.java \
java/util/concurrent/Callable.java \
java/util/concurrent/CancellationException.java \
+ java/util/concurrent/CompletableFuture.java \
+ java/util/concurrent/CompletionException.java \
java/util/concurrent/CompletionService.java \
java/util/concurrent/ConcurrentHashMap.java \
java/util/concurrent/ConcurrentLinkedDeque.java \
diff -r b3ca7ed8e44f -r fcce5e09e23b jdk/src/share/classes/java/util/concurrent/CompletableFuture.java
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/jdk/src/share/classes/java/util/concurrent/CompletableFuture.java Tue Apr 09 17:27:47 2013 +0100
@@ -0,0 +1,3305 @@
+/*
+ * 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.
+ */
+
+/*
+ * This file is available under and governed by the GNU General Public
+ * License version 2 only, as published by the Free Software Foundation.
+ * However, the following notice accompanied the original version of this
+ * file:
+ *
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/publicdomain/zero/1.0/
+ */
+
+package java.util.concurrent;
+import java.util.function.Supplier;
+import java.util.function.Consumer;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.function.BiFunction;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+
+/**
+ * A {@link Future} that may be explicitly completed (setting its
+ * value and status), and may include dependent functions and actions
+ * that trigger upon its completion.
+ *
+ *
When two or more threads attempt to
+ * {@link #complete complete},
+ * {@link #completeExceptionally completeExceptionally}, or
+ * {@link #cancel cancel}
+ * a CompletableFuture, only one of them succeeds.
+ *
+ *
Methods are available for adding dependents based on
+ * user-provided Functions, Consumers, or Runnables. The appropriate
+ * form to use depends on whether actions require arguments and/or
+ * produce results. Completion of a dependent action will trigger the
+ * completion of another CompletableFuture. Actions may also be
+ * triggered after either or both the current and another
+ * CompletableFuture complete. Multiple CompletableFutures may also
+ * be grouped as one using {@link #anyOf(CompletableFuture...)} and
+ * {@link #allOf(CompletableFuture...)}.
+ *
+ *
CompletableFutures themselves do not execute asynchronously.
+ * However, actions supplied for dependent completions of another
+ * CompletableFuture may do so, depending on whether they are provided
+ * via one of the async methods (that is, methods with names
+ * of the form xxxAsync). The async
+ * methods provide a way to commence asynchronous processing of an
+ * action using either a given {@link Executor} or by default the
+ * {@link ForkJoinPool#commonPool()}. To simplify monitoring,
+ * debugging, and tracking, all generated asynchronous tasks are
+ * instances of the marker interface {@link AsynchronousCompletionTask}.
+ *
+ *
Actions supplied for dependent completions of non-async
+ * methods may be performed by the thread that completes the current
+ * CompletableFuture, or by any other caller of these methods. There
+ * are no guarantees about the order of processing completions unless
+ * constrained by these methods.
+ *
+ *
Since (unlike {@link FutureTask}) this class has no direct
+ * control over the computation that causes it to be completed,
+ * cancellation is treated as just another form of exceptional completion.
+ * Method {@link #cancel cancel} has the same effect as
+ * {@code completeExceptionally(new CancellationException())}.
+ *
+ *
Upon exceptional completion (including cancellation), or when a
+ * completion entails an additional computation which terminates
+ * abruptly with an (unchecked) exception or error, then all of their
+ * dependent completions (and their dependents in turn) generally act
+ * as {@code completeExceptionally} with a {@link CompletionException}
+ * holding that exception as its cause. However, the {@link
+ * #exceptionally exceptionally} and {@link #handle handle}
+ * completions are able to handle exceptional completions of
+ * the CompletableFutures they depend on.
+ *
+ *
In case of exceptional completion with a CompletionException,
+ * methods {@link #get()} and {@link #get(long, TimeUnit)} throw an
+ * {@link ExecutionException} with the same cause as held in the
+ * corresponding CompletionException. However, in these cases,
+ * methods {@link #join()} and {@link #getNow} throw the
+ * CompletionException, which simplifies usage.
+ *
+ *
Arguments used to pass a completion result (that is, for parameters
+ * of type {@code T}) may be null, but passing a null value for any other
+ * parameter will result in a {@link NullPointerException} being thrown.
+ *
+ * @author Doug Lea
+ * @since 1.8
+ */
+public class CompletableFuture implements Future {
+
+ /*
+ * Overview:
+ *
+ * 1. Non-nullness of field result (set via CAS) indicates done.
+ * An AltResult is used to box null as a result, as well as to
+ * hold exceptions. Using a single field makes completion fast
+ * and simple to detect and trigger, at the expense of a lot of
+ * encoding and decoding that infiltrates many methods. One minor
+ * simplification relies on the (static) NIL (to box null results)
+ * being the only AltResult with a null exception field, so we
+ * don't usually need explicit comparisons with NIL. The CF
+ * exception propagation mechanics surrounding decoding rely on
+ * unchecked casts of decoded results really being unchecked,
+ * where user type errors are caught at point of use, as is
+ * currently the case in Java. These are highlighted by using
+ * SuppressWarnings-annotated temporaries.
+ *
+ * 2. Waiters are held in a Treiber stack similar to the one used
+ * in FutureTask, Phaser, and SynchronousQueue. See their
+ * internal documentation for algorithmic details.
+ *
+ * 3. Completions are also kept in a list/stack, and pulled off
+ * and run when completion is triggered. (We could even use the
+ * same stack as for waiters, but would give up the potential
+ * parallelism obtained because woken waiters help release/run
+ * others -- see method postComplete). Because post-processing
+ * may race with direct calls, class Completion opportunistically
+ * extends AtomicInteger so callers can claim the action via
+ * compareAndSet(0, 1). The Completion.run methods are all
+ * written a boringly similar uniform way (that sometimes includes
+ * unnecessary-looking checks, kept to maintain uniformity).
+ * There are enough dimensions upon which they differ that
+ * attempts to factor commonalities while maintaining efficiency
+ * require more lines of code than they would save.
+ *
+ * 4. The exported then/and/or methods do support a bit of
+ * factoring (see doThenApply etc). They must cope with the
+ * intrinsic races surrounding addition of a dependent action
+ * versus performing the action directly because the task is
+ * already complete. For example, a CF may not be complete upon
+ * entry, so a dependent completion is added, but by the time it
+ * is added, the target CF is complete, so must be directly
+ * executed. This is all done while avoiding unnecessary object
+ * construction in safe-bypass cases.
+ */
+
+ // preliminaries
+
+ static final class AltResult {
+ final Throwable ex; // null only for NIL
+ AltResult(Throwable ex) { this.ex = ex; }
+ }
+
+ static final AltResult NIL = new AltResult(null);
+
+ // Fields
+
+ volatile Object result; // Either the result or boxed AltResult
+ volatile WaitNode waiters; // Treiber stack of threads blocked on get()
+ volatile CompletionNode completions; // list (Treiber stack) of completions
+
+ // Basic utilities for triggering and processing completions
+
+ /**
+ * Removes and signals all waiting threads and runs all completions.
+ */
+ final void postComplete() {
+ WaitNode q; Thread t;
+ while ((q = waiters) != null) {
+ if (UNSAFE.compareAndSwapObject(this, WAITERS, q, q.next) &&
+ (t = q.thread) != null) {
+ q.thread = null;
+ LockSupport.unpark(t);
+ }
+ }
+
+ CompletionNode h; Completion c;
+ while ((h = completions) != null) {
+ if (UNSAFE.compareAndSwapObject(this, COMPLETIONS, h, h.next) &&
+ (c = h.completion) != null)
+ c.run();
+ }
+ }
+
+ /**
+ * Triggers completion with the encoding of the given arguments:
+ * if the exception is non-null, encodes it as a wrapped
+ * CompletionException unless it is one already. Otherwise uses
+ * the given result, boxed as NIL if null.
+ */
+ final void internalComplete(T v, Throwable ex) {
+ if (result == null)
+ UNSAFE.compareAndSwapObject
+ (this, RESULT, null,
+ (ex == null) ? (v == null) ? NIL : v :
+ new AltResult((ex instanceof CompletionException) ? ex :
+ new CompletionException(ex)));
+ postComplete(); // help out even if not triggered
+ }
+
+ /**
+ * If triggered, helps release and/or process completions.
+ */
+ final void helpPostComplete() {
+ if (result != null)
+ postComplete();
+ }
+
+ /* ------------- waiting for completions -------------- */
+
+ /** Number of processors, for spin control */
+ static final int NCPU = Runtime.getRuntime().availableProcessors();
+
+ /**
+ * Heuristic spin value for waitingGet() before blocking on
+ * multiprocessors
+ */
+ static final int SPINS = (NCPU > 1) ? 1 << 8 : 0;
+
+ /**
+ * Linked nodes to record waiting threads in a Treiber stack. See
+ * other classes such as Phaser and SynchronousQueue for more
+ * detailed explanation. This class implements ManagedBlocker to
+ * avoid starvation when blocking actions pile up in
+ * ForkJoinPools.
+ */
+ static final class WaitNode implements ForkJoinPool.ManagedBlocker {
+ long nanos; // wait time if timed
+ final long deadline; // non-zero if timed
+ volatile int interruptControl; // > 0: interruptible, < 0: interrupted
+ volatile Thread thread;
+ volatile WaitNode next;
+ WaitNode(boolean interruptible, long nanos, long deadline) {
+ this.thread = Thread.currentThread();
+ this.interruptControl = interruptible ? 1 : 0;
+ this.nanos = nanos;
+ this.deadline = deadline;
+ }
+ public boolean isReleasable() {
+ if (thread == null)
+ return true;
+ if (Thread.interrupted()) {
+ int i = interruptControl;
+ interruptControl = -1;
+ if (i > 0)
+ return true;
+ }
+ if (deadline != 0L &&
+ (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {
+ thread = null;
+ return true;
+ }
+ return false;
+ }
+ public boolean block() {
+ if (isReleasable())
+ return true;
+ else if (deadline == 0L)
+ LockSupport.park(this);
+ else if (nanos > 0L)
+ LockSupport.parkNanos(this, nanos);
+ return isReleasable();
+ }
+ }
+
+ /**
+ * Returns raw result after waiting, or null if interruptible and
+ * interrupted.
+ */
+ private Object waitingGet(boolean interruptible) {
+ WaitNode q = null;
+ boolean queued = false;
+ int spins = SPINS;
+ for (Object r;;) {
+ if ((r = result) != null) {
+ if (q != null) { // suppress unpark
+ q.thread = null;
+ if (q.interruptControl < 0) {
+ if (interruptible) {
+ removeWaiter(q);
+ return null;
+ }
+ Thread.currentThread().interrupt();
+ }
+ }
+ postComplete(); // help release others
+ return r;
+ }
+ else if (spins > 0) {
+ int rnd = ThreadLocalRandom.nextSecondarySeed();
+ if (rnd == 0)
+ rnd = ThreadLocalRandom.current().nextInt();
+ if (rnd >= 0)
+ --spins;
+ }
+ else if (q == null)
+ q = new WaitNode(interruptible, 0L, 0L);
+ else if (!queued)
+ queued = UNSAFE.compareAndSwapObject(this, WAITERS,
+ q.next = waiters, q);
+ else if (interruptible && q.interruptControl < 0) {
+ removeWaiter(q);
+ return null;
+ }
+ else if (q.thread != null && result == null) {
+ try {
+ ForkJoinPool.managedBlock(q);
+ } catch (InterruptedException ex) {
+ q.interruptControl = -1;
+ }
+ }
+ }
+ }
+
+ /**
+ * Awaits completion or aborts on interrupt or timeout.
+ *
+ * @param nanos time to wait
+ * @return raw result
+ */
+ private Object timedAwaitDone(long nanos)
+ throws InterruptedException, TimeoutException {
+ WaitNode q = null;
+ boolean queued = false;
+ for (Object r;;) {
+ if ((r = result) != null) {
+ if (q != null) {
+ q.thread = null;
+ if (q.interruptControl < 0) {
+ removeWaiter(q);
+ throw new InterruptedException();
+ }
+ }
+ postComplete();
+ return r;
+ }
+ else if (q == null) {
+ if (nanos <= 0L)
+ throw new TimeoutException();
+ long d = System.nanoTime() + nanos;
+ q = new WaitNode(true, nanos, d == 0L ? 1L : d); // avoid 0
+ }
+ else if (!queued)
+ queued = UNSAFE.compareAndSwapObject(this, WAITERS,
+ q.next = waiters, q);
+ else if (q.interruptControl < 0) {
+ removeWaiter(q);
+ throw new InterruptedException();
+ }
+ else if (q.nanos <= 0L) {
+ if (result == null) {
+ removeWaiter(q);
+ throw new TimeoutException();
+ }
+ }
+ else if (q.thread != null && result == null) {
+ try {
+ ForkJoinPool.managedBlock(q);
+ } catch (InterruptedException ex) {
+ q.interruptControl = -1;
+ }
+ }
+ }
+ }
+
+ /**
+ * Tries to unlink a timed-out or interrupted wait node to avoid
+ * accumulating garbage. Internal nodes are simply unspliced
+ * without CAS since it is harmless if they are traversed anyway
+ * by releasers. To avoid effects of unsplicing from already
+ * removed nodes, the list is retraversed in case of an apparent
+ * race. This is slow when there are a lot of nodes, but we don't
+ * expect lists to be long enough to outweigh higher-overhead
+ * schemes.
+ */
+ private void removeWaiter(WaitNode node) {
+ if (node != null) {
+ node.thread = null;
+ retry:
+ for (;;) { // restart on removeWaiter race
+ for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
+ s = q.next;
+ if (q.thread != null)
+ pred = q;
+ else if (pred != null) {
+ pred.next = s;
+ if (pred.thread == null) // check for race
+ continue retry;
+ }
+ else if (!UNSAFE.compareAndSwapObject(this, WAITERS, q, s))
+ continue retry;
+ }
+ break;
+ }
+ }
+ }
+
+ /* ------------- Async tasks -------------- */
+
+ /**
+ * A marker interface identifying asynchronous tasks produced by
+ * {@code async} methods. This may be useful for monitoring,
+ * debugging, and tracking asynchronous activities.
+ *
+ * @since 1.8
+ */
+ public static interface AsynchronousCompletionTask {
+ }
+
+ /** Base class can act as either FJ or plain Runnable */
+ abstract static class Async extends ForkJoinTask
+ implements Runnable, AsynchronousCompletionTask {
+ public final Void getRawResult() { return null; }
+ public final void setRawResult(Void v) { }
+ public final void run() { exec(); }
+ }
+
+ static final class AsyncRun extends Async {
+ final Runnable fn;
+ final CompletableFuture dst;
+ AsyncRun(Runnable fn, CompletableFuture dst) {
+ this.fn = fn; this.dst = dst;
+ }
+ public final boolean exec() {
+ CompletableFuture d; Throwable ex;
+ if ((d = this.dst) != null && d.result == null) {
+ try {
+ fn.run();
+ ex = null;
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ d.internalComplete(null, ex);
+ }
+ return true;
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class AsyncSupply extends Async {
+ final Supplier fn;
+ final CompletableFuture dst;
+ AsyncSupply(Supplier fn, CompletableFuture dst) {
+ this.fn = fn; this.dst = dst;
+ }
+ public final boolean exec() {
+ CompletableFuture d; U u; Throwable ex;
+ if ((d = this.dst) != null && d.result == null) {
+ try {
+ u = fn.get();
+ ex = null;
+ } catch (Throwable rex) {
+ ex = rex;
+ u = null;
+ }
+ d.internalComplete(u, ex);
+ }
+ return true;
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class AsyncApply extends Async {
+ final T arg;
+ final Function super T,? extends U> fn;
+ final CompletableFuture dst;
+ AsyncApply(T arg, Function super T,? extends U> fn,
+ CompletableFuture dst) {
+ this.arg = arg; this.fn = fn; this.dst = dst;
+ }
+ public final boolean exec() {
+ CompletableFuture d; U u; Throwable ex;
+ if ((d = this.dst) != null && d.result == null) {
+ try {
+ u = fn.apply(arg);
+ ex = null;
+ } catch (Throwable rex) {
+ ex = rex;
+ u = null;
+ }
+ d.internalComplete(u, ex);
+ }
+ return true;
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class AsyncCombine extends Async {
+ final T arg1;
+ final U arg2;
+ final BiFunction super T,? super U,? extends V> fn;
+ final CompletableFuture dst;
+ AsyncCombine(T arg1, U arg2,
+ BiFunction super T,? super U,? extends V> fn,
+ CompletableFuture dst) {
+ this.arg1 = arg1; this.arg2 = arg2; this.fn = fn; this.dst = dst;
+ }
+ public final boolean exec() {
+ CompletableFuture d; V v; Throwable ex;
+ if ((d = this.dst) != null && d.result == null) {
+ try {
+ v = fn.apply(arg1, arg2);
+ ex = null;
+ } catch (Throwable rex) {
+ ex = rex;
+ v = null;
+ }
+ d.internalComplete(v, ex);
+ }
+ return true;
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class AsyncAccept extends Async {
+ final T arg;
+ final Consumer super T> fn;
+ final CompletableFuture dst;
+ AsyncAccept(T arg, Consumer super T> fn,
+ CompletableFuture dst) {
+ this.arg = arg; this.fn = fn; this.dst = dst;
+ }
+ public final boolean exec() {
+ CompletableFuture d; Throwable ex;
+ if ((d = this.dst) != null && d.result == null) {
+ try {
+ fn.accept(arg);
+ ex = null;
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ d.internalComplete(null, ex);
+ }
+ return true;
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class AsyncAcceptBoth extends Async {
+ final T arg1;
+ final U arg2;
+ final BiConsumer super T,? super U> fn;
+ final CompletableFuture dst;
+ AsyncAcceptBoth(T arg1, U arg2,
+ BiConsumer super T,? super U> fn,
+ CompletableFuture dst) {
+ this.arg1 = arg1; this.arg2 = arg2; this.fn = fn; this.dst = dst;
+ }
+ public final boolean exec() {
+ CompletableFuture d; Throwable ex;
+ if ((d = this.dst) != null && d.result == null) {
+ try {
+ fn.accept(arg1, arg2);
+ ex = null;
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ d.internalComplete(null, ex);
+ }
+ return true;
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class AsyncCompose extends Async {
+ final T arg;
+ final Function super T, CompletableFuture> fn;
+ final CompletableFuture dst;
+ AsyncCompose(T arg,
+ Function super T, CompletableFuture> fn,
+ CompletableFuture dst) {
+ this.arg = arg; this.fn = fn; this.dst = dst;
+ }
+ public final boolean exec() {
+ CompletableFuture d, fr; U u; Throwable ex;
+ if ((d = this.dst) != null && d.result == null) {
+ try {
+ fr = fn.apply(arg);
+ ex = (fr == null) ? new NullPointerException() : null;
+ } catch (Throwable rex) {
+ ex = rex;
+ fr = null;
+ }
+ if (ex != null)
+ u = null;
+ else {
+ Object r = fr.result;
+ if (r == null)
+ r = fr.waitingGet(false);
+ if (r instanceof AltResult) {
+ ex = ((AltResult)r).ex;
+ u = null;
+ }
+ else {
+ @SuppressWarnings("unchecked") U ur = (U) r;
+ u = ur;
+ }
+ }
+ d.internalComplete(u, ex);
+ }
+ return true;
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ /* ------------- Completions -------------- */
+
+ /**
+ * Simple linked list nodes to record completions, used in
+ * basically the same way as WaitNodes. (We separate nodes from
+ * the Completions themselves mainly because for the And and Or
+ * methods, the same Completion object resides in two lists.)
+ */
+ static final class CompletionNode {
+ final Completion completion;
+ volatile CompletionNode next;
+ CompletionNode(Completion completion) { this.completion = completion; }
+ }
+
+ // Opportunistically subclass AtomicInteger to use compareAndSet to claim.
+ abstract static class Completion extends AtomicInteger implements Runnable {
+ }
+
+ static final class ThenApply extends Completion {
+ final CompletableFuture extends T> src;
+ final Function super T,? extends U> fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ ThenApply(CompletableFuture extends T> src,
+ Function super T,? extends U> fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture extends T> a;
+ final Function super T,? extends U> fn;
+ final CompletableFuture dst;
+ Object r; T t; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (a = this.src) != null &&
+ (r = a.result) != null &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult) {
+ ex = ((AltResult)r).ex;
+ t = null;
+ }
+ else {
+ ex = null;
+ @SuppressWarnings("unchecked") T tr = (T) r;
+ t = tr;
+ }
+ Executor e = executor;
+ U u = null;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncApply(t, fn, dst));
+ else
+ u = fn.apply(t);
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(u, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class ThenAccept extends Completion {
+ final CompletableFuture extends T> src;
+ final Consumer super T> fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ ThenAccept(CompletableFuture extends T> src,
+ Consumer super T> fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture extends T> a;
+ final Consumer super T> fn;
+ final CompletableFuture dst;
+ Object r; T t; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (a = this.src) != null &&
+ (r = a.result) != null &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult) {
+ ex = ((AltResult)r).ex;
+ t = null;
+ }
+ else {
+ ex = null;
+ @SuppressWarnings("unchecked") T tr = (T) r;
+ t = tr;
+ }
+ Executor e = executor;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncAccept(t, fn, dst));
+ else
+ fn.accept(t);
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(null, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class ThenRun extends Completion {
+ final CompletableFuture> src;
+ final Runnable fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ ThenRun(CompletableFuture> src,
+ Runnable fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture> a;
+ final Runnable fn;
+ final CompletableFuture dst;
+ Object r; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (a = this.src) != null &&
+ (r = a.result) != null &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult)
+ ex = ((AltResult)r).ex;
+ else
+ ex = null;
+ Executor e = executor;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncRun(fn, dst));
+ else
+ fn.run();
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(null, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class ThenCombine extends Completion {
+ final CompletableFuture extends T> src;
+ final CompletableFuture extends U> snd;
+ final BiFunction super T,? super U,? extends V> fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ ThenCombine(CompletableFuture extends T> src,
+ CompletableFuture extends U> snd,
+ BiFunction super T,? super U,? extends V> fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.snd = snd;
+ this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture extends T> a;
+ final CompletableFuture extends U> b;
+ final BiFunction super T,? super U,? extends V> fn;
+ final CompletableFuture dst;
+ Object r, s; T t; U u; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (a = this.src) != null &&
+ (r = a.result) != null &&
+ (b = this.snd) != null &&
+ (s = b.result) != null &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult) {
+ ex = ((AltResult)r).ex;
+ t = null;
+ }
+ else {
+ ex = null;
+ @SuppressWarnings("unchecked") T tr = (T) r;
+ t = tr;
+ }
+ if (ex != null)
+ u = null;
+ else if (s instanceof AltResult) {
+ ex = ((AltResult)s).ex;
+ u = null;
+ }
+ else {
+ @SuppressWarnings("unchecked") U us = (U) s;
+ u = us;
+ }
+ Executor e = executor;
+ V v = null;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncCombine(t, u, fn, dst));
+ else
+ v = fn.apply(t, u);
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(v, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class ThenAcceptBoth extends Completion {
+ final CompletableFuture extends T> src;
+ final CompletableFuture extends U> snd;
+ final BiConsumer super T,? super U> fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ ThenAcceptBoth(CompletableFuture extends T> src,
+ CompletableFuture extends U> snd,
+ BiConsumer super T,? super U> fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.snd = snd;
+ this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture extends T> a;
+ final CompletableFuture extends U> b;
+ final BiConsumer super T,? super U> fn;
+ final CompletableFuture dst;
+ Object r, s; T t; U u; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (a = this.src) != null &&
+ (r = a.result) != null &&
+ (b = this.snd) != null &&
+ (s = b.result) != null &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult) {
+ ex = ((AltResult)r).ex;
+ t = null;
+ }
+ else {
+ ex = null;
+ @SuppressWarnings("unchecked") T tr = (T) r;
+ t = tr;
+ }
+ if (ex != null)
+ u = null;
+ else if (s instanceof AltResult) {
+ ex = ((AltResult)s).ex;
+ u = null;
+ }
+ else {
+ @SuppressWarnings("unchecked") U us = (U) s;
+ u = us;
+ }
+ Executor e = executor;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncAcceptBoth(t, u, fn, dst));
+ else
+ fn.accept(t, u);
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(null, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class RunAfterBoth extends Completion {
+ final CompletableFuture> src;
+ final CompletableFuture> snd;
+ final Runnable fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ RunAfterBoth(CompletableFuture> src,
+ CompletableFuture> snd,
+ Runnable fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.snd = snd;
+ this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture> a;
+ final CompletableFuture> b;
+ final Runnable fn;
+ final CompletableFuture dst;
+ Object r, s; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (a = this.src) != null &&
+ (r = a.result) != null &&
+ (b = this.snd) != null &&
+ (s = b.result) != null &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult)
+ ex = ((AltResult)r).ex;
+ else
+ ex = null;
+ if (ex == null && (s instanceof AltResult))
+ ex = ((AltResult)s).ex;
+ Executor e = executor;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncRun(fn, dst));
+ else
+ fn.run();
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(null, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class AndCompletion extends Completion {
+ final CompletableFuture> src;
+ final CompletableFuture> snd;
+ final CompletableFuture dst;
+ AndCompletion(CompletableFuture> src,
+ CompletableFuture> snd,
+ CompletableFuture dst) {
+ this.src = src; this.snd = snd; this.dst = dst;
+ }
+ public final void run() {
+ final CompletableFuture> a;
+ final CompletableFuture> b;
+ final CompletableFuture dst;
+ Object r, s; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (a = this.src) != null &&
+ (r = a.result) != null &&
+ (b = this.snd) != null &&
+ (s = b.result) != null &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult)
+ ex = ((AltResult)r).ex;
+ else
+ ex = null;
+ if (ex == null && (s instanceof AltResult))
+ ex = ((AltResult)s).ex;
+ dst.internalComplete(null, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class ApplyToEither extends Completion {
+ final CompletableFuture extends T> src;
+ final CompletableFuture extends T> snd;
+ final Function super T,? extends U> fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ ApplyToEither(CompletableFuture extends T> src,
+ CompletableFuture extends T> snd,
+ Function super T,? extends U> fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.snd = snd;
+ this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture extends T> a;
+ final CompletableFuture extends T> b;
+ final Function super T,? extends U> fn;
+ final CompletableFuture dst;
+ Object r; T t; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (((a = this.src) != null && (r = a.result) != null) ||
+ ((b = this.snd) != null && (r = b.result) != null)) &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult) {
+ ex = ((AltResult)r).ex;
+ t = null;
+ }
+ else {
+ ex = null;
+ @SuppressWarnings("unchecked") T tr = (T) r;
+ t = tr;
+ }
+ Executor e = executor;
+ U u = null;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncApply(t, fn, dst));
+ else
+ u = fn.apply(t);
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(u, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class AcceptEither extends Completion {
+ final CompletableFuture extends T> src;
+ final CompletableFuture extends T> snd;
+ final Consumer super T> fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ AcceptEither(CompletableFuture extends T> src,
+ CompletableFuture extends T> snd,
+ Consumer super T> fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.snd = snd;
+ this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture extends T> a;
+ final CompletableFuture extends T> b;
+ final Consumer super T> fn;
+ final CompletableFuture dst;
+ Object r; T t; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (((a = this.src) != null && (r = a.result) != null) ||
+ ((b = this.snd) != null && (r = b.result) != null)) &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult) {
+ ex = ((AltResult)r).ex;
+ t = null;
+ }
+ else {
+ ex = null;
+ @SuppressWarnings("unchecked") T tr = (T) r;
+ t = tr;
+ }
+ Executor e = executor;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncAccept(t, fn, dst));
+ else
+ fn.accept(t);
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(null, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class RunAfterEither extends Completion {
+ final CompletableFuture> src;
+ final CompletableFuture> snd;
+ final Runnable fn;
+ final CompletableFuture dst;
+ final Executor executor;
+ RunAfterEither(CompletableFuture> src,
+ CompletableFuture> snd,
+ Runnable fn,
+ CompletableFuture dst,
+ Executor executor) {
+ this.src = src; this.snd = snd;
+ this.fn = fn; this.dst = dst;
+ this.executor = executor;
+ }
+ public final void run() {
+ final CompletableFuture> a;
+ final CompletableFuture> b;
+ final Runnable fn;
+ final CompletableFuture dst;
+ Object r; Throwable ex;
+ if ((dst = this.dst) != null &&
+ (fn = this.fn) != null &&
+ (((a = this.src) != null && (r = a.result) != null) ||
+ ((b = this.snd) != null && (r = b.result) != null)) &&
+ compareAndSet(0, 1)) {
+ if (r instanceof AltResult)
+ ex = ((AltResult)r).ex;
+ else
+ ex = null;
+ Executor e = executor;
+ if (ex == null) {
+ try {
+ if (e != null)
+ e.execute(new AsyncRun(fn, dst));
+ else
+ fn.run();
+ } catch (Throwable rex) {
+ ex = rex;
+ }
+ }
+ if (e == null || ex != null)
+ dst.internalComplete(null, ex);
+ }
+ }
+ private static final long serialVersionUID = 5232453952276885070L;
+ }
+
+ static final class OrCompletion extends Completion {
+ final CompletableFuture> src;
+ final CompletableFuture> snd;
+ final CompletableFuture