ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.219
Committed: Mon Sep 24 01:23:53 2018 UTC (5 years, 8 months ago) by jsr166
Branch: MAIN
Changes since 1.218: +17 -26 lines
Log Message:
uniCompose methods should not invoke Executor.execute unnecessarily

File Contents

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