ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.192
Committed: Thu Jun 2 13:16:27 2016 UTC (8 years ago) by dl
Branch: MAIN
Changes since 1.191: +27 -45 lines
Log Message:
VarHandles conversion; pass 1

File Contents

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