archive/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2025-08-27 06:45:07 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

8277090: jsr166 refresh for jdk19

Browse source 
Reviewed-by: alanb, psandoz
This commit is contained in:
Doug Lea 2022-05-04 19:24:19 +00:00
parent c5a0687f80
commit 00e6c63cd1
13 changed files with 2726 additions and 1640 deletions
Download patch file Download diff file Expand all files Collapse all files

54
src/java.base/share/classes/java/util/concurrent/CompletableFuture.java
View file

@ -438,7 +438,7 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
/** Fallback if ForkJoinPool.commonPool() cannot support parallelism */
static final class ThreadPerTaskExecutor implements Executor {
private static final class ThreadPerTaskExecutor implements Executor {
public void execute(Runnable r) {
Objects.requireNonNull(r);
new Thread(r).start();
@ -2134,6 +2134,38 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
return ((r = result) == null) ? valueIfAbsent : (T) reportJoin(r);
}
@Override
public T resultNow() {
Object r = result;
if (r != null) {
if (r instanceof AltResult alt) {
if (alt.ex == null) return null;
} else {
@SuppressWarnings("unchecked")
T t = (T) r;
return t;
}
}
throw new IllegalStateException();
}
@Override
public Throwable exceptionNow() {
Object r = result;
Throwable x;
if (r instanceof AltResult alt
&& ((x = alt.ex) != null)
&& !(x instanceof CancellationException)) {
if (x instanceof CompletionException) {
Throwable cause = x.getCause();
if (cause != null)
x = cause;
}
return x;
}
throw new IllegalStateException();
}
/**
* If not already completed, sets the value returned by {@link
* #get()} and related methods to the given value.
@ -2509,6 +2541,20 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
return ((r = result) instanceof AltResult) && r != NIL;
}
@Override
public State state() {
Object r = result;
if (r == null)
return State.RUNNING;
if (r != NIL && r instanceof AltResult alt) {
if (alt.ex instanceof CancellationException)
return State.CANCELLED;
else
return State.FAILED;
}
return State.SUCCESS;
}
/**
* Forcibly sets or resets the value subsequently returned by
* method {@link #get()} and related methods, whether or not
@ -2912,6 +2958,10 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
throw new UnsupportedOperationException(); }
@Override public T join() {
throw new UnsupportedOperationException(); }
@Override public T resultNow() {
throw new UnsupportedOperationException(); }
@Override public Throwable exceptionNow() {
throw new UnsupportedOperationException(); }
@Override public boolean complete(T value) {
throw new UnsupportedOperationException(); }
@Override public boolean completeExceptionally(Throwable ex) {
@ -2928,6 +2978,8 @@ public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
throw new UnsupportedOperationException(); }
@Override public boolean isCompletedExceptionally() {
throw new UnsupportedOperationException(); }
@Override public State state() {
throw new UnsupportedOperationException(); }
@Override public int getNumberOfDependents() {
throw new UnsupportedOperationException(); }
@Override public CompletableFuture<T> completeAsync

29
src/java.base/share/classes/java/util/concurrent/CountedCompleter.java
View file

@ -35,8 +35,7 @@
package java.util.concurrent;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import jdk.internal.misc.Unsafe;
/**
* A {@link ForkJoinTask} with a completion action performed when
@ -540,7 +539,7 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
* @param delta the value to add
*/
public final void addToPendingCount(int delta) {
PENDING.getAndAdd(this, delta);
U.getAndAddInt(this, PENDING, delta);
}
/**
@ -552,12 +551,12 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
* @return {@code true} if successful
*/
public final boolean compareAndSetPendingCount(int expected, int count) {
return PENDING.compareAndSet(this, expected, count);
return U.compareAndSetInt(this, PENDING, expected, count);
}
// internal-only weak version
final boolean weakCompareAndSetPendingCount(int expected, int count) {
return PENDING.weakCompareAndSet(this, expected, count);
return U.weakCompareAndSetInt(this, PENDING, expected, count);
}
/**
@ -733,7 +732,6 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
if (q != null && maxTasks > 0)
q.helpComplete(this, owned, maxTasks);
}
// ForkJoinTask overrides
/**
@ -779,15 +777,16 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
@Override
protected void setRawResult(T t) { }
// VarHandle mechanics
private static final VarHandle PENDING;
/*
* This class uses jdk-internal Unsafe for atomics and special
* memory modes, rather than VarHandles, to avoid initialization
* dependencies in other jdk components that require early
* parallelism.
*/
private static final Unsafe U;
private static final long PENDING;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
PENDING = l.findVarHandle(CountedCompleter.class, "pending", int.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
U = Unsafe.getUnsafe();
PENDING = U.objectFieldOffset(CountedCompleter.class, "pending");
}
}

67
src/java.base/share/classes/java/util/concurrent/ExecutorService.java
View file

@ -102,9 +102,20 @@ import java.util.List;
* }
* }}</pre>
*
* The following method shuts down an {@code ExecutorService} in two phases,
* first by calling {@code shutdown} to reject incoming tasks, and then
* calling {@code shutdownNow}, if necessary, to cancel any lingering tasks:
* An {@code ExecutorService} may also be established and closed
* (shutdown, blocking until terminated) as follows; illustrating with
* a different {@code Executors} factory method:
*
* <pre> {@code
* try (ExecutorService e = Executors.newWorkStealingPool()) {
* // submit or execute many tasks with e ...
* }}</pre>
*
* Further customization is also possible. For example, the following
* method shuts down an {@code ExecutorService} in two phases, first
* by calling {@code shutdown} to reject incoming tasks, and then
* calling {@code shutdownNow}, if necessary, to cancel any lingering
* tasks:
*
* <pre> {@code
* void shutdownAndAwaitTermination(ExecutorService pool) {
@ -135,7 +146,7 @@ import java.util.List;
* @since 1.5
* @author Doug Lea
*/
public interface ExecutorService extends Executor {
public interface ExecutorService extends Executor, AutoCloseable {
/**
* Initiates an orderly shutdown in which previously submitted
@ -368,4 +379,52 @@ public interface ExecutorService extends Executor {
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
/**
* Initiates an orderly shutdown in which previously submitted tasks are
* executed, but no new tasks will be accepted. This method waits until all
* tasks have completed execution and the executor has terminated.
*
* <p> If interrupted while waiting, this method stops all executing tasks as
* if by invoking {@link #shutdownNow()}. It then continues to wait until all
* actively executing tasks have completed. Tasks that were awaiting
* execution are not executed. The interrupt status will be re-asserted
* before this method returns.
*
* <p> If already terminated, invoking this method has no effect.
*
* @implSpec
* The default implementation invokes {@code shutdown()} and waits for tasks
* to complete execution with {@code awaitTermination}.
*
* @throws SecurityException if a security manager exists and
* shutting down this ExecutorService may manipulate
* threads that the caller is not permitted to modify
* because it does not hold {@link
* java.lang.RuntimePermission}{@code ("modifyThread")},
* or the security manager's {@code checkAccess} method
* denies access.
* @since 19
*/
@Override
default void close() {
boolean terminated = isTerminated();
if (!terminated) {
shutdown();
boolean interrupted = false;
while (!terminated) {
try {
terminated = awaitTermination(1L, TimeUnit.DAYS);
} catch (InterruptedException e) {
if (!interrupted) {
shutdownNow();
interrupted = true;
}
}
}
if (interrupted) {
Thread.currentThread().interrupt();
}
}
}
}

2963
src/java.base/share/classes/java/util/concurrent/ForkJoinPool.java
View file

File diff suppressed because it is too large Load diff

395
src/java.base/share/classes/java/util/concurrent/ForkJoinTask.java
View file

@ -36,13 +36,12 @@
package java.util.concurrent;
import java.io.Serializable;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.lang.reflect.Constructor;
import java.util.Collection;
import java.util.List;
import java.util.RandomAccess;
import java.util.concurrent.locks.LockSupport;
import jdk.internal.misc.Unsafe;
/**
* Abstract base class for tasks that run within a {@link ForkJoinPool}.
@ -218,9 +217,11 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* This is sometimes hard to see because this file orders exported
* methods in a way that flows well in javadocs.
*
* Revision notes: The use of "Aux" field replaces previous
* reliance on a table to hold exceptions and synchronized blocks
* and monitors to wait for completion.
* Revision notes: This class uses jdk-internal Unsafe for atomics
* and special memory modes, rather than VarHandles, to avoid
* initialization dependencies in other jdk components that
* require early parallelism. It also simplifies handling of
* pool-submitted tasks, among other minor improvements.
*/
/**
@ -238,16 +239,13 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
this.ex = ex;
}
final boolean casNext(Aux c, Aux v) { // used only in cancellation
return NEXT.compareAndSet(this, c, v);
return U.compareAndSetReference(this, NEXT, c, v);
}
private static final VarHandle NEXT;
private static final Unsafe U;
private static final long NEXT;
static {
try {
NEXT = MethodHandles.lookup()
.findVarHandle(Aux.class, "next", Aux.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
U = Unsafe.getUnsafe();
NEXT = U.objectFieldOffset(Aux.class, "next");
}
}
@ -261,27 +259,41 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* control bits occupy only (some of) the upper half (16 bits) of
* status field. The lower bits are used for user-defined tags.
*/
private static final int DONE = 1 << 31; // must be negative
private static final int ABNORMAL = 1 << 16;
private static final int THROWN = 1 << 17;
private static final int SMASK = 0xffff; // short bits for tags
private static final int UNCOMPENSATE = 1 << 16; // helpJoin return sentinel
static final int DONE = 1 << 31; // must be negative
static final int ABNORMAL = 1 << 16;
static final int THROWN = 1 << 17;
static final int SMASK = 0xffff; // short bits for tags
static final int UNCOMPENSATE = 1 << 16; // helpJoin return sentinel
static final int POOLSUBMIT = 1 << 18; // for pool.submit vs fork
// flags for awaitDone (in addition to above)
static final int RAN = 1;
static final int INTERRUPTIBLE = 2;
static final int TIMED = 4;
// Fields
volatile int status; // accessed directly by pool and workers
private transient volatile Aux aux; // either waiters or thrown Exception
// Support for atomic operations
private static final VarHandle STATUS;
private static final VarHandle AUX;
private static final Unsafe U;
private static final long STATUS;
private static final long AUX;
private int getAndBitwiseOrStatus(int v) {
return (int)STATUS.getAndBitwiseOr(this, v);
return U.getAndBitwiseOrInt(this, STATUS, v);
}
private boolean casStatus(int c, int v) {
return STATUS.compareAndSet(this, c, v);
return U.compareAndSetInt(this, STATUS, c, v);
}
private boolean casAux(Aux c, Aux v) {
return AUX.compareAndSet(this, c, v);
return U.compareAndSetReference(this, AUX, c, v);
}
/**
* Marks this task as an external pool submission.
*/
final void markPoolSubmission() {
getAndBitwiseOrStatus(POOLSUBMIT);
}
/** Removes and unparks waiters */
@ -387,105 +399,69 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* Helps and/or waits for completion from join, get, or invoke;
* called from either internal or external threads.
*
* @param pool if nonnull, known submitted pool, else assumes current pool
* @param ran true if task known to have been exec'd
* @param interruptible true if park interruptibly when external
* @param timed true if use timed wait
* @param nanos if timed, timeout value
* @param how flags for POOLSUBMIT, RAN, INTERRUPTIBLE, TIMED
* @param deadline if timed, timeout deadline
* @return ABNORMAL if interrupted, else status on exit
*/
private int awaitDone(ForkJoinPool pool, boolean ran,
boolean interruptible, boolean timed,
long nanos) {
ForkJoinPool p; boolean internal; int s; Thread t;
private int awaitDone(int how, long deadline) {
int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool p;
ForkJoinPool.WorkQueue q = null;
boolean timed = (how & TIMED) != 0;
boolean owned = false, uncompensate = false;
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
owned = true;
q = (wt = (ForkJoinWorkerThread)t).workQueue;
p = wt.pool;
if (pool == null)
pool = p;
if (internal = (pool == p))
q = wt.workQueue;
}
else {
internal = false;
p = ForkJoinPool.common;
if (pool == null)
pool = p;
if (pool == p && p != null)
q = p.externalQueue();
else if ((p = ForkJoinPool.common) != null && (how & POOLSUBMIT) == 0)
q = p.externalQueue();
if (q != null && p != null) { // try helping
if (this instanceof CountedCompleter)
s = p.helpComplete(this, q, owned, timed);
else if ((how & RAN) != 0 ||
(s = q.tryRemoveAndExec(this, owned)) >= 0)
s = (owned) ? p.helpJoin(this, q, timed) : 0;
if (s < 0)
return s;
if (s == UNCOMPENSATE)
uncompensate = true;
}
if (interruptible && Thread.interrupted())
return ABNORMAL;
if ((s = status) < 0)
return s;
long deadline = 0L;
if (timed) {
if (nanos <= 0L)
return 0;
else if ((deadline = nanos + System.nanoTime()) == 0L)
deadline = 1L;
}
boolean uncompensate = false;
if (q != null && p != null) { // try helping
// help even in timed mode if pool has no parallelism
boolean canHelp = !timed || (p.mode & SMASK) == 0;
if (canHelp) {
if ((this instanceof CountedCompleter) &&
(s = p.helpComplete(this, q, internal)) < 0)
return s;
if (!ran && ((!internal && q.externalTryUnpush(this)) ||
q.tryRemove(this, internal)) && (s = doExec()) < 0)
return s;
}
if (internal) {
if ((s = p.helpJoin(this, q, canHelp)) < 0)
return s;
if (s == UNCOMPENSATE)
uncompensate = true;
}
}
// block until done or cancelled wait
boolean interrupted = false, queued = false;
boolean parked = false, fail = false;
Aux node = null;
while ((s = status) >= 0) {
Aux a; long ns;
if (fail || (fail = (pool != null && pool.mode < 0)))
casStatus(s, s | (DONE | ABNORMAL)); // try to cancel
else if (parked && Thread.interrupted()) {
if (interruptible) {
long ns = 0L;
boolean interrupted = false, queued = false;
for (;;) { // install node and await signal
Aux a;
if ((s = status) < 0)
break;
else if (node == null)
node = new Aux(Thread.currentThread(), null);
else if (!queued) {
if (((a = aux) == null || a.ex == null) &&
(queued = casAux(node.next = a, node)))
LockSupport.setCurrentBlocker(this);
}
else if (timed && (ns = deadline - System.nanoTime()) <= 0) {
s = 0;
break;
}
else if (Thread.interrupted()) {
interrupted = true;
if ((how & POOLSUBMIT) != 0 && p != null && p.runState < 0)
cancelIgnoringExceptions(this); // cancel on shutdown
else if ((how & INTERRUPTIBLE) != 0) {
s = ABNORMAL;
break;
}
interrupted = true;
}
else if (queued) {
if (deadline != 0L) {
if ((ns = deadline - System.nanoTime()) <= 0L)
break;
LockSupport.parkNanos(ns);
}
else
LockSupport.park();
parked = true;
}
else if (node != null) {
if ((a = aux) != null && a.ex != null)
Thread.onSpinWait(); // exception in progress
else if (queued = casAux(node.next = a, node))
LockSupport.setCurrentBlocker(this);
}
else {
try {
node = new Aux(Thread.currentThread(), null);
} catch (Throwable ex) { // cannot create
fail = true;
}
}
else if ((s = status) < 0) // recheck
break;
else if (timed)
LockSupport.parkNanos(ns);
else
LockSupport.park();
}
if (pool != null && uncompensate)
pool.uncompensate();
if (uncompensate)
p.uncompensate();
if (queued) {
LockSupport.setCurrentBlocker(null);
@ -578,8 +554,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
*/
private Throwable getException(int s) {
Throwable ex = null;
if ((s & ABNORMAL) != 0 &&
((s & THROWN) == 0 || (ex = getThrowableException()) == null))
if ((s & ABNORMAL) != 0 && (ex = getThrowableException()) == null)
ex = new CancellationException();
return ex;
}
@ -589,8 +564,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* CancellationException if none recorded.
*/
private void reportException(int s) {
ForkJoinTask.<RuntimeException>uncheckedThrow(
(s & THROWN) != 0 ? getThrowableException() : null);
ForkJoinTask.<RuntimeException>uncheckedThrow(getThrowableException());
}
/**
@ -598,13 +572,13 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* necessary in an ExecutionException.
*/
private void reportExecutionException(int s) {
Throwable ex = null;
Throwable ex = null, rx;
if (s == ABNORMAL)
ex = new InterruptedException();
else if (s >= 0)
ex = new TimeoutException();
else if ((s & THROWN) != 0 && (ex = getThrowableException()) != null)
ex = new ExecutionException(ex);
else if ((rx = getThrowableException()) != null)
ex = new ExecutionException(rx);
ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
}
@ -647,11 +621,16 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* @return {@code this}, to simplify usage
*/
public final ForkJoinTask<V> fork() {
Thread t; ForkJoinWorkerThread w;
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
(w = (ForkJoinWorkerThread)t).workQueue.push(this, w.pool);
Thread t; ForkJoinWorkerThread wt;
ForkJoinPool p; ForkJoinPool.WorkQueue q;
U.storeStoreFence(); // ensure safely publishable
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
p = (wt = (ForkJoinWorkerThread)t).pool;
q = wt.workQueue;
}
else
ForkJoinPool.common.externalPush(this);
q = (p = ForkJoinPool.common).submissionQueue(false);
q.push(this, p, true);
return this;
}
@ -669,7 +648,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
public final V join() {
int s;
if ((s = status) >= 0)
s = awaitDone(null, false, false, false, 0L);
s = awaitDone(s & POOLSUBMIT, 0L);
if ((s & ABNORMAL) != 0)
reportException(s);
return getRawResult();
@ -686,7 +665,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
public final V invoke() {
int s;
if ((s = doExec()) >= 0)
s = awaitDone(null, true, false, false, 0L);
s = awaitDone(RAN, 0L);
if ((s & ABNORMAL) != 0)
reportException(s);
return getRawResult();
@ -715,13 +694,17 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
throw new NullPointerException();
t2.fork();
if ((s1 = t1.doExec()) >= 0)
s1 = t1.awaitDone(null, true, false, false, 0L);
s1 = t1.awaitDone(RAN, 0L);
if ((s1 & ABNORMAL) != 0) {
cancelIgnoringExceptions(t2);
t1.reportException(s1);
}
else if (((s2 = t2.awaitDone(null, false, false, false, 0L)) & ABNORMAL) != 0)
t2.reportException(s2);
else {
if ((s2 = t2.status) >= 0)
s2 = t2.awaitDone(0, 0L);
if ((s2 & ABNORMAL) != 0)
t2.reportException(s2);
}
}
/**
@ -751,7 +734,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
if (i == 0) {
int s;
if ((s = t.doExec()) >= 0)
s = t.awaitDone(null, true, false, false, 0L);
s = t.awaitDone(RAN, 0L);
if ((s & ABNORMAL) != 0)
ex = t.getException(s);
break;
@ -764,7 +747,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
if ((t = tasks[i]) != null) {
int s;
if ((s = t.status) >= 0)
s = t.awaitDone(null, false, false, false, 0L);
s = t.awaitDone(0, 0L);
if ((s & ABNORMAL) != 0 && (ex = t.getException(s)) != null)
break;
}
@ -814,7 +797,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
if (i == 0) {
int s;
if ((s = t.doExec()) >= 0)
s = t.awaitDone(null, true, false, false, 0L);
s = t.awaitDone(RAN, 0L);
if ((s & ABNORMAL) != 0)
ex = t.getException(s);
break;
@ -827,7 +810,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
if ((t = ts.get(i)) != null) {
int s;
if ((s = t.status) >= 0)
s = t.awaitDone(null, false, false, false, 0L);
s = t.awaitDone(0, 0L);
if ((s & ABNORMAL) != 0 && (ex = t.getException(s)) != null)
break;
}
@ -900,6 +883,29 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
return (status & (DONE | ABNORMAL)) == DONE;
}
@Override
public State state() {
int s = status;
return (s >= 0) ? State.RUNNING :
((s & (DONE | ABNORMAL)) == DONE) ? State.SUCCESS:
((s & (ABNORMAL | THROWN)) == (ABNORMAL | THROWN)) ? State.FAILED :
State.CANCELLED;
}
@Override
public V resultNow() {
if (!isCompletedNormally())
throw new IllegalStateException();
return getRawResult();
}
@Override
public Throwable exceptionNow() {
if ((status & (ABNORMAL | THROWN)) != (ABNORMAL | THROWN))
throw new IllegalStateException();
return getThrowableException();
}
/**
* Returns the exception thrown by the base computation, or a
* {@code CancellationException} if cancelled, or {@code null} if
@ -978,7 +984,11 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* member of a ForkJoinPool and was interrupted while waiting
*/
public final V get() throws InterruptedException, ExecutionException {
int s = awaitDone(null, false, true, false, 0L);
int s;
if (Thread.interrupted())
s = ABNORMAL;
else if ((s = status) >= 0)
s = awaitDone((s & POOLSUBMIT) | INTERRUPTIBLE, 0L);
if ((s & ABNORMAL) != 0)
reportExecutionException(s);
return getRawResult();
@ -1001,7 +1011,12 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
public final V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
long nanos = unit.toNanos(timeout);
int s = awaitDone(null, false, true, true, nanos);
int s;
if (Thread.interrupted())
s = ABNORMAL;
else if ((s = status) >= 0 && nanos > 0L)
s = awaitDone((s & POOLSUBMIT) | INTERRUPTIBLE | TIMED,
nanos + System.nanoTime());
if (s >= 0 || (s & ABNORMAL) != 0)
reportExecutionException(s);
return getRawResult();
@ -1014,49 +1029,65 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* known to have aborted.
*/
public final void quietlyJoin() {
if (status >= 0)
awaitDone(null, false, false, false, 0L);
int s;
if ((s = status) >= 0)
awaitDone(s & POOLSUBMIT, 0L);
}
/**
* Commences performing this task and awaits its completion if
* necessary, without returning its result or throwing its
* exception.
*/
public final void quietlyInvoke() {
if (doExec() >= 0)
awaitDone(null, true, false, false, 0L);
int s;
if ((s = doExec()) >= 0)
awaitDone(RAN, 0L);
}
// Versions of join/get for pool.invoke* methods that use external,
// possibly-non-commonPool submits
/**
* Tries to join this task, returning true if it completed
* (possibly exceptionally) before the given timeout and
* the current thread has not been interrupted.
*
* @param timeout the maximum time to wait
* @param unit the time unit of the timeout argument
* @return true if this task completed
* @throws InterruptedException if the current thread was
* interrupted while waiting
* @since 19
*/
public final boolean quietlyJoin(long timeout, TimeUnit unit)
throws InterruptedException {
int s;
long nanos = unit.toNanos(timeout);
if (Thread.interrupted())
s = ABNORMAL;
else if ((s = status) >= 0 && nanos > 0L)
s = awaitDone((s & POOLSUBMIT) | INTERRUPTIBLE | TIMED,
nanos + System.nanoTime());
if (s == ABNORMAL)
throw new InterruptedException();
else
return (s < 0);
}
final void awaitPoolInvoke(ForkJoinPool pool) {
awaitDone(pool, false, false, false, 0L);
}
final void awaitPoolInvoke(ForkJoinPool pool, long nanos) {
awaitDone(pool, false, true, true, nanos);
}
final V joinForPoolInvoke(ForkJoinPool pool) {
int s = awaitDone(pool, false, false, false, 0L);
if ((s & ABNORMAL) != 0)
reportException(s);
return getRawResult();
}
final V getForPoolInvoke(ForkJoinPool pool)
throws InterruptedException, ExecutionException {
int s = awaitDone(pool, false, true, false, 0L);
if ((s & ABNORMAL) != 0)
reportExecutionException(s);
return getRawResult();
}
final V getForPoolInvoke(ForkJoinPool pool, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
int s = awaitDone(pool, false, true, true, nanos);
if (s >= 0 || (s & ABNORMAL) != 0)
reportExecutionException(s);
return getRawResult();
/**
* Tries to join this task, returning true if it completed
* (possibly exceptionally) before the given timeout.
*
* @param timeout the maximum time to wait
* @param unit the time unit of the timeout argument
* @return true if this task completed
* @since 19
*/
public final boolean quietlyJoinUninterruptibly(long timeout,
TimeUnit unit) {
int s;
long nanos = unit.toNanos(timeout);
if ((s = status) >= 0 && nanos > 0L)
s = awaitDone((s & POOLSUBMIT) | TIMED, nanos + System.nanoTime());
return (s < 0);
}
/**
@ -1067,12 +1098,7 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* all are processed.
*/
public static void helpQuiesce() {
Thread t; ForkJoinWorkerThread w; ForkJoinPool p;
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread &&
(p = (w = (ForkJoinWorkerThread)t).pool) != null)
p.helpQuiescePool(w.workQueue, Long.MAX_VALUE, false);
else
ForkJoinPool.common.externalHelpQuiescePool(Long.MAX_VALUE, false);
ForkJoinPool.helpQuiescePool(null, Long.MAX_VALUE, false);
}
/**
@ -1134,12 +1160,12 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* @return {@code true} if unforked
*/
public boolean tryUnfork() {
Thread t; ForkJoinPool.WorkQueue q;
return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
? (q = ((ForkJoinWorkerThread)t).workQueue) != null
&& q.tryUnpush(this)
: (q = ForkJoinPool.commonQueue()) != null
&& q.externalTryUnpush(this);
Thread t; ForkJoinPool.WorkQueue q; boolean owned;
if (owned = (t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
q = ((ForkJoinWorkerThread)t).workQueue;
else
q = ForkJoinPool.commonQueue();
return (q != null && q.tryUnpush(this, owned));
}
/**
@ -1541,11 +1567,10 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
* @param <T> the type of the callable's result
* @return the task
*
* @since 17
* @since 19
*/
// adaptInterruptible deferred to its own independent change
// https://bugs.openjdk.java.net/browse/JDK-8246587
/* TODO: public */ private static <T> ForkJoinTask<T> adaptInterruptible(Callable<? extends T> callable) {
public static <T> ForkJoinTask<T> adaptInterruptible(Callable<? extends T> callable) {
// https://bugs.openjdk.java.net/browse/JDK-8246587
return new AdaptedInterruptibleCallable<T>(callable);
}
@ -1584,13 +1609,11 @@ public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
}
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
STATUS = l.findVarHandle(ForkJoinTask.class, "status", int.class);
AUX = l.findVarHandle(ForkJoinTask.class, "aux", Aux.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
U = Unsafe.getUnsafe();
STATUS = U.objectFieldOffset(ForkJoinTask.class, "status");
AUX = U.objectFieldOffset(ForkJoinTask.class, "aux");
Class<?> dep1 = LockSupport.class; // ensure loaded
Class<?> dep2 = Aux.class;
}
}

74
src/java.base/share/classes/java/util/concurrent/ForkJoinWorkerThread.java
View file

@ -36,7 +36,9 @@
package java.util.concurrent;
import java.security.AccessController;
import java.security.AccessControlContext;
import java.security.PrivilegedAction;
import java.security.ProtectionDomain;
/**
* A thread managed by a {@link ForkJoinPool}, which executes
@ -70,10 +72,13 @@ public class ForkJoinWorkerThread extends Thread {
* Full nonpublic constructor.
*/
ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool,
boolean useSystemClassLoader, boolean isInnocuous) {
super(group, null, pool.nextWorkerThreadName(), 0L);
boolean useSystemClassLoader,
boolean clearThreadLocals) {
super(group, null, pool.nextWorkerThreadName(), 0L, !clearThreadLocals);
UncaughtExceptionHandler handler = (this.pool = pool).ueh;
this.workQueue = new ForkJoinPool.WorkQueue(this, isInnocuous);
this.workQueue = new ForkJoinPool.WorkQueue(this, 0);
if (clearThreadLocals)
workQueue.setClearThreadLocals();
super.setDaemon(true);
if (handler != null)
super.setUncaughtExceptionHandler(handler);
@ -83,14 +88,21 @@ public class ForkJoinWorkerThread extends Thread {
/**
* Creates a ForkJoinWorkerThread operating in the given thread group and
* pool.
* pool, and with the given policy for preserving ThreadLocals.
*
* @param group if non-null, the thread group for this thread
* @param group if non-null, the thread group for this
* thread. Otherwise, the thread group is chosen by the security
* manager if present, else set to the current thread's thread
* group.
* @param pool the pool this thread works in
* @param preserveThreadLocals if true, always preserve the values of
* ThreadLocal variables across tasks; otherwise they may be cleared.
* @throws NullPointerException if pool is null
* @since 19
*/
/* TODO: protected */ ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool) {
this(group, pool, false, false);
protected ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool,
boolean preserveThreadLocals) {
this(group, pool, false, !preserveThreadLocals);
}
/**
@ -181,26 +193,24 @@ public class ForkJoinWorkerThread extends Thread {
/**
* A worker thread that has no permissions, is not a member of any
* user-defined ThreadGroup, uses the system class loader as
* thread context class loader, and erases all ThreadLocals after
* thread context class loader, and clears all ThreadLocals after
* running each top-level task.
*/
static final class InnocuousForkJoinWorkerThread extends ForkJoinWorkerThread {
/** The ThreadGroup for all InnocuousForkJoinWorkerThreads */
private static final ThreadGroup innocuousThreadGroup;
@SuppressWarnings("removal")
private static final ThreadGroup innocuousThreadGroup =
AccessController.doPrivileged(new PrivilegedAction<>() {
public ThreadGroup run() {
ThreadGroup group = Thread.currentThread().getThreadGroup();
for (ThreadGroup p; (p = group.getParent()) != null; )
group = p;
return new ThreadGroup(
group, "InnocuousForkJoinWorkerThreadGroup");
}});
private static final AccessControlContext innocuousACC;
InnocuousForkJoinWorkerThread(ForkJoinPool pool) {
super(innocuousThreadGroup, pool, true, true);
}
@Override @SuppressWarnings("removal")
protected void onStart() {
Thread t = Thread.currentThread();
ThreadLocalRandom.setInheritedAccessControlContext(t, innocuousACC);
}
@Override // to silently fail
public void setUncaughtExceptionHandler(UncaughtExceptionHandler x) { }
@ -209,5 +219,33 @@ public class ForkJoinWorkerThread extends Thread {
if (cl != null && ClassLoader.getSystemClassLoader() != cl)
throw new SecurityException("setContextClassLoader");
}
@SuppressWarnings("removal")
static AccessControlContext createACC() {
return new AccessControlContext(
new ProtectionDomain[] { new ProtectionDomain(null, null) });
}
static ThreadGroup createGroup() {
ThreadGroup group = Thread.currentThread().getThreadGroup();
for (ThreadGroup p; (p = group.getParent()) != null; )
group = p;
return new ThreadGroup(group, "InnocuousForkJoinWorkerThreadGroup");
}
static {
@SuppressWarnings("removal")
SecurityManager sm = System.getSecurityManager();
@SuppressWarnings("removal")
ThreadGroup g = innocuousThreadGroup =
(sm == null) ? createGroup() :
AccessController.doPrivileged(new PrivilegedAction<>() {
public ThreadGroup run() {
return createGroup(); }});
@SuppressWarnings("removal")
AccessControlContext a = innocuousACC =
(sm == null) ? createACC() :
AccessController.doPrivileged(new PrivilegedAction<>() {
public AccessControlContext run() {
return createACC(); }});
}
}
}

139
src/java.base/share/classes/java/util/concurrent/Future.java
View file

@ -161,4 +161,143 @@ public interface Future<V> {
*/
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
/**
* Returns the computed result, without waiting.
*
* <p> This method is for cases where the caller knows that the task has
* already completed successfully, for example when filtering a stream
* of Future objects for the successful tasks and using a mapping
* operation to obtain a stream of results.
* {@snippet lang=java :
* results = futures.stream()
* .filter(f -> f.state() == Future.State.SUCCESS)
* .map(Future::resultNow)
* .toList();
* }
*
* @implSpec
* The default implementation invokes {@code isDone()} to test if the task
* has completed. If done, it invokes {@code get()} to obtain the result.
*
* @return the computed result
* @throws IllegalStateException if the task has not completed or the task
* did not complete with a result
* @since 19
*/
default V resultNow() {
if (!isDone())
throw new IllegalStateException("Task has not completed");
boolean interrupted = false;
try {
while (true) {
try {
return get();
} catch (InterruptedException e) {
interrupted = true;
} catch (ExecutionException e) {
throw new IllegalStateException("Task completed with exception");
} catch (CancellationException e) {
throw new IllegalStateException("Task was cancelled");
}
}
} finally {
if (interrupted) Thread.currentThread().interrupt();
}
}
/**
* Returns the exception thrown by the task, without waiting.
*
* <p> This method is for cases where the caller knows that the task
* has already completed with an exception.
*
* @implSpec
* The default implementation invokes {@code isDone()} to test if the task
* has completed. If done and not cancelled, it invokes {@code get()} and
* catches the {@code ExecutionException} to obtain the exception.
*
* @return the exception thrown by the task
* @throws IllegalStateException if the task has not completed, the task
* completed normally, or the task was cancelled
* @since 19
*/
default Throwable exceptionNow() {
if (!isDone())
throw new IllegalStateException("Task has not completed");
if (isCancelled())
throw new IllegalStateException("Task was cancelled");
boolean interrupted = false;
try {
while (true) {
try {
get();
throw new IllegalStateException("Task completed with a result");
} catch (InterruptedException e) {
interrupted = true;
} catch (ExecutionException e) {
return e.getCause();
}
}
} finally {
if (interrupted) Thread.currentThread().interrupt();
}
}
/**
* Represents the computation state.
* @since 19
*/
enum State {
/**
* The task has not completed.
*/
RUNNING,
/**
* The task completed with a result.
* @see Future#resultNow()
*/
SUCCESS,
/**
* The task completed with an exception.
* @see Future#exceptionNow()
*/
FAILED,
/**
* The task was cancelled.
* @see #cancel(boolean)
*/
CANCELLED
}
/**
* {@return the computation state}
*
* @implSpec
* The default implementation uses {@code isDone()}, {@code isCancelled()},
* and {@code get()} to determine the state.
*
* @since 19
*/
default State state() {
if (!isDone())
return State.RUNNING;
if (isCancelled())
return State.CANCELLED;
boolean interrupted = false;
try {
while (true) {
try {
get(); // may throw InterruptedException when done
return State.SUCCESS;
} catch (InterruptedException e) {
interrupted = true;
} catch (ExecutionException e) {
return State.FAILED;
}
}
} finally {
if (interrupted) Thread.currentThread().interrupt();
}
}
}

53
src/java.base/share/classes/java/util/concurrent/FutureTask.java
View file

@ -205,6 +205,59 @@ public class FutureTask<V> implements RunnableFuture<V> {
return report(s);
}
@Override
public V resultNow() {
switch (state()) { // Future.State
case SUCCESS:
@SuppressWarnings("unchecked")
V result = (V) outcome;
return result;
case FAILED:
throw new IllegalStateException("Task completed with exception");
case CANCELLED:
throw new IllegalStateException("Task was cancelled");
default:
throw new IllegalStateException("Task has not completed");
}
}
@Override
public Throwable exceptionNow() {
switch (state()) { // Future.State
case SUCCESS:
throw new IllegalStateException("Task completed with a result");
case FAILED:
Object x = outcome;
return (Throwable) x;
case CANCELLED:
throw new IllegalStateException("Task was cancelled");
default:
throw new IllegalStateException("Task has not completed");
}
}
@Override
public State state() {
int s = state;
while (s == COMPLETING) {
// waiting for transition to NORMAL or EXCEPTIONAL
Thread.yield();
s = state;
}
switch (s) {
case NORMAL:
return State.SUCCESS;
case EXCEPTIONAL:
return State.FAILED;
case CANCELLED:
case INTERRUPTING:
case INTERRUPTED:
return State.CANCELLED;
default:
return State.RUNNING;
}
}
/**
* Protected method invoked when this task transitions to state
* {@code isDone} (whether normally or via cancellation). The

46
src/java.base/share/classes/java/util/concurrent/RecursiveTask.java
View file

@ -38,30 +38,38 @@ package java.util.concurrent;
/**
* A recursive result-bearing {@link ForkJoinTask}.
*
* <p>For a classic example, here is a task computing Fibonacci numbers:
* <p>For example, here is a task-based program for computing Factorials:
*
* <pre> {@code
* class Fibonacci extends RecursiveTask<Integer> {
* final int n;
* Fibonacci(int n) { this.n = n; }
* protected Integer compute() {
* if (n <= 1)
* return n;
* Fibonacci f1 = new Fibonacci(n - 1);
* f1.fork();
* Fibonacci f2 = new Fibonacci(n - 2);
* return f2.compute() + f1.join();
* import java.util.concurrent.RecursiveTask;
* import java.math.BigInteger;
* public class Factorial {
* static class FactorialTask extends RecursiveTask<BigInteger> {
* private final int from, to;
* FactorialTask(int from, int to) { this.from = from; this.to = to; }
* protected BigInteger compute() {
* int range = to - from;
* if (range == 0) { // base case
* return BigInteger.valueOf(from);
* } else if (range == 1) { // too small to parallelize
* return BigInteger.valueOf(from).multiply(BigInteger.valueOf(to));
* } else { // split in half
* int mid = from + range / 2;
* FactorialTask leftTask = new FactorialTask(from, mid);
* leftTask.fork(); // perform about half the work locally
* return new FactorialTask(mid + 1, to).compute()
* .multiply(leftTask.join());
* }
* }
* }
* static BigInteger factorial(int n) { // uses ForkJoinPool.commonPool()
* return (n <= 1) ? BigInteger.ONE : new FactorialTask(1, n).invoke();
* }
* public static void main(String[] args) {
* System.out.println(factorial(Integer.parseInt(args[0])));
* }
* }}</pre>
*
* However, besides being a dumb way to compute Fibonacci functions
* (there is a simple fast linear algorithm that you'd use in
* practice), this is likely to perform poorly because the smallest
* subtasks are too small to be worthwhile splitting up. Instead, as
* is the case for nearly all fork/join applications, you'd pick some
* minimum granularity size (for example 10 here) for which you always
* sequentially solve rather than subdividing.
*
* @param <V> the type of the result of the task
*
* @since 1.7

4
src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedLongSynchronizer.java
View file

@ -1156,7 +1156,9 @@ public abstract class AbstractQueuedLongSynchronizer
*/
private boolean canReacquire(ConditionNode node) {
// check links, not status to avoid enqueue race
return node != null && node.prev != null && isEnqueued(node);
Node p; // traverse unless known to be bidirectionally linked
return node != null && (p = node.prev) != null &&
(p.next == node || isEnqueued(node));
}
/**

4
src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedSynchronizer.java
View file

@ -1524,7 +1524,9 @@ public abstract class AbstractQueuedSynchronizer
*/
private boolean canReacquire(ConditionNode node) {
// check links, not status to avoid enqueue race
return node != null && node.prev != null && isEnqueued(node);
Node p; // traverse unless known to be bidirectionally linked
return node != null && (p = node.prev) != null &&
(p.next == node || isEnqueued(node));
}
/**

495
test/jdk/java/util/concurrent/tck/ForkJoinPool19Test.java Normal file
View file

@ -0,0 +1,495 @@
/*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
/*
* This file is available under and governed by the GNU General Public
* License version 2 only, as published by the Free Software Foundation.
* However, the following notice accompanied the original version of this
* file:
*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/publicdomain/zero/1.0/
*/
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import java.util.HashSet;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountedCompleter;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.Callable;
import junit.framework.Test;
import junit.framework.TestSuite;
/**
* Tests for ForkJoinPool and corresponding ForkJoinTask additions.
*/
public class ForkJoinPool19Test extends JSR166TestCase {
public static void main(String[] args) {
main(suite(), args);
}
public static Test suite() {
return new TestSuite(ForkJoinPool8Test.class);
}
/**
* SetParallelism sets reported parallellism and returns previous value
*/
public void testSetParallelism() {
final ForkJoinPool p = new ForkJoinPool(2);
assertEquals(2, p.getParallelism());
assertEquals(2, p.setParallelism(3));
assertEquals(3, p.setParallelism(2));
p.shutdown();
}
/**
* SetParallelism throws exception if argument out of bounds
*/
public void testSetParallelismBadArgs() {
final ForkJoinPool p = new ForkJoinPool(2);
try {
p.setParallelism(0);
shouldThrow();
} catch (Exception success) {
}
try {
p.setParallelism(Integer.MAX_VALUE);
shouldThrow();
} catch (Exception success) {
}
assertEquals(2, p.getParallelism());
p.shutdown();
}
/*
* Some test methods adapted from RecursiveAction
*/
private static ForkJoinPool mainPool() {
return new ForkJoinPool();
}
private void testInvokeOnPool(ForkJoinPool pool, RecursiveAction a) {
try (PoolCleaner cleaner = cleaner(pool)) {
checkNotDone(a);
assertNull(pool.invoke(a));
checkCompletedNormally(a);
}
}
private void checkInvoke(ForkJoinTask<?> a) {
checkNotDone(a);
assertNull(a.invoke());
checkCompletedNormally(a);
}
void checkNotDone(ForkJoinTask<?> a) {
assertFalse(a.isDone());
assertFalse(a.isCompletedNormally());
assertFalse(a.isCompletedAbnormally());
assertFalse(a.isCancelled());
assertNull(a.getException());
assertNull(a.getRawResult());
if (! ForkJoinTask.inForkJoinPool()) {
Thread.currentThread().interrupt();
try {
a.get();
shouldThrow();
} catch (InterruptedException success) {
} catch (Throwable fail) { threadUnexpectedException(fail); }
Thread.currentThread().interrupt();
try {
a.get(randomTimeout(), randomTimeUnit());
shouldThrow();
} catch (InterruptedException success) {
} catch (Throwable fail) { threadUnexpectedException(fail); }
}
try {
a.get(randomExpiredTimeout(), randomTimeUnit());
shouldThrow();
} catch (TimeoutException success) {
} catch (Throwable fail) { threadUnexpectedException(fail); }
}
void checkCompletedNormally(ForkJoinTask<?> a) {
assertTrue(a.isDone());
assertFalse(a.isCancelled());
assertTrue(a.isCompletedNormally());
assertFalse(a.isCompletedAbnormally());
assertNull(a.getException());
assertNull(a.getRawResult());
assertNull(a.join());
assertFalse(a.cancel(false));
assertFalse(a.cancel(true));
Object v1 = null, v2 = null;
try {
v1 = a.get();
v2 = a.get(randomTimeout(), randomTimeUnit());
} catch (Throwable fail) { threadUnexpectedException(fail); }
assertNull(v1);
assertNull(v2);
}
void checkCancelled(ForkJoinTask<?> a) {
assertTrue(a.isDone());
assertTrue(a.isCancelled());
assertFalse(a.isCompletedNormally());
assertTrue(a.isCompletedAbnormally());
assertTrue(a.getException() instanceof CancellationException);
assertNull(a.getRawResult());
try {
a.join();
shouldThrow();
} catch (CancellationException success) {
} catch (Throwable fail) { threadUnexpectedException(fail); }
try {
a.get();
shouldThrow();
} catch (CancellationException success) {
} catch (Throwable fail) { threadUnexpectedException(fail); }
try {
a.get(randomTimeout(), randomTimeUnit());
shouldThrow();
} catch (CancellationException success) {
} catch (Throwable fail) { threadUnexpectedException(fail); }
}
void checkCompletedAbnormally(ForkJoinTask<?> a, Throwable t) {
assertTrue(a.isDone());
assertFalse(a.isCancelled());
assertFalse(a.isCompletedNormally());
assertTrue(a.isCompletedAbnormally());
assertSame(t.getClass(), a.getException().getClass());
assertNull(a.getRawResult());
assertFalse(a.cancel(false));
assertFalse(a.cancel(true));
try {
a.join();
shouldThrow();
} catch (Throwable expected) {
assertSame(expected.getClass(), t.getClass());
}
try {
a.get();
shouldThrow();
} catch (ExecutionException success) {
assertSame(t.getClass(), success.getCause().getClass());
} catch (Throwable fail) { threadUnexpectedException(fail); }
try {
a.get(randomTimeout(), randomTimeUnit());
shouldThrow();
} catch (ExecutionException success) {
assertSame(t.getClass(), success.getCause().getClass());
} catch (Throwable fail) { threadUnexpectedException(fail); }
}
public static final class FJException extends RuntimeException {
public FJException() { super(); }
public FJException(Throwable cause) { super(cause); }
}
/** A simple recursive action for testing. */
final class FibAction extends CheckedRecursiveAction {
final int number;
int result;
FibAction(int n) { number = n; }
protected void realCompute() {
int n = number;
if (n <= 1)
result = n;
else {
FibAction f1 = new FibAction(n - 1);
FibAction f2 = new FibAction(n - 2);
invokeAll(f1, f2);
result = f1.result + f2.result;
}
}
}
/** A recursive action failing in base case. */
static final class FailingFibAction extends RecursiveAction {
final int number;
int result;
FailingFibAction(int n) { number = n; }
public void compute() {
int n = number;
if (n <= 1)
throw new FJException();
else {
FailingFibAction f1 = new FailingFibAction(n - 1);
FailingFibAction f2 = new FailingFibAction(n - 2);
invokeAll(f1, f2);
result = f1.result + f2.result;
}
}
}
/**
* lazySubmit submits a task that is not executed until new
* workers are created or it is explicitly joined by a worker.
*/
public void testLazySubmit() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
final ForkJoinPool p = mainPool();
FibAction f = new FibAction(8);
p.lazySubmit(f);
checkNotDone(f);
FibAction g = new FibAction(8);
p.submit(g);
g.join();
f.join();
assertEquals(21, f.result);
checkCompletedNormally(f);
}};
testInvokeOnPool(mainPool(), a);
}
/**
* quietlyInvoke task returns when task completes normally.
* isCompletedAbnormally and isCancelled return false for normally
* completed tasks
*/
public void testQuietlyInvoke() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
FibAction f = new FibAction(8);
f.quietlyInvoke();
assertEquals(21, f.result);
checkCompletedNormally(f);
}};
checkInvoke(a);
}
/**
* join of a forked task returns when task completes
*/
public void testForkJoin() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
FibAction f = new FibAction(8);
assertSame(f, f.fork());
assertNull(f.join());
assertEquals(21, f.result);
checkCompletedNormally(f);
}};
checkInvoke(a);
}
/**
* timed quietlyJoinUninterruptibly of a forked task succeeds in
* the presence of interrupts
*/
public void testTimedQuietlyJoinUninterruptiblyInterrupts() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
FibAction f;
final Thread currentThread = Thread.currentThread();
// test quietlyJoin()
f = new FibAction(8);
assertSame(f, f.fork());
currentThread.interrupt();
f.quietlyJoinUninterruptibly(LONG_DELAY_MS, MILLISECONDS);
Thread.interrupted();
assertEquals(21, f.result);
checkCompletedNormally(f);
f = new FibAction(8);
f.cancel(true);
assertSame(f, f.fork());
currentThread.interrupt();
f.quietlyJoinUninterruptibly(LONG_DELAY_MS, MILLISECONDS);
Thread.interrupted();
checkCancelled(f);
f = new FibAction(8);
f.completeExceptionally(new FJException());
assertSame(f, f.fork());
currentThread.interrupt();
f.quietlyJoinUninterruptibly(LONG_DELAY_MS, MILLISECONDS);
Thread.interrupted();
checkCompletedAbnormally(f, f.getException());
}};
checkInvoke(a);
a.reinitialize();
checkInvoke(a);
}
/**
* timed quietlyJoin throws IE in the presence of interrupts
*/
public void testTimedQuietlyJoinInterrupts() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
FibAction f;
final Thread currentThread = Thread.currentThread();
f = new FibAction(8);
assertSame(f, f.fork());
currentThread.interrupt();
try {
f.quietlyJoin(LONG_DELAY_MS, MILLISECONDS);
} catch (InterruptedException success) {
}
Thread.interrupted();
f.quietlyJoin();
f = new FibAction(8);
f.cancel(true);
assertSame(f, f.fork());
currentThread.interrupt();
try {
f.quietlyJoin(LONG_DELAY_MS, MILLISECONDS);
} catch (InterruptedException success) {
}
f.quietlyJoin();
checkCancelled(f);
}};
checkInvoke(a);
a.reinitialize();
checkInvoke(a);
}
/**
* timed quietlyJoin of a forked task returns when task completes
*/
public void testForkTimedQuietlyJoin() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
FibAction f = new FibAction(8);
assertSame(f, f.fork());
assertTrue(f.quietlyJoin(LONG_DELAY_MS, MILLISECONDS));
assertEquals(21, f.result);
checkCompletedNormally(f);
}};
checkInvoke(a);
}
/**
* timed quietlyJoin with null time unit throws NPE
*/
public void testForkTimedQuietlyJoinNPE() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
FibAction f = new FibAction(8);
assertSame(f, f.fork());
try {
f.quietlyJoin(randomTimeout(), null);
shouldThrow();
} catch (NullPointerException success) {}
}};
checkInvoke(a);
}
/**
* quietlyInvoke task returns when task completes abnormally
*/
public void testAbnormalTimedQuietlyJoin() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
FailingFibAction f = new FailingFibAction(8);
assertSame(f, f.fork());
assertTrue(f.quietlyJoin(LONG_DELAY_MS, MILLISECONDS));
assertTrue(f.getException() instanceof FJException);
checkCompletedAbnormally(f, f.getException());
}};
checkInvoke(a);
}
/**
* timed quietlyJoinUninterruptibly of a forked task returns when task completes
*/
public void testForkTimedQuietlyJoinUninterruptibly() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
FibAction f = new FibAction(8);
assertSame(f, f.fork());
assertTrue(f.quietlyJoinUninterruptibly(LONG_DELAY_MS, MILLISECONDS));
assertEquals(21, f.result);
checkCompletedNormally(f);
}};
checkInvoke(a);
}
/**
* timed quietlyJoinUninterruptibly with null time unit throws NPE
*/
public void testForkTimedQuietlyJoinUninterruptiblyNPE() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
FibAction f = new FibAction(8);
assertSame(f, f.fork());
try {
f.quietlyJoinUninterruptibly(randomTimeout(), null);
shouldThrow();
} catch (NullPointerException success) {}
}};
checkInvoke(a);
}
/**
* quietlyInvoke task returns when task completes abnormally
*/
public void testAbnormalTimedQuietlyJoinUninterruptibly() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
FailingFibAction f = new FailingFibAction(8);
assertSame(f, f.fork());
assertTrue(f.quietlyJoinUninterruptibly(LONG_DELAY_MS, MILLISECONDS));
assertTrue(f.getException() instanceof FJException);
checkCompletedAbnormally(f, f.getException());
}};
checkInvoke(a);
}
/**
* adaptInterruptible(callable).toString() contains toString of wrapped task
*/
public void testAdaptInterruptible_Callable_toString() {
if (testImplementationDetails) {
Callable<String> c = () -> "";
ForkJoinTask<String> task = ForkJoinTask.adaptInterruptible(c);
assertEquals(
identityString(task) + "[Wrapped task = " + c.toString() + "]",
task.toString());
}
}
}

