Unify the handling of thread pools and their lifecycle. Fixes JCS-248

Author: tvand

commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/CacheEventQueue.java

@@ -56,22 +56,20 @@ public CacheEventQueue( final ICacheListener<K, V> listener, final long listener
     public CacheEventQueue( final ICacheListener<K, V> listener, final long listenerId, final String cacheName, final int maxFailure,
                             final int waitBeforeRetry )
     {
-        super( listener, listenerId, cacheName, maxFailure, waitBeforeRetry, null );
+        super( listener, listenerId, cacheName, maxFailure, waitBeforeRetry, "CacheEventQueue.QProcessor-" + cacheName);
     }
 
     /**
      * Create the thread pool.
      *
-     * @param threadPoolName
      * @since 3.1
      */
     @Override
-    protected ExecutorService createPool(final String threadPoolName)
+    protected ExecutorService createPool()
     {
         // create a default pool with one worker thread to mimic the SINGLE queue behavior
-        return ThreadPoolManager.getInstance().createPool(
-                new PoolConfiguration(false, 0, 1, 1, getWaitToDie(), WhenBlockedPolicy.RUN, 1),
-                "CacheEventQueue.QProcessor-" + getCacheName());
+        return ThreadPoolManager.getInstance().getExecutorService(poolName,
+                new PoolConfiguration(false, 0, 1, 1, getWaitToDie(), WhenBlockedPolicy.RUN, 1));
     }
 
     /**

commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/PooledCacheEventQueue.java

@@ -24,7 +24,6 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.jcs4.engine.behavior.ICacheListener;
 import org.apache.commons.jcs4.engine.stats.Stats;
@@ -51,6 +50,9 @@ public class PooledCacheEventQueue<K, V>
     /** The Thread Pool to execute events with. */
     protected ExecutorService pool;
 
+    /** The Thread Pool name in ThreadPoolManager. */
+    protected String poolName;
+
     /** The Thread Pool queue */
     protected BlockingQueue<Runnable> queue;
 
@@ -73,14 +75,12 @@ public PooledCacheEventQueue( final ICacheListener<K, V> listener, final long li
     /**
      * Create the thread pool.
      *
-     * @param threadPoolName
      * @since 3.1
      */
-    protected ExecutorService createPool(final String threadPoolName)
+    protected ExecutorService createPool()
     {
         // this will share the same pool with other event queues by default.
-        return ThreadPoolManager.getInstance().getExecutorService(
-                threadPoolName == null ? "cache_event_queue" : threadPoolName );
+        return ThreadPoolManager.getInstance().getExecutorService(poolName);
     }
 
     /**
@@ -94,23 +94,7 @@ public synchronized void destroy(final Duration wait)
         if ( isWorking() )
         {
             setWorking(false);
-            pool.shutdown();
-
-            if (wait.toSeconds() > 0)
-            {
-                try
-                {
-                    if (!pool.awaitTermination(wait.toSeconds(), TimeUnit.SECONDS))
-                    {
-                        log.info( "No longer waiting for event queue to finish: {0}",
-                                this::getStatistics);
-                    }
-                }
-                catch (final InterruptedException e)
-                {
-                    // ignore
-                }
-            }
+            ThreadPoolManager.getInstance().disposeExecutorService(poolName, wait);
             log.info( "Cache event queue destroyed: {0}", this );
         }
     }
@@ -160,7 +144,8 @@ protected void initialize( final ICacheListener<K, V> listener, final long liste
     {
         super.initialize(listener, listenerId, cacheName, maxFailure, waitBeforeRetry);
 
-        pool = createPool(threadPoolName);
+        poolName = threadPoolName == null ? "cache_event_queue" : threadPoolName;
+        pool = createPool();
 
         if (pool instanceof ThreadPoolExecutor tpe)
         {

commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/event/ElementEventQueue.java

@@ -38,7 +38,7 @@
 public class ElementEventQueue
     implements IElementEventQueue
 {
-    private static final String THREAD_PREFIX = "JCS-ElementEventQueue-";
+    protected static final String POOL_NAME = "ElementEventQueue";
 
     /** The logger */
     private static final Log log = Log.getLog( ElementEventQueue.class );
@@ -54,8 +54,8 @@ public class ElementEventQueue
      */
     public ElementEventQueue()
     {
-        queueProcessor = ThreadPoolManager.getInstance().createPool(
-        		new PoolConfiguration(false, 0, 1, 1, Duration.ZERO, WhenBlockedPolicy.RUN, 1), THREAD_PREFIX);
+        queueProcessor = ThreadPoolManager.getInstance().getExecutorService(POOL_NAME,
+        		new PoolConfiguration(false, 0, 1, 1, Duration.ZERO, WhenBlockedPolicy.RUN, 1));
 
         log.debug( "Constructed: {0}", this );
     }
@@ -91,8 +91,7 @@ public void dispose()
     {
         if (destroyed.compareAndSet(false, true))
         {
-            // Pool will be shut down by the ThreadPoolManager
-            // queueProcessor.shutdownNow();
+            ThreadPoolManager.getInstance().disposeExecutorService(POOL_NAME);
             log.info( "Element event queue destroyed: {0}", this );
         }
     }

commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/discovery/UDPDiscoveryReceiver.java

@@ -105,10 +105,9 @@ public UDPDiscoveryReceiver( final Consumer<UDPDiscoveryMessage> service,
         setService(service);
 
         // create a small thread pool to handle a barrage
-        this.pooledExecutor = ThreadPoolManager.getInstance().createPool(
+        this.pooledExecutor = ThreadPoolManager.getInstance().getExecutorService("UDPDiscoveryReceiver",
                 new PoolConfiguration(false, 0, maxPoolSize, maxPoolSize, Duration.ZERO,
-                        WhenBlockedPolicy.DISCARDOLDEST, maxPoolSize),
-                "JCS-UDPDiscoveryReceiver-", Thread.MIN_PRIORITY);
+                        WhenBlockedPolicy.DISCARDOLDEST, maxPoolSize, Thread.MIN_PRIORITY));
 
         log.info( "Constructing listener, [{0}:{1}]", multicastAddress, multicastPort );
         createSocket( multicastInterfaceString, multicastAddress, multicastPort );

commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/threadpool/PoolConfiguration.java

@@ -47,7 +47,10 @@ public record PoolConfiguration(
         WhenBlockedPolicy whenBlockedPolicy,
 
         /** The number of threads to create on startup */
-        int startUpSize
+        int startUpSize,
+
+        /** The thread priority */
+        int threadPriority
 ) implements Cloneable
 {
     public enum WhenBlockedPolicy
@@ -88,12 +91,15 @@ public enum WhenBlockedPolicy
     /** Default startup size */
     private static final int DEFAULT_STARTUP_SIZE = DEFAULT_MINIMUM_POOL_SIZE;
 
+    /** Default thread priority */
+    private static final int DEFAULT_THREAD_PRIORITY = Thread.NORM_PRIORITY;
+
     /**
      * Default
      */
     private static PoolConfiguration DEFAULT = new PoolConfiguration(DEFAULT_USE_BOUNDARY,
             DEFAULT_BOUNDARY_SIZE, DEFAULT_MAXIMUM_POOL_SIZE, DEFAULT_MINIMUM_POOL_SIZE,
-            DEFAULT_KEEPALIVE_TIME, DEFAULT_WHEN_BLOCKED_POLICY, DEFAULT_STARTUP_SIZE);
+            DEFAULT_KEEPALIVE_TIME, DEFAULT_WHEN_BLOCKED_POLICY, DEFAULT_STARTUP_SIZE, DEFAULT_THREAD_PRIORITY);
 
     /**
      * @return An object containing the default settings
@@ -103,6 +109,22 @@ public static PoolConfiguration defaults()
         return DEFAULT;
     }
 
+    /**
+     * Convenience constructor
+     *
+     * @param useBoundary Should we bound the queue
+     * @param boundarySize If the queue is bounded, how big can it get
+     * @param maximumPoolSize Only has meaning if a boundary is used
+     * @param minimumPoolSize the exact number that will be used in a boundless queue
+     * @param keepAliveTime How long idle threads above the minimum should be kept alive
+     * @param whenBlockedPolicy Should be ABORT, BLOCK, RUN, WAIT, DISCARDOLDEST
+     * @param startUpSize The number of threads to create on startup
+     */
+    public PoolConfiguration(boolean useBoundary, int boundarySize, int maximumPoolSize, int minimumPoolSize, Duration keepAliveTime,
+            WhenBlockedPolicy whenBlockedPolicy, int startUpSize)
+    {
+        this(useBoundary, boundarySize, maximumPoolSize, minimumPoolSize, keepAliveTime, whenBlockedPolicy, startUpSize, DEFAULT_THREAD_PRIORITY);
+    }
 
     /**
      * To string for debugging purposes.
@@ -118,7 +140,8 @@ public String toString()
         buf.append("minimumPoolSize = [").append(minimumPoolSize()).append("] ");
         buf.append("keepAliveTime = [").append(keepAliveTime()).append("] ");
         buf.append("whenBlockedPolicy = [").append(whenBlockedPolicy()).append("] ");
-        buf.append("startUpSize = [").append(startUpSize()).append("]" );
+        buf.append("startUpSize = [").append(startUpSize()).append("] " );
+        buf.append("threadPriority = [").append(threadPriority()).append("]" );
         return buf.toString();
     }
 }