ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.185
Committed: Sun Apr 3 14:42:04 2016 UTC (8 years, 2 months ago) by dl
Branch: MAIN
Changes since 1.184: +45 -38 lines
Log Message:
Faster cleanup

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