80  * which case, a new Thread is used). To simplify monitoring,

    79  * which case, a new Thread is created to run each task).  To simplify

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

    80  * monitoring, debugging, and tracking, all generated asynchronous

    82  * instances of the marker interface {@link

    81  * tasks are instances of the marker interface {@link

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

   115      * A CompletableFuture may have dependent completion actions,

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

   116      * collected in a linked stack. It atomically completes by CASing

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

   117      * a result field, and then pops off and runs those actions. This

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

   118      * applies across normal vs exceptional outcomes, sync vs async

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

   119      * actions, binary triggers, and various forms of completions.

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

   120      *

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

   121      * Non-nullness of field result (set via CAS) indicates done.  An

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

   122      * AltResult is used to box null as a result, as well as to hold

   124      * exception propagation mechanics surrounding decoding rely on

   123      * exceptions.  Using a single field makes completion simple to

   125      * unchecked casts of decoded results really being unchecked,

   124      * detect and trigger.  Encoding and decoding is straightforward

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

   125      * but adds to the sprawl of trapping and associating exceptions

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

   126      * with targets.  Minor simplifications rely on (static) NIL (to

   128      * SuppressWarnings-annotated temporaries.

   127      * box null results) being the only AltResult with a null

   129      *

   128      * exception field, so we don't usually need explicit comparisons.

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

   129      * Even though some of the generics casts are unchecked (see

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

   130      * SuppressWarnings annotations), they are placed to be

   132      * internal documentation for algorithmic details.

   131      * appropriate even if checked.

   133      *

   132      *

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

   133      * Dependent actions are represented by Completion objects linked

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

   134      * as Treiber stacks headed by field "stack". There are Completion

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

   135      * classes for each kind of action, grouped into single-input

   137      * parallelism obtained because woken waiters help release/run

   136      * (UniCompletion), two-input (BiCompletion), projected

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

   137      * (BiCompletions using either (not both) of two inputs), shared

   139      * may race with direct calls, class Completion opportunistically

   138      * (CoCompletion, used by the second of two sources), zero-input

   140      * extends AtomicInteger so callers can claim the action via

   139      * source actions, and Signallers that unblock waiters. Class

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

   140      * Completion extends ForkJoinTask to enable async execution

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

   141      * (adding no space overhead because we exploit its "tag" methods

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

   142      * to maintain claims). It is also declared as Runnable to allow

   144      * There are enough dimensions upon which they differ that

   143      * usage with arbitrary executors.

   145      * attempts to factor commonalities while maintaining efficiency

   144      *

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

   145      * Support for each kind of CompletionStage relies on a separate

   147      *

   146      * class, along with two CompletableFuture methods:

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

   147      *

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

   148      * * A Completion class with name X corresponding to function,

   150      * intrinsic races surrounding addition of a dependent action

   149      *   prefaced with "Uni", "Bi", or "Or". Each class contains

   151      * versus performing the action directly because the task is

   150      *   fields for source(s), actions, and dependent. They are

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

   151      *   boringly similar, differing from others only with respect to

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

   152      *   underlying functional forms. We do this so that users don't

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

   153      *   encounter layers of adaptors in common usages. We also

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

   154      *   include "Relay" classes/methods that don't correspond to user

   156      * construction in safe-bypass cases.

   155      *   methods; they copy results from one stage to another.

   157      */

   156      *

   158 

   157      * * Boolean CompletableFuture method x(...) (for example

   159     // preliminaries

   158      *   uniApply) takes all of the arguments needed to check that an

   160 

   159      *   action is triggerable, and then either runs the action or

   161     static final class AltResult {

   160      *   arranges its async execution by executing its Completion

   162         final Throwable ex; // null only for NIL

   161      *   argument, if present. The method returns true if known to be

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

   162      *   complete.

   164     }

   163      *

   165 

   164      * * Completion method tryFire(int mode) invokes the associated x

   165      *   method with its held arguments, and on success cleans up.

   166      *   The mode argument allows tryFire to be called twice (SYNC,

   167      *   then ASYNC); the first to screen and trap exceptions while

   168      *   arranging to execute, and the second when called from a

   169      *   task. (A few classes are not used async so take slightly

   170      *   different forms.)  The claim() callback suppresses function

   171      *   invocation if already claimed by another thread.

   172      *

   173      * * CompletableFuture method xStage(...) is called from a public

   174      *   stage method of CompletableFuture x. It screens user

   175      *   arguments and invokes and/or creates the stage object.  If

   176      *   not async and x is already complete, the action is run

   177      *   immediately.  Otherwise a Completion c is created, pushed to

   178      *   x's stack (unless done), and started or triggered via

   179      *   c.tryFire.  This also covers races possible if x completes

   180      *   while pushing.  Classes with two inputs (for example BiApply)

   181      *   deal with races across both while pushing actions.  The

   182      *   second completion is a CoCompletion pointing to the first,

   183      *   shared so that at most one performs the action.  The

   184      *   multiple-arity methods allOf and anyOf do this pairwise to

   185      *   form trees of completions.

   186      *

   187      * Note that the generic type parameters of methods vary according

   188      * to whether "this" is a source, dependent, or completion.

   189      *

   190      * Method postComplete is called upon completion unless the target

   191      * is guaranteed not to be observable (i.e., not yet returned or

   192      * linked). Multiple threads can call postComplete, which

   193      * atomically pops each dependent action, and tries to trigger it

   194      * via method tryFire, in NESTED mode.  Triggering can propagate

   195      * recursively, so NESTED mode returns its completed dependent (if

   196      * one exists) for further processing by its caller (see method

   197      * postFire).

   198      *

   199      * Blocking methods get() and join() rely on Signaller Completions

   200      * that wake up waiting threads.  The mechanics are similar to

   201      * Treiber stack wait-nodes used in FutureTask, Phaser, and

   202      * SynchronousQueue. See their internal documentation for

   203      * algorithmic details.

   204      *

   205      * Without precautions, CompletableFutures would be prone to

   206      * garbage accumulation as chains of Completions build up, each

   207      * pointing back to its sources. So we null out fields as soon as

   208      * possible (see especially method Completion.detach). The

   209      * screening checks needed anyway harmlessly ignore null arguments

   210      * that may have been obtained during races with threads nulling

   211      * out fields.  We also try to unlink fired Completions from

   212      * stacks that might never be popped (see method postFire).

   213      * Completion fields need not be declared as final or volatile

   214      * because they are only visible to other threads upon safe

   215      * publication.

   216      */

   217 

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

   219     volatile Completion stack;    // Top of Treiber stack of dependent actions

   220 

   221     final boolean internalComplete(Object r) { // CAS from null to r

   222         return UNSAFE.compareAndSwapObject(this, RESULT, null, r);

   223     }

   224 

   225     final boolean casStack(Completion cmp, Completion val) {

   226         return UNSAFE.compareAndSwapObject(this, STACK, cmp, val);

   227     }

   228 

   229     /** Returns true if successfully pushed c onto stack. */

   230     final boolean tryPushStack(Completion c) {

   231         Completion h = stack;

   232         lazySetNext(c, h);

   233         return UNSAFE.compareAndSwapObject(this, STACK, h, c);

   234     }

   235 

   236     /** Unconditionally pushes c onto stack, retrying if necessary. */

   237     final void pushStack(Completion c) {

   238         do {} while (!tryPushStack(c));

   239     }

   240 

   241     /* ------------- Encoding and decoding outcomes -------------- */

   242 

   243     static final class AltResult { // See above

   244         final Throwable ex;        // null only for NIL

   245         AltResult(Throwable x) { this.ex = x; }

   246     }

   247 

   248     /** The encoding of the null value. */

   168     // Fields

   251     /** Completes with the null value, unless already completed. */

   169 

   252     final boolean completeNull() {

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

   253         return UNSAFE.compareAndSwapObject(this, RESULT, null,

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

   254                                            NIL);

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

   255     }

   173 

   256 

   174     // Basic utilities for triggering and processing completions

   257     /** Returns the encoding of the given non-exceptional value. */

   175 

   258     final Object encodeValue(T t) {

   176     /**

   259         return (t == null) ? NIL : t;

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

   260     }

   261 

   262     /** Completes with a non-exceptional result, unless already completed. */

   263     final boolean completeValue(T t) {

   264         return UNSAFE.compareAndSwapObject(this, RESULT, null,

   265                                            (t == null) ? NIL : t);

   266     }

   267 

   268     /**

   269      * Returns the encoding of the given (non-null) exception as a

   270      * wrapped CompletionException unless it is one already.

   271      */

   272     static AltResult encodeThrowable(Throwable x) {

   273         return new AltResult((x instanceof CompletionException) ? x :

   274                              new CompletionException(x));

   275     }

   276 

   277     /** Completes with an exceptional result, unless already completed. */

   278     final boolean completeThrowable(Throwable x) {

   279         return UNSAFE.compareAndSwapObject(this, RESULT, null,

   280                                            encodeThrowable(x));

   281     }

   282 

   283     /**

   284      * Returns the encoding of the given (non-null) exception as a

   285      * wrapped CompletionException unless it is one already.  May

   286      * return the given Object r (which must have been the result of a

   287      * source future) if it is equivalent, i.e. if this is a simple

   288      * relay of an existing CompletionException.

   289      */

   290     static Object encodeThrowable(Throwable x, Object r) {

   291         if (!(x instanceof CompletionException))

   292             x = new CompletionException(x);

   293         else if (r instanceof AltResult && x == ((AltResult)r).ex)

   294             return r;

   295         return new AltResult(x);

   296     }

   297 

   298     /**

   299      * Completes with the given (non-null) exceptional result as a

   300      * wrapped CompletionException unless it is one already, unless

   301      * already completed.  May complete with the given Object r

   302      * (which must have been the result of a source future) if it is

   303      * equivalent, i.e. if this is a simple propagation of an

   304      * existing CompletionException.

   305      */

   306     final boolean completeThrowable(Throwable x, Object r) {

   307         return UNSAFE.compareAndSwapObject(this, RESULT, null,

   308                                            encodeThrowable(x, r));

   309     }

   310 

   311     /**

   312      * Returns the encoding of the given arguments: if the exception

   313      * is non-null, encodes as AltResult.  Otherwise uses the given

   314      * value, boxed as NIL if null.

   315      */

   316     Object encodeOutcome(T t, Throwable x) {

   317         return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x);

   318     }

   319 

   320     /**

   321      * Returns the encoding of a copied outcome; if exceptional,

   322      * rewraps as a CompletionException, else returns argument.

   323      */

   324     static Object encodeRelay(Object r) {

   325         Throwable x;

   326         return (((r instanceof AltResult) &&

   327                  (x = ((AltResult)r).ex) != null &&

   328                  !(x instanceof CompletionException)) ?

   329                 new AltResult(new CompletionException(x)) : r);

   330     }

   331 

   332     /**

   333      * Completes with r or a copy of r, unless already completed.

   334      * If exceptional, r is first coerced to a CompletionException.

   335      */

   336     final boolean completeRelay(Object r) {

   337         return UNSAFE.compareAndSwapObject(this, RESULT, null,

   338                                            encodeRelay(r));

   339     }

   340 

   341     /**

   342      * Reports result using Future.get conventions.

   343      */

   344     private static <T> T reportGet(Object r)

   345         throws InterruptedException, ExecutionException {

   346         if (r == null) // by convention below, null means interrupted

   347             throw new InterruptedException();

   348         if (r instanceof AltResult) {

   349             Throwable x, cause;

   350             if ((x = ((AltResult)r).ex) == null)

   351                 return null;

   352             if (x instanceof CancellationException)

   353                 throw (CancellationException)x;

   354             if ((x instanceof CompletionException) &&

   355                 (cause = x.getCause()) != null)

   356                 x = cause;

   357             throw new ExecutionException(x);

   358         }

   359         @SuppressWarnings("unchecked") T t = (T) r;

   360         return t;

   361     }

   362 

   363     /**

   364      * Decodes outcome to return result or throw unchecked exception.

   365      */

   366     private static <T> T reportJoin(Object r) {

   367         if (r instanceof AltResult) {

   368             Throwable x;

   369             if ((x = ((AltResult)r).ex) == null)

   370                 return null;

   371             if (x instanceof CancellationException)

   372                 throw (CancellationException)x;

   373             if (x instanceof CompletionException)

   374                 throw (CompletionException)x;

   375             throw new CompletionException(x);

   376         }

   377         @SuppressWarnings("unchecked") T t = (T) r;

   378         return t;

   379     }

   380 

   381     /* ------------- Async task preliminaries -------------- */

   382 

   383     /**

   384      * A marker interface identifying asynchronous tasks produced by

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

   386      * debugging, and tracking asynchronous activities.

   387      *

   388      * @since 1.8

   389      */

   390     public static interface AsynchronousCompletionTask {

   391     }

   392 

   393     private static final boolean useCommonPool =

   394         (ForkJoinPool.getCommonPoolParallelism() > 1);

   395 

   396     /**

   397      * Default executor -- ForkJoinPool.commonPool() unless it cannot

   398      * support parallelism.

   399      */

   400     private static final Executor asyncPool = useCommonPool ?

   401         ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

   402 

   403     /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */

   404     static final class ThreadPerTaskExecutor implements Executor {

   405         public void execute(Runnable r) { new Thread(r).start(); }

   406     }

   407 

   408     /**

   409      * Null-checks user executor argument, and translates uses of

   410      * commonPool to asyncPool in case parallelism disabled.

   411      */

   412     static Executor screenExecutor(Executor e) {

   413         if (!useCommonPool && e == ForkJoinPool.commonPool())

   414             return asyncPool;

   415         if (e == null) throw new NullPointerException();

   416         return e;

   417     }

   418 

   419     // Modes for Completion.tryFire. Signedness matters.

   420     static final int SYNC   =  0;

   421     static final int ASYNC  =  1;

   422     static final int NESTED = -1;

   423 

   424     /* ------------- Base Completion classes and operations -------------- */

   425 

   426     @SuppressWarnings("serial")

   427     abstract static class Completion extends ForkJoinTask<Void>

   428         implements Runnable, AsynchronousCompletionTask {

   429         volatile Completion next;      // Treiber stack link

   430 

   431         /**

   432          * Performs completion action if triggered, returning a

   433          * dependent that may need propagation, if one exists.

   434          *

   435          * @param mode SYNC, ASYNC, or NESTED

   436          */

   437         abstract CompletableFuture<?> tryFire(int mode);

   438 

   439         /** Returns true if possibly still triggerable. Used by cleanStack. */

   440         abstract boolean isLive();

   441 

   442         public final void run()                { tryFire(ASYNC); }

   443         public final boolean exec()            { tryFire(ASYNC); return true; }

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

   445         public final void setRawResult(Void v) {}

   446     }

   447 

   448     static void lazySetNext(Completion c, Completion next) {

   449         UNSAFE.putOrderedObject(c, NEXT, next);

   450     }

   451 

   452     /**

   453      * Pops and tries to trigger all reachable dependents.  Call only

   454      * when known to be done.

   180         WaitNode q; Thread t;

   457         /*

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

   458          * On each step, variable f holds current dependents to pop

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

   459          * and run.  It is extended along only one path at a time,

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

   460          * pushing others to avoid unbounded recursion.

   184                 q.thread = null;

   461          */

   185                 LockSupport.unpark(t);

   462         CompletableFuture<?> f = this; Completion h;

   186             }

   463         while ((h = f.stack) != null ||

   187         }

   464                (f != this && (h = (f = this).stack) != null)) {

   188 

   465             CompletableFuture<?> d; Completion t;

   189         CompletionNode h; Completion c;

   466             if (f.casStack(h, t = h.next)) {

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

   467                 if (t != null) {

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

   468                     if (f != this) {

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

   469                         pushStack(h);

   193                 c.run();

   470                         continue;

   194         }

   471                     }

   195     }

   472                     h.next = null;    // detach

   196 

   473                 }

   197     /**

   474                 f = (d = h.tryFire(NESTED)) == null ? this : d;

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

   475             }

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

   476         }

   200      * CompletionException unless it is one already.  Otherwise uses

   477     }

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

   478 

   202      */

   479     /** Traverses stack and unlinks dead Completions. */

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

   480     final void cleanStack() {

   481         for (Completion p = null, q = stack; q != null;) {

   482             Completion s = q.next;

   483             if (q.isLive()) {

   484                 p = q;

   485                 q = s;

   486             }

   487             else if (p == null) {

   488                 casStack(q, s);

   489                 q = stack;

   490             }

   491             else {

   492                 p.next = s;

   493                 if (p.isLive())

   494                     q = s;

   495                 else {

   496                     p = null;  // restart

   497                     q = stack;

   498                 }

   499             }

   500         }

   501     }

   502 

   503     /* ------------- One-input Completions -------------- */

   504 

   505     /** A Completion with a source, dependent, and executor. */

   506     @SuppressWarnings("serial")

   507     abstract static class UniCompletion<T,V> extends Completion {

   508         Executor executor;                 // executor to use (null if none)

   509         CompletableFuture<V> dep;          // the dependent to complete

   510         CompletableFuture<T> src;          // source for action

   511 

   512         UniCompletion(Executor executor, CompletableFuture<V> dep,

   513                       CompletableFuture<T> src) {

   514             this.executor = executor; this.dep = dep; this.src = src;

   515         }

   516 

   517         /**

   518          * Returns true if action can be run. Call only when known to

   519          * be triggerable. Uses FJ tag bit to ensure that only one

   520          * thread claims ownership.  If async, starts as task -- a

   521          * later call to tryFire will run action.

   522          */

   523         final boolean claim() {

   524             Executor e = executor;

   525             if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {

   526                 if (e == null)

   527                     return true;

   528                 executor = null; // disable

   529                 e.execute(this);

   530             }

   531             return false;

   532         }

   533 

   534         final boolean isLive() { return dep != null; }

   535     }

   536 

   537     /** Pushes the given completion (if it exists) unless done. */

   538     final void push(UniCompletion<?,?> c) {

   539         if (c != null) {

   540             while (result == null && !tryPushStack(c))

   541                 lazySetNext(c, null); // clear on failure

   542         }

   543     }

   544 

   545     /**

   546      * Post-processing by dependent after successful UniCompletion

   547      * tryFire.  Tries to clean stack of source a, and then either runs

   548      * postComplete or returns this to caller, depending on mode.

   549      */

   550     final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {

   551         if (a != null && a.stack != null) {

   552             if (mode < 0 || a.result == null)

   553                 a.cleanStack();

   554             else

   555                 a.postComplete();

   556         }

   557         if (result != null && stack != null) {

   558             if (mode < 0)

   559                 return this;

   560             else

   561                 postComplete();

   562         }

   563         return null;

   564     }

   565 

   566     @SuppressWarnings("serial")

   567     static final class UniApply<T,V> extends UniCompletion<T,V> {

   568         Function<? super T,? extends V> fn;

   569         UniApply(Executor executor, CompletableFuture<V> dep,

   570                  CompletableFuture<T> src,

   571                  Function<? super T,? extends V> fn) {

   572             super(executor, dep, src); this.fn = fn;

   573         }

   574         final CompletableFuture<V> tryFire(int mode) {

   575             CompletableFuture<V> d; CompletableFuture<T> a;

   576             if ((d = dep) == null ||

   577                 !d.uniApply(a = src, fn, mode > 0 ? null : this))

   578                 return null;

   579             dep = null; src = null; fn = null;

   580             return d.postFire(a, mode);

   581         }

   582     }

   583 

   584     final <S> boolean uniApply(CompletableFuture<S> a,

   585                                Function<? super S,? extends T> f,

   586                                UniApply<S,T> c) {

   587         Object r; Throwable x;

   588         if (a == null || (r = a.result) == null || f == null)

   589             return false;

   590         tryComplete: if (result == null) {

   591             if (r instanceof AltResult) {

   592                 if ((x = ((AltResult)r).ex) != null) {

   593                     completeThrowable(x, r);

   594                     break tryComplete;

   595                 }

   596                 r = null;

   597             }

   598             try {

   599                 if (c != null && !c.claim())

   600                     return false;

   601                 @SuppressWarnings("unchecked") S s = (S) r;

   602                 completeValue(f.apply(s));

   603             } catch (Throwable ex) {

   604                 completeThrowable(ex);

   605             }

   606         }

   607         return true;

   608     }

   609 

   610     private <V> CompletableFuture<V> uniApplyStage(

   611         Executor e, Function<? super T,? extends V> f) {

   612         if (f == null) throw new NullPointerException();

   613         CompletableFuture<V> d =  new CompletableFuture<V>();

   614         if (e != null || !d.uniApply(this, f, null)) {

   615             UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);

   616             push(c);

   617             c.tryFire(SYNC);

   618         }

   619         return d;

   620     }

   621 

   622     @SuppressWarnings("serial")

   623     static final class UniAccept<T> extends UniCompletion<T,Void> {

   624         Consumer<? super T> fn;

   625         UniAccept(Executor executor, CompletableFuture<Void> dep,

   626                   CompletableFuture<T> src, Consumer<? super T> fn) {

   627             super(executor, dep, src); this.fn = fn;

   628         }

   629         final CompletableFuture<Void> tryFire(int mode) {

   630             CompletableFuture<Void> d; CompletableFuture<T> a;

   631             if ((d = dep) == null ||

   632                 !d.uniAccept(a = src, fn, mode > 0 ? null : this))

   633                 return null;

   634             dep = null; src = null; fn = null;

   635             return d.postFire(a, mode);

   636         }

   637     }

   638 

   639     final <S> boolean uniAccept(CompletableFuture<S> a,

   640                                 Consumer<? super S> f, UniAccept<S> c) {

   641         Object r; Throwable x;

   642         if (a == null || (r = a.result) == null || f == null)

   643             return false;

   644         tryComplete: if (result == null) {

   645             if (r instanceof AltResult) {

   646                 if ((x = ((AltResult)r).ex) != null) {

   647                     completeThrowable(x, r);

   648                     break tryComplete;

   649                 }

   650                 r = null;

   651             }

   652             try {

   653                 if (c != null && !c.claim())

   654                     return false;

   655                 @SuppressWarnings("unchecked") S s = (S) r;

   656                 f.accept(s);

   657                 completeNull();

   658             } catch (Throwable ex) {

   659                 completeThrowable(ex);

   660             }

   661         }

   662         return true;

   663     }

   664 

   665     private CompletableFuture<Void> uniAcceptStage(Executor e,

   666                                                    Consumer<? super T> f) {

   667         if (f == null) throw new NullPointerException();

   668         CompletableFuture<Void> d = new CompletableFuture<Void>();

   669         if (e != null || !d.uniAccept(this, f, null)) {

   670             UniAccept<T> c = new UniAccept<T>(e, d, this, f);

   671             push(c);

   672             c.tryFire(SYNC);

   673         }

   674         return d;

   675     }

   676 

   677     @SuppressWarnings("serial")

   678     static final class UniRun<T> extends UniCompletion<T,Void> {

   679         Runnable fn;

   680         UniRun(Executor executor, CompletableFuture<Void> dep,

   681                CompletableFuture<T> src, Runnable fn) {

   682             super(executor, dep, src); this.fn = fn;

   683         }

   684         final CompletableFuture<Void> tryFire(int mode) {

   685             CompletableFuture<Void> d; CompletableFuture<T> a;

   686             if ((d = dep) == null ||

   687                 !d.uniRun(a = src, fn, mode > 0 ? null : this))

   688                 return null;

   689             dep = null; src = null; fn = null;

   690             return d.postFire(a, mode);

   691         }

   692     }

   693 

   694     final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) {

   695         Object r; Throwable x;

   696         if (a == null || (r = a.result) == null || f == null)

   697             return false;

   698         if (result == null) {

   699             if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)

   700                 completeThrowable(x, r);

   701             else

   702                 try {

   703                     if (c != null && !c.claim())

   704                         return false;

   705                     f.run();

   706                     completeNull();

   707                 } catch (Throwable ex) {

   708                     completeThrowable(ex);

   709                 }

   710         }

   711         return true;

   712     }

   713 

   714     private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {

   715         if (f == null) throw new NullPointerException();

   716         CompletableFuture<Void> d = new CompletableFuture<Void>();

   717         if (e != null || !d.uniRun(this, f, null)) {

   718             UniRun<T> c = new UniRun<T>(e, d, this, f);

   719             push(c);

   720             c.tryFire(SYNC);

   721         }

   722         return d;

   723     }

   724 

   725     @SuppressWarnings("serial")

   726     static final class UniWhenComplete<T> extends UniCompletion<T,T> {

   727         BiConsumer<? super T, ? super Throwable> fn;

   728         UniWhenComplete(Executor executor, CompletableFuture<T> dep,

   729                         CompletableFuture<T> src,

   730                         BiConsumer<? super T, ? super Throwable> fn) {

   731             super(executor, dep, src); this.fn = fn;

   732         }

   733         final CompletableFuture<T> tryFire(int mode) {

   734             CompletableFuture<T> d; CompletableFuture<T> a;

   735             if ((d = dep) == null ||

   736                 !d.uniWhenComplete(a = src, fn, mode > 0 ? null : this))

   737                 return null;

   738             dep = null; src = null; fn = null;

   739             return d.postFire(a, mode);

   740         }

   741     }

   742 

   743     final boolean uniWhenComplete(CompletableFuture<T> a,

   744                                   BiConsumer<? super T,? super Throwable> f,

   745                                   UniWhenComplete<T> c) {

   746         Object r; T t; Throwable x = null;

   747         if (a == null || (r = a.result) == null || f == null)

   748             return false;

   749         if (result == null) {

   750             try {

   751                 if (c != null && !c.claim())

   752                     return false;

   753                 if (r instanceof AltResult) {

   754                     x = ((AltResult)r).ex;

   755                     t = null;

   756                 } else {

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

   758                     t = tr;

   759                 }

   760                 f.accept(t, x);

   761                 if (x == null) {

   762                     internalComplete(r);

   763                     return true;

   764                 }

   765             } catch (Throwable ex) {

   766                 if (x == null)

   767                     x = ex;

   768             }

   769             completeThrowable(x, r);

   770         }

   771         return true;

   772     }

   773 

   774     private CompletableFuture<T> uniWhenCompleteStage(

   775         Executor e, BiConsumer<? super T, ? super Throwable> f) {

   776         if (f == null) throw new NullPointerException();

   777         CompletableFuture<T> d = new CompletableFuture<T>();

   778         if (e != null || !d.uniWhenComplete(this, f, null)) {

   779             UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);

   780             push(c);

   781             c.tryFire(SYNC);

   782         }

   783         return d;

   784     }

   785 

   786     @SuppressWarnings("serial")

   787     static final class UniHandle<T,V> extends UniCompletion<T,V> {

   788         BiFunction<? super T, Throwable, ? extends V> fn;

   789         UniHandle(Executor executor, CompletableFuture<V> dep,

   790                   CompletableFuture<T> src,

   791                   BiFunction<? super T, Throwable, ? extends V> fn) {

   792             super(executor, dep, src); this.fn = fn;

   793         }

   794         final CompletableFuture<V> tryFire(int mode) {

   795             CompletableFuture<V> d; CompletableFuture<T> a;

   796             if ((d = dep) == null ||

   797                 !d.uniHandle(a = src, fn, mode > 0 ? null : this))

   798                 return null;

   799             dep = null; src = null; fn = null;

   800             return d.postFire(a, mode);

   801         }

   802     }

   803 

   804     final <S> boolean uniHandle(CompletableFuture<S> a,

   805                                 BiFunction<? super S, Throwable, ? extends T> f,

   806                                 UniHandle<S,T> c) {

   807         Object r; S s; Throwable x;

   808         if (a == null || (r = a.result) == null || f == null)

   809             return false;

   810         if (result == null) {

   811             try {

   812                 if (c != null && !c.claim())

   813                     return false;

   814                 if (r instanceof AltResult) {

   815                     x = ((AltResult)r).ex;

   816                     s = null;

   817                 } else {

   818                     x = null;

   819                     @SuppressWarnings("unchecked") S ss = (S) r;

   820                     s = ss;

   821                 }

   822                 completeValue(f.apply(s, x));

   823             } catch (Throwable ex) {

   824                 completeThrowable(ex);

   825             }

   826         }

   827         return true;

   828     }

   829 

   830     private <V> CompletableFuture<V> uniHandleStage(

   831         Executor e, BiFunction<? super T, Throwable, ? extends V> f) {

   832         if (f == null) throw new NullPointerException();

   833         CompletableFuture<V> d = new CompletableFuture<V>();

   834         if (e != null || !d.uniHandle(this, f, null)) {

   835             UniHandle<T,V> c = new UniHandle<T,V>(e, d, this, f);

   836             push(c);

   837             c.tryFire(SYNC);

   838         }

   839         return d;

   840     }

   841 

   842     @SuppressWarnings("serial")

   843     static final class UniExceptionally<T> extends UniCompletion<T,T> {

   844         Function<? super Throwable, ? extends T> fn;

   845         UniExceptionally(CompletableFuture<T> dep, CompletableFuture<T> src,

   846                          Function<? super Throwable, ? extends T> fn) {

   847             super(null, dep, src); this.fn = fn;

   848         }

   849         final CompletableFuture<T> tryFire(int mode) { // never ASYNC

   850             // assert mode != ASYNC;

   851             CompletableFuture<T> d; CompletableFuture<T> a;

   852             if ((d = dep) == null || !d.uniExceptionally(a = src, fn, this))

   853                 return null;

   854             dep = null; src = null; fn = null;

   855             return d.postFire(a, mode);

   856         }

   857     }

   858 

   859     final boolean uniExceptionally(CompletableFuture<T> a,

   860                                    Function<? super Throwable, ? extends T> f,

   861                                    UniExceptionally<T> c) {

   862         Object r; Throwable x;

   863         if (a == null || (r = a.result) == null || f == null)

   864             return false;

   865         if (result == null) {

   866             try {

   867                 if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) {

   868                     if (c != null && !c.claim())

   869                         return false;

   870                     completeValue(f.apply(x));

   871                 } else

   872                     internalComplete(r);

   873             } catch (Throwable ex) {

   874                 completeThrowable(ex);

   875             }

   876         }

   877         return true;

   878     }

   879 

   880     private CompletableFuture<T> uniExceptionallyStage(

   881         Function<Throwable, ? extends T> f) {

   882         if (f == null) throw new NullPointerException();

   883         CompletableFuture<T> d = new CompletableFuture<T>();

   884         if (!d.uniExceptionally(this, f, null)) {

   885             UniExceptionally<T> c = new UniExceptionally<T>(d, this, f);

   886             push(c);

   887             c.tryFire(SYNC);

   888         }

   889         return d;

   890     }

   891 

   892     @SuppressWarnings("serial")

   893     static final class UniRelay<T> extends UniCompletion<T,T> { // for Compose

   894         UniRelay(CompletableFuture<T> dep, CompletableFuture<T> src) {

   895             super(null, dep, src);

   896         }

   897         final CompletableFuture<T> tryFire(int mode) {

   898             CompletableFuture<T> d; CompletableFuture<T> a;

   899             if ((d = dep) == null || !d.uniRelay(a = src))

   900                 return null;

   901             src = null; dep = null;

   902             return d.postFire(a, mode);

   903         }

   904     }

   905 

   906     final boolean uniRelay(CompletableFuture<T> a) {

   907         Object r;

   908         if (a == null || (r = a.result) == null)

   909             return false;

   910         if (result == null) // no need to claim

   911             completeRelay(r);

   912         return true;

   913     }

   914 

   915     @SuppressWarnings("serial")

   916     static final class UniCompose<T,V> extends UniCompletion<T,V> {

   917         Function<? super T, ? extends CompletionStage<V>> fn;

   918         UniCompose(Executor executor, CompletableFuture<V> dep,

   919                    CompletableFuture<T> src,

   920                    Function<? super T, ? extends CompletionStage<V>> fn) {

   921             super(executor, dep, src); this.fn = fn;

   922         }

   923         final CompletableFuture<V> tryFire(int mode) {

   924             CompletableFuture<V> d; CompletableFuture<T> a;

   925             if ((d = dep) == null ||

   926                 !d.uniCompose(a = src, fn, mode > 0 ? null : this))

   927                 return null;

   928             dep = null; src = null; fn = null;

   929             return d.postFire(a, mode);

   930         }

   931     }

   932 

   933     final <S> boolean uniCompose(

   934         CompletableFuture<S> a,

   935         Function<? super S, ? extends CompletionStage<T>> f,

   936         UniCompose<S,T> c) {

   937         Object r; Throwable x;

   938         if (a == null || (r = a.result) == null || f == null)

   939             return false;

   940         tryComplete: if (result == null) {

   941             if (r instanceof AltResult) {

   942                 if ((x = ((AltResult)r).ex) != null) {

   943                     completeThrowable(x, r);

   944                     break tryComplete;

   945                 }

   946                 r = null;

   947             }

   948             try {

   949                 if (c != null && !c.claim())

   950                     return false;

   951                 @SuppressWarnings("unchecked") S s = (S) r;

   952                 CompletableFuture<T> g = f.apply(s).toCompletableFuture();

   953                 if (g.result == null || !uniRelay(g)) {

   954                     UniRelay<T> copy = new UniRelay<T>(this, g);

   955                     g.push(copy);

   956                     copy.tryFire(SYNC);

   957                     if (result == null)

   958                         return false;

   959                 }

   960             } catch (Throwable ex) {

   961                 completeThrowable(ex);

   962             }

   963         }

   964         return true;

   965     }

   966 

   967     private <V> CompletableFuture<V> uniComposeStage(

   968         Executor e, Function<? super T, ? extends CompletionStage<V>> f) {

   969         if (f == null) throw new NullPointerException();

   970         Object r; Throwable x;

   971         if (e == null && (r = result) != null) {

   972             // try to return function result directly

   973             if (r instanceof AltResult) {

   974                 if ((x = ((AltResult)r).ex) != null) {

   975                     return new CompletableFuture<V>(encodeThrowable(x, r));

   976                 }

   977                 r = null;

   978             }

   979             try {

   980                 @SuppressWarnings("unchecked") T t = (T) r;

   981                 return f.apply(t).toCompletableFuture();

   982             } catch (Throwable ex) {

   983                 return new CompletableFuture<V>(encodeThrowable(ex));

   984             }

   985         }

   986         CompletableFuture<V> d = new CompletableFuture<V>();

   987         UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);

   988         push(c);

   989         c.tryFire(SYNC);

   990         return d;

   991     }

   992 

   993     /* ------------- Two-input Completions -------------- */

   994 

   995     /** A Completion for an action with two sources */

   996     @SuppressWarnings("serial")

   997     abstract static class BiCompletion<T,U,V> extends UniCompletion<T,V> {

   998         CompletableFuture<U> snd; // second source for action

   999         BiCompletion(Executor executor, CompletableFuture<V> dep,

  1000                      CompletableFuture<T> src, CompletableFuture<U> snd) {

  1001             super(executor, dep, src); this.snd = snd;

  1002         }

  1003     }

  1004 

  1005     /** A Completion delegating to a BiCompletion */

  1006     @SuppressWarnings("serial")

  1007     static final class CoCompletion extends Completion {

  1008         BiCompletion<?,?,?> base;

  1009         CoCompletion(BiCompletion<?,?,?> base) { this.base = base; }

  1010         final CompletableFuture<?> tryFire(int mode) {

  1011             BiCompletion<?,?,?> c; CompletableFuture<?> d;

  1012             if ((c = base) == null || (d = c.tryFire(mode)) == null)

  1013                 return null;

  1014             base = null; // detach

  1015             return d;

  1016         }

  1017         final boolean isLive() {

  1018             BiCompletion<?,?,?> c;

  1019             return (c = base) != null && c.dep != null;

  1020         }

  1021     }

  1022 

  1023     /** Pushes completion to this and b unless both done. */

  1024     final void bipush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {

  1025         if (c != null) {

  1026             Object r;

  1027             while ((r = result) == null && !tryPushStack(c))

  1028                 lazySetNext(c, null); // clear on failure

  1029             if (b != null && b != this && b.result == null) {

  1030                 Completion q = (r != null) ? c : new CoCompletion(c);

  1031                 while (b.result == null && !b.tryPushStack(q))

  1032                     lazySetNext(q, null); // clear on failure

  1033             }

  1034         }

  1035     }

  1036 

  1037     /** Post-processing after successful BiCompletion tryFire. */

  1038     final CompletableFuture<T> postFire(CompletableFuture<?> a,

  1039                                         CompletableFuture<?> b, int mode) {

  1040         if (b != null && b.stack != null) { // clean second source

  1041             if (mode < 0 || b.result == null)

  1042                 b.cleanStack();

  1043             else

  1044                 b.postComplete();

  1045         }

  1046         return postFire(a, mode);

  1047     }

  1048 

  1049     @SuppressWarnings("serial")

  1050     static final class BiApply<T,U,V> extends BiCompletion<T,U,V> {

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

  1052         BiApply(Executor executor, CompletableFuture<V> dep,

  1053                 CompletableFuture<T> src, CompletableFuture<U> snd,

  1054                 BiFunction<? super T,? super U,? extends V> fn) {

  1055             super(executor, dep, src, snd); this.fn = fn;

  1056         }

  1057         final CompletableFuture<V> tryFire(int mode) {

  1058             CompletableFuture<V> d;

  1059             CompletableFuture<T> a;

  1060             CompletableFuture<U> b;

  1061             if ((d = dep) == null ||

  1062                 !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this))

  1063                 return null;

  1064             dep = null; src = null; snd = null; fn = null;

  1065             return d.postFire(a, b, mode);

  1066         }

  1067     }

  1068 

  1069     final <R,S> boolean biApply(CompletableFuture<R> a,

  1070                                 CompletableFuture<S> b,

  1071                                 BiFunction<? super R,? super S,? extends T> f,

  1072                                 BiApply<R,S,T> c) {

  1073         Object r, s; Throwable x;

  1074         if (a == null || (r = a.result) == null ||

  1075             b == null || (s = b.result) == null || f == null)

  1076             return false;

  1077         tryComplete: if (result == null) {

  1078             if (r instanceof AltResult) {

  1079                 if ((x = ((AltResult)r).ex) != null) {

  1080                     completeThrowable(x, r);

  1081                     break tryComplete;

  1082                 }

  1083                 r = null;

  1084             }

  1085             if (s instanceof AltResult) {

  1086                 if ((x = ((AltResult)s).ex) != null) {

  1087                     completeThrowable(x, s);

  1088                     break tryComplete;

  1089                 }

  1090                 s = null;

  1091             }

  1092             try {

  1093                 if (c != null && !c.claim())

  1094                     return false;

  1095                 @SuppressWarnings("unchecked") R rr = (R) r;

  1096                 @SuppressWarnings("unchecked") S ss = (S) s;

  1097                 completeValue(f.apply(rr, ss));

  1098             } catch (Throwable ex) {

  1099                 completeThrowable(ex);

  1100             }

  1101         }

  1102         return true;

  1103     }

  1104 

  1105     private <U,V> CompletableFuture<V> biApplyStage(

  1106         Executor e, CompletionStage<U> o,

  1107         BiFunction<? super T,? super U,? extends V> f) {

  1108         CompletableFuture<U> b;

  1109         if (f == null || (b = o.toCompletableFuture()) == null)

  1110             throw new NullPointerException();

  1111         CompletableFuture<V> d = new CompletableFuture<V>();

  1112         if (e != null || !d.biApply(this, b, f, null)) {

  1113             BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);

  1114             bipush(b, c);

  1115             c.tryFire(SYNC);

  1116         }

  1117         return d;

  1118     }

  1119 

  1120     @SuppressWarnings("serial")

  1121     static final class BiAccept<T,U> extends BiCompletion<T,U,Void> {

  1122         BiConsumer<? super T,? super U> fn;

  1123         BiAccept(Executor executor, CompletableFuture<Void> dep,

  1124                  CompletableFuture<T> src, CompletableFuture<U> snd,

  1125                  BiConsumer<? super T,? super U> fn) {

  1126             super(executor, dep, src, snd); this.fn = fn;

  1127         }

  1128         final CompletableFuture<Void> tryFire(int mode) {

  1129             CompletableFuture<Void> d;

  1130             CompletableFuture<T> a;

  1131             CompletableFuture<U> b;

  1132             if ((d = dep) == null ||

  1133                 !d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this))

  1134                 return null;

  1135             dep = null; src = null; snd = null; fn = null;

  1136             return d.postFire(a, b, mode);

  1137         }

  1138     }

  1139 

  1140     final <R,S> boolean biAccept(CompletableFuture<R> a,

  1141                                  CompletableFuture<S> b,

  1142                                  BiConsumer<? super R,? super S> f,

  1143                                  BiAccept<R,S> c) {

  1144         Object r, s; Throwable x;

  1145         if (a == null || (r = a.result) == null ||

  1146             b == null || (s = b.result) == null || f == null)

  1147             return false;

  1148         tryComplete: if (result == null) {

  1149             if (r instanceof AltResult) {

  1150                 if ((x = ((AltResult)r).ex) != null) {

  1151                     completeThrowable(x, r);

  1152                     break tryComplete;

  1153                 }

  1154                 r = null;

  1155             }

  1156             if (s instanceof AltResult) {

  1157                 if ((x = ((AltResult)s).ex) != null) {

  1158                     completeThrowable(x, s);

  1159                     break tryComplete;

  1160                 }

  1161                 s = null;

  1162             }

  1163             try {

  1164                 if (c != null && !c.claim())

  1165                     return false;

  1166                 @SuppressWarnings("unchecked") R rr = (R) r;

  1167                 @SuppressWarnings("unchecked") S ss = (S) s;

  1168                 f.accept(rr, ss);

  1169                 completeNull();

  1170             } catch (Throwable ex) {

  1171                 completeThrowable(ex);

  1172             }

  1173         }

  1174         return true;

  1175     }

  1176 

  1177     private <U> CompletableFuture<Void> biAcceptStage(

  1178         Executor e, CompletionStage<U> o,

  1179         BiConsumer<? super T,? super U> f) {

  1180         CompletableFuture<U> b;

  1181         if (f == null || (b = o.toCompletableFuture()) == null)

  1182             throw new NullPointerException();

  1183         CompletableFuture<Void> d = new CompletableFuture<Void>();

  1184         if (e != null || !d.biAccept(this, b, f, null)) {

  1185             BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);

  1186             bipush(b, c);

  1187             c.tryFire(SYNC);

  1188         }

  1189         return d;

  1190     }

  1191 

  1192     @SuppressWarnings("serial")

  1193     static final class BiRun<T,U> extends BiCompletion<T,U,Void> {

  1194         Runnable fn;

  1195         BiRun(Executor executor, CompletableFuture<Void> dep,

  1196               CompletableFuture<T> src,

  1197               CompletableFuture<U> snd,

  1198               Runnable fn) {

  1199             super(executor, dep, src, snd); this.fn = fn;

  1200         }

  1201         final CompletableFuture<Void> tryFire(int mode) {

  1202             CompletableFuture<Void> d;

  1203             CompletableFuture<T> a;

  1204             CompletableFuture<U> b;

  1205             if ((d = dep) == null ||

  1206                 !d.biRun(a = src, b = snd, fn, mode > 0 ? null : this))

  1207                 return null;

  1208             dep = null; src = null; snd = null; fn = null;

  1209             return d.postFire(a, b, mode);

  1210         }

  1211     }

  1212 

  1213     final boolean biRun(CompletableFuture<?> a, CompletableFuture<?> b,

  1214                         Runnable f, BiRun<?,?> c) {

  1215         Object r, s; Throwable x;

  1216         if (a == null || (r = a.result) == null ||

  1217             b == null || (s = b.result) == null || f == null)

  1218             return false;

  1219         if (result == null) {