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

8246585: ForkJoin updates

Browse source 
8229253: forkjoin/FJExceptionTableLeak.java fails "AssertionError: failed to satisfy condition"

Reviewed-by: dl
This commit is contained in:
Martin Buchholz 2021-01-09 20:57:52 +00:00
parent 6472104e18
commit 5cfa8c94d6
9 changed files with 2716 additions and 2219 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -357,7 +357,7 @@ import java.lang.invoke.VarHandle;
* within this method to ensure thread safety of accesses to fields of
* this task or other completed tasks.
*
* <p><b>Completion Traversals</b>. If using {@code onCompletion} to
* <p><b>Completion Traversals.</b> If using {@code onCompletion} to
* process completions is inapplicable or inconvenient, you can use
* methods {@link #firstComplete} and {@link #nextComplete} to create
* custom traversals. For example, to define a MapReducer that only
@ -553,6 +553,11 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
return PENDING.compareAndSet(this, expected, count);
}
// internal-only weak version
final boolean weakCompareAndSetPendingCount(int expected, int count) {
return PENDING.weakCompareAndSet(this, expected, count);
}
/**
* If the pending count is nonzero, (atomically) decrements it.
*
@ -562,7 +567,7 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
public final int decrementPendingCountUnlessZero() {
int c;
do {} while ((c = pending) != 0 &&
!PENDING.weakCompareAndSet(this, c, c - 1));
!weakCompareAndSetPendingCount(c, c - 1));
return c;
}
@ -595,7 +600,7 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
return;
}
}
else if (PENDING.weakCompareAndSet(a, c, c - 1))
else if (a.weakCompareAndSetPendingCount(c, c - 1))
return;
}
}
@ -618,7 +623,7 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
return;
}
}
else if (PENDING.weakCompareAndSet(a, c, c - 1))
else if (a.weakCompareAndSetPendingCount(c, c - 1))
return;
}
}
@ -663,7 +668,7 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
for (int c;;) {
if ((c = pending) == 0)
return this;
else if (PENDING.weakCompareAndSet(this, c, c - 1))
else if (weakCompareAndSetPendingCount(c, c - 1))
return null;
}
}
@ -718,30 +723,33 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
* processed.
*/
public final void helpComplete(int maxTasks) {
Thread t; ForkJoinWorkerThread wt;
if (maxTasks > 0 && status >= 0) {
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
(wt = (ForkJoinWorkerThread)t).pool.
helpComplete(wt.workQueue, this, maxTasks);
else
ForkJoinPool.common.externalHelpComplete(this, maxTasks);
}
ForkJoinPool.WorkQueue q; Thread t; boolean owned;
if (owned = (t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
q = ((ForkJoinWorkerThread)t).workQueue;
else
q = ForkJoinPool.commonQueue();
if (q != null && maxTasks > 0)
q.helpComplete(this, owned, maxTasks);
}
// ForkJoinTask overrides
/**
* Supports ForkJoinTask exception propagation.
*/
void internalPropagateException(Throwable ex) {
CountedCompleter<?> a = this, s = a;
while (a.onExceptionalCompletion(ex, s) &&
(a = (s = a).completer) != null && a.status >= 0 &&
isExceptionalStatus(a.recordExceptionalCompletion(ex)))
;
@Override
final int trySetException(Throwable ex) {
CountedCompleter<?> a = this, p = a;
do {} while (isExceptionalStatus(a.trySetThrown(ex)) &&
a.onExceptionalCompletion(ex, p) &&
(a = (p = a).completer) != null && a.status >= 0);
return status;
}
/**
* Implements execution conventions for CountedCompleters.
*/
@Override
protected final boolean exec() {
compute();
return false;
@ -756,6 +764,7 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
*
* @return the result of the computation
*/
@Override
public T getRawResult() { return null; }
/**
@ -765,6 +774,7 @@ public abstract class CountedCompleter<T> extends ForkJoinTask<T> {
* overridden to update existing objects or fields, then it must
* in general be defined to be thread-safe.
*/
@Override
protected void setRawResult(T t) { }
// VarHandle mechanics

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

File diff suppressed because it is too large Load diff

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

File diff suppressed because it is too large Load diff

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

@ -35,10 +35,8 @@
package java.util.concurrent;
import java.security.AccessControlContext;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.security.ProtectionDomain;
/**
* A thread managed by a {@link ForkJoinPool}, which executes
@ -62,26 +60,38 @@ public class ForkJoinWorkerThread extends Thread {
* ForkJoinTasks. For explanation, see the internal documentation
* of class ForkJoinPool.
*
* This class just maintains links to its pool and WorkQueue. The
* pool field is set immediately upon construction, but the
* workQueue field is not set until a call to registerWorker
* completes. This leads to a visibility race, that is tolerated
* by requiring that the workQueue field is only accessed by the
* owning thread.
*
* Support for (non-public) subclass InnocuousForkJoinWorkerThread
* requires that we break quite a lot of encapsulation (via helper
* methods in ThreadLocalRandom) both here and in the subclass to
* access and set Thread fields.
* This class just maintains links to its pool and WorkQueue.
*/
final ForkJoinPool pool; // the pool this thread works in
final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics
/** An AccessControlContext supporting no privileges */
private static final AccessControlContext INNOCUOUS_ACC =
new AccessControlContext(
new ProtectionDomain[] { new ProtectionDomain(null, null) });
/**
* Full nonpublic constructor.
*/
ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool,
boolean useSystemClassLoader, boolean isInnocuous) {
super(group, null, pool.nextWorkerThreadName(), 0L);
UncaughtExceptionHandler handler = (this.pool = pool).ueh;
this.workQueue = new ForkJoinPool.WorkQueue(this, isInnocuous);
super.setDaemon(true);
if (handler != null)
super.setUncaughtExceptionHandler(handler);
if (useSystemClassLoader)
super.setContextClassLoader(ClassLoader.getSystemClassLoader());
}
/**
* Creates a ForkJoinWorkerThread operating in the given thread group and
* pool.
*
* @param group if non-null, the thread group for this thread
* @param pool the pool this thread works in
* @throws NullPointerException if pool is null
*/
/* TODO: protected */ ForkJoinWorkerThread(ThreadGroup group, ForkJoinPool pool) {
this(group, pool, false, false);
}
/**
* Creates a ForkJoinWorkerThread operating in the given pool.
@ -90,38 +100,7 @@ public class ForkJoinWorkerThread extends Thread {
* @throws NullPointerException if pool is null
*/
protected ForkJoinWorkerThread(ForkJoinPool pool) {
// Use a placeholder until a useful name can be set in registerWorker
super("aForkJoinWorkerThread");
this.pool = pool;
this.workQueue = pool.registerWorker(this);
}
/**
* Version for use by the default pool. Supports setting the
* context class loader. This is a separate constructor to avoid
* affecting the protected constructor.
*/
ForkJoinWorkerThread(ForkJoinPool pool, ClassLoader ccl) {
super("aForkJoinWorkerThread");
super.setContextClassLoader(ccl);
ThreadLocalRandom.setInheritedAccessControlContext(this, INNOCUOUS_ACC);
this.pool = pool;
this.workQueue = pool.registerWorker(this);
}
/**
* Version for InnocuousForkJoinWorkerThread.
*/
ForkJoinWorkerThread(ForkJoinPool pool,
ClassLoader ccl,
ThreadGroup threadGroup,
AccessControlContext acc) {
super(threadGroup, null, "aForkJoinWorkerThread");
super.setContextClassLoader(ccl);
ThreadLocalRandom.setInheritedAccessControlContext(this, acc);
ThreadLocalRandom.eraseThreadLocals(this); // clear before registering
this.pool = pool;
this.workQueue = pool.registerWorker(this);
this(null, pool, false, false);
}
/**
@ -176,11 +155,14 @@ public class ForkJoinWorkerThread extends Thread {
* {@link ForkJoinTask}s.
*/
public void run() {
if (workQueue.array == null) { // only run once
Throwable exception = null;
Throwable exception = null;
ForkJoinPool p = pool;
ForkJoinPool.WorkQueue w = workQueue;
if (p != null && w != null) { // skip on failed initialization
try {
p.registerWorker(w);
onStart();
pool.runWorker(workQueue);
p.runWorker(w);
} catch (Throwable ex) {
exception = ex;
} finally {
@ -190,18 +172,12 @@ public class ForkJoinWorkerThread extends Thread {
if (exception == null)
exception = ex;
} finally {
pool.deregisterWorker(this, exception);
p.deregisterWorker(this, exception);
}
}
}
}
/**
* Non-public hook method for InnocuousForkJoinWorkerThread.
*/
void afterTopLevelExec() {
}
/**
* A worker thread that has no permissions, is not a member of any
* user-defined ThreadGroup, uses the system class loader as
@ -221,15 +197,7 @@ public class ForkJoinWorkerThread extends Thread {
}});
InnocuousForkJoinWorkerThread(ForkJoinPool pool) {
super(pool,
ClassLoader.getSystemClassLoader(),
innocuousThreadGroup,
INNOCUOUS_ACC);
}
@Override // to erase ThreadLocals
void afterTopLevelExec() {
ThreadLocalRandom.eraseThreadLocals(this);
super(innocuousThreadGroup, pool, true, true);
}
@Override // to silently fail