8319447: Improve performance of delayed task handling

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

@@ -330,14 +330,9 @@ private void setDone() {
      */
     final int trySetCancelled() {
         int s;
-        for (;;) {
-            if ((s = status) < 0)
-                break;
-            if (casStatus(s, s | (DONE | ABNORMAL))) {
-                signalWaiters();
-                break;
-            }
-        }
+        if ((s = status) >= 0 &&
+            (s = getAndBitwiseOrStatus(DONE | ABNORMAL)) >= 0)
+            signalWaiters();
         return s;
     }
 
@@ -642,7 +637,7 @@ public final ForkJoinTask<V> fork() {
             p = wt.pool;
         }
         else
-            q = (p = ForkJoinPool.common).externalSubmissionQueue();
+            q = (p = ForkJoinPool.common).externalSubmissionQueue(false);
         q.push(this, p, internal);
         return this;
     }
@@ -1415,7 +1410,8 @@ public final boolean compareAndSetForkJoinTaskTag(short expect, short update) {
      * @return the task
      */
     public static ForkJoinTask<?> adapt(Runnable runnable) {
-        return new AdaptedRunnableAction(runnable);
+        return new AdaptedRunnableAction(
+            Objects.requireNonNull(runnable));
     }
 
     /**
@@ -1429,7 +1425,8 @@ public static ForkJoinTask<?> adapt(Runnable runnable) {
      * @return the task
      */
     public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
-        return new AdaptedRunnable<T>(runnable, result);
+        return new AdaptedRunnable<T>(
+            Objects.requireNonNull(runnable), result);
     }
 
     /**
@@ -1443,7 +1440,8 @@ public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
      * @return the task
      */
     public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
-        return new AdaptedCallable<T>(callable);
+        return new AdaptedCallable<T>(
+            Objects.requireNonNull(callable));
     }
 
     /**
@@ -1461,7 +1459,8 @@ public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
      * @since 19
      */
     public static <T> ForkJoinTask<T> adaptInterruptible(Callable<? extends T> callable) {
-        return new AdaptedInterruptibleCallable<T>(callable);
+        return new AdaptedInterruptibleCallable<T>(
+            Objects.requireNonNull(callable));
     }
 
     /**
@@ -1480,7 +1479,8 @@ public static <T> ForkJoinTask<T> adaptInterruptible(Callable<? extends T> calla
      * @since 22
      */
     public static <T> ForkJoinTask<T> adaptInterruptible(Runnable runnable, T result) {
-        return new AdaptedInterruptibleRunnable<T>(runnable, result);
+        return new AdaptedInterruptibleRunnable<T>(
+            Objects.requireNonNull(runnable), result);
     }
 
     /**
@@ -1498,7 +1498,8 @@ public static <T> ForkJoinTask<T> adaptInterruptible(Runnable runnable, T result
      * @since 22
      */
     public static ForkJoinTask<?> adaptInterruptible(Runnable runnable) {
-        return new AdaptedInterruptibleRunnable<Void>(runnable, null);
+        return new AdaptedInterruptibleRunnable<Void>(
+            Objects.requireNonNull(runnable), null);
     }
 
     // Serialization support
@@ -1557,7 +1558,6 @@ static final class AdaptedRunnable<T> extends ForkJoinTask<T>
         @SuppressWarnings("serial") // Conditionally serializable
         T result;
         AdaptedRunnable(Runnable runnable, T result) {
-            Objects.requireNonNull(runnable);
             this.runnable = runnable;
             this.result = result; // OK to set this even before completion
         }
@@ -1579,7 +1579,6 @@ static final class AdaptedRunnableAction extends ForkJoinTask<Void>
         @SuppressWarnings("serial") // Conditionally serializable
         final Runnable runnable;
         AdaptedRunnableAction(Runnable runnable) {
-            Objects.requireNonNull(runnable);
             this.runnable = runnable;
         }
         public final Void getRawResult() { return null; }
@@ -1602,7 +1601,6 @@ static final class AdaptedCallable<T> extends ForkJoinTask<T>
         @SuppressWarnings("serial") // Conditionally serializable
         T result;
         AdaptedCallable(Callable<? extends T> callable) {
-            Objects.requireNonNull(callable);
             this.callable = callable;
         }
         public final T getRawResult() { return result; }
@@ -1656,20 +1654,29 @@ public final boolean exec() {
             } finally {
                 runner = null;
             }
+            return postExec();
+        }
+        boolean postExec() { // cleanup and return completion status to doExec
             return true;
         }
