ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/CompletableFuture.java
Revision: 1.200
Committed: Sat Jun 25 15:11:11 2016 UTC (7 years, 11 months ago) by jsr166
Branch: MAIN
Changes since 1.199: +229 -234 lines
Log Message:
improve efficiency of Uni/Or completions when immediately complete

File Contents

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