TaskExecutor ScheduledExecutorService etc. cleanup

src/dev/enola/be/task/TaskExecutor.java

@@ -2,14 +2,16 @@
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
+import dev.enola.common.concurrent.Executors;
+
+import java.time.Duration;
 import java.time.Instant;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.FutureTask;
 import java.util.concurrent.ScheduledExecutorService;
@@ -22,20 +24,43 @@ public class TaskExecutor implements AutoCloseable {
     // TODO Synthetic "root" task, to which all running tasks are children?
     // This could be useful for managing task hierarchies and dependencies.
 
-    // TODO Eviction policy of completed tasks?! As-is, this leaks memory...
-    // E.g. periodically scan the map and remove tasks that are in a terminal state.
-    // Persist them first, so that get() can still find them later; with separate
-    // eviction policy.
+    // This map has a basic time-based eviction policy; see constructor.
+    // A more sophisticated implementation could persist completed tasks, so that get()
+    // can still find them later, with a separate eviction policy for that persistent store.
     private final Map<UUID, Task<?, ?>> tasks = new ConcurrentHashMap<>();
 
-    private final ExecutorService executor = TaskExecutorServices.newVirtualThreadPerTaskExecutor();
-
-    // TODO Can the timeoutScheduler also use virtual threads?
-    //   (Would need to check if ScheduledExecutorService supports that.)
+    // Nota bene: In *THEORY* we should *NEVER* have *ANY* uncaught exceptions from Task,
+    // because any exception thrown by the task's `execute()` method would be caught and
+    // wrapped in an ExecutionException by the Future returned by ExecutorService.submit().
     //
-    // TODO Either way, use LoggingScheduledExecutorService from Enola Commons?
+    // But in practice, who knows what the future holds, so we better log them just in case;
+    // just because "swallowed" lost exceptions are seriously the worst kind of bugs to diagnose!
+    private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(LOG);
+
     private final ScheduledExecutorService timeoutScheduler =
-            Executors.newSingleThreadScheduledExecutor();
+            Executors.newSingleThreadScheduledExecutor("TaskExecutor-Timeout", LOG);
+
+    private final ScheduledExecutorService cleanupScheduler =
+            Executors.newSingleThreadScheduledExecutor("TaskExecutor-Cleanup", LOG);
+
+    public TaskExecutor(Duration completedTaskEvictionInterval) {
+        if (completedTaskEvictionInterval == null) {
+            throw new IllegalArgumentException("completedTaskEvictionInterval must not be null");
+        }
+        if (completedTaskEvictionInterval.isNegative() || completedTaskEvictionInterval.isZero()) {
+            throw new IllegalArgumentException("completedTaskEvictionInterval must be positive");
+        }
+        var m = completedTaskEvictionInterval.toMillis();
+        cleanupScheduler.scheduleAtFixedRate(this::evictCompletedTasks, m, m, MILLISECONDS);
+    }
+
+    public TaskExecutor() {
+        this(Duration.ofHours(1));
+    }
+
+    private void evictCompletedTasks() {
+        tasks.values().removeIf(task -> task.status().isTerminal());
+    }
 
     private static class LoggingFutureTask<V> extends FutureTask<V> {
         private final Task<?, V> task;
@@ -127,7 +152,8 @@ public void close() {
             task.cancel();
         }
 
-        executor.close();
+        cleanupScheduler.close();
         timeoutScheduler.close();
+        executor.close();
     }
 }

src/dev/enola/common/concurrent/Executors.java

@@ -0,0 +1,56 @@
+package dev.enola.common.concurrent;
+
+import static dev.enola.common.concurrent.LoggingThreadUncaughtExceptionHandler.toLogger;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.logging.Logger;
+
+public final class Executors {
+
+    public static ScheduledExecutorService newSingleThreadScheduledExecutor(
+            String namePrefix, Logger logger) {
+        var tf = createThreadFactory(namePrefix, logger);
+        return java.util.concurrent.Executors.newSingleThreadScheduledExecutor(tf);
+    }
+
+    public static ExecutorService newVirtualThreadPerTaskExecutor(
+            String namePrefix, Logger logger) {
+        var tf = createVirtualThreadFactory(namePrefix, logger);
+        return java.util.concurrent.Executors.newThreadPerTaskExecutor(tf);
+    }
+
+    public static ExecutorService newVirtualThreadPerTaskExecutor(Logger logger) {
+        var tf = createVirtualThreadFactory(logger);
+        return java.util.concurrent.Executors.newThreadPerTaskExecutor(tf);
+    }
+
+    private static ThreadFactory createThreadFactory(String namePrefix, Logger logger) {
+        return Thread.ofPlatform()
+                .name(namePrefix, 1)
+                .uncaughtExceptionHandler(toLogger(logger))
+                .daemon(true)
+                // TODO new ContextAwareThreadFactory() how-to?
+                // TODO .inheritInheritableThreadLocals(true) ?
+                .factory();
+    }
+
+    private static ThreadFactory createVirtualThreadFactory(String namePrefix, Logger logger) {
+        var builder = Thread.ofVirtual();
+        if (namePrefix != null) {
+            builder = builder.name(namePrefix, 1);
+        }
+        return builder.uncaughtExceptionHandler(toLogger(logger))
+                // NB: Virtual threads are always daemon threads, so no: .setDaemon(true)
+                // TODO new ContextAwareThreadFactory() how-to?
+                // TODO .inheritInheritableThreadLocals(true) ?
+                .factory();
+    }
+
+    private static ThreadFactory createVirtualThreadFactory(Logger logger) {
+        return createVirtualThreadFactory(null, logger);
+    }
+
+    private Executors() {}
+}

src/dev/enola/common/concurrent/LoggingThreadUncaughtExceptionHandler.java

@@ -0,0 +1,27 @@
+package dev.enola.common.concurrent;
+
+import static java.util.Objects.requireNonNull;
+
+import java.lang.Thread.UncaughtExceptionHandler;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+class LoggingThreadUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
+
+    private final Logger logger;
+
+    private LoggingThreadUncaughtExceptionHandler(Logger logger) {
+        this.logger = requireNonNull(logger, "logger");
+    }
+
+    /** Factory method to obtain an instance of this bound to the passed JUL Logger. */
+    public static UncaughtExceptionHandler toLogger(Logger logger) {
+        return new LoggingThreadUncaughtExceptionHandler(logger);
+    }
+
+    @Override
+    public void uncaughtException(Thread thread, Throwable throwable) {
+        logger.log(
+                Level.SEVERE, "Uncaught exception in thread '" + thread.getName() + "'", throwable);
+    }
+}