/* * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * This code is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License version 2 only, as * published by the Free Software Foundation. Oracle designates this * particular file as subject to the "Classpath" exception as provided * by Oracle in the LICENSE file that accompanied this code. * * This code is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * version 2 for more details (a copy is included in the LICENSE file that * accompanied this code). * * You should have received a copy of the GNU General Public License version * 2 along with this work; if not, write to the Free Software Foundation, * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. * * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA * or visit www.oracle.com if you need additional information or have any * questions. */ /* * This file is available under and governed by the GNU General Public * License version 2 only, as published by the Free Software Foundation. * However, the following notice accompanied the original version of this * file: * * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at * http://creativecommons.org/publicdomain/zero/1.0/ */ package java.util.concurrent; import java.util.concurrent.locks.LockSupport; import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; /** * A {@link Future} that may be explicitly completed (setting its * value and status), and may be used as a {@link CompletionStage}, * supporting dependent functions and actions that trigger upon its * completion. * *

When two or more threads attempt to * {@link #complete complete}, * {@link #completeExceptionally completeExceptionally}, or * {@link #cancel cancel} * a CompletableFuture, only one of them succeeds. * *

In addition to these and related methods for directly * manipulating status and results, CompletableFuture implements * interface {@link CompletionStage} with the following policies:

* *

CompletableFuture also implements {@link Future} with the following * policies:

* *

Arguments used to pass a completion result (that is, for * parameters of type {@code T}) for methods accepting them may be * null, but passing a null value for any other parameter will result * in a {@link NullPointerException} being thrown. * *

Subclasses of this class should normally override the "virtual * constructor" method {@link #newIncompleteFuture}, which establishes * the concrete type returned by CompletionStage methods. For example, * here is a class that substitutes a different default Executor and * disables the {@code obtrude} methods: * *

 {@code
 * class MyCompletableFuture extends CompletableFuture {
 *   static final Executor myExecutor = ...;
 *   public MyCompletableFuture() { }
 *   public  CompletableFuture newIncompleteFuture() {
 *     return new MyCompletableFuture(); }
 *   public Executor defaultExecutor() {
 *     return myExecutor; }
 *   public void obtrudeValue(T value) {
 *     throw new UnsupportedOperationException(); }
 *   public void obtrudeException(Throwable ex) {
 *     throw new UnsupportedOperationException(); }
 * }}
* * @author Doug Lea * @since 1.8 * @param The result type returned by this future's {@code join} * and {@code get} methods */ public class CompletableFuture implements Future, CompletionStage { /* * Overview: * * A CompletableFuture may have dependent completion actions, * collected in a linked stack. It atomically completes by CASing * a result field, and then pops off and runs those actions. This * applies across normal vs exceptional outcomes, sync vs async * actions, binary triggers, and various forms of completions. * * Non-nullness of field result (set via CAS) indicates done. An * AltResult is used to box null as a result, as well as to hold * exceptions. Using a single field makes completion simple to * detect and trigger. Encoding and decoding is straightforward * but adds to the sprawl of trapping and associating exceptions * with targets. Minor simplifications rely on (static) NIL (to * box null results) being the only AltResult with a null * exception field, so we don't usually need explicit comparisons. * Even though some of the generics casts are unchecked (see * SuppressWarnings annotations), they are placed to be * appropriate even if checked. * * Dependent actions are represented by Completion objects linked * as Treiber stacks headed by field "stack". There are Completion * classes for each kind of action, grouped into single-input * (UniCompletion), two-input (BiCompletion), projected * (BiCompletions using either (not both) of two inputs), shared * (CoCompletion, used by the second of two sources), zero-input * source actions, and Signallers that unblock waiters. Class * Completion extends ForkJoinTask to enable async execution * (adding no space overhead because we exploit its "tag" methods * to maintain claims). It is also declared as Runnable to allow * usage with arbitrary executors. * * Support for each kind of CompletionStage relies on a separate * class, along with two CompletableFuture methods: * * * A Completion class with name X corresponding to function, * prefaced with "Uni", "Bi", or "Or". Each class contains * fields for source(s), actions, and dependent. They are * boringly similar, differing from others only with respect to * underlying functional forms. We do this so that users don't * encounter layers of adapters in common usages. * * * Boolean CompletableFuture method x(...) (for example * uniApply) takes all of the arguments needed to check that an * action is triggerable, and then either runs the action or * arranges its async execution by executing its Completion * argument, if present. The method returns true if known to be * complete. * * * Completion method tryFire(int mode) invokes the associated x * method with its held arguments, and on success cleans up. * The mode argument allows tryFire to be called twice (SYNC, * then ASYNC); the first to screen and trap exceptions while * arranging to execute, and the second when called from a * task. (A few classes are not used async so take slightly * different forms.) The claim() callback suppresses function * invocation if already claimed by another thread. * * * CompletableFuture method xStage(...) is called from a public * stage method of CompletableFuture x. It screens user * arguments and invokes and/or creates the stage object. If * not async and x is already complete, the action is run * immediately. Otherwise a Completion c is created, pushed to * x's stack (unless done), and started or triggered via * c.tryFire. This also covers races possible if x completes * while pushing. Classes with two inputs (for example BiApply) * deal with races across both while pushing actions. The * second completion is a CoCompletion pointing to the first, * shared so that at most one performs the action. The * multiple-arity methods allOf and anyOf do this pairwise to * form trees of completions. * * Note that the generic type parameters of methods vary according * to whether "this" is a source, dependent, or completion. * * Method postComplete is called upon completion unless the target * is guaranteed not to be observable (i.e., not yet returned or * linked). Multiple threads can call postComplete, which * atomically pops each dependent action, and tries to trigger it * via method tryFire, in NESTED mode. Triggering can propagate * recursively, so NESTED mode returns its completed dependent (if * one exists) for further processing by its caller (see method * postFire). * * Blocking methods get() and join() rely on Signaller Completions * that wake up waiting threads. The mechanics are similar to * Treiber stack wait-nodes used in FutureTask, Phaser, and * SynchronousQueue. See their internal documentation for * algorithmic details. * * Without precautions, CompletableFutures would be prone to * garbage accumulation as chains of Completions build up, each * pointing back to its sources. So we null out fields as soon as * possible (see especially method Completion.detach). The * screening checks needed anyway harmlessly ignore null arguments * that may have been obtained during races with threads nulling * out fields. We also try to unlink fired Completions from * stacks that might never be popped (see method postFire). * Completion fields need not be declared as final or volatile * because they are only visible to other threads upon safe * publication. */ volatile Object result; // Either the result or boxed AltResult volatile Completion stack; // Top of Treiber stack of dependent actions final boolean internalComplete(Object r) { // CAS from null to r return U.compareAndSwapObject(this, RESULT, null, r); } final boolean casStack(Completion cmp, Completion val) { return U.compareAndSwapObject(this, STACK, cmp, val); } /** Returns true if successfully pushed c onto stack. */ final boolean tryPushStack(Completion c) { Completion h = stack; lazySetNext(c, h); return U.compareAndSwapObject(this, STACK, h, c); } /** Unconditionally pushes c onto stack, retrying if necessary. */ final void pushStack(Completion c) { do {} while (!tryPushStack(c)); } /* ------------- Encoding and decoding outcomes -------------- */ static final class AltResult { // See above final Throwable ex; // null only for NIL AltResult(Throwable x) { this.ex = x; } } /** The encoding of the null value. */ static final AltResult NIL = new AltResult(null); /** Completes with the null value, unless already completed. */ final boolean completeNull() { return U.compareAndSwapObject(this, RESULT, null, NIL); } /** Returns the encoding of the given non-exceptional value. */ final Object encodeValue(T t) { return (t == null) ? NIL : t; } /** Completes with a non-exceptional result, unless already completed. */ final boolean completeValue(T t) { return U.compareAndSwapObject(this, RESULT, null, (t == null) ? NIL : t); } /** * Returns the encoding of the given (non-null) exception as a * wrapped CompletionException unless it is one already. */ static AltResult encodeThrowable(Throwable x) { return new AltResult((x instanceof CompletionException) ? x : new CompletionException(x)); } /** Completes with an exceptional result, unless already completed. */ final boolean completeThrowable(Throwable x) { return U.compareAndSwapObject(this, RESULT, null, encodeThrowable(x)); } /** * Returns the encoding of the given (non-null) exception as a * wrapped CompletionException unless it is one already. May * return the given Object r (which must have been the result of a * source future) if it is equivalent, i.e. if this is a simple * relay of an existing CompletionException. */ static Object encodeThrowable(Throwable x, Object r) { if (!(x instanceof CompletionException)) x = new CompletionException(x); else if (r instanceof AltResult && x == ((AltResult)r).ex) return r; return new AltResult(x); } /** * Completes with the given (non-null) exceptional result as a * wrapped CompletionException unless it is one already, unless * already completed. May complete with the given Object r * (which must have been the result of a source future) if it is * equivalent, i.e. if this is a simple propagation of an * existing CompletionException. */ final boolean completeThrowable(Throwable x, Object r) { return U.compareAndSwapObject(this, RESULT, null, encodeThrowable(x, r)); } /** * Returns the encoding of the given arguments: if the exception * is non-null, encodes as AltResult. Otherwise uses the given * value, boxed as NIL if null. */ Object encodeOutcome(T t, Throwable x) { return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x); } /** * Returns the encoding of a copied outcome; if exceptional, * rewraps as a CompletionException, else returns argument. */ static Object encodeRelay(Object r) { Throwable x; return (((r instanceof AltResult) && (x = ((AltResult)r).ex) != null && !(x instanceof CompletionException)) ? new AltResult(new CompletionException(x)) : r); } /** * Completes with r or a copy of r, unless already completed. * If exceptional, r is first coerced to a CompletionException. */ final boolean completeRelay(Object r) { return U.compareAndSwapObject(this, RESULT, null, encodeRelay(r)); } /** * Reports result using Future.get conventions. */ private static T reportGet(Object r) throws InterruptedException, ExecutionException { if (r == null) // by convention below, null means interrupted throw new InterruptedException(); if (r instanceof AltResult) { Throwable x, cause; if ((x = ((AltResult)r).ex) == null) return null; if (x instanceof CancellationException) throw (CancellationException)x; if ((x instanceof CompletionException) && (cause = x.getCause()) != null) x = cause; throw new ExecutionException(x); } @SuppressWarnings("unchecked") T t = (T) r; return t; } /** * Decodes outcome to return result or throw unchecked exception. */ private static T reportJoin(Object r) { if (r instanceof AltResult) { Throwable x; if ((x = ((AltResult)r).ex) == null) return null; if (x instanceof CancellationException) throw (CancellationException)x; if (x instanceof CompletionException) throw (CompletionException)x; throw new CompletionException(x); } @SuppressWarnings("unchecked") T t = (T) r; return t; } /* ------------- Async task preliminaries -------------- */ /** * A marker interface identifying asynchronous tasks produced by * {@code async} methods. This may be useful for monitoring, * debugging, and tracking asynchronous activities. * * @since 1.8 */ public static interface AsynchronousCompletionTask { } private static final boolean USE_COMMON_POOL = (ForkJoinPool.getCommonPoolParallelism() > 1); /** * Default executor -- ForkJoinPool.commonPool() unless it cannot * support parallelism. */ private static final Executor ASYNC_POOL = USE_COMMON_POOL ? ForkJoinPool.commonPool() : new ThreadPerTaskExecutor(); /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */ static final class ThreadPerTaskExecutor implements Executor { public void execute(Runnable r) { new Thread(r).start(); } } /** * Null-checks user executor argument, and translates uses of * commonPool to ASYNC_POOL in case parallelism disabled. */ static Executor screenExecutor(Executor e) { if (!USE_COMMON_POOL && e == ForkJoinPool.commonPool()) return ASYNC_POOL; if (e == null) throw new NullPointerException(); return e; } // Modes for Completion.tryFire. Signedness matters. static final int SYNC = 0; static final int ASYNC = 1; static final int NESTED = -1; /** * Spins before blocking in waitingGet */ static final int SPINS = (Runtime.getRuntime().availableProcessors() > 1 ? 1 << 8 : 0); /* ------------- Base Completion classes and operations -------------- */ @SuppressWarnings("serial") abstract static class Completion extends ForkJoinTask implements Runnable, AsynchronousCompletionTask { volatile Completion next; // Treiber stack link /** * Performs completion action if triggered, returning a * dependent that may need propagation, if one exists. * * @param mode SYNC, ASYNC, or NESTED */ abstract CompletableFuture tryFire(int mode); /** Returns true if possibly still triggerable. Used by cleanStack. */ abstract boolean isLive(); public final void run() { tryFire(ASYNC); } public final boolean exec() { tryFire(ASYNC); return false; } public final Void getRawResult() { return null; } public final void setRawResult(Void v) {} } static void lazySetNext(Completion c, Completion next) { U.putOrderedObject(c, NEXT, next); } /** * Pops and tries to trigger all reachable dependents. Call only * when known to be done. */ final void postComplete() { /* * On each step, variable f holds current dependents to pop * and run. It is extended along only one path at a time, * pushing others to avoid unbounded recursion. */ CompletableFuture f = this; Completion h; while ((h = f.stack) != null || (f != this && (h = (f = this).stack) != null)) { CompletableFuture d; Completion t; if (f.casStack(h, t = h.next)) { if (t != null) { if (f != this) { pushStack(h); continue; } h.next = null; // detach } f = (d = h.tryFire(NESTED)) == null ? this : d; } } } /** Traverses stack and unlinks dead Completions. */ final void cleanStack() { for (Completion p = null, q = stack; q != null;) { Completion s = q.next; if (q.isLive()) { p = q; q = s; } else if (p == null) { casStack(q, s); q = stack; } else { p.next = s; if (p.isLive()) q = s; else { p = null; // restart q = stack; } } } } /* ------------- One-input Completions -------------- */ /** A Completion with a source, dependent, and executor. */ @SuppressWarnings("serial") abstract static class UniCompletion extends Completion { Executor executor; // executor to use (null if none) CompletableFuture dep; // the dependent to complete CompletableFuture src; // source for action UniCompletion(Executor executor, CompletableFuture dep, CompletableFuture src) { this.executor = executor; this.dep = dep; this.src = src; } /** * Returns true if action can be run. Call only when known to * be triggerable. Uses FJ tag bit to ensure that only one * thread claims ownership. If async, starts as task -- a * later call to tryFire will run action. */ final boolean claim() { Executor e = executor; if (compareAndSetForkJoinTaskTag((short)0, (short)1)) { if (e == null) return true; executor = null; // disable e.execute(this); } return false; } final boolean isLive() { return dep != null; } } /** Pushes the given completion (if it exists) unless done. */ final void push(UniCompletion c) { if (c != null) { while (result == null && !tryPushStack(c)) lazySetNext(c, null); // clear on failure } } /** * Post-processing by dependent after successful UniCompletion * tryFire. Tries to clean stack of source a, and then either runs * postComplete or returns this to caller, depending on mode. */ final CompletableFuture postFire(CompletableFuture a, int mode) { if (a != null && a.stack != null) { if (mode < 0 || a.result == null) a.cleanStack(); else a.postComplete(); } if (result != null && stack != null) { if (mode < 0) return this; else postComplete(); } return null; } @SuppressWarnings("serial") static final class UniApply extends UniCompletion { Function fn; UniApply(Executor executor, CompletableFuture dep, CompletableFuture src, Function fn) { super(executor, dep, src); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; if ((d = dep) == null || !d.uniApply(a = src, fn, mode > 0 ? null : this)) return null; dep = null; src = null; fn = null; return d.postFire(a, mode); } } final boolean uniApply(CompletableFuture a, Function f, UniApply c) { Object r; Throwable x; if (a == null || (r = a.result) == null || f == null) return false; tryComplete: if (result == null) { if (r instanceof AltResult) { if ((x = ((AltResult)r).ex) != null) { completeThrowable(x, r); break tryComplete; } r = null; } try { if (c != null && !c.claim()) return false; @SuppressWarnings("unchecked") S s = (S) r; completeValue(f.apply(s)); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture uniApplyStage( Executor e, Function f) { if (f == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.uniApply(this, f, null)) { UniApply c = new UniApply(e, d, this, f); push(c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class UniAccept extends UniCompletion { Consumer fn; UniAccept(Executor executor, CompletableFuture dep, CompletableFuture src, Consumer fn) { super(executor, dep, src); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; if ((d = dep) == null || !d.uniAccept(a = src, fn, mode > 0 ? null : this)) return null; dep = null; src = null; fn = null; return d.postFire(a, mode); } } final boolean uniAccept(CompletableFuture a, Consumer f, UniAccept c) { Object r; Throwable x; if (a == null || (r = a.result) == null || f == null) return false; tryComplete: if (result == null) { if (r instanceof AltResult) { if ((x = ((AltResult)r).ex) != null) { completeThrowable(x, r); break tryComplete; } r = null; } try { if (c != null && !c.claim()) return false; @SuppressWarnings("unchecked") S s = (S) r; f.accept(s); completeNull(); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture uniAcceptStage(Executor e, Consumer f) { if (f == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.uniAccept(this, f, null)) { UniAccept c = new UniAccept(e, d, this, f); push(c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class UniRun extends UniCompletion { Runnable fn; UniRun(Executor executor, CompletableFuture dep, CompletableFuture src, Runnable fn) { super(executor, dep, src); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; if ((d = dep) == null || !d.uniRun(a = src, fn, mode > 0 ? null : this)) return null; dep = null; src = null; fn = null; return d.postFire(a, mode); } } final boolean uniRun(CompletableFuture a, Runnable f, UniRun c) { Object r; Throwable x; if (a == null || (r = a.result) == null || f == null) return false; if (result == null) { if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) completeThrowable(x, r); else try { if (c != null && !c.claim()) return false; f.run(); completeNull(); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture uniRunStage(Executor e, Runnable f) { if (f == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.uniRun(this, f, null)) { UniRun c = new UniRun(e, d, this, f); push(c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class UniWhenComplete extends UniCompletion { BiConsumer fn; UniWhenComplete(Executor executor, CompletableFuture dep, CompletableFuture src, BiConsumer fn) { super(executor, dep, src); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; if ((d = dep) == null || !d.uniWhenComplete(a = src, fn, mode > 0 ? null : this)) return null; dep = null; src = null; fn = null; return d.postFire(a, mode); } } final boolean uniWhenComplete(CompletableFuture a, BiConsumer f, UniWhenComplete c) { Object r; T t; Throwable x = null; if (a == null || (r = a.result) == null || f == null) return false; if (result == null) { try { if (c != null && !c.claim()) return false; if (r instanceof AltResult) { x = ((AltResult)r).ex; t = null; } else { @SuppressWarnings("unchecked") T tr = (T) r; t = tr; } f.accept(t, x); if (x == null) { internalComplete(r); return true; } } catch (Throwable ex) { if (x == null) x = ex; } completeThrowable(x, r); } return true; } private CompletableFuture uniWhenCompleteStage( Executor e, BiConsumer f) { if (f == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.uniWhenComplete(this, f, null)) { UniWhenComplete c = new UniWhenComplete(e, d, this, f); push(c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class UniHandle extends UniCompletion { BiFunction fn; UniHandle(Executor executor, CompletableFuture dep, CompletableFuture src, BiFunction fn) { super(executor, dep, src); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; if ((d = dep) == null || !d.uniHandle(a = src, fn, mode > 0 ? null : this)) return null; dep = null; src = null; fn = null; return d.postFire(a, mode); } } final boolean uniHandle(CompletableFuture a, BiFunction f, UniHandle c) { Object r; S s; Throwable x; if (a == null || (r = a.result) == null || f == null) return false; if (result == null) { try { if (c != null && !c.claim()) return false; if (r instanceof AltResult) { x = ((AltResult)r).ex; s = null; } else { x = null; @SuppressWarnings("unchecked") S ss = (S) r; s = ss; } completeValue(f.apply(s, x)); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture uniHandleStage( Executor e, BiFunction f) { if (f == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.uniHandle(this, f, null)) { UniHandle c = new UniHandle(e, d, this, f); push(c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class UniExceptionally extends UniCompletion { Function fn; UniExceptionally(CompletableFuture dep, CompletableFuture src, Function fn) { super(null, dep, src); this.fn = fn; } final CompletableFuture tryFire(int mode) { // never ASYNC // assert mode != ASYNC; CompletableFuture d; CompletableFuture a; if ((d = dep) == null || !d.uniExceptionally(a = src, fn, this)) return null; dep = null; src = null; fn = null; return d.postFire(a, mode); } } final boolean uniExceptionally(CompletableFuture a, Function f, UniExceptionally c) { Object r; Throwable x; if (a == null || (r = a.result) == null || f == null) return false; if (result == null) { try { if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) { if (c != null && !c.claim()) return false; completeValue(f.apply(x)); } else internalComplete(r); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture uniExceptionallyStage( Function f) { if (f == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (!d.uniExceptionally(this, f, null)) { UniExceptionally c = new UniExceptionally(d, this, f); push(c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class UniRelay extends UniCompletion { // for Compose UniRelay(CompletableFuture dep, CompletableFuture src) { super(null, dep, src); } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; if ((d = dep) == null || !d.uniRelay(a = src)) return null; src = null; dep = null; return d.postFire(a, mode); } } final boolean uniRelay(CompletableFuture a) { Object r; if (a == null || (r = a.result) == null) return false; if (result == null) // no need to claim completeRelay(r); return true; } private CompletableFuture uniCopyStage() { Object r; CompletableFuture d = newIncompleteFuture(); if ((r = result) != null) d.completeRelay(r); else { UniRelay c = new UniRelay(d, this); push(c); c.tryFire(SYNC); } return d; } private MinimalStage uniAsMinimalStage() { Object r; if ((r = result) != null) return new MinimalStage(encodeRelay(r)); MinimalStage d = new MinimalStage(); UniRelay c = new UniRelay(d, this); push(c); c.tryFire(SYNC); return d; } @SuppressWarnings("serial") static final class UniCompose extends UniCompletion { Function> fn; UniCompose(Executor executor, CompletableFuture dep, CompletableFuture src, Function> fn) { super(executor, dep, src); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; if ((d = dep) == null || !d.uniCompose(a = src, fn, mode > 0 ? null : this)) return null; dep = null; src = null; fn = null; return d.postFire(a, mode); } } final boolean uniCompose( CompletableFuture a, Function> f, UniCompose c) { Object r; Throwable x; if (a == null || (r = a.result) == null || f == null) return false; tryComplete: if (result == null) { if (r instanceof AltResult) { if ((x = ((AltResult)r).ex) != null) { completeThrowable(x, r); break tryComplete; } r = null; } try { if (c != null && !c.claim()) return false; @SuppressWarnings("unchecked") S s = (S) r; CompletableFuture g = f.apply(s).toCompletableFuture(); if (g.result == null || !uniRelay(g)) { UniRelay copy = new UniRelay(this, g); g.push(copy); copy.tryFire(SYNC); if (result == null) return false; } } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture uniComposeStage( Executor e, Function> f) { if (f == null) throw new NullPointerException(); Object r, s; Throwable x; CompletableFuture d = newIncompleteFuture(); if (e == null && (r = result) != null) { if (r instanceof AltResult) { if ((x = ((AltResult)r).ex) != null) { d.result = encodeThrowable(x, r); return d; } r = null; } try { @SuppressWarnings("unchecked") T t = (T) r; CompletableFuture g = f.apply(t).toCompletableFuture(); if ((s = g.result) != null) d.completeRelay(s); else { UniRelay c = new UniRelay(d, g); g.push(c); c.tryFire(SYNC); } return d; } catch (Throwable ex) { d.result = encodeThrowable(ex); return d; } } UniCompose c = new UniCompose(e, d, this, f); push(c); c.tryFire(SYNC); return d; } /* ------------- Two-input Completions -------------- */ /** A Completion for an action with two sources */ @SuppressWarnings("serial") abstract static class BiCompletion extends UniCompletion { CompletableFuture snd; // second source for action BiCompletion(Executor executor, CompletableFuture dep, CompletableFuture src, CompletableFuture snd) { super(executor, dep, src); this.snd = snd; } } /** A Completion delegating to a BiCompletion */ @SuppressWarnings("serial") static final class CoCompletion extends Completion { BiCompletion base; CoCompletion(BiCompletion base) { this.base = base; } final CompletableFuture tryFire(int mode) { BiCompletion c; CompletableFuture d; if ((c = base) == null || (d = c.tryFire(mode)) == null) return null; base = null; // detach return d; } final boolean isLive() { BiCompletion c; return (c = base) != null && c.dep != null; } } /** Pushes completion to this and b unless both done. */ final void bipush(CompletableFuture b, BiCompletion c) { if (c != null) { Object r; while ((r = result) == null && !tryPushStack(c)) lazySetNext(c, null); // clear on failure if (b != null && b != this && b.result == null) { Completion q = (r != null) ? c : new CoCompletion(c); while (b.result == null && !b.tryPushStack(q)) lazySetNext(q, null); // clear on failure } } } /** Post-processing after successful BiCompletion tryFire. */ final CompletableFuture postFire(CompletableFuture a, CompletableFuture b, int mode) { if (b != null && b.stack != null) { // clean second source if (mode < 0 || b.result == null) b.cleanStack(); else b.postComplete(); } return postFire(a, mode); } @SuppressWarnings("serial") static final class BiApply extends BiCompletion { BiFunction fn; BiApply(Executor executor, CompletableFuture dep, CompletableFuture src, CompletableFuture snd, BiFunction fn) { super(executor, dep, src, snd); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; CompletableFuture b; if ((d = dep) == null || !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this)) return null; dep = null; src = null; snd = null; fn = null; return d.postFire(a, b, mode); } } final boolean biApply(CompletableFuture a, CompletableFuture b, BiFunction f, BiApply c) { Object r, s; Throwable x; if (a == null || (r = a.result) == null || b == null || (s = b.result) == null || f == null) return false; tryComplete: if (result == null) { if (r instanceof AltResult) { if ((x = ((AltResult)r).ex) != null) { completeThrowable(x, r); break tryComplete; } r = null; } if (s instanceof AltResult) { if ((x = ((AltResult)s).ex) != null) { completeThrowable(x, s); break tryComplete; } s = null; } try { if (c != null && !c.claim()) return false; @SuppressWarnings("unchecked") R rr = (R) r; @SuppressWarnings("unchecked") S ss = (S) s; completeValue(f.apply(rr, ss)); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture biApplyStage( Executor e, CompletionStage o, BiFunction f) { CompletableFuture b; if (f == null || (b = o.toCompletableFuture()) == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.biApply(this, b, f, null)) { BiApply c = new BiApply(e, d, this, b, f); bipush(b, c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class BiAccept extends BiCompletion { BiConsumer fn; BiAccept(Executor executor, CompletableFuture dep, CompletableFuture src, CompletableFuture snd, BiConsumer fn) { super(executor, dep, src, snd); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; CompletableFuture b; if ((d = dep) == null || !d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this)) return null; dep = null; src = null; snd = null; fn = null; return d.postFire(a, b, mode); } } final boolean biAccept(CompletableFuture a, CompletableFuture b, BiConsumer f, BiAccept c) { Object r, s; Throwable x; if (a == null || (r = a.result) == null || b == null || (s = b.result) == null || f == null) return false; tryComplete: if (result == null) { if (r instanceof AltResult) { if ((x = ((AltResult)r).ex) != null) { completeThrowable(x, r); break tryComplete; } r = null; } if (s instanceof AltResult) { if ((x = ((AltResult)s).ex) != null) { completeThrowable(x, s); break tryComplete; } s = null; } try { if (c != null && !c.claim()) return false; @SuppressWarnings("unchecked") R rr = (R) r; @SuppressWarnings("unchecked") S ss = (S) s; f.accept(rr, ss); completeNull(); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture biAcceptStage( Executor e, CompletionStage o, BiConsumer f) { CompletableFuture b; if (f == null || (b = o.toCompletableFuture()) == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.biAccept(this, b, f, null)) { BiAccept c = new BiAccept(e, d, this, b, f); bipush(b, c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class BiRun extends BiCompletion { Runnable fn; BiRun(Executor executor, CompletableFuture dep, CompletableFuture src, CompletableFuture snd, Runnable fn) { super(executor, dep, src, snd); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; CompletableFuture b; if ((d = dep) == null || !d.biRun(a = src, b = snd, fn, mode > 0 ? null : this)) return null; dep = null; src = null; snd = null; fn = null; return d.postFire(a, b, mode); } } final boolean biRun(CompletableFuture a, CompletableFuture b, Runnable f, BiRun c) { Object r, s; Throwable x; if (a == null || (r = a.result) == null || b == null || (s = b.result) == null || f == null) return false; if (result == null) { if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) completeThrowable(x, r); else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null) completeThrowable(x, s); else try { if (c != null && !c.claim()) return false; f.run(); completeNull(); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture biRunStage(Executor e, CompletionStage o, Runnable f) { CompletableFuture b; if (f == null || (b = o.toCompletableFuture()) == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.biRun(this, b, f, null)) { BiRun c = new BiRun<>(e, d, this, b, f); bipush(b, c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class BiRelay extends BiCompletion { // for And BiRelay(CompletableFuture dep, CompletableFuture src, CompletableFuture snd) { super(null, dep, src, snd); } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; CompletableFuture b; if ((d = dep) == null || !d.biRelay(a = src, b = snd)) return null; src = null; snd = null; dep = null; return d.postFire(a, b, mode); } } boolean biRelay(CompletableFuture a, CompletableFuture b) { Object r, s; Throwable x; if (a == null || (r = a.result) == null || b == null || (s = b.result) == null) return false; if (result == null) { if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) completeThrowable(x, r); else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null) completeThrowable(x, s); else completeNull(); } return true; } /** Recursively constructs a tree of completions. */ static CompletableFuture andTree(CompletableFuture[] cfs, int lo, int hi) { CompletableFuture d = new CompletableFuture(); if (lo > hi) // empty d.result = NIL; else { CompletableFuture a, b; int mid = (lo + hi) >>> 1; if ((a = (lo == mid ? cfs[lo] : andTree(cfs, lo, mid))) == null || (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] : andTree(cfs, mid+1, hi))) == null) throw new NullPointerException(); if (!d.biRelay(a, b)) { BiRelay c = new BiRelay<>(d, a, b); a.bipush(b, c); c.tryFire(SYNC); } } return d; } /* ------------- Projected (Ored) BiCompletions -------------- */ /** Pushes completion to this and b unless either done. */ final void orpush(CompletableFuture b, BiCompletion c) { if (c != null) { while ((b == null || b.result == null) && result == null) { if (tryPushStack(c)) { if (b != null && b != this && b.result == null) { Completion q = new CoCompletion(c); while (result == null && b.result == null && !b.tryPushStack(q)) lazySetNext(q, null); // clear on failure } break; } lazySetNext(c, null); // clear on failure } } } @SuppressWarnings("serial") static final class OrApply extends BiCompletion { Function fn; OrApply(Executor executor, CompletableFuture dep, CompletableFuture src, CompletableFuture snd, Function fn) { super(executor, dep, src, snd); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; CompletableFuture b; if ((d = dep) == null || !d.orApply(a = src, b = snd, fn, mode > 0 ? null : this)) return null; dep = null; src = null; snd = null; fn = null; return d.postFire(a, b, mode); } } final boolean orApply(CompletableFuture a, CompletableFuture b, Function f, OrApply c) { Object r; Throwable x; if (a == null || b == null || ((r = a.result) == null && (r = b.result) == null) || f == null) return false; tryComplete: if (result == null) { try { if (c != null && !c.claim()) return false; if (r instanceof AltResult) { if ((x = ((AltResult)r).ex) != null) { completeThrowable(x, r); break tryComplete; } r = null; } @SuppressWarnings("unchecked") R rr = (R) r; completeValue(f.apply(rr)); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture orApplyStage( Executor e, CompletionStage o, Function f) { CompletableFuture b; if (f == null || (b = o.toCompletableFuture()) == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.orApply(this, b, f, null)) { OrApply c = new OrApply(e, d, this, b, f); orpush(b, c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class OrAccept extends BiCompletion { Consumer fn; OrAccept(Executor executor, CompletableFuture dep, CompletableFuture src, CompletableFuture snd, Consumer fn) { super(executor, dep, src, snd); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; CompletableFuture b; if ((d = dep) == null || !d.orAccept(a = src, b = snd, fn, mode > 0 ? null : this)) return null; dep = null; src = null; snd = null; fn = null; return d.postFire(a, b, mode); } } final boolean orAccept(CompletableFuture a, CompletableFuture b, Consumer f, OrAccept c) { Object r; Throwable x; if (a == null || b == null || ((r = a.result) == null && (r = b.result) == null) || f == null) return false; tryComplete: if (result == null) { try { if (c != null && !c.claim()) return false; if (r instanceof AltResult) { if ((x = ((AltResult)r).ex) != null) { completeThrowable(x, r); break tryComplete; } r = null; } @SuppressWarnings("unchecked") R rr = (R) r; f.accept(rr); completeNull(); } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture orAcceptStage( Executor e, CompletionStage o, Consumer f) { CompletableFuture b; if (f == null || (b = o.toCompletableFuture()) == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.orAccept(this, b, f, null)) { OrAccept c = new OrAccept(e, d, this, b, f); orpush(b, c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class OrRun extends BiCompletion { Runnable fn; OrRun(Executor executor, CompletableFuture dep, CompletableFuture src, CompletableFuture snd, Runnable fn) { super(executor, dep, src, snd); this.fn = fn; } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; CompletableFuture b; if ((d = dep) == null || !d.orRun(a = src, b = snd, fn, mode > 0 ? null : this)) return null; dep = null; src = null; snd = null; fn = null; return d.postFire(a, b, mode); } } final boolean orRun(CompletableFuture a, CompletableFuture b, Runnable f, OrRun c) { Object r; Throwable x; if (a == null || b == null || ((r = a.result) == null && (r = b.result) == null) || f == null) return false; if (result == null) { try { if (c != null && !c.claim()) return false; if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) completeThrowable(x, r); else { f.run(); completeNull(); } } catch (Throwable ex) { completeThrowable(ex); } } return true; } private CompletableFuture orRunStage(Executor e, CompletionStage o, Runnable f) { CompletableFuture b; if (f == null || (b = o.toCompletableFuture()) == null) throw new NullPointerException(); CompletableFuture d = newIncompleteFuture(); if (e != null || !d.orRun(this, b, f, null)) { OrRun c = new OrRun<>(e, d, this, b, f); orpush(b, c); c.tryFire(SYNC); } return d; } @SuppressWarnings("serial") static final class OrRelay extends BiCompletion { // for Or OrRelay(CompletableFuture dep, CompletableFuture src, CompletableFuture snd) { super(null, dep, src, snd); } final CompletableFuture tryFire(int mode) { CompletableFuture d; CompletableFuture a; CompletableFuture b; if ((d = dep) == null || !d.orRelay(a = src, b = snd)) return null; src = null; snd = null; dep = null; return d.postFire(a, b, mode); } } final boolean orRelay(CompletableFuture a, CompletableFuture b) { Object r; if (a == null || b == null || ((r = a.result) == null && (r = b.result) == null)) return false; if (result == null) completeRelay(r); return true; } /** Recursively constructs a tree of completions. */ static CompletableFuture orTree(CompletableFuture[] cfs, int lo, int hi) { CompletableFuture d = new CompletableFuture(); if (lo <= hi) { CompletableFuture a, b; int mid = (lo + hi) >>> 1; if ((a = (lo == mid ? cfs[lo] : orTree(cfs, lo, mid))) == null || (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] : orTree(cfs, mid+1, hi))) == null) throw new NullPointerException(); if (!d.orRelay(a, b)) { OrRelay c = new OrRelay<>(d, a, b); a.orpush(b, c); c.tryFire(SYNC); } } return d; } /* ------------- Zero-input Async forms -------------- */ @SuppressWarnings("serial") static final class AsyncSupply extends ForkJoinTask implements Runnable, AsynchronousCompletionTask { CompletableFuture dep; Supplier fn; AsyncSupply(CompletableFuture dep, Supplier fn) { this.dep = dep; this.fn = fn; } public final Void getRawResult() { return null; } public final void setRawResult(Void v) {} public final boolean exec() { run(); return true; } public void run() { CompletableFuture d; Supplier f; if ((d = dep) != null && (f = fn) != null) { dep = null; fn = null; if (d.result == null) { try { d.completeValue(f.get()); } catch (Throwable ex) { d.completeThrowable(ex); } } d.postComplete(); } } } static CompletableFuture asyncSupplyStage(Executor e, Supplier f) { if (f == null) throw new NullPointerException(); CompletableFuture d = new CompletableFuture(); e.execute(new AsyncSupply(d, f)); return d; } @SuppressWarnings("serial") static final class AsyncRun extends ForkJoinTask implements Runnable, AsynchronousCompletionTask { CompletableFuture dep; Runnable fn; AsyncRun(CompletableFuture dep, Runnable fn) { this.dep = dep; this.fn = fn; } public final Void getRawResult() { return null; } public final void setRawResult(Void v) {} public final boolean exec() { run(); return true; } public void run() { CompletableFuture d; Runnable f; if ((d = dep) != null && (f = fn) != null) { dep = null; fn = null; if (d.result == null) { try { f.run(); d.completeNull(); } catch (Throwable ex) { d.completeThrowable(ex); } } d.postComplete(); } } } static CompletableFuture asyncRunStage(Executor e, Runnable f) { if (f == null) throw new NullPointerException(); CompletableFuture d = new CompletableFuture(); e.execute(new AsyncRun(d, f)); return d; } /* ------------- Signallers -------------- */ /** * Completion for recording and releasing a waiting thread. This * class implements ManagedBlocker to avoid starvation when * blocking actions pile up in ForkJoinPools. */ @SuppressWarnings("serial") static final class Signaller extends Completion implements ForkJoinPool.ManagedBlocker { long nanos; // remaining wait time if timed final long deadline; // non-zero if timed final boolean interruptible; boolean interrupted; volatile Thread thread; Signaller(boolean interruptible, long nanos, long deadline) { this.thread = Thread.currentThread(); this.interruptible = interruptible; this.nanos = nanos; this.deadline = deadline; } final CompletableFuture tryFire(int ignore) { Thread w; // no need to atomically claim if ((w = thread) != null) { thread = null; LockSupport.unpark(w); } return null; } public boolean isReleasable() { if (Thread.interrupted()) interrupted = true; return ((interrupted && interruptible) || (deadline != 0L && (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) || thread == null); } public boolean block() { while (!isReleasable()) { if (deadline == 0L) LockSupport.park(this); else LockSupport.parkNanos(this, nanos); } return true; } final boolean isLive() { return thread != null; } } /** * Returns raw result after waiting, or null if interruptible and * interrupted. */ private Object waitingGet(boolean interruptible) { Signaller q = null; boolean queued = false; int spins = SPINS; Object r; while ((r = result) == null) { if (spins > 0) { if (ThreadLocalRandom.nextSecondarySeed() >= 0) --spins; } else if (q == null) q = new Signaller(interruptible, 0L, 0L); else if (!queued) queued = tryPushStack(q); else { try { ForkJoinPool.managedBlock(q); } catch (InterruptedException ie) { // currently cannot happen q.interrupted = true; } if (q.interrupted && interruptible) break; } } if (q != null) { q.thread = null; if (q.interrupted) { if (interruptible) cleanStack(); else Thread.currentThread().interrupt(); } } if (r != null) postComplete(); return r; } /** * Returns raw result after waiting, or null if interrupted, or * throws TimeoutException on timeout. */ private Object timedGet(long nanos) throws TimeoutException { if (Thread.interrupted()) return null; if (nanos > 0L) { long d = System.nanoTime() + nanos; long deadline = (d == 0L) ? 1L : d; // avoid 0 Signaller q = null; boolean queued = false; Object r; while ((r = result) == null) { // similar to untimed, without spins if (q == null) q = new Signaller(true, nanos, deadline); else if (!queued) queued = tryPushStack(q); else if (q.nanos <= 0L) break; else { try { ForkJoinPool.managedBlock(q); } catch (InterruptedException ie) { q.interrupted = true; } if (q.interrupted) break; } } if (q != null) q.thread = null; if (r != null) postComplete(); else cleanStack(); if (r != null || (q != null && q.interrupted)) return r; } throw new TimeoutException(); } /* ------------- public methods -------------- */ /** * Creates a new incomplete CompletableFuture. */ public CompletableFuture() { } /** * Creates a new complete CompletableFuture with given encoded result. */ CompletableFuture(Object r) { this.result = r; } /** * Returns a new CompletableFuture that is asynchronously completed * by a task running in the {@link ForkJoinPool#commonPool()} with * the value obtained by calling the given Supplier. * * @param supplier a function returning the value to be used * to complete the returned CompletableFuture * @param the function's return type * @return the new CompletableFuture */ public static CompletableFuture supplyAsync(Supplier supplier) { return asyncSupplyStage(ASYNC_POOL, supplier); } /** * Returns a new CompletableFuture that is asynchronously completed * by a task running in the given executor with the value obtained * by calling the given Supplier. * * @param supplier a function returning the value to be used * to complete the returned CompletableFuture * @param executor the executor to use for asynchronous execution * @param the function's return type * @return the new CompletableFuture */ public static CompletableFuture supplyAsync(Supplier supplier, Executor executor) { return asyncSupplyStage(screenExecutor(executor), supplier); } /** * Returns a new CompletableFuture that is asynchronously completed * by a task running in the {@link ForkJoinPool#commonPool()} after * it runs the given action. * * @param runnable the action to run before completing the * returned CompletableFuture * @return the new CompletableFuture */ public static CompletableFuture runAsync(Runnable runnable) { return asyncRunStage(ASYNC_POOL, runnable); } /** * Returns a new CompletableFuture that is asynchronously completed * by a task running in the given executor after it runs the given * action. * * @param runnable the action to run before completing the * returned CompletableFuture * @param executor the executor to use for asynchronous execution * @return the new CompletableFuture */ public static CompletableFuture runAsync(Runnable runnable, Executor executor) { return asyncRunStage(screenExecutor(executor), runnable); } /** * Returns a new CompletableFuture that is already completed with * the given value. * * @param value the value * @param the type of the value * @return the completed CompletableFuture */ public static CompletableFuture completedFuture(U value) { return new CompletableFuture((value == null) ? NIL : value); } /** * Returns {@code true} if completed in any fashion: normally, * exceptionally, or via cancellation. * * @return {@code true} if completed */ public boolean isDone() { return result != null; } /** * Waits if necessary for this future to complete, and then * returns its result. * * @return the result value * @throws CancellationException if this future was cancelled * @throws ExecutionException if this future completed exceptionally * @throws InterruptedException if the current thread was interrupted * while waiting */ public T get() throws InterruptedException, ExecutionException { Object r; return reportGet((r = result) == null ? waitingGet(true) : r); } /** * Waits if necessary for at most the given time for this future * to complete, and then returns its result, if available. * * @param timeout the maximum time to wait * @param unit the time unit of the timeout argument * @return the result value * @throws CancellationException if this future was cancelled * @throws ExecutionException if this future completed exceptionally * @throws InterruptedException if the current thread was interrupted * while waiting * @throws TimeoutException if the wait timed out */ public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { Object r; long nanos = unit.toNanos(timeout); return reportGet((r = result) == null ? timedGet(nanos) : r); } /** * Returns the result value when complete, or throws an * (unchecked) exception if completed exceptionally. To better * conform with the use of common functional forms, if a * computation involved in the completion of this * CompletableFuture threw an exception, this method throws an * (unchecked) {@link CompletionException} with the underlying * exception as its cause. * * @return the result value * @throws CancellationException if the computation was cancelled * @throws CompletionException if this future completed * exceptionally or a completion computation threw an exception */ public T join() { Object r; return reportJoin((r = result) == null ? waitingGet(false) : r); } /** * Returns the result value (or throws any encountered exception) * if completed, else returns the given valueIfAbsent. * * @param valueIfAbsent the value to return if not completed * @return the result value, if completed, else the given valueIfAbsent * @throws CancellationException if the computation was cancelled * @throws CompletionException if this future completed * exceptionally or a completion computation threw an exception */ public T getNow(T valueIfAbsent) { Object r; return ((r = result) == null) ? valueIfAbsent : reportJoin(r); } /** * If not already completed, sets the value returned by {@link * #get()} and related methods to the given value. * * @param value the result value * @return {@code true} if this invocation caused this CompletableFuture * to transition to a completed state, else {@code false} */ public boolean complete(T value) { boolean triggered = completeValue(value); postComplete(); return triggered; } /** * If not already completed, causes invocations of {@link #get()} * and related methods to throw the given exception. * * @param ex the exception * @return {@code true} if this invocation caused this CompletableFuture * to transition to a completed state, else {@code false} */ public boolean completeExceptionally(Throwable ex) { if (ex == null) throw new NullPointerException(); boolean triggered = internalComplete(new AltResult(ex)); postComplete(); return triggered; } public CompletableFuture thenApply( Function fn) { return uniApplyStage(null, fn); } public CompletableFuture thenApplyAsync( Function fn) { return uniApplyStage(defaultExecutor(), fn); } public CompletableFuture thenApplyAsync( Function fn, Executor executor) { return uniApplyStage(screenExecutor(executor), fn); } public CompletableFuture thenAccept(Consumer action) { return uniAcceptStage(null, action); } public CompletableFuture thenAcceptAsync(Consumer action) { return uniAcceptStage(defaultExecutor(), action); } public CompletableFuture thenAcceptAsync(Consumer action, Executor executor) { return uniAcceptStage(screenExecutor(executor), action); } public CompletableFuture thenRun(Runnable action) { return uniRunStage(null, action); } public CompletableFuture thenRunAsync(Runnable action) { return uniRunStage(defaultExecutor(), action); } public CompletableFuture thenRunAsync(Runnable action, Executor executor) { return uniRunStage(screenExecutor(executor), action); } public CompletableFuture thenCombine( CompletionStage other, BiFunction fn) { return biApplyStage(null, other, fn); } public CompletableFuture thenCombineAsync( CompletionStage other, BiFunction fn) { return biApplyStage(defaultExecutor(), other, fn); } public CompletableFuture thenCombineAsync( CompletionStage other, BiFunction fn, Executor executor) { return biApplyStage(screenExecutor(executor), other, fn); } public CompletableFuture thenAcceptBoth( CompletionStage other, BiConsumer action) { return biAcceptStage(null, other, action); } public CompletableFuture thenAcceptBothAsync( CompletionStage other, BiConsumer action) { return biAcceptStage(defaultExecutor(), other, action); } public CompletableFuture thenAcceptBothAsync( CompletionStage other, BiConsumer action, Executor executor) { return biAcceptStage(screenExecutor(executor), other, action); } public CompletableFuture runAfterBoth(CompletionStage other, Runnable action) { return biRunStage(null, other, action); } public CompletableFuture runAfterBothAsync(CompletionStage other, Runnable action) { return biRunStage(defaultExecutor(), other, action); } public CompletableFuture runAfterBothAsync(CompletionStage other, Runnable action, Executor executor) { return biRunStage(screenExecutor(executor), other, action); } public CompletableFuture applyToEither( CompletionStage other, Function fn) { return orApplyStage(null, other, fn); } public CompletableFuture applyToEitherAsync( CompletionStage other, Function fn) { return orApplyStage(defaultExecutor(), other, fn); } public CompletableFuture applyToEitherAsync( CompletionStage other, Function fn, Executor executor) { return orApplyStage(screenExecutor(executor), other, fn); } public CompletableFuture acceptEither( CompletionStage other, Consumer action) { return orAcceptStage(null, other, action); } public CompletableFuture acceptEitherAsync( CompletionStage other, Consumer action) { return orAcceptStage(defaultExecutor(), other, action); } public CompletableFuture acceptEitherAsync( CompletionStage other, Consumer action, Executor executor) { return orAcceptStage(screenExecutor(executor), other, action); } public CompletableFuture runAfterEither(CompletionStage other, Runnable action) { return orRunStage(null, other, action); } public CompletableFuture runAfterEitherAsync(CompletionStage other, Runnable action) { return orRunStage(defaultExecutor(), other, action); } public CompletableFuture runAfterEitherAsync(CompletionStage other, Runnable action, Executor executor) { return orRunStage(screenExecutor(executor), other, action); } public CompletableFuture thenCompose( Function> fn) { return uniComposeStage(null, fn); } public CompletableFuture thenComposeAsync( Function> fn) { return uniComposeStage(defaultExecutor(), fn); } public CompletableFuture thenComposeAsync( Function> fn, Executor executor) { return uniComposeStage(screenExecutor(executor), fn); } public CompletableFuture whenComplete( BiConsumer action) { return uniWhenCompleteStage(null, action); } public CompletableFuture whenCompleteAsync( BiConsumer action) { return uniWhenCompleteStage(defaultExecutor(), action); } public CompletableFuture whenCompleteAsync( BiConsumer action, Executor executor) { return uniWhenCompleteStage(screenExecutor(executor), action); } public CompletableFuture handle( BiFunction fn) { return uniHandleStage(null, fn); } public CompletableFuture handleAsync( BiFunction fn) { return uniHandleStage(defaultExecutor(), fn); } public CompletableFuture handleAsync( BiFunction fn, Executor executor) { return uniHandleStage(screenExecutor(executor), fn); } /** * Returns this CompletableFuture. * * @return this CompletableFuture */ public CompletableFuture toCompletableFuture() { return this; } // not in interface CompletionStage /** * Returns a new CompletableFuture that is completed when this * CompletableFuture completes, with the result of the given * function of the exception triggering this CompletableFuture's * completion when it completes exceptionally; otherwise, if this * CompletableFuture completes normally, then the returned * CompletableFuture also completes normally with the same value. * Note: More flexible versions of this functionality are * available using methods {@code whenComplete} and {@code handle}. * * @param fn the function to use to compute the value of the * returned CompletableFuture if this CompletableFuture completed * exceptionally * @return the new CompletableFuture */ public CompletableFuture exceptionally( Function fn) { return uniExceptionallyStage(fn); } /* ------------- Arbitrary-arity constructions -------------- */ /** * Returns a new CompletableFuture that is completed when all of * the given CompletableFutures complete. If any of the given * CompletableFutures complete exceptionally, then the returned * CompletableFuture also does so, with a CompletionException * holding this exception as its cause. Otherwise, the results, * if any, of the given CompletableFutures are not reflected in * the returned CompletableFuture, but may be obtained by * inspecting them individually. If no CompletableFutures are * provided, returns a CompletableFuture completed with the value * {@code null}. * *

Among the applications of this method is to await completion * of a set of independent CompletableFutures before continuing a * program, as in: {@code CompletableFuture.allOf(c1, c2, * c3).join();}. * * @param cfs the CompletableFutures * @return a new CompletableFuture that is completed when all of the * given CompletableFutures complete * @throws NullPointerException if the array or any of its elements are * {@code null} */ public static CompletableFuture allOf(CompletableFuture... cfs) { return andTree(cfs, 0, cfs.length - 1); } /** * Returns a new CompletableFuture that is completed when any of * the given CompletableFutures complete, with the same result. * Otherwise, if it completed exceptionally, the returned * CompletableFuture also does so, with a CompletionException * holding this exception as its cause. If no CompletableFutures * are provided, returns an incomplete CompletableFuture. * * @param cfs the CompletableFutures * @return a new CompletableFuture that is completed with the * result or exception of any of the given CompletableFutures when * one completes * @throws NullPointerException if the array or any of its elements are * {@code null} */ public static CompletableFuture anyOf(CompletableFuture... cfs) { return orTree(cfs, 0, cfs.length - 1); } /* ------------- Control and status methods -------------- */ /** * If not already completed, completes this CompletableFuture with * a {@link CancellationException}. Dependent CompletableFutures * that have not already completed will also complete * exceptionally, with a {@link CompletionException} caused by * this {@code CancellationException}. * * @param mayInterruptIfRunning this value has no effect in this * implementation because interrupts are not used to control * processing. * * @return {@code true} if this task is now cancelled */ public boolean cancel(boolean mayInterruptIfRunning) { boolean cancelled = (result == null) && internalComplete(new AltResult(new CancellationException())); postComplete(); return cancelled || isCancelled(); } /** * Returns {@code true} if this CompletableFuture was cancelled * before it completed normally. * * @return {@code true} if this CompletableFuture was cancelled * before it completed normally */ public boolean isCancelled() { Object r; return ((r = result) instanceof AltResult) && (((AltResult)r).ex instanceof CancellationException); } /** * Returns {@code true} if this CompletableFuture completed * exceptionally, in any way. Possible causes include * cancellation, explicit invocation of {@code * completeExceptionally}, and abrupt termination of a * CompletionStage action. * * @return {@code true} if this CompletableFuture completed * exceptionally */ public boolean isCompletedExceptionally() { Object r; return ((r = result) instanceof AltResult) && r != NIL; } /** * Forcibly sets or resets the value subsequently returned by * method {@link #get()} and related methods, whether or not * already completed. This method is designed for use only in * error recovery actions, and even in such situations may result * in ongoing dependent completions using established versus * overwritten outcomes. * * @param value the completion value */ public void obtrudeValue(T value) { result = (value == null) ? NIL : value; postComplete(); } /** * Forcibly causes subsequent invocations of method {@link #get()} * and related methods to throw the given exception, whether or * not already completed. This method is designed for use only in * error recovery actions, and even in such situations may result * in ongoing dependent completions using established versus * overwritten outcomes. * * @param ex the exception * @throws NullPointerException if the exception is null */ public void obtrudeException(Throwable ex) { if (ex == null) throw new NullPointerException(); result = new AltResult(ex); postComplete(); } /** * Returns the estimated number of CompletableFutures whose * completions are awaiting completion of this CompletableFuture. * This method is designed for use in monitoring system state, not * for synchronization control. * * @return the number of dependent CompletableFutures */ public int getNumberOfDependents() { int count = 0; for (Completion p = stack; p != null; p = p.next) ++count; return count; } /** * Returns a string identifying this CompletableFuture, as well as * its completion state. The state, in brackets, contains the * String {@code "Completed Normally"} or the String {@code * "Completed Exceptionally"}, or the String {@code "Not * completed"} followed by the number of CompletableFutures * dependent upon its completion, if any. * * @return a string identifying this CompletableFuture, as well as its state */ public String toString() { Object r = result; int count = 0; // avoid call to getNumberOfDependents in case disabled for (Completion p = stack; p != null; p = p.next) ++count; return super.toString() + ((r == null) ? ((count == 0) ? "[Not completed]" : "[Not completed, " + count + " dependents]") : (((r instanceof AltResult) && ((AltResult)r).ex != null) ? "[Completed exceptionally]" : "[Completed normally]")); } // jdk9 additions /** * Returns a new incomplete CompletableFuture of the type to be * returned by a CompletionStage method. Subclasses should * normally override this method to return an instance of the same * class as this CompletableFuture. The default implementation * returns an instance of class CompletableFuture. * * @param the type of the value * @return a new CompletableFuture * @since 1.9 */ public CompletableFuture newIncompleteFuture() { return new CompletableFuture(); } /** * Returns the default Executor used for async methods that do not * specify an Executor. This class uses the {@link * ForkJoinPool#commonPool()} if it supports more than one * parallel thread, or else an Executor using one thread per async * task. This method may be overridden in subclasses to return * an Executor that provides at least one independent thread. * * @return the executor * @since 1.9 */ public Executor defaultExecutor() { return ASYNC_POOL; } /** * Returns a new CompletableFuture that is completed normally with * the same value as this CompletableFuture when it completes * normally. If this CompletableFuture completes exceptionally, * then the returned CompletableFuture completes exceptionally * with a CompletionException with this exception as cause. The * behavior is equivalent to {@code thenApply(x -> x)}. This * method may be useful as a form of "defensive copying", to * prevent clients from completing, while still being able to * arrange dependent actions. * * @return the new CompletableFuture * @since 1.9 */ public CompletableFuture copy() { return uniCopyStage(); } /** * Returns a new CompletionStage that is completed normally with * the same value as this CompletableFuture when it completes * normally, and cannot be independently completed or otherwise * used in ways not defined by the methods of interface {@link * CompletionStage}. If this CompletableFuture completes * exceptionally, then the returned CompletionStage completes * exceptionally with a CompletionException with this exception as * cause. * * @return the new CompletionStage * @since 1.9 */ public CompletionStage minimalCompletionStage() { return uniAsMinimalStage(); } /** * Completes this CompletableFuture with the result of * the given Supplier function invoked from an asynchronous * task using the given executor. * * @param supplier a function returning the value to be used * to complete this CompletableFuture * @param executor the executor to use for asynchronous execution * @return this CompletableFuture * @since 1.9 */ public CompletableFuture completeAsync(Supplier supplier, Executor executor) { if (supplier == null || executor == null) throw new NullPointerException(); executor.execute(new AsyncSupply(this, supplier)); return this; } /** * Completes this CompletableFuture with the result of the given * Supplier function invoked from an asynchronous task using the * default executor. * * @param supplier a function returning the value to be used * to complete this CompletableFuture * @return this CompletableFuture * @since 1.9 */ public CompletableFuture completeAsync(Supplier supplier) { return completeAsync(supplier, defaultExecutor()); } /** * Exceptionally completes this CompletableFuture with * a {@link TimeoutException} if not otherwise completed * before the given timeout. * * @param timeout how long to wait before completing exceptionally * with a TimeoutException, in units of {@code unit} * @param unit a {@code TimeUnit} determining how to interpret the * {@code timeout} parameter * @return this CompletableFuture * @since 1.9 */ public CompletableFuture orTimeout(long timeout, TimeUnit unit) { if (unit == null) throw new NullPointerException(); if (result == null) whenComplete(new Canceller(Delayer.delay(new Timeout(this), timeout, unit))); return this; } /** * Completes this CompletableFuture with the given value if not * otherwise completed before the given timeout. * * @param value the value to use upon timeout * @param timeout how long to wait before completing normally * with the given value, in units of {@code unit} * @param unit a {@code TimeUnit} determining how to interpret the * {@code timeout} parameter * @return this CompletableFuture * @since 1.9 */ public CompletableFuture completeOnTimeout(T value, long timeout, TimeUnit unit) { if (unit == null) throw new NullPointerException(); if (result == null) whenComplete(new Canceller(Delayer.delay( new DelayedCompleter(this, value), timeout, unit))); return this; } /** * Returns a new Executor that submits a task to the given base * executor after the given delay (or no delay if non-positive). * Each delay commences upon invocation of the returned executor's * {@code execute} method. * * @param delay how long to delay, in units of {@code unit} * @param unit a {@code TimeUnit} determining how to interpret the * {@code delay} parameter * @param executor the base executor * @return the new delayed executor * @since 1.9 */ public static Executor delayedExecutor(long delay, TimeUnit unit, Executor executor) { if (unit == null || executor == null) throw new NullPointerException(); return new DelayedExecutor(delay, unit, executor); } /** * Returns a new Executor that submits a task to the default * executor after the given delay (or no delay if non-positive). * Each delay commences upon invocation of the returned executor's * {@code execute} method. * * @param delay how long to delay, in units of {@code unit} * @param unit a {@code TimeUnit} determining how to interpret the * {@code delay} parameter * @return the new delayed executor * @since 1.9 */ public static Executor delayedExecutor(long delay, TimeUnit unit) { if (unit == null) throw new NullPointerException(); return new DelayedExecutor(delay, unit, ASYNC_POOL); } /** * Returns a new CompletionStage that is already completed with * the given value and supports only those methods in * interface {@link CompletionStage}. * * @param value the value * @param the type of the value * @return the completed CompletionStage * @since 1.9 */ public static CompletionStage completedStage(U value) { return new MinimalStage((value == null) ? NIL : value); } /** * Returns a new CompletableFuture that is already completed * exceptionally with the given exception. * * @param ex the exception * @param the type of the value * @return the exceptionally completed CompletableFuture * @since 1.9 */ public static CompletableFuture failedFuture(Throwable ex) { if (ex == null) throw new NullPointerException(); return new CompletableFuture(new AltResult(ex)); } /** * Returns a new CompletionStage that is already completed * exceptionally with the given exception and supports only those * methods in interface {@link CompletionStage}. * * @param ex the exception * @param the type of the value * @return the exceptionally completed CompletionStage * @since 1.9 */ public static CompletionStage failedStage(Throwable ex) { if (ex == null) throw new NullPointerException(); return new MinimalStage(new AltResult(ex)); } /** * Singleton delay scheduler, used only for starting and * cancelling tasks. */ static final class Delayer { static ScheduledFuture delay(Runnable command, long delay, TimeUnit unit) { return delayer.schedule(command, delay, unit); } static final class DaemonThreadFactory implements ThreadFactory { public Thread newThread(Runnable r) { Thread t = new Thread(r); t.setDaemon(true); t.setName("CompletableFutureDelayScheduler"); return t; } } static final ScheduledThreadPoolExecutor delayer; static { (delayer = new ScheduledThreadPoolExecutor( 1, new DaemonThreadFactory())). setRemoveOnCancelPolicy(true); } } // Little class-ified lambdas to better support monitoring static final class DelayedExecutor implements Executor { final long delay; final TimeUnit unit; final Executor executor; DelayedExecutor(long delay, TimeUnit unit, Executor executor) { this.delay = delay; this.unit = unit; this.executor = executor; } public void execute(Runnable r) { Delayer.delay(new TaskSubmitter(executor, r), delay, unit); } } /** Action to submit user task */ static final class TaskSubmitter implements Runnable { final Executor executor; final Runnable action; TaskSubmitter(Executor executor, Runnable action) { this.executor = executor; this.action = action; } public void run() { executor.execute(action); } } /** Action to completeExceptionally on timeout */ static final class Timeout implements Runnable { final CompletableFuture f; Timeout(CompletableFuture f) { this.f = f; } public void run() { if (f != null && !f.isDone()) f.completeExceptionally(new TimeoutException()); } } /** Action to complete on timeout */ static final class DelayedCompleter implements Runnable { final CompletableFuture f; final U u; DelayedCompleter(CompletableFuture f, U u) { this.f = f; this.u = u; } public void run() { if (f != null) f.complete(u); } } /** Action to cancel unneeded timeouts */ static final class Canceller implements BiConsumer { final Future f; Canceller(Future f) { this.f = f; } public void accept(Object ignore, Throwable ex) { if (ex == null && f != null && !f.isDone()) f.cancel(false); } } /** * A subclass that just throws UOE for most non-CompletionStage methods. */ static final class MinimalStage extends CompletableFuture { MinimalStage() { } MinimalStage(Object r) { super(r); } @Override public CompletableFuture newIncompleteFuture() { return new MinimalStage(); } @Override public T get() { throw new UnsupportedOperationException(); } @Override public T get(long timeout, TimeUnit unit) { throw new UnsupportedOperationException(); } @Override public T getNow(T valueIfAbsent) { throw new UnsupportedOperationException(); } @Override public T join() { throw new UnsupportedOperationException(); } @Override public boolean complete(T value) { throw new UnsupportedOperationException(); } @Override public boolean completeExceptionally(Throwable ex) { throw new UnsupportedOperationException(); } @Override public boolean cancel(boolean mayInterruptIfRunning) { throw new UnsupportedOperationException(); } @Override public void obtrudeValue(T value) { throw new UnsupportedOperationException(); } @Override public void obtrudeException(Throwable ex) { throw new UnsupportedOperationException(); } @Override public boolean isDone() { throw new UnsupportedOperationException(); } @Override public boolean isCancelled() { throw new UnsupportedOperationException(); } @Override public boolean isCompletedExceptionally() { throw new UnsupportedOperationException(); } @Override public int getNumberOfDependents() { throw new UnsupportedOperationException(); } @Override public CompletableFuture completeAsync (Supplier supplier, Executor executor) { throw new UnsupportedOperationException(); } @Override public CompletableFuture completeAsync (Supplier supplier) { throw new UnsupportedOperationException(); } @Override public CompletableFuture orTimeout (long timeout, TimeUnit unit) { throw new UnsupportedOperationException(); } @Override public CompletableFuture completeOnTimeout (T value, long timeout, TimeUnit unit) { throw new UnsupportedOperationException(); } } // Unsafe mechanics private static final jdk.internal.misc.Unsafe U = jdk.internal.misc.Unsafe.getUnsafe(); private static final long RESULT; private static final long STACK; private static final long NEXT; static { try { RESULT = U.objectFieldOffset (CompletableFuture.class.getDeclaredField("result")); STACK = U.objectFieldOffset (CompletableFuture.class.getDeclaredField("stack")); NEXT = U.objectFieldOffset (Completion.class.getDeclaredField("next")); } catch (ReflectiveOperationException e) { throw new Error(e); } // Reduce the risk of rare disastrous classloading in first call to // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773 Class ensureLoaded = LockSupport.class; } }