ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jdk8/java/util/concurrent/CompletableFuture.java
Revision: 1.10
Committed: Mon Oct 1 03:58:18 2018 UTC (5 years, 7 months ago) by jsr166
Branch: MAIN
CVS Tags: HEAD
Changes since 1.9: +112 -30 lines
Log Message:
backport exceptionally* methods to fix 4jdk8-tck target

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