+ Terminable api change

q3769 · q3769 · commit f01ab4a2bb48 · 2023-09-18T22:03:10.000-05:00
+ replaced ThreadPoolExecutor with ForkJointPool
diff --git a/pom.xml b/pom.xml
@@ -27,7 +27,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>io.github.q3769</groupId>
     <artifactId>conseq4j</artifactId>
-    <version>20230619.20230729.20230730</version>
+    <version>20230918.0.0</version>
     <packaging>jar</packaging>
     <name>conseq4j</name>
     <description>A Java concurrent API to sequence related tasks while concurring unrelated ones</description>
diff --git a/src/main/java/conseq4j/Terminable.java b/src/main/java/conseq4j/Terminable.java
@@ -24,6 +24,8 @@
 
 package conseq4j;
 
+import java.util.List;
+
 /**
  *
  */
@@ -43,7 +45,7 @@ public interface Terminable {
     boolean isTerminated();
 
     /**
-     * @return true if, by estimation, no worker/thread is actively executing tasks.
+     * @return Tasks submitted but never started executing
      */
-    boolean isIdle();
+    List<Runnable> shutdownNow();
 }
diff --git a/src/main/java/conseq4j/execute/ConseqExecutor.java b/src/main/java/conseq4j/execute/ConseqExecutor.java
@@ -31,6 +31,7 @@
 
 import javax.annotation.Nonnull;
 import javax.annotation.concurrent.ThreadSafe;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.*;
 
