1 /*

     2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     3  *

     4  * This code is free software; you can redistribute it and/or modify it

     5  * under the terms of the GNU General Public License version 2 only, as

     6  * published by the Free Software Foundation.  Oracle designates this

     7  * particular file as subject to the "Classpath" exception as provided

     8  * by Oracle in the LICENSE file that accompanied this code.

     9  *

    10  * This code is distributed in the hope that it will be useful, but WITHOUT

    11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    13  * version 2 for more details (a copy is included in the LICENSE file that

    14  * accompanied this code).

    15  *

    16  * You should have received a copy of the GNU General Public License version

    17  * 2 along with this work; if not, write to the Free Software Foundation,

    18  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    19  *

    20  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    21  * or visit www.oracle.com if you need additional information or have any

    22  * questions.

    23  */

    24 

    25 /*

    26  * This file is available under and governed by the GNU General Public

    27  * License version 2 only, as published by the Free Software Foundation.

    28  * However, the following notice accompanied the original version of this

    29  * file:

    30  *

    31  * Written by Doug Lea with assistance from members of JCP JSR-166

    32  * Expert Group and released to the public domain, as explained at

    33  * http://creativecommons.org/publicdomain/zero/1.0/

    34  */

    35 

    36 package java.util.concurrent;

    37 import java.util.function.Supplier;

    38 import java.util.function.Consumer;

    39 import java.util.function.BiConsumer;

    40 import java.util.function.Function;

    41 import java.util.function.BiFunction;

    42 import java.util.concurrent.Future;

    43 import java.util.concurrent.TimeUnit;

    44 import java.util.concurrent.ForkJoinPool;

    45 import java.util.concurrent.ForkJoinTask;

    46 import java.util.concurrent.Executor;

    47 import java.util.concurrent.ThreadLocalRandom;

    48 import java.util.concurrent.ExecutionException;

    49 import java.util.concurrent.TimeoutException;

    50 import java.util.concurrent.CancellationException;

    51 import java.util.concurrent.atomic.AtomicInteger;

    52 import java.util.concurrent.locks.LockSupport;

    53 

    54 /**

    55  * A {@link Future} that may be explicitly completed (setting its

    56  * value and status), and may include dependent functions and actions

    57  * that trigger upon its completion.

    58  *

    59  * <p>When two or more threads attempt to

    60  * {@link #complete complete},

    61  * {@link #completeExceptionally completeExceptionally}, or

    62  * {@link #cancel cancel}

    63  * a CompletableFuture, only one of them succeeds.

    64  *

    65  * <p>Methods are available for adding dependents based on

    66  * user-provided Functions, Consumers, or Runnables. The appropriate

    67  * form to use depends on whether actions require arguments and/or

    68  * produce results.  Completion of a dependent action will trigger the

    69  * completion of another CompletableFuture.  Actions may also be

    70  * triggered after either or both the current and another

    71  * CompletableFuture complete.  Multiple CompletableFutures may also

    72  * be grouped as one using {@link #anyOf(CompletableFuture...)} and

    73  * {@link #allOf(CompletableFuture...)}.

    74  *

    75  * <p>CompletableFutures themselves do not execute asynchronously.

    76  * However, actions supplied for dependent completions of another

    77  * CompletableFuture may do so, depending on whether they are provided

    78  * via one of the <em>async</em> methods (that is, methods with names

    79  * of the form <tt><var>xxx</var>Async</tt>).  The <em>async</em>

    80  * methods provide a way to commence asynchronous processing of an

    81  * action using either a given {@link Executor} or by default the

    82  * {@link ForkJoinPool#commonPool()}. To simplify monitoring,

    83  * debugging, and tracking, all generated asynchronous tasks are

    84  * instances of the marker interface {@link AsynchronousCompletionTask}.

    85  *

    86  * <p>Actions supplied for dependent completions of <em>non-async</em>

    87  * methods may be performed by the thread that completes the current

    88  * CompletableFuture, or by any other caller of these methods.  There

    89  * are no guarantees about the order of processing completions unless

    90  * constrained by these methods.

    91  *

    92  * <p>Since (unlike {@link FutureTask}) this class has no direct

    93  * control over the computation that causes it to be completed,

    94  * cancellation is treated as just another form of exceptional completion.

    95  * Method {@link #cancel cancel} has the same effect as

    96  * {@code completeExceptionally(new CancellationException())}.

    97  *

    98  * <p>Upon exceptional completion (including cancellation), or when a

    99  * completion entails an additional computation which terminates

   100  * abruptly with an (unchecked) exception or error, then all of their

   101  * dependent completions (and their dependents in turn) generally act

   102  * as {@code completeExceptionally} with a {@link CompletionException}

   103  * holding that exception as its cause.  However, the {@link

   104  * #exceptionally exceptionally} and {@link #handle handle}

   105  * completions <em>are</em> able to handle exceptional completions of

   106  * the CompletableFutures they depend on.

   107  *

   108  * <p>In case of exceptional completion with a CompletionException,

   109  * methods {@link #get()} and {@link #get(long, TimeUnit)} throw an

   110  * {@link ExecutionException} with the same cause as held in the

   111  * corresponding CompletionException.  However, in these cases,

   112  * methods {@link #join()} and {@link #getNow} throw the

   113  * CompletionException, which simplifies usage.

   114  *

   115  * <p>Arguments used to pass a completion result (that is, for parameters

   116  * of type {@code T}) may be null, but passing a null value for any other

   117  * parameter will result in a {@link NullPointerException} being thrown.

   118  *

   119  * @author Doug Lea

   120  * @since 1.8

   121  */

   122 public class CompletableFuture<T> implements Future<T> {

   123 

   124     /*

   125      * Overview:

   126      *

   127      * 1. Non-nullness of field result (set via CAS) indicates done.

   128      * An AltResult is used to box null as a result, as well as to

   129      * hold exceptions.  Using a single field makes completion fast

   130      * and simple to detect and trigger, at the expense of a lot of

   131      * encoding and decoding that infiltrates many methods. One minor

   132      * simplification relies on the (static) NIL (to box null results)

   133      * being the only AltResult with a null exception field, so we

   134      * don't usually need explicit comparisons with NIL. The CF

   135      * exception propagation mechanics surrounding decoding rely on

   136      * unchecked casts of decoded results really being unchecked,

   137      * where user type errors are caught at point of use, as is

   138      * currently the case in Java. These are highlighted by using

   139      * SuppressWarnings-annotated temporaries.

   140      *

   141      * 2. Waiters are held in a Treiber stack similar to the one used

   142      * in FutureTask, Phaser, and SynchronousQueue. See their

   143      * internal documentation for algorithmic details.

   144      *

   145      * 3. Completions are also kept in a list/stack, and pulled off

   146      * and run when completion is triggered. (We could even use the

   147      * same stack as for waiters, but would give up the potential

   148      * parallelism obtained because woken waiters help release/run

   149      * others -- see method postComplete).  Because post-processing

   150      * may race with direct calls, class Completion opportunistically

   151      * extends AtomicInteger so callers can claim the action via

   152      * compareAndSet(0, 1).  The Completion.run methods are all

   153      * written a boringly similar uniform way (that sometimes includes

   154      * unnecessary-looking checks, kept to maintain uniformity).

   155      * There are enough dimensions upon which they differ that

   156      * attempts to factor commonalities while maintaining efficiency

   157      * require more lines of code than they would save.

   158      *

   159      * 4. The exported then/and/or methods do support a bit of

   160      * factoring (see doThenApply etc). They must cope with the

   161      * intrinsic races surrounding addition of a dependent action

   162      * versus performing the action directly because the task is

   163      * already complete.  For example, a CF may not be complete upon

   164      * entry, so a dependent completion is added, but by the time it

   165      * is added, the target CF is complete, so must be directly

   166      * executed. This is all done while avoiding unnecessary object

   167      * construction in safe-bypass cases.

   168      */

   169 

   170     // preliminaries

   171 

   172     static final class AltResult {

   173         final Throwable ex; // null only for NIL

   174         AltResult(Throwable ex) { this.ex = ex; }

   175     }

   176 

   177     static final AltResult NIL = new AltResult(null);

   178 

   179     // Fields

   180 

   181     volatile Object result;    // Either the result or boxed AltResult

   182     volatile WaitNode waiters; // Treiber stack of threads blocked on get()

   183     volatile CompletionNode completions; // list (Treiber stack) of completions

   184 

   185     // Basic utilities for triggering and processing completions

   186 

   187     /**

   188      * Removes and signals all waiting threads and runs all completions.

   189      */

   190     final void postComplete() {

   191         WaitNode q; Thread t;

   192         while ((q = waiters) != null) {

   193             if (UNSAFE.compareAndSwapObject(this, WAITERS, q, q.next) &&

   194                 (t = q.thread) != null) {

   195                 q.thread = null;

   196                 LockSupport.unpark(t);

   197             }

   198         }

   199 

   200         CompletionNode h; Completion c;

   201         while ((h = completions) != null) {

   202             if (UNSAFE.compareAndSwapObject(this, COMPLETIONS, h, h.next) &&

   203                 (c = h.completion) != null)

   204                 c.run();

   205         }

   206     }

   207 

   208     /**

   209      * Triggers completion with the encoding of the given arguments:

   210      * if the exception is non-null, encodes it as a wrapped

   211      * CompletionException unless it is one already.  Otherwise uses

   212      * the given result, boxed as NIL if null.

   213      */

   214     final void internalComplete(T v, Throwable ex) {

   215         if (result == null)

   216             UNSAFE.compareAndSwapObject

   217                 (this, RESULT, null,

   218                  (ex == null) ? (v == null) ? NIL : v :

   219                  new AltResult((ex instanceof CompletionException) ? ex :

   220                                new CompletionException(ex)));

   221         postComplete(); // help out even if not triggered

   222     }

   223 

   224     /**

   225      * If triggered, helps release and/or process completions.

   226      */

   227     final void helpPostComplete() {

   228         if (result != null)

   229             postComplete();

   230     }

   231 

   232     /* ------------- waiting for completions -------------- */

   233 

   234     /** Number of processors, for spin control */

   235     static final int NCPU = Runtime.getRuntime().availableProcessors();

   236 

   237     /**

   238      * Heuristic spin value for waitingGet() before blocking on

   239      * multiprocessors

   240      */

   241     static final int SPINS = (NCPU > 1) ? 1 << 8 : 0;

   242 

   243     /**

   244      * Linked nodes to record waiting threads in a Treiber stack.  See

   245      * other classes such as Phaser and SynchronousQueue for more

   246      * detailed explanation. This class implements ManagedBlocker to

   247      * avoid starvation when blocking actions pile up in

   248      * ForkJoinPools.

   249      */

   250     static final class WaitNode implements ForkJoinPool.ManagedBlocker {

   251         long nanos;          // wait time if timed

   252         final long deadline; // non-zero if timed

   253         volatile int interruptControl; // > 0: interruptible, < 0: interrupted

   254         volatile Thread thread;

   255         volatile WaitNode next;

   256         WaitNode(boolean interruptible, long nanos, long deadline) {

   257             this.thread = Thread.currentThread();

   258             this.interruptControl = interruptible ? 1 : 0;

   259             this.nanos = nanos;

   260             this.deadline = deadline;

   261         }

   262         public boolean isReleasable() {

   263             if (thread == null)

   264                 return true;

   265             if (Thread.interrupted()) {

   266                 int i = interruptControl;

   267                 interruptControl = -1;

   268                 if (i > 0)

   269                     return true;

   270             }

   271             if (deadline != 0L &&

   272                 (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {

   273                 thread = null;

   274                 return true;

   275             }

   276             return false;

   277         }

   278         public boolean block() {

   279             if (isReleasable())

   280                 return true;

   281             else if (deadline == 0L)

   282                 LockSupport.park(this);

   283             else if (nanos > 0L)

   284                 LockSupport.parkNanos(this, nanos);

   285             return isReleasable();

   286         }

   287     }

   288 

   289     /**

   290      * Returns raw result after waiting, or null if interruptible and

   291      * interrupted.

   292      */

   293     private Object waitingGet(boolean interruptible) {

   294         WaitNode q = null;

   295         boolean queued = false;

   296         int spins = SPINS;

   297         for (Object r;;) {

   298             if ((r = result) != null) {

   299                 if (q != null) { // suppress unpark

   300                     q.thread = null;

   301                     if (q.interruptControl < 0) {

   302                         if (interruptible) {

   303                             removeWaiter(q);

   304                             return null;

   305                         }

   306                         Thread.currentThread().interrupt();

   307                     }

   308                 }

   309                 postComplete(); // help release others

   310                 return r;

   311             }

   312             else if (spins > 0) {

   313                 int rnd = ThreadLocalRandom.nextSecondarySeed();

   314                 if (rnd == 0)

   315                     rnd = ThreadLocalRandom.current().nextInt();

   316                 if (rnd >= 0)

   317                     --spins;

   318             }

   319             else if (q == null)

   320                 q = new WaitNode(interruptible, 0L, 0L);

   321             else if (!queued)

   322                 queued = UNSAFE.compareAndSwapObject(this, WAITERS,

   323                                                      q.next = waiters, q);

   324             else if (interruptible && q.interruptControl < 0) {

   325                 removeWaiter(q);

   326                 return null;

   327             }

   328             else if (q.thread != null && result == null) {

   329                 try {

   330                     ForkJoinPool.managedBlock(q);

   331                 } catch (InterruptedException ex) {

   332                     q.interruptControl = -1;

   333                 }

   334             }

   335         }

   336     }

   337 

   338     /**

   339      * Awaits completion or aborts on interrupt or timeout.

   340      *

   341      * @param nanos time to wait

   342      * @return raw result

   343      */

   344     private Object timedAwaitDone(long nanos)

   345         throws InterruptedException, TimeoutException {

   346         WaitNode q = null;

   347         boolean queued = false;

   348         for (Object r;;) {

   349             if ((r = result) != null) {

   350                 if (q != null) {

   351                     q.thread = null;

   352                     if (q.interruptControl < 0) {

   353                         removeWaiter(q);

   354                         throw new InterruptedException();

   355                     }

   356                 }

   357                 postComplete();

   358                 return r;

   359             }

   360             else if (q == null) {

   361                 if (nanos <= 0L)

   362                     throw new TimeoutException();

   363                 long d = System.nanoTime() + nanos;

   364                 q = new WaitNode(true, nanos, d == 0L ? 1L : d); // avoid 0

   365             }

   366             else if (!queued)

   367                 queued = UNSAFE.compareAndSwapObject(this, WAITERS,

   368                                                      q.next = waiters, q);

   369             else if (q.interruptControl < 0) {

   370                 removeWaiter(q);

   371                 throw new InterruptedException();

   372             }

   373             else if (q.nanos <= 0L) {

   374                 if (result == null) {

   375                     removeWaiter(q);

   376                     throw new TimeoutException();

   377                 }

   378             }

   379             else if (q.thread != null && result == null) {

   380                 try {

   381                     ForkJoinPool.managedBlock(q);

   382                 } catch (InterruptedException ex) {

   383                     q.interruptControl = -1;

   384                 }

   385             }

   386         }

   387     }

   388 

   389     /**

   390      * Tries to unlink a timed-out or interrupted wait node to avoid

   391      * accumulating garbage.  Internal nodes are simply unspliced

   392      * without CAS since it is harmless if they are traversed anyway

   393      * by releasers.  To avoid effects of unsplicing from already

   394      * removed nodes, the list is retraversed in case of an apparent

   395      * race.  This is slow when there are a lot of nodes, but we don't

   396      * expect lists to be long enough to outweigh higher-overhead

   397      * schemes.

   398      */

   399     private void removeWaiter(WaitNode node) {

   400         if (node != null) {

   401             node.thread = null;

   402             retry:

   403             for (;;) {          // restart on removeWaiter race

   404                 for (WaitNode pred = null, q = waiters, s; q != null; q = s) {

   405                     s = q.next;

   406                     if (q.thread != null)

   407                         pred = q;

   408                     else if (pred != null) {

   409                         pred.next = s;

   410                         if (pred.thread == null) // check for race

   411                             continue retry;

   412                     }

   413                     else if (!UNSAFE.compareAndSwapObject(this, WAITERS, q, s))

   414                         continue retry;

   415                 }

   416                 break;

   417             }

   418         }

   419     }

   420 

   421     /* ------------- Async tasks -------------- */

   422 

   423     /**

   424      * A marker interface identifying asynchronous tasks produced by

   425      * {@code async} methods. This may be useful for monitoring,

   426      * debugging, and tracking asynchronous activities.

   427      *

   428      * @since 1.8

   429      */

   430     public static interface AsynchronousCompletionTask {

   431     }

   432 

   433     /** Base class can act as either FJ or plain Runnable */

   434     abstract static class Async extends ForkJoinTask<Void>

   435         implements Runnable, AsynchronousCompletionTask {

   436         public final Void getRawResult() { return null; }

   437         public final void setRawResult(Void v) { }

   438         public final void run() { exec(); }

   439     }

   440 

   441     static final class AsyncRun extends Async {

   442         final Runnable fn;

   443         final CompletableFuture<Void> dst;

   444         AsyncRun(Runnable fn, CompletableFuture<Void> dst) {

   445             this.fn = fn; this.dst = dst;

   446         }

   447         public final boolean exec() {

   448             CompletableFuture<Void> d; Throwable ex;

   449             if ((d = this.dst) != null && d.result == null) {

   450                 try {

   451                     fn.run();

   452                     ex = null;

   453                 } catch (Throwable rex) {

   454                     ex = rex;

   455                 }

   456                 d.internalComplete(null, ex);

   457             }

   458             return true;

   459         }

   460         private static final long serialVersionUID = 5232453952276885070L;

   461     }

   462 

   463     static final class AsyncSupply<U> extends Async {

   464         final Supplier<U> fn;

   465         final CompletableFuture<U> dst;

   466         AsyncSupply(Supplier<U> fn, CompletableFuture<U> dst) {

   467             this.fn = fn; this.dst = dst;

   468         }

   469         public final boolean exec() {

   470             CompletableFuture<U> d; U u; Throwable ex;

   471             if ((d = this.dst) != null && d.result == null) {

   472                 try {

   473                     u = fn.get();

   474                     ex = null;

   475                 } catch (Throwable rex) {

   476                     ex = rex;

   477                     u = null;

   478                 }

   479                 d.internalComplete(u, ex);

   480             }

   481             return true;

   482         }

   483         private static final long serialVersionUID = 5232453952276885070L;

   484     }

   485 

   486     static final class AsyncApply<T,U> extends Async {

   487         final T arg;

   488         final Function<? super T,? extends U> fn;

   489         final CompletableFuture<U> dst;

   490         AsyncApply(T arg, Function<? super T,? extends U> fn,

   491                    CompletableFuture<U> dst) {

   492             this.arg = arg; this.fn = fn; this.dst = dst;

   493         }

   494         public final boolean exec() {

   495             CompletableFuture<U> d; U u; Throwable ex;

   496             if ((d = this.dst) != null && d.result == null) {

   497                 try {

   498                     u = fn.apply(arg);

   499                     ex = null;

   500                 } catch (Throwable rex) {

   501                     ex = rex;

   502                     u = null;

   503                 }

   504                 d.internalComplete(u, ex);

   505             }

   506             return true;

   507         }

   508         private static final long serialVersionUID = 5232453952276885070L;

   509     }

   510 

   511     static final class AsyncCombine<T,U,V> extends Async {

   512         final T arg1;

   513         final U arg2;

   514         final BiFunction<? super T,? super U,? extends V> fn;

   515         final CompletableFuture<V> dst;

   516         AsyncCombine(T arg1, U arg2,

   517                      BiFunction<? super T,? super U,? extends V> fn,

   518                      CompletableFuture<V> dst) {

   519             this.arg1 = arg1; this.arg2 = arg2; this.fn = fn; this.dst = dst;

   520         }

   521         public final boolean exec() {

   522             CompletableFuture<V> d; V v; Throwable ex;

   523             if ((d = this.dst) != null && d.result == null) {

   524                 try {

   525                     v = fn.apply(arg1, arg2);

   526                     ex = null;

   527                 } catch (Throwable rex) {

   528                     ex = rex;

   529                     v = null;

   530                 }

   531                 d.internalComplete(v, ex);

   532             }

   533             return true;

   534         }

   535         private static final long serialVersionUID = 5232453952276885070L;

   536     }

   537 

   538     static final class AsyncAccept<T> extends Async {

   539         final T arg;

   540         final Consumer<? super T> fn;

   541         final CompletableFuture<Void> dst;

   542         AsyncAccept(T arg, Consumer<? super T> fn,

   543                     CompletableFuture<Void> dst) {

   544             this.arg = arg; this.fn = fn; this.dst = dst;

   545         }

   546         public final boolean exec() {

   547             CompletableFuture<Void> d; Throwable ex;

   548             if ((d = this.dst) != null && d.result == null) {

   549                 try {

   550                     fn.accept(arg);

   551                     ex = null;

   552                 } catch (Throwable rex) {

   553                     ex = rex;

   554                 }

   555                 d.internalComplete(null, ex);

   556             }

   557             return true;

   558         }

   559         private static final long serialVersionUID = 5232453952276885070L;

   560     }

   561 

   562     static final class AsyncAcceptBoth<T,U> extends Async {

   563         final T arg1;

   564         final U arg2;

   565         final BiConsumer<? super T,? super U> fn;

   566         final CompletableFuture<Void> dst;

   567         AsyncAcceptBoth(T arg1, U arg2,

   568                         BiConsumer<? super T,? super U> fn,

   569                         CompletableFuture<Void> dst) {

   570             this.arg1 = arg1; this.arg2 = arg2; this.fn = fn; this.dst = dst;

   571         }

   572         public final boolean exec() {

   573             CompletableFuture<Void> d; Throwable ex;

   574             if ((d = this.dst) != null && d.result == null) {

   575                 try {

   576                     fn.accept(arg1, arg2);

   577                     ex = null;

   578                 } catch (Throwable rex) {

   579                     ex = rex;

   580                 }

   581                 d.internalComplete(null, ex);

   582             }

   583             return true;

   584         }

   585         private static final long serialVersionUID = 5232453952276885070L;

   586     }

   587 

   588     static final class AsyncCompose<T,U> extends Async {

   589         final T arg;

   590         final Function<? super T, CompletableFuture<U>> fn;

   591         final CompletableFuture<U> dst;

   592         AsyncCompose(T arg,

   593                      Function<? super T, CompletableFuture<U>> fn,

   594                      CompletableFuture<U> dst) {

   595             this.arg = arg; this.fn = fn; this.dst = dst;

   596         }

   597         public final boolean exec() {

   598             CompletableFuture<U> d, fr; U u; Throwable ex;

   599             if ((d = this.dst) != null && d.result == null) {

   600                 try {

   601                     fr = fn.apply(arg);

   602                     ex = (fr == null) ? new NullPointerException() : null;

   603                 } catch (Throwable rex) {

   604                     ex = rex;

   605                     fr = null;

   606                 }

   607                 if (ex != null)

   608                     u = null;

   609                 else {

   610                     Object r = fr.result;

   611                     if (r == null)

   612                         r = fr.waitingGet(false);

   613                     if (r instanceof AltResult) {

   614                         ex = ((AltResult)r).ex;

   615                         u = null;

   616                     }

   617                     else {

   618                         @SuppressWarnings("unchecked") U ur = (U) r;

   619                         u = ur;

   620                     }

   621                 }

   622                 d.internalComplete(u, ex);

   623             }

   624             return true;

   625         }

   626         private static final long serialVersionUID = 5232453952276885070L;

   627     }

   628 

   629     /* ------------- Completions -------------- */

   630 

   631     /**

   632      * Simple linked list nodes to record completions, used in

   633      * basically the same way as WaitNodes. (We separate nodes from

   634      * the Completions themselves mainly because for the And and Or

   635      * methods, the same Completion object resides in two lists.)

   636      */

   637     static final class CompletionNode {

   638         final Completion completion;

   639         volatile CompletionNode next;

   640         CompletionNode(Completion completion) { this.completion = completion; }

   641     }

   642 

   643     // Opportunistically subclass AtomicInteger to use compareAndSet to claim.

   644     abstract static class Completion extends AtomicInteger implements Runnable {

   645     }

   646 

   647     static final class ThenApply<T,U> extends Completion {

   648         final CompletableFuture<? extends T> src;

   649         final Function<? super T,? extends U> fn;

   650         final CompletableFuture<U> dst;

   651         final Executor executor;

   652         ThenApply(CompletableFuture<? extends T> src,

   653                   Function<? super T,? extends U> fn,

   654                   CompletableFuture<U> dst,

   655                   Executor executor) {

   656             this.src = src; this.fn = fn; this.dst = dst;

   657             this.executor = executor;

   658         }

   659         public final void run() {

   660             final CompletableFuture<? extends T> a;

   661             final Function<? super T,? extends U> fn;

   662             final CompletableFuture<U> dst;

   663             Object r; T t; Throwable ex;

   664             if ((dst = this.dst) != null &&

   665                 (fn = this.fn) != null &&

   666                 (a = this.src) != null &&

   667                 (r = a.result) != null &&

   668                 compareAndSet(0, 1)) {

   669                 if (r instanceof AltResult) {

   670                     ex = ((AltResult)r).ex;

   671                     t = null;

   672                 }

   673                 else {

   674                     ex = null;

   675                     @SuppressWarnings("unchecked") T tr = (T) r;

   676                     t = tr;

   677                 }

   678                 Executor e = executor;

   679                 U u = null;

   680                 if (ex == null) {

   681                     try {

   682                         if (e != null)

   683                             e.execute(new AsyncApply<T,U>(t, fn, dst));

   684                         else

   685                             u = fn.apply(t);

   686                     } catch (Throwable rex) {

   687                         ex = rex;

   688                     }

   689                 }

   690                 if (e == null || ex != null)

   691                     dst.internalComplete(u, ex);

   692             }

   693         }

   694         private static final long serialVersionUID = 5232453952276885070L;

   695     }

   696 

   697     static final class ThenAccept<T> extends Completion {

   698         final CompletableFuture<? extends T> src;

   699         final Consumer<? super T> fn;

   700         final CompletableFuture<Void> dst;

   701         final Executor executor;

   702         ThenAccept(CompletableFuture<? extends T> src,

   703                    Consumer<? super T> fn,

   704                    CompletableFuture<Void> dst,

   705                    Executor executor) {

   706             this.src = src; this.fn = fn; this.dst = dst;

   707             this.executor = executor;

   708         }

   709         public final void run() {

   710             final CompletableFuture<? extends T> a;

   711             final Consumer<? super T> fn;

   712             final CompletableFuture<Void> dst;

   713             Object r; T t; Throwable ex;

   714             if ((dst = this.dst) != null &&

   715                 (fn = this.fn) != null &&

   716                 (a = this.src) != null &&

   717                 (r = a.result) != null &&

   718                 compareAndSet(0, 1)) {

   719                 if (r instanceof AltResult) {

   720                     ex = ((AltResult)r).ex;

   721                     t = null;

   722                 }

   723                 else {

   724                     ex = null;

   725                     @SuppressWarnings("unchecked") T tr = (T) r;

   726                     t = tr;

   727                 }

   728                 Executor e = executor;

   729                 if (ex == null) {

   730                     try {

   731                         if (e != null)

   732                             e.execute(new AsyncAccept<T>(t, fn, dst));

   733                         else

   734                             fn.accept(t);

   735                     } catch (Throwable rex) {

   736                         ex = rex;

   737                     }

   738                 }

   739                 if (e == null || ex != null)

   740                     dst.internalComplete(null, ex);

   741             }

   742         }

   743         private static final long serialVersionUID = 5232453952276885070L;

   744     }

   745 

   746     static final class ThenRun extends Completion {

   747         final CompletableFuture<?> src;

   748         final Runnable fn;

   749         final CompletableFuture<Void> dst;

   750         final Executor executor;

   751         ThenRun(CompletableFuture<?> src,

   752                 Runnable fn,

   753                 CompletableFuture<Void> dst,

   754                 Executor executor) {

   755             this.src = src; this.fn = fn; this.dst = dst;

   756             this.executor = executor;

   757         }

   758         public final void run() {

   759             final CompletableFuture<?> a;

   760             final Runnable fn;

   761             final CompletableFuture<Void> dst;

   762             Object r; Throwable ex;

   763             if ((dst = this.dst) != null &&

   764                 (fn = this.fn) != null &&

   765                 (a = this.src) != null &&

   766                 (r = a.result) != null &&

   767                 compareAndSet(0, 1)) {

   768                 if (r instanceof AltResult)

   769                     ex = ((AltResult)r).ex;

   770                 else

   771                     ex = null;

   772                 Executor e = executor;

   773                 if (ex == null) {

   774                     try {

   775                         if (e != null)

   776                             e.execute(new AsyncRun(fn, dst));

   777                         else

   778                             fn.run();

   779                     } catch (Throwable rex) {

   780                         ex = rex;

   781                     }

   782                 }

   783                 if (e == null || ex != null)

   784                     dst.internalComplete(null, ex);

   785             }

   786         }

   787         private static final long serialVersionUID = 5232453952276885070L;

   788     }

   789 

   790     static final class ThenCombine<T,U,V> extends Completion {

   791         final CompletableFuture<? extends T> src;

   792         final CompletableFuture<? extends U> snd;

   793         final BiFunction<? super T,? super U,? extends V> fn;

   794         final CompletableFuture<V> dst;

   795         final Executor executor;

   796         ThenCombine(CompletableFuture<? extends T> src,

   797                     CompletableFuture<? extends U> snd,

   798                     BiFunction<? super T,? super U,? extends V> fn,

   799                     CompletableFuture<V> dst,

   800                     Executor executor) {

   801             this.src = src; this.snd = snd;

   802             this.fn = fn; this.dst = dst;

   803             this.executor = executor;

   804         }

   805         public final void run() {

   806             final CompletableFuture<? extends T> a;

   807             final CompletableFuture<? extends U> b;

   808             final BiFunction<? super T,? super U,? extends V> fn;

   809             final CompletableFuture<V> dst;

   810             Object r, s; T t; U u; Throwable ex;

   811             if ((dst = this.dst) != null &&

   812                 (fn = this.fn) != null &&

   813                 (a = this.src) != null &&

   814                 (r = a.result) != null &&

   815                 (b = this.snd) != null &&

   816                 (s = b.result) != null &&

   817                 compareAndSet(0, 1)) {

   818                 if (r instanceof AltResult) {

   819                     ex = ((AltResult)r).ex;

   820                     t = null;

   821                 }

   822                 else {

   823                     ex = null;

   824                     @SuppressWarnings("unchecked") T tr = (T) r;

   825                     t = tr;

   826                 }

   827                 if (ex != null)

   828                     u = null;

   829                 else if (s instanceof AltResult) {

   830                     ex = ((AltResult)s).ex;

   831                     u = null;

   832                 }

   833                 else {

   834                     @SuppressWarnings("unchecked") U us = (U) s;

   835                     u = us;

   836                 }

   837                 Executor e = executor;

   838                 V v = null;

   839                 if (ex == null) {

   840                     try {

   841                         if (e != null)

   842                             e.execute(new AsyncCombine<T,U,V>(t, u, fn, dst));

   843                         else

   844                             v = fn.apply(t, u);

   845                     } catch (Throwable rex) {

   846                         ex = rex;

   847                     }

   848                 }

   849                 if (e == null || ex != null)

   850                     dst.internalComplete(v, ex);

   851             }

   852         }

   853         private static final long serialVersionUID = 5232453952276885070L;

   854     }

   855 

   856     static final class ThenAcceptBoth<T,U> extends Completion {

   857         final CompletableFuture<? extends T> src;

   858         final CompletableFuture<? extends U> snd;

   859         final BiConsumer<? super T,? super U> fn;

   860         final CompletableFuture<Void> dst;

   861         final Executor executor;

   862         ThenAcceptBoth(CompletableFuture<? extends T> src,

   863                        CompletableFuture<? extends U> snd,

   864                        BiConsumer<? super T,? super U> fn,

   865                        CompletableFuture<Void> dst,

   866                        Executor executor) {

   867             this.src = src; this.snd = snd;

   868             this.fn = fn; this.dst = dst;

   869             this.executor = executor;

   870         }

   871         public final void run() {

   872             final CompletableFuture<? extends T> a;

   873             final CompletableFuture<? extends U> b;

   874             final BiConsumer<? super T,? super U> fn;

   875             final CompletableFuture<Void> dst;

   876             Object r, s; T t; U u; Throwable ex;

   877             if ((dst = this.dst) != null &&

   878                 (fn = this.fn) != null &&

   879                 (a = this.src) != null &&

   880                 (r = a.result) != null &&

   881                 (b = this.snd) != null &&

   882                 (s = b.result) != null &&

   883                 compareAndSet(0, 1)) {

   884                 if (r instanceof AltResult) {

   885                     ex = ((AltResult)r).ex;

   886                     t = null;

   887                 }

   888                 else {

   889                     ex = null;

   890                     @SuppressWarnings("unchecked") T tr = (T) r;

   891                     t = tr;

   892                 }

   893                 if (ex != null)

   894                     u = null;

   895                 else if (s instanceof AltResult) {

   896                     ex = ((AltResult)s).ex;

   897                     u = null;

   898                 }

   899                 else {

   900                     @SuppressWarnings("unchecked") U us = (U) s;

   901                     u = us;

   902                 }

   903                 Executor e = executor;

   904                 if (ex == null) {

   905                     try {

   906                         if (e != null)

   907                             e.execute(new AsyncAcceptBoth<T,U>(t, u, fn, dst));

   908                         else

   909                             fn.accept(t, u);

   910                     } catch (Throwable rex) {

   911                         ex = rex;

   912                     }

   913                 }

   914                 if (e == null || ex != null)

   915                     dst.internalComplete(null, ex);

   916             }

   917         }

   918         private static final long serialVersionUID = 5232453952276885070L;

   919     }

   920 

   921     static final class RunAfterBoth extends Completion {

   922         final CompletableFuture<?> src;

   923         final CompletableFuture<?> snd;

   924         final Runnable fn;

   925         final CompletableFuture<Void> dst;

   926         final Executor executor;

   927         RunAfterBoth(CompletableFuture<?> src,

   928                      CompletableFuture<?> snd,

   929                      Runnable fn,

   930                      CompletableFuture<Void> dst,

   931                      Executor executor) {

   932             this.src = src; this.snd = snd;

   933             this.fn = fn; this.dst = dst;

   934             this.executor = executor;

   935         }

   936         public final void run() {

   937             final CompletableFuture<?> a;

   938             final CompletableFuture<?> b;

   939             final Runnable fn;

   940             final CompletableFuture<Void> dst;

   941             Object r, s; Throwable ex;

   942             if ((dst = this.dst) != null &&

   943                 (fn = this.fn) != null &&

   944                 (a = this.src) != null &&

   945                 (r = a.result) != null &&

   946                 (b = this.snd) != null &&

   947                 (s = b.result) != null &&

   948                 compareAndSet(0, 1)) {

   949                 if (r instanceof AltResult)

   950                     ex = ((AltResult)r).ex;

   951                 else

   952                     ex = null;

   953                 if (ex == null && (s instanceof AltResult))

   954                     ex = ((AltResult)s).ex;

   955                 Executor e = executor;

   956                 if (ex == null) {

   957                     try {

   958                         if (e != null)

   959                             e.execute(new AsyncRun(fn, dst));

   960                         else

   961                             fn.run();

   962                     } catch (Throwable rex) {

   963                         ex = rex;

   964                     }

   965                 }

   966                 if (e == null || ex != null)

   967                     dst.internalComplete(null, ex);

   968             }

   969         }

   970         private static final long serialVersionUID = 5232453952276885070L;

   971     }

   972 

   973     static final class AndCompletion extends Completion {

   974         final CompletableFuture<?> src;

   975         final CompletableFuture<?> snd;

   976         final CompletableFuture<Void> dst;

   977         AndCompletion(CompletableFuture<?> src,

   978                       CompletableFuture<?> snd,

   979                       CompletableFuture<Void> dst) {

   980             this.src = src; this.snd = snd; this.dst = dst;

   981         }

   982         public final void run() {

   983             final CompletableFuture<?> a;

   984             final CompletableFuture<?> b;

   985             final CompletableFuture<Void> dst;

   986             Object r, s; Throwable ex;

   987             if ((dst = this.dst) != null &&

   988                 (a = this.src) != null &&

   989                 (r = a.result) != null &&

   990                 (b = this.snd) != null &&

   991                 (s = b.result) != null &&

   992                 compareAndSet(0, 1)) {

   993                 if (r instanceof AltResult)

   994                     ex = ((AltResult)r).ex;

   995                 else

   996                     ex = null;

   997                 if (ex == null && (s instanceof AltResult))

   998                     ex = ((AltResult)s).ex;

   999                 dst.internalComplete(null, ex);

  1000             }

  1001         }

  1002         private static final long serialVersionUID = 5232453952276885070L;

  1003     }

  1004 

  1005     static final class ApplyToEither<T,U> extends Completion {

  1006         final CompletableFuture<? extends T> src;

  1007         final CompletableFuture<? extends T> snd;

  1008         final Function<? super T,? extends U> fn;

  1009         final CompletableFuture<U> dst;

  1010         final Executor executor;

  1011         ApplyToEither(CompletableFuture<? extends T> src,

  1012                       CompletableFuture<? extends T> snd,

  1013                       Function<? super T,? extends U> fn,

  1014                       CompletableFuture<U> dst,

  1015                       Executor executor) {

  1016             this.src = src; this.snd = snd;

  1017             this.fn = fn; this.dst = dst;

  1018             this.executor = executor;

  1019         }

  1020         public final void run() {

  1021             final CompletableFuture<? extends T> a;

  1022             final CompletableFuture<? extends T> b;

  1023             final Function<? super T,? extends U> fn;

  1024             final CompletableFuture<U> dst;

  1025             Object r; T t; Throwable ex;

  1026             if ((dst = this.dst) != null &&

  1027                 (fn = this.fn) != null &&

  1028                 (((a = this.src) != null && (r = a.result) != null) ||

  1029                  ((b = this.snd) != null && (r = b.result) != null)) &&

  1030                 compareAndSet(0, 1)) {

  1031                 if (r instanceof AltResult) {

  1032                     ex = ((AltResult)r).ex;

  1033                     t = null;

  1034                 }

  1035                 else {

  1036                     ex = null;

  1037                     @SuppressWarnings("unchecked") T tr = (T) r;

  1038                     t = tr;

  1039                 }

  1040                 Executor e = executor;

  1041                 U u = null;

  1042                 if (ex == null) {

  1043                     try {

  1044                         if (e != null)

  1045                             e.execute(new AsyncApply<T,U>(t, fn, dst));

  1046                         else

  1047                             u = fn.apply(t);

  1048                     } catch (Throwable rex) {

  1049                         ex = rex;

  1050                     }

  1051                 }

  1052                 if (e == null || ex != null)

  1053                     dst.internalComplete(u, ex);

  1054             }

  1055         }

  1056         private static final long serialVersionUID = 5232453952276885070L;

  1057     }

  1058 

  1059     static final class AcceptEither<T> extends Completion {

  1060         final CompletableFuture<? extends T> src;

  1061         final CompletableFuture<? extends T> snd;

  1062         final Consumer<? super T> fn;

  1063         final CompletableFuture<Void> dst;

  1064         final Executor executor;

  1065         AcceptEither(CompletableFuture<? extends T> src,

  1066                      CompletableFuture<? extends T> snd,

  1067                      Consumer<? super T> fn,

  1068                      CompletableFuture<Void> dst,

  1069                      Executor executor) {

  1070             this.src = src; this.snd = snd;

  1071             this.fn = fn; this.dst = dst;

  1072             this.executor = executor;

  1073         }

  1074         public final void run() {

  1075             final CompletableFuture<? extends T> a;

  1076             final CompletableFuture<? extends T> b;

  1077             final Consumer<? super T> fn;

  1078             final CompletableFuture<Void> dst;

  1079             Object r; T t; Throwable ex;

  1080             if ((dst = this.dst) != null &&

  1081                 (fn = this.fn) != null &&

  1082                 (((a = this.src) != null && (r = a.result) != null) ||

  1083                  ((b = this.snd) != null && (r = b.result) != null)) &&

  1084                 compareAndSet(0, 1)) {

  1085                 if (r instanceof AltResult) {

  1086                     ex = ((AltResult)r).ex;

  1087                     t = null;

  1088                 }

  1089                 else {

  1090                     ex = null;

  1091                     @SuppressWarnings("unchecked") T tr = (T) r;

  1092                     t = tr;

  1093                 }

  1094                 Executor e = executor;

  1095                 if (ex == null) {

  1096                     try {

  1097                         if (e != null)

  1098                             e.execute(new AsyncAccept<T>(t, fn, dst));

  1099                         else

  1100                             fn.accept(t);

  1101                     } catch (Throwable rex) {

  1102                         ex = rex;

  1103                     }

  1104                 }

  1105                 if (e == null || ex != null)

  1106                     dst.internalComplete(null, ex);

  1107             }

  1108         }

  1109         private static final long serialVersionUID = 5232453952276885070L;

  1110     }

  1111 

  1112     static final class RunAfterEither extends Completion {

  1113         final CompletableFuture<?> src;

  1114         final CompletableFuture<?> snd;

  1115         final Runnable fn;

  1116         final CompletableFuture<Void> dst;

  1117         final Executor executor;

  1118         RunAfterEither(CompletableFuture<?> src,

  1119                        CompletableFuture<?> snd,

  1120                        Runnable fn,

  1121                        CompletableFuture<Void> dst,

  1122                        Executor executor) {

  1123             this.src = src; this.snd = snd;

  1124             this.fn = fn; this.dst = dst;

  1125             this.executor = executor;

  1126         }

  1127         public final void run() {

  1128             final CompletableFuture<?> a;

  1129             final CompletableFuture<?> b;

  1130             final Runnable fn;

  1131             final CompletableFuture<Void> dst;

  1132             Object r; Throwable ex;

  1133             if ((dst = this.dst) != null &&

  1134                 (fn = this.fn) != null &&

  1135                 (((a = this.src) != null && (r = a.result) != null) ||