43
test/jdk/java/util/concurrent/tck/JSR166TestCase.java
View file

@ -225,8 +225,9 @@ import junit.framework.TestSuite;
* </ul>
*/
public class JSR166TestCase extends TestCase {
// No longer run with custom securityManagers
private static final boolean useSecurityManager =
Boolean.getBoolean("jsr166.useSecurityManager");
Boolean.getBoolean("jsr166.useSecurityManager");
protected static final boolean expensiveTests =
Boolean.getBoolean("jsr166.expensiveTests");
@ -437,11 +438,15 @@ public class JSR166TestCase extends TestCase {
* Runs all unit tests in the given test suite.
* Actual behavior influenced by jsr166.* system properties.
*/
@SuppressWarnings("removal")
static void main(Test suite, String[] args) {
if (useSecurityManager) {
System.err.println("Setting a permissive security manager");
Policy.setPolicy(permissivePolicy());
System.setSecurityManager(new SecurityManager());
try {
System.setSecurityManager(new SecurityManager());
} catch(Throwable ok) { // failure OK during deprecation
}
}
for (int i = 0; i < suiteRuns; i++) {
TestResult result = newPithyTestRunner().doRun(suite);
@ -482,14 +487,18 @@ public class JSR166TestCase extends TestCase {
public static final String JAVA_SPECIFICATION_VERSION;
static {
try {
JAVA_CLASS_VERSION = java.security.AccessController.doPrivileged(
@SuppressWarnings("removal") double jcv =
java.security.AccessController.doPrivileged(
new java.security.PrivilegedAction<Double>() {
public Double run() {
return Double.valueOf(System.getProperty("java.class.version"));}});
JAVA_SPECIFICATION_VERSION = java.security.AccessController.doPrivileged(
JAVA_CLASS_VERSION = jcv;
@SuppressWarnings("removal") String jsv =
java.security.AccessController.doPrivileged(
new java.security.PrivilegedAction<String>() {
public String run() {
return System.getProperty("java.specification.version");}});
JAVA_SPECIFICATION_VERSION = jsv;
} catch (Throwable t) {
throw new Error(t);
}
@ -626,6 +635,12 @@ public class JSR166TestCase extends TestCase {
addNamedTestClasses(suite, java9TestClassNames);
}
if (atLeastJava17()) {
String[] java17TestClassNames = {
"ForkJoinPool19Test",
};
addNamedTestClasses(suite, java17TestClassNames);
}
return suite;
}
@ -1084,7 +1099,7 @@ public class JSR166TestCase extends TestCase {
void joinPool(ExecutorService pool) {
try {
pool.shutdown();
if (!pool.awaitTermination(2 * LONG_DELAY_MS, MILLISECONDS)) {
if (!pool.awaitTermination(20 * LONG_DELAY_MS, MILLISECONDS)) {
try {
threadFail("ExecutorService " + pool +
" did not terminate in a timely manner");
@ -1168,6 +1183,7 @@ public class JSR166TestCase extends TestCase {
* A debugging tool to print stack traces of most threads, as jstack does.
* Uninteresting threads are filtered out.
*/
@SuppressWarnings("removal")
static void dumpTestThreads() {
SecurityManager sm = System.getSecurityManager();
if (sm != null) {
@ -1376,9 +1392,7 @@ public class JSR166TestCase extends TestCase {
assertTrue(c.remove(i));
}
static void mustNotRemove(Collection<Item> c, int i) {
Item[] items = defaultItems;
Item x = (i >= 0 && i < items.length) ? items[i] : new Item(i);
assertFalse(c.remove(x));
assertFalse(c.remove(itemFor(i)));
}
static void mustNotRemove(Collection<Item> c, Item i) {
assertFalse(c.remove(i));
@ -1403,6 +1417,7 @@ public class JSR166TestCase extends TestCase {
* security manager. We require that any security manager permit
* getPolicy/setPolicy.
*/
@SuppressWarnings("removal")
public void runWithPermissions(Runnable r, Permission... permissions) {
SecurityManager sm = System.getSecurityManager();
if (sm == null) {
@ -1418,8 +1433,10 @@ public class JSR166TestCase extends TestCase {
* Runnable. We require that any security manager permit
* getPolicy/setPolicy.
*/
@SuppressWarnings("removal")
public void runWithSecurityManagerWithPermissions(Runnable r,
Permission... permissions) {
if (!useSecurityManager) return;
SecurityManager sm = System.getSecurityManager();
if (sm == null) {
Policy savedPolicy = Policy.getPolicy();
@ -1427,9 +1444,13 @@ public class JSR166TestCase extends TestCase {
Policy.setPolicy(permissivePolicy());
System.setSecurityManager(new SecurityManager());
runWithSecurityManagerWithPermissions(r, permissions);
} catch (UnsupportedOperationException ok) {
} finally {
System.setSecurityManager(null);
Policy.setPolicy(savedPolicy);
try {
System.setSecurityManager(null);
Policy.setPolicy(savedPolicy);
} catch (Exception ok) {
}
}
} else {
Policy savedPolicy = Policy.getPolicy();
@ -1456,6 +1477,7 @@ public class JSR166TestCase extends TestCase {
* A security policy where new permissions can be dynamically added
* or all cleared.
*/
@SuppressWarnings("removal")
public static class AdjustablePolicy extends java.security.Policy {
Permissions perms = new Permissions();
AdjustablePolicy(Permission... permissions) {
@ -1485,6 +1507,7 @@ public class JSR166TestCase extends TestCase {
/**
* Returns a policy containing all the permissions we ever need.
*/
@SuppressWarnings("removal")
public static Policy permissivePolicy() {
return new AdjustablePolicy
// Permissions j.u.c. needs directly