8319447: Improve performance of delayed task handling

src/java.base/share/classes/java/util/concurrent/ForkJoinPool.java

@@ -387,18 +387,18 @@ public class ForkJoinPool extends AbstractExecutorService
      * WorkQueues are also used in a similar way for tasks submitted
      * to the pool. We cannot mix these tasks in the same queues used
      * by workers. Instead, we randomly associate submission queues
-     * with submitting threads, using a form of hashing.  The
-     * ThreadLocalRandom probe value serves as a hash code for
-     * choosing existing queues, and may be randomly repositioned upon
-     * contention with other submitters.  In essence, submitters act
-     * like workers except that they are restricted to executing local
-     * tasks that they submitted (or when known, subtasks thereof).
-     * Insertion of tasks in shared mode requires a lock. We use only
-     * a simple spinlock (as one role of field "phase") because
-     * submitters encountering a busy queue move to a different
-     * position to use or create other queues.  They (spin) block when
-     * registering new queues, or indirectly elsewhere, by revisiting
-     * later.
+     * with submitting threads (or carriers when using VirtualThreads)
+     * using a form of hashing.  The ThreadLocalRandom probe value
+     * serves as a hash code for choosing existing queues, and may be
+     * randomly repositioned upon contention with other submitters.
+     * In essence, submitters act like workers except that they are
+     * restricted to executing local tasks that they submitted (or
+     * when known, subtasks thereof).  Insertion of tasks in shared
+     * mode requires a lock. We use only a simple spinlock (as one
+     * role of field "phase") because submitters encountering a busy
+     * queue move to a different position to use or create other
+     * queues.  They (spin) block when registering new queues, or
+     * indirectly elsewhere, by revisiting later.
      *
      * Management
      * ==========
@@ -885,7 +885,8 @@ public class ForkJoinPool extends AbstractExecutorService
      * To comply with ExecutorService specs, we use subclasses of
      * abstract class InterruptibleTask for tasks that require
      * stronger interruption and cancellation guarantees.  External
-     * submitters never run these tasks, even if in the common pool.
+     * submitters never run these tasks, even if in the common pool
+     * (as indicated by ForkJoinTask.noUserHelp status bit).
      * InterruptibleTasks include a "runner" field (implemented
      * similarly to FutureTask) to support cancel(true).  Upon pool
      * shutdown, runners are interrupted so they can cancel. Since
@@ -1295,10 +1296,9 @@ final void push(ForkJoinTask<?> task, ForkJoinPool pool,
                     unlockPhase();
                 if (room < 0)
                     throw new RejectedExecutionException("Queue capacity exceeded");
-                if ((room == 0 ||        // pad for InterruptibleTasks
+                if ((room == 0 ||        // pad if no caller-run
                      a[m & (s - ((internal || task == null ||
-                                  task.getClass().getSuperclass() !=
-                                  interruptibleTaskClass) ? 1 : 2))] == null) &&
+                                  task.noUserHelp() == 0) ? 1 : 2))] == null) &&
                     pool != null)
                     pool.signalWork();   // may have appeared empty
             }
@@ -1637,11 +1637,6 @@ final boolean isApparentlyUnblocked() {
      */
     static volatile RuntimePermission modifyThreadPermission;
 
-    /**
-     * Cached for faster type tests.
-     */
-    static final Class<?> interruptibleTaskClass;
-
     /**
      * For VirtualThread intrinsics
      */
@@ -2023,12 +2018,11 @@ final void runWorker(WorkQueue w) {
                                 boolean propagate;
                                 int nb = q.base = b + 1;
                                 w.nsteals = ++nsteals;
+                                int ts = t.status;
                                 w.source = j;             // volatile
                                 rescan = true;
                                 if (propagate =
-                                    ((src != (src = j) ||
-                                      t.getClass().getSuperclass() ==
-                                      interruptibleTaskClass) &&
+                                    ((src != (src = j) || t.noUserHelp() != 0) &&
                                      a[nb & m] != null))
                                     signalWork();
                                 w.topLevelExec(t, fifo);
@@ -4406,7 +4400,6 @@ protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
         if ((scale & (scale - 1)) != 0)
             throw new Error("array index scale not a power of two");
 
-        interruptibleTaskClass = ForkJoinTask.InterruptibleTask.class;
         Class<?> dep = LockSupport.class; // ensure loaded
         // allow access to non-public methods
         JLA = SharedSecrets.getJavaLangAccess();

src/java.base/share/classes/java/util/concurrent/ForkJoinTask.java

@@ -273,6 +273,7 @@ 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 MARKER         = 1 << 30; // utility marker
     static final int SMASK          = 0xffff;  // short bits for tags
     static final int UNCOMPENSATE   = 1 << 16; // helpJoin sentinel
@@ -292,6 +293,12 @@ 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 void setNoUserHelp() { // for use in constructors only
+        U.putInt(this, STATUS, NO_USER_HELP);
+    }
 
     // Support for waiting and signalling
 
@@ -476,7 +483,7 @@ else if (casAux(a, next))
      */
     private int awaitDone(boolean interruptible, long deadline) {
         ForkJoinWorkerThread wt; ForkJoinPool p; ForkJoinPool.WorkQueue q;
-        Thread t; boolean internal; int s;
+        Thread t; boolean internal; int s, ss;
         if (internal =
             (t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
             p = (wt = (ForkJoinWorkerThread)t).pool;
@@ -487,7 +494,7 @@ private int awaitDone(boolean interruptible, long deadline) {
         return (((s = (p == null) ? 0 :
                   ((this instanceof CountedCompleter) ?
                    p.helpComplete(this, q, internal) :
-                   (this instanceof InterruptibleTask) && !internal ? status :
+                   !internal && ((ss = status) & NO_USER_HELP) != 0 ? ss :
                    p.helpJoin(this, q, internal))) < 0)) ? s :
             awaitDone(internal ? p : null, s, interruptible, deadline);
     }
@@ -1155,7 +1162,7 @@ public static void helpQuiesce() {
      */
     public void reinitialize() {
         aux = null;
-        status = 0;
+        status &= NO_USER_HELP;
     }
 
     /**
@@ -1634,6 +1641,9 @@ public String toString() {
     abstract static class InterruptibleTask<T> extends ForkJoinTask<T>
         implements RunnableFuture<T> {
         transient volatile Thread runner;
+        InterruptibleTask() {
+            setNoUserHelp();
+        }
         abstract T compute() throws Exception;
         public final boolean exec() {
             Thread.interrupted();