+        final boolean interruptIfRunning(boolean enabled) {
+             Thread t;
+             if ((t = runner) == null) // return false if not running
+                 return false;
+             if (enabled) {
+                 try {
+                     t.interrupt();
+                 } catch (Throwable ignore) {
+                 }
+             }
+             return true;
+        }
         public boolean cancel(boolean mayInterruptIfRunning) {
-            Thread t;
-            if (trySetCancelled() >= 0) {
-                if (mayInterruptIfRunning && (t = runner) != null) {
-                    try {
-                        t.interrupt();
-                    } catch (Throwable ignore) {
-                    }
-                }
-                return true;
-            }
-            return isCancelled();
+            int s;
+            if ((s = trySetCancelled()) < 0)
+                return ((s & (ABNORMAL | THROWN)) == ABNORMAL);
+            interruptIfRunning(mayInterruptIfRunning);
+            return true;
         }
         public final void run() { quietlyInvoke(); }
         Object adaptee() { return null; } // for printing and diagnostics
@@ -1691,7 +1698,6 @@ static final class AdaptedInterruptibleCallable<T> extends InterruptibleTask<T>
         @SuppressWarnings("serial") // Conditionally serializable
         T result;
         AdaptedInterruptibleCallable(Callable<? extends T> callable) {
-            Objects.requireNonNull(callable);
             this.callable = callable;
         }
         public final T getRawResult() { return result; }
@@ -1710,7 +1716,6 @@ static final class AdaptedInterruptibleRunnable<T> extends InterruptibleTask<T>
         @SuppressWarnings("serial") // Conditionally serializable
         final T result;
         AdaptedInterruptibleRunnable(Runnable runnable, T result) {
-            Objects.requireNonNull(runnable);
             this.runnable = runnable;
             this.result = result;
         }
