Rework FJP-DS connections

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

@@ -54,8 +54,10 @@ final class DelayScheduler extends Thread {
      *
      * To reduce memory contention, the heap is maintained solely via
      * local variables in method loop() (forcing noticeable code
-     * sprawl), recording only the heap array to allow method
-     * canShutDown to conservatively check emptiness.
+     * sprawl), recording only the current heap size when blocked to
+     * allow method canShutDown to conservatively check emptiness, and
+     * to support an approximate reporting of current size for
+     * monitoring.
      *
      * The pending queue uses a design similar to ForkJoinTask.Aux
      * queues: Incoming requests prepend (Treiber-stack-style) to the
@@ -104,7 +106,7 @@ final class DelayScheduler extends Thread {
 
     private static final int INITIAL_HEAP_CAPACITY = 1 << 6;
     private final ForkJoinPool pool; // read only once
-    private DelayedTask<?>[] heap;   // written only when (re)allocated
+    int restingSize;                 // written only before parking
     private volatile int active;     // 0: inactive, -1: stopped, +1: running
     @jdk.internal.vm.annotation.Contended()
     private volatile DelayedTask<?> pending;
@@ -164,24 +166,28 @@ final void pend(DelayedTask<?> task) {
      * Returns true if (momentarily) inactive and heap is empty
      */
     final boolean canShutDown() {
-        DelayedTask<?>[] h;
-        return (active <= 0 &&
-                ((h = heap) == null || h.length <= 0 || h[0] == null) &&
-                active <= 0);
+        return (active <= 0 && restingSize <= 0);
     }
 
     /**
-     * Setup and run scheduling loop
+     * Returns an approximate number of elements in heap
+     */
+    final int approximateSize() {
+        return (active < 0) ? 0 : restingSize;
+    }
+
+    /**
+     * Sets up and runs scheduling loop
      */
     public final void run() {
         ForkJoinPool p;
-        ThreadLocalRandom.localInit();
         if ((p = pool) != null) {
             try {
                 loop(p);
             } finally {
+                restingSize = 0;
                 active = -1;
-                ForkJoinPool.canTerminate(p);
+                ForkJoinPool.poolCanTerminate(p);
             }
         }
     }
@@ -195,9 +201,9 @@ public final void run() {
      *    else park until next trigger time, or indefinitely if none
      */
     private void loop(ForkJoinPool p) {
-        DelayedTask<?>[] h = new DelayedTask<?>[INITIAL_HEAP_CAPACITY];
-        heap = h;
+        p.onDelaySchedulerStart();
         active = 1;
+        DelayedTask<?>[] h = new DelayedTask<?>[INITIAL_HEAP_CAPACITY];
         boolean purgedPeriodic = false;
         for (int n = 0;;) {                    // n is heap size
             DelayedTask<?> t;
@@ -236,7 +242,7 @@ else if (t.status >= 0) {
                             }
                             t.heapIndex = k;
                             h[k] = t;
-                            if (n >= cap && (nh = growHeap(h)) != null)
+                            if (n >= cap && (nh = growHeap(h, cap)) != null)
                                 cap = (h = nh).length;
                         }
                     }
@@ -261,13 +267,14 @@ else if (t.status >= 0) {
                             break;
                         }
                         f.heapIndex = -1;
-                        if (stat >= 0 && p != null)
+                        if (stat >= 0)
                             p.executeReadyDelayedTask(f);
                     }
                 } while ((n = replace(h, 0, n)) > 0);
             }
 
             if (pending == null) {
+                restingSize = n;
                 Thread.interrupted();       // clear before park
                 if (active == 0)
                     U.park(false, parkTime);
@@ -281,20 +288,16 @@ else if (t.status >= 0) {
      * Tries to reallocate the heap array, returning existing
      * array on failure.
      */
-    private DelayedTask<?>[] growHeap(DelayedTask<?>[] h) {
-        int cap, newCap;
-        if (h != null && (cap = h.length) < (newCap = cap << 1)) {
-            DelayedTask<?>[] a = null;
+    private DelayedTask<?>[] growHeap(DelayedTask<?>[] h, int cap) {
+        int newCap  = cap << 1;
+        DelayedTask<?>[] nh = h;
+        if (h != null && h.length == cap && cap < newCap) {
             try {
-                a = Arrays.copyOf(h, newCap);
+                nh = Arrays.copyOf(h, newCap);
             } catch (Error | RuntimeException ex) {
             }
-            if (a != null && a.length > cap) {
-                heap = h = a;
-                U.storeFence();
-            }
         }
-        return h;
+        return nh;
     }
 
     /**
@@ -361,7 +364,7 @@ private int tryStop(ForkJoinPool p, DelayedTask<?>[] h,
                         }
                     }
                 }
-                if (n > 0 || !ForkJoinPool.canTerminate(p))
+                if (n > 0 || !ForkJoinPool.poolCanTerminate(p))
                     return n;
             }
             for (int i = 0; i < n; ++i) {
@@ -378,25 +381,6 @@ private int tryStop(ForkJoinPool p, DelayedTask<?>[] h,
         return -1;
     }
 
-    /**
-     * Returns an approximate count by finding the highest used
-     * heap array slot. This is very racy and not fast, but
-     * useful enough for monitoring purposes.
-     */
-    final int approximateSize() {
-        DelayedTask<?>[] h;
-        int size = 0;
-        if (active >= 0 && (h = heap) != null) {
-            for (int i = h.length - 1; i >= 0; --i) {
-                if (h[i] != null) {
-                    size = i + 1;
-                    break;
-                }
-            }
-        }
-        return size;
-    }
-
     /**
      * Task class for DelayScheduler operations
      */

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

@@ -902,19 +902,21 @@ public class ForkJoinPool extends AbstractExecutorService
      *
      * This class supports ScheduledExecutorService methods by
      * creating and starting a DelayScheduler on first use of these
-     * methods. The scheduler operates independently in its own
-     * thread, relaying tasks to the pool to execute as they become
-     * ready (see method executeReadyDelayedTask). The only other
-     * interactions with the delayScheduler are to control shutdown
-     * and maintain shutdown-related policies in methods quiescent()
-     * and tryTerminate(). In particular, to conform to policies,
-     * shutdown-related processing must deal with cases in which tasks
-     * are submitted before shutdown, but not ready until afterwards,
-     * in which case they must bypass some screening to be allowed to
-     * run. Conversely, the DelayScheduler interacts with the pool
-     * only to check runState status (via mehods poolIsStopping and
-     * poolIsShutdown) and complete termination (via canTerminate)
-     * that invoke corresponding private pool implementations.
+     * methods (via startDelayScheduler, with callback
+     * onDelaySchedulerStart). The scheduler operates independently in
+     * its own thread, relaying tasks to the pool to execute as they
+     * become ready (see method executeReadyDelayedTask). The only
+     * other interactions with the delayScheduler are to control
+     * shutdown and maintain shutdown-related policies in methods
+     * quiescent() and tryTerminate(). In particular, to conform to
+     * policies, shutdown-related processing must deal with cases in
+     * which tasks are submitted before shutdown, but not ready until
+     * afterwards, in which case they must bypass some screening to be
+     * allowed to run. Conversely, the DelayScheduler interacts with
+     * the pool only to check runState status (via mehods
+     * poolIsStopping and poolIsShutdown) and complete termination
+     * (via poolCanTerminate) that invoke corresponding private pool
+     * implementations.
      *
      * Memory placement
      * ================
@@ -1720,17 +1722,6 @@ private long spinLockRunState() {            // spin/sleep
         }
     }
 
-    // status methods used by other classes in this package
-    static boolean poolIsStopping(ForkJoinPool p) {
-        return p != null && (p.runState & STOP) != 0L;
-    }
-    static boolean poolIsShutdown(ForkJoinPool p) {
-        return p != null && (p.runState & SHUTDOWN) != 0L;
-    }
-    static boolean canTerminate(ForkJoinPool p) { // true if terminated
-        return p != null && (p.tryTerminate(false, false) & STOP) != 0L;
-    }
-
     // Creating, registering, and deregistering workers
 
     /**
@@ -3415,6 +3406,17 @@ public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
 
     // Support for delayed tasks
 
+    // status methods used by Delayscheduler and other classes in this package
+    static boolean poolIsStopping(ForkJoinPool p) {
+        return p != null && (p.runState & STOP) != 0L;
+    }
+    static boolean poolIsShutdown(ForkJoinPool p) {
+        return p != null && (p.runState & SHUTDOWN) != 0L;
+    }
+    static boolean poolCanTerminate(ForkJoinPool p) { // true if terminated
+        return p != null && (p.tryTerminate(false, false) & STOP) != 0L;
+    }
+
     /**
      * Creates and starts DelayScheduler unless pool is in shutdown mode
      */
@@ -3443,6 +3445,15 @@ private DelayScheduler startDelayScheduler() {
         return ds;
     }
 
+    /**
+     * Callback upon starting DelayScheduler
+     */
+    final void onDelaySchedulerStart() {
+        WorkQueue q;           // set up default submission queue
+        if ((q = submissionQueue(0, false)) != null)
+            q.unlockPhase();
+    }
+
     /**
      * Arranges execution of a ready task from DelayScheduler
      */
@@ -3753,6 +3764,19 @@ public int getQueuedSubmissionCount() {
         return count;
     }
 
+    /**
+     * Returns an estimate of the number of delayed (including
+     * periodic) tasks scheduled in this pool that are not yet ready
+     * to submit for execution. The returned value is innacurate when
+     * delayed tasks are being processed.
+     *
+     * @return an estimate of the number of delayed tasks
+     */
+    public int getDelayedTaskCount() {
+        DelayScheduler ds;
+        return ((ds = delayScheduler) == null ? 0 : ds.approximateSize());
+    }
+
     /**
      * Returns {@code true} if there are any tasks submitted to this
      * pool that have not yet begun executing.
@@ -3816,6 +3840,7 @@ protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
      */
     public String toString() {
         // Use a single pass through queues to collect counts
+        DelayScheduler ds;
         long e = runState;
         long st = stealCount;
         long qt = 0L, ss = 0L; int rc = 0;
@@ -3835,7 +3860,8 @@ public String toString() {
                 }
             }
         }
-
+        String delayed = ((ds = delayScheduler) == null ? "" :
+                          ", delayed = " + ds.approximateSize());
         int pc = parallelism;
         long c = ctl;
         int tc = (short)(c >>> TC_SHIFT);
@@ -3855,6 +3881,7 @@ public String toString() {
             ", steals = " + st +
             ", tasks = " + qt +
             ", submissions = " + ss +
+            delayed +
             "]";
     }