ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jdk8/java/util/concurrent/CompletableFuture.java
Revision: 1.2
Committed: Sat Apr 2 17:45:34 2016 UTC (8 years, 1 month ago) by dl
Branch: MAIN
Changes since 1.1: +140 -39 lines
Log Message:
reduce async overhead

File Contents

# User Rev Content
1 jsr166 1.1 /*
2     * Written by Doug Lea with assistance from members of JCP JSR-166
3     * Expert Group and released to the public domain, as explained at
4     * http://creativecommons.org/publicdomain/zero/1.0/
5     */
6    
7     package java.util.concurrent;
8    
9     import java.util.concurrent.locks.LockSupport;
10     import java.util.function.BiConsumer;
11     import java.util.function.BiFunction;
12     import java.util.function.Consumer;
13     import java.util.function.Function;
14     import java.util.function.Supplier;
15    
16     /**
17     * A {@link Future} that may be explicitly completed (setting its
18     * value and status), and may be used as a {@link CompletionStage},
19     * supporting dependent functions and actions that trigger upon its
20     * completion.
21     *
22     * <p>When two or more threads attempt to
23     * {@link #complete complete},
24     * {@link #completeExceptionally completeExceptionally}, or
25     * {@link #cancel cancel}
26     * a CompletableFuture, only one of them succeeds.
27     *
28     * <p>In addition to these and related methods for directly
29     * manipulating status and results, CompletableFuture implements
30     * interface {@link CompletionStage} with the following policies: <ul>
31     *
32     * <li>Actions supplied for dependent completions of
33     * <em>non-async</em> methods may be performed by the thread that
34     * completes the current CompletableFuture, or by any other caller of
35     * a completion method.
36     *
37     * <li>All <em>async</em> methods without an explicit Executor
38     * argument are performed using the {@link ForkJoinPool#commonPool()}
39     * (unless it does not support a parallelism level of at least two, in
40     * which case, a new Thread is created to run each task). This may be
41     * overridden for non-static methods in subclasses by defining method
42     * {@link #defaultExecutor()}. To simplify monitoring, debugging,
43     * and tracking, all generated asynchronous tasks are instances of the
44     * marker interface {@link AsynchronousCompletionTask}. Operations
45     * with time-delays can use adapter methods defined in this class, for
46     * example: {@code supplyAsync(supplier, delayedExecutor(timeout,
47     * timeUnit))}. To support methods with delays and timeouts, this
48     * class maintains at most one daemon thread for triggering and
49     * cancelling actions, not for running them.
50     *
51     * <li>All CompletionStage methods are implemented independently of
52     * other public methods, so the behavior of one method is not impacted
53     * by overrides of others in subclasses.
54     *
55     * <li>All CompletionStage methods return CompletableFutures. To
56     * restrict usages to only those methods defined in interface
57     * CompletionStage, use method {@link #minimalCompletionStage}. Or to
58     * ensure only that clients do not themselves modify a future, use
59     * method {@link #copy}.
60     * </ul>
61     *
62     * <p>CompletableFuture also implements {@link Future} with the following
63     * policies: <ul>
64     *
65     * <li>Since (unlike {@link FutureTask}) this class has no direct
66     * control over the computation that causes it to be completed,
67     * cancellation is treated as just another form of exceptional
68     * completion. Method {@link #cancel cancel} has the same effect as
69     * {@code completeExceptionally(new CancellationException())}. Method
70     * {@link #isCompletedExceptionally} can be used to determine if a
71     * CompletableFuture completed in any exceptional fashion.
72     *
73     * <li>In case of exceptional completion with a CompletionException,
74     * methods {@link #get()} and {@link #get(long, TimeUnit)} throw an
75     * {@link ExecutionException} with the same cause as held in the
76     * corresponding CompletionException. To simplify usage in most
77     * contexts, this class also defines methods {@link #join()} and
78     * {@link #getNow} that instead throw the CompletionException directly
79     * in these cases.
80     * </ul>
81     *
82     * <p>Arguments used to pass a completion result (that is, for
83     * parameters of type {@code T}) for methods accepting them may be
84     * null, but passing a null value for any other parameter will result
85     * in a {@link NullPointerException} being thrown.
86     *
87     * <p>Subclasses of this class should normally override the "virtual
88     * constructor" method {@link #newIncompleteFuture}, which establishes
89     * the concrete type returned by CompletionStage methods. For example,
90     * here is a class that substitutes a different default Executor and
91     * disables the {@code obtrude} methods:
92     *
93     * <pre> {@code
94     * class MyCompletableFuture<T> extends CompletableFuture<T> {
95     * static final Executor myExecutor = ...;
96     * public MyCompletableFuture() { }
97     * public <U> CompletableFuture<U> newIncompleteFuture() {
98     * return new MyCompletableFuture<U>(); }
99     * public Executor defaultExecutor() {
100     * return myExecutor; }
101     * public void obtrudeValue(T value) {
102     * throw new UnsupportedOperationException(); }
103     * public void obtrudeException(Throwable ex) {
104     * throw new UnsupportedOperationException(); }
105     * }}</pre>
106     *
107     * @author Doug Lea
108     * @since 1.8
109     * @param <T> The result type returned by this future's {@code join}
110     * and {@code get} methods
111     */
112     public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
113    
114     /*
115     * Overview:
116     *
117     * A CompletableFuture may have dependent completion actions,
118     * collected in a linked stack. It atomically completes by CASing
119     * a result field, and then pops off and runs those actions. This
120     * applies across normal vs exceptional outcomes, sync vs async
121     * actions, binary triggers, and various forms of completions.
122     *
123     * Non-nullness of field result (set via CAS) indicates done. An
124     * AltResult is used to box null as a result, as well as to hold
125     * exceptions. Using a single field makes completion simple to
126     * detect and trigger. Encoding and decoding is straightforward
127     * but adds to the sprawl of trapping and associating exceptions
128     * with targets. Minor simplifications rely on (static) NIL (to
129     * box null results) being the only AltResult with a null
130     * exception field, so we don't usually need explicit comparisons.
131     * Even though some of the generics casts are unchecked (see
132     * SuppressWarnings annotations), they are placed to be
133     * appropriate even if checked.
134     *
135     * Dependent actions are represented by Completion objects linked
136     * as Treiber stacks headed by field "stack". There are Completion
137     * classes for each kind of action, grouped into single-input
138     * (UniCompletion), two-input (BiCompletion), projected
139     * (BiCompletions using either (not both) of two inputs), shared
140     * (CoCompletion, used by the second of two sources), zero-input
141     * source actions, and Signallers that unblock waiters. Class
142     * Completion extends ForkJoinTask to enable async execution
143     * (adding no space overhead because we exploit its "tag" methods
144     * to maintain claims). It is also declared as Runnable to allow
145     * usage with arbitrary executors.
146     *
147     * Support for each kind of CompletionStage relies on a separate
148     * class, along with two CompletableFuture methods:
149     *
150     * * A Completion class with name X corresponding to function,
151     * prefaced with "Uni", "Bi", or "Or". Each class contains
152     * fields for source(s), actions, and dependent. They are
153     * boringly similar, differing from others only with respect to
154     * underlying functional forms. We do this so that users don't
155     * encounter layers of adapters in common usages.
156     *
157     * * Boolean CompletableFuture method x(...) (for example
158     * uniApply) takes all of the arguments needed to check that an
159     * action is triggerable, and then either runs the action or
160     * arranges its async execution by executing its Completion
161     * argument, if present. The method returns true if known to be
162     * complete.
163     *
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. The screening checks needed anyway harmlessly ignore
209     * null arguments that may have been obtained during races with
210     * threads nulling out fields. We also try to unlink fired
211     * Completions from stacks that might never be popped (see method
212     * postFire). Completion fields need not be declared as final or
213     * volatile because they are only visible to other threads upon
214     * safe publication.
215     */
216    
217     volatile Object result; // Either the result or boxed AltResult
218     volatile Completion stack; // Top of Treiber stack of dependent actions
219    
220     final boolean internalComplete(Object r) { // CAS from null to r
221     return U.compareAndSwapObject(this, RESULT, null, r);
222     }
223    
224     final boolean casStack(Completion cmp, Completion val) {
225     return U.compareAndSwapObject(this, STACK, cmp, val);
226     }
227    
228     /** Returns true if successfully pushed c onto stack. */
229     final boolean tryPushStack(Completion c) {
230     Completion h = stack;
231     lazySetNext(c, h);
232     return U.compareAndSwapObject(this, STACK, h, c);
233     }
234    
235     /** Unconditionally pushes c onto stack, retrying if necessary. */
236     final void pushStack(Completion c) {
237     do {} while (!tryPushStack(c));
238     }
239    
240     /* ------------- Encoding and decoding outcomes -------------- */
241    
242     static final class AltResult { // See above
243     final Throwable ex; // null only for NIL
244     AltResult(Throwable x) { this.ex = x; }
245     }
246    
247     /** The encoding of the null value. */
248     static final AltResult NIL = new AltResult(null);
249    
250     /** Completes with the null value, unless already completed. */
251     final boolean completeNull() {
252     return U.compareAndSwapObject(this, RESULT, null,
253     NIL);
254     }
255    
256     /** Returns the encoding of the given non-exceptional value. */
257     final Object encodeValue(T t) {
258     return (t == null) ? NIL : t;
259     }
260    
261     /** Completes with a non-exceptional result, unless already completed. */
262     final boolean completeValue(T t) {
263     return U.compareAndSwapObject(this, RESULT, null,
264     (t == null) ? NIL : t);
265     }
266    
267     /**
268     * Returns the encoding of the given (non-null) exception as a
269     * wrapped CompletionException unless it is one already.
270     */
271     static AltResult encodeThrowable(Throwable x) {
272     return new AltResult((x instanceof CompletionException) ? x :
273     new CompletionException(x));
274     }
275    
276     /** Completes with an exceptional result, unless already completed. */
277     final boolean completeThrowable(Throwable x) {
278     return U.compareAndSwapObject(this, RESULT, null,
279     encodeThrowable(x));
280     }
281    
282     /**
283     * Returns the encoding of the given (non-null) exception as a
284     * wrapped CompletionException unless it is one already. May
285     * return the given Object r (which must have been the result of a
286     * source future) if it is equivalent, i.e. if this is a simple
287     * relay of an existing CompletionException.
288     */
289     static Object encodeThrowable(Throwable x, Object r) {
290     if (!(x instanceof CompletionException))
291     x = new CompletionException(x);
292     else if (r instanceof AltResult && x == ((AltResult)r).ex)
293     return r;
294     return new AltResult(x);
295     }
296    
297     /**
298     * Completes with the given (non-null) exceptional result as a
299     * wrapped CompletionException unless it is one already, unless
300     * already completed. May complete with the given Object r
301     * (which must have been the result of a source future) if it is
302     * equivalent, i.e. if this is a simple propagation of an
303     * existing CompletionException.
304     */
305     final boolean completeThrowable(Throwable x, Object r) {
306     return U.compareAndSwapObject(this, RESULT, null,
307     encodeThrowable(x, r));
308     }
309    
310     /**
311     * Returns the encoding of the given arguments: if the exception
312     * is non-null, encodes as AltResult. Otherwise uses the given
313     * value, boxed as NIL if null.
314     */
315     Object encodeOutcome(T t, Throwable x) {
316     return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x);
317     }
318    
319     /**
320     * Returns the encoding of a copied outcome; if exceptional,
321     * rewraps as a CompletionException, else returns argument.
322     */
323     static Object encodeRelay(Object r) {
324     Throwable x;
325     return (((r instanceof AltResult) &&
326     (x = ((AltResult)r).ex) != null &&
327     !(x instanceof CompletionException)) ?
328     new AltResult(new CompletionException(x)) : r);
329     }
330    
331     /**
332     * Completes with r or a copy of r, unless already completed.
333     * If exceptional, r is first coerced to a CompletionException.
334     */
335     final boolean completeRelay(Object r) {
336     return U.compareAndSwapObject(this, RESULT, null,
337     encodeRelay(r));
338     }
339    
340     /**
341     * Reports result using Future.get conventions.
342     */
343     private static <T> T reportGet(Object r)
344     throws InterruptedException, ExecutionException {
345     if (r == null) // by convention below, null means interrupted
346     throw new InterruptedException();
347     if (r instanceof AltResult) {
348     Throwable x, cause;
349     if ((x = ((AltResult)r).ex) == null)
350     return null;
351     if (x instanceof CancellationException)
352     throw (CancellationException)x;
353     if ((x instanceof CompletionException) &&
354     (cause = x.getCause()) != null)
355     x = cause;
356     throw new ExecutionException(x);
357     }
358     @SuppressWarnings("unchecked") T t = (T) r;
359     return t;
360     }
361    
362     /**
363     * Decodes outcome to return result or throw unchecked exception.
364     */
365     private static <T> T reportJoin(Object r) {
366     if (r instanceof AltResult) {
367     Throwable x;
368     if ((x = ((AltResult)r).ex) == null)
369     return null;
370     if (x instanceof CancellationException)
371     throw (CancellationException)x;
372     if (x instanceof CompletionException)
373     throw (CompletionException)x;
374     throw new CompletionException(x);
375     }
376     @SuppressWarnings("unchecked") T t = (T) r;
377     return t;
378     }
379    
380     /* ------------- Async task preliminaries -------------- */
381    
382     /**
383     * A marker interface identifying asynchronous tasks produced by
384     * {@code async} methods. This may be useful for monitoring,
385     * debugging, and tracking asynchronous activities.
386     *
387     * @since 1.8
388     */
389     public static interface AsynchronousCompletionTask {
390     }
391    
392     private static final boolean USE_COMMON_POOL =
393     (ForkJoinPool.getCommonPoolParallelism() > 1);
394    
395     /**
396     * Default executor -- ForkJoinPool.commonPool() unless it cannot
397     * support parallelism.
398     */
399     private static final Executor ASYNC_POOL = USE_COMMON_POOL ?
400     ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
401    
402     /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */
403     static final class ThreadPerTaskExecutor implements Executor {
404     public void execute(Runnable r) { new Thread(r).start(); }
405     }
406    
407     /**
408     * Null-checks user executor argument, and translates uses of
409     * commonPool to ASYNC_POOL in case parallelism disabled.
410     */
411     static Executor screenExecutor(Executor e) {
412     if (!USE_COMMON_POOL && e == ForkJoinPool.commonPool())
413     return ASYNC_POOL;
414     if (e == null) throw new NullPointerException();
415     return e;
416     }
417    
418     // Modes for Completion.tryFire. Signedness matters.
419     static final int SYNC = 0;
420     static final int ASYNC = 1;
421     static final int NESTED = -1;
422    
423     /* ------------- Base Completion classes and operations -------------- */
424    
425     @SuppressWarnings("serial")
426     abstract static class Completion extends ForkJoinTask<Void>
427     implements Runnable, AsynchronousCompletionTask {
428     volatile Completion next; // Treiber stack link
429    
430     /**
431     * Performs completion action if triggered, returning a
432     * dependent that may need propagation, if one exists.
433     *
434     * @param mode SYNC, ASYNC, or NESTED
435     */
436     abstract CompletableFuture<?> tryFire(int mode);
437    
438     /** Returns true if possibly still triggerable. Used by cleanStack. */
439     abstract boolean isLive();
440    
441     public final void run() { tryFire(ASYNC); }
442     public final boolean exec() { tryFire(ASYNC); return false; }
443     public final Void getRawResult() { return null; }
444     public final void setRawResult(Void v) {}
445     }
446    
447     static void lazySetNext(Completion c, Completion next) {
448     U.putOrderedObject(c, NEXT, next);
449     }
450    
451     /**
452     * Pops and tries to trigger all reachable dependents. Call only
453     * when known to be done.
454     */
455     final void postComplete() {
456     /*
457     * On each step, variable f holds current dependents to pop
458     * and run. It is extended along only one path at a time,
459     * pushing others to avoid unbounded recursion.
460     */
461     CompletableFuture<?> f = this; Completion h;
462     while ((h = f.stack) != null ||
463     (f != this && (h = (f = this).stack) != null)) {
464     CompletableFuture<?> d; Completion t;
465     if (f.casStack(h, t = h.next)) {
466     if (t != null) {
467     if (f != this) {
468     pushStack(h);
469     continue;
470     }
471     h.next = null; // detach
472     }
473     f = (d = h.tryFire(NESTED)) == null ? this : d;
474     }
475     }
476     }
477    
478     /** Traverses stack and unlinks dead Completions. */
479     final void cleanStack() {
480     for (Completion p = null, q = stack; q != null;) {
481     Completion s = q.next;
482     if (q.isLive()) {
483     p = q;
484     q = s;
485     }
486     else if (p == null) {
487     casStack(q, s);
488     q = stack;
489     }
490     else {
491     p.next = s;
492     if (p.isLive())
493     q = s;
494     else {
495     p = null; // restart
496     q = stack;
497     }
498     }
499     }
500     }
501    
502     /* ------------- One-input Completions -------------- */
503    
504     /** A Completion with a source, dependent, and executor. */
505     @SuppressWarnings("serial")
506     abstract static class UniCompletion<T,V> extends Completion {
507     Executor executor; // executor to use (null if none)
508     CompletableFuture<V> dep; // the dependent to complete
509     CompletableFuture<T> src; // source for action
510    
511     UniCompletion(Executor executor, CompletableFuture<V> dep,
512     CompletableFuture<T> src) {
513     this.executor = executor; this.dep = dep; this.src = src;
514     }
515    
516     /**
517     * Returns true if action can be run. Call only when known to
518     * be triggerable. Uses FJ tag bit to ensure that only one
519     * thread claims ownership. If async, starts as task -- a
520     * later call to tryFire will run action.
521     */
522     final boolean claim() {
523     Executor e = executor;
524     if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {
525     if (e == null)
526     return true;
527     executor = null; // disable
528     e.execute(this);
529     }
530     return false;
531     }
532    
533     final boolean isLive() { return dep != null; }
534     }
535    
536     /** Pushes the given completion (if it exists) unless done. */
537     final void push(UniCompletion<?,?> c) {
538     if (c != null) {
539     while (result == null && !tryPushStack(c))
540     lazySetNext(c, null); // clear on failure
541     }
542     }
543    
544     /**
545     * Post-processing by dependent after successful UniCompletion
546     * tryFire. Tries to clean stack of source a, and then either runs
547     * postComplete or returns this to caller, depending on mode.
548     */
549     final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
550     if (a != null && a.stack != null) {
551     if (a.result == null)
552     a.cleanStack();
553     else if (mode >= 0)
554     a.postComplete();
555     }
556     if (result != null && stack != null) {
557     if (mode < 0)
558     return this;
559     else
560     postComplete();
561     }
562     return null;
563     }
564    
565     @SuppressWarnings("serial")
566     static final class UniApply<T,V> extends UniCompletion<T,V> {
567     Function<? super T,? extends V> fn;
568     UniApply(Executor executor, CompletableFuture<V> dep,
569     CompletableFuture<T> src,
570     Function<? super T,? extends V> fn) {
571     super(executor, dep, src); this.fn = fn;
572     }
573     final CompletableFuture<V> tryFire(int mode) {
574     CompletableFuture<V> d; CompletableFuture<T> a;
575     if ((d = dep) == null ||
576     !d.uniApply(a = src, fn, mode > 0 ? null : this))
577     return null;
578     dep = null; src = null; fn = null;
579     return d.postFire(a, mode);
580     }
581     }
582    
583     final <S> boolean uniApply(CompletableFuture<S> a,
584     Function<? super S,? extends T> f,
585     UniApply<S,T> c) {
586     Object r; Throwable x;
587     if (a == null || (r = a.result) == null || f == null)
588     return false;
589     tryComplete: if (result == null) {
590     if (r instanceof AltResult) {
591     if ((x = ((AltResult)r).ex) != null) {
592     completeThrowable(x, r);
593     break tryComplete;
594     }
595     r = null;
596     }
597     try {
598     if (c != null && !c.claim())
599     return false;
600     @SuppressWarnings("unchecked") S s = (S) r;
601     completeValue(f.apply(s));
602     } catch (Throwable ex) {
603     completeThrowable(ex);
604     }
605     }
606     return true;
607     }
608    
609     private <V> CompletableFuture<V> uniApplyStage(
610     Executor e, Function<? super T,? extends V> f) {
611     if (f == null) throw new NullPointerException();
612     CompletableFuture<V> d = newIncompleteFuture();
613     if (e != null || !d.uniApply(this, f, null)) {
614     UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
615 dl 1.2 if (e != null && result != null) {
616     try {
617     e.execute(c);
618     } catch (Throwable ex) {
619     d.completeThrowable(ex);
620     }
621     }
622     else {
623     push(c);
624     c.tryFire(SYNC);
625     }
626 jsr166 1.1 }
627     return d;
628     }
629    
630     @SuppressWarnings("serial")
631     static final class UniAccept<T> extends UniCompletion<T,Void> {
632     Consumer<? super T> fn;
633     UniAccept(Executor executor, CompletableFuture<Void> dep,
634     CompletableFuture<T> src, Consumer<? super T> fn) {
635     super(executor, dep, src); this.fn = fn;
636     }
637     final CompletableFuture<Void> tryFire(int mode) {
638     CompletableFuture<Void> d; CompletableFuture<T> a;
639     if ((d = dep) == null ||
640     !d.uniAccept(a = src, fn, mode > 0 ? null : this))
641     return null;
642     dep = null; src = null; fn = null;
643     return d.postFire(a, mode);
644     }
645     }
646    
647     final <S> boolean uniAccept(CompletableFuture<S> a,
648     Consumer<? super S> f, UniAccept<S> c) {
649     Object r; Throwable x;
650     if (a == null || (r = a.result) == null || f == null)
651     return false;
652     tryComplete: if (result == null) {
653     if (r instanceof AltResult) {
654     if ((x = ((AltResult)r).ex) != null) {
655     completeThrowable(x, r);
656     break tryComplete;
657     }
658     r = null;
659     }
660     try {
661     if (c != null && !c.claim())
662     return false;
663     @SuppressWarnings("unchecked") S s = (S) r;
664     f.accept(s);
665     completeNull();
666     } catch (Throwable ex) {
667     completeThrowable(ex);
668     }
669     }
670     return true;
671     }
672    
673     private CompletableFuture<Void> uniAcceptStage(Executor e,
674     Consumer<? super T> f) {
675     if (f == null) throw new NullPointerException();
676     CompletableFuture<Void> d = newIncompleteFuture();
677     if (e != null || !d.uniAccept(this, f, null)) {
678     UniAccept<T> c = new UniAccept<T>(e, d, this, f);
679 dl 1.2 if (e != null && result != null) {
680     try {
681     e.execute(c);
682     } catch (Throwable ex) {
683     d.completeThrowable(ex);
684     }
685     }
686     else {
687     push(c);
688     c.tryFire(SYNC);
689     }
690 jsr166 1.1 }
691     return d;
692     }
693    
694     @SuppressWarnings("serial")
695     static final class UniRun<T> extends UniCompletion<T,Void> {
696     Runnable fn;
697     UniRun(Executor executor, CompletableFuture<Void> dep,
698     CompletableFuture<T> src, Runnable fn) {
699     super(executor, dep, src); this.fn = fn;
700     }
701     final CompletableFuture<Void> tryFire(int mode) {
702     CompletableFuture<Void> d; CompletableFuture<T> a;
703     if ((d = dep) == null ||
704     !d.uniRun(a = src, fn, mode > 0 ? null : this))
705     return null;
706     dep = null; src = null; fn = null;
707     return d.postFire(a, mode);
708     }
709     }
710    
711     final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) {
712     Object r; Throwable x;
713     if (a == null || (r = a.result) == null || f == null)
714     return false;
715     if (result == null) {
716     if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
717     completeThrowable(x, r);
718     else
719     try {
720     if (c != null && !c.claim())
721     return false;
722     f.run();
723     completeNull();
724     } catch (Throwable ex) {
725     completeThrowable(ex);
726     }
727     }
728     return true;
729     }
730    
731     private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {
732     if (f == null) throw new NullPointerException();
733     CompletableFuture<Void> d = newIncompleteFuture();
734     if (e != null || !d.uniRun(this, f, null)) {
735     UniRun<T> c = new UniRun<T>(e, d, this, f);
736 dl 1.2 if (e != null && result != null) {
737     try {
738     e.execute(c);
739     } catch (Throwable ex) {
740     d.completeThrowable(ex);
741     }
742     }
743     else {
744     push(c);
745     c.tryFire(SYNC);
746     }
747 jsr166 1.1 }
748     return d;
749     }
750    
751     @SuppressWarnings("serial")
752     static final class UniWhenComplete<T> extends UniCompletion<T,T> {
753     BiConsumer<? super T, ? super Throwable> fn;
754     UniWhenComplete(Executor executor, CompletableFuture<T> dep,
755     CompletableFuture<T> src,
756     BiConsumer<? super T, ? super Throwable> fn) {
757     super(executor, dep, src); this.fn = fn;
758     }
759     final CompletableFuture<T> tryFire(int mode) {
760     CompletableFuture<T> d; CompletableFuture<T> a;
761     if ((d = dep) == null ||
762     !d.uniWhenComplete(a = src, fn, mode > 0 ? null : this))
763     return null;
764     dep = null; src = null; fn = null;
765     return d.postFire(a, mode);
766     }
767     }
768    
769     final boolean uniWhenComplete(CompletableFuture<T> a,
770     BiConsumer<? super T,? super Throwable> f,
771     UniWhenComplete<T> c) {
772     Object r; T t; Throwable x = null;
773     if (a == null || (r = a.result) == null || f == null)
774     return false;
775     if (result == null) {
776     try {
777     if (c != null && !c.claim())
778     return false;
779     if (r instanceof AltResult) {
780     x = ((AltResult)r).ex;
781     t = null;
782     } else {
783     @SuppressWarnings("unchecked") T tr = (T) r;
784     t = tr;
785     }
786     f.accept(t, x);
787     if (x == null) {
788     internalComplete(r);
789     return true;
790     }
791     } catch (Throwable ex) {
792     if (x == null)
793     x = ex;
794     else if (x != ex)
795     x.addSuppressed(ex);
796     }
797     completeThrowable(x, r);
798     }
799     return true;
800     }
801    
802     private CompletableFuture<T> uniWhenCompleteStage(
803     Executor e, BiConsumer<? super T, ? super Throwable> f) {
804     if (f == null) throw new NullPointerException();
805     CompletableFuture<T> d = newIncompleteFuture();
806     if (e != null || !d.uniWhenComplete(this, f, null)) {
807     UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);
808 dl 1.2 if (e != null && result != null) {
809     try {
810     e.execute(c);
811     } catch (Throwable ex) {
812     d.completeThrowable(ex);
813     }
814     }
815     else {
816     push(c);
817     c.tryFire(SYNC);
818     }
819 jsr166 1.1 }
820     return d;
821     }
822    
823     @SuppressWarnings("serial")
824     static final class UniHandle<T,V> extends UniCompletion<T,V> {
825     BiFunction<? super T, Throwable, ? extends V> fn;
826     UniHandle(Executor executor, CompletableFuture<V> dep,
827     CompletableFuture<T> src,
828     BiFunction<? super T, Throwable, ? extends V> fn) {
829     super(executor, dep, src); this.fn = fn;
830     }
831     final CompletableFuture<V> tryFire(int mode) {
832     CompletableFuture<V> d; CompletableFuture<T> a;
833     if ((d = dep) == null ||
834     !d.uniHandle(a = src, fn, mode > 0 ? null : this))
835     return null;
836     dep = null; src = null; fn = null;
837     return d.postFire(a, mode);
838     }
839     }
840    
841     final <S> boolean uniHandle(CompletableFuture<S> a,
842     BiFunction<? super S, Throwable, ? extends T> f,
843     UniHandle<S,T> c) {
844     Object r; S s; Throwable x;
845     if (a == null || (r = a.result) == null || f == null)
846     return false;
847     if (result == null) {
848     try {
849     if (c != null && !c.claim())
850     return false;
851     if (r instanceof AltResult) {
852     x = ((AltResult)r).ex;
853     s = null;
854     } else {
855     x = null;
856     @SuppressWarnings("unchecked") S ss = (S) r;
857     s = ss;
858     }
859     completeValue(f.apply(s, x));
860     } catch (Throwable ex) {
861     completeThrowable(ex);
862     }
863     }
864     return true;
865     }
866    
867     private <V> CompletableFuture<V> uniHandleStage(
868     Executor e, BiFunction<? super T, Throwable, ? extends V> f) {
869     if (f == null) throw new NullPointerException();
870     CompletableFuture<V> d = newIncompleteFuture();
871     if (e != null || !d.uniHandle(this, f, null)) {
872     UniHandle<T,V> c = new UniHandle<T,V>(e, d, this, f);
873 dl 1.2 if (e != null && result != null) {
874     try {
875     e.execute(c);
876     } catch (Throwable ex) {
877     d.completeThrowable(ex);
878     }
879     }
880     else {
881     push(c);
882     c.tryFire(SYNC);
883     }
884 jsr166 1.1 }
885     return d;
886     }
887    
888     @SuppressWarnings("serial")
889     static final class UniExceptionally<T> extends UniCompletion<T,T> {
890     Function<? super Throwable, ? extends T> fn;
891     UniExceptionally(CompletableFuture<T> dep, CompletableFuture<T> src,
892     Function<? super Throwable, ? extends T> fn) {
893     super(null, dep, src); this.fn = fn;
894     }
895     final CompletableFuture<T> tryFire(int mode) { // never ASYNC
896     // assert mode != ASYNC;
897     CompletableFuture<T> d; CompletableFuture<T> a;
898     if ((d = dep) == null || !d.uniExceptionally(a = src, fn, this))
899     return null;
900     dep = null; src = null; fn = null;
901     return d.postFire(a, mode);
902     }
903     }
904    
905     final boolean uniExceptionally(CompletableFuture<T> a,
906     Function<? super Throwable, ? extends T> f,
907     UniExceptionally<T> c) {
908     Object r; Throwable x;
909     if (a == null || (r = a.result) == null || f == null)
910     return false;
911     if (result == null) {
912     try {
913     if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) {
914     if (c != null && !c.claim())
915     return false;
916     completeValue(f.apply(x));
917     } else
918     internalComplete(r);
919     } catch (Throwable ex) {
920     completeThrowable(ex);
921     }
922     }
923     return true;
924     }
925    
926     private CompletableFuture<T> uniExceptionallyStage(
927     Function<Throwable, ? extends T> f) {
928     if (f == null) throw new NullPointerException();
929     CompletableFuture<T> d = newIncompleteFuture();
930     if (!d.uniExceptionally(this, f, null)) {
931     UniExceptionally<T> c = new UniExceptionally<T>(d, this, f);
932     push(c);
933     c.tryFire(SYNC);
934     }
935     return d;
936     }
937    
938     @SuppressWarnings("serial")
939     static final class UniRelay<T> extends UniCompletion<T,T> { // for Compose
940     UniRelay(CompletableFuture<T> dep, CompletableFuture<T> src) {
941     super(null, dep, src);
942     }
943     final CompletableFuture<T> tryFire(int mode) {
944     CompletableFuture<T> d; CompletableFuture<T> a;
945     if ((d = dep) == null || !d.uniRelay(a = src))
946     return null;
947     src = null; dep = null;
948     return d.postFire(a, mode);
949     }
950     }
951    
952     final boolean uniRelay(CompletableFuture<T> a) {
953     Object r;
954     if (a == null || (r = a.result) == null)
955     return false;
956     if (result == null) // no need to claim
957     completeRelay(r);
958     return true;
959     }
960    
961     private CompletableFuture<T> uniCopyStage() {
962     Object r;
963     CompletableFuture<T> d = newIncompleteFuture();
964     if ((r = result) != null)
965     d.completeRelay(r);
966     else {
967     UniRelay<T> c = new UniRelay<T>(d, this);
968     push(c);
969     c.tryFire(SYNC);
970     }
971     return d;
972     }
973    
974     private MinimalStage<T> uniAsMinimalStage() {
975     Object r;
976     if ((r = result) != null)
977     return new MinimalStage<T>(encodeRelay(r));
978     MinimalStage<T> d = new MinimalStage<T>();
979     UniRelay<T> c = new UniRelay<T>(d, this);
980     push(c);
981     c.tryFire(SYNC);
982     return d;
983     }
984    
985     @SuppressWarnings("serial")
986     static final class UniCompose<T,V> extends UniCompletion<T,V> {
987     Function<? super T, ? extends CompletionStage<V>> fn;
988     UniCompose(Executor executor, CompletableFuture<V> dep,
989     CompletableFuture<T> src,
990     Function<? super T, ? extends CompletionStage<V>> fn) {
991     super(executor, dep, src); this.fn = fn;
992     }
993     final CompletableFuture<V> tryFire(int mode) {
994     CompletableFuture<V> d; CompletableFuture<T> a;
995     if ((d = dep) == null ||
996     !d.uniCompose(a = src, fn, mode > 0 ? null : this))
997     return null;
998     dep = null; src = null; fn = null;
999     return d.postFire(a, mode);
1000     }
1001     }
1002    
1003     final <S> boolean uniCompose(
1004     CompletableFuture<S> a,
1005     Function<? super S, ? extends CompletionStage<T>> f,
1006     UniCompose<S,T> c) {
1007     Object r; Throwable x;
1008     if (a == null || (r = a.result) == null || f == null)
1009     return false;
1010     tryComplete: if (result == null) {
1011     if (r instanceof AltResult) {
1012     if ((x = ((AltResult)r).ex) != null) {
1013     completeThrowable(x, r);
1014     break tryComplete;
1015     }
1016     r = null;
1017     }
1018     try {
1019     if (c != null && !c.claim())
1020     return false;
1021     @SuppressWarnings("unchecked") S s = (S) r;
1022     CompletableFuture<T> g = f.apply(s).toCompletableFuture();
1023     if (g.result == null || !uniRelay(g)) {
1024     UniRelay<T> copy = new UniRelay<T>(this, g);
1025     g.push(copy);
1026     copy.tryFire(SYNC);
1027     if (result == null)
1028     return false;
1029     }
1030     } catch (Throwable ex) {
1031     completeThrowable(ex);
1032     }
1033     }
1034     return true;
1035     }
1036    
1037     private <V> CompletableFuture<V> uniComposeStage(
1038     Executor e, Function<? super T, ? extends CompletionStage<V>> f) {
1039     if (f == null) throw new NullPointerException();
1040     Object r, s; Throwable x;
1041     CompletableFuture<V> d = newIncompleteFuture();
1042 dl 1.2 if ((r = result) != null && e == null) {
1043 jsr166 1.1 if (r instanceof AltResult) {
1044     if ((x = ((AltResult)r).ex) != null) {
1045     d.result = encodeThrowable(x, r);
1046     return d;
1047     }
1048     r = null;
1049     }
1050     try {
1051     @SuppressWarnings("unchecked") T t = (T) r;
1052     CompletableFuture<V> g = f.apply(t).toCompletableFuture();
1053     if ((s = g.result) != null)
1054     d.completeRelay(s);
1055     else {
1056     UniRelay<V> c = new UniRelay<V>(d, g);
1057     g.push(c);
1058     c.tryFire(SYNC);
1059     }
1060     return d;
1061     } catch (Throwable ex) {
1062     d.result = encodeThrowable(ex);
1063     return d;
1064     }
1065     }
1066     UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);
1067 dl 1.2 if (r != null && e != null) {
1068     try {
1069     e.execute(new UniCompose<T,V>(null, d, this, f));
1070     } catch (Throwable ex) {
1071     d.completeThrowable(ex);
1072     }
1073     }
1074     else {
1075     push(c);
1076     c.tryFire(SYNC);
1077     }
1078 jsr166 1.1 return d;
1079     }
1080    
1081     /* ------------- Two-input Completions -------------- */
1082    
1083     /** A Completion for an action with two sources */
1084     @SuppressWarnings("serial")
1085     abstract static class BiCompletion<T,U,V> extends UniCompletion<T,V> {
1086     CompletableFuture<U> snd; // second source for action
1087     BiCompletion(Executor executor, CompletableFuture<V> dep,
1088     CompletableFuture<T> src, CompletableFuture<U> snd) {
1089     super(executor, dep, src); this.snd = snd;
1090     }
1091     }
1092    
1093     /** A Completion delegating to a BiCompletion */
1094     @SuppressWarnings("serial")
1095     static final class CoCompletion extends Completion {
1096     BiCompletion<?,?,?> base;
1097     CoCompletion(BiCompletion<?,?,?> base) { this.base = base; }
1098     final CompletableFuture<?> tryFire(int mode) {
1099     BiCompletion<?,?,?> c; CompletableFuture<?> d;
1100     if ((c = base) == null || (d = c.tryFire(mode)) == null)
1101     return null;
1102     base = null; // detach
1103     return d;
1104     }
1105     final boolean isLive() {
1106     BiCompletion<?,?,?> c;
1107     return (c = base) != null && c.dep != null;
1108     }
1109     }
1110    
1111     /** Pushes completion to this and b unless both done. */
1112     final void bipush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
1113     if (c != null) {
1114     Object r;
1115     while ((r = result) == null && !tryPushStack(c))
1116     lazySetNext(c, null); // clear on failure
1117     if (b != null && b != this && b.result == null) {
1118     Completion q = (r != null) ? c : new CoCompletion(c);
1119     while (b.result == null && !b.tryPushStack(q))
1120     lazySetNext(q, null); // clear on failure
1121     }
1122     }
1123     }
1124    
1125     /** Post-processing after successful BiCompletion tryFire. */
1126     final CompletableFuture<T> postFire(CompletableFuture<?> a,
1127     CompletableFuture<?> b, int mode) {
1128     if (b != null && b.stack != null) { // clean second source
1129     if (b.result == null)
1130     b.cleanStack();
1131     else if (mode >= 0)
1132     b.postComplete();
1133     }
1134     return postFire(a, mode);
1135     }
1136    
1137     @SuppressWarnings("serial")
1138     static final class BiApply<T,U,V> extends BiCompletion<T,U,V> {
1139     BiFunction<? super T,? super U,? extends V> fn;
1140     BiApply(Executor executor, CompletableFuture<V> dep,
1141     CompletableFuture<T> src, CompletableFuture<U> snd,
1142     BiFunction<? super T,? super U,? extends V> fn) {
1143     super(executor, dep, src, snd); this.fn = fn;
1144     }
1145     final CompletableFuture<V> tryFire(int mode) {
1146     CompletableFuture<V> d;
1147     CompletableFuture<T> a;
1148     CompletableFuture<U> b;
1149     if ((d = dep) == null ||
1150     !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this))
1151     return null;
1152     dep = null; src = null; snd = null; fn = null;
1153     return d.postFire(a, b, mode);
1154     }
1155     }
1156    
1157     final <R,S> boolean biApply(CompletableFuture<R> a,
1158     CompletableFuture<S> b,
1159     BiFunction<? super R,? super S,? extends T> f,
1160     BiApply<R,S,T> c) {
1161     Object r, s; Throwable x;
1162     if (a == null || (r = a.result) == null ||
1163     b == null || (s = b.result) == null || f == null)
1164     return false;
1165     tryComplete: if (result == null) {
1166     if (r instanceof AltResult) {
1167     if ((x = ((AltResult)r).ex) != null) {
1168     completeThrowable(x, r);
1169     break tryComplete;
1170     }
1171     r = null;
1172     }
1173     if (s instanceof AltResult) {
1174     if ((x = ((AltResult)s).ex) != null) {
1175     completeThrowable(x, s);
1176     break tryComplete;
1177     }
1178     s = null;
1179     }
1180     try {
1181     if (c != null && !c.claim())
1182     return false;
1183     @SuppressWarnings("unchecked") R rr = (R) r;
1184     @SuppressWarnings("unchecked") S ss = (S) s;
1185     completeValue(f.apply(rr, ss));
1186     } catch (Throwable ex) {
1187     completeThrowable(ex);
1188     }
1189     }
1190     return true;
1191     }
1192    
1193     private <U,V> CompletableFuture<V> biApplyStage(
1194     Executor e, CompletionStage<U> o,
1195     BiFunction<? super T,? super U,? extends V> f) {
1196     CompletableFuture<U> b;
1197     if (f == null || (b = o.toCompletableFuture()) == null)
1198     throw new NullPointerException();
1199     CompletableFuture<V> d = newIncompleteFuture();
1200     if (e != null || !d.biApply(this, b, f, null)) {
1201     BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);
1202 dl 1.2 if (e != null && result != null && b.result != null) {
1203     try {
1204     e.execute(c);
1205     } catch (Throwable ex) {
1206     d.completeThrowable(ex);
1207     }
1208     }
1209     else {
1210     bipush(b, c);
1211     c.tryFire(SYNC);
1212     }
1213 jsr166 1.1 }
1214     return d;
1215     }
1216    
1217     @SuppressWarnings("serial")
1218     static final class BiAccept<T,U> extends BiCompletion<T,U,Void> {
1219     BiConsumer<? super T,? super U> fn;
1220     BiAccept(Executor executor, CompletableFuture<Void> dep,
1221     CompletableFuture<T> src, CompletableFuture<U> snd,
1222     BiConsumer<? super T,? super U> fn) {
1223     super(executor, dep, src, snd); this.fn = fn;
1224     }
1225     final CompletableFuture<Void> tryFire(int mode) {
1226     CompletableFuture<Void> d;
1227     CompletableFuture<T> a;
1228     CompletableFuture<U> b;
1229     if ((d = dep) == null ||
1230     !d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this))
1231     return null;
1232     dep = null; src = null; snd = null; fn = null;
1233     return d.postFire(a, b, mode);
1234     }
1235     }
1236    
1237     final <R,S> boolean biAccept(CompletableFuture<R> a,
1238     CompletableFuture<S> b,
1239     BiConsumer<? super R,? super S> f,
1240     BiAccept<R,S> c) {
1241     Object r, s; Throwable x;
1242     if (a == null || (r = a.result) == null ||
1243     b == null || (s = b.result) == null || f == null)
1244     return false;
1245     tryComplete: if (result == null) {
1246     if (r instanceof AltResult) {
1247     if ((x = ((AltResult)r).ex) != null) {
1248     completeThrowable(x, r);
1249     break tryComplete;
1250     }
1251     r = null;
1252     }
1253     if (s instanceof AltResult) {
1254     if ((x = ((AltResult)s).ex) != null) {
1255     completeThrowable(x, s);
1256     break tryComplete;
1257     }
1258     s = null;
1259     }
1260     try {
1261     if (c != null && !c.claim())
1262     return false;
1263     @SuppressWarnings("unchecked") R rr = (R) r;
1264     @SuppressWarnings("unchecked") S ss = (S) s;
1265     f.accept(rr, ss);
1266     completeNull();
1267     } catch (Throwable ex) {
1268     completeThrowable(ex);
1269     }
1270     }
1271     return true;
1272     }
1273    
1274     private <U> CompletableFuture<Void> biAcceptStage(
1275     Executor e, CompletionStage<U> o,
1276     BiConsumer<? super T,? super U> f) {
1277     CompletableFuture<U> b;
1278     if (f == null || (b = o.toCompletableFuture()) == null)
1279     throw new NullPointerException();
1280     CompletableFuture<Void> d = newIncompleteFuture();
1281     if (e != null || !d.biAccept(this, b, f, null)) {
1282     BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
1283 dl 1.2 if (e != null && result != null && b.result != null) {
1284     try {
1285     e.execute(c);
1286     } catch (Throwable ex) {
1287     d.completeThrowable(ex);
1288     }
1289     }
1290     else {
1291     bipush(b, c);
1292     c.tryFire(SYNC);
1293     }
1294 jsr166 1.1 }
1295     return d;
1296     }
1297    
1298     @SuppressWarnings("serial")
1299     static final class BiRun<T,U> extends BiCompletion<T,U,Void> {
1300     Runnable fn;
1301     BiRun(Executor executor, CompletableFuture<Void> dep,
1302     CompletableFuture<T> src,
1303     CompletableFuture<U> snd,
1304     Runnable fn) {
1305     super(executor, dep, src, snd); this.fn = fn;
1306     }
1307     final CompletableFuture<Void> tryFire(int mode) {
1308     CompletableFuture<Void> d;
1309     CompletableFuture<T> a;
1310     CompletableFuture<U> b;
1311     if ((d = dep) == null ||
1312     !d.biRun(a = src, b = snd, fn, mode > 0 ? null : this))
1313     return null;
1314     dep = null; src = null; snd = null; fn = null;
1315     return d.postFire(a, b, mode);
1316     }
1317     }
1318    
1319     final boolean biRun(CompletableFuture<?> a, CompletableFuture<?> b,
1320     Runnable f, BiRun<?,?> c) {
1321     Object r, s; Throwable x;
1322     if (a == null || (r = a.result) == null ||
1323     b == null || (s = b.result) == null || f == null)
1324     return false;
1325     if (result == null) {
1326     if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1327     completeThrowable(x, r);
1328     else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
1329     completeThrowable(x, s);
1330     else
1331     try {
1332     if (c != null && !c.claim())
1333     return false;
1334     f.run();
1335     completeNull();
1336     } catch (Throwable ex) {
1337     completeThrowable(ex);
1338     }
1339     }
1340     return true;
1341     }
1342    
1343     private CompletableFuture<Void> biRunStage(Executor e, CompletionStage<?> o,
1344     Runnable f) {
1345     CompletableFuture<?> b;
1346     if (f == null || (b = o.toCompletableFuture()) == null)
1347     throw new NullPointerException();
1348     CompletableFuture<Void> d = newIncompleteFuture();
1349     if (e != null || !d.biRun(this, b, f, null)) {
1350     BiRun<T,?> c = new BiRun<>(e, d, this, b, f);
1351 dl 1.2 if (e != null && result != null && b.result != null) {
1352     try {
1353     e.execute(c);
1354     } catch (Throwable ex) {
1355     d.completeThrowable(ex);
1356     }
1357     }
1358     else {
1359     bipush(b, c);
1360     c.tryFire(SYNC);
1361     }
1362 jsr166 1.1 }
1363     return d;
1364     }
1365    
1366     @SuppressWarnings("serial")
1367     static final class BiRelay<T,U> extends BiCompletion<T,U,Void> { // for And
1368     BiRelay(CompletableFuture<Void> dep,
1369     CompletableFuture<T> src,
1370     CompletableFuture<U> snd) {
1371     super(null, dep, src, snd);
1372     }
1373     final CompletableFuture<Void> tryFire(int mode) {
1374     CompletableFuture<Void> d;
1375     CompletableFuture<T> a;
1376     CompletableFuture<U> b;
1377     if ((d = dep) == null || !d.biRelay(a = src, b = snd))
1378     return null;
1379     src = null; snd = null; dep = null;
1380     return d.postFire(a, b, mode);
1381     }
1382     }
1383    
1384     boolean biRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
1385     Object r, s; Throwable x;
1386     if (a == null || (r = a.result) == null ||
1387     b == null || (s = b.result) == null)
1388     return false;
1389     if (result == null) {
1390     if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1391     completeThrowable(x, r);
1392     else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
1393     completeThrowable(x, s);
1394     else
1395     completeNull();
1396     }
1397     return true;
1398     }
1399    
1400     /** Recursively constructs a tree of completions. */
1401     static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs,
1402     int lo, int hi) {
1403     CompletableFuture<Void> d = new CompletableFuture<Void>();
1404     if (lo > hi) // empty
1405     d.result = NIL;
1406     else {
1407     CompletableFuture<?> a, b;
1408     int mid = (lo + hi) >>> 1;
1409     if ((a = (lo == mid ? cfs[lo] :
1410     andTree(cfs, lo, mid))) == null ||
1411     (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
1412     andTree(cfs, mid+1, hi))) == null)
1413     throw new NullPointerException();
1414     if (!d.biRelay(a, b)) {
1415     BiRelay<?,?> c = new BiRelay<>(d, a, b);
1416     a.bipush(b, c);
1417     c.tryFire(SYNC);
1418     }
1419     }
1420     return d;
1421     }
1422    
1423     /* ------------- Projected (Ored) BiCompletions -------------- */
1424    
1425     /** Pushes completion to this and b unless either done. */
1426     final void orpush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
1427     if (c != null) {
1428     while ((b == null || b.result == null) && result == null) {
1429     if (tryPushStack(c)) {
1430     if (b != null && b != this && b.result == null) {
1431     Completion q = new CoCompletion(c);
1432     while (result == null && b.result == null &&
1433     !b.tryPushStack(q))
1434     lazySetNext(q, null); // clear on failure
1435     }
1436     break;
1437     }
1438     lazySetNext(c, null); // clear on failure
1439     }
1440     }
1441     }
1442    
1443     @SuppressWarnings("serial")
1444     static final class OrApply<T,U extends T,V> extends BiCompletion<T,U,V> {
1445     Function<? super T,? extends V> fn;
1446     OrApply(Executor executor, CompletableFuture<V> dep,
1447     CompletableFuture<T> src,
1448     CompletableFuture<U> snd,
1449     Function<? super T,? extends V> fn) {
1450     super(executor, dep, src, snd); this.fn = fn;
1451     }
1452     final CompletableFuture<V> tryFire(int mode) {
1453     CompletableFuture<V> d;
1454     CompletableFuture<T> a;
1455     CompletableFuture<U> b;
1456     if ((d = dep) == null ||
1457     !d.orApply(a = src, b = snd, fn, mode > 0 ? null : this))
1458     return null;
1459     dep = null; src = null; snd = null; fn = null;
1460     return d.postFire(a, b, mode);
1461     }
1462     }
1463    
1464     final <R,S extends R> boolean orApply(CompletableFuture<R> a,
1465     CompletableFuture<S> b,
1466     Function<? super R, ? extends T> f,
1467     OrApply<R,S,T> c) {
1468     Object r; Throwable x;
1469     if (a == null || b == null ||
1470     ((r = a.result) == null && (r = b.result) == null) || f == null)
1471     return false;
1472     tryComplete: if (result == null) {
1473     try {
1474     if (c != null && !c.claim())
1475     return false;
1476     if (r instanceof AltResult) {
1477     if ((x = ((AltResult)r).ex) != null) {
1478     completeThrowable(x, r);
1479     break tryComplete;
1480     }
1481     r = null;
1482     }
1483     @SuppressWarnings("unchecked") R rr = (R) r;
1484     completeValue(f.apply(rr));
1485     } catch (Throwable ex) {
1486     completeThrowable(ex);
1487     }
1488     }
1489     return true;
1490     }
1491    
1492     private <U extends T,V> CompletableFuture<V> orApplyStage(
1493     Executor e, CompletionStage<U> o,
1494     Function<? super T, ? extends V> f) {
1495     CompletableFuture<U> b;
1496     if (f == null || (b = o.toCompletableFuture()) == null)
1497     throw new NullPointerException();
1498     CompletableFuture<V> d = newIncompleteFuture();
1499     if (e != null || !d.orApply(this, b, f, null)) {
1500     OrApply<T,U,V> c = new OrApply<T,U,V>(e, d, this, b, f);
1501 dl 1.2 if (e != null && (result != null || b.result != null)) {
1502     try {
1503     e.execute(c);
1504     } catch (Throwable ex) {
1505     d.completeThrowable(ex);
1506     }
1507     }
1508     else {
1509     orpush(b, c);
1510     c.tryFire(SYNC);
1511     }
1512 jsr166 1.1 }
1513     return d;
1514     }
1515    
1516     @SuppressWarnings("serial")
1517     static final class OrAccept<T,U extends T> extends BiCompletion<T,U,Void> {
1518     Consumer<? super T> fn;
1519     OrAccept(Executor executor, CompletableFuture<Void> dep,
1520     CompletableFuture<T> src,
1521     CompletableFuture<U> snd,
1522     Consumer<? super T> fn) {
1523     super(executor, dep, src, snd); this.fn = fn;
1524     }
1525     final CompletableFuture<Void> tryFire(int mode) {
1526     CompletableFuture<Void> d;
1527     CompletableFuture<T> a;
1528     CompletableFuture<U> b;
1529     if ((d = dep) == null ||
1530     !d.orAccept(a = src, b = snd, fn, mode > 0 ? null : this))
1531     return null;
1532     dep = null; src = null; snd = null; fn = null;
1533     return d.postFire(a, b, mode);
1534     }
1535     }
1536    
1537     final <R,S extends R> boolean orAccept(CompletableFuture<R> a,
1538     CompletableFuture<S> b,
1539     Consumer<? super R> f,
1540     OrAccept<R,S> c) {
1541     Object r; Throwable x;
1542     if (a == null || b == null ||
1543     ((r = a.result) == null && (r = b.result) == null) || f == null)
1544     return false;
1545     tryComplete: if (result == null) {
1546     try {
1547     if (c != null && !c.claim())
1548     return false;
1549     if (r instanceof AltResult) {
1550     if ((x = ((AltResult)r).ex) != null) {
1551     completeThrowable(x, r);
1552     break tryComplete;
1553     }
1554     r = null;
1555     }
1556     @SuppressWarnings("unchecked") R rr = (R) r;
1557     f.accept(rr);
1558     completeNull();
1559     } catch (Throwable ex) {
1560     completeThrowable(ex);
1561     }
1562     }
1563     return true;
1564     }
1565    
1566     private <U extends T> CompletableFuture<Void> orAcceptStage(
1567     Executor e, CompletionStage<U> o, Consumer<? super T> f) {
1568     CompletableFuture<U> b;
1569     if (f == null || (b = o.toCompletableFuture()) == null)
1570     throw new NullPointerException();
1571     CompletableFuture<Void> d = newIncompleteFuture();
1572     if (e != null || !d.orAccept(this, b, f, null)) {
1573     OrAccept<T,U> c = new OrAccept<T,U>(e, d, this, b, f);
1574 dl 1.2 if (e != null && (result != null || b.result != null)) {
1575     try {
1576     e.execute(c);
1577     } catch (Throwable ex) {
1578     d.completeThrowable(ex);
1579     }
1580     }
1581     else {
1582     orpush(b, c);
1583     c.tryFire(SYNC);
1584     }
1585 jsr166 1.1 }
1586     return d;
1587     }
1588    
1589     @SuppressWarnings("serial")
1590     static final class OrRun<T,U> extends BiCompletion<T,U,Void> {
1591     Runnable fn;
1592     OrRun(Executor executor, CompletableFuture<Void> dep,
1593     CompletableFuture<T> src,
1594     CompletableFuture<U> snd,
1595     Runnable fn) {
1596     super(executor, dep, src, snd); this.fn = fn;
1597     }
1598     final CompletableFuture<Void> tryFire(int mode) {
1599     CompletableFuture<Void> d;
1600     CompletableFuture<T> a;
1601     CompletableFuture<U> b;
1602     if ((d = dep) == null ||
1603     !d.orRun(a = src, b = snd, fn, mode > 0 ? null : this))
1604     return null;
1605     dep = null; src = null; snd = null; fn = null;
1606     return d.postFire(a, b, mode);
1607     }
1608     }
1609    
1610     final boolean orRun(CompletableFuture<?> a, CompletableFuture<?> b,
1611     Runnable f, OrRun<?,?> c) {
1612     Object r; Throwable x;
1613     if (a == null || b == null ||
1614     ((r = a.result) == null && (r = b.result) == null) || f == null)
1615     return false;
1616     if (result == null) {
1617     try {
1618     if (c != null && !c.claim())
1619     return false;
1620     if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
1621     completeThrowable(x, r);
1622     else {
1623     f.run();
1624     completeNull();
1625     }
1626     } catch (Throwable ex) {
1627     completeThrowable(ex);
1628     }
1629     }
1630     return true;
1631     }
1632    
1633     private CompletableFuture<Void> orRunStage(Executor e, CompletionStage<?> o,
1634     Runnable f) {
1635     CompletableFuture<?> b;
1636     if (f == null || (b = o.toCompletableFuture()) == null)
1637     throw new NullPointerException();
1638     CompletableFuture<Void> d = newIncompleteFuture();
1639     if (e != null || !d.orRun(this, b, f, null)) {
1640     OrRun<T,?> c = new OrRun<>(e, d, this, b, f);
1641 dl 1.2 if (e != null && (result != null || b.result != null)) {
1642     try {
1643     e.execute(c);
1644     } catch (Throwable ex) {
1645     d.completeThrowable(ex);
1646     }
1647     }
1648     else {
1649     orpush(b, c);
1650     c.tryFire(SYNC);
1651     }
1652 jsr166 1.1 }
1653     return d;
1654     }
1655    
1656     @SuppressWarnings("serial")
1657     static final class OrRelay<T,U> extends BiCompletion<T,U,Object> { // for Or
1658     OrRelay(CompletableFuture<Object> dep, CompletableFuture<T> src,
1659     CompletableFuture<U> snd) {
1660     super(null, dep, src, snd);
1661     }
1662     final CompletableFuture<Object> tryFire(int mode) {
1663     CompletableFuture<Object> d;
1664     CompletableFuture<T> a;
1665     CompletableFuture<U> b;
1666     if ((d = dep) == null || !d.orRelay(a = src, b = snd))
1667     return null;
1668     src = null; snd = null; dep = null;
1669     return d.postFire(a, b, mode);
1670     }
1671     }
1672    
1673     final boolean orRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
1674     Object r;
1675     if (a == null || b == null ||
1676     ((r = a.result) == null && (r = b.result) == null))
1677     return false;
1678     if (result == null)
1679     completeRelay(r);
1680     return true;
1681     }
1682    
1683     /** Recursively constructs a tree of completions. */
1684     static CompletableFuture<Object> orTree(CompletableFuture<?>[] cfs,
1685     int lo, int hi) {
1686     CompletableFuture<Object> d = new CompletableFuture<Object>();
1687     if (lo <= hi) {
1688     CompletableFuture<?> a, b;
1689     int mid = (lo + hi) >>> 1;
1690     if ((a = (lo == mid ? cfs[lo] :
1691     orTree(cfs, lo, mid))) == null ||
1692     (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
1693     orTree(cfs, mid+1, hi))) == null)
1694     throw new NullPointerException();
1695     if (!d.orRelay(a, b)) {
1696     OrRelay<?,?> c = new OrRelay<>(d, a, b);
1697     a.orpush(b, c);
1698     c.tryFire(SYNC);
1699     }
1700     }
1701     return d;
1702     }
1703    
1704     /* ------------- Zero-input Async forms -------------- */
1705    
1706     @SuppressWarnings("serial")
1707     static final class AsyncSupply<T> extends ForkJoinTask<Void>
1708     implements Runnable, AsynchronousCompletionTask {
1709     CompletableFuture<T> dep; Supplier<? extends T> fn;
1710     AsyncSupply(CompletableFuture<T> dep, Supplier<? extends T> fn) {
1711     this.dep = dep; this.fn = fn;
1712     }
1713    
1714     public final Void getRawResult() { return null; }
1715     public final void setRawResult(Void v) {}
1716     public final boolean exec() { run(); return true; }
1717    
1718     public void run() {
1719     CompletableFuture<T> d; Supplier<? extends T> f;
1720     if ((d = dep) != null && (f = fn) != null) {
1721     dep = null; fn = null;
1722     if (d.result == null) {
1723     try {
1724     d.completeValue(f.get());
1725     } catch (Throwable ex) {
1726     d.completeThrowable(ex);
1727     }
1728     }
1729     d.postComplete();
1730     }
1731     }
1732     }
1733    
1734     static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
1735     Supplier<U> f) {
1736     if (f == null) throw new NullPointerException();
1737     CompletableFuture<U> d = new CompletableFuture<U>();
1738     e.execute(new AsyncSupply<U>(d, f));
1739     return d;
1740     }
1741    
1742     @SuppressWarnings("serial")
1743     static final class AsyncRun extends ForkJoinTask<Void>
1744     implements Runnable, AsynchronousCompletionTask {
1745     CompletableFuture<Void> dep; Runnable fn;
1746     AsyncRun(CompletableFuture<Void> dep, Runnable fn) {
1747     this.dep = dep; this.fn = fn;
1748     }
1749    
1750     public final Void getRawResult() { return null; }
1751     public final void setRawResult(Void v) {}
1752     public final boolean exec() { run(); return true; }
1753    
1754     public void run() {
1755     CompletableFuture<Void> d; Runnable f;
1756     if ((d = dep) != null && (f = fn) != null) {
1757     dep = null; fn = null;
1758     if (d.result == null) {
1759     try {
1760     f.run();
1761     d.completeNull();
1762     } catch (Throwable ex) {
1763     d.completeThrowable(ex);
1764     }
1765     }
1766     d.postComplete();
1767     }
1768     }
1769     }
1770    
1771     static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
1772     if (f == null) throw new NullPointerException();
1773     CompletableFuture<Void> d = new CompletableFuture<Void>();
1774     e.execute(new AsyncRun(d, f));
1775     return d;
1776     }
1777    
1778     /* ------------- Signallers -------------- */
1779    
1780     /**
1781     * Completion for recording and releasing a waiting thread. This
1782     * class implements ManagedBlocker to avoid starvation when
1783     * blocking actions pile up in ForkJoinPools.
1784     */
1785     @SuppressWarnings("serial")
1786     static final class Signaller extends Completion
1787     implements ForkJoinPool.ManagedBlocker {
1788     long nanos; // remaining wait time if timed
1789     final long deadline; // non-zero if timed
1790     final boolean interruptible;
1791     boolean interrupted;
1792     volatile Thread thread;
1793    
1794     Signaller(boolean interruptible, long nanos, long deadline) {
1795     this.thread = Thread.currentThread();
1796     this.interruptible = interruptible;
1797     this.nanos = nanos;
1798     this.deadline = deadline;
1799     }
1800     final CompletableFuture<?> tryFire(int ignore) {
1801     Thread w; // no need to atomically claim
1802     if ((w = thread) != null) {
1803     thread = null;
1804     LockSupport.unpark(w);
1805     }
1806     return null;
1807     }
1808     public boolean isReleasable() {
1809     if (Thread.interrupted())
1810     interrupted = true;
1811     return ((interrupted && interruptible) ||
1812     (deadline != 0L &&
1813     (nanos <= 0L ||
1814     (nanos = deadline - System.nanoTime()) <= 0L)) ||
1815     thread == null);
1816     }
1817     public boolean block() {
1818     while (!isReleasable()) {
1819     if (deadline == 0L)
1820     LockSupport.park(this);
1821     else
1822     LockSupport.parkNanos(this, nanos);
1823     }
1824     return true;
1825     }
1826     final boolean isLive() { return thread != null; }
1827     }
1828    
1829     /**
1830     * Returns raw result after waiting, or null if interruptible and
1831     * interrupted.
1832     */
1833     private Object waitingGet(boolean interruptible) {
1834     Signaller q = null;
1835     boolean queued = false;
1836     Object r;
1837     while ((r = result) == null) {
1838 dl 1.2 if (q == null) {
1839     q = new Signaller(interruptible, 0L, 0L);
1840     ForkJoinPool.helpAsyncBlocker(defaultExecutor(), q);
1841 jsr166 1.1 }
1842     else if (!queued)
1843     queued = tryPushStack(q);
1844     else {
1845     try {
1846     ForkJoinPool.managedBlock(q);
1847     } catch (InterruptedException ie) { // currently cannot happen
1848     q.interrupted = true;
1849     }
1850     if (q.interrupted && interruptible)
1851     break;
1852     }
1853     }
1854     if (q != null) {
1855     q.thread = null;
1856     if (q.interrupted) {
1857     if (interruptible)
1858     cleanStack();
1859     else
1860     Thread.currentThread().interrupt();
1861     }
1862     }
1863     if (r != null)
1864     postComplete();
1865     return r;
1866     }
1867    
1868     /**
1869     * Returns raw result after waiting, or null if interrupted, or
1870     * throws TimeoutException on timeout.
1871     */
1872     private Object timedGet(long nanos) throws TimeoutException {
1873     if (Thread.interrupted())
1874     return null;
1875     if (nanos > 0L) {
1876     long d = System.nanoTime() + nanos;
1877     long deadline = (d == 0L) ? 1L : d; // avoid 0
1878     Signaller q = null;
1879     boolean queued = false;
1880     Object r;
1881 dl 1.2 while ((r = result) == null) { // similar to untimed
1882     if (q == null) {
1883 jsr166 1.1 q = new Signaller(true, nanos, deadline);
1884 dl 1.2 ForkJoinPool.helpAsyncBlocker(defaultExecutor(), q);
1885     }
1886 jsr166 1.1 else if (!queued)
1887     queued = tryPushStack(q);
1888     else if (q.nanos <= 0L)
1889     break;
1890     else {
1891     try {
1892     ForkJoinPool.managedBlock(q);
1893     } catch (InterruptedException ie) {
1894     q.interrupted = true;
1895     }
1896     if (q.interrupted)
1897     break;
1898     }
1899     }
1900     if (q != null)
1901     q.thread = null;
1902     if (r != null)
1903     postComplete();
1904     else
1905     cleanStack();
1906     if (r != null || (q != null && q.interrupted))
1907     return r;
1908     }
1909     throw new TimeoutException();
1910     }
1911    
1912     /* ------------- public methods -------------- */
1913    
1914     /**
1915     * Creates a new incomplete CompletableFuture.
1916     */
1917     public CompletableFuture() {
1918     }
1919    
1920     /**
1921     * Creates a new complete CompletableFuture with given encoded result.
1922     */
1923     CompletableFuture(Object r) {
1924     this.result = r;
1925     }
1926    
1927     /**
1928     * Returns a new CompletableFuture that is asynchronously completed
1929     * by a task running in the {@link ForkJoinPool#commonPool()} with
1930     * the value obtained by calling the given Supplier.
1931     *
1932     * @param supplier a function returning the value to be used
1933     * to complete the returned CompletableFuture
1934     * @param <U> the function's return type
1935     * @return the new CompletableFuture
1936     */
1937     public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
1938     return asyncSupplyStage(ASYNC_POOL, supplier);
1939     }
1940    
1941     /**
1942     * Returns a new CompletableFuture that is asynchronously completed
1943     * by a task running in the given executor with the value obtained
1944     * by calling the given Supplier.
1945     *
1946     * @param supplier a function returning the value to be used
1947     * to complete the returned CompletableFuture
1948     * @param executor the executor to use for asynchronous execution
1949     * @param <U> the function's return type
1950     * @return the new CompletableFuture
1951     */
1952     public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
1953     Executor executor) {
1954     return asyncSupplyStage(screenExecutor(executor), supplier);
1955     }
1956    
1957     /**
1958     * Returns a new CompletableFuture that is asynchronously completed
1959     * by a task running in the {@link ForkJoinPool#commonPool()} after
1960     * it runs the given action.
1961     *
1962     * @param runnable the action to run before completing the
1963     * returned CompletableFuture
1964     * @return the new CompletableFuture
1965     */
1966     public static CompletableFuture<Void> runAsync(Runnable runnable) {
1967     return asyncRunStage(ASYNC_POOL, runnable);
1968     }
1969    
1970     /**
1971     * Returns a new CompletableFuture that is asynchronously completed
1972     * by a task running in the given executor after it runs the given
1973     * action.
1974     *
1975     * @param runnable the action to run before completing the
1976     * returned CompletableFuture
1977     * @param executor the executor to use for asynchronous execution
1978     * @return the new CompletableFuture
1979     */
1980     public static CompletableFuture<Void> runAsync(Runnable runnable,
1981     Executor executor) {
1982     return asyncRunStage(screenExecutor(executor), runnable);
1983     }
1984    
1985     /**
1986     * Returns a new CompletableFuture that is already completed with
1987     * the given value.
1988     *
1989     * @param value the value
1990     * @param <U> the type of the value
1991     * @return the completed CompletableFuture
1992     */
1993     public static <U> CompletableFuture<U> completedFuture(U value) {
1994     return new CompletableFuture<U>((value == null) ? NIL : value);
1995     }
1996    
1997     /**
1998     * Returns {@code true} if completed in any fashion: normally,
1999     * exceptionally, or via cancellation.
2000     *
2001     * @return {@code true} if completed
2002     */
2003     public boolean isDone() {
2004     return result != null;
2005     }
2006    
2007     /**
2008     * Waits if necessary for this future to complete, and then
2009     * returns its result.
2010     *
2011     * @return the result value
2012     * @throws CancellationException if this future was cancelled
2013     * @throws ExecutionException if this future completed exceptionally
2014     * @throws InterruptedException if the current thread was interrupted
2015     * while waiting
2016     */
2017     public T get() throws InterruptedException, ExecutionException {
2018     Object r;
2019     return reportGet((r = result) == null ? waitingGet(true) : r);
2020     }
2021    
2022     /**
2023     * Waits if necessary for at most the given time for this future
2024     * to complete, and then returns its result, if available.
2025     *
2026     * @param timeout the maximum time to wait
2027     * @param unit the time unit of the timeout argument
2028     * @return the result value
2029     * @throws CancellationException if this future was cancelled
2030     * @throws ExecutionException if this future completed exceptionally
2031     * @throws InterruptedException if the current thread was interrupted
2032     * while waiting
2033     * @throws TimeoutException if the wait timed out
2034     */
2035     public T get(long timeout, TimeUnit unit)
2036     throws InterruptedException, ExecutionException, TimeoutException {
2037     Object r;
2038     long nanos = unit.toNanos(timeout);
2039     return reportGet((r = result) == null ? timedGet(nanos) : r);
2040     }
2041    
2042     /**
2043     * Returns the result value when complete, or throws an
2044     * (unchecked) exception if completed exceptionally. To better
2045     * conform with the use of common functional forms, if a
2046     * computation involved in the completion of this
2047     * CompletableFuture threw an exception, this method throws an
2048     * (unchecked) {@link CompletionException} with the underlying
2049     * exception as its cause.
2050     *
2051     * @return the result value
2052     * @throws CancellationException if the computation was cancelled
2053     * @throws CompletionException if this future completed
2054     * exceptionally or a completion computation threw an exception
2055     */
2056     public T join() {
2057     Object r;
2058     return reportJoin((r = result) == null ? waitingGet(false) : r);
2059     }
2060    
2061     /**
2062     * Returns the result value (or throws any encountered exception)
2063     * if completed, else returns the given valueIfAbsent.
2064     *
2065     * @param valueIfAbsent the value to return if not completed
2066     * @return the result value, if completed, else the given valueIfAbsent
2067     * @throws CancellationException if the computation was cancelled
2068     * @throws CompletionException if this future completed
2069     * exceptionally or a completion computation threw an exception
2070     */
2071     public T getNow(T valueIfAbsent) {
2072     Object r;
2073     return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
2074     }
2075    
2076     /**
2077     * If not already completed, sets the value returned by {@link
2078     * #get()} and related methods to the given value.
2079     *
2080     * @param value the result value
2081     * @return {@code true} if this invocation caused this CompletableFuture
2082     * to transition to a completed state, else {@code false}
2083     */
2084     public boolean complete(T value) {
2085     boolean triggered = completeValue(value);
2086     postComplete();
2087     return triggered;
2088     }
2089    
2090     /**
2091     * If not already completed, causes invocations of {@link #get()}
2092     * and related methods to throw the given exception.
2093     *
2094     * @param ex the exception
2095     * @return {@code true} if this invocation caused this CompletableFuture
2096     * to transition to a completed state, else {@code false}
2097     */
2098     public boolean completeExceptionally(Throwable ex) {
2099     if (ex == null) throw new NullPointerException();
2100     boolean triggered = internalComplete(new AltResult(ex));
2101     postComplete();
2102     return triggered;
2103     }
2104    
2105     public <U> CompletableFuture<U> thenApply(
2106     Function<? super T,? extends U> fn) {
2107     return uniApplyStage(null, fn);
2108     }
2109    
2110     public <U> CompletableFuture<U> thenApplyAsync(
2111     Function<? super T,? extends U> fn) {
2112     return uniApplyStage(defaultExecutor(), fn);
2113     }
2114    
2115     public <U> CompletableFuture<U> thenApplyAsync(
2116     Function<? super T,? extends U> fn, Executor executor) {
2117     return uniApplyStage(screenExecutor(executor), fn);
2118     }
2119    
2120     public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
2121     return uniAcceptStage(null, action);
2122     }
2123    
2124     public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
2125     return uniAcceptStage(defaultExecutor(), action);
2126     }
2127    
2128     public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
2129     Executor executor) {
2130     return uniAcceptStage(screenExecutor(executor), action);
2131     }
2132    
2133     public CompletableFuture<Void> thenRun(Runnable action) {
2134     return uniRunStage(null, action);
2135     }
2136    
2137     public CompletableFuture<Void> thenRunAsync(Runnable action) {
2138     return uniRunStage(defaultExecutor(), action);
2139     }
2140    
2141     public CompletableFuture<Void> thenRunAsync(Runnable action,
2142     Executor executor) {
2143     return uniRunStage(screenExecutor(executor), action);
2144     }
2145    
2146     public <U,V> CompletableFuture<V> thenCombine(
2147     CompletionStage<? extends U> other,
2148     BiFunction<? super T,? super U,? extends V> fn) {
2149     return biApplyStage(null, other, fn);
2150     }
2151    
2152     public <U,V> CompletableFuture<V> thenCombineAsync(
2153     CompletionStage<? extends U> other,
2154     BiFunction<? super T,? super U,? extends V> fn) {
2155     return biApplyStage(defaultExecutor(), other, fn);
2156     }
2157    
2158     public <U,V> CompletableFuture<V> thenCombineAsync(
2159     CompletionStage<? extends U> other,
2160     BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
2161     return biApplyStage(screenExecutor(executor), other, fn);
2162     }
2163    
2164     public <U> CompletableFuture<Void> thenAcceptBoth(
2165     CompletionStage<? extends U> other,
2166     BiConsumer<? super T, ? super U> action) {
2167     return biAcceptStage(null, other, action);
2168     }
2169    
2170     public <U> CompletableFuture<Void> thenAcceptBothAsync(
2171     CompletionStage<? extends U> other,
2172     BiConsumer<? super T, ? super U> action) {
2173     return biAcceptStage(defaultExecutor(), other, action);
2174     }
2175    
2176     public <U> CompletableFuture<Void> thenAcceptBothAsync(
2177     CompletionStage<? extends U> other,
2178     BiConsumer<? super T, ? super U> action, Executor executor) {
2179     return biAcceptStage(screenExecutor(executor), other, action);
2180     }
2181    
2182     public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
2183     Runnable action) {
2184     return biRunStage(null, other, action);
2185     }
2186    
2187     public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
2188     Runnable action) {
2189     return biRunStage(defaultExecutor(), other, action);
2190     }
2191    
2192     public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
2193     Runnable action,
2194     Executor executor) {
2195     return biRunStage(screenExecutor(executor), other, action);
2196     }
2197    
2198     public <U> CompletableFuture<U> applyToEither(
2199     CompletionStage<? extends T> other, Function<? super T, U> fn) {
2200     return orApplyStage(null, other, fn);
2201     }
2202    
2203     public <U> CompletableFuture<U> applyToEitherAsync(
2204     CompletionStage<? extends T> other, Function<? super T, U> fn) {
2205     return orApplyStage(defaultExecutor(), other, fn);
2206     }
2207    
2208     public <U> CompletableFuture<U> applyToEitherAsync(
2209     CompletionStage<? extends T> other, Function<? super T, U> fn,
2210     Executor executor) {
2211     return orApplyStage(screenExecutor(executor), other, fn);
2212     }
2213    
2214     public CompletableFuture<Void> acceptEither(
2215     CompletionStage<? extends T> other, Consumer<? super T> action) {
2216     return orAcceptStage(null, other, action);
2217     }
2218    
2219     public CompletableFuture<Void> acceptEitherAsync(
2220     CompletionStage<? extends T> other, Consumer<? super T> action) {
2221     return orAcceptStage(defaultExecutor(), other, action);
2222     }
2223    
2224     public CompletableFuture<Void> acceptEitherAsync(
2225     CompletionStage<? extends T> other, Consumer<? super T> action,
2226     Executor executor) {
2227     return orAcceptStage(screenExecutor(executor), other, action);
2228     }
2229    
2230     public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
2231     Runnable action) {
2232     return orRunStage(null, other, action);
2233     }
2234    
2235     public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
2236     Runnable action) {
2237     return orRunStage(defaultExecutor(), other, action);
2238     }
2239    
2240     public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
2241     Runnable action,
2242     Executor executor) {
2243     return orRunStage(screenExecutor(executor), other, action);
2244     }
2245    
2246     public <U> CompletableFuture<U> thenCompose(
2247     Function<? super T, ? extends CompletionStage<U>> fn) {
2248     return uniComposeStage(null, fn);
2249     }
2250    
2251     public <U> CompletableFuture<U> thenComposeAsync(
2252     Function<? super T, ? extends CompletionStage<U>> fn) {
2253     return uniComposeStage(defaultExecutor(), fn);
2254     }
2255    
2256     public <U> CompletableFuture<U> thenComposeAsync(
2257     Function<? super T, ? extends CompletionStage<U>> fn,
2258     Executor executor) {
2259     return uniComposeStage(screenExecutor(executor), fn);
2260     }
2261    
2262     public CompletableFuture<T> whenComplete(
2263     BiConsumer<? super T, ? super Throwable> action) {
2264     return uniWhenCompleteStage(null, action);
2265     }
2266    
2267     public CompletableFuture<T> whenCompleteAsync(
2268     BiConsumer<? super T, ? super Throwable> action) {
2269     return uniWhenCompleteStage(defaultExecutor(), action);
2270     }
2271    
2272     public CompletableFuture<T> whenCompleteAsync(
2273     BiConsumer<? super T, ? super Throwable> action, Executor executor) {
2274     return uniWhenCompleteStage(screenExecutor(executor), action);
2275     }
2276    
2277     public <U> CompletableFuture<U> handle(
2278     BiFunction<? super T, Throwable, ? extends U> fn) {
2279     return uniHandleStage(null, fn);
2280     }
2281    
2282     public <U> CompletableFuture<U> handleAsync(
2283     BiFunction<? super T, Throwable, ? extends U> fn) {
2284     return uniHandleStage(defaultExecutor(), fn);
2285     }
2286    
2287     public <U> CompletableFuture<U> handleAsync(
2288     BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
2289     return uniHandleStage(screenExecutor(executor), fn);
2290     }
2291    
2292     /**
2293     * Returns this CompletableFuture.
2294     *
2295     * @return this CompletableFuture
2296     */
2297     public CompletableFuture<T> toCompletableFuture() {
2298     return this;
2299     }
2300    
2301     // not in interface CompletionStage
2302    
2303     /**
2304     * Returns a new CompletableFuture that is completed when this
2305     * CompletableFuture completes, with the result of the given
2306     * function of the exception triggering this CompletableFuture's
2307     * completion when it completes exceptionally; otherwise, if this
2308     * CompletableFuture completes normally, then the returned
2309     * CompletableFuture also completes normally with the same value.
2310     * Note: More flexible versions of this functionality are
2311     * available using methods {@code whenComplete} and {@code handle}.
2312     *
2313     * @param fn the function to use to compute the value of the
2314     * returned CompletableFuture if this CompletableFuture completed
2315     * exceptionally
2316     * @return the new CompletableFuture
2317     */
2318     public CompletableFuture<T> exceptionally(
2319     Function<Throwable, ? extends T> fn) {
2320     return uniExceptionallyStage(fn);
2321     }
2322    
2323    
2324     /* ------------- Arbitrary-arity constructions -------------- */
2325    
2326     /**
2327     * Returns a new CompletableFuture that is completed when all of
2328     * the given CompletableFutures complete. If any of the given
2329     * CompletableFutures complete exceptionally, then the returned
2330     * CompletableFuture also does so, with a CompletionException
2331     * holding this exception as its cause. Otherwise, the results,
2332     * if any, of the given CompletableFutures are not reflected in
2333     * the returned CompletableFuture, but may be obtained by
2334     * inspecting them individually. If no CompletableFutures are
2335     * provided, returns a CompletableFuture completed with the value
2336     * {@code null}.
2337     *
2338     * <p>Among the applications of this method is to await completion
2339     * of a set of independent CompletableFutures before continuing a
2340     * program, as in: {@code CompletableFuture.allOf(c1, c2,
2341     * c3).join();}.
2342     *
2343     * @param cfs the CompletableFutures
2344     * @return a new CompletableFuture that is completed when all of the
2345     * given CompletableFutures complete
2346     * @throws NullPointerException if the array or any of its elements are
2347     * {@code null}
2348     */
2349     public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
2350     return andTree(cfs, 0, cfs.length - 1);
2351     }
2352    
2353     /**
2354     * Returns a new CompletableFuture that is completed when any of
2355     * the given CompletableFutures complete, with the same result.
2356     * Otherwise, if it completed exceptionally, the returned
2357     * CompletableFuture also does so, with a CompletionException
2358     * holding this exception as its cause. If no CompletableFutures
2359     * are provided, returns an incomplete CompletableFuture.
2360     *
2361     * @param cfs the CompletableFutures
2362     * @return a new CompletableFuture that is completed with the
2363     * result or exception of any of the given CompletableFutures when
2364     * one completes
2365     * @throws NullPointerException if the array or any of its elements are
2366     * {@code null}
2367     */
2368     public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
2369     return orTree(cfs, 0, cfs.length - 1);
2370     }
2371    
2372     /* ------------- Control and status methods -------------- */
2373    
2374     /**
2375     * If not already completed, completes this CompletableFuture with
2376     * a {@link CancellationException}. Dependent CompletableFutures
2377     * that have not already completed will also complete
2378     * exceptionally, with a {@link CompletionException} caused by
2379     * this {@code CancellationException}.
2380     *
2381     * @param mayInterruptIfRunning this value has no effect in this
2382     * implementation because interrupts are not used to control
2383     * processing.
2384     *
2385     * @return {@code true} if this task is now cancelled
2386     */
2387     public boolean cancel(boolean mayInterruptIfRunning) {
2388     boolean cancelled = (result == null) &&
2389     internalComplete(new AltResult(new CancellationException()));
2390     postComplete();
2391     return cancelled || isCancelled();
2392     }
2393    
2394     /**
2395     * Returns {@code true} if this CompletableFuture was cancelled
2396     * before it completed normally.
2397     *
2398     * @return {@code true} if this CompletableFuture was cancelled
2399     * before it completed normally
2400     */
2401     public boolean isCancelled() {
2402     Object r;
2403     return ((r = result) instanceof AltResult) &&
2404     (((AltResult)r).ex instanceof CancellationException);
2405     }
2406    
2407     /**
2408     * Returns {@code true} if this CompletableFuture completed
2409     * exceptionally, in any way. Possible causes include
2410     * cancellation, explicit invocation of {@code
2411     * completeExceptionally}, and abrupt termination of a
2412     * CompletionStage action.
2413     *
2414     * @return {@code true} if this CompletableFuture completed
2415     * exceptionally
2416     */
2417     public boolean isCompletedExceptionally() {
2418     Object r;
2419     return ((r = result) instanceof AltResult) && r != NIL;
2420     }
2421    
2422     /**
2423     * Forcibly sets or resets the value subsequently returned by
2424     * method {@link #get()} and related methods, whether or not
2425     * already completed. This method is designed for use only in
2426     * error recovery actions, and even in such situations may result
2427     * in ongoing dependent completions using established versus
2428     * overwritten outcomes.
2429     *
2430     * @param value the completion value
2431     */
2432     public void obtrudeValue(T value) {
2433     result = (value == null) ? NIL : value;
2434     postComplete();
2435     }
2436    
2437     /**
2438     * Forcibly causes subsequent invocations of method {@link #get()}
2439     * and related methods to throw the given exception, whether or
2440     * not already completed. This method is designed for use only in
2441     * error recovery actions, and even in such situations may result
2442     * in ongoing dependent completions using established versus
2443     * overwritten outcomes.
2444     *
2445     * @param ex the exception
2446     * @throws NullPointerException if the exception is null
2447     */
2448     public void obtrudeException(Throwable ex) {
2449     if (ex == null) throw new NullPointerException();
2450     result = new AltResult(ex);
2451     postComplete();
2452     }
2453    
2454     /**
2455     * Returns the estimated number of CompletableFutures whose
2456     * completions are awaiting completion of this CompletableFuture.
2457     * This method is designed for use in monitoring system state, not
2458     * for synchronization control.
2459     *
2460     * @return the number of dependent CompletableFutures
2461     */
2462     public int getNumberOfDependents() {
2463     int count = 0;
2464     for (Completion p = stack; p != null; p = p.next)
2465     ++count;
2466     return count;
2467     }
2468    
2469     /**
2470     * Returns a string identifying this CompletableFuture, as well as
2471     * its completion state. The state, in brackets, contains the
2472     * String {@code "Completed Normally"} or the String {@code
2473     * "Completed Exceptionally"}, or the String {@code "Not
2474     * completed"} followed by the number of CompletableFutures
2475     * dependent upon its completion, if any.
2476     *
2477     * @return a string identifying this CompletableFuture, as well as its state
2478     */
2479     public String toString() {
2480     Object r = result;
2481     int count = 0; // avoid call to getNumberOfDependents in case disabled
2482     for (Completion p = stack; p != null; p = p.next)
2483     ++count;
2484     return super.toString() +
2485     ((r == null) ?
2486     ((count == 0) ?
2487     "[Not completed]" :
2488     "[Not completed, " + count + " dependents]") :
2489     (((r instanceof AltResult) && ((AltResult)r).ex != null) ?
2490     "[Completed exceptionally]" :
2491     "[Completed normally]"));
2492     }
2493    
2494     // jdk9 additions
2495    
2496     /**
2497     * Returns a new incomplete CompletableFuture of the type to be
2498     * returned by a CompletionStage method. Subclasses should
2499     * normally override this method to return an instance of the same
2500     * class as this CompletableFuture. The default implementation
2501     * returns an instance of class CompletableFuture.
2502     *
2503     * @param <U> the type of the value
2504     * @return a new CompletableFuture
2505     * @since 9
2506     */
2507     public <U> CompletableFuture<U> newIncompleteFuture() {
2508     return new CompletableFuture<U>();
2509     }
2510    
2511     /**
2512     * Returns the default Executor used for async methods that do not
2513     * specify an Executor. This class uses the {@link
2514     * ForkJoinPool#commonPool()} if it supports more than one
2515     * parallel thread, or else an Executor using one thread per async
2516     * task. This method may be overridden in subclasses to return
2517     * an Executor that provides at least one independent thread.
2518     *
2519     * @return the executor
2520     * @since 9
2521     */
2522     public Executor defaultExecutor() {
2523     return ASYNC_POOL;
2524     }
2525    
2526     /**
2527     * Returns a new CompletableFuture that is completed normally with
2528     * the same value as this CompletableFuture when it completes
2529     * normally. If this CompletableFuture completes exceptionally,
2530     * then the returned CompletableFuture completes exceptionally
2531     * with a CompletionException with this exception as cause. The
2532     * behavior is equivalent to {@code thenApply(x -> x)}. This
2533     * method may be useful as a form of "defensive copying", to
2534     * prevent clients from completing, while still being able to
2535     * arrange dependent actions.
2536     *
2537     * @return the new CompletableFuture
2538     * @since 9
2539     */
2540     public CompletableFuture<T> copy() {
2541     return uniCopyStage();
2542     }
2543    
2544     /**
2545     * Returns a new CompletionStage that is completed normally with
2546     * the same value as this CompletableFuture when it completes
2547     * normally, and cannot be independently completed or otherwise
2548     * used in ways not defined by the methods of interface {@link
2549     * CompletionStage}. If this CompletableFuture completes
2550     * exceptionally, then the returned CompletionStage completes
2551     * exceptionally with a CompletionException with this exception as
2552     * cause.
2553     *
2554     * @return the new CompletionStage
2555     * @since 9
2556     */
2557     public CompletionStage<T> minimalCompletionStage() {
2558     return uniAsMinimalStage();
2559     }
2560    
2561     /**
2562     * Completes this CompletableFuture with the result of
2563     * the given Supplier function invoked from an asynchronous
2564     * task using the given executor.
2565     *
2566     * @param supplier a function returning the value to be used
2567     * to complete this CompletableFuture
2568     * @param executor the executor to use for asynchronous execution
2569     * @return this CompletableFuture
2570     * @since 9
2571     */
2572     public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier,
2573     Executor executor) {
2574     if (supplier == null || executor == null)
2575     throw new NullPointerException();
2576     executor.execute(new AsyncSupply<T>(this, supplier));
2577     return this;
2578     }
2579    
2580     /**
2581     * Completes this CompletableFuture with the result of the given
2582     * Supplier function invoked from an asynchronous task using the
2583     * default executor.
2584     *
2585     * @param supplier a function returning the value to be used
2586     * to complete this CompletableFuture
2587     * @return this CompletableFuture
2588     * @since 9
2589     */
2590     public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier) {
2591     return completeAsync(supplier, defaultExecutor());
2592     }
2593    
2594     /**
2595     * Exceptionally completes this CompletableFuture with
2596     * a {@link TimeoutException} if not otherwise completed
2597     * before the given timeout.
2598     *
2599     * @param timeout how long to wait before completing exceptionally
2600     * with a TimeoutException, in units of {@code unit}
2601     * @param unit a {@code TimeUnit} determining how to interpret the
2602     * {@code timeout} parameter
2603     * @return this CompletableFuture
2604     * @since 9
2605     */
2606     public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit) {
2607     if (unit == null)
2608     throw new NullPointerException();
2609     if (result == null)
2610     whenComplete(new Canceller(Delayer.delay(new Timeout(this),
2611     timeout, unit)));
2612     return this;
2613     }
2614    
2615     /**
2616     * Completes this CompletableFuture with the given value if not
2617     * otherwise completed before the given timeout.
2618     *
2619     * @param value the value to use upon timeout
2620     * @param timeout how long to wait before completing normally
2621     * with the given value, in units of {@code unit}
2622     * @param unit a {@code TimeUnit} determining how to interpret the
2623     * {@code timeout} parameter
2624     * @return this CompletableFuture
2625     * @since 9
2626     */
2627     public CompletableFuture<T> completeOnTimeout(T value, long timeout,
2628     TimeUnit unit) {
2629     if (unit == null)
2630     throw new NullPointerException();
2631     if (result == null)
2632     whenComplete(new Canceller(Delayer.delay(
2633     new DelayedCompleter<T>(this, value),
2634     timeout, unit)));
2635     return this;
2636     }
2637    
2638     /**
2639     * Returns a new Executor that submits a task to the given base
2640     * executor after the given delay (or no delay if non-positive).
2641     * Each delay commences upon invocation of the returned executor's
2642     * {@code execute} method.
2643     *
2644     * @param delay how long to delay, in units of {@code unit}
2645     * @param unit a {@code TimeUnit} determining how to interpret the
2646     * {@code delay} parameter
2647     * @param executor the base executor
2648     * @return the new delayed executor
2649     * @since 9
2650     */
2651     public static Executor delayedExecutor(long delay, TimeUnit unit,
2652     Executor executor) {
2653     if (unit == null || executor == null)
2654     throw new NullPointerException();
2655     return new DelayedExecutor(delay, unit, executor);
2656     }
2657    
2658     /**
2659     * Returns a new Executor that submits a task to the default
2660     * executor after the given delay (or no delay if non-positive).
2661     * Each delay commences upon invocation of the returned executor's
2662     * {@code execute} method.
2663     *
2664     * @param delay how long to delay, in units of {@code unit}
2665     * @param unit a {@code TimeUnit} determining how to interpret the
2666     * {@code delay} parameter
2667     * @return the new delayed executor
2668     * @since 9
2669     */
2670     public static Executor delayedExecutor(long delay, TimeUnit unit) {
2671     if (unit == null)
2672     throw new NullPointerException();
2673     return new DelayedExecutor(delay, unit, ASYNC_POOL);
2674     }
2675    
2676     /**
2677     * Returns a new CompletionStage that is already completed with
2678     * the given value and supports only those methods in
2679     * interface {@link CompletionStage}.
2680     *
2681     * @param value the value
2682     * @param <U> the type of the value
2683     * @return the completed CompletionStage
2684     * @since 9
2685     */
2686     public static <U> CompletionStage<U> completedStage(U value) {
2687     return new MinimalStage<U>((value == null) ? NIL : value);
2688     }
2689    
2690     /**
2691     * Returns a new CompletableFuture that is already completed
2692     * exceptionally with the given exception.
2693     *
2694     * @param ex the exception
2695     * @param <U> the type of the value
2696     * @return the exceptionally completed CompletableFuture
2697     * @since 9
2698     */
2699     public static <U> CompletableFuture<U> failedFuture(Throwable ex) {
2700     if (ex == null) throw new NullPointerException();
2701     return new CompletableFuture<U>(new AltResult(ex));
2702     }
2703    
2704     /**
2705     * Returns a new CompletionStage that is already completed
2706     * exceptionally with the given exception and supports only those
2707     * methods in interface {@link CompletionStage}.
2708     *
2709     * @param ex the exception
2710     * @param <U> the type of the value
2711     * @return the exceptionally completed CompletionStage
2712     * @since 9
2713     */
2714     public static <U> CompletionStage<U> failedStage(Throwable ex) {
2715     if (ex == null) throw new NullPointerException();
2716     return new MinimalStage<U>(new AltResult(ex));
2717     }
2718    
2719     /**
2720     * Singleton delay scheduler, used only for starting and
2721     * cancelling tasks.
2722     */
2723     static final class Delayer {
2724     static ScheduledFuture<?> delay(Runnable command, long delay,
2725     TimeUnit unit) {
2726     return delayer.schedule(command, delay, unit);
2727     }
2728    
2729     static final class DaemonThreadFactory implements ThreadFactory {
2730     public Thread newThread(Runnable r) {
2731     Thread t = new Thread(r);
2732     t.setDaemon(true);
2733     t.setName("CompletableFutureDelayScheduler");
2734     return t;
2735     }
2736     }
2737    
2738     static final ScheduledThreadPoolExecutor delayer;
2739     static {
2740     (delayer = new ScheduledThreadPoolExecutor(
2741     1, new DaemonThreadFactory())).
2742     setRemoveOnCancelPolicy(true);
2743     }
2744     }
2745    
2746     // Little class-ified lambdas to better support monitoring
2747    
2748     static final class DelayedExecutor implements Executor {
2749     final long delay;
2750     final TimeUnit unit;
2751     final Executor executor;
2752     DelayedExecutor(long delay, TimeUnit unit, Executor executor) {
2753     this.delay = delay; this.unit = unit; this.executor = executor;
2754     }
2755     public void execute(Runnable r) {
2756     Delayer.delay(new TaskSubmitter(executor, r), delay, unit);
2757     }
2758     }
2759    
2760     /** Action to submit user task */
2761     static final class TaskSubmitter implements Runnable {
2762     final Executor executor;
2763     final Runnable action;
2764     TaskSubmitter(Executor executor, Runnable action) {
2765     this.executor = executor;
2766     this.action = action;
2767     }
2768     public void run() { executor.execute(action); }
2769     }
2770    
2771     /** Action to completeExceptionally on timeout */
2772     static final class Timeout implements Runnable {
2773     final CompletableFuture<?> f;
2774     Timeout(CompletableFuture<?> f) { this.f = f; }
2775     public void run() {
2776     if (f != null && !f.isDone())
2777     f.completeExceptionally(new TimeoutException());
2778     }
2779     }
2780    
2781     /** Action to complete on timeout */
2782     static final class DelayedCompleter<U> implements Runnable {
2783     final CompletableFuture<U> f;
2784     final U u;
2785     DelayedCompleter(CompletableFuture<U> f, U u) { this.f = f; this.u = u; }
2786     public void run() {
2787     if (f != null)
2788     f.complete(u);
2789     }
2790     }
2791    
2792     /** Action to cancel unneeded timeouts */
2793     static final class Canceller implements BiConsumer<Object, Throwable> {
2794     final Future<?> f;
2795     Canceller(Future<?> f) { this.f = f; }
2796     public void accept(Object ignore, Throwable ex) {
2797     if (ex == null && f != null && !f.isDone())
2798     f.cancel(false);
2799     }
2800     }
2801    
2802     /**
2803     * A subclass that just throws UOE for most non-CompletionStage methods.
2804     */
2805     static final class MinimalStage<T> extends CompletableFuture<T> {
2806     MinimalStage() { }
2807     MinimalStage(Object r) { super(r); }
2808     @Override public <U> CompletableFuture<U> newIncompleteFuture() {
2809     return new MinimalStage<U>(); }
2810     @Override public T get() {
2811     throw new UnsupportedOperationException(); }
2812     @Override public T get(long timeout, TimeUnit unit) {
2813     throw new UnsupportedOperationException(); }
2814     @Override public T getNow(T valueIfAbsent) {
2815     throw new UnsupportedOperationException(); }
2816     @Override public T join() {
2817     throw new UnsupportedOperationException(); }
2818     @Override public boolean complete(T value) {
2819     throw new UnsupportedOperationException(); }
2820     @Override public boolean completeExceptionally(Throwable ex) {
2821     throw new UnsupportedOperationException(); }
2822     @Override public boolean cancel(boolean mayInterruptIfRunning) {
2823     throw new UnsupportedOperationException(); }
2824     @Override public void obtrudeValue(T value) {
2825     throw new UnsupportedOperationException(); }
2826     @Override public void obtrudeException(Throwable ex) {
2827     throw new UnsupportedOperationException(); }
2828     @Override public boolean isDone() {
2829     throw new UnsupportedOperationException(); }
2830     @Override public boolean isCancelled() {
2831     throw new UnsupportedOperationException(); }
2832     @Override public boolean isCompletedExceptionally() {
2833     throw new UnsupportedOperationException(); }
2834     @Override public int getNumberOfDependents() {
2835     throw new UnsupportedOperationException(); }
2836     @Override public CompletableFuture<T> completeAsync
2837     (Supplier<? extends T> supplier, Executor executor) {
2838     throw new UnsupportedOperationException(); }
2839     @Override public CompletableFuture<T> completeAsync
2840     (Supplier<? extends T> supplier) {
2841     throw new UnsupportedOperationException(); }
2842     @Override public CompletableFuture<T> orTimeout
2843     (long timeout, TimeUnit unit) {
2844     throw new UnsupportedOperationException(); }
2845     @Override public CompletableFuture<T> completeOnTimeout
2846     (T value, long timeout, TimeUnit unit) {
2847     throw new UnsupportedOperationException(); }
2848     }
2849    
2850     // Unsafe mechanics
2851     private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
2852     private static final long RESULT;
2853     private static final long STACK;
2854     private static final long NEXT;
2855     static {
2856     try {
2857     RESULT = U.objectFieldOffset
2858     (CompletableFuture.class.getDeclaredField("result"));
2859     STACK = U.objectFieldOffset
2860     (CompletableFuture.class.getDeclaredField("stack"));
2861     NEXT = U.objectFieldOffset
2862     (Completion.class.getDeclaredField("next"));
2863     } catch (ReflectiveOperationException e) {
2864     throw new Error(e);
2865     }
2866    
2867     // Reduce the risk of rare disastrous classloading in first call to
2868     // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
2869     Class<?> ensureLoaded = LockSupport.class;
2870     }
2871     }