@@ -1728,7 +1733,6 @@ static final class RunnableExecuteAction extends InterruptibleTask<Void> {
         @SuppressWarnings("serial") // Conditionally serializable
         final Runnable runnable;
         RunnableExecuteAction(Runnable runnable) {
-            Objects.requireNonNull(runnable);
             this.runnable = runnable;
         }
         public final Void getRawResult() { return null; }
@@ -1794,9 +1798,12 @@ final T invokeAny(Collection<? extends Callable<T>> tasks,
                 throw new NullPointerException();
             InvokeAnyTask<T> t = null; // list of submitted tasks
             try {
-                for (Callable<T> c : tasks)
+                for (Callable<T> c : tasks) {
+                    if (c == null)
+                        throw new NullPointerException();
                     pool.execute((ForkJoinTask<?>)
                                  (t = new InvokeAnyTask<T>(c, this, t)));
+                }
                 return timed ? get(nanos, TimeUnit.NANOSECONDS) : get();
             } finally {
                 for (; t != null; t = t.pred)
@@ -1823,7 +1830,6 @@ static final class InvokeAnyTask<T> extends InterruptibleTask<Void> {
         final InvokeAnyTask<T> pred; // to traverse on cancellation
         InvokeAnyTask(Callable<T> callable, InvokeAnyRoot<T> root,
                       InvokeAnyTask<T> pred) {
-            Objects.requireNonNull(callable);
             this.callable = callable;
             this.root = root;
             this.pred = pred;
@@ -1858,4 +1864,39 @@ final void onRootCompletion() {
         public final void setRawResult(Void v) { }
         final Object adaptee() { return callable; }
     }
+
+    /**
+     * Adapter for Callable-based interruptible tasks with timeout actions.
+     */
+    @SuppressWarnings("serial") // Conditionally serializable
+    static final class CallableWithTimeout<T> extends InterruptibleTask<T> {
+        Callable<? extends T> callable; // nulled out after use
+        ForkJoinTask<?> timeoutAction;
+        T result;
+        CallableWithTimeout(Callable<? extends T> callable,
+                            ForkJoinTask<?> timeoutAction) {
+            this.callable = callable;
+            this.timeoutAction = timeoutAction;
+        }
+        public final T getRawResult() { return result; }
+        public final void setRawResult(T v) { result = v; }
+        final Object adaptee() { return callable; }
+        final T compute() throws Exception {
+            Callable<? extends T> c;
+            return ((c = callable) != null) ? c.call() : null;
+        }
+        final boolean postExec() {       // cancel timeout action
+            ForkJoinTask<?> t;
+            callable = null;
+            if ((t = timeoutAction) != null) {
+                timeoutAction = null;
+                try {
+                    t.cancel(false);
+                } catch (Error | RuntimeException ex) {
+                }
+            }
+            return true;
+        }
+    }
+
 }

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

@@ -206,22 +206,6 @@ static final int roundCapacity(int cap) {
             (n >= BUFFER_CAPACITY_LIMIT) ? BUFFER_CAPACITY_LIMIT : n + 1;
     }
 
-    // default Executor setup; nearly the same as CompletableFuture
-
-    /**
-     * Default executor -- ForkJoinPool.commonPool() unless it cannot
-     * support parallelism.
-     */
-    private static final Executor ASYNC_POOL =
-        (ForkJoinPool.getCommonPoolParallelism() > 1) ?
-        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
-
-    /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */
-    private static final class ThreadPerTaskExecutor implements Executor {
-        ThreadPerTaskExecutor() {}      // prevent access constructor creation
-        public void execute(Runnable r) { new Thread(r).start(); }
-    }
-
     /**
      * Clients (BufferedSubscriptions) are maintained in a linked list
      * (via their "next" fields). This works well for publish loops.
@@ -316,7 +300,7 @@ public SubmissionPublisher(Executor executor, int maxBufferCapacity) {
      * Flow.Subscriber#onNext(Object) onNext}.
      */
     public SubmissionPublisher() {
-        this(ASYNC_POOL, Flow.defaultBufferSize(), null);
+        this(ForkJoinPool.asyncCommonPool(), Flow.defaultBufferSize(), null);
     }
 
     /**

test/jdk/java/util/concurrent/CompletableFuture/CompletableFutureOrTimeoutExceptionallyTest.java

@@ -30,39 +30,25 @@
  */
 
 import java.time.Duration;
-import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 class CompletableFutureOrTimeoutExceptionallyTest {
-    static final BlockingQueue<Runnable> delayerQueue;
-    static {
-        try {
-            var delayerClass = Class.forName("java.util.concurrent.CompletableFuture$Delayer",
-                                             true,
-                                             CompletableFuture.class.getClassLoader());
-            var delayerField = delayerClass.getDeclaredField("delayer");
-            delayerField.setAccessible(true);
-            delayerQueue = ((ScheduledThreadPoolExecutor)delayerField.get(null)).getQueue();
-        } catch (Throwable t) {
-            throw new ExceptionInInitializerError(t);
-        }
-    }
-
+    // updated February 2025 to adapt to CompletableFuture DelayScheduler changes
     /**
      * Test that orTimeout task is cancelled if the CompletableFuture is completed Exceptionally
      */
     @Test
     void testOrTimeoutWithCompleteExceptionallyDoesNotLeak() throws InterruptedException {
-        assertTrue(delayerQueue.peek() == null);
+        ForkJoinPool delayer = ForkJoinPool.commonPool();
+        assertEquals(delayer.getDelayedTaskCount(), 0);
         var future = new CompletableFuture<>().orTimeout(12, TimeUnit.HOURS);
-        assertTrue(delayerQueue.peek() != null);
         future.completeExceptionally(new RuntimeException("This is fine"));
-        while (delayerQueue.peek() != null) {
+        while (delayer.getDelayedTaskCount() != 0) {
             Thread.sleep(100);
         };
     }
@@ -72,12 +58,13 @@ void testOrTimeoutWithCompleteExceptionallyDoesNotLeak() throws InterruptedExcep
      */
     @Test
     void testCompleteOnTimeoutWithCompleteExceptionallyDoesNotLeak() throws InterruptedException {
-        assertTrue(delayerQueue.peek() == null);
+        ForkJoinPool delayer = ForkJoinPool.commonPool();
+        assertEquals(delayer.getDelayedTaskCount(), 0);
         var future = new CompletableFuture<>().completeOnTimeout(null, 12, TimeUnit.HOURS);
-        assertTrue(delayerQueue.peek() != null);
         future.completeExceptionally(new RuntimeException("This is fine"));
-        while (delayerQueue.peek() != null) {
+        while (delayer.getDelayedTaskCount() != 0) {
             Thread.sleep(100);
         };
     }
+
 }

test/jdk/java/util/concurrent/tck/CompletableFutureTest.java

@@ -658,8 +658,6 @@ public void execute(Runnable r) {
         }
     }
 
-    static final boolean defaultExecutorIsCommonPool
-        = ForkJoinPool.getCommonPoolParallelism() > 1;
 
     /**
      * Permits the testing of parallel code for the 3 different
@@ -750,8 +748,7 @@ public <U> CompletableFuture<U> supplyAsync(Supplier<U> a) {
         },
         ASYNC {
             public void checkExecutionMode() {
-                mustEqual(defaultExecutorIsCommonPool,
-                             (ForkJoinPool.commonPool() == ForkJoinTask.getPool()));
+                mustEqual(ForkJoinPool.commonPool(), ForkJoinTask.getPool());
             }
             public CompletableFuture<Void> runAsync(Runnable a) {
                 return CompletableFuture.runAsync(a);
@@ -3794,10 +3791,7 @@ public void testDefaultExecutor() {
         CompletableFuture<Item> f = new CompletableFuture<>();
         Executor e = f.defaultExecutor();
         Executor c = ForkJoinPool.commonPool();
-        if (ForkJoinPool.getCommonPoolParallelism() > 1)
-            assertSame(e, c);
-        else
-            assertNotSame(e, c);
+        assertSame(e, c);
     }
 
     /**