Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

8319447: Improve performance of delayed task handling #23702

Open
wants to merge 47 commits into
base: master
Choose a base branch
from
Open
Changes from 1 commit
Commits
Show all changes
47 commits
Select commit Hold shift + click to select a range
cb1aedf
In-progress snapshot
DougLea Jan 6, 2025
5c4ca21
Better conform to STPE
DougLea Jan 7, 2025
bb0e6a6
Refactorings
DougLea Jan 9, 2025
f683d7f
Use pendingRemval queue
DougLea Jan 9, 2025
3801ba0
Reduce unparks
DougLea Jan 10, 2025
d630b40
Better pending queues
DougLea Jan 16, 2025
0dd4ab7
Merge branch 'openjdk:master' into JDK-8319447
DougLea Jan 16, 2025
7e04af5
Conform to default STPE policies
DougLea Jan 19, 2025
7288e32
Comment out racy test
DougLea Jan 19, 2025
b14e31c
Merge branch 'openjdk:master' into JDK-8319447
DougLea Jan 19, 2025
cb202b6
Reduce memory contention
DougLea Jan 21, 2025
1f0a5cf
Use nanoTimeOrigin
DougLea Jan 21, 2025
97a2920
Reduce nanoTime usage; extend tck tests
DougLea Jan 23, 2025
f9aa135
Simplify scheduler state tracking
DougLea Jan 23, 2025
798fe64
Refactor delay scheduler pool submissions
DougLea Jan 24, 2025
d083e91
improve removal cost balance
DougLea Jan 25, 2025
1a7f77c
Ensure negative nanotime offset
DougLea Jan 26, 2025
f832335
Merge branch 'openjdk:master' into JDK-8319447
DougLea Feb 2, 2025
3da4fd7
Solidify design; add documentation
DougLea Feb 2, 2025
49b1699
Separate out DelayScheduler.java
DougLea Feb 3, 2025
229da14
Rework FJP-DS connections
DougLea Feb 4, 2025
9fad8d4
Deal with commonPool parallelism zero; use in other juc classes; remo…
DougLea Feb 7, 2025
2e60dc9
Refactor schedule methods
DougLea Feb 8, 2025
a0db427
Isolate screening
DougLea Feb 8, 2025
d0f4af1
Support STPE policy methods
DougLea Feb 9, 2025
a6290ab
Reduce memory accesses
DougLea Feb 9, 2025
c839299
Simplify policy methods; improve layout
DougLea Feb 10, 2025
f1394c4
Rename DelayedTask to ScheduledForkJoinTask; misc other improvements
DougLea Feb 12, 2025
0e13955
Better accommodate CompletableFuture; use 4-ary heap; add javadocs; o…
DougLea Feb 15, 2025
14a7a6f
Reduce garbage retention; use trailing padding; add tests
DougLea Feb 16, 2025
bd58f41
Merge branch 'openjdk:master' into JDK-8319447
DougLea Feb 16, 2025
93aac79
Merge remote-tracking branch 'refs/remotes/origin/JDK-8319447' into J…
DougLea Feb 16, 2025
753d0e0
Add optional SubmitWithTimeout action
DougLea Feb 17, 2025
53516e9
Misc minor improvements and renamings for clarity
DougLea Feb 19, 2025
16815cc
Address feedback
DougLea Feb 21, 2025
84eaab0
Address review comments; ensure new internal methods can't clash with…
DougLea Feb 22, 2025
c9bc41a
Standardize parameter checking
DougLea Feb 23, 2025
b40513f
Address review comments; reactivation tweak
DougLea Feb 28, 2025
0c5d22a
Reduce volatile reads
DougLea Mar 1, 2025
5c0355b
Associate probes with carriers if Virtual (no doc updates yet)
DougLea Mar 8, 2025
f670910
Merge branch 'openjdk:master' into JDK-8319447
DougLea Mar 8, 2025
6fe1a3b
Disambiguate caller-runs vs Interruptible
DougLea Mar 9, 2025
9cc670b
Use SharedSecrets for ThreadLocalRandomProbe; other tweaks
DougLea Mar 11, 2025
172a235
Reword javadoc
DougLea Mar 13, 2025
9b51b7a
Use TC_MASK in accord with https://bugs.openjdk.org/browse/JDK-833001…
DougLea Mar 14, 2025
24422e4
Match indent of naster changes
DougLea Mar 22, 2025
b552c22
Merge branch 'openjdk:master' into JDK-8319447
DougLea Mar 22, 2025
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Use SharedSecrets for ThreadLocalRandomProbe; other tweaks
DougLea committed Mar 11, 2025
commit 9cc670bc2cdf35c1055d6f469d4210d27d26bffc
25 changes: 12 additions & 13 deletions src/java.base/share/classes/java/util/concurrent/ForkJoinPool.java
Original file line number Diff line number Diff line change
@@ -1273,15 +1273,17 @@ final int queueSize() {
/**
* Pushes a task. Called only by owner or if already locked
*
* @param task the task. Caller must ensure non-null.
* @param task the task; no-op if null
* @param pool the pool to signal if was previously empty, else null
* @param internal if caller owns this queue
* @throws RejectedExecutionException if array could not be resized
*/
final void push(ForkJoinTask<?> task, ForkJoinPool pool,
boolean internal) {
int s = top, b = base, m, cap, room; ForkJoinTask<?>[] a;
if ((a = array) != null && (cap = a.length) > 0) { // else disabled
if ((a = array) != null && (cap = a.length) > 0 && // else disabled
task != null) {
int pk = task.noUserHelp() + 1; // prev slot offset
if ((room = (m = cap - 1) - (s - b)) >= 0) {
top = s + 1;
long pos = slotOffset(m & s);
@@ -1296,9 +1298,7 @@ final void push(ForkJoinTask<?> task, ForkJoinPool pool,
unlockPhase();
if (room < 0)
throw new RejectedExecutionException("Queue capacity exceeded");
if ((room == 0 || // pad if no caller-run
a[m & (s - ((internal || task == null ||
task.noUserHelp() == 0) ? 1 : 2))] == null) &&
if ((room == 0 || a[m & (s - pk)] == null) &&
pool != null)
pool.signalWork(); // may have appeared empty
}
@@ -2016,14 +2016,13 @@ final void runWorker(WorkQueue w) {
}
else {
boolean propagate;
int nb = q.base = b + 1;
int nb = q.base = b + 1, prevSrc = src;
w.nsteals = ++nsteals;
int ts = t.status;
w.source = j; // volatile
w.source = src = j; // volatile
rescan = true;
int nh = t.noUserHelp();
if (propagate =
((src != (src = j) || t.noUserHelp() != 0) &&
a[nb & m] != null))
(prevSrc != src || nh != 0) && a[nb & m] != null)
signalWork();
w.topLevelExec(t, fifo);
if ((b = q.base) != nb && !propagate)
@@ -2062,8 +2061,8 @@ private int deactivate(WorkQueue w, int phase) {
((e & SHUTDOWN) != 0L && ac == 0 && quiescent() > 0) ||
(qs = queues) == null || (n = qs.length) <= 0)
return IDLE; // terminating
int prechecks = 3; // reactivation threshold
for (int k = Math.max(n << 2, SPIN_WAITS << 1);;) {
int k = Math.max(n << 2, SPIN_WAITS << 1);
for (int prechecks = k / n;;) { // reactivation threshold
WorkQueue q; int cap; ForkJoinTask<?>[] a; long c;
if (w.phase == activePhase)
return activePhase;
@@ -2072,7 +2071,7 @@ private int deactivate(WorkQueue w, int phase) {
if ((q = qs[k & (n - 1)]) == null)
Thread.onSpinWait();
else if ((a = q.array) != null && (cap = a.length) > 0 &&
a[q.base & (cap - 1)] != null && --prechecks < 0 &&
a[q.base & (cap - 1)] != null && --prechecks <= 0 &&
(int)(c = ctl) == activePhase &&
compareAndSetCtl(c, (sp & LMASK) | ((c + RC_UNIT) & UMASK)))
return w.phase = activePhase; // reactivate
Original file line number Diff line number Diff line change
@@ -273,7 +273,8 @@ final boolean casNext(Aux c, Aux v) { // used only in cancellation
static final int ABNORMAL = 1 << 16;
static final int THROWN = 1 << 17;
static final int HAVE_EXCEPTION = DONE | ABNORMAL | THROWN;
static final int NO_USER_HELP = 1 << 24; // no external caller-run helping
static final int NUH_BIT = 24; // no external caller helping
static final int NO_USER_HELP = 1 << NUH_BIT;
static final int MARKER = 1 << 30; // utility marker
static final int SMASK = 0xffff; // short bits for tags
static final int UNCOMPENSATE = 1 << 16; // helpJoin sentinel
@@ -293,8 +294,8 @@ private int getAndBitwiseOrStatus(int v) {
private boolean casStatus(int c, int v) {
return U.compareAndSetInt(this, STATUS, c, v);
}
final int noUserHelp() { // nonvolatile read
return U.getInt(this, STATUS) & NO_USER_HELP;
final int noUserHelp() { // nonvolatile read; return 0 or 1
return (U.getInt(this, STATUS) & NO_USER_HELP) >>> NUH_BIT;
}
final void setNoUserHelp() { // for use in constructors only
U.putInt(this, STATUS, NO_USER_HELP);
Original file line number Diff line number Diff line change
@@ -242,16 +242,18 @@ protected int next(int bits) {
* the classes that use them. Briefly, a thread's "probe" value is
* a non-zero hash code that (probably) does not collide with
* other existing threads with respect to any power of two
* collision space. When it does collide, it is pseudo-randomly
* adjusted (using a Marsaglia XorShift). The nextSecondarySeed
* method is used in the same contexts as ThreadLocalRandom, but
* only for transient usages such as random adaptive spin/block
* sequences for which a cheap RNG suffices and for which it could
* in principle disrupt user-visible statistical properties of the
* main ThreadLocalRandom if we were to use it.
* collision space, based on carrier threads in the case of
* VirtualThreads to reduce the expected collision rate. When it
* does collide, it is pseudo-randomly adjusted (using a Marsaglia
* XorShift). The nextSecondarySeed method is used in the same
* contexts as ThreadLocalRandom, but only for transient usages
* such as random adaptive spin/block sequences for which a cheap
* RNG suffices and for which it could in principle disrupt
* user-visible statistical properties of the main
* ThreadLocalRandom if we were to use it.
*
* Note: Because of package-protection issues, versions of some
* these methods also appear in some subpackage classes.
* Note: jdk SharedSecrets are used enable use in jdk classes
* outside this package.
*/

/**
@@ -397,6 +399,12 @@ private static class Access {
public int nextSecondaryThreadLocalRandomSeed() {
return nextSecondarySeed();
}
public int getThreadLocalRandomProbe() {
return getProbe();
}
public int advanceThreadLocalRandomProbe(int r) {
return advanceProbe(r);
}
}
);
}
Original file line number Diff line number Diff line change
@@ -43,6 +43,8 @@
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.DoubleBinaryOperator;
import java.util.function.LongBinaryOperator;
import jdk.internal.access.SharedSecrets;
import jdk.internal.access.JavaUtilConcurrentTLRAccess;

/**
* A package-local class holding common representation and mechanics
@@ -188,24 +190,18 @@ final boolean casCellsBusy() {
}

/**
* Returns the probe value for the current thread.
* Duplicated from ThreadLocalRandom because of packaging restrictions.
* Returns the ThreadLocalRandom probe value for the current thread.
*/
static final int getProbe() {
return (int) THREAD_PROBE.get(Thread.currentThread());
return TLR.getThreadLocalRandomProbe();
}

/**
* Pseudo-randomly advances and records the given probe value for the
* given thread.
* Duplicated from ThreadLocalRandom because of packaging restrictions.
*/
static final int advanceProbe(int probe) {
probe ^= probe << 13; // xorshift
probe ^= probe >>> 17;
probe ^= probe << 5;
THREAD_PROBE.set(Thread.currentThread(), probe);
return probe;
return TLR.advanceThreadLocalRandomProbe(probe);
}

/**
@@ -371,21 +367,17 @@ else if (casBase(v = base, apply(fn, v, x)))
}
}

private static final JavaUtilConcurrentTLRAccess TLR =
SharedSecrets.getJavaUtilConcurrentTLRAccess();

// VarHandle mechanics
private static final VarHandle BASE;
private static final VarHandle CELLSBUSY;
private static final VarHandle THREAD_PROBE;
static {
MethodHandles.Lookup l1 = MethodHandles.lookup();

BASE = MhUtil.findVarHandle(l1, "base", long.class);
CELLSBUSY = MhUtil.findVarHandle(l1, "cellsBusy", int.class);
try {
MethodHandles.Lookup l2 = MethodHandles.privateLookupIn(Thread.class, l1);
THREAD_PROBE = MhUtil.findVarHandle(l2, "threadLocalRandomProbe", int.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
}

}
Original file line number Diff line number Diff line change
@@ -27,4 +27,6 @@

public interface JavaUtilConcurrentTLRAccess {
int nextSecondaryThreadLocalRandomSeed();
int getThreadLocalRandomProbe();
int advanceThreadLocalRandomProbe(int r);
}