@@ -42,10 +43,6 @@
 @ThreadSafe
 @ToString
 public final class ConseqExecutor implements SequentialExecutor {
-    private static final int DEFAULT_CONCURRENCY = Math.max(16, Runtime.getRuntime().availableProcessors());
-    private static final int DEFAULT_WORK_QUEUE_CAPACITY = Integer.MAX_VALUE;
-    private static final RejectedExecutionHandler DEFAULT_REJECTED_HANDLER = new ThreadPoolExecutor.AbortPolicy();
-    private static final Builder.WorkQueueType DEFAULT_WORK_QUEUE_TYPE = Builder.WorkQueueType.LINKED;
 
     private final Map<Object, CompletableFuture<?>> activeSequentialTasks = new ConcurrentHashMap<>();
     private final ExecutorService adminService = Executors.newSingleThreadExecutor();
@@ -54,49 +51,36 @@ public final class ConseqExecutor implements SequentialExecutor {
      * from the pool can be used to execute any task, regardless of sequence keys. The pool capacity decides the overall
      * max parallelism of task execution.
      */
-    private final ThreadPoolExecutor workerThreadPool;
+    private final ExecutorService workerExecutorService;
     private final ConditionFactory await = Awaitility.await().forever();
 
-    private ConseqExecutor(@Nonnull Builder builder) {
-        this(new ThreadPoolExecutor(builder.concurrency,
-                builder.concurrency,
-                0,
-                TimeUnit.MILLISECONDS,
-                builder.workQueueType == Builder.WorkQueueType.ARRAY ?
-                        new ArrayBlockingQueue<>(builder.workQueueCapacity) :
-                        new LinkedBlockingQueue<>(builder.workQueueCapacity),
-                Executors.defaultThreadFactory(),
-                builder.rejectedExecutionHandler));
-    }
-
-    private ConseqExecutor(ThreadPoolExecutor workerThreadPool) {
-        this.workerThreadPool = workerThreadPool;
+    private ConseqExecutor(ExecutorService workerExecutorService) {
+        this.workerExecutorService = workerExecutorService;
     }
 
     /**
      * @return conseq executor with default concurrency
      */
-    public static @Nonnull ConseqExecutor newInstance() {
-        return new Builder().build();
+    public static @Nonnull ConseqExecutor instance() {
+        return instance(Runtime.getRuntime().availableProcessors());
     }
 
     /**
      * @param concurrency
-     *         max number of tasks that can be run in parallel by the returned executor instance. This is set as the max
-     *         capacity of the {@link #workerThreadPool}
+     *         max number of tasks that can be run in parallel by the returned executor instance.
      * @return conseq executor with given concurrency
      */
-    public static @Nonnull ConseqExecutor newInstance(int concurrency) {
-        return new Builder().concurrency(concurrency).build();
+    public static @Nonnull ConseqExecutor instance(int concurrency) {
+        return instance(new ForkJoinPool(concurrency, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true));
     }
 
     /**
-     * @param workerThreadPool
-     *         the worker thread pool that facilitates the overall async execution, independent of the submitted tasks.
-     * @return new ConseqExecutor instance backed by the specified workerThreadPool
+     * @param workerExecutorService
+     *         ExecutorService that backs the async operations of worker threads
+     * @return instance of {@link ConseqExecutor}
      */
-    public static ConseqExecutor from(ThreadPoolExecutor workerThreadPool) {
-        return new ConseqExecutor(workerThreadPool);
+    public static @Nonnull ConseqExecutor instance(ExecutorService workerExecutorService) {
+        return new ConseqExecutor(workerExecutorService);
     }
 
     private static <T> T call(Callable<T> task) {
@@ -122,7 +106,7 @@ public CompletableFuture<Void> execute(@NonNull Runnable command, @NonNull Objec
 
     /**
      * Tasks of different sequence keys execute in parallel, pending thread availability from the backing
-     * {@link #workerThreadPool}.
+     * {@link #workerExecutorService}.
      * <p>
      * Sequential execution of tasks under the same/equal sequence key is achieved by linearly chaining/queuing the
      * task/work stages of the same key, leveraging the {@link CompletableFuture} API.
@@ -160,8 +144,8 @@ public CompletableFuture<Void> execute(@NonNull Runnable command, @NonNull Objec
     public <T> CompletableFuture<T> submit(@NonNull Callable<T> task, @NonNull Object sequenceKey) {
         CompletableFuture<?> latestTask = activeSequentialTasks.compute(sequenceKey,
                 (k, presentTask) -> (presentTask == null) ?
-                        CompletableFuture.supplyAsync(() -> call(task), workerThreadPool) :
-                        presentTask.handleAsync((r, e) -> call(task), workerThreadPool));
+                        CompletableFuture.supplyAsync(() -> call(task), workerExecutorService) :
+                        presentTask.handleAsync((r, e) -> call(task), workerExecutorService));
         latestTask.whenCompleteAsync((r, e) -> activeSequentialTasks.computeIfPresent(sequenceKey,
                 (k, checkedTask) -> checkedTask.isDone() ? null : checkedTask), adminService);
         return (CompletableFuture<T>) latestTask.thenApply(r -> r);
@@ -171,113 +155,26 @@ public <T> CompletableFuture<T> submit(@NonNull Callable<T> task, @NonNull Objec
     public void shutdown() {
         ExecutorService asyncThread = Executors.newSingleThreadExecutor();
         asyncThread.execute(() -> {
-            workerThreadPool.shutdown();
-            await.until(this::isIdle);
+            workerExecutorService.shutdown();
+            await.until(activeSequentialTasks::isEmpty);
             adminService.shutdown();
         });
         asyncThread.shutdown();
     }
 
     @Override
     public boolean isTerminated() {
-        return this.workerThreadPool.isTerminated() && this.adminService.isTerminated();
+        return this.workerExecutorService.isTerminated() && this.adminService.isTerminated();
     }
 
     @Override
-    public boolean isIdle() {
-        return activeSequentialTasks.isEmpty() && workerThreadPool.getActiveCount() == 0;
+    public List<Runnable> shutdownNow() {
+        List<Runnable> neverStartedTasks = workerExecutorService.shutdownNow();
+        adminService.shutdownNow();
+        return neverStartedTasks;
     }
 
     int estimateActiveExecutorCount() {
         return activeSequentialTasks.size();
     }
-
-    /**
-     * {@code ConseqExecutor} builder static inner class.
-     */
-    public static final class Builder {
-        private int concurrency;
-        private int workQueueCapacity;
-        private WorkQueueType workQueueType;
-        private RejectedExecutionHandler rejectedExecutionHandler;
-
-        /**
-         *
-         */
-        public Builder() {
-            concurrency = DEFAULT_CONCURRENCY;
-            workQueueCapacity = DEFAULT_WORK_QUEUE_CAPACITY;
-            rejectedExecutionHandler = DEFAULT_REJECTED_HANDLER;
-            workQueueType = DEFAULT_WORK_QUEUE_TYPE;
-        }
-
-        /**
-         * Sets the {@code concurrency} and returns a reference to this Builder enabling method chaining.
-         *
-         * @param concurrency
-         *         the {@code concurrency} to set
-         * @return a reference to this Builder
-         */
-        public Builder concurrency(int concurrency) {
-            this.concurrency = concurrency;
-            return this;
-        }
-
-        /**
-         * Sets the {@code workQueueCapacity} and returns a reference to this Builder enabling method chaining.
-         *
-         * @param workQueueCapacity
-         *         the {@code workQueueCapacity} to set
-         * @return a reference to this Builder
-         */
-        public Builder workQueueCapacity(int workQueueCapacity) {
-            this.workQueueCapacity = workQueueCapacity;
-            return this;
-        }
-
-        /**
-         * @param rejectedExecutionHandler
-         *         handler executed by the caller thread if a task is rejected by the conseq executor, maybe e.g.
-         *         because of work queue is full. Default is {@link ThreadPoolExecutor.AbortPolicy}
-         * @return a reference to this Builder
-         */
-        public Builder rejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) {
-            this.rejectedExecutionHandler = rejectedExecutionHandler;
-            return this;
-        }
-
-        /**
-         * Returns a {@code ConseqExecutor} built from the parameters previously set.
-         *
-         * @return a {@code ConseqExecutor} built with parameters of this {@code ConseqExecutor.Builder}
-         */
-        public @Nonnull ConseqExecutor build() {
-            return new ConseqExecutor(this);
-        }
-
-        /**
-         * @param workQueueType
-         *         {@link WorkQueueType#LINKED} meaning {@link LinkedBlockingQueue}, or  {@link WorkQueueType#ARRAY}
-         *         meaning {@link ArrayBlockingQueue}, default to {@link WorkQueueType#LINKED}
-         * @return a reference to this Builder
-         */
-        public Builder workQueueType(WorkQueueType workQueueType) {
-            this.workQueueType = workQueueType;
-            return this;
-        }
-
-        /**
-         *
-         */
-        public enum WorkQueueType {
-            /**
-             *
-             */
-            LINKED,
-            /**
-             *
-             */
-            ARRAY
-        }
-    }
 }
diff --git a/src/main/java/conseq4j/summon/ConseqServiceFactory.java b/src/main/java/conseq4j/summon/ConseqServiceFactory.java
@@ -26,11 +26,15 @@
 import lombok.ToString;
 import lombok.experimental.Delegate;
 
+import javax.annotation.Nonnull;
 import javax.annotation.concurrent.ThreadSafe;
-import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
 import static java.lang.Math.floorMod;
@@ -89,25 +93,21 @@ public ExecutorService getExecutorService(Object sequenceKey) {
 
     @Override
     public void shutdown() {
-        List<ExecutorService> executorServices = new ArrayList<>(this.sequentialExecutors.values());
-        List<ExecutorService> shutdownDisabledExecutorServices = executorServices.stream()
-                .filter(ShutdownDisabledExecutorService.class::isInstance)
-                .collect(Collectors.toList());
-        shutdownDisabledExecutorServices.stream()
-                .map(ShutdownDisabledExecutorService.class::cast)
-                .forEach(ShutdownDisabledExecutorService::shutdownDelegate);
-        executorServices.removeAll(shutdownDisabledExecutorServices);
-        executorServices.forEach(ExecutorService::shutdown);
+        this.sequentialExecutors.values().forEach(ShutdownDisabledExecutorService::shutdownDelegate);
     }
 
     @Override
     public boolean isTerminated() {
-        return this.sequentialExecutors.values().parallelStream().allMatch(ExecutorService::isTerminated);
+        return this.sequentialExecutors.values().stream().allMatch(ExecutorService::isTerminated);
     }
 
     @Override
-    public boolean isIdle() {
-        return sequentialExecutors.values().stream().allMatch(e -> e.getActiveCount() == 0);
+    public List<Runnable> shutdownNow() {
+        return sequentialExecutors.values()
+                .parallelStream()
+                .map(ShutdownDisabledExecutorService::shutdownDelegateNow)
+                .flatMap(Collection::stream)
+                .collect(Collectors.toList());
     }
 
     private int bucketOf(Object sequenceKey) {
@@ -149,22 +149,20 @@ public void shutdown() {
          * @see #shutdown()
          */
         @Override
+        @Nonnull
         public List<Runnable> shutdownNow() {
             throw new UnsupportedOperationException(SHUTDOWN_UNSUPPORTED_MESSAGE);
         }
 
-        public int getActiveCount() {
-            if (!(delegate instanceof ThreadPoolExecutor)) {
-                throw new UnsupportedOperationException(
-                        "Delegate " + delegate + " is not of type " + ThreadPoolExecutor.class);
-            }
-            return ((ThreadPoolExecutor) delegate).getActiveCount();
-        }
-
         void shutdownDelegate() {
             this.delegate.shutdown();
         }
 
+        @Nonnull
+        List<Runnable> shutdownDelegateNow() {
+            return this.delegate.shutdownNow();
+        }
+
         /**
          * Methods that require complete overriding instead of delegation/decoration
          */
diff --git a/src/test/java/conseq4j/execute/ConseqExecutorTest.java b/src/test/java/conseq4j/execute/ConseqExecutorTest.java
