--- a/jdk/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java Fri Jul 15 13:55:51 2016 -0700
+++ b/jdk/src/java.base/share/classes/java/util/concurrent/CompletableFuture.java Fri Jul 15 13:59:58 2016 -0700
@@ -35,6 +35,8 @@
package java.util.concurrent;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
import java.util.concurrent.locks.LockSupport;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
@@ -149,26 +151,29 @@
* applies across normal vs exceptional outcomes, sync vs async
* actions, binary triggers, and various forms of completions.
*
- * Non-nullness of field result (set via CAS) indicates done. An
- * AltResult is used to box null as a result, as well as to hold
- * exceptions. Using a single field makes completion simple to
- * detect and trigger. Encoding and decoding is straightforward
- * but adds to the sprawl of trapping and associating exceptions
- * with targets. Minor simplifications rely on (static) NIL (to
- * box null results) being the only AltResult with a null
- * exception field, so we don't usually need explicit comparisons.
- * Even though some of the generics casts are unchecked (see
- * SuppressWarnings annotations), they are placed to be
- * appropriate even if checked.
+ * Non-nullness of volatile field "result" indicates done. It may
+ * be set directly if known to be thread-confined, else via CAS.
+ * An AltResult is used to box null as a result, as well as to
+ * hold exceptions. Using a single field makes completion simple
+ * to detect and trigger. Result encoding and decoding is
+ * straightforward but tedious and adds to the sprawl of trapping
+ * and associating exceptions with targets. Minor simplifications
+ * rely on (static) NIL (to box null results) being the only
+ * AltResult with a null exception field, so we don't usually need
+ * explicit comparisons. Even though some of the generics casts
+ * are unchecked (see SuppressWarnings annotations), they are
+ * placed to be appropriate even if checked.
*
* Dependent actions are represented by Completion objects linked
* as Treiber stacks headed by field "stack". There are Completion
- * classes for each kind of action, grouped into single-input
- * (UniCompletion), two-input (BiCompletion), projected
- * (BiCompletions using either (not both) of two inputs), shared
- * (CoCompletion, used by the second of two sources), zero-input
- * source actions, and Signallers that unblock waiters. Class
- * Completion extends ForkJoinTask to enable async execution
+ * classes for each kind of action, grouped into:
+ * - single-input (UniCompletion),
+ * - two-input (BiCompletion),
+ * - projected (BiCompletions using exactly one of two inputs),
+ * - shared (CoCompletion, used by the second of two sources),
+ * - zero-input source actions,
+ * - Signallers that unblock waiters.
+ * Class Completion extends ForkJoinTask to enable async execution
* (adding no space overhead because we exploit its "tag" methods
* to maintain claims). It is also declared as Runnable to allow
* usage with arbitrary executors.
@@ -184,7 +189,7 @@
* encounter layers of adapters in common usages.
*
* * Boolean CompletableFuture method x(...) (for example
- * uniApply) takes all of the arguments needed to check that an
+ * biApply) 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
@@ -194,24 +199,29 @@
* 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.
+ * 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.
+ *
+ * * Some classes (for example UniApply) have separate handling
+ * code for when known to be thread-confined ("now" methods) and
+ * for when shared (in tryFire), for efficiency.
*
* * CompletableFuture method xStage(...) is called from a public
- * stage method of CompletableFuture x. It screens user
+ * stage method of CompletableFuture f. 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.
+ * not async and already triggerable, the action is run
+ * immediately. Otherwise a Completion c is created, and
+ * submitted to the executor if triggerable, or pushed onto f's
+ * stack if not. Completion actions are started via c.tryFire.
+ * We recheck after pushing to a source future's stack to cover
+ * possible races if the source 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.
@@ -236,29 +246,30 @@
* pointing back to its sources. So we null out fields as soon as
* possible. 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.
+ * threads nulling out fields. We also try to unlink non-isLive
+ * (fired or cancelled) Completions from stacks that might
+ * otherwise never be popped: Method cleanStack always unlinks non
+ * isLive completions from the head of stack; others may
+ * occasionally remain if racing with other cancellations or
+ * removals.
+ *
+ * Completion fields need not be declared as final or volatile
+ * because they are only visible to other threads upon safe
+ * publication.
*/
volatile Object result; // Either the result or boxed AltResult
volatile Completion stack; // Top of Treiber stack of dependent actions
final boolean internalComplete(Object r) { // CAS from null to r
- return U.compareAndSwapObject(this, RESULT, null, r);
- }
-
- final boolean casStack(Completion cmp, Completion val) {
- return U.compareAndSwapObject(this, STACK, cmp, val);
+ return RESULT.compareAndSet(this, null, r);
}
/** Returns true if successfully pushed c onto stack. */
final boolean tryPushStack(Completion c) {
Completion h = stack;
- lazySetNext(c, h);
- return U.compareAndSwapObject(this, STACK, h, c);
+ NEXT.set(c, h); // CAS piggyback
+ return STACK.compareAndSet(this, h, c);
}
/** Unconditionally pushes c onto stack, retrying if necessary. */
@@ -278,8 +289,7 @@
/** Completes with the null value, unless already completed. */
final boolean completeNull() {
- return U.compareAndSwapObject(this, RESULT, null,
- NIL);
+ return RESULT.compareAndSet(this, null, NIL);
}
/** Returns the encoding of the given non-exceptional value. */
@@ -289,8 +299,7 @@
/** Completes with a non-exceptional result, unless already completed. */
final boolean completeValue(T t) {
- return U.compareAndSwapObject(this, RESULT, null,
- (t == null) ? NIL : t);
+ return RESULT.compareAndSet(this, null, (t == null) ? NIL : t);
}
/**
@@ -304,8 +313,7 @@
/** Completes with an exceptional result, unless already completed. */
final boolean completeThrowable(Throwable x) {
- return U.compareAndSwapObject(this, RESULT, null,
- encodeThrowable(x));
+ return RESULT.compareAndSet(this, null, encodeThrowable(x));
}
/**
@@ -332,8 +340,7 @@
* existing CompletionException.
*/
final boolean completeThrowable(Throwable x, Object r) {
- return U.compareAndSwapObject(this, RESULT, null,
- encodeThrowable(x, r));
+ return RESULT.compareAndSet(this, null, encodeThrowable(x, r));
}
/**
@@ -351,10 +358,11 @@
*/
static Object encodeRelay(Object r) {
Throwable x;
- return (((r instanceof AltResult) &&
- (x = ((AltResult)r).ex) != null &&
- !(x instanceof CompletionException)) ?
- new AltResult(new CompletionException(x)) : r);
+ if (r instanceof AltResult
+ && (x = ((AltResult)r).ex) != null
+ && !(x instanceof CompletionException))
+ r = new AltResult(new CompletionException(x));
+ return r;
}
/**
@@ -362,14 +370,13 @@
* If exceptional, r is first coerced to a CompletionException.
*/
final boolean completeRelay(Object r) {
- return U.compareAndSwapObject(this, RESULT, null,
- encodeRelay(r));
+ return RESULT.compareAndSet(this, null, encodeRelay(r));
}
/**
* Reports result using Future.get conventions.
*/
- private static <T> T reportGet(Object r)
+ private static Object reportGet(Object r)
throws InterruptedException, ExecutionException {
if (r == null) // by convention below, null means interrupted
throw new InterruptedException();
@@ -384,14 +391,13 @@
x = cause;
throw new ExecutionException(x);
}
- @SuppressWarnings("unchecked") T t = (T) r;
- return t;
+ return r;
}
/**
* Decodes outcome to return result or throw unchecked exception.
*/
- private static <T> T reportJoin(Object r) {
+ private static Object reportJoin(Object r) {
if (r instanceof AltResult) {
Throwable x;
if ((x = ((AltResult)r).ex) == null)
@@ -402,8 +408,7 @@
throw (CompletionException)x;
throw new CompletionException(x);
}
- @SuppressWarnings("unchecked") T t = (T) r;
- return t;
+ return r;
}
/* ------------- Async task preliminaries -------------- */
@@ -449,12 +454,6 @@
static final int ASYNC = 1;
static final int NESTED = -1;
- /**
- * Spins before blocking in waitingGet
- */
- static final int SPINS = (Runtime.getRuntime().availableProcessors() > 1 ?
- 1 << 8 : 0);
-
/* ------------- Base Completion classes and operations -------------- */
@SuppressWarnings("serial")
@@ -479,10 +478,6 @@
public final void setRawResult(Void v) {}
}
- static void lazySetNext(Completion c, Completion next) {
- U.putObjectRelease(c, NEXT, next);
- }
-
/**
* Pops and tries to trigger all reachable dependents. Call only
* when known to be done.
@@ -497,40 +492,47 @@
while ((h = f.stack) != null ||
(f != this && (h = (f = this).stack) != null)) {
CompletableFuture<?> d; Completion t;
- if (f.casStack(h, t = h.next)) {
+ if (STACK.compareAndSet(f, h, t = h.next)) {
if (t != null) {
if (f != this) {
pushStack(h);
continue;
}
- h.next = null; // detach
+ NEXT.compareAndSet(h, t, null); // try to detach
}
f = (d = h.tryFire(NESTED)) == null ? this : d;
}
}
}
- /** Traverses stack and unlinks dead Completions. */
+ /** Traverses stack and unlinks one or more dead Completions, if found. */
final void cleanStack() {
- for (Completion p = null, q = stack; q != null;) {
+ Completion p = stack;
+ // ensure head of stack live
+ for (boolean unlinked = false;;) {
+ if (p == null)
+ return;
+ else if (p.isLive()) {
+ if (unlinked)
+ return;
+ else
+ break;
+ }
+ else if (STACK.weakCompareAndSetVolatile(this, p, (p = p.next)))
+ unlinked = true;
+ else
+ p = stack;
+ }
+ // try to unlink first non-live
+ for (Completion q = p.next; q != null;) {
Completion s = q.next;
if (q.isLive()) {
p = q;
q = s;
- }
- else if (p == null) {
- casStack(q, s);
- q = stack;
- }
- else {
- p.next = s;
- if (p.isLive())
- q = s;
- else {
- p = null; // restart
- q = stack;
- }
- }
+ } else if (NEXT.weakCompareAndSetVolatile(p, q, s))
+ break;
+ else
+ q = p.next;
}
}
@@ -568,11 +570,20 @@
final boolean isLive() { return dep != null; }
}
- /** Pushes the given completion (if it exists) unless done. */
- final void push(UniCompletion<?,?> c) {
+ /**
+ * Pushes the given completion unless it completes while trying.
+ * Caller should first check that result is null.
+ */
+ final void unipush(Completion c) {
if (c != null) {
- while (result == null && !tryPushStack(c))
- lazySetNext(c, null); // clear on failure
+ while (!tryPushStack(c)) {
+ if (result != null) {
+ NEXT.set(c, null);
+ break;
+ }
+ }
+ if (result != null)
+ c.tryFire(SYNC);
}
}
@@ -583,9 +594,10 @@
*/
final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
if (a != null && a.stack != null) {
- if (a.result == null)
+ Object r;
+ if ((r = a.result) == null)
a.cleanStack();
- else if (mode >= 0)
+ if (mode >= 0 && (r != null || a.result != null))
a.postComplete();
}
if (result != null && stack != null) {
@@ -607,48 +619,65 @@
}
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))
+ Object r; Throwable x; Function<? super T,? extends V> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null)
return null;
+ tryComplete: if (d.result == null) {
+ if (r instanceof AltResult) {
+ if ((x = ((AltResult)r).ex) != null) {
+ d.completeThrowable(x, r);
+ break tryComplete;
+ }
+ r = null;
+ }
+ try {
+ if (mode <= 0 && !claim())
+ return null;
+ else {
+ @SuppressWarnings("unchecked") T t = (T) r;
+ d.completeValue(f.apply(t));
+ }
+ } catch (Throwable ex) {
+ d.completeThrowable(ex);
+ }
+ }
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
- final <S> boolean uniApply(CompletableFuture<S> a,
- Function<? super S,? extends T> f,
- UniApply<S,T> c) {
- Object r; Throwable x;
- if (a == null || (r = a.result) == null || f == null)
- return false;
- tryComplete: if (result == null) {
- if (r instanceof AltResult) {
- if ((x = ((AltResult)r).ex) != null) {
- completeThrowable(x, r);
- break tryComplete;
- }
- r = null;
- }
- try {
- if (c != null && !c.claim())
- return false;
- @SuppressWarnings("unchecked") S s = (S) r;
- completeValue(f.apply(s));
- } catch (Throwable ex) {
- completeThrowable(ex);
- }
- }
- return true;
- }
-
private <V> CompletableFuture<V> uniApplyStage(
Executor e, Function<? super T,? extends V> f) {
if (f == null) throw new NullPointerException();
+ Object r;
+ if ((r = result) != null)
+ return uniApplyNow(r, e, f);
CompletableFuture<V> d = newIncompleteFuture();
- 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);
+ unipush(new UniApply<T,V>(e, d, this, f));
+ return d;
+ }
+
+ private <V> CompletableFuture<V> uniApplyNow(
+ Object r, Executor e, Function<? super T,? extends V> f) {
+ Throwable x;
+ CompletableFuture<V> d = newIncompleteFuture();
+ if (r instanceof AltResult) {
+ if ((x = ((AltResult)r).ex) != null) {
+ d.result = encodeThrowable(x, r);
+ return d;
+ }
+ r = null;
+ }
+ try {
+ if (e != null) {
+ e.execute(new UniApply<T,V>(null, d, this, f));
+ } else {
+ @SuppressWarnings("unchecked") T t = (T) r;
+ d.result = d.encodeValue(f.apply(t));
+ }
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
}
return d;
}
@@ -662,48 +691,67 @@
}
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))
+ Object r; Throwable x; Consumer<? super T> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null)
return null;
+ tryComplete: if (d.result == null) {
+ if (r instanceof AltResult) {
+ if ((x = ((AltResult)r).ex) != null) {
+ d.completeThrowable(x, r);
+ break tryComplete;
+ }
+ r = null;
+ }
+ try {
+ if (mode <= 0 && !claim())
+ return null;
+ else {
+ @SuppressWarnings("unchecked") T t = (T) r;
+ f.accept(t);
+ d.completeNull();
+ }
+ } catch (Throwable ex) {
+ d.completeThrowable(ex);
+ }
+ }
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
- final <S> boolean uniAccept(CompletableFuture<S> a,
- Consumer<? super S> f, UniAccept<S> c) {
- Object r; Throwable x;
- if (a == null || (r = a.result) == null || f == null)
- return false;
- tryComplete: if (result == null) {
- if (r instanceof AltResult) {
- if ((x = ((AltResult)r).ex) != null) {
- completeThrowable(x, r);
- break tryComplete;
- }
- r = null;
- }
- try {
- if (c != null && !c.claim())
- return false;
- @SuppressWarnings("unchecked") S s = (S) r;
- f.accept(s);
- completeNull();
- } catch (Throwable ex) {
- completeThrowable(ex);
- }
- }
- return true;
- }
-
private CompletableFuture<Void> uniAcceptStage(Executor e,
Consumer<? super T> f) {
if (f == null) throw new NullPointerException();
+ Object r;
+ if ((r = result) != null)
+ return uniAcceptNow(r, e, f);
CompletableFuture<Void> d = newIncompleteFuture();
- if (e != null || !d.uniAccept(this, f, null)) {
- UniAccept<T> c = new UniAccept<T>(e, d, this, f);
- push(c);
- c.tryFire(SYNC);
+ unipush(new UniAccept<T>(e, d, this, f));
+ return d;
+ }
+
+ private CompletableFuture<Void> uniAcceptNow(
+ Object r, Executor e, Consumer<? super T> f) {
+ Throwable x;
+ CompletableFuture<Void> d = newIncompleteFuture();
+ if (r instanceof AltResult) {
+ if ((x = ((AltResult)r).ex) != null) {
+ d.result = encodeThrowable(x, r);
+ return d;
+ }
+ r = null;
+ }
+ try {
+ if (e != null) {
+ e.execute(new UniAccept<T>(null, d, this, f));
+ } else {
+ @SuppressWarnings("unchecked") T t = (T) r;
+ f.accept(t);
+ d.result = NIL;
+ }
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
}
return d;
}
@@ -717,42 +765,56 @@
}
final CompletableFuture<Void> tryFire(int mode) {
CompletableFuture<Void> d; CompletableFuture<T> a;
- if ((d = dep) == null ||
- !d.uniRun(a = src, fn, mode > 0 ? null : this))
+ Object r; Throwable x; Runnable f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null)
return null;
+ if (d.result == null) {
+ if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
+ d.completeThrowable(x, r);
+ else
+ try {
+ if (mode <= 0 && !claim())
+ return null;
+ else {
+ f.run();
+ d.completeNull();
+ }
+ } catch (Throwable ex) {
+ d.completeThrowable(ex);
+ }
+ }
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
- final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) {
- Object r; Throwable x;
- if (a == null || (r = a.result) == null || f == null)
- return false;
- if (result == null) {
- if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
- completeThrowable(x, r);
- else
- try {
- if (c != null && !c.claim())
- return false;
- f.run();
- completeNull();
- } catch (Throwable ex) {
- completeThrowable(ex);
- }
- }
- return true;
+ private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {
+ if (f == null) throw new NullPointerException();
+ Object r;
+ if ((r = result) != null)
+ return uniRunNow(r, e, f);
+ CompletableFuture<Void> d = newIncompleteFuture();
+ unipush(new UniRun<T>(e, d, this, f));
+ return d;
}
- private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {
- if (f == null) throw new NullPointerException();
+ private CompletableFuture<Void> uniRunNow(Object r, Executor e, Runnable f) {
+ Throwable x;
CompletableFuture<Void> d = newIncompleteFuture();
- if (e != null || !d.uniRun(this, f, null)) {
- UniRun<T> c = new UniRun<T>(e, d, this, f);
- push(c);
- c.tryFire(SYNC);
- }
+ if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
+ d.result = encodeThrowable(x, r);
+ else
+ try {
+ if (e != null) {
+ e.execute(new UniRun<T>(null, d, this, f));
+ } else {
+ f.run();
+ d.result = NIL;
+ }
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
return d;
}
@@ -766,20 +828,20 @@
}
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))
+ Object r; BiConsumer<? super T, ? super Throwable> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null
+ || !d.uniWhenComplete(r, f, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
- final boolean uniWhenComplete(CompletableFuture<T> a,
+ final boolean uniWhenComplete(Object r,
BiConsumer<? super T,? super Throwable> f,
UniWhenComplete<T> c) {
- Object r; T t; Throwable x = null;
- if (a == null || (r = a.result) == null || f == null)
- return false;
+ T t; Throwable x = null;
if (result == null) {
try {
if (c != null && !c.claim())
@@ -811,10 +873,17 @@
Executor e, BiConsumer<? super T, ? super Throwable> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<T> d = newIncompleteFuture();
- if (e != null || !d.uniWhenComplete(this, f, null)) {
- UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);
- push(c);
- c.tryFire(SYNC);
+ Object r;
+ if ((r = result) == null)
+ unipush(new UniWhenComplete<T>(e, d, this, f));
+ else if (e == null)
+ d.uniWhenComplete(r, f, null);
+ else {
+ try {
+ e.execute(new UniWhenComplete<T>(null, d, this, f));
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
}
return d;
}
@@ -829,20 +898,20 @@
}
final CompletableFuture<V> tryFire(int mode) {
CompletableFuture<V> d; CompletableFuture<T> a;
- if ((d = dep) == null ||
- !d.uniHandle(a = src, fn, mode > 0 ? null : this))
+ Object r; BiFunction<? super T, Throwable, ? extends V> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null
+ || !d.uniHandle(r, f, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
- final <S> boolean uniHandle(CompletableFuture<S> a,
+ final <S> boolean uniHandle(Object r,
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;
+ S s; Throwable x;
if (result == null) {
try {
if (c != null && !c.claim())
@@ -867,10 +936,17 @@
Executor e, BiFunction<? super T, Throwable, ? extends V> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<V> d = newIncompleteFuture();
- 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);
+ Object r;
+ if ((r = result) == null)
+ unipush(new UniHandle<T,V>(e, d, this, f));
+ else if (e == null)
+ d.uniHandle(r, f, null);
+ else {
+ try {
+ e.execute(new UniHandle<T,V>(null, d, this, f));
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
}
return d;
}
@@ -885,19 +961,20 @@
final CompletableFuture<T> tryFire(int mode) { // never ASYNC
// assert mode != ASYNC;
CompletableFuture<T> d; CompletableFuture<T> a;
- if ((d = dep) == null || !d.uniExceptionally(a = src, fn, this))
+ Object r; Function<? super Throwable, ? extends T> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null
+ || !d.uniExceptionally(r, f, this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
- final boolean uniExceptionally(CompletableFuture<T> a,
+ final boolean uniExceptionally(Object r,
Function<? super Throwable, ? extends T> f,
UniExceptionally<T> c) {
- Object r; Throwable x;
- if (a == null || (r = a.result) == null || f == null)
- return false;
+ Throwable x;
if (result == null) {
try {
if (r instanceof AltResult && (x = ((AltResult)r).ex) != null) {
@@ -917,47 +994,38 @@
Function<Throwable, ? extends T> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<T> d = newIncompleteFuture();
- if (!d.uniExceptionally(this, f, null)) {
- UniExceptionally<T> c = new UniExceptionally<T>(d, this, f);
- push(c);
- c.tryFire(SYNC);
- }
+ Object r;
+ if ((r = result) == null)
+ unipush(new UniExceptionally<T>(d, this, f));
+ else
+ d.uniExceptionally(r, f, null);
return d;
}
@SuppressWarnings("serial")
- static final class UniRelay<T> extends UniCompletion<T,T> { // for Compose
+ static final class UniRelay<T> extends UniCompletion<T,T> {
UniRelay(CompletableFuture<T> dep, CompletableFuture<T> src) {
super(null, dep, src);
}
final CompletableFuture<T> tryFire(int mode) {
- CompletableFuture<T> d; CompletableFuture<T> a;
- if ((d = dep) == null || !d.uniRelay(a = src))
+ CompletableFuture<T> d; CompletableFuture<T> a; Object r;
+ if ((d = dep) == null
+ || (a = src) == null || (r = a.result) == null)
return null;
+ if (d.result == null)
+ d.completeRelay(r);
src = null; dep = null;
return d.postFire(a, mode);
}
}
- final boolean uniRelay(CompletableFuture<T> a) {
- Object r;
- if (a == null || (r = a.result) == null)
- return false;
- if (result == null) // no need to claim
- completeRelay(r);
- return true;
- }
-
private CompletableFuture<T> uniCopyStage() {
Object r;
CompletableFuture<T> d = newIncompleteFuture();
if ((r = result) != null)
- d.completeRelay(r);
- else {
- UniRelay<T> c = new UniRelay<T>(d, this);
- push(c);
- c.tryFire(SYNC);
- }
+ d.result = encodeRelay(r);
+ else
+ unipush(new UniRelay<T>(d, this));
return d;
}
@@ -966,9 +1034,7 @@
if ((r = result) != null)
return new MinimalStage<T>(encodeRelay(r));
MinimalStage<T> d = new MinimalStage<T>();
- UniRelay<T> c = new UniRelay<T>(d, this);
- push(c);
- c.tryFire(SYNC);
+ unipush(new UniRelay<T>(d, this));
return d;
}
@@ -982,54 +1048,48 @@
}
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))
+ Function<? super T, ? extends CompletionStage<V>> f;
+ Object r; Throwable x;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null)
return null;
+ tryComplete: if (d.result == null) {
+ if (r instanceof AltResult) {
+ if ((x = ((AltResult)r).ex) != null) {
+ d.completeThrowable(x, r);
+ break tryComplete;
+ }
+ r = null;
+ }
+ try {
+ if (mode <= 0 && !claim())
+ return null;
+ @SuppressWarnings("unchecked") T t = (T) r;
+ CompletableFuture<V> g = f.apply(t).toCompletableFuture();
+ if ((r = g.result) != null)
+ d.completeRelay(r);
+ else {
+ g.unipush(new UniRelay<V>(d, g));
+ if (d.result == null)
+ return null;
+ }
+ } catch (Throwable ex) {
+ d.completeThrowable(ex);
+ }
+ }
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
- final <S> boolean uniCompose(
- CompletableFuture<S> a,
- Function<? super S, ? extends CompletionStage<T>> f,
- UniCompose<S,T> c) {
- Object r; Throwable x;
- if (a == null || (r = a.result) == null || f == null)
- return false;
- tryComplete: if (result == null) {
- if (r instanceof AltResult) {
- if ((x = ((AltResult)r).ex) != null) {
- completeThrowable(x, r);
- break tryComplete;
- }
- r = null;
- }
- try {
- if (c != null && !c.claim())
- return false;
- @SuppressWarnings("unchecked") S s = (S) r;
- CompletableFuture<T> g = f.apply(s).toCompletableFuture();
- if (g.result == null || !uniRelay(g)) {
- UniRelay<T> copy = new UniRelay<T>(this, g);
- g.push(copy);
- copy.tryFire(SYNC);
- if (result == null)
- return false;
- }
- } catch (Throwable ex) {
- completeThrowable(ex);
- }
- }
- return true;
- }
-
private <V> CompletableFuture<V> uniComposeStage(
Executor e, Function<? super T, ? extends CompletionStage<V>> f) {
if (f == null) throw new NullPointerException();
+ CompletableFuture<V> d = newIncompleteFuture();
Object r, s; Throwable x;
- CompletableFuture<V> d = newIncompleteFuture();
- if (e == null && (r = result) != null) {
+ if ((r = result) == null)
+ unipush(new UniCompose<T,V>(e, d, this, f));
+ else if (e == null) {
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
d.result = encodeThrowable(x, r);
@@ -1041,21 +1101,20 @@
@SuppressWarnings("unchecked") T t = (T) r;
CompletableFuture<V> g = f.apply(t).toCompletableFuture();
if ((s = g.result) != null)
- d.completeRelay(s);
+ d.result = encodeRelay(s);
else {
- UniRelay<V> c = new UniRelay<V>(d, g);
- g.push(c);
- c.tryFire(SYNC);
+ g.unipush(new UniRelay<V>(d, g));
}
- return d;
} catch (Throwable ex) {
d.result = encodeThrowable(ex);
- return d;
}
}
- UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);
- push(c);
- c.tryFire(SYNC);
+ else
+ try {
+ e.execute(new UniCompose<T,V>(null, d, this, f));
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
return d;
}
@@ -1085,21 +1144,28 @@
}
final boolean isLive() {
BiCompletion<?,?,?> c;
- return (c = base) != null && c.dep != null;
+ return (c = base) != null
+ // && c.isLive()
+ && c.dep != null;
}
}
- /** Pushes completion to this and b unless both done. */
+ /**
+ * Pushes completion to this and b unless both done.
+ * Caller should first check that either result or b.result is null.
+ */
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
+ while (result == null) {
+ if (tryPushStack(c)) {
+ if (b.result == null)
+ b.unipush(new CoCompletion(c));
+ else if (result != null)
+ c.tryFire(SYNC);
+ return;
+ }
}
+ b.unipush(c);
}
}
@@ -1107,9 +1173,10 @@
final CompletableFuture<T> postFire(CompletableFuture<?> a,
CompletableFuture<?> b, int mode) {
if (b != null && b.stack != null) { // clean second source
- if (b.result == null)
+ Object r;
+ if ((r = b.result) == null)
b.cleanStack();
- else if (mode >= 0)
+ if (mode >= 0 && (r != null || b.result != null))
b.postComplete();
}
return postFire(a, mode);
@@ -1127,22 +1194,21 @@
CompletableFuture<V> d;
CompletableFuture<T> a;
CompletableFuture<U> b;
- if ((d = dep) == null ||
- !d.biApply(a = src, b = snd, fn, mode > 0 ? null : this))
+ Object r, s; BiFunction<? super T,? super U,? extends V> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null
+ || (b = snd) == null || (s = b.result) == null
+ || !d.biApply(r, s, f, mode > 0 ? null : this))
return null;
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
}
}
- final <R,S> boolean biApply(CompletableFuture<R> a,
- CompletableFuture<S> b,
+ final <R,S> boolean biApply(Object r, Object s,
BiFunction<? super R,? super S,? extends T> f,
BiApply<R,S,T> c) {
- Object r, s; Throwable x;
- if (a == null || (r = a.result) == null ||
- b == null || (s = b.result) == null || f == null)
- return false;
+ Throwable x;
tryComplete: if (result == null) {
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
@@ -1174,15 +1240,20 @@
private <U,V> CompletableFuture<V> biApplyStage(
Executor e, CompletionStage<U> o,
BiFunction<? super T,? super U,? extends V> f) {
- CompletableFuture<U> b;
+ CompletableFuture<U> b; Object r, s;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
CompletableFuture<V> d = newIncompleteFuture();
- if (e != null || !d.biApply(this, b, f, null)) {
- BiApply<T,U,V> c = new BiApply<T,U,V>(e, d, this, b, f);
- bipush(b, c);
- c.tryFire(SYNC);
- }
+ if ((r = result) == null || (s = b.result) == null)
+ bipush(b, new BiApply<T,U,V>(e, d, this, b, f));
+ else if (e == null)
+ d.biApply(r, s, f, null);
+ else
+ try {
+ e.execute(new BiApply<T,U,V>(null, d, this, b, f));
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
return d;
}
@@ -1198,22 +1269,21 @@
CompletableFuture<Void> d;
CompletableFuture<T> a;
CompletableFuture<U> b;
- if ((d = dep) == null ||
- !d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this))
+ Object r, s; BiConsumer<? super T,? super U> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null
+ || (b = snd) == null || (s = b.result) == null
+ || !d.biAccept(r, s, f, mode > 0 ? null : this))
return null;
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
}
}
- final <R,S> boolean biAccept(CompletableFuture<R> a,
- CompletableFuture<S> b,
+ final <R,S> boolean biAccept(Object r, Object s,
BiConsumer<? super R,? super S> f,
BiAccept<R,S> c) {
- Object r, s; Throwable x;
- if (a == null || (r = a.result) == null ||
- b == null || (s = b.result) == null || f == null)
- return false;
+ Throwable x;
tryComplete: if (result == null) {
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
@@ -1246,15 +1316,20 @@
private <U> CompletableFuture<Void> biAcceptStage(
Executor e, CompletionStage<U> o,
BiConsumer<? super T,? super U> f) {
- CompletableFuture<U> b;
+ CompletableFuture<U> b; Object r, s;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
CompletableFuture<Void> d = newIncompleteFuture();
- if (e != null || !d.biAccept(this, b, f, null)) {
- BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
- bipush(b, c);
- c.tryFire(SYNC);
- }
+ if ((r = result) == null || (s = b.result) == null)
+ bipush(b, new BiAccept<T,U>(e, d, this, b, f));
+ else if (e == null)
+ d.biAccept(r, s, f, null);
+ else
+ try {
+ e.execute(new BiAccept<T,U>(null, d, this, b, f));
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
return d;
}
@@ -1262,8 +1337,7 @@
static final class BiRun<T,U> extends BiCompletion<T,U,Void> {
Runnable fn;
BiRun(Executor executor, CompletableFuture<Void> dep,
- CompletableFuture<T> src,
- CompletableFuture<U> snd,
+ CompletableFuture<T> src, CompletableFuture<U> snd,
Runnable fn) {
super(executor, dep, src, snd); this.fn = fn;
}
@@ -1271,25 +1345,25 @@
CompletableFuture<Void> d;
CompletableFuture<T> a;
CompletableFuture<U> b;
- if ((d = dep) == null ||
- !d.biRun(a = src, b = snd, fn, mode > 0 ? null : this))
+ Object r, s; Runnable f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (r = a.result) == null
+ || (b = snd) == null || (s = b.result) == null
+ || !d.biRun(r, s, f, mode > 0 ? null : this))
return null;
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
}
}
- final boolean biRun(CompletableFuture<?> a, CompletableFuture<?> b,
- Runnable f, BiRun<?,?> c) {
- Object r, s; Throwable x;
- if (a == null || (r = a.result) == null ||
- b == null || (s = b.result) == null || f == null)
- return false;
+ final boolean biRun(Object r, Object s, Runnable f, BiRun<?,?> c) {
+ Throwable x; Object z;
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);
+ if ((r instanceof AltResult
+ && (x = ((AltResult)(z = r)).ex) != null) ||
+ (s instanceof AltResult
+ && (x = ((AltResult)(z = s)).ex) != null))
+ completeThrowable(x, z);
else
try {
if (c != null && !c.claim())
@@ -1305,52 +1379,52 @@
private CompletableFuture<Void> biRunStage(Executor e, CompletionStage<?> o,
Runnable f) {
- CompletableFuture<?> b;
+ CompletableFuture<?> b; Object r, s;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
CompletableFuture<Void> d = newIncompleteFuture();
- if (e != null || !d.biRun(this, b, f, null)) {
- BiRun<T,?> c = new BiRun<>(e, d, this, b, f);
- bipush(b, c);
- c.tryFire(SYNC);
- }
+ if ((r = result) == null || (s = b.result) == null)
+ bipush(b, new BiRun<>(e, d, this, b, f));
+ else if (e == null)
+ d.biRun(r, s, f, null);
+ else
+ try {
+ e.execute(new BiRun<>(null, d, this, b, f));
+ } catch (Throwable ex) {
+ d.result = encodeThrowable(ex);
+ }
return d;
}
@SuppressWarnings("serial")
static final class BiRelay<T,U> extends BiCompletion<T,U,Void> { // for And
BiRelay(CompletableFuture<Void> dep,
- CompletableFuture<T> src,
- CompletableFuture<U> snd) {
+ 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))
+ Object r, s, z; Throwable x;
+ if ((d = dep) == null
+ || (a = src) == null || (r = a.result) == null
+ || (b = snd) == null || (s = b.result) == null)
return null;
+ if (d.result == null) {
+ if ((r instanceof AltResult
+ && (x = ((AltResult)(z = r)).ex) != null) ||
+ (s instanceof AltResult
+ && (x = ((AltResult)(z = s)).ex) != null))
+ d.completeThrowable(x, z);
+ else
+ d.completeNull();
+ }
src = null; snd = null; dep = null;
return d.postFire(a, b, mode);
}
}
- boolean biRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
- Object r, s; Throwable x;
- if (a == null || (r = a.result) == null ||
- b == null || (s = b.result) == null)
- return false;
- if (result == null) {
- if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
- completeThrowable(x, r);
- else if (s instanceof AltResult && (x = ((AltResult)s).ex) != null)
- completeThrowable(x, s);
- else
- completeNull();
- }
- return true;
- }
-
/** Recursively constructs a tree of completions. */
static CompletableFuture<Void> andTree(CompletableFuture<?>[] cfs,
int lo, int hi) {
@@ -1358,39 +1432,44 @@
if (lo > hi) // empty
d.result = NIL;
else {
- CompletableFuture<?> a, b;
+ CompletableFuture<?> a, b; Object r, s, z; Throwable x;
int mid = (lo + hi) >>> 1;
if ((a = (lo == mid ? cfs[lo] :
andTree(cfs, lo, mid))) == null ||
(b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
andTree(cfs, mid+1, hi))) == null)
throw new NullPointerException();
- if (!d.biRelay(a, b)) {
- BiRelay<?,?> c = new BiRelay<>(d, a, b);
- a.bipush(b, c);
- c.tryFire(SYNC);
- }
+ if ((r = a.result) == null || (s = b.result) == null)
+ a.bipush(b, new BiRelay<>(d, a, b));
+ else if ((r instanceof AltResult
+ && (x = ((AltResult)(z = r)).ex) != null) ||
+ (s instanceof AltResult
+ && (x = ((AltResult)(z = s)).ex) != null))
+ d.result = encodeThrowable(x, z);
+ else
+ d.result = NIL;
}
return d;
}
/* ------------- Projected (Ored) BiCompletions -------------- */
- /** Pushes completion to this and b unless either done. */
+ /**
+ * Pushes completion to this and b unless either done.
+ * Caller should first check that result and b.result are both null.
+ */
final void orpush(CompletableFuture<?> b, BiCompletion<?,?,?> c) {
if (c != null) {
- while ((b == null || b.result == null) && result == null) {
- if (tryPushStack(c)) {
- if (b != null && b != this && b.result == null) {
- Completion q = new CoCompletion(c);
- while (result == null && b.result == null &&
- !b.tryPushStack(q))
- lazySetNext(q, null); // clear on failure
- }
+ while (!tryPushStack(c)) {
+ if (result != null) {
+ NEXT.set(c, null);
break;
}
- lazySetNext(c, null); // clear on failure
}
+ if (result != null)
+ c.tryFire(SYNC);
+ else
+ b.unipush(new CoCompletion(c));
}
}
@@ -1398,8 +1477,7 @@
static final class OrApply<T,U extends T,V> extends BiCompletion<T,U,V> {
Function<? super T,? extends V> fn;
OrApply(Executor executor, CompletableFuture<V> dep,
- CompletableFuture<T> src,
- CompletableFuture<U> snd,
+ CompletableFuture<T> src, CompletableFuture<U> snd,
Function<? super T,? extends V> fn) {
super(executor, dep, src, snd); this.fn = fn;
}
@@ -1407,54 +1485,46 @@
CompletableFuture<V> d;
CompletableFuture<T> a;
CompletableFuture<U> b;
- if ((d = dep) == null ||
- !d.orApply(a = src, b = snd, fn, mode > 0 ? null : this))
+ Object r; Throwable x; Function<? super T,? extends V> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (b = snd) == null
+ || ((r = a.result) == null && (r = b.result) == null))
return null;
+ tryComplete: if (d.result == null) {
+ try {
+ if (mode <= 0 && !claim())
+ return null;
+ if (r instanceof AltResult) {
+ if ((x = ((AltResult)r).ex) != null) {
+ d.completeThrowable(x, r);
+ break tryComplete;
+ }
+ r = null;
+ }
+ @SuppressWarnings("unchecked") T t = (T) r;
+ d.completeValue(f.apply(t));
+ } catch (Throwable ex) {
+ d.completeThrowable(ex);
+ }
+ }
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
}
}
- 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 {
- if (c != null && !c.claim())
- return false;
- if (r instanceof AltResult) {
- if ((x = ((AltResult)r).ex) != null) {
- completeThrowable(x, r);
- break tryComplete;
- }
- r = null;
- }
- @SuppressWarnings("unchecked") R rr = (R) r;
- completeValue(f.apply(rr));
- } catch (Throwable ex) {
- completeThrowable(ex);
- }
- }
- return true;
- }
-
private <U extends T,V> CompletableFuture<V> orApplyStage(
- Executor e, CompletionStage<U> o,
- Function<? super T, ? extends V> f) {
+ Executor e, CompletionStage<U> o, Function<? super T, ? extends V> f) {
CompletableFuture<U> b;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
+
+ Object r; CompletableFuture<? extends T> z;
+ if ((r = (z = this).result) != null ||
+ (r = (z = b).result) != null)
+ return z.uniApplyNow(r, e, f);
+
CompletableFuture<V> d = newIncompleteFuture();
- 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);
- }
+ orpush(b, new OrApply<T,U,V>(e, d, this, b, f));
return d;
}
@@ -1462,8 +1532,7 @@
static final class OrAccept<T,U extends T> extends BiCompletion<T,U,Void> {
Consumer<? super T> fn;
OrAccept(Executor executor, CompletableFuture<Void> dep,
- CompletableFuture<T> src,
- CompletableFuture<U> snd,
+ CompletableFuture<T> src, CompletableFuture<U> snd,
Consumer<? super T> fn) {
super(executor, dep, src, snd); this.fn = fn;
}
@@ -1471,54 +1540,47 @@
CompletableFuture<Void> d;
CompletableFuture<T> a;
CompletableFuture<U> b;
- if ((d = dep) == null ||
- !d.orAccept(a = src, b = snd, fn, mode > 0 ? null : this))
+ Object r; Throwable x; Consumer<? super T> f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (b = snd) == null
+ || ((r = a.result) == null && (r = b.result) == null))
return null;
+ tryComplete: if (d.result == null) {
+ try {
+ if (mode <= 0 && !claim())
+ return null;
+ if (r instanceof AltResult) {
+ if ((x = ((AltResult)r).ex) != null) {
+ d.completeThrowable(x, r);
+ break tryComplete;
+ }
+ r = null;
+ }
+ @SuppressWarnings("unchecked") T t = (T) r;
+ f.accept(t);
+ d.completeNull();
+ } catch (Throwable ex) {
+ d.completeThrowable(ex);
+ }
+ }
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
}
}
- 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);
- }
- }
- return true;
- }
-
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();
+
+ Object r; CompletableFuture<? extends T> z;
+ if ((r = (z = this).result) != null ||
+ (r = (z = b).result) != null)
+ return z.uniAcceptNow(r, e, f);
+
CompletableFuture<Void> d = newIncompleteFuture();
- 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);
- }
+ orpush(b, new OrAccept<T,U>(e, d, this, b, f));
return d;
}
@@ -1526,8 +1588,7 @@
static final class OrRun<T,U> extends BiCompletion<T,U,Void> {
Runnable fn;
OrRun(Executor executor, CompletableFuture<Void> dep,
- CompletableFuture<T> src,
- CompletableFuture<U> snd,
+ CompletableFuture<T> src, CompletableFuture<U> snd,
Runnable fn) {
super(executor, dep, src, snd); this.fn = fn;
}
@@ -1535,95 +1596,84 @@
CompletableFuture<Void> d;
CompletableFuture<T> a;
CompletableFuture<U> b;
- if ((d = dep) == null ||
- !d.orRun(a = src, b = snd, fn, mode > 0 ? null : this))
+ Object r; Throwable x; Runnable f;
+ if ((d = dep) == null || (f = fn) == null
+ || (a = src) == null || (b = snd) == null
+ || ((r = a.result) == null && (r = b.result) == null))
return null;
+ if (d.result == null) {
+ try {
+ if (mode <= 0 && !claim())
+ return null;
+ else if (r instanceof AltResult
+ && (x = ((AltResult)r).ex) != null)
+ d.completeThrowable(x, r);
+ else {
+ f.run();
+ d.completeNull();
+ }
+ } catch (Throwable ex) {
+ d.completeThrowable(ex);
+ }
+ }
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
}
}
- final boolean orRun(CompletableFuture<?> a, CompletableFuture<?> b,
- Runnable f, OrRun<?,?> c) {
- Object r; Throwable x;
- if (a == null || b == null ||
- ((r = a.result) == null && (r = b.result) == null) || f == null)
- return false;
- if (result == null) {
- try {
- if (c != null && !c.claim())
- return false;
- if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
- completeThrowable(x, r);
- else {
- f.run();
- completeNull();
- }
- } catch (Throwable ex) {
- completeThrowable(ex);
- }
- }
- return true;
- }
-
private CompletableFuture<Void> orRunStage(Executor e, CompletionStage<?> o,
Runnable f) {
CompletableFuture<?> b;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
+
+ Object r; CompletableFuture<?> z;
+ if ((r = (z = this).result) != null ||
+ (r = (z = b).result) != null)
+ return z.uniRunNow(r, e, f);
+
CompletableFuture<Void> d = newIncompleteFuture();
- 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);
- }
+ orpush(b, new OrRun<>(e, d, this, b, f));
return d;
}
@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) {
+ 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))
+ Object r;
+ if ((d = dep) == null
+ || (a = src) == null || (b = snd) == null
+ || ((r = a.result) == null && (r = b.result) == null))
return null;
+ d.completeRelay(r);
src = null; snd = null; dep = null;
return d.postFire(a, b, mode);
}
}
- final boolean orRelay(CompletableFuture<?> a, CompletableFuture<?> b) {
- Object r;
- if (a == null || b == null ||
- ((r = a.result) == null && (r = b.result) == null))
- return false;
- if (result == null)
- completeRelay(r);
- return true;
- }
-
/** Recursively constructs a tree of completions. */
static CompletableFuture<Object> orTree(CompletableFuture<?>[] cfs,
int lo, int hi) {
CompletableFuture<Object> d = new CompletableFuture<Object>();
if (lo <= hi) {
- CompletableFuture<?> a, b;
+ CompletableFuture<?> a, b; Object r;
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 = a.result) != null && (r = b.result) != null)
+ d.result = encodeRelay(r);
+ else
+ a.orpush(b, new OrRelay<>(d, a, b));
}
return d;
}
@@ -1640,7 +1690,7 @@
public final Void getRawResult() { return null; }
public final void setRawResult(Void v) {}
- public final boolean exec() { run(); return true; }
+ public final boolean exec() { run(); return false; }
public void run() {
CompletableFuture<T> d; Supplier<? extends T> f;
@@ -1676,7 +1726,7 @@
public final Void getRawResult() { return null; }
public final void setRawResult(Void v) {}
- public final boolean exec() { run(); return true; }
+ public final boolean exec() { run(); return false; }
public void run() {
CompletableFuture<Void> d; Runnable f;
@@ -1760,15 +1810,13 @@
private Object waitingGet(boolean interruptible) {
Signaller q = null;
boolean queued = false;
- int spins = SPINS;
Object r;
while ((r = result) == null) {
- if (spins > 0) {
- if (ThreadLocalRandom.nextSecondarySeed() >= 0)
- --spins;
+ if (q == null) {
+ q = new Signaller(interruptible, 0L, 0L);
+ if (Thread.currentThread() instanceof ForkJoinWorkerThread)
+ ForkJoinPool.helpAsyncBlocker(defaultExecutor(), q);
}
- else if (q == null)
- q = new Signaller(interruptible, 0L, 0L);
else if (!queued)
queued = tryPushStack(q);
else {
@@ -1781,16 +1829,14 @@
break;
}
}
- if (q != null) {
+ if (q != null && queued) {
q.thread = null;
- if (q.interrupted) {
- if (interruptible)
- cleanStack();
- else
- Thread.currentThread().interrupt();
- }
+ if (!interruptible && q.interrupted)
+ Thread.currentThread().interrupt();
+ if (r == null)
+ cleanStack();
}
- if (r != null)
+ if (r != null || (r = result) != null)
postComplete();
return r;
}
@@ -1808,9 +1854,12 @@
Signaller q = null;
boolean queued = false;
Object r;
- while ((r = result) == null) { // similar to untimed, without spins
- if (q == null)
+ while ((r = result) == null) { // similar to untimed
+ if (q == null) {
q = new Signaller(true, nanos, deadline);
+ if (Thread.currentThread() instanceof ForkJoinWorkerThread)
+ ForkJoinPool.helpAsyncBlocker(defaultExecutor(), q);
+ }
else if (!queued)
queued = tryPushStack(q);
else if (q.nanos <= 0L)
@@ -1825,12 +1874,13 @@
break;
}
}
- if (q != null)
+ if (q != null && queued) {
q.thread = null;
- if (r != null)
+ if (r == null)
+ cleanStack();
+ }
+ if (r != null || (r = result) != null)
postComplete();
- else
- cleanStack();
if (r != null || (q != null && q.interrupted))
return r;
}
@@ -1942,9 +1992,12 @@
* @throws InterruptedException if the current thread was interrupted
* while waiting
*/
+ @SuppressWarnings("unchecked")
public T get() throws InterruptedException, ExecutionException {
Object r;
- return reportGet((r = result) == null ? waitingGet(true) : r);
+ if ((r = result) == null)
+ r = waitingGet(true);
+ return (T) reportGet(r);
}
/**
@@ -1960,11 +2013,14 @@
* while waiting
* @throws TimeoutException if the wait timed out
*/
+ @SuppressWarnings("unchecked")
public T get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
+ long nanos = unit.toNanos(timeout);
Object r;
- long nanos = unit.toNanos(timeout);
- return reportGet((r = result) == null ? timedGet(nanos) : r);
+ if ((r = result) == null)
+ r = timedGet(nanos);
+ return (T) reportGet(r);
}
/**
@@ -1981,9 +2037,12 @@
* @throws CompletionException if this future completed
* exceptionally or a completion computation threw an exception
*/
+ @SuppressWarnings("unchecked")
public T join() {
Object r;
- return reportJoin((r = result) == null ? waitingGet(false) : r);
+ if ((r = result) == null)
+ r = waitingGet(false);
+ return (T) reportJoin(r);
}
/**
@@ -1996,9 +2055,10 @@
* @throws CompletionException if this future completed
* exceptionally or a completion computation threw an exception
*/
+ @SuppressWarnings("unchecked")
public T getNow(T valueIfAbsent) {
Object r;
- return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
+ return ((r = result) == null) ? valueIfAbsent : (T) reportJoin(r);
}
/**
@@ -2775,19 +2835,16 @@
throw new UnsupportedOperationException(); }
}
- // Unsafe mechanics
- private static final jdk.internal.misc.Unsafe U = jdk.internal.misc.Unsafe.getUnsafe();
- private static final long RESULT;
- private static final long STACK;
- private static final long NEXT;
+ // VarHandle mechanics
+ private static final VarHandle RESULT;
+ private static final VarHandle STACK;
+ private static final VarHandle NEXT;
static {
try {
- RESULT = U.objectFieldOffset
- (CompletableFuture.class.getDeclaredField("result"));
- STACK = U.objectFieldOffset
- (CompletableFuture.class.getDeclaredField("stack"));
- NEXT = U.objectFieldOffset
- (Completion.class.getDeclaredField("next"));
+ MethodHandles.Lookup l = MethodHandles.lookup();
+ RESULT = l.findVarHandle(CompletableFuture.class, "result", Object.class);
+ STACK = l.findVarHandle(CompletableFuture.class, "stack", Completion.class);
+ NEXT = l.findVarHandle(Completion.class, "next", Completion.class);
} catch (ReflectiveOperationException e) {
throw new Error(e);
}
--- a/jdk/test/java/util/concurrent/tck/CompletableFutureTest.java Fri Jul 15 13:55:51 2016 -0700
+++ b/jdk/test/java/util/concurrent/tck/CompletableFutureTest.java Fri Jul 15 13:59:58 2016 -0700
@@ -57,6 +57,7 @@
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -486,62 +487,68 @@
class FailingSupplier extends CheckedAction
implements Supplier<Integer>
{
- FailingSupplier(ExecutionMode m) { super(m); }
+ final CFException ex;
+ FailingSupplier(ExecutionMode m) { super(m); ex = new CFException(); }
public Integer get() {
invoked();
- throw new CFException();
+ throw ex;
}
}
class FailingConsumer extends CheckedIntegerAction
implements Consumer<Integer>
{
- FailingConsumer(ExecutionMode m) { super(m); }
+ final CFException ex;
+ FailingConsumer(ExecutionMode m) { super(m); ex = new CFException(); }
public void accept(Integer x) {
invoked();
value = x;
- throw new CFException();
+ throw ex;
}
}
class FailingBiConsumer extends CheckedIntegerAction
implements BiConsumer<Integer, Integer>
{
- FailingBiConsumer(ExecutionMode m) { super(m); }
+ final CFException ex;
+ FailingBiConsumer(ExecutionMode m) { super(m); ex = new CFException(); }
public void accept(Integer x, Integer y) {
invoked();
value = subtract(x, y);
- throw new CFException();
+ throw ex;
}
}
class FailingFunction extends CheckedIntegerAction
implements Function<Integer, Integer>
{
- FailingFunction(ExecutionMode m) { super(m); }
+ final CFException ex;
+ FailingFunction(ExecutionMode m) { super(m); ex = new CFException(); }
public Integer apply(Integer x) {
invoked();
value = x;
- throw new CFException();
+ throw ex;
}
}
class FailingBiFunction extends CheckedIntegerAction
implements BiFunction<Integer, Integer, Integer>
{
- FailingBiFunction(ExecutionMode m) { super(m); }
+ final CFException ex;
+ FailingBiFunction(ExecutionMode m) { super(m); ex = new CFException(); }
public Integer apply(Integer x, Integer y) {
invoked();
value = subtract(x, y);
- throw new CFException();
+ throw ex;
}
}
class FailingRunnable extends CheckedAction implements Runnable {
- FailingRunnable(ExecutionMode m) { super(m); }
+ final CFException ex;
+ FailingRunnable(ExecutionMode m) { super(m); ex = new CFException(); }
public void run() {
invoked();
- throw new CFException();
+ throw ex;
}
}
@@ -561,11 +568,21 @@
class FailingCompletableFutureFunction extends CheckedIntegerAction
implements Function<Integer, CompletableFuture<Integer>>
{
- FailingCompletableFutureFunction(ExecutionMode m) { super(m); }
+ final CFException ex;
+ FailingCompletableFutureFunction(ExecutionMode m) { super(m); ex = new CFException(); }
public CompletableFuture<Integer> apply(Integer x) {
invoked();
value = x;
- throw new CFException();
+ throw ex;
+ }
+ }
+
+ static class CountingRejectingExecutor implements Executor {
+ final RejectedExecutionException ex = new RejectedExecutionException();
+ final AtomicInteger count = new AtomicInteger(0);
+ public void execute(Runnable r) {
+ count.getAndIncrement();
+ throw ex;
}
}
@@ -1249,10 +1266,22 @@
{
final FailingRunnable r = new FailingRunnable(m);
final CompletableFuture<Void> f = m.runAsync(r);
- checkCompletedWithWrappedCFException(f);
+ checkCompletedWithWrappedException(f, r.ex);
r.assertInvoked();
}}
+ public void testRunAsync_rejectingExecutor() {
+ CountingRejectingExecutor e = new CountingRejectingExecutor();
+ try {
+ CompletableFuture.runAsync(() -> {}, e);
+ shouldThrow();
+ } catch (Throwable t) {
+ assertSame(e.ex, t);
+ }
+
+ assertEquals(1, e.count.get());
+ }
+
/**
* supplyAsync completes with result of supplier
*/
@@ -1283,10 +1312,22 @@
{
FailingSupplier r = new FailingSupplier(m);
CompletableFuture<Integer> f = m.supplyAsync(r);
- checkCompletedWithWrappedCFException(f);
+ checkCompletedWithWrappedException(f, r.ex);
r.assertInvoked();
}}
+ public void testSupplyAsync_rejectingExecutor() {
+ CountingRejectingExecutor e = new CountingRejectingExecutor();
+ try {
+ CompletableFuture.supplyAsync(() -> null, e);
+ shouldThrow();
+ } catch (Throwable t) {
+ assertSame(e.ex, t);
+ }
+
+ assertEquals(1, e.count.get());
+ }
+
// seq completion methods
/**
@@ -1405,12 +1446,12 @@
final CompletableFuture<Void> h4 = m.runAfterBoth(f, f, rs[4]);
final CompletableFuture<Void> h5 = m.runAfterEither(f, f, rs[5]);
- checkCompletedWithWrappedCFException(h0);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
- checkCompletedWithWrappedCFException(h4);
- checkCompletedWithWrappedCFException(h5);
+ checkCompletedWithWrappedException(h0, rs[0].ex);
+ checkCompletedWithWrappedException(h1, rs[1].ex);
+ checkCompletedWithWrappedException(h2, rs[2].ex);
+ checkCompletedWithWrappedException(h3, rs[3].ex);
+ checkCompletedWithWrappedException(h4, rs[4].ex);
+ checkCompletedWithWrappedException(h5, rs[5].ex);
checkCompletedNormally(f, v1);
}}
@@ -1509,10 +1550,10 @@
final CompletableFuture<Integer> h2 = m.thenApply(f, rs[2]);
final CompletableFuture<Integer> h3 = m.applyToEither(f, f, rs[3]);
- checkCompletedWithWrappedCFException(h0);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
+ checkCompletedWithWrappedException(h0, rs[0].ex);
+ checkCompletedWithWrappedException(h1, rs[1].ex);
+ checkCompletedWithWrappedException(h2, rs[2].ex);
+ checkCompletedWithWrappedException(h3, rs[3].ex);
checkCompletedNormally(f, v1);
}}
@@ -1611,10 +1652,10 @@
final CompletableFuture<Void> h2 = m.thenAccept(f, rs[2]);
final CompletableFuture<Void> h3 = m.acceptEither(f, f, rs[3]);
- checkCompletedWithWrappedCFException(h0);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
+ checkCompletedWithWrappedException(h0, rs[0].ex);
+ checkCompletedWithWrappedException(h1, rs[1].ex);
+ checkCompletedWithWrappedException(h2, rs[2].ex);
+ checkCompletedWithWrappedException(h3, rs[3].ex);
checkCompletedNormally(f, v1);
}}
@@ -1776,9 +1817,9 @@
assertTrue(snd.complete(w2));
final CompletableFuture<Integer> h3 = m.thenCombine(f, g, r3);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
+ checkCompletedWithWrappedException(h1, r1.ex);
+ checkCompletedWithWrappedException(h2, r2.ex);
+ checkCompletedWithWrappedException(h3, r3.ex);
r1.assertInvoked();
r2.assertInvoked();
r3.assertInvoked();
@@ -1940,9 +1981,9 @@
assertTrue(snd.complete(w2));
final CompletableFuture<Void> h3 = m.thenAcceptBoth(f, g, r3);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
+ checkCompletedWithWrappedException(h1, r1.ex);
+ checkCompletedWithWrappedException(h2, r2.ex);
+ checkCompletedWithWrappedException(h3, r3.ex);
r1.assertInvoked();
r2.assertInvoked();
r3.assertInvoked();
@@ -2104,9 +2145,9 @@
assertTrue(snd.complete(w2));
final CompletableFuture<Void> h3 = m.runAfterBoth(f, g, r3);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
+ checkCompletedWithWrappedException(h1, r1.ex);
+ checkCompletedWithWrappedException(h2, r2.ex);
+ checkCompletedWithWrappedException(h3, r3.ex);
r1.assertInvoked();
r2.assertInvoked();
r3.assertInvoked();
@@ -2396,10 +2437,10 @@
f.complete(v1);
final CompletableFuture<Integer> h2 = m.applyToEither(f, g, rs[2]);
final CompletableFuture<Integer> h3 = m.applyToEither(g, f, rs[3]);
- checkCompletedWithWrappedCFException(h0);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
+ checkCompletedWithWrappedException(h0, rs[0].ex);
+ checkCompletedWithWrappedException(h1, rs[1].ex);
+ checkCompletedWithWrappedException(h2, rs[2].ex);
+ checkCompletedWithWrappedException(h3, rs[3].ex);
for (int i = 0; i < 4; i++) rs[i].assertValue(v1);
g.complete(v2);
@@ -2408,10 +2449,10 @@
final CompletableFuture<Integer> h4 = m.applyToEither(f, g, rs[4]);
final CompletableFuture<Integer> h5 = m.applyToEither(g, f, rs[5]);
- checkCompletedWithWrappedCFException(h4);
+ checkCompletedWithWrappedException(h4, rs[4].ex);
assertTrue(Objects.equals(v1, rs[4].value) ||
Objects.equals(v2, rs[4].value));
- checkCompletedWithWrappedCFException(h5);
+ checkCompletedWithWrappedException(h5, rs[5].ex);
assertTrue(Objects.equals(v1, rs[5].value) ||
Objects.equals(v2, rs[5].value));
@@ -2655,10 +2696,10 @@
f.complete(v1);
final CompletableFuture<Void> h2 = m.acceptEither(f, g, rs[2]);
final CompletableFuture<Void> h3 = m.acceptEither(g, f, rs[3]);
- checkCompletedWithWrappedCFException(h0);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
+ checkCompletedWithWrappedException(h0, rs[0].ex);
+ checkCompletedWithWrappedException(h1, rs[1].ex);
+ checkCompletedWithWrappedException(h2, rs[2].ex);
+ checkCompletedWithWrappedException(h3, rs[3].ex);
for (int i = 0; i < 4; i++) rs[i].assertValue(v1);
g.complete(v2);
@@ -2667,10 +2708,10 @@
final CompletableFuture<Void> h4 = m.acceptEither(f, g, rs[4]);
final CompletableFuture<Void> h5 = m.acceptEither(g, f, rs[5]);
- checkCompletedWithWrappedCFException(h4);
+ checkCompletedWithWrappedException(h4, rs[4].ex);
assertTrue(Objects.equals(v1, rs[4].value) ||
Objects.equals(v2, rs[4].value));
- checkCompletedWithWrappedCFException(h5);
+ checkCompletedWithWrappedException(h5, rs[5].ex);
assertTrue(Objects.equals(v1, rs[5].value) ||
Objects.equals(v2, rs[5].value));
@@ -2686,6 +2727,7 @@
for (ExecutionMode m : ExecutionMode.values())
for (Integer v1 : new Integer[] { 1, null })
for (Integer v2 : new Integer[] { 2, null })
+ for (boolean pushNop : new boolean[] { true, false })
{
final CompletableFuture<Integer> f = new CompletableFuture<>();
final CompletableFuture<Integer> g = new CompletableFuture<>();
@@ -2698,6 +2740,10 @@
checkIncomplete(h1);
rs[0].assertNotInvoked();
rs[1].assertNotInvoked();
+ if (pushNop) { // ad hoc test of intra-completion interference
+ m.thenRun(f, () -> {});
+ m.thenRun(g, () -> {});
+ }
f.complete(v1);
checkCompletedNormally(h0, null);
checkCompletedNormally(h1, null);
@@ -2910,16 +2956,16 @@
assertTrue(f.complete(v1));
final CompletableFuture<Void> h2 = m.runAfterEither(f, g, rs[2]);
final CompletableFuture<Void> h3 = m.runAfterEither(g, f, rs[3]);
- checkCompletedWithWrappedCFException(h0);
- checkCompletedWithWrappedCFException(h1);
- checkCompletedWithWrappedCFException(h2);
- checkCompletedWithWrappedCFException(h3);
+ checkCompletedWithWrappedException(h0, rs[0].ex);
+ checkCompletedWithWrappedException(h1, rs[1].ex);
+ checkCompletedWithWrappedException(h2, rs[2].ex);
+ checkCompletedWithWrappedException(h3, rs[3].ex);
for (int i = 0; i < 4; i++) rs[i].assertInvoked();
assertTrue(g.complete(v2));
final CompletableFuture<Void> h4 = m.runAfterEither(f, g, rs[4]);
final CompletableFuture<Void> h5 = m.runAfterEither(g, f, rs[5]);
- checkCompletedWithWrappedCFException(h4);
- checkCompletedWithWrappedCFException(h5);
+ checkCompletedWithWrappedException(h4, rs[4].ex);
+ checkCompletedWithWrappedException(h5, rs[5].ex);
checkCompletedNormally(f, v1);
checkCompletedNormally(g, v2);
@@ -2980,7 +3026,7 @@
final CompletableFuture<Integer> g = m.thenCompose(f, r);
if (createIncomplete) assertTrue(f.complete(v1));
- checkCompletedWithWrappedCFException(g);
+ checkCompletedWithWrappedException(g, r.ex);
checkCompletedNormally(f, v1);
}}
@@ -3089,7 +3135,7 @@
}
}
- public void testAllOf_backwards() throws Exception {
+ public void testAllOf_normal_backwards() throws Exception {
for (int k = 1; k < 10; k++) {
CompletableFuture<Integer>[] fs
= (CompletableFuture<Integer>[]) new CompletableFuture[k];
@@ -3337,6 +3383,151 @@
}
/**
+ * Test submissions to an executor that rejects all tasks.
+ */
+ public void testRejectingExecutor() {
+ for (Integer v : new Integer[] { 1, null })
+ {
+ final CountingRejectingExecutor e = new CountingRejectingExecutor();
+
+ final CompletableFuture<Integer> complete = CompletableFuture.completedFuture(v);
+ final CompletableFuture<Integer> incomplete = new CompletableFuture<>();
+
+ List<CompletableFuture<?>> futures = new ArrayList<>();
+
+ List<CompletableFuture<Integer>> srcs = new ArrayList<>();
+ srcs.add(complete);
+ srcs.add(incomplete);
+
+ for (CompletableFuture<Integer> src : srcs) {
+ List<CompletableFuture<?>> fs = new ArrayList<>();
+ fs.add(src.thenRunAsync(() -> {}, e));
+ fs.add(src.thenAcceptAsync((z) -> {}, e));
+ fs.add(src.thenApplyAsync((z) -> z, e));
+
+ fs.add(src.thenCombineAsync(src, (x, y) -> x, e));
+ fs.add(src.thenAcceptBothAsync(src, (x, y) -> {}, e));
+ fs.add(src.runAfterBothAsync(src, () -> {}, e));
+
+ fs.add(src.applyToEitherAsync(src, (z) -> z, e));
+ fs.add(src.acceptEitherAsync(src, (z) -> {}, e));
+ fs.add(src.runAfterEitherAsync(src, () -> {}, e));
+
+ fs.add(src.thenComposeAsync((z) -> null, e));
+ fs.add(src.whenCompleteAsync((z, t) -> {}, e));
+ fs.add(src.handleAsync((z, t) -> null, e));
+
+ for (CompletableFuture<?> future : fs) {
+ if (src.isDone())
+ checkCompletedWithWrappedException(future, e.ex);
+ else
+ checkIncomplete(future);
+ }
+ futures.addAll(fs);
+ }
+
+ {
+ List<CompletableFuture<?>> fs = new ArrayList<>();
+
+ fs.add(complete.thenCombineAsync(incomplete, (x, y) -> x, e));
+ fs.add(incomplete.thenCombineAsync(complete, (x, y) -> x, e));
+
+ fs.add(complete.thenAcceptBothAsync(incomplete, (x, y) -> {}, e));
+ fs.add(incomplete.thenAcceptBothAsync(complete, (x, y) -> {}, e));
+
+ fs.add(complete.runAfterBothAsync(incomplete, () -> {}, e));
+ fs.add(incomplete.runAfterBothAsync(complete, () -> {}, e));
+
+ for (CompletableFuture<?> future : fs)
+ checkIncomplete(future);
+ futures.addAll(fs);
+ }
+
+ {
+ List<CompletableFuture<?>> fs = new ArrayList<>();
+
+ fs.add(complete.applyToEitherAsync(incomplete, (z) -> z, e));
+ fs.add(incomplete.applyToEitherAsync(complete, (z) -> z, e));
+
+ fs.add(complete.acceptEitherAsync(incomplete, (z) -> {}, e));
+ fs.add(incomplete.acceptEitherAsync(complete, (z) -> {}, e));
+
+ fs.add(complete.runAfterEitherAsync(incomplete, () -> {}, e));
+ fs.add(incomplete.runAfterEitherAsync(complete, () -> {}, e));
+
+ for (CompletableFuture<?> future : fs)
+ checkCompletedWithWrappedException(future, e.ex);
+ futures.addAll(fs);
+ }
+
+ incomplete.complete(v);
+
+ for (CompletableFuture<?> future : futures)
+ checkCompletedWithWrappedException(future, e.ex);
+
+ assertEquals(futures.size(), e.count.get());
+ }}
+
+ /**
+ * Test submissions to an executor that rejects all tasks, but
+ * should never be invoked because the dependent future is
+ * explicitly completed.
+ */
+ public void testRejectingExecutorNeverInvoked() {
+ for (Integer v : new Integer[] { 1, null })
+ {
+ final CountingRejectingExecutor e = new CountingRejectingExecutor();
+
+ final CompletableFuture<Integer> complete = CompletableFuture.completedFuture(v);
+ final CompletableFuture<Integer> incomplete = new CompletableFuture<>();
+
+ List<CompletableFuture<?>> futures = new ArrayList<>();
+
+ List<CompletableFuture<Integer>> srcs = new ArrayList<>();
+ srcs.add(complete);
+ srcs.add(incomplete);
+
+ List<CompletableFuture<?>> fs = new ArrayList<>();
+ fs.add(incomplete.thenRunAsync(() -> {}, e));
+ fs.add(incomplete.thenAcceptAsync((z) -> {}, e));
+ fs.add(incomplete.thenApplyAsync((z) -> z, e));
+
+ fs.add(incomplete.thenCombineAsync(incomplete, (x, y) -> x, e));
+ fs.add(incomplete.thenAcceptBothAsync(incomplete, (x, y) -> {}, e));
+ fs.add(incomplete.runAfterBothAsync(incomplete, () -> {}, e));
+
+ fs.add(incomplete.applyToEitherAsync(incomplete, (z) -> z, e));
+ fs.add(incomplete.acceptEitherAsync(incomplete, (z) -> {}, e));
+ fs.add(incomplete.runAfterEitherAsync(incomplete, () -> {}, e));
+
+ fs.add(incomplete.thenComposeAsync((z) -> null, e));
+ fs.add(incomplete.whenCompleteAsync((z, t) -> {}, e));
+ fs.add(incomplete.handleAsync((z, t) -> null, e));
+
+ fs.add(complete.thenCombineAsync(incomplete, (x, y) -> x, e));
+ fs.add(incomplete.thenCombineAsync(complete, (x, y) -> x, e));
+
+ fs.add(complete.thenAcceptBothAsync(incomplete, (x, y) -> {}, e));
+ fs.add(incomplete.thenAcceptBothAsync(complete, (x, y) -> {}, e));
+
+ fs.add(complete.runAfterBothAsync(incomplete, () -> {}, e));
+ fs.add(incomplete.runAfterBothAsync(complete, () -> {}, e));
+
+ for (CompletableFuture<?> future : fs)
+ checkIncomplete(future);
+
+ for (CompletableFuture<?> future : fs)
+ future.complete(null);
+
+ incomplete.complete(v);
+
+ for (CompletableFuture<?> future : fs)
+ checkCompletedNormally(future, null);
+
+ assertEquals(0, e.count.get());
+ }}
+
+ /**
* toCompletableFuture returns this CompletableFuture.
*/
public void testToCompletableFuture() {
@@ -3659,12 +3850,25 @@
//--- tests of implementation details; not part of official tck ---
Object resultOf(CompletableFuture<?> f) {
+ SecurityManager sm = System.getSecurityManager();
+ if (sm != null) {
+ try {
+ System.setSecurityManager(null);
+ } catch (SecurityException giveUp) {
+ return "Reflection not available";
+ }
+ }
+
try {
java.lang.reflect.Field resultField
= CompletableFuture.class.getDeclaredField("result");
resultField.setAccessible(true);
return resultField.get(f);
- } catch (Throwable t) { throw new AssertionError(t); }
+ } catch (Throwable t) {
+ throw new AssertionError(t);
+ } finally {
+ if (sm != null) System.setSecurityManager(sm);
+ }
}
public void testExceptionPropagationReusesResultObject() {
@@ -3675,33 +3879,44 @@
final CompletableFuture<Integer> v42 = CompletableFuture.completedFuture(42);
final CompletableFuture<Integer> incomplete = new CompletableFuture<>();
+ final Runnable noopRunnable = new Noop(m);
+ final Consumer<Integer> noopConsumer = new NoopConsumer(m);
+ final Function<Integer, Integer> incFunction = new IncFunction(m);
+
List<Function<CompletableFuture<Integer>, CompletableFuture<?>>> funs
= new ArrayList<>();
- funs.add((y) -> m.thenRun(y, new Noop(m)));
- funs.add((y) -> m.thenAccept(y, new NoopConsumer(m)));
- funs.add((y) -> m.thenApply(y, new IncFunction(m)));
-
- funs.add((y) -> m.runAfterEither(y, incomplete, new Noop(m)));
- funs.add((y) -> m.acceptEither(y, incomplete, new NoopConsumer(m)));
- funs.add((y) -> m.applyToEither(y, incomplete, new IncFunction(m)));
-
- funs.add((y) -> m.runAfterBoth(y, v42, new Noop(m)));
+ funs.add((y) -> m.thenRun(y, noopRunnable));
+ funs.add((y) -> m.thenAccept(y, noopConsumer));
+ funs.add((y) -> m.thenApply(y, incFunction));
+
+ funs.add((y) -> m.runAfterEither(y, incomplete, noopRunnable));
+ funs.add((y) -> m.acceptEither(y, incomplete, noopConsumer));
+ funs.add((y) -> m.applyToEither(y, incomplete, incFunction));
+
+ funs.add((y) -> m.runAfterBoth(y, v42, noopRunnable));
+ funs.add((y) -> m.runAfterBoth(v42, y, noopRunnable));
funs.add((y) -> m.thenAcceptBoth(y, v42, new SubtractAction(m)));
+ funs.add((y) -> m.thenAcceptBoth(v42, y, new SubtractAction(m)));
funs.add((y) -> m.thenCombine(y, v42, new SubtractFunction(m)));
+ funs.add((y) -> m.thenCombine(v42, y, new SubtractFunction(m)));
funs.add((y) -> m.whenComplete(y, (Integer r, Throwable t) -> {}));
funs.add((y) -> m.thenCompose(y, new CompletableFutureInc(m)));
- funs.add((y) -> CompletableFuture.allOf(new CompletableFuture<?>[] {y, v42}));
- funs.add((y) -> CompletableFuture.anyOf(new CompletableFuture<?>[] {y, incomplete}));
+ funs.add((y) -> CompletableFuture.allOf(y));
+ funs.add((y) -> CompletableFuture.allOf(y, v42));
+ funs.add((y) -> CompletableFuture.allOf(v42, y));
+ funs.add((y) -> CompletableFuture.anyOf(y));
+ funs.add((y) -> CompletableFuture.anyOf(y, incomplete));
+ funs.add((y) -> CompletableFuture.anyOf(incomplete, y));
for (Function<CompletableFuture<Integer>, CompletableFuture<?>>
fun : funs) {
CompletableFuture<Integer> f = new CompletableFuture<>();
f.completeExceptionally(ex);
- CompletableFuture<Integer> src = m.thenApply(f, new IncFunction(m));
+ CompletableFuture<Integer> src = m.thenApply(f, incFunction);
checkCompletedWithWrappedException(src, ex);
CompletableFuture<?> dep = fun.apply(src);
checkCompletedWithWrappedException(dep, ex);
@@ -3711,7 +3926,7 @@
for (Function<CompletableFuture<Integer>, CompletableFuture<?>>
fun : funs) {
CompletableFuture<Integer> f = new CompletableFuture<>();
- CompletableFuture<Integer> src = m.thenApply(f, new IncFunction(m));
+ CompletableFuture<Integer> src = m.thenApply(f, incFunction);
CompletableFuture<?> dep = fun.apply(src);
f.completeExceptionally(ex);
checkCompletedWithWrappedException(src, ex);
@@ -3725,7 +3940,7 @@
CompletableFuture<Integer> f = new CompletableFuture<>();
f.cancel(mayInterruptIfRunning);
checkCancelled(f);
- CompletableFuture<Integer> src = m.thenApply(f, new IncFunction(m));
+ CompletableFuture<Integer> src = m.thenApply(f, incFunction);
checkCompletedWithWrappedCancellationException(src);
CompletableFuture<?> dep = fun.apply(src);
checkCompletedWithWrappedCancellationException(dep);
@@ -3736,7 +3951,7 @@
for (Function<CompletableFuture<Integer>, CompletableFuture<?>>
fun : funs) {
CompletableFuture<Integer> f = new CompletableFuture<>();
- CompletableFuture<Integer> src = m.thenApply(f, new IncFunction(m));
+ CompletableFuture<Integer> src = m.thenApply(f, incFunction);
CompletableFuture<?> dep = fun.apply(src);
f.cancel(mayInterruptIfRunning);
checkCancelled(f);
@@ -3747,7 +3962,7 @@
}}
/**
- * Minimal completion stages throw UOE for all non-CompletionStage methods
+ * Minimal completion stages throw UOE for most non-CompletionStage methods
*/
public void testMinimalCompletionStage_minimality() {
if (!testImplementationDetails) return;
@@ -3776,8 +3991,10 @@
.filter((method) -> !permittedMethodSignatures.contains(toSignature.apply(method)))
.collect(Collectors.toList());
- CompletionStage<Integer> minimalStage =
- new CompletableFuture<Integer>().minimalCompletionStage();
+ List<CompletionStage<Integer>> stages = new ArrayList<>();
+ stages.add(new CompletableFuture<Integer>().minimalCompletionStage());
+ stages.add(CompletableFuture.completedStage(1));
+ stages.add(CompletableFuture.failedStage(new CFException()));
List<Method> bugs = new ArrayList<>();
for (Method method : allMethods) {
@@ -3793,20 +4010,22 @@
else if (parameterTypes[i] == long.class)
args[i] = 0L;
}
- try {
- method.invoke(minimalStage, args);
- bugs.add(method);
+ for (CompletionStage<Integer> stage : stages) {
+ try {
+ method.invoke(stage, args);
+ bugs.add(method);
+ }
+ catch (java.lang.reflect.InvocationTargetException expected) {
+ if (! (expected.getCause() instanceof UnsupportedOperationException)) {
+ bugs.add(method);
+ // expected.getCause().printStackTrace();
+ }
+ }
+ catch (ReflectiveOperationException bad) { throw new Error(bad); }
}
- catch (java.lang.reflect.InvocationTargetException expected) {
- if (! (expected.getCause() instanceof UnsupportedOperationException)) {
- bugs.add(method);
- // expected.getCause().printStackTrace();
- }
- }
- catch (ReflectiveOperationException bad) { throw new Error(bad); }
}
if (!bugs.isEmpty())
- throw new Error("Methods did not throw UOE: " + bugs.toString());
+ throw new Error("Methods did not throw UOE: " + bugs);
}
static class Monad {
@@ -3955,12 +4174,33 @@
Monad.plus(godot, Monad.unit(5L)));
}
+ /** Test long recursive chains of CompletableFutures with cascading completions */
+ public void testRecursiveChains() throws Throwable {
+ for (ExecutionMode m : ExecutionMode.values())
+ for (boolean addDeadEnds : new boolean[] { true, false })
+ {
+ final int val = 42;
+ final int n = expensiveTests ? 1_000 : 2;
+ CompletableFuture<Integer> head = new CompletableFuture<>();
+ CompletableFuture<Integer> tail = head;
+ for (int i = 0; i < n; i++) {
+ if (addDeadEnds) m.thenApply(tail, v -> v + 1);
+ tail = m.thenApply(tail, v -> v + 1);
+ if (addDeadEnds) m.applyToEither(tail, tail, v -> v + 1);
+ tail = m.applyToEither(tail, tail, v -> v + 1);
+ if (addDeadEnds) m.thenCombine(tail, tail, (v, w) -> v + 1);
+ tail = m.thenCombine(tail, tail, (v, w) -> v + 1);
+ }
+ head.complete(val);
+ assertEquals(val + 3 * n, (int) tail.join());
+ }}
+
/**
* A single CompletableFuture with many dependents.
* A demo of scalability - runtime is O(n).
*/
public void testManyDependents() throws Throwable {
- final int n = 1_000;
+ final int n = expensiveTests ? 1_000_000 : 10;
final CompletableFuture<Void> head = new CompletableFuture<>();
final CompletableFuture<Void> complete = CompletableFuture.completedFuture((Void)null);
final AtomicInteger count = new AtomicInteger(0);
@@ -3987,6 +4227,78 @@
assertEquals(5 * 3 * n, count.get());
}
+ /** ant -Dvmoptions=-Xmx8m -Djsr166.expensiveTests=true -Djsr166.tckTestClass=CompletableFutureTest tck */
+ public void testCoCompletionGarbageRetention() throws Throwable {
+ final int n = expensiveTests ? 1_000_000 : 10;
+ final CompletableFuture<Integer> incomplete = new CompletableFuture<>();
+ CompletableFuture<Integer> f;
+ for (int i = 0; i < n; i++) {
+ f = new CompletableFuture<>();
+ f.runAfterEither(incomplete, () -> {});
+ f.complete(null);
+
+ f = new CompletableFuture<>();
+ f.acceptEither(incomplete, (x) -> {});
+ f.complete(null);
+
+ f = new CompletableFuture<>();
+ f.applyToEither(incomplete, (x) -> x);
+ f.complete(null);
+
+ f = new CompletableFuture<>();
+ CompletableFuture.anyOf(new CompletableFuture<?>[] { f, incomplete });
+ f.complete(null);
+ }
+
+ for (int i = 0; i < n; i++) {
+ f = new CompletableFuture<>();
+ incomplete.runAfterEither(f, () -> {});
+ f.complete(null);
+
+ f = new CompletableFuture<>();
+ incomplete.acceptEither(f, (x) -> {});
+ f.complete(null);
+
+ f = new CompletableFuture<>();
+ incomplete.applyToEither(f, (x) -> x);
+ f.complete(null);
+
+ f = new CompletableFuture<>();
+ CompletableFuture.anyOf(new CompletableFuture<?>[] { incomplete, f });
+ f.complete(null);
+ }
+ }
+
+ /*
+ * Tests below currently fail in stress mode due to memory retention.
+ * ant -Dvmoptions=-Xmx8m -Djsr166.expensiveTests=true -Djsr166.tckTestClass=CompletableFutureTest tck
+ */
+
+ /** Checks for garbage retention with anyOf. */
+ public void testAnyOfGarbageRetention() throws Throwable {
+ for (Integer v : new Integer[] { 1, null })
+ {
+ final int n = expensiveTests ? 100_000 : 10;
+ CompletableFuture<Integer>[] fs
+ = (CompletableFuture<Integer>[]) new CompletableFuture<?>[100];
+ for (int i = 0; i < fs.length; i++)
+ fs[i] = new CompletableFuture<>();
+ fs[fs.length - 1].complete(v);
+ for (int i = 0; i < n; i++)
+ checkCompletedNormally(CompletableFuture.anyOf(fs), v);
+ }}
+
+ /** Checks for garbage retention with allOf. */
+ public void testCancelledAllOfGarbageRetention() throws Throwable {
+ final int n = expensiveTests ? 100_000 : 10;
+ CompletableFuture<Integer>[] fs
+ = (CompletableFuture<Integer>[]) new CompletableFuture<?>[100];
+ for (int i = 0; i < fs.length; i++)
+ fs[i] = new CompletableFuture<>();
+ for (int i = 0; i < n; i++)
+ assertTrue(CompletableFuture.allOf(fs).cancel(false));
+ }
+
// static <U> U join(CompletionStage<U> stage) {
// CompletableFuture<U> f = new CompletableFuture<>();
// stage.whenComplete((v, ex) -> {