Skip to content

D
dragonwell8_jdk

openanolis / dragonwell8_jdk

通知 4
Star 2
Fork 0
D
dragonwell8_jdk
提交 b51cc74c 编写于 作者: D dl
浏览文件
操作

8056249: Improve CompletableFuture resource usage 

Reviewed-by: psandoz, chegar, martin
上级 4352a351
显示空白变更内容
内联 并排
Showing with 1656 addition and 2271 deletion
src/share/classes/java/util/concurrent/CompletableFuture.java
浏览文件 @ b51cc74c
...@@ -50,7 +50,6 @@ import java.util.concurrent.TimeoutException; ...@@ -50,7 +50,6 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.CancellationException; import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletionException; import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage; import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport; import java.util.concurrent.locks.LockSupport;
/** /**
...@@ -77,9 +76,9 @@ import java.util.concurrent.locks.LockSupport; ...@@ -77,9 +76,9 @@ import java.util.concurrent.locks.LockSupport;
* <li>All <em>async</em> methods without an explicit Executor * <li>All <em>async</em> methods without an explicit Executor
* argument are performed using the {@link ForkJoinPool#commonPool()} * argument are performed using the {@link ForkJoinPool#commonPool()}
* (unless it does not support a parallelism level of at least two, in * (unless it does not support a parallelism level of at least two, in
* which case, a new Thread is used). To simplify monitoring, * which case, a new Thread is created to run each task). To simplify
* debugging, and tracking, all generated asynchronous tasks are * monitoring, debugging, and tracking, all generated asynchronous
* instances of the marker interface {@link * tasks are instances of the marker interface {@link
* AsynchronousCompletionTask}. </li> * AsynchronousCompletionTask}. </li>
* *
* <li>All CompletionStage methods are implemented independently of * <li>All CompletionStage methods are implemented independently of
...@@ -113,301 +112,273 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -113,301 +112,273 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
/* /*
* Overview: * Overview:
* *
* 1. Non-nullness of field result (set via CAS) indicates done. * A CompletableFuture may have dependent completion actions,
* An AltResult is used to box null as a result, as well as to * collected in a linked stack. It atomically completes by CASing
* hold exceptions. Using a single field makes completion fast * a result field, and then pops off and runs those actions. This
* and simple to detect and trigger, at the expense of a lot of * applies across normal vs exceptional outcomes, sync vs async
* encoding and decoding that infiltrates many methods. One minor * actions, binary triggers, and various forms of completions.
* simplification relies on the (static) NIL (to box null results)
* being the only AltResult with a null exception field, so we
* don't usually need explicit comparisons with NIL. The CF
* exception propagation mechanics surrounding decoding rely on
* unchecked casts of decoded results really being unchecked,
* where user type errors are caught at point of use, as is
* currently the case in Java. These are highlighted by using
* SuppressWarnings-annotated temporaries.
* *
* 2. Waiters are held in a Treiber stack similar to the one used * Non-nullness of field result (set via CAS) indicates done. An
* in FutureTask, Phaser, and SynchronousQueue. See their * AltResult is used to box null as a result, as well as to hold
* internal documentation for algorithmic details. * 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.
* *
* 3. Completions are also kept in a list/stack, and pulled off * Dependent actions are represented by Completion objects linked
* and run when completion is triggered. (We could even use the * as Treiber stacks headed by field "stack". There are Completion
* same stack as for waiters, but would give up the potential * classes for each kind of action, grouped into single-input
* parallelism obtained because woken waiters help release/run * (UniCompletion), two-input (BiCompletion), projected
* others -- see method postComplete). Because post-processing * (BiCompletions using either (not both) of two inputs), shared
* may race with direct calls, class Completion opportunistically * (CoCompletion, used by the second of two sources), zero-input
* extends AtomicInteger so callers can claim the action via * source actions, and Signallers that unblock waiters. Class
* compareAndSet(0, 1). The Completion.run methods are all * Completion extends ForkJoinTask to enable async execution
* written a boringly similar uniform way (that sometimes includes * (adding no space overhead because we exploit its "tag" methods
* unnecessary-looking checks, kept to maintain uniformity). * to maintain claims). It is also declared as Runnable to allow
* There are enough dimensions upon which they differ that * usage with arbitrary executors.
* attempts to factor commonalities while maintaining efficiency
* require more lines of code than they would save.
* *
* 4. The exported then/and/or methods do support a bit of * Support for each kind of CompletionStage relies on a separate
* factoring (see doThenApply etc). They must cope with the * class, along with two CompletableFuture methods:
* intrinsic races surrounding addition of a dependent action *
* versus performing the action directly because the task is * * A Completion class with name X corresponding to function,
* already complete. For example, a CF may not be complete upon * prefaced with "Uni", "Bi", or "Or". Each class contains
* entry, so a dependent completion is added, but by the time it * fields for source(s), actions, and dependent. They are
* is added, the target CF is complete, so must be directly * boringly similar, differing from others only with respect to
* executed. This is all done while avoiding unnecessary object * underlying functional forms. We do this so that users don't
* construction in safe-bypass cases. * encounter layers of adaptors in common usages. We also
* include "Relay" classes/methods that don't correspond to user
* methods; they copy results from one stage to another.
*
* * 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.
*/ */
// preliminaries 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 UNSAFE.compareAndSwapObject(this, RESULT, null, r);
}
final boolean casStack(Completion cmp, Completion val) {
return UNSAFE.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 UNSAFE.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 { static final class AltResult { // See above
final Throwable ex; // null only for NIL final Throwable ex; // null only for NIL
AltResult(Throwable ex) { this.ex = ex; } AltResult(Throwable x) { this.ex = x; }
} }
/** The encoding of the null value. */
static final AltResult NIL = new AltResult(null); static final AltResult NIL = new AltResult(null);
// Fields /** Completes with the null value, unless already completed. */
final boolean completeNull() {
return UNSAFE.compareAndSwapObject(this, RESULT, null,
NIL);
}
volatile Object result; // Either the result or boxed AltResult /** Returns the encoding of the given non-exceptional value. */
volatile WaitNode waiters; // Treiber stack of threads blocked on get() final Object encodeValue(T t) {
volatile CompletionNode completions; // list (Treiber stack) of completions return (t == null) ? NIL : t;
}
// Basic utilities for triggering and processing completions /** Completes with a non-exceptional result, unless already completed. */
final boolean completeValue(T t) {
return UNSAFE.compareAndSwapObject(this, RESULT, null,
(t == null) ? NIL : t);
}
/** /**
* Removes and signals all waiting threads and runs all completions. * Returns the encoding of the given (non-null) exception as a
* wrapped CompletionException unless it is one already.
*/ */
final void postComplete() { static AltResult encodeThrowable(Throwable x) {
WaitNode q; Thread t; return new AltResult((x instanceof CompletionException) ? x :
while ((q = waiters) != null) { new CompletionException(x));
if (UNSAFE.compareAndSwapObject(this, WAITERS, q, q.next) &&
(t = q.thread) != null) {
q.thread = null;
LockSupport.unpark(t);
}
} }
CompletionNode h; Completion c; /** Completes with an exceptional result, unless already completed. */
while ((h = completions) != null) { final boolean completeThrowable(Throwable x) {
if (UNSAFE.compareAndSwapObject(this, COMPLETIONS, h, h.next) && return UNSAFE.compareAndSwapObject(this, RESULT, null,
(c = h.completion) != null) encodeThrowable(x));
c.run();
}
} }
/** /**
* Triggers completion with the encoding of the given arguments: * Returns the encoding of the given (non-null) exception as a
* if the exception is non-null, encodes it as a wrapped * wrapped CompletionException unless it is one already. May
* CompletionException unless it is one already. Otherwise uses * return the given Object r (which must have been the result of a
* the given result, boxed as NIL if null. * source future) if it is equivalent, i.e. if this is a simple
* relay of an existing CompletionException.
*/ */
final void internalComplete(T v, Throwable ex) { static Object encodeThrowable(Throwable x, Object r) {
if (result == null) if (!(x instanceof CompletionException))
UNSAFE.compareAndSwapObject x = new CompletionException(x);
(this, RESULT, null, else if (r instanceof AltResult && x == ((AltResult)r).ex)
(ex == null) ? (v == null) ? NIL : v : return r;
new AltResult((ex instanceof CompletionException) ? ex : return new AltResult(x);
new CompletionException(ex)));
postComplete(); // help out even if not triggered
} }
/** /**
* If triggered, helps release and/or process completions. * 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 void helpPostComplete() { final boolean completeThrowable(Throwable x, Object r) {
if (result != null) return UNSAFE.compareAndSwapObject(this, RESULT, null,
postComplete(); encodeThrowable(x, r));
} }
/* ------------- waiting for completions -------------- */
/** Number of processors, for spin control */
static final int NCPU = Runtime.getRuntime().availableProcessors();
/** /**
* Heuristic spin value for waitingGet() before blocking on * Returns the encoding of the given arguments: if the exception
* multiprocessors * is non-null, encodes as AltResult. Otherwise uses the given
* value, boxed as NIL if null.
*/ */
static final int SPINS = (NCPU > 1) ? 1 << 8 : 0; Object encodeOutcome(T t, Throwable x) {
return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x);
}
/** /**
* Linked nodes to record waiting threads in a Treiber stack. See * Returns the encoding of a copied outcome; if exceptional,
* other classes such as Phaser and SynchronousQueue for more * rewraps as a CompletionException, else returns argument.
* detailed explanation. This class implements ManagedBlocker to
* avoid starvation when blocking actions pile up in
* ForkJoinPools.
*/ */
static final class WaitNode implements ForkJoinPool.ManagedBlocker { static Object encodeRelay(Object r) {
long nanos; // wait time if timed Throwable x;
final long deadline; // non-zero if timed return (((r instanceof AltResult) &&
volatile int interruptControl; // > 0: interruptible, < 0: interrupted (x = ((AltResult)r).ex) != null &&
volatile Thread thread; !(x instanceof CompletionException)) ?
volatile WaitNode next; new AltResult(new CompletionException(x)) : r);
WaitNode(boolean interruptible, long nanos, long deadline) {
this.thread = Thread.currentThread();
this.interruptControl = interruptible ? 1 : 0;
this.nanos = nanos;
this.deadline = deadline;
}
public boolean isReleasable() {
if (thread == null)
return true;
if (Thread.interrupted()) {
int i = interruptControl;
interruptControl = -1;
if (i > 0)
return true;
}
if (deadline != 0L &&
(nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {
thread = null;
return true;
}
return false;
}
public boolean block() {
if (isReleasable())
return true;
else if (deadline == 0L)
LockSupport.park(this);
else if (nanos > 0L)
LockSupport.parkNanos(this, nanos);
return isReleasable();
}
} }
/** /**
* Returns raw result after waiting, or null if interruptible and * Completes with r or a copy of r, unless already completed.
* interrupted. * If exceptional, r is first coerced to a CompletionException.
*/ */
private Object waitingGet(boolean interruptible) { final boolean completeRelay(Object r) {
WaitNode q = null; return UNSAFE.compareAndSwapObject(this, RESULT, null,
boolean queued = false; encodeRelay(r));
int spins = SPINS;
for (Object r;;) {
if ((r = result) != null) {
if (q != null) { // suppress unpark
q.thread = null;
if (q.interruptControl < 0) {
if (interruptible) {
removeWaiter(q);
return null;
}
Thread.currentThread().interrupt();
}
}
postComplete(); // help release others
return r;
}
else if (spins > 0) {
int rnd = ThreadLocalRandom.nextSecondarySeed();
if (rnd == 0)
rnd = ThreadLocalRandom.current().nextInt();
if (rnd >= 0)
--spins;
}
else if (q == null)
q = new WaitNode(interruptible, 0L, 0L);
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, WAITERS,
q.next = waiters, q);
else if (interruptible && q.interruptControl < 0) {
removeWaiter(q);
return null;
}
else if (q.thread != null && result == null) {
try {
ForkJoinPool.managedBlock(q);
} catch (InterruptedException ex) {
q.interruptControl = -1;
}
}
}
} }
/** /**
* Awaits completion or aborts on interrupt or timeout. * Reports result using Future.get conventions.
*
* @param nanos time to wait
* @return raw result
*/ */
private Object timedAwaitDone(long nanos) private static <T> T reportGet(Object r)
throws InterruptedException, TimeoutException { throws InterruptedException, ExecutionException {
WaitNode q = null; if (r == null) // by convention below, null means interrupted
boolean queued = false;
for (Object r;;) {
if ((r = result) != null) {
if (q != null) {
q.thread = null;
if (q.interruptControl < 0) {
removeWaiter(q);
throw new InterruptedException();
}
}
postComplete();
return r;
}
else if (q == null) {
if (nanos <= 0L)
throw new TimeoutException();
long d = System.nanoTime() + nanos;
q = new WaitNode(true, nanos, d == 0L ? 1L : d); // avoid 0
}
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, WAITERS,
q.next = waiters, q);
else if (q.interruptControl < 0) {
removeWaiter(q);
throw new InterruptedException(); 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);
} }
else if (q.nanos <= 0L) { @SuppressWarnings("unchecked") T t = (T) r;
if (result == null) { return t;
removeWaiter(q);
throw new TimeoutException();
}
}
else if (q.thread != null && result == null) {
try {
ForkJoinPool.managedBlock(q);
} catch (InterruptedException ex) {
q.interruptControl = -1;
}
}
}
} }
/** /**
* Tries to unlink a timed-out or interrupted wait node to avoid * Decodes outcome to return result or throw unchecked exception.
* accumulating garbage. Internal nodes are simply unspliced
* without CAS since it is harmless if they are traversed anyway
* by releasers. To avoid effects of unsplicing from already
* removed nodes, the list is retraversed in case of an apparent
* race. This is slow when there are a lot of nodes, but we don't
* expect lists to be long enough to outweigh higher-overhead
* schemes.
*/ */
private void removeWaiter(WaitNode node) { private static <T> T reportJoin(Object r) {
if (node != null) { if (r instanceof AltResult) {
node.thread = null; Throwable x;
retry: if ((x = ((AltResult)r).ex) == null)
for (;;) { // restart on removeWaiter race return null;
for (WaitNode pred = null, q = waiters, s; q != null; q = s) { if (x instanceof CancellationException)
s = q.next; throw (CancellationException)x;
if (q.thread != null) if (x instanceof CompletionException)
pred = q; throw (CompletionException)x;
else if (pred != null) { throw new CompletionException(x);
pred.next = s;
if (pred.thread == null) // check for race
continue retry;
}
else if (!UNSAFE.compareAndSwapObject(this, WAITERS, q, s))
continue retry;
}
break;
}
} }
@SuppressWarnings("unchecked") T t = (T) r;
return t;
} }
/* ------------- Async tasks -------------- */ /* ------------- Async task preliminaries -------------- */
/** /**
* A marker interface identifying asynchronous tasks produced by * A marker interface identifying asynchronous tasks produced by
...@@ -419,1693 +390,1394 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -419,1693 +390,1394 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
public static interface AsynchronousCompletionTask { public static interface AsynchronousCompletionTask {
} }
/** Base class can act as either FJ or plain Runnable */ private static final boolean useCommonPool =
(ForkJoinPool.getCommonPoolParallelism() > 1);
/**
* Default executor -- ForkJoinPool.commonPool() unless it cannot
* support parallelism.
*/
private static final Executor asyncPool = useCommonPool ?
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 asyncPool in case parallelism disabled.
*/
static Executor screenExecutor(Executor e) {
if (!useCommonPool && e == ForkJoinPool.commonPool())
return asyncPool;
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;
/* ------------- Base Completion classes and operations -------------- */
@SuppressWarnings("serial") @SuppressWarnings("serial")
abstract static class Async extends ForkJoinTask<Void> abstract static class Completion extends ForkJoinTask<Void>
implements Runnable, AsynchronousCompletionTask { implements Runnable, AsynchronousCompletionTask {
public final Void getRawResult() { return null; } volatile Completion next; // Treiber stack link
public final void setRawResult(Void v) { }
public final void run() { exec(); }
}
/** /**
* Starts the given async task using the given executor, unless * Performs completion action if triggered, returning a
* the executor is ForkJoinPool.commonPool and it has been * dependent that may need propagation, if one exists.
* disabled, in which case starts a new thread. *
* @param mode SYNC, ASYNC, or NESTED
*/ */
static void execAsync(Executor e, Async r) { abstract CompletableFuture<?> tryFire(int mode);
if (e == ForkJoinPool.commonPool() &&
ForkJoinPool.getCommonPoolParallelism() <= 1) /** Returns true if possibly still triggerable. Used by cleanStack. */
new Thread(r).start(); abstract boolean isLive();
else
e.execute(r); public final void run() { tryFire(ASYNC); }
public final boolean exec() { tryFire(ASYNC); return true; }
public final Void getRawResult() { return null; }
public final void setRawResult(Void v) {}
} }
static final class AsyncRun extends Async { static void lazySetNext(Completion c, Completion next) {
final Runnable fn; UNSAFE.putOrderedObject(c, NEXT, next);
final CompletableFuture<Void> dst;
AsyncRun(Runnable fn, CompletableFuture<Void> dst) {
this.fn = fn; this.dst = dst;
} }
public final boolean exec() {
CompletableFuture<Void> d; Throwable ex; /**
if ((d = this.dst) != null && d.result == null) { * Pops and tries to trigger all reachable dependents. Call only
try { * when known to be done.
fn.run(); */
ex = null; final void postComplete() {
} catch (Throwable rex) { /*
ex = rex; * 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;
} }
d.internalComplete(null, ex); h.next = null; // detach
} }
return true; f = (d = h.tryFire(NESTED)) == null ? this : d;
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class AsyncSupply<U> extends Async {
final Supplier<U> fn;
final CompletableFuture<U> dst;
AsyncSupply(Supplier<U> fn, CompletableFuture<U> dst) {
this.fn = fn; this.dst = dst;
} }
public final boolean exec() {
CompletableFuture<U> d; U u; Throwable ex; /** Traverses stack and unlinks dead Completions. */
if ((d = this.dst) != null && d.result == null) { final void cleanStack() {
try { for (Completion p = null, q = stack; q != null;) {
u = fn.get(); Completion s = q.next;
ex = null; if (q.isLive()) {
} catch (Throwable rex) { p = q;
ex = rex; q = s;
u = null;
} }
d.internalComplete(u, ex); else if (p == null) {
casStack(q, s);
q = stack;
} }
return true; else {
p.next = s;
if (p.isLive())
q = s;
else {
p = null; // restart
q = stack;
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class AsyncApply<T,U> extends Async {
final T arg;
final Function<? super T,? extends U> fn;
final CompletableFuture<U> dst;
AsyncApply(T arg, Function<? super T,? extends U> fn,
CompletableFuture<U> dst) {
this.arg = arg; this.fn = fn; this.dst = dst;
} }
public final boolean exec() {
CompletableFuture<U> d; U u; Throwable ex;
if ((d = this.dst) != null && d.result == null) {
try {
u = fn.apply(arg);
ex = null;
} catch (Throwable rex) {
ex = rex;
u = null;
} }
d.internalComplete(u, ex);
/* ------------- One-input Completions -------------- */
/** A Completion with a source, dependent, and executor. */
@SuppressWarnings("serial")
abstract static class UniCompletion<T,V> extends Completion {
Executor executor; // executor to use (null if none)
CompletableFuture<V> dep; // the dependent to complete
CompletableFuture<T> src; // source for action
UniCompletion(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> 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; return true;
executor = null; // disable
e.execute(this);
} }
private static final long serialVersionUID = 5232453952276885070L; return false;
} }
static final class AsyncCombine<T,U,V> extends Async { final boolean isLive() { return dep != null; }
final T arg1;
final U arg2;
final BiFunction<? super T,? super U,? extends V> fn;
final CompletableFuture<V> dst;
AsyncCombine(T arg1, U arg2,
BiFunction<? super T,? super U,? extends V> fn,
CompletableFuture<V> dst) {
this.arg1 = arg1; this.arg2 = arg2; this.fn = fn; this.dst = dst;
} }
public final boolean exec() {
CompletableFuture<V> d; V v; Throwable ex; /** Pushes the given completion (if it exists) unless done. */
if ((d = this.dst) != null && d.result == null) { final void push(UniCompletion<?,?> c) {
try { if (c != null) {
v = fn.apply(arg1, arg2); while (result == null && !tryPushStack(c))
ex = null; lazySetNext(c, null); // clear on failure
} catch (Throwable rex) {
ex = rex;
v = null;
}
d.internalComplete(v, ex);
}
return true;
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class AsyncAccept<T> extends Async { /**
final T arg; * Post-processing by dependent after successful UniCompletion
final Consumer<? super T> fn; * tryFire. Tries to clean stack of source a, and then either runs
final CompletableFuture<?> dst; * postComplete or returns this to caller, depending on mode.
AsyncAccept(T arg, Consumer<? super T> fn, */
CompletableFuture<?> dst) { final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
this.arg = arg; this.fn = fn; this.dst = dst; if (a != null && a.stack != null) {
if (mode < 0 || a.result == null)
a.cleanStack();
else
a.postComplete();
} }
public final boolean exec() { if (result != null && stack != null) {
CompletableFuture<?> d; Throwable ex; if (mode < 0)
if ((d = this.dst) != null && d.result == null) { return this;
try { else
fn.accept(arg); postComplete();
ex = null;
} catch (Throwable rex) {
ex = rex;
} }
d.internalComplete(null, ex); return null;
} }
return true;
@SuppressWarnings("serial")
static final class UniApply<T,V> extends UniCompletion<T,V> {
Function<? super T,? extends V> fn;
UniApply(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src,
Function<? super T,? extends V> fn) {
super(executor, dep, src); this.fn = fn;
}
final CompletableFuture<V> tryFire(int mode) {
CompletableFuture<V> d; CompletableFuture<T> 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);
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class AsyncAcceptBoth<T,U> extends Async { final <S> boolean uniApply(CompletableFuture<S> a,
final T arg1; Function<? super S,? extends T> f,
final U arg2; UniApply<S,T> c) {
final BiConsumer<? super T,? super U> fn; Object r; Throwable x;
final CompletableFuture<?> dst; if (a == null || (r = a.result) == null || f == null)
AsyncAcceptBoth(T arg1, U arg2, return false;
BiConsumer<? super T,? super U> fn, tryComplete: if (result == null) {
CompletableFuture<?> dst) { if (r instanceof AltResult) {
this.arg1 = arg1; this.arg2 = arg2; this.fn = fn; this.dst = dst; if ((x = ((AltResult)r).ex) != null) {
completeThrowable(x, r);
break tryComplete;
}
r = null;
} }
public final boolean exec() {
CompletableFuture<?> d; Throwable ex;
if ((d = this.dst) != null && d.result == null) {
try { try {
fn.accept(arg1, arg2); if (c != null && !c.claim())
ex = null; return false;
} catch (Throwable rex) { @SuppressWarnings("unchecked") S s = (S) r;
ex = rex; completeValue(f.apply(s));
} catch (Throwable ex) {
completeThrowable(ex);
} }
d.internalComplete(null, ex);
} }
return true; return true;
} }
private static final long serialVersionUID = 5232453952276885070L;
private <V> CompletableFuture<V> uniApplyStage(
Executor e, Function<? super T,? extends V> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<V> d = new CompletableFuture<V>();
if (e != null || !d.uniApply(this, f, null)) {
UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
push(c);
c.tryFire(SYNC);
}
return d;
} }
static final class AsyncCompose<T,U> extends Async { @SuppressWarnings("serial")
final T arg; static final class UniAccept<T> extends UniCompletion<T,Void> {
final Function<? super T, ? extends CompletionStage<U>> fn; Consumer<? super T> fn;
final CompletableFuture<U> dst; UniAccept(Executor executor, CompletableFuture<Void> dep,
AsyncCompose(T arg, CompletableFuture<T> src, Consumer<? super T> fn) {
Function<? super T, ? extends CompletionStage<U>> fn, super(executor, dep, src); this.fn = fn;
CompletableFuture<U> dst) { }
this.arg = arg; this.fn = fn; this.dst = dst; final CompletableFuture<Void> tryFire(int mode) {
CompletableFuture<Void> d; CompletableFuture<T> 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);
} }
public final boolean exec() { }
CompletableFuture<U> d, fr; U u; Throwable ex;
if ((d = this.dst) != null && d.result == null) { final <S> boolean uniAccept(CompletableFuture<S> a,
try { Consumer<? super S> f, UniAccept<S> c) {
CompletionStage<U> cs = fn.apply(arg); Object r; Throwable x;
fr = (cs == null) ? null : cs.toCompletableFuture(); if (a == null || (r = a.result) == null || f == null)
ex = (fr == null) ? new NullPointerException() : null; return false;
} catch (Throwable rex) { tryComplete: if (result == null) {
ex = rex;
fr = null;
}
if (ex != null)
u = null;
else {
Object r = fr.result;
if (r == null)
r = fr.waitingGet(false);
if (r instanceof AltResult) { if (r instanceof AltResult) {
ex = ((AltResult)r).ex; if ((x = ((AltResult)r).ex) != null) {
u = null; completeThrowable(x, r);
break tryComplete;
} }
else { r = null;
@SuppressWarnings("unchecked") U ur = (U) r;
u = ur;
} }
try {
if (c != null && !c.claim())
return false;
@SuppressWarnings("unchecked") S s = (S) r;
f.accept(s);
completeNull();
} catch (Throwable ex) {
completeThrowable(ex);
} }
d.internalComplete(u, ex);
} }
return true; return true;
} }
private static final long serialVersionUID = 5232453952276885070L;
private CompletableFuture<Void> uniAcceptStage(Executor e,
Consumer<? super T> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
if (e != null || !d.uniAccept(this, f, null)) {
UniAccept<T> c = new UniAccept<T>(e, d, this, f);
push(c);
c.tryFire(SYNC);
}
return d;
} }
static final class AsyncWhenComplete<T> extends Async { @SuppressWarnings("serial")
final T arg1; static final class UniRun<T> extends UniCompletion<T,Void> {
final Throwable arg2; Runnable fn;
final BiConsumer<? super T,? super Throwable> fn; UniRun(Executor executor, CompletableFuture<Void> dep,
final CompletableFuture<T> dst; CompletableFuture<T> src, Runnable fn) {
AsyncWhenComplete(T arg1, Throwable arg2, super(executor, dep, src); this.fn = fn;
BiConsumer<? super T,? super Throwable> fn, }
CompletableFuture<T> dst) { final CompletableFuture<Void> tryFire(int mode) {
this.arg1 = arg1; this.arg2 = arg2; this.fn = fn; this.dst = dst; CompletableFuture<Void> d; CompletableFuture<T> 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);
}
} }
public final boolean exec() {
CompletableFuture<T> d; final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) {
if ((d = this.dst) != null && d.result == null) { Object r; Throwable x;
Throwable ex = arg2; 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 { try {
fn.accept(arg1, ex); if (c != null && !c.claim())
} catch (Throwable rex) { return false;
if (ex == null) f.run();
ex = rex; completeNull();
} catch (Throwable ex) {
completeThrowable(ex);
} }
d.internalComplete(arg1, ex);
} }
return true; return true;
} }
private static final long serialVersionUID = 5232453952276885070L;
}
/* ------------- Completions -------------- */ private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {
if (f == null) throw new NullPointerException();
/** CompletableFuture<Void> d = new CompletableFuture<Void>();
* Simple linked list nodes to record completions, used in if (e != null || !d.uniRun(this, f, null)) {
* basically the same way as WaitNodes. (We separate nodes from UniRun<T> c = new UniRun<T>(e, d, this, f);
* the Completions themselves mainly because for the And and Or push(c);
* methods, the same Completion object resides in two lists.) c.tryFire(SYNC);
*/ }
static final class CompletionNode { return d;
final Completion completion;
volatile CompletionNode next;
CompletionNode(Completion completion) { this.completion = completion; }
} }
// Opportunistically subclass AtomicInteger to use compareAndSet to claim.
@SuppressWarnings("serial") @SuppressWarnings("serial")
abstract static class Completion extends AtomicInteger implements Runnable { static final class UniWhenComplete<T> extends UniCompletion<T,T> {
BiConsumer<? super T, ? super Throwable> fn;
UniWhenComplete(Executor executor, CompletableFuture<T> dep,
CompletableFuture<T> src,
BiConsumer<? super T, ? super Throwable> fn) {
super(executor, dep, src); this.fn = fn;
}
final CompletableFuture<T> tryFire(int mode) {
CompletableFuture<T> d; CompletableFuture<T> 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);
}
} }
static final class ThenApply<T,U> extends Completion { final boolean uniWhenComplete(CompletableFuture<T> a,
final CompletableFuture<? extends T> src; BiConsumer<? super T,? super Throwable> f,
final Function<? super T,? extends U> fn; UniWhenComplete<T> c) {
final CompletableFuture<U> dst; Object r; T t; Throwable x = null;
final Executor executor; if (a == null || (r = a.result) == null || f == null)
ThenApply(CompletableFuture<? extends T> src, return false;
Function<? super T,? extends U> fn, if (result == null) {
CompletableFuture<U> dst, try {
Executor executor) { if (c != null && !c.claim())
this.src = src; this.fn = fn; this.dst = dst; return false;
this.executor = executor;
}
public final void run() {
final CompletableFuture<? extends T> a;
final Function<? super T,? extends U> fn;
final CompletableFuture<U> dst;
Object r; T t; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) { if (r instanceof AltResult) {
ex = ((AltResult)r).ex; x = ((AltResult)r).ex;
t = null; t = null;
} } else {
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r; @SuppressWarnings("unchecked") T tr = (T) r;
t = tr; t = tr;
} }
Executor e = executor; f.accept(t, x);
U u = null; if (x == null) {
if (ex == null) { internalComplete(r);
try { return true;
if (e != null) }
execAsync(e, new AsyncApply<T,U>(t, fn, dst)); } catch (Throwable ex) {
else if (x == null)
u = fn.apply(t); x = ex;
} catch (Throwable rex) {
ex = rex;
} }
completeThrowable(x, r);
} }
if (e == null || ex != null) return true;
dst.internalComplete(u, ex);
} }
private CompletableFuture<T> uniWhenCompleteStage(
Executor e, BiConsumer<? super T, ? super Throwable> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<T> d = new CompletableFuture<T>();
if (e != null || !d.uniWhenComplete(this, f, null)) {
UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);
push(c);
c.tryFire(SYNC);
} }
private static final long serialVersionUID = 5232453952276885070L; return d;
} }
static final class ThenAccept<T> extends Completion { @SuppressWarnings("serial")
final CompletableFuture<? extends T> src; static final class UniHandle<T,V> extends UniCompletion<T,V> {
final Consumer<? super T> fn; BiFunction<? super T, Throwable, ? extends V> fn;
final CompletableFuture<?> dst; UniHandle(Executor executor, CompletableFuture<V> dep,
final Executor executor; CompletableFuture<T> src,
ThenAccept(CompletableFuture<? extends T> src, BiFunction<? super T, Throwable, ? extends V> fn) {
Consumer<? super T> fn, super(executor, dep, src); this.fn = fn;
CompletableFuture<?> dst, }
Executor executor) { final CompletableFuture<V> tryFire(int mode) {
this.src = src; this.fn = fn; this.dst = dst; CompletableFuture<V> d; CompletableFuture<T> a;
this.executor = executor; if ((d = dep) == null ||
} !d.uniHandle(a = src, fn, mode > 0 ? null : this))
public final void run() { return null;
final CompletableFuture<? extends T> a; dep = null; src = null; fn = null;
final Consumer<? super T> fn; return d.postFire(a, mode);
final CompletableFuture<?> dst;
Object r; T t; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
} }
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
Executor e = executor;
if (ex == null) { final <S> boolean uniHandle(CompletableFuture<S> a,
BiFunction<? super S, Throwable, ? extends T> f,
UniHandle<S,T> c) {
Object r; S s; Throwable x;
if (a == null || (r = a.result) == null || f == null)
return false;
if (result == null) {
try { try {
if (e != null) if (c != null && !c.claim())
execAsync(e, new AsyncAccept<T>(t, fn, dst)); return false;
else if (r instanceof AltResult) {
fn.accept(t); x = ((AltResult)r).ex;
} catch (Throwable rex) { s = null;
ex = rex; } else {
x = null;
@SuppressWarnings("unchecked") S ss = (S) r;
s = ss;
}
completeValue(f.apply(s, x));
} catch (Throwable ex) {
completeThrowable(ex);
} }
} }
if (e == null || ex != null) return true;
dst.internalComplete(null, ex);
} }
private <V> CompletableFuture<V> uniHandleStage(
Executor e, BiFunction<? super T, Throwable, ? extends V> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<V> d = new CompletableFuture<V>();
if (e != null || !d.uniHandle(this, f, null)) {
UniHandle<T,V> c = new UniHandle<T,V>(e, d, this, f);
push(c);
c.tryFire(SYNC);
} }
private static final long serialVersionUID = 5232453952276885070L; return d;
} }
static final class ThenRun extends Completion { @SuppressWarnings("serial")
final CompletableFuture<?> src; static final class UniExceptionally<T> extends UniCompletion<T,T> {
final Runnable fn; Function<? super Throwable, ? extends T> fn;
final CompletableFuture<Void> dst; UniExceptionally(CompletableFuture<T> dep, CompletableFuture<T> src,
final Executor executor; Function<? super Throwable, ? extends T> fn) {
ThenRun(CompletableFuture<?> src, super(null, dep, src); this.fn = fn;
Runnable fn, }
CompletableFuture<Void> dst, final CompletableFuture<T> tryFire(int mode) { // never ASYNC
Executor executor) { // assert mode != ASYNC;
this.src = src; this.fn = fn; this.dst = dst; CompletableFuture<T> d; CompletableFuture<T> a;
this.executor = executor; if ((d = dep) == null || !d.uniExceptionally(a = src, fn, this))
} return null;
public final void run() { dep = null; src = null; fn = null;
final CompletableFuture<?> a; return d.postFire(a, mode);
final Runnable fn;
final CompletableFuture<Void> dst;
Object r; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult)
ex = ((AltResult)r).ex;
else
ex = null;
Executor e = executor;
if (ex == null) {
try {
if (e != null)
execAsync(e, new AsyncRun(fn, dst));
else
fn.run();
} catch (Throwable rex) {
ex = rex;
}
}
if (e == null || ex != null)
dst.internalComplete(null, ex);
}
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class ThenCombine<T,U,V> extends Completion { final boolean uniExceptionally(CompletableFuture<T> a,
final CompletableFuture<? extends T> src; Function<? super Throwable, ? extends T> f,
final CompletableFuture<? extends U> snd; UniExceptionally<T> c) {
final BiFunction<? super T,? super U,? extends V> fn; Object r; Throwable x;
final CompletableFuture<V> dst; if (a == null || (r = a.result) == null || f == null)
final Executor executor; return false;
ThenCombine(CompletableFuture<? extends T> src, if (result == null) {
CompletableFuture<? extends U> snd,
BiFunction<? super T,? super U,? extends V> fn,
CompletableFuture<V> dst,
Executor executor) {
this.src = src; this.snd = snd;
this.fn = fn; this.dst = dst;
this.executor = executor;
}
public final void run() {
final CompletableFuture<? extends T> a;
final CompletableFuture<? extends U> b;
final BiFunction<? super T,? super U,? extends V> fn;
final CompletableFuture<V> dst;
Object r, s; T t; U u; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
(b = this.snd) != null &&
(s = b.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
}
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
}
if (ex != null)
u = null;
else if (s instanceof AltResult) {
ex = ((AltResult)s).ex;
u = null;
}
else {
@SuppressWarnings("unchecked") U us = (U) s;
u = us;
}
Executor e = executor;
V v = null;
if (ex == null) {
try { try {
if (e != null) if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) {
execAsync(e, new AsyncCombine<T,U,V>(t, u, fn, dst)); if (c != null && !c.claim())
else return false;
v = fn.apply(t, u); completeValue(f.apply(x));
} catch (Throwable rex) { } else
ex = rex; internalComplete(r);
} } catch (Throwable ex) {
} completeThrowable(ex);
if (e == null || ex != null)
dst.internalComplete(v, ex);
} }
} }
private static final long serialVersionUID = 5232453952276885070L; return true;
} }
static final class ThenAcceptBoth<T,U> extends Completion { private CompletableFuture<T> uniExceptionallyStage(
final CompletableFuture<? extends T> src; Function<Throwable, ? extends T> f) {
final CompletableFuture<? extends U> snd; if (f == null) throw new NullPointerException();
final BiConsumer<? super T,? super U> fn; CompletableFuture<T> d = new CompletableFuture<T>();
final CompletableFuture<Void> dst; if (!d.uniExceptionally(this, f, null)) {
final Executor executor; UniExceptionally<T> c = new UniExceptionally<T>(d, this, f);
ThenAcceptBoth(CompletableFuture<? extends T> src, push(c);
CompletableFuture<? extends U> snd, c.tryFire(SYNC);
BiConsumer<? super T,? super U> fn,
CompletableFuture<Void> dst,
Executor executor) {
this.src = src; this.snd = snd;
this.fn = fn; this.dst = dst;
this.executor = executor;
}
public final void run() {
final CompletableFuture<? extends T> a;
final CompletableFuture<? extends U> b;
final BiConsumer<? super T,? super U> fn;
final CompletableFuture<Void> dst;
Object r, s; T t; U u; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
(b = this.snd) != null &&
(s = b.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
}
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
if (ex != null) return d;
u = null;
else if (s instanceof AltResult) {
ex = ((AltResult)s).ex;
u = null;
} }
else {
@SuppressWarnings("unchecked") U us = (U) s; @SuppressWarnings("serial")
u = us; static final class UniRelay<T> extends UniCompletion<T,T> { // for Compose
} UniRelay(CompletableFuture<T> dep, CompletableFuture<T> src) {
Executor e = executor; super(null, dep, src);
if (ex == null) {
try {
if (e != null)
execAsync(e, new AsyncAcceptBoth<T,U>(t, u, fn, dst));
else
fn.accept(t, u);
} catch (Throwable rex) {
ex = rex;
}
}
if (e == null || ex != null)
dst.internalComplete(null, ex);
} }
final CompletableFuture<T> tryFire(int mode) {
CompletableFuture<T> d; CompletableFuture<T> a;
if ((d = dep) == null || !d.uniRelay(a = src))
return null;
src = null; dep = null;
return d.postFire(a, mode);
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class RunAfterBoth extends Completion { final boolean uniRelay(CompletableFuture<T> a) {
final CompletableFuture<?> src; Object r;
final CompletableFuture<?> snd; if (a == null || (r = a.result) == null)
final Runnable fn; return false;
final CompletableFuture<Void> dst; if (result == null) // no need to claim
final Executor executor; completeRelay(r);
RunAfterBoth(CompletableFuture<?> src, return true;
CompletableFuture<?> snd,
Runnable fn,
CompletableFuture<Void> dst,
Executor executor) {
this.src = src; this.snd = snd;
this.fn = fn; this.dst = dst;
this.executor = executor;
}
public final void run() {
final CompletableFuture<?> a;
final CompletableFuture<?> b;
final Runnable fn;
final CompletableFuture<Void> dst;
Object r, s; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
(b = this.snd) != null &&
(s = b.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult)
ex = ((AltResult)r).ex;
else
ex = null;
if (ex == null && (s instanceof AltResult))
ex = ((AltResult)s).ex;
Executor e = executor;
if (ex == null) {
try {
if (e != null)
execAsync(e, new AsyncRun(fn, dst));
else
fn.run();
} catch (Throwable rex) {
ex = rex;
}
}
if (e == null || ex != null)
dst.internalComplete(null, ex);
}
}
private static final long serialVersionUID = 5232453952276885070L;
}
static final class AndCompletion extends Completion {
final CompletableFuture<?> src;
final CompletableFuture<?> snd;
final CompletableFuture<Void> dst;
AndCompletion(CompletableFuture<?> src,
CompletableFuture<?> snd,
CompletableFuture<Void> dst) {
this.src = src; this.snd = snd; this.dst = dst;
}
public final void run() {
final CompletableFuture<?> a;
final CompletableFuture<?> b;
final CompletableFuture<Void> dst;
Object r, s; Throwable ex;
if ((dst = this.dst) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
(b = this.snd) != null &&
(s = b.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult)
ex = ((AltResult)r).ex;
else
ex = null;
if (ex == null && (s instanceof AltResult))
ex = ((AltResult)s).ex;
dst.internalComplete(null, ex);
} }
@SuppressWarnings("serial")
static final class UniCompose<T,V> extends UniCompletion<T,V> {
Function<? super T, ? extends CompletionStage<V>> fn;
UniCompose(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src,
Function<? super T, ? extends CompletionStage<V>> fn) {
super(executor, dep, src); this.fn = fn;
}
final CompletableFuture<V> tryFire(int mode) {
CompletableFuture<V> d; CompletableFuture<T> 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);
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class ApplyToEither<T,U> extends Completion { final <S> boolean uniCompose(
final CompletableFuture<? extends T> src; CompletableFuture<S> a,
final CompletableFuture<? extends T> snd; Function<? super S, ? extends CompletionStage<T>> f,
final Function<? super T,? extends U> fn; UniCompose<S,T> c) {
final CompletableFuture<U> dst; Object r; Throwable x;
final Executor executor; if (a == null || (r = a.result) == null || f == null)
ApplyToEither(CompletableFuture<? extends T> src, return false;
CompletableFuture<? extends T> snd, tryComplete: if (result == null) {
Function<? super T,? extends U> fn,
CompletableFuture<U> dst,
Executor executor) {
this.src = src; this.snd = snd;
this.fn = fn; this.dst = dst;
this.executor = executor;
}
public final void run() {
final CompletableFuture<? extends T> a;
final CompletableFuture<? extends T> b;
final Function<? super T,? extends U> fn;
final CompletableFuture<U> dst;
Object r; T t; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(((a = this.src) != null && (r = a.result) != null) ||
((b = this.snd) != null && (r = b.result) != null)) &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) { if (r instanceof AltResult) {
ex = ((AltResult)r).ex; if ((x = ((AltResult)r).ex) != null) {
t = null; completeThrowable(x, r);
break tryComplete;
} }
else { r = null;
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
Executor e = executor;
U u = null;
if (ex == null) {
try { try {
if (e != null) if (c != null && !c.claim())
execAsync(e, new AsyncApply<T,U>(t, fn, dst)); return false;
else @SuppressWarnings("unchecked") S s = (S) r;
u = fn.apply(t); CompletableFuture<T> g = f.apply(s).toCompletableFuture();
} catch (Throwable rex) { if (g.result == null || !uniRelay(g)) {
ex = rex; UniRelay<T> copy = new UniRelay<T>(this, g);
} g.push(copy);
copy.tryFire(SYNC);
if (result == null)
return false;
} }
if (e == null || ex != null) } catch (Throwable ex) {
dst.internalComplete(u, ex); completeThrowable(ex);
} }
} }
private static final long serialVersionUID = 5232453952276885070L; return true;
} }
static final class AcceptEither<T> extends Completion { private <V> CompletableFuture<V> uniComposeStage(
final CompletableFuture<? extends T> src; Executor e, Function<? super T, ? extends CompletionStage<V>> f) {
final CompletableFuture<? extends T> snd; if (f == null) throw new NullPointerException();
final Consumer<? super T> fn; Object r; Throwable x;
final CompletableFuture<Void> dst; if (e == null && (r = result) != null) {
final Executor executor; // try to return function result directly
AcceptEither(CompletableFuture<? extends T> src,
CompletableFuture<? extends T> snd,
Consumer<? super T> fn,
CompletableFuture<Void> dst,
Executor executor) {
this.src = src; this.snd = snd;
this.fn = fn; this.dst = dst;
this.executor = executor;
}
public final void run() {
final CompletableFuture<? extends T> a;
final CompletableFuture<? extends T> b;
final Consumer<? super T> fn;
final CompletableFuture<Void> dst;
Object r; T t; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(((a = this.src) != null && (r = a.result) != null) ||
((b = this.snd) != null && (r = b.result) != null)) &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) { if (r instanceof AltResult) {
ex = ((AltResult)r).ex; if ((x = ((AltResult)r).ex) != null) {
t = null; return new CompletableFuture<V>(encodeThrowable(x, r));
} }
else { r = null;
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
Executor e = executor;
if (ex == null) {
try { try {
if (e != null) @SuppressWarnings("unchecked") T t = (T) r;
execAsync(e, new AsyncAccept<T>(t, fn, dst)); return f.apply(t).toCompletableFuture();
else } catch (Throwable ex) {
fn.accept(t); return new CompletableFuture<V>(encodeThrowable(ex));
} catch (Throwable rex) {
ex = rex;
}
} }
if (e == null || ex != null)
dst.internalComplete(null, ex);
} }
} CompletableFuture<V> d = new CompletableFuture<V>();
private static final long serialVersionUID = 5232453952276885070L; UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);
push(c);
c.tryFire(SYNC);
return d;
} }
static final class RunAfterEither extends Completion { /* ------------- Two-input Completions -------------- */
final CompletableFuture<?> src;
final CompletableFuture<?> snd;
final Runnable fn;
final CompletableFuture<Void> dst;
final Executor executor;
RunAfterEither(CompletableFuture<?> src,
CompletableFuture<?> snd,
Runnable fn,
CompletableFuture<Void> dst,
Executor executor) {
this.src = src; this.snd = snd;
this.fn = fn; this.dst = dst;
this.executor = executor;
}
public final void run() {
final CompletableFuture<?> a;
final CompletableFuture<?> b;
final Runnable fn;
final CompletableFuture<Void> dst;
Object r; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(((a = this.src) != null && (r = a.result) != null) ||
((b = this.snd) != null && (r = b.result) != null)) &&
compareAndSet(0, 1)) {
if (r instanceof AltResult)
ex = ((AltResult)r).ex;
else
ex = null;
Executor e = executor;
if (ex == null) {
try {
if (e != null)
execAsync(e, new AsyncRun(fn, dst));
else
fn.run();
} catch (Throwable rex) {
ex = rex;
}
}
if (e == null || ex != null)
dst.internalComplete(null, ex);
}
}
private static final long serialVersionUID = 5232453952276885070L;
}
static final class OrCompletion extends Completion { /** A Completion for an action with two sources */
final CompletableFuture<?> src; @SuppressWarnings("serial")
final CompletableFuture<?> snd; abstract static class BiCompletion<T,U,V> extends UniCompletion<T,V> {
final CompletableFuture<Object> dst; CompletableFuture<U> snd; // second source for action
OrCompletion(CompletableFuture<?> src, BiCompletion(Executor executor, CompletableFuture<V> dep,
CompletableFuture<?> snd, CompletableFuture<T> src, CompletableFuture<U> snd) {
CompletableFuture<Object> dst) { super(executor, dep, src); this.snd = snd;
this.src = src; this.snd = snd; this.dst = dst;
} }
public final void run() {
final CompletableFuture<?> a;
final CompletableFuture<?> b;
final CompletableFuture<Object> dst;
Object r, t; Throwable ex;
if ((dst = this.dst) != null &&
(((a = this.src) != null && (r = a.result) != null) ||
((b = this.snd) != null && (r = b.result) != null)) &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
} }
else {
ex = null; /** A Completion delegating to a BiCompletion */
t = r; @SuppressWarnings("serial")
} static final class CoCompletion extends Completion {
dst.internalComplete(t, ex); BiCompletion<?,?,?> base;
} CoCompletion(BiCompletion<?,?,?> base) { this.base = base; }
} final CompletableFuture<?> tryFire(int mode) {
private static final long serialVersionUID = 5232453952276885070L; BiCompletion<?,?,?> c; CompletableFuture<?> d;
} if ((c = base) == null || (d = c.tryFire(mode)) == null)
return null;
static final class ExceptionCompletion<T> extends Completion { base = null; // detach
final CompletableFuture<? extends T> src; return d;
final Function<? super Throwable, ? extends T> fn;
final CompletableFuture<T> dst;
ExceptionCompletion(CompletableFuture<? extends T> src,
Function<? super Throwable, ? extends T> fn,
CompletableFuture<T> dst) {
this.src = src; this.fn = fn; this.dst = dst;
}
public final void run() {
final CompletableFuture<? extends T> a;
final Function<? super Throwable, ? extends T> fn;
final CompletableFuture<T> dst;
Object r; T t = null; Throwable ex, dx = null;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
compareAndSet(0, 1)) {
if ((r instanceof AltResult) &&
(ex = ((AltResult)r).ex) != null) {
try {
t = fn.apply(ex);
} catch (Throwable rex) {
dx = rex;
} }
final boolean isLive() {
BiCompletion<?,?,?> c;
return (c = base) != null && c.dep != null;
} }
else {
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
dst.internalComplete(t, dx);
/** 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
} }
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class WhenCompleteCompletion<T> extends Completion { /** Post-processing after successful BiCompletion tryFire. */
final CompletableFuture<? extends T> src; final CompletableFuture<T> postFire(CompletableFuture<?> a,
final BiConsumer<? super T, ? super Throwable> fn; CompletableFuture<?> b, int mode) {
final CompletableFuture<T> dst; if (b != null && b.stack != null) { // clean second source
final Executor executor; if (mode < 0 || b.result == null)
WhenCompleteCompletion(CompletableFuture<? extends T> src, b.cleanStack();
BiConsumer<? super T, ? super Throwable> fn,
CompletableFuture<T> dst,
Executor executor) {
this.src = src; this.fn = fn; this.dst = dst;
this.executor = executor;
}
public final void run() {
final CompletableFuture<? extends T> a;
final BiConsumer<? super T, ? super Throwable> fn;
final CompletableFuture<T> dst;
Object r; T t; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
}
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
}
Executor e = executor;
Throwable dx = null;
try {
if (e != null)
execAsync(e, new AsyncWhenComplete<T>(t, ex, fn, dst));
else else
fn.accept(t, ex); b.postComplete();
} catch (Throwable rex) {
dx = rex;
} }
if (e == null || dx != null) return postFire(a, mode);
dst.internalComplete(t, ex != null ? ex : dx);
} }
@SuppressWarnings("serial")
static final class BiApply<T,U,V> extends BiCompletion<T,U,V> {
BiFunction<? super T,? super U,? extends V> fn;
BiApply(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src, CompletableFuture<U> snd,
BiFunction<? super T,? super U,? extends V> fn) {
super(executor, dep, src, snd); this.fn = fn;
}
final CompletableFuture<V> tryFire(int mode) {
CompletableFuture<V> d;
CompletableFuture<T> a;
CompletableFuture<U> 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);
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class ThenCopy<T> extends Completion { final <R,S> boolean biApply(CompletableFuture<R> a,
final CompletableFuture<?> src; CompletableFuture<S> b,
final CompletableFuture<T> dst; BiFunction<? super R,? super S,? extends T> f,
ThenCopy(CompletableFuture<?> src, BiApply<R,S,T> c) {
CompletableFuture<T> dst) { Object r, s; Throwable x;
this.src = src; this.dst = dst; if (a == null || (r = a.result) == null ||
} b == null || (s = b.result) == null || f == null)
public final void run() { return false;
final CompletableFuture<?> a; tryComplete: if (result == null) {
final CompletableFuture<T> dst;
Object r; T t; Throwable ex;
if ((dst = this.dst) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) { if (r instanceof AltResult) {
ex = ((AltResult)r).ex; if ((x = ((AltResult)r).ex) != null) {
t = null; completeThrowable(x, r);
} break tryComplete;
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
}
dst.internalComplete(t, ex);
} }
r = null;
} }
private static final long serialVersionUID = 5232453952276885070L; if (s instanceof AltResult) {
if ((x = ((AltResult)s).ex) != null) {
completeThrowable(x, s);
break tryComplete;
} }
s = null;
// version of ThenCopy for CompletableFuture<Void> dst
static final class ThenPropagate extends Completion {
final CompletableFuture<?> src;
final CompletableFuture<Void> dst;
ThenPropagate(CompletableFuture<?> src,
CompletableFuture<Void> dst) {
this.src = src; this.dst = dst;
} }
public final void run() { try {
final CompletableFuture<?> a; if (c != null && !c.claim())
final CompletableFuture<Void> dst; return false;
Object r; Throwable ex; @SuppressWarnings("unchecked") R rr = (R) r;
if ((dst = this.dst) != null && @SuppressWarnings("unchecked") S ss = (S) s;
(a = this.src) != null && completeValue(f.apply(rr, ss));
(r = a.result) != null && } catch (Throwable ex) {
compareAndSet(0, 1)) { completeThrowable(ex);
if (r instanceof AltResult)
ex = ((AltResult)r).ex;
else
ex = null;
dst.internalComplete(null, ex);
} }
} }
private static final long serialVersionUID = 5232453952276885070L; return true;
} }
static final class HandleCompletion<T,U> extends Completion { private <U,V> CompletableFuture<V> biApplyStage(
final CompletableFuture<? extends T> src; Executor e, CompletionStage<U> o,
final BiFunction<? super T, Throwable, ? extends U> fn; BiFunction<? super T,? super U,? extends V> f) {
final CompletableFuture<U> dst; CompletableFuture<U> b;
final Executor executor; if (f == null || (b = o.toCompletableFuture()) == null)
HandleCompletion(CompletableFuture<? extends T> src, throw new NullPointerException();
BiFunction<? super T, Throwable, ? extends U> fn, CompletableFuture<V> d = new CompletableFuture<V>();
CompletableFuture<U> dst, if (e != null || !d.biApply(this, b, f, null)) {
Executor executor) { BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);
this.src = src; this.fn = fn; this.dst = dst; bipush(b, c);
this.executor = executor; c.tryFire(SYNC);
}
public final void run() {
final CompletableFuture<? extends T> a;
final BiFunction<? super T, Throwable, ? extends U> fn;
final CompletableFuture<U> dst;
Object r; T t; Throwable ex;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
}
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
}
Executor e = executor;
U u = null;
Throwable dx = null;
try {
if (e != null)
execAsync(e, new AsyncCombine<T,Throwable,U>(t, ex, fn, dst));
else
u = fn.apply(t, ex);
} catch (Throwable rex) {
dx = rex;
} }
if (e == null || dx != null) return d;
dst.internalComplete(u, dx);
} }
@SuppressWarnings("serial")
static final class BiAccept<T,U> extends BiCompletion<T,U,Void> {
BiConsumer<? super T,? super U> fn;
BiAccept(Executor executor, CompletableFuture<Void> dep,
CompletableFuture<T> src, CompletableFuture<U> snd,
BiConsumer<? super T,? super U> fn) {
super(executor, dep, src, snd); this.fn = fn;
}
final CompletableFuture<Void> tryFire(int mode) {
CompletableFuture<Void> d;
CompletableFuture<T> a;
CompletableFuture<U> 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);
} }
private static final long serialVersionUID = 5232453952276885070L;
} }
static final class ThenCompose<T,U> extends Completion { final <R,S> boolean biAccept(CompletableFuture<R> a,
final CompletableFuture<? extends T> src; CompletableFuture<S> b,
final Function<? super T, ? extends CompletionStage<U>> fn; BiConsumer<? super R,? super S> f,
final CompletableFuture<U> dst; BiAccept<R,S> c) {
final Executor executor; Object r, s; Throwable x;
ThenCompose(CompletableFuture<? extends T> src, if (a == null || (r = a.result) == null ||
Function<? super T, ? extends CompletionStage<U>> fn, b == null || (s = b.result) == null || f == null)
CompletableFuture<U> dst, return false;
Executor executor) { tryComplete: if (result == null) {
this.src = src; this.fn = fn; this.dst = dst;
this.executor = executor;
}
public final void run() {
final CompletableFuture<? extends T> a;
final Function<? super T, ? extends CompletionStage<U>> fn;
final CompletableFuture<U> dst;
Object r; T t; Throwable ex; Executor e;
if ((dst = this.dst) != null &&
(fn = this.fn) != null &&
(a = this.src) != null &&
(r = a.result) != null &&
compareAndSet(0, 1)) {
if (r instanceof AltResult) { if (r instanceof AltResult) {
ex = ((AltResult)r).ex; if ((x = ((AltResult)r).ex) != null) {
t = null; completeThrowable(x, r);
} break tryComplete;
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
}
CompletableFuture<U> c = null;
U u = null;
boolean complete = false;
if (ex == null) {
if ((e = executor) != null)
execAsync(e, new AsyncCompose<T,U>(t, fn, dst));
else {
try {
CompletionStage<U> cs = fn.apply(t);
c = (cs == null) ? null : cs.toCompletableFuture();
if (c == null)
ex = new NullPointerException();
} catch (Throwable rex) {
ex = rex;
}
}
}
if (c != null) {
ThenCopy<U> d = null;
Object s;
if ((s = c.result) == null) {
CompletionNode p = new CompletionNode
(d = new ThenCopy<U>(c, dst));
while ((s = c.result) == null) {
if (UNSAFE.compareAndSwapObject
(c, COMPLETIONS, p.next = c.completions, p))
break;
} }
r = null;
} }
if (s != null && (d == null || d.compareAndSet(0, 1))) {
complete = true;
if (s instanceof AltResult) { if (s instanceof AltResult) {
ex = ((AltResult)s).ex; // no rewrap if ((x = ((AltResult)s).ex) != null) {
u = null; completeThrowable(x, s);
} break tryComplete;
else {
@SuppressWarnings("unchecked") U us = (U) s;
u = us;
} }
s = null;
} }
} try {
if (complete || ex != null) if (c != null && !c.claim())
dst.internalComplete(u, ex); return false;
if (c != null) @SuppressWarnings("unchecked") R rr = (R) r;
c.helpPostComplete(); @SuppressWarnings("unchecked") S ss = (S) s;
f.accept(rr, ss);
completeNull();
} catch (Throwable ex) {
completeThrowable(ex);
} }
} }
private static final long serialVersionUID = 5232453952276885070L; return true;
} }
// Implementations of stage methods with (plain, async, Executor) forms private <U> CompletableFuture<Void> biAcceptStage(
Executor e, CompletionStage<U> o,
private <U> CompletableFuture<U> doThenApply BiConsumer<? super T,? super U> f) {
(Function<? super T,? extends U> fn, CompletableFuture<U> b;
Executor e) { if (f == null || (b = o.toCompletableFuture()) == null)
if (fn == null) throw new NullPointerException(); throw new NullPointerException();
CompletableFuture<U> dst = new CompletableFuture<U>(); CompletableFuture<Void> d = new CompletableFuture<Void>();
ThenApply<T,U> d = null; if (e != null || !d.biAccept(this, b, f, null)) {
Object r; BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
if ((r = result) == null) { bipush(b, c);
CompletionNode p = new CompletionNode c.tryFire(SYNC);
(d = new ThenApply<T,U>(this, fn, dst, e));
while ((r = result) == null) {
if (UNSAFE.compareAndSwapObject
(this, COMPLETIONS, p.next = completions, p))
break;
} }
return d;
} }
if (r != null && (d == null || d.compareAndSet(0, 1))) {
T t; Throwable ex; @SuppressWarnings("serial")
if (r instanceof AltResult) { static final class BiRun<T,U> extends BiCompletion<T,U,Void> {
ex = ((AltResult)r).ex; Runnable fn;
t = null; BiRun(Executor executor, CompletableFuture<Void> dep,
CompletableFuture<T> src,
CompletableFuture<U> snd,
Runnable fn) {
super(executor, dep, src, snd); this.fn = fn;
}
final CompletableFuture<Void> tryFire(int mode) {
CompletableFuture<Void> d;
CompletableFuture<T> a;
CompletableFuture<U> 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);
} }
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
U u = null;
if (ex == null) { final boolean biRun(CompletableFuture<?> a, CompletableFuture<?> b,
try { Runnable f, BiRun<?,?> c) {
if (e != null) Object r, s; Throwable x;
execAsync(e, new AsyncApply<T,U>(t, fn, dst)); 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 else
u = fn.apply(t); try {
} catch (Throwable rex) { if (c != null && !c.claim())
ex = rex; return false;
} f.run();
completeNull();
} catch (Throwable ex) {
completeThrowable(ex);
} }
if (e == null || ex != null)
dst.internalComplete(u, ex);
} }
helpPostComplete(); return true;
return dst;
} }
private CompletableFuture<Void> doThenAccept(Consumer<? super T> fn, private CompletableFuture<Void> biRunStage(Executor e, CompletionStage<?> o,
Executor e) { Runnable f) {
if (fn == null) throw new NullPointerException(); CompletableFuture<?> b;
CompletableFuture<Void> dst = new CompletableFuture<Void>(); if (f == null || (b = o.toCompletableFuture()) == null)
ThenAccept<T> d = null; throw new NullPointerException();
Object r; CompletableFuture<Void> d = new CompletableFuture<Void>();
if ((r = result) == null) { if (e != null || !d.biRun(this, b, f, null)) {
CompletionNode p = new CompletionNode BiRun<T,?> c = new BiRun<>(e, d, this, b, f);
(d = new ThenAccept<T>(this, fn, dst, e)); bipush(b, c);
while ((r = result) == null) { c.tryFire(SYNC);
if (UNSAFE.compareAndSwapObject
(this, COMPLETIONS, p.next = completions, p))
break;
} }
return d;
} }
if (r != null && (d == null || d.compareAndSet(0, 1))) {
T t; Throwable ex; @SuppressWarnings("serial")
if (r instanceof AltResult) { static final class BiRelay<T,U> extends BiCompletion<T,U,Void> { // for And
ex = ((AltResult)r).ex; BiRelay(CompletableFuture<Void> dep,
t = null; CompletableFuture<T> src,
CompletableFuture<U> snd) {
super(null, dep, src, snd);
}
final CompletableFuture<Void> tryFire(int mode) {
CompletableFuture<Void> d;
CompletableFuture<T> a;
CompletableFuture<U> 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);
} }
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
if (ex == null) {
try { boolean biRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
if (e != null) Object r, s; Throwable x;
execAsync(e, new AsyncAccept<T>(t, fn, dst)); 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 else
fn.accept(t); completeNull();
} catch (Throwable rex) {
ex = rex;
}
} }
if (e == null || ex != null) return true;
dst.internalComplete(null, ex);
}
helpPostComplete();
return dst;
} }
private CompletableFuture<Void> doThenRun(Runnable action, /** Recursively constructs a tree of completions. */
Executor e) { static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs,
if (action == null) throw new NullPointerException(); int lo, int hi) {
CompletableFuture<Void> dst = new CompletableFuture<Void>(); CompletableFuture<Void> d = new CompletableFuture<Void>();
ThenRun d = null; if (lo > hi) // empty
Object r; d.result = NIL;
if ((r = result) == null) { else {
CompletionNode p = new CompletionNode CompletableFuture<?> a, b;
(d = new ThenRun(this, action, dst, e)); int mid = (lo + hi) >>> 1;
while ((r = result) == null) { if ((a = (lo == mid ? cfs[lo] :
if (UNSAFE.compareAndSwapObject andTree(cfs, lo, mid))) == null ||
(this, COMPLETIONS, p.next = completions, p)) (b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
break; 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);
} }
} }
if (r != null && (d == null || d.compareAndSet(0, 1))) { return d;
Throwable ex;
if (r instanceof AltResult)
ex = ((AltResult)r).ex;
else
ex = null;
if (ex == null) {
try {
if (e != null)
execAsync(e, new AsyncRun(action, dst));
else
action.run();
} catch (Throwable rex) {
ex = rex;
}
}
if (e == null || ex != null)
dst.internalComplete(null, ex);
}
helpPostComplete();
return dst;
}
private <U,V> CompletableFuture<V> doThenCombine
(CompletableFuture<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn,
Executor e) {
if (other == null || fn == null) throw new NullPointerException();
CompletableFuture<V> dst = new CompletableFuture<V>();
ThenCombine<T,U,V> d = null;
Object r, s = null;
if ((r = result) == null || (s = other.result) == null) {
d = new ThenCombine<T,U,V>(this, other, fn, dst, e);
CompletionNode q = null, p = new CompletionNode(d);
while ((r == null && (r = result) == null) ||
(s == null && (s = other.result) == null)) {
if (q != null) {
if (s != null ||
UNSAFE.compareAndSwapObject
(other, COMPLETIONS, q.next = other.completions, q))
break;
} }
else if (r != null ||
UNSAFE.compareAndSwapObject /* ------------- Projected (Ored) BiCompletions -------------- */
(this, COMPLETIONS, p.next = completions, p)) {
if (s != null) /** Pushes completion to this and b unless either done. */
break; final void orpush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
q = new CompletionNode(d); 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
} }
if (r != null && s != null && (d == null || d.compareAndSet(0, 1))) {
T t; U u; Throwable ex;
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
} }
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
if (ex != null)
u = null; @SuppressWarnings("serial")
else if (s instanceof AltResult) { static final class OrApply<T,U extends T,V> extends BiCompletion<T,U,V> {
ex = ((AltResult)s).ex; Function<? super T,? extends V> fn;
u = null; OrApply(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src,
CompletableFuture<U> snd,
Function<? super T,? extends V> fn) {
super(executor, dep, src, snd); this.fn = fn;
}
final CompletableFuture<V> tryFire(int mode) {
CompletableFuture<V> d;
CompletableFuture<T> a;
CompletableFuture<U> 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);
} }
else {
@SuppressWarnings("unchecked") U us = (U) s;
u = us;
} }
V v = null;
if (ex == null) { final <R,S extends R> boolean orApply(CompletableFuture<R> a,
CompletableFuture<S> b,
Function<? super R, ? extends T> f,
OrApply<R,S,T> 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 { try {
if (e != null) if (c != null && !c.claim())
execAsync(e, new AsyncCombine<T,U,V>(t, u, fn, dst)); return false;
else
v = fn.apply(t, u);
} catch (Throwable rex) {
ex = rex;
}
}
if (e == null || ex != null)
dst.internalComplete(v, ex);
}
helpPostComplete();
other.helpPostComplete();
return dst;
}
private <U> CompletableFuture<Void> doThenAcceptBoth
(CompletableFuture<? extends U> other,
BiConsumer<? super T,? super U> fn,
Executor e) {
if (other == null || fn == null) throw new NullPointerException();
CompletableFuture<Void> dst = new CompletableFuture<Void>();
ThenAcceptBoth<T,U> d = null;
Object r, s = null;
if ((r = result) == null || (s = other.result) == null) {
d = new ThenAcceptBoth<T,U>(this, other, fn, dst, e);
CompletionNode q = null, p = new CompletionNode(d);
while ((r == null && (r = result) == null) ||
(s == null && (s = other.result) == null)) {
if (q != null) {
if (s != null ||
UNSAFE.compareAndSwapObject
(other, COMPLETIONS, q.next = other.completions, q))
break;
}
else if (r != null ||
UNSAFE.compareAndSwapObject
(this, COMPLETIONS, p.next = completions, p)) {
if (s != null)
break;
q = new CompletionNode(d);
}
}
}
if (r != null && s != null && (d == null || d.compareAndSet(0, 1))) {
T t; U u; Throwable ex;
if (r instanceof AltResult) { if (r instanceof AltResult) {
ex = ((AltResult)r).ex; if ((x = ((AltResult)r).ex) != null) {
t = null; completeThrowable(x, r);
break tryComplete;
} }
else { r = null;
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
if (ex != null) @SuppressWarnings("unchecked") R rr = (R) r;
u = null; completeValue(f.apply(rr));
else if (s instanceof AltResult) { } catch (Throwable ex) {
ex = ((AltResult)s).ex; completeThrowable(ex);
u = null;
}
else {
@SuppressWarnings("unchecked") U us = (U) s;
u = us;
} }
if (ex == null) {
try {
if (e != null)
execAsync(e, new AsyncAcceptBoth<T,U>(t, u, fn, dst));
else
fn.accept(t, u);
} catch (Throwable rex) {
ex = rex;
} }
return true;
} }
if (e == null || ex != null)
dst.internalComplete(null, ex); private <U extends T,V> CompletableFuture<V> orApplyStage(
Executor e, CompletionStage<U> o,
Function<? super T, ? extends V> f) {
CompletableFuture<U> b;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
CompletableFuture<V> d = new CompletableFuture<V>();
if (e != null || !d.orApply(this, b, f, null)) {
OrApply<T,U,V> c = new OrApply<T,U,V>(e, d, this, b, f);
orpush(b, c);
c.tryFire(SYNC);
} }
helpPostComplete(); return d;
other.helpPostComplete();
return dst;
} }
private CompletableFuture<Void> doRunAfterBoth(CompletableFuture<?> other, @SuppressWarnings("serial")
Runnable action, static final class OrAccept<T,U extends T> extends BiCompletion<T,U,Void> {
Executor e) { Consumer<? super T> fn;
if (other == null || action == null) throw new NullPointerException(); OrAccept(Executor executor, CompletableFuture<Void> dep,
CompletableFuture<Void> dst = new CompletableFuture<Void>(); CompletableFuture<T> src,
RunAfterBoth d = null; CompletableFuture<U> snd,
Object r, s = null; Consumer<? super T> fn) {
if ((r = result) == null || (s = other.result) == null) { super(executor, dep, src, snd); this.fn = fn;
d = new RunAfterBoth(this, other, action, dst, e); }
CompletionNode q = null, p = new CompletionNode(d); final CompletableFuture<Void> tryFire(int mode) {
while ((r == null && (r = result) == null) || CompletableFuture<Void> d;
(s == null && (s = other.result) == null)) { CompletableFuture<T> a;
if (q != null) { CompletableFuture<U> b;
if (s != null || if ((d = dep) == null ||
UNSAFE.compareAndSwapObject !d.orAccept(a = src, b = snd, fn, mode > 0 ? null : this))
(other, COMPLETIONS, q.next = other.completions, q)) return null;
break; dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
} }
else if (r != null ||
UNSAFE.compareAndSwapObject
(this, COMPLETIONS, p.next = completions, p)) {
if (s != null)
break;
q = new CompletionNode(d);
} }
final <R,S extends R> boolean orAccept(CompletableFuture<R> a,
CompletableFuture<S> b,
Consumer<? super R> f,
OrAccept<R,S> 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);
} }
if (r != null && s != null && (d == null || d.compareAndSet(0, 1))) {
Throwable ex;
if (r instanceof AltResult)
ex = ((AltResult)r).ex;
else
ex = null;
if (ex == null && (s instanceof AltResult))
ex = ((AltResult)s).ex;
if (ex == null) {
try {
if (e != null)
execAsync(e, new AsyncRun(action, dst));
else
action.run();
} catch (Throwable rex) {
ex = rex;
} }
return true;
} }
if (e == null || ex != null)
dst.internalComplete(null, ex); private <U extends T> CompletableFuture<Void> orAcceptStage(
Executor e, CompletionStage<U> o, Consumer<? super T> f) {
CompletableFuture<U> b;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
if (e != null || !d.orAccept(this, b, f, null)) {
OrAccept<T,U> c = new OrAccept<T,U>(e, d, this, b, f);
orpush(b, c);
c.tryFire(SYNC);
} }
helpPostComplete(); return d;
other.helpPostComplete();
return dst;
} }
private <U> CompletableFuture<U> doApplyToEither @SuppressWarnings("serial")
(CompletableFuture<? extends T> other, static final class OrRun<T,U> extends BiCompletion<T,U,Void> {
Function<? super T, U> fn, Runnable fn;
Executor e) { OrRun(Executor executor, CompletableFuture<Void> dep,
if (other == null || fn == null) throw new NullPointerException(); CompletableFuture<T> src,
CompletableFuture<U> dst = new CompletableFuture<U>(); CompletableFuture<U> snd,
ApplyToEither<T,U> d = null; Runnable fn) {
Object r; super(executor, dep, src, snd); this.fn = fn;
if ((r = result) == null && (r = other.result) == null) { }
d = new ApplyToEither<T,U>(this, other, fn, dst, e); final CompletableFuture<Void> tryFire(int mode) {
CompletionNode q = null, p = new CompletionNode(d); CompletableFuture<Void> d;
while ((r = result) == null && (r = other.result) == null) { CompletableFuture<T> a;
if (q != null) { CompletableFuture<U> b;
if (UNSAFE.compareAndSwapObject if ((d = dep) == null ||
(other, COMPLETIONS, q.next = other.completions, q)) !d.orRun(a = src, b = snd, fn, mode > 0 ? null : this))
break; return null;
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
} }
else if (UNSAFE.compareAndSwapObject
(this, COMPLETIONS, p.next = completions, p))
q = new CompletionNode(d);
} }
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();
} }
if (r != null && (d == null || d.compareAndSet(0, 1))) { } catch (Throwable ex) {
T t; Throwable ex; completeThrowable(ex);
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
} }
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
U u = null; return true;
if (ex == null) { }
try {
if (e != null) private CompletableFuture<Void> orRunStage(Executor e, CompletionStage<?> o,
execAsync(e, new AsyncApply<T,U>(t, fn, dst)); Runnable f) {
else CompletableFuture<?> b;
u = fn.apply(t); if (f == null || (b = o.toCompletableFuture()) == null)
} catch (Throwable rex) { throw new NullPointerException();
ex = rex; CompletableFuture<Void> d = new CompletableFuture<Void>();
if (e != null || !d.orRun(this, b, f, null)) {
OrRun<T,?> c = new OrRun<>(e, d, this, b, f);
orpush(b, c);
c.tryFire(SYNC);
} }
return d;
} }
if (e == null || ex != null)
dst.internalComplete(u, ex); @SuppressWarnings("serial")
static final class OrRelay<T,U> extends BiCompletion<T,U,Object> { // for Or
OrRelay(CompletableFuture<Object> dep, CompletableFuture<T> src,
CompletableFuture<U> snd) {
super(null, dep, src, snd);
}
final CompletableFuture<Object> tryFire(int mode) {
CompletableFuture<Object> d;
CompletableFuture<T> a;
CompletableFuture<U> 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);
} }
helpPostComplete();
other.helpPostComplete();
return dst;
} }
private CompletableFuture<Void> doAcceptEither final boolean orRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
(CompletableFuture<? extends T> other,
Consumer<? super T> fn,
Executor e) {
if (other == null || fn == null) throw new NullPointerException();
CompletableFuture<Void> dst = new CompletableFuture<Void>();
AcceptEither<T> d = null;
Object r; Object r;
if ((r = result) == null && (r = other.result) == null) { if (a == null || b == null ||
d = new AcceptEither<T>(this, other, fn, dst, e); ((r = a.result) == null && (r = b.result) == null))
CompletionNode q = null, p = new CompletionNode(d); return false;
while ((r = result) == null && (r = other.result) == null) { if (result == null)
if (q != null) { completeRelay(r);
if (UNSAFE.compareAndSwapObject return true;
(other, COMPLETIONS, q.next = other.completions, q))
break;
} }
else if (UNSAFE.compareAndSwapObject
(this, COMPLETIONS, p.next = completions, p)) /** Recursively constructs a tree of completions. */
q = new CompletionNode(d); static CompletableFuture<Object> orTree(CompletableFuture<?>[] cfs,
int lo, int hi) {
CompletableFuture<Object> d = new CompletableFuture<Object>();
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);
} }
} }
if (r != null && (d == null || d.compareAndSet(0, 1))) { return d;
T t; Throwable ex;
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
} }
else {
ex = null; /* ------------- Zero-input Async forms -------------- */
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr; @SuppressWarnings("serial")
static final class AsyncSupply<T> extends ForkJoinTask<Void>
implements Runnable, AsynchronousCompletionTask {
CompletableFuture<T> dep; Supplier<T> fn;
AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {
this.dep = dep; this.fn = fn;
} }
if (ex == null) {
public final Void getRawResult() { return null; }
public final void setRawResult(Void v) {}
public final boolean exec() { run(); return true; }
public void run() {
CompletableFuture<T> d; Supplier<T> f;
if ((d = dep) != null && (f = fn) != null) {
dep = null; fn = null;
if (d.result == null) {
try { try {
if (e != null) d.completeValue(f.get());
execAsync(e, new AsyncAccept<T>(t, fn, dst)); } catch (Throwable ex) {
else d.completeThrowable(ex);
fn.accept(t);
} catch (Throwable rex) {
ex = rex;
} }
} }
if (e == null || ex != null) d.postComplete();
dst.internalComplete(null, ex);
} }
helpPostComplete();
other.helpPostComplete();
return dst;
} }
private CompletableFuture<Void> doRunAfterEither
(CompletableFuture<?> other,
Runnable action,
Executor e) {
if (other == null || action == null) throw new NullPointerException();
CompletableFuture<Void> dst = new CompletableFuture<Void>();
RunAfterEither d = null;
Object r;
if ((r = result) == null && (r = other.result) == null) {
d = new RunAfterEither(this, other, action, dst, e);
CompletionNode q = null, p = new CompletionNode(d);
while ((r = result) == null && (r = other.result) == null) {
if (q != null) {
if (UNSAFE.compareAndSwapObject
(other, COMPLETIONS, q.next = other.completions, q))
break;
} }
else if (UNSAFE.compareAndSwapObject
(this, COMPLETIONS, p.next = completions, p)) static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
q = new CompletionNode(d); Supplier<U> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<U> d = new CompletableFuture<U>();
e.execute(new AsyncSupply<U>(d, f));
return d;
} }
@SuppressWarnings("serial")
static final class AsyncRun extends ForkJoinTask<Void>
implements Runnable, AsynchronousCompletionTask {
CompletableFuture<Void> dep; Runnable fn;
AsyncRun(CompletableFuture<Void> dep, Runnable fn) {
this.dep = dep; this.fn = fn;
} }
if (r != null && (d == null || d.compareAndSet(0, 1))) {
Throwable ex; public final Void getRawResult() { return null; }
if (r instanceof AltResult) public final void setRawResult(Void v) {}
ex = ((AltResult)r).ex; public final boolean exec() { run(); return true; }
else
ex = null; public void run() {
if (ex == null) { CompletableFuture<Void> d; Runnable f;
if ((d = dep) != null && (f = fn) != null) {
dep = null; fn = null;
if (d.result == null) {
try { try {
if (e != null) f.run();
execAsync(e, new AsyncRun(action, dst)); d.completeNull();
else } catch (Throwable ex) {
action.run(); d.completeThrowable(ex);
} catch (Throwable rex) {
ex = rex;
} }
} }
if (e == null || ex != null) d.postComplete();
dst.internalComplete(null, ex);
} }
helpPostComplete();
other.helpPostComplete();
return dst;
} }
private <U> CompletableFuture<U> doThenCompose
(Function<? super T, ? extends CompletionStage<U>> fn,
Executor e) {
if (fn == null) throw new NullPointerException();
CompletableFuture<U> dst = null;
ThenCompose<T,U> d = null;
Object r;
if ((r = result) == null) {
dst = new CompletableFuture<U>();
CompletionNode p = new CompletionNode
(d = new ThenCompose<T,U>(this, fn, dst, e));
while ((r = result) == null) {
if (UNSAFE.compareAndSwapObject
(this, COMPLETIONS, p.next = completions, p))
break;
} }
static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
if (f == null) throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
e.execute(new AsyncRun(d, f));
return d;
} }
if (r != null && (d == null || d.compareAndSet(0, 1))) {
T t; Throwable ex; /* ------------- Signallers -------------- */
if (r instanceof AltResult) {
ex = ((AltResult)r).ex; /**
t = null; * 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; // wait time if timed
final long deadline; // non-zero if timed
volatile int interruptControl; // > 0: interruptible, < 0: interrupted
volatile Thread thread;
Signaller(boolean interruptible, long nanos, long deadline) {
this.thread = Thread.currentThread();
this.interruptControl = interruptible ? 1 : 0;
this.nanos = nanos;
this.deadline = deadline;
} }
else { final CompletableFuture<?> tryFire(int ignore) {
ex = null; Thread w; // no need to atomically claim
@SuppressWarnings("unchecked") T tr = (T) r; if ((w = thread) != null) {
t = tr; thread = null;
LockSupport.unpark(w);
} }
if (ex == null) { return null;
if (e != null) {
if (dst == null)
dst = new CompletableFuture<U>();
execAsync(e, new AsyncCompose<T,U>(t, fn, dst));
} }
else { public boolean isReleasable() {
try { if (thread == null)
CompletionStage<U> cs = fn.apply(t); return true;
if (cs == null || if (Thread.interrupted()) {
(dst = cs.toCompletableFuture()) == null) int i = interruptControl;
ex = new NullPointerException(); interruptControl = -1;
} catch (Throwable rex) { if (i > 0)
ex = rex; return true;
} }
if (deadline != 0L &&
(nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {
thread = null;
return true;
} }
return false;
} }
if (dst == null) public boolean block() {
dst = new CompletableFuture<U>(); if (isReleasable())
if (ex != null) return true;
dst.internalComplete(null, ex); else if (deadline == 0L)
LockSupport.park(this);
else if (nanos > 0L)
LockSupport.parkNanos(this, nanos);
return isReleasable();
} }
helpPostComplete(); final boolean isLive() { return thread != null; }
dst.helpPostComplete();
return dst;
} }
private CompletableFuture<T> doWhenComplete /**
(BiConsumer<? super T, ? super Throwable> fn, * Returns raw result after waiting, or null if interruptible and
Executor e) { * interrupted.
if (fn == null) throw new NullPointerException(); */
CompletableFuture<T> dst = new CompletableFuture<T>(); private Object waitingGet(boolean interruptible) {
WhenCompleteCompletion<T> d = null; Signaller q = null;
boolean queued = false;
int spins = -1;
Object r; Object r;
if ((r = result) == null) {
CompletionNode p =
new CompletionNode(d = new WhenCompleteCompletion<T>
(this, fn, dst, e));
while ((r = result) == null) { while ((r = result) == null) {
if (UNSAFE.compareAndSwapObject(this, COMPLETIONS, if (spins < 0)
p.next = completions, p)) spins = (Runtime.getRuntime().availableProcessors() > 1) ?
break; 1 << 8 : 0; // Use brief spin-wait on multiprocessors
else 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 if (interruptible && q.interruptControl < 0) {
q.thread = null;
cleanStack();
return null;
} }
if (r != null && (d == null || d.compareAndSet(0, 1))) { else if (q.thread != null && result == null) {
T t; Throwable ex; try {
if (r instanceof AltResult) { ForkJoinPool.managedBlock(q);
ex = ((AltResult)r).ex; } catch (InterruptedException ie) {
t = null; q.interruptControl = -1;
} }
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
Throwable dx = null; }
try { if (q != null) {
if (e != null) q.thread = null;
execAsync(e, new AsyncWhenComplete<T>(t, ex, fn, dst)); if (q.interruptControl < 0) {
if (interruptible)
r = null; // report interruption
else else
fn.accept(t, ex); Thread.currentThread().interrupt();
} catch (Throwable rex) {
dx = rex;
} }
if (e == null || dx != null)
dst.internalComplete(t, ex != null ? ex : dx);
} }
helpPostComplete(); postComplete();
return dst; return r;
} }
private <U> CompletableFuture<U> doHandle /**
(BiFunction<? super T, Throwable, ? extends U> fn, * Returns raw result after waiting, or null if interrupted, or
Executor e) { * throws TimeoutException on timeout.
if (fn == null) throw new NullPointerException(); */
CompletableFuture<U> dst = new CompletableFuture<U>(); private Object timedGet(long nanos) throws TimeoutException {
HandleCompletion<T,U> d = null; if (Thread.interrupted())
return null;
if (nanos <= 0L)
throw new TimeoutException();
long d = System.nanoTime() + nanos;
Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0
boolean queued = false;
Object r; Object r;
if ((r = result) == null) { // We intentionally don't spin here (as waitingGet does) because
CompletionNode p = // the call to nanoTime() above acts much like a spin.
new CompletionNode(d = new HandleCompletion<T,U>
(this, fn, dst, e));
while ((r = result) == null) { while ((r = result) == null) {
if (UNSAFE.compareAndSwapObject(this, COMPLETIONS, if (!queued)
p.next = completions, p)) queued = tryPushStack(q);
break; else if (q.interruptControl < 0 || q.nanos <= 0L) {
} q.thread = null;
} cleanStack();
if (r != null && (d == null || d.compareAndSet(0, 1))) { if (q.interruptControl < 0)
T t; Throwable ex; return null;
if (r instanceof AltResult) { throw new TimeoutException();
ex = ((AltResult)r).ex;
t = null;
}
else {
ex = null;
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
} }
U u = null; else if (q.thread != null && result == null) {
Throwable dx = null;
try { try {
if (e != null) ForkJoinPool.managedBlock(q);
execAsync(e, new AsyncCombine<T,Throwable,U>(t, ex, fn, dst)); } catch (InterruptedException ie) {
else { q.interruptControl = -1;
u = fn.apply(t, ex);
dx = null;
} }
} catch (Throwable rex) {
dx = rex;
u = null;
} }
if (e == null || dx != null)
dst.internalComplete(u, dx);
} }
helpPostComplete(); if (q.interruptControl < 0)
return dst; r = null;
q.thread = null;
postComplete();
return r;
} }
/* ------------- public methods -------------- */
// public methods
/** /**
* Creates a new incomplete CompletableFuture. * Creates a new incomplete CompletableFuture.
...@@ -2113,6 +1785,13 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2113,6 +1785,13 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
public CompletableFuture() { public CompletableFuture() {
} }
/**
* Creates a new complete CompletableFuture with given encoded result.
*/
private CompletableFuture(Object r) {
this.result = r;
}
/** /**
* Returns a new CompletableFuture that is asynchronously completed * Returns a new CompletableFuture that is asynchronously completed
* by a task running in the {@link ForkJoinPool#commonPool()} with * by a task running in the {@link ForkJoinPool#commonPool()} with
...@@ -2124,10 +1803,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2124,10 +1803,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* @return the new CompletableFuture * @return the new CompletableFuture
*/ */
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) { public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
if (supplier == null) throw new NullPointerException(); return asyncSupplyStage(asyncPool, supplier);
CompletableFuture<U> f = new CompletableFuture<U>();
execAsync(ForkJoinPool.commonPool(), new AsyncSupply<U>(supplier, f));
return f;
} }
/** /**
...@@ -2143,11 +1819,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2143,11 +1819,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
*/ */
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
Executor executor) { Executor executor) {
if (executor == null || supplier == null) return asyncSupplyStage(screenExecutor(executor), supplier);
throw new NullPointerException();
CompletableFuture<U> f = new CompletableFuture<U>();
execAsync(executor, new AsyncSupply<U>(supplier, f));
return f;
} }
/** /**
...@@ -2160,10 +1832,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2160,10 +1832,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* @return the new CompletableFuture * @return the new CompletableFuture
*/ */
public static CompletableFuture<Void> runAsync(Runnable runnable) { public static CompletableFuture<Void> runAsync(Runnable runnable) {
if (runnable == null) throw new NullPointerException(); return asyncRunStage(asyncPool, runnable);
CompletableFuture<Void> f = new CompletableFuture<Void>();
execAsync(ForkJoinPool.commonPool(), new AsyncRun(runnable, f));
return f;
} }
/** /**
...@@ -2178,11 +1847,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2178,11 +1847,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
*/ */
public static CompletableFuture<Void> runAsync(Runnable runnable, public static CompletableFuture<Void> runAsync(Runnable runnable,
Executor executor) { Executor executor) {
if (executor == null || runnable == null) return asyncRunStage(screenExecutor(executor), runnable);
throw new NullPointerException();
CompletableFuture<Void> f = new CompletableFuture<Void>();
execAsync(executor, new AsyncRun(runnable, f));
return f;
} }
/** /**
...@@ -2194,9 +1859,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2194,9 +1859,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* @return the completed CompletableFuture * @return the completed CompletableFuture
*/ */
public static <U> CompletableFuture<U> completedFuture(U value) { public static <U> CompletableFuture<U> completedFuture(U value) {
CompletableFuture<U> f = new CompletableFuture<U>(); return new CompletableFuture<U>((value == null) ? NIL : value);
f.result = (value == null) ? NIL : value;
return f;
} }
/** /**
...@@ -2220,21 +1883,8 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2220,21 +1883,8 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* while waiting * while waiting
*/ */
public T get() throws InterruptedException, ExecutionException { public T get() throws InterruptedException, ExecutionException {
Object r; Throwable ex, cause; Object r;
if ((r = result) == null && (r = waitingGet(true)) == null) return reportGet((r = result) == null ? waitingGet(true) : r);
throw new InterruptedException();
if (!(r instanceof AltResult)) {
@SuppressWarnings("unchecked") T tr = (T) r;
return tr;
}
if ((ex = ((AltResult)r).ex) == null)
return null;
if (ex instanceof CancellationException)
throw (CancellationException)ex;
if ((ex instanceof CompletionException) &&
(cause = ex.getCause()) != null)
ex = cause;
throw new ExecutionException(ex);
} }
/** /**
...@@ -2252,24 +1902,9 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2252,24 +1902,9 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
*/ */
public T get(long timeout, TimeUnit unit) public T get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException { throws InterruptedException, ExecutionException, TimeoutException {
Object r; Throwable ex, cause; Object r;
long nanos = unit.toNanos(timeout); long nanos = unit.toNanos(timeout);
if (Thread.interrupted()) return reportGet((r = result) == null ? timedGet(nanos) : r);
throw new InterruptedException();
if ((r = result) == null)
r = timedAwaitDone(nanos);
if (!(r instanceof AltResult)) {
@SuppressWarnings("unchecked") T tr = (T) r;
return tr;
}
if ((ex = ((AltResult)r).ex) == null)
return null;
if (ex instanceof CancellationException)
throw (CancellationException)ex;
if ((ex instanceof CompletionException) &&
(cause = ex.getCause()) != null)
ex = cause;
throw new ExecutionException(ex);
} }
/** /**
...@@ -2287,20 +1922,8 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2287,20 +1922,8 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* exceptionally or a completion computation threw an exception * exceptionally or a completion computation threw an exception
*/ */
public T join() { public T join() {
Object r; Throwable ex; Object r;
if ((r = result) == null) return reportJoin((r = result) == null ? waitingGet(false) : r);
r = waitingGet(false);
if (!(r instanceof AltResult)) {
@SuppressWarnings("unchecked") T tr = (T) r;
return tr;
}
if ((ex = ((AltResult)r).ex) == null)
return null;
if (ex instanceof CancellationException)
throw (CancellationException)ex;
if (ex instanceof CompletionException)
throw (CompletionException)ex;
throw new CompletionException(ex);
} }
/** /**
...@@ -2314,20 +1937,8 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2314,20 +1937,8 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* exceptionally or a completion computation threw an exception * exceptionally or a completion computation threw an exception
*/ */
public T getNow(T valueIfAbsent) { public T getNow(T valueIfAbsent) {
Object r; Throwable ex; Object r;
if ((r = result) == null) return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
return valueIfAbsent;
if (!(r instanceof AltResult)) {
@SuppressWarnings("unchecked") T tr = (T) r;
return tr;
}
if ((ex = ((AltResult)r).ex) == null)
return null;
if (ex instanceof CancellationException)
throw (CancellationException)ex;
if (ex instanceof CompletionException)
throw (CompletionException)ex;
throw new CompletionException(ex);
} }
/** /**
...@@ -2339,9 +1950,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2339,9 +1950,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* to transition to a completed state, else {@code false} * to transition to a completed state, else {@code false}
*/ */
public boolean complete(T value) { public boolean complete(T value) {
boolean triggered = result == null && boolean triggered = completeValue(value);
UNSAFE.compareAndSwapObject(this, RESULT, null,
value == null ? NIL : value);
postComplete(); postComplete();
return triggered; return triggered;
} }
...@@ -2356,244 +1965,200 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2356,244 +1965,200 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
*/ */
public boolean completeExceptionally(Throwable ex) { public boolean completeExceptionally(Throwable ex) {
if (ex == null) throw new NullPointerException(); if (ex == null) throw new NullPointerException();
boolean triggered = result == null && boolean triggered = internalComplete(new AltResult(ex));
UNSAFE.compareAndSwapObject(this, RESULT, null, new AltResult(ex));
postComplete(); postComplete();
return triggered; return triggered;
} }
// CompletionStage methods public <U> CompletableFuture<U> thenApply(
Function<? super T,? extends U> fn) {
public <U> CompletableFuture<U> thenApply return uniApplyStage(null, fn);
(Function<? super T,? extends U> fn) {
return doThenApply(fn, null);
} }
public <U> CompletableFuture<U> thenApplyAsync public <U> CompletableFuture<U> thenApplyAsync(
(Function<? super T,? extends U> fn) { Function<? super T,? extends U> fn) {
return doThenApply(fn, ForkJoinPool.commonPool()); return uniApplyStage(asyncPool, fn);
} }
public <U> CompletableFuture<U> thenApplyAsync public <U> CompletableFuture<U> thenApplyAsync(
(Function<? super T,? extends U> fn, Function<? super T,? extends U> fn, Executor executor) {
Executor executor) { return uniApplyStage(screenExecutor(executor), fn);
if (executor == null) throw new NullPointerException();
return doThenApply(fn, executor);
} }
public CompletableFuture<Void> thenAccept public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
(Consumer<? super T> action) { return uniAcceptStage(null, action);
return doThenAccept(action, null);
} }
public CompletableFuture<Void> thenAcceptAsync public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
(Consumer<? super T> action) { return uniAcceptStage(asyncPool, action);
return doThenAccept(action, ForkJoinPool.commonPool());
} }
public CompletableFuture<Void> thenAcceptAsync public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
(Consumer<? super T> action,
Executor executor) { Executor executor) {
if (executor == null) throw new NullPointerException(); return uniAcceptStage(screenExecutor(executor), action);
return doThenAccept(action, executor);
} }
public CompletableFuture<Void> thenRun public CompletableFuture<Void> thenRun(Runnable action) {
(Runnable action) { return uniRunStage(null, action);
return doThenRun(action, null);
} }
public CompletableFuture<Void> thenRunAsync public CompletableFuture<Void> thenRunAsync(Runnable action) {
(Runnable action) { return uniRunStage(asyncPool, action);
return doThenRun(action, ForkJoinPool.commonPool());
} }
public CompletableFuture<Void> thenRunAsync public CompletableFuture<Void> thenRunAsync(Runnable action,
(Runnable action,
Executor executor) { Executor executor) {
if (executor == null) throw new NullPointerException(); return uniRunStage(screenExecutor(executor), action);
return doThenRun(action, executor);
} }
public <U,V> CompletableFuture<V> thenCombine public <U,V> CompletableFuture<V> thenCombine(
(CompletionStage<? extends U> other, CompletionStage<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn) { BiFunction<? super T,? super U,? extends V> fn) {
return doThenCombine(other.toCompletableFuture(), fn, null); return biApplyStage(null, other, fn);
} }
public <U,V> CompletableFuture<V> thenCombineAsync public <U,V> CompletableFuture<V> thenCombineAsync(
(CompletionStage<? extends U> other, CompletionStage<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn) { BiFunction<? super T,? super U,? extends V> fn) {
return doThenCombine(other.toCompletableFuture(), fn, return biApplyStage(asyncPool, other, fn);
ForkJoinPool.commonPool());
} }
public <U,V> CompletableFuture<V> thenCombineAsync public <U,V> CompletableFuture<V> thenCombineAsync(
(CompletionStage<? extends U> other, CompletionStage<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn, BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
Executor executor) { return biApplyStage(screenExecutor(executor), other, fn);
if (executor == null) throw new NullPointerException();
return doThenCombine(other.toCompletableFuture(), fn, executor);
} }
public <U> CompletableFuture<Void> thenAcceptBoth public <U> CompletableFuture<Void> thenAcceptBoth(
(CompletionStage<? extends U> other, CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action) { BiConsumer<? super T, ? super U> action) {
return doThenAcceptBoth(other.toCompletableFuture(), action, null); return biAcceptStage(null, other, action);
} }
public <U> CompletableFuture<Void> thenAcceptBothAsync public <U> CompletableFuture<Void> thenAcceptBothAsync(
(CompletionStage<? extends U> other, CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action) { BiConsumer<? super T, ? super U> action) {
return doThenAcceptBoth(other.toCompletableFuture(), action, return biAcceptStage(asyncPool, other, action);
ForkJoinPool.commonPool());
} }
public <U> CompletableFuture<Void> thenAcceptBothAsync public <U> CompletableFuture<Void> thenAcceptBothAsync(
(CompletionStage<? extends U> other, CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action, BiConsumer<? super T, ? super U> action, Executor executor) {
Executor executor) { return biAcceptStage(screenExecutor(executor), other, action);
if (executor == null) throw new NullPointerException();
return doThenAcceptBoth(other.toCompletableFuture(), action, executor);
} }
public CompletableFuture<Void> runAfterBoth public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
(CompletionStage<?> other,
Runnable action) { Runnable action) {
return doRunAfterBoth(other.toCompletableFuture(), action, null); return biRunStage(null, other, action);
} }
public CompletableFuture<Void> runAfterBothAsync public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
(CompletionStage<?> other,
Runnable action) { Runnable action) {
return doRunAfterBoth(other.toCompletableFuture(), action, return biRunStage(asyncPool, other, action);
ForkJoinPool.commonPool());
} }
public CompletableFuture<Void> runAfterBothAsync public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
(CompletionStage<?> other,
Runnable action, Runnable action,
Executor executor) { Executor executor) {
if (executor == null) throw new NullPointerException(); return biRunStage(screenExecutor(executor), other, action);
return doRunAfterBoth(other.toCompletableFuture(), action, executor);
} }
public <U> CompletableFuture<U> applyToEither(
public <U> CompletableFuture<U> applyToEither CompletionStage<? extends T> other, Function<? super T, U> fn) {
(CompletionStage<? extends T> other, return orApplyStage(null, other, fn);
Function<? super T, U> fn) {
return doApplyToEither(other.toCompletableFuture(), fn, null);
} }
public <U> CompletableFuture<U> applyToEitherAsync public <U> CompletableFuture<U> applyToEitherAsync(
(CompletionStage<? extends T> other, CompletionStage<? extends T> other, Function<? super T, U> fn) {
Function<? super T, U> fn) { return orApplyStage(asyncPool, other, fn);
return doApplyToEither(other.toCompletableFuture(), fn,
ForkJoinPool.commonPool());
} }
public <U> CompletableFuture<U> applyToEitherAsync public <U> CompletableFuture<U> applyToEitherAsync(
(CompletionStage<? extends T> other, CompletionStage<? extends T> other, Function<? super T, U> fn,
Function<? super T, U> fn,
Executor executor) { Executor executor) {
if (executor == null) throw new NullPointerException(); return orApplyStage(screenExecutor(executor), other, fn);
return doApplyToEither(other.toCompletableFuture(), fn, executor);
} }
public CompletableFuture<Void> acceptEither public CompletableFuture<Void> acceptEither(
(CompletionStage<? extends T> other, CompletionStage<? extends T> other, Consumer<? super T> action) {
Consumer<? super T> action) { return orAcceptStage(null, other, action);
return doAcceptEither(other.toCompletableFuture(), action, null);
} }
public CompletableFuture<Void> acceptEitherAsync public CompletableFuture<Void> acceptEitherAsync(
(CompletionStage<? extends T> other, CompletionStage<? extends T> other, Consumer<? super T> action) {
Consumer<? super T> action) { return orAcceptStage(asyncPool, other, action);
return doAcceptEither(other.toCompletableFuture(), action,
ForkJoinPool.commonPool());
} }
public CompletableFuture<Void> acceptEitherAsync public CompletableFuture<Void> acceptEitherAsync(
(CompletionStage<? extends T> other, CompletionStage<? extends T> other, Consumer<? super T> action,
Consumer<? super T> action,
Executor executor) { Executor executor) {
if (executor == null) throw new NullPointerException(); return orAcceptStage(screenExecutor(executor), other, action);
return doAcceptEither(other.toCompletableFuture(), action, executor);
} }
public CompletableFuture<Void> runAfterEither(CompletionStage<?> other, public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
Runnable action) { Runnable action) {
return doRunAfterEither(other.toCompletableFuture(), action, null); return orRunStage(null, other, action);
} }
public CompletableFuture<Void> runAfterEitherAsync public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
(CompletionStage<?> other,
Runnable action) { Runnable action) {
return doRunAfterEither(other.toCompletableFuture(), action, return orRunStage(asyncPool, other, action);
ForkJoinPool.commonPool());
} }
public CompletableFuture<Void> runAfterEitherAsync public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
(CompletionStage<?> other,
Runnable action, Runnable action,
Executor executor) { Executor executor) {
if (executor == null) throw new NullPointerException(); return orRunStage(screenExecutor(executor), other, action);
return doRunAfterEither(other.toCompletableFuture(), action, executor);
} }
public <U> CompletableFuture<U> thenCompose public <U> CompletableFuture<U> thenCompose(
(Function<? super T, ? extends CompletionStage<U>> fn) { Function<? super T, ? extends CompletionStage<U>> fn) {
return doThenCompose(fn, null); return uniComposeStage(null, fn);
} }
public <U> CompletableFuture<U> thenComposeAsync public <U> CompletableFuture<U> thenComposeAsync(
(Function<? super T, ? extends CompletionStage<U>> fn) { Function<? super T, ? extends CompletionStage<U>> fn) {
return doThenCompose(fn, ForkJoinPool.commonPool()); return uniComposeStage(asyncPool, fn);
} }
public <U> CompletableFuture<U> thenComposeAsync public <U> CompletableFuture<U> thenComposeAsync(
(Function<? super T, ? extends CompletionStage<U>> fn, Function<? super T, ? extends CompletionStage<U>> fn,
Executor executor) { Executor executor) {
if (executor == null) throw new NullPointerException(); return uniComposeStage(screenExecutor(executor), fn);
return doThenCompose(fn, executor);
} }
public CompletableFuture<T> whenComplete public CompletableFuture<T> whenComplete(
(BiConsumer<? super T, ? super Throwable> action) { BiConsumer<? super T, ? super Throwable> action) {
return doWhenComplete(action, null); return uniWhenCompleteStage(null, action);
} }
public CompletableFuture<T> whenCompleteAsync public CompletableFuture<T> whenCompleteAsync(
(BiConsumer<? super T, ? super Throwable> action) { BiConsumer<? super T, ? super Throwable> action) {
return doWhenComplete(action, ForkJoinPool.commonPool()); return uniWhenCompleteStage(asyncPool, action);
} }
public CompletableFuture<T> whenCompleteAsync public CompletableFuture<T> whenCompleteAsync(
(BiConsumer<? super T, ? super Throwable> action, BiConsumer<? super T, ? super Throwable> action, Executor executor) {
Executor executor) { return uniWhenCompleteStage(screenExecutor(executor), action);
if (executor == null) throw new NullPointerException();
return doWhenComplete(action, executor);
} }
public <U> CompletableFuture<U> handle public <U> CompletableFuture<U> handle(
(BiFunction<? super T, Throwable, ? extends U> fn) { BiFunction<? super T, Throwable, ? extends U> fn) {
return doHandle(fn, null); return uniHandleStage(null, fn);
} }
public <U> CompletableFuture<U> handleAsync public <U> CompletableFuture<U> handleAsync(
(BiFunction<? super T, Throwable, ? extends U> fn) { BiFunction<? super T, Throwable, ? extends U> fn) {
return doHandle(fn, ForkJoinPool.commonPool()); return uniHandleStage(asyncPool, fn);
} }
public <U> CompletableFuture<U> handleAsync public <U> CompletableFuture<U> handleAsync(
(BiFunction<? super T, Throwable, ? extends U> fn, BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
Executor executor) { return uniHandleStage(screenExecutor(executor), fn);
if (executor == null) throw new NullPointerException();
return doHandle(fn, executor);
} }
/** /**
* Returns this CompletableFuture * Returns this CompletableFuture.
* *
* @return this CompletableFuture * @return this CompletableFuture
*/ */
...@@ -2618,52 +2183,13 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2618,52 +2183,13 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* exceptionally * exceptionally
* @return the new CompletableFuture * @return the new CompletableFuture
*/ */
public CompletableFuture<T> exceptionally public CompletableFuture<T> exceptionally(
(Function<Throwable, ? extends T> fn) { Function<Throwable, ? extends T> fn) {
if (fn == null) throw new NullPointerException(); return uniExceptionallyStage(fn);
CompletableFuture<T> dst = new CompletableFuture<T>();
ExceptionCompletion<T> d = null;
Object r;
if ((r = result) == null) {
CompletionNode p =
new CompletionNode(d = new ExceptionCompletion<T>
(this, fn, dst));
while ((r = result) == null) {
if (UNSAFE.compareAndSwapObject(this, COMPLETIONS,
p.next = completions, p))
break;
}
}
if (r != null && (d == null || d.compareAndSet(0, 1))) {
T t = null; Throwable ex, dx = null;
if (r instanceof AltResult) {
if ((ex = ((AltResult)r).ex) != null) {
try {
t = fn.apply(ex);
} catch (Throwable rex) {
dx = rex;
}
}
}
else {
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
}
dst.internalComplete(t, dx);
}
helpPostComplete();
return dst;
} }
/* ------------- Arbitrary-arity constructions -------------- */ /* ------------- Arbitrary-arity constructions -------------- */
/*
* The basic plan of attack is to recursively form binary
* completion trees of elements. This can be overkill for small
* sets, but scales nicely. The And/All vs Or/Any forms use the
* same idea, but details differ.
*/
/** /**
* Returns a new CompletableFuture that is completed when all of * Returns a new CompletableFuture that is completed when all of
* the given CompletableFutures complete. If any of the given * the given CompletableFutures complete. If any of the given
...@@ -2688,82 +2214,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2688,82 +2214,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* {@code null} * {@code null}
*/ */
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) { public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
int len = cfs.length; // Directly handle empty and singleton cases return andTree(cfs, 0, cfs.length - 1);
if (len > 1)
return allTree(cfs, 0, len - 1);
else {
CompletableFuture<Void> dst = new CompletableFuture<Void>();
CompletableFuture<?> f;
if (len == 0)
dst.result = NIL;
else if ((f = cfs[0]) == null)
throw new NullPointerException();
else {
ThenPropagate d = null;
CompletionNode p = null;
Object r;
while ((r = f.result) == null) {
if (d == null)
d = new ThenPropagate(f, dst);
else if (p == null)
p = new CompletionNode(d);
else if (UNSAFE.compareAndSwapObject
(f, COMPLETIONS, p.next = f.completions, p))
break;
}
if (r != null && (d == null || d.compareAndSet(0, 1)))
dst.internalComplete(null, (r instanceof AltResult) ?
((AltResult)r).ex : null);
f.helpPostComplete();
}
return dst;
}
}
/**
* Recursively constructs an And'ed tree of CompletableFutures.
* Called only when array known to have at least two elements.
*/
private static CompletableFuture<Void> allTree(CompletableFuture<?>[] cfs,
int lo, int hi) {
CompletableFuture<?> fst, snd;
int mid = (lo + hi) >>> 1;
if ((fst = (lo == mid ? cfs[lo] : allTree(cfs, lo, mid))) == null ||
(snd = (hi == mid+1 ? cfs[hi] : allTree(cfs, mid+1, hi))) == null)
throw new NullPointerException();
CompletableFuture<Void> dst = new CompletableFuture<Void>();
AndCompletion d = null;
CompletionNode p = null, q = null;
Object r = null, s = null;
while ((r = fst.result) == null || (s = snd.result) == null) {
if (d == null)
d = new AndCompletion(fst, snd, dst);
else if (p == null)
p = new CompletionNode(d);
else if (q == null) {
if (UNSAFE.compareAndSwapObject
(fst, COMPLETIONS, p.next = fst.completions, p))
q = new CompletionNode(d);
}
else if (UNSAFE.compareAndSwapObject
(snd, COMPLETIONS, q.next = snd.completions, q))
break;
}
if ((r != null || (r = fst.result) != null) &&
(s != null || (s = snd.result) != null) &&
(d == null || d.compareAndSet(0, 1))) {
Throwable ex;
if (r instanceof AltResult)
ex = ((AltResult)r).ex;
else
ex = null;
if (ex == null && (s instanceof AltResult))
ex = ((AltResult)s).ex;
dst.internalComplete(null, ex);
}
fst.helpPostComplete();
snd.helpPostComplete();
return dst;
} }
/** /**
...@@ -2782,92 +2233,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2782,92 +2233,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* {@code null} * {@code null}
*/ */
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) { public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
int len = cfs.length; // Same idea as allOf return orTree(cfs, 0, cfs.length - 1);
if (len > 1)
return anyTree(cfs, 0, len - 1);
else {
CompletableFuture<Object> dst = new CompletableFuture<Object>();
CompletableFuture<?> f;
if (len == 0)
; // skip
else if ((f = cfs[0]) == null)
throw new NullPointerException();
else {
ThenCopy<Object> d = null;
CompletionNode p = null;
Object r;
while ((r = f.result) == null) {
if (d == null)
d = new ThenCopy<Object>(f, dst);
else if (p == null)
p = new CompletionNode(d);
else if (UNSAFE.compareAndSwapObject
(f, COMPLETIONS, p.next = f.completions, p))
break;
}
if (r != null && (d == null || d.compareAndSet(0, 1))) {
Throwable ex; Object t;
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
}
else {
ex = null;
t = r;
}
dst.internalComplete(t, ex);
}
f.helpPostComplete();
}
return dst;
}
}
/**
* Recursively constructs an Or'ed tree of CompletableFutures.
*/
private static CompletableFuture<Object> anyTree(CompletableFuture<?>[] cfs,
int lo, int hi) {
CompletableFuture<?> fst, snd;
int mid = (lo + hi) >>> 1;
if ((fst = (lo == mid ? cfs[lo] : anyTree(cfs, lo, mid))) == null ||
(snd = (hi == mid+1 ? cfs[hi] : anyTree(cfs, mid+1, hi))) == null)
throw new NullPointerException();
CompletableFuture<Object> dst = new CompletableFuture<Object>();
OrCompletion d = null;
CompletionNode p = null, q = null;
Object r;
while ((r = fst.result) == null && (r = snd.result) == null) {
if (d == null)
d = new OrCompletion(fst, snd, dst);
else if (p == null)
p = new CompletionNode(d);
else if (q == null) {
if (UNSAFE.compareAndSwapObject
(fst, COMPLETIONS, p.next = fst.completions, p))
q = new CompletionNode(d);
}
else if (UNSAFE.compareAndSwapObject
(snd, COMPLETIONS, q.next = snd.completions, q))
break;
}
if ((r != null || (r = fst.result) != null ||
(r = snd.result) != null) &&
(d == null || d.compareAndSet(0, 1))) {
Throwable ex; Object t;
if (r instanceof AltResult) {
ex = ((AltResult)r).ex;
t = null;
}
else {
ex = null;
t = r;
}
dst.internalComplete(t, ex);
}
fst.helpPostComplete();
snd.helpPostComplete();
return dst;
} }
/* ------------- Control and status methods -------------- */ /* ------------- Control and status methods -------------- */
...@@ -2887,8 +2253,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2887,8 +2253,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
*/ */
public boolean cancel(boolean mayInterruptIfRunning) { public boolean cancel(boolean mayInterruptIfRunning) {
boolean cancelled = (result == null) && boolean cancelled = (result == null) &&
UNSAFE.compareAndSwapObject internalComplete(new AltResult(new CancellationException()));
(this, RESULT, null, new AltResult(new CancellationException()));
postComplete(); postComplete();
return cancelled || isCancelled(); return cancelled || isCancelled();
} }
...@@ -2940,11 +2305,12 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2940,11 +2305,12 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
* Forcibly causes subsequent invocations of method {@link #get()} * Forcibly causes subsequent invocations of method {@link #get()}
* and related methods to throw the given exception, whether or * and related methods to throw the given exception, whether or
* not already completed. This method is designed for use only in * not already completed. This method is designed for use only in
* recovery actions, and even in such situations may result in * error recovery actions, and even in such situations may result
* ongoing dependent completions using established versus * in ongoing dependent completions using established versus
* overwritten outcomes. * overwritten outcomes.
* *
* @param ex the exception * @param ex the exception
* @throws NullPointerException if the exception is null
*/ */
public void obtrudeException(Throwable ex) { public void obtrudeException(Throwable ex) {
if (ex == null) throw new NullPointerException(); if (ex == null) throw new NullPointerException();
...@@ -2962,7 +2328,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2962,7 +2328,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
*/ */
public int getNumberOfDependents() { public int getNumberOfDependents() {
int count = 0; int count = 0;
for (CompletionNode p = completions; p != null; p = p.next) for (Completion p = stack; p != null; p = p.next)
++count; ++count;
return count; return count;
} }
...@@ -2993,20 +2359,19 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> { ...@@ -2993,20 +2359,19 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
// Unsafe mechanics // Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE; private static final sun.misc.Unsafe UNSAFE;
private static final long RESULT; private static final long RESULT;
private static final long WAITERS; private static final long STACK;
private static final long COMPLETIONS; private static final long NEXT;
static { static {
try { try {
UNSAFE = sun.misc.Unsafe.getUnsafe(); final sun.misc.Unsafe u;
UNSAFE = u = sun.misc.Unsafe.getUnsafe();
Class<?> k = CompletableFuture.class; Class<?> k = CompletableFuture.class;
RESULT = UNSAFE.objectFieldOffset RESULT = u.objectFieldOffset(k.getDeclaredField("result"));
(k.getDeclaredField("result")); STACK = u.objectFieldOffset(k.getDeclaredField("stack"));
WAITERS = UNSAFE.objectFieldOffset NEXT = u.objectFieldOffset
(k.getDeclaredField("waiters")); (Completion.class.getDeclaredField("next"));
COMPLETIONS = UNSAFE.objectFieldOffset } catch (Exception x) {
(k.getDeclaredField("completions")); throw new Error(x);
} catch (Exception e) {
throw new Error(e);
} }
} }
} }
src/share/classes/java/util/concurrent/CompletionStage.java
浏览文件 @ b51cc74c
...@@ -407,7 +407,7 @@ public interface CompletionStage<T> { ...@@ -407,7 +407,7 @@ public interface CompletionStage<T> {
/** /**
* Returns a new CompletionStage that, when this and the other * Returns a new CompletionStage that, when this and the other
* given stage complete normally, executes the given action using * given stage complete normally, executes the given action using
* the supplied executor * the supplied executor.
* *
* See the {@link CompletionStage} documentation for rules * See the {@link CompletionStage} documentation for rules
* covering exceptional completion. * covering exceptional completion.
...@@ -569,7 +569,7 @@ public interface CompletionStage<T> { ...@@ -569,7 +569,7 @@ public interface CompletionStage<T> {
/** /**
* Returns a new CompletionStage that, when either this or the * Returns a new CompletionStage that, when either this or the
* other given stage complete normally, executes the given action * other given stage complete normally, executes the given action
* using supplied executor. * using the supplied executor.
* *
* See the {@link CompletionStage} documentation for rules * See the {@link CompletionStage} documentation for rules
* covering exceptional completion. * covering exceptional completion.
...@@ -649,10 +649,15 @@ public interface CompletionStage<T> { ...@@ -649,10 +649,15 @@ public interface CompletionStage<T> {
(Function<Throwable, ? extends T> fn); (Function<Throwable, ? extends T> fn);
/** /**
* Returns a new CompletionStage with the same result or exception * Returns a new CompletionStage with the same result or exception as
* as this stage, and when this stage completes, executes the * this stage, that executes the given action when this stage completes.
* given action with the result (or {@code null} if none) and the *
* exception (or {@code null} if none) of this stage. * <p>When this stage is complete, the given action is invoked with the
* result (or {@code null} if none) and the exception (or {@code null}
* if none) of this stage as arguments. The returned stage is completed
* when the action returns. If the supplied action itself encounters an
* exception, then the returned stage exceptionally completes with this
* exception unless this stage also completed exceptionally.
* *
* @param action the action to perform * @param action the action to perform
* @return the new CompletionStage * @return the new CompletionStage
...@@ -661,12 +666,16 @@ public interface CompletionStage<T> { ...@@ -661,12 +666,16 @@ public interface CompletionStage<T> {
(BiConsumer<? super T, ? super Throwable> action); (BiConsumer<? super T, ? super Throwable> action);
/** /**
* Returns a new CompletionStage with the same result or exception * Returns a new CompletionStage with the same result or exception as
* as this stage, and when this stage completes, executes the * this stage, that executes the given action using this stage's
* given action executes the given action using this stage's * default asynchronous execution facility when this stage completes.
* default asynchronous execution facility, with the result (or *
* {@code null} if none) and the exception (or {@code null} if * <p>When this stage is complete, the given action is invoked with the
* none) of this stage as arguments. * result (or {@code null} if none) and the exception (or {@code null}
* if none) of this stage as arguments. The returned stage is completed
* when the action returns. If the supplied action itself encounters an
* exception, then the returned stage exceptionally completes with this
* exception unless this stage also completed exceptionally.
* *
* @param action the action to perform * @param action the action to perform
* @return the new CompletionStage * @return the new CompletionStage
...@@ -675,11 +684,16 @@ public interface CompletionStage<T> { ...@@ -675,11 +684,16 @@ public interface CompletionStage<T> {
(BiConsumer<? super T, ? super Throwable> action); (BiConsumer<? super T, ? super Throwable> action);
/** /**
* Returns a new CompletionStage with the same result or exception * Returns a new CompletionStage with the same result or exception as
* as this stage, and when this stage completes, executes using * this stage, that executes the given action using the supplied
* the supplied Executor, the given action with the result (or * Executor when this stage completes.
* {@code null} if none) and the exception (or {@code null} if *
* none) of this stage as arguments. * <p>When this stage is complete, the given action is invoked with the
* result (or {@code null} if none) and the exception (or {@code null}
* if none) of this stage as arguments. The returned stage is completed
* when the action returns. If the supplied action itself encounters an
* exception, then the returned stage exceptionally completes with this
* exception unless this stage also completed exceptionally.
* *
* @param action the action to perform * @param action the action to perform
* @param executor the executor to use for asynchronous execution * @param executor the executor to use for asynchronous execution
...@@ -693,9 +707,11 @@ public interface CompletionStage<T> { ...@@ -693,9 +707,11 @@ public interface CompletionStage<T> {
* Returns a new CompletionStage that, when this stage completes * Returns a new CompletionStage that, when this stage completes
* either normally or exceptionally, is executed with this stage's * either normally or exceptionally, is executed with this stage's
* result and exception as arguments to the supplied function. * result and exception as arguments to the supplied function.
* The given function is invoked with the result (or {@code null} *
* if none) and the exception (or {@code null} if none) of this * <p>When this stage is complete, the given function is invoked
* stage when complete as arguments. * with the result (or {@code null} if none) and the exception (or
* {@code null} if none) of this stage as arguments, and the
* function's result is used to complete the returned stage.
* *
* @param fn the function to use to compute the value of the * @param fn the function to use to compute the value of the
* returned CompletionStage * returned CompletionStage
...@@ -710,9 +726,11 @@ public interface CompletionStage<T> { ...@@ -710,9 +726,11 @@ public interface CompletionStage<T> {
* either normally or exceptionally, is executed using this stage's * either normally or exceptionally, is executed using this stage's
* default asynchronous execution facility, with this stage's * default asynchronous execution facility, with this stage's
* result and exception as arguments to the supplied function. * result and exception as arguments to the supplied function.
* The given function is invoked with the result (or {@code null} *
* if none) and the exception (or {@code null} if none) of this * <p>When this stage is complete, the given function is invoked
* stage when complete as arguments. * with the result (or {@code null} if none) and the exception (or
* {@code null} if none) of this stage as arguments, and the
* function's result is used to complete the returned stage.
* *
* @param fn the function to use to compute the value of the * @param fn the function to use to compute the value of the
* returned CompletionStage * returned CompletionStage
...@@ -726,10 +744,12 @@ public interface CompletionStage<T> { ...@@ -726,10 +744,12 @@ public interface CompletionStage<T> {
* Returns a new CompletionStage that, when this stage completes * Returns a new CompletionStage that, when this stage completes
* either normally or exceptionally, is executed using the * either normally or exceptionally, is executed using the
* supplied executor, with this stage's result and exception as * supplied executor, with this stage's result and exception as
* arguments to the supplied function. The given function is * arguments to the supplied function.
* invoked with the result (or {@code null} if none) and the *
* exception (or {@code null} if none) of this stage when complete * <p>When this stage is complete, the given function is invoked
* as arguments. * with the result (or {@code null} if none) and the exception (or
* {@code null} if none) of this stage as arguments, and the
* function's result is used to complete the returned stage.
* *
* @param fn the function to use to compute the value of the * @param fn the function to use to compute the value of the
* returned CompletionStage * returned CompletionStage
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或