Trampoline#forkAll optimisation

Author: alexandru

tasks-jvm/src/main/java/org/funfix/tasks/jvm/DelayedFun.java

@@ -2,7 +2,6 @@
 
 import org.jspecify.annotations.Nullable;
 
-import java.util.concurrent.Delayed;
 /**
  * Represents a delayed computation (a thunk).
  * <p>

tasks-jvm/src/main/java/org/funfix/tasks/jvm/Task.java

@@ -518,8 +518,8 @@ private T awaitInline(final Cancellable cancelToken, final AwaitFunction await)
             } catch (final TimeoutException | InterruptedException e) {
                 if (!isCancelled) {
                     isCancelled = true;
-                    if (e instanceof TimeoutException)
-                        timedOut = (TimeoutException) e;
+                    if (e instanceof TimeoutException te)
+                        timedOut = te;
                     cancelToken.cancel();
                 }
             }

tasks-jvm/src/main/java/org/funfix/tasks/jvm/TaskExecutor.java

@@ -23,7 +23,9 @@ final class TaskLocalContext {
     static void signalTheStartOfBlockingCall() {
         // clears the trampoline first
         final var executor = localExecutor.get();
-        Trampoline.forkAll(executor);
+        Trampoline.forkAll(
+            executor != null ? executor : TaskExecutors.sharedBlockingIO()
+        );
     }
 
     static boolean isCurrentExecutor(final TaskExecutor executor) {
@@ -64,7 +66,7 @@ public void execute(final Runnable command) {
     @Override
     public void resumeOnExecutor(final Runnable runnable) {
         if (TaskLocalContext.isCurrentExecutor(this)) {
-            Trampoline.INSTANCE.execute(runnable);
+            Trampoline.execute(runnable);
         } else {
             execute(runnable);
         }

tasks-jvm/src/main/java/org/funfix/tasks/jvm/TaskExecutors.java

@@ -76,7 +76,7 @@ private static Executor sharedVirtualIO() {
      * On Java 21 and above, the created {@code Executor} will run tasks on virtual threads.
      * On older JVM versions, it returns a plain {@code Executors.newCachedThreadPool}.
      */
-    @SuppressWarnings("deprecation")
+    @SuppressWarnings({"deprecation", "EmptyCatch"})
     public static ExecutorService unlimitedThreadPoolForIO(final String prefix) {
         if (VirtualThreads.areVirtualThreadsSupported())
             try {

tasks-jvm/src/main/java/org/funfix/tasks/jvm/Trampoline.java

@@ -3,7 +3,7 @@
 import org.jetbrains.annotations.ApiStatus;
 import org.jspecify.annotations.Nullable;
 
-import java.util.LinkedList;
+import java.util.ArrayDeque;
 import java.util.concurrent.Executor;
 
 /**
@@ -17,7 +17,7 @@
 final class Trampoline {
     private Trampoline() {}
 
-    private static final ThreadLocal<@Nullable LinkedList<Runnable>> queue =
+    private static final ThreadLocal<@Nullable ArrayDeque<Runnable>> queue =
             new ThreadLocal<>();
 
     private static void eventLoop() {
@@ -49,7 +49,7 @@ public void resumeOnExecutor(Runnable runnable) {
             public void execute(Runnable command) {
                 var current = queue.get();
                 if (current == null) {
-                    current = new LinkedList<>();
+                    current = new ArrayDeque<>();
                     current.add(command);
                     queue.set(current);
                     try {
@@ -63,20 +63,17 @@ public void execute(Runnable command) {
             }
         };
 
-    public static void forkAll(final @Nullable Executor executor) {
+    public static void forkAll(final Executor executor) {
         final var current = queue.get();
-        while (current != null && !current.isEmpty()) {
-            final var next = current.pollFirst();
-            if (executor != null) {
-                executor.execute(next);
-            } else {
-                try {
-                    next.run();
-                } catch (final Throwable e) {
-                    UncaughtExceptionHandler.logOrRethrow(e);
-                }
+        if (current == null) return;
+
+        final var copy = new ArrayDeque<>(current);
+        executor.execute(() -> Trampoline.execute(() -> {
+            while (!copy.isEmpty()) {
+                final var next = copy.pollFirst();
+                Trampoline.execute(next);
             }
-        }
+        }));
     }
 
     public static void execute(final Runnable command) {

tasks-jvm/src/main/java/org/funfix/tasks/jvm/UncaughtExceptionHandler.java

@@ -10,8 +10,8 @@ public static void rethrowIfFatal(final Throwable e) {
             // the process
             return;
         }
-        if (e instanceof Error) {
-            throw (Error) e;
+        if (e instanceof Error error) {
+            throw error;
         }
     }