You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@commons.apache.org by tv...@apache.org on 2026/08/27 11:33:22 UTC
(commons-jcs) 02/02: Second round of thread pool lifecycle fixes. JCS-248
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/commons-jcs.git
commit 0de0497adf13aa68a4294a85cee73706cd7356a0
Author: Thomas Vandahl <tv...@apache.org>
AuthorDate: Thu Aug 27 13:33:15 2026 +0200
Second round of thread pool lifecycle fixes. JCS-248
---
.../jcs4/engine/control/CompositeCache.java | 7 -
.../jcs4/engine/control/CompositeCacheManager.java | 2 +-
.../engine/control/event/ElementEventQueue.java | 1 -
.../jcs4/utils/threadpool/ThreadPoolManager.java | 166 +++++++++++++--------
.../control/event/ElementEventQueueUnitTest.java | 87 -----------
.../threadpool/ThreadPoolManagerUnitTest.java | 44 ++++++
6 files changed, 148 insertions(+), 159 deletions(-)
diff --git a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/CompositeCache.java b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/CompositeCache.java
index bfa90bfc..e99b7947 100644
--- a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/CompositeCache.java
+++ b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/CompositeCache.java
@@ -245,13 +245,6 @@ public class CompositeCache<K, V>
future.cancel(true);
}
- // Now, shut down the event queue
- if (elementEventQ != null)
- {
- elementEventQ.dispose();
- elementEventQ = null;
- }
-
// Dispose of each auxiliary cache, Remote auxiliaries will be
// skipped if 'fromRemote' is true.
for (final ICache<K, V> aux : auxCaches)
diff --git a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/CompositeCacheManager.java b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/CompositeCacheManager.java
index 3626716b..85498462 100644
--- a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/CompositeCacheManager.java
+++ b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/CompositeCacheManager.java
@@ -847,7 +847,7 @@ public class CompositeCacheManager
this.scheduledExecutor.shutdownNow();
// shutdown all thread pools
- ThreadPoolManager.dispose();
+ ThreadPoolManager.getInstance().dispose();
if (shutdownHook != null)
{
diff --git a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/event/ElementEventQueue.java b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/event/ElementEventQueue.java
index cffc420f..29432870 100644
--- a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/event/ElementEventQueue.java
+++ b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/engine/control/event/ElementEventQueue.java
@@ -91,7 +91,6 @@ public class ElementEventQueue
{
if (destroyed.compareAndSet(false, true))
{
- ThreadPoolManager.getInstance().disposeExecutorService(POOL_NAME);
log.info( "Element event queue destroyed: {0}", this );
}
}
diff --git a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/threadpool/ThreadPoolManager.java b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/threadpool/ThreadPoolManager.java
index b9e807ea..3be4c30b 100644
--- a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/threadpool/ThreadPoolManager.java
+++ b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/threadpool/ThreadPoolManager.java
@@ -33,6 +33,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.jcs4.log.Log;
import org.apache.commons.jcs4.utils.config.ConfigurationBuilder;
@@ -112,10 +113,10 @@ public class ThreadPoolManager
/**
* Dispose of the instance of the ThreadPoolManger and shut down all thread pools
*/
- public static void dispose()
+ public synchronized void dispose()
{
- for ( final Iterator<Map.Entry<String, ExecutorService>> i =
- getInstance().pools.entrySet().iterator(); i.hasNext(); )
+ for (final Iterator<Map.Entry<String, ExecutorService>> i =
+ pools.entrySet().iterator(); i.hasNext();)
{
final Map.Entry<String, ExecutorService> entry = i.next();
try
@@ -127,10 +128,11 @@ public class ThreadPoolManager
log.warn("Failed to close pool {0}", entry.getKey(), t);
}
i.remove();
+ poolUseCounts.remove(entry.getKey());
}
- for ( final Iterator<Map.Entry<String, ScheduledExecutorService>> i =
- getInstance().schedulerPools.entrySet().iterator(); i.hasNext(); )
+ for (final Iterator<Map.Entry<String, ScheduledExecutorService>> i =
+ schedulerPools.entrySet().iterator(); i.hasNext();)
{
final Map.Entry<String, ScheduledExecutorService> entry = i.next();
try
@@ -142,6 +144,7 @@ public class ThreadPoolManager
log.warn("Failed to close pool {0}", entry.getKey(), t);
}
i.remove();
+ schedulerPoolUseCounts.remove(entry.getKey());
}
}
@@ -150,7 +153,7 @@ public class ThreadPoolManager
*
* @param poolName the name of the pool
*/
- public void disposeExecutorService(String poolName)
+ public synchronized void disposeExecutorService(String poolName)
{
disposeExecutorService(poolName, Duration.ZERO);
}
@@ -161,40 +164,45 @@ public class ThreadPoolManager
* @param poolName the name of the pool
* @param wait Duration to wait for termination
*/
- public void disposeExecutorService(String poolName, Duration wait)
+ public synchronized void disposeExecutorService(String poolName, Duration wait)
{
- ExecutorService pool = pools.remove(poolName);
- if (pool == null)
+ AtomicInteger useCount = poolUseCounts.computeIfAbsent(poolName, k -> new AtomicInteger());
+ if (useCount.decrementAndGet() == 0)
{
- log.warn("Failed to close non-existing pool {0}", poolName);
- }
- else
- {
- try
+ poolUseCounts.remove(poolName, useCount);
+ ExecutorService pool = pools.remove(poolName);
+ if (pool == null)
{
- if (wait == null || wait.isZero())
- {
- pool.shutdownNow();
- }
- else
+ log.warn("Failed to close non-existing pool {0}", poolName);
+ }
+ else
+ {
+ try
{
- pool.shutdown();
- try
+ if (wait == null || wait.isZero())
{
- if (!pool.awaitTermination(wait.toSeconds(), TimeUnit.SECONDS))
- {
- log.info( "No longer waiting for pool {0} to terminate", poolName);
- }
+ pool.shutdownNow();
}
- catch (final InterruptedException e)
+ else
{
- // ignore
+ pool.shutdown();
+ try
+ {
+ if (!pool.awaitTermination(wait.toSeconds(), TimeUnit.SECONDS))
+ {
+ log.info( "No longer waiting for pool {0} to terminate", poolName);
+ }
+ }
+ catch (final InterruptedException e)
+ {
+ // ignore
+ }
}
}
- }
- catch (final Throwable t)
- {
- log.warn("Failed to close pool {0}", poolName, t);
+ catch (final Throwable t)
+ {
+ log.warn("Failed to close pool {0}", poolName, t);
+ }
}
}
}
@@ -204,7 +212,7 @@ public class ThreadPoolManager
*
* @param poolName the name of the pool
*/
- public void disposeSchedulerPool(String poolName)
+ public synchronized void disposeSchedulerPool(String poolName)
{
disposeSchedulerPool(poolName, Duration.ZERO);
}
@@ -215,40 +223,45 @@ public class ThreadPoolManager
* @param poolName the name of the pool
* @param wait Duration to wait for termination
*/
- public void disposeSchedulerPool(String poolName, Duration wait)
+ public synchronized void disposeSchedulerPool(String poolName, Duration wait)
{
- ExecutorService pool = schedulerPools.remove(poolName);
- if (pool == null)
+ AtomicInteger useCount = schedulerPoolUseCounts.computeIfAbsent(poolName, k -> new AtomicInteger());
+ if (useCount.decrementAndGet() == 0)
{
- log.warn("Failed to close non-existing pool {0}", poolName);
- }
- else
- {
- try
+ schedulerPoolUseCounts.remove(poolName, useCount);
+ ExecutorService pool = schedulerPools.remove(poolName);
+ if (pool == null)
{
- if (wait == null || wait.isZero())
- {
- pool.shutdownNow();
- }
- else
+ log.warn("Failed to close non-existing pool {0}", poolName);
+ }
+ else
+ {
+ try
{
- pool.shutdown();
- try
+ if (wait == null || wait.isZero())
{
- if (!pool.awaitTermination(wait.toSeconds(), TimeUnit.SECONDS))
- {
- log.info( "No longer waiting for pool {0} to terminate", poolName);
- }
+ pool.shutdownNow();
}
- catch (final InterruptedException e)
+ else
{
- // ignore
+ pool.shutdown();
+ try
+ {
+ if (!pool.awaitTermination(wait.toMillis(), TimeUnit.MILLISECONDS))
+ {
+ log.info( "No longer waiting for pool {0} to terminate", poolName);
+ }
+ }
+ catch (final InterruptedException e)
+ {
+ // ignore
+ }
}
}
- }
- catch (final Throwable t)
- {
- log.warn("Failed to close pool {0}", poolName, t);
+ catch (final Throwable t)
+ {
+ log.warn("Failed to close pool {0}", poolName, t);
+ }
}
}
}
@@ -306,6 +319,12 @@ public class ThreadPoolManager
/** Map of names to scheduler pools. */
private final ConcurrentHashMap<String, ScheduledExecutorService> schedulerPools;
+ /** Map of names to pool use counts. */
+ private final ConcurrentHashMap<String, AtomicInteger> poolUseCounts;
+
+ /** Map of names to scheduler pool use counts. */
+ private final ConcurrentHashMap<String, AtomicInteger> schedulerPoolUseCounts;
+
/**
* No instances please. This is a singleton.
*/
@@ -313,6 +332,8 @@ public class ThreadPoolManager
{
this.pools = new ConcurrentHashMap<>();
this.schedulerPools = new ConcurrentHashMap<>();
+ this.poolUseCounts = new ConcurrentHashMap<>();
+ this.schedulerPoolUseCounts = new ConcurrentHashMap<>();
configure();
}
@@ -411,9 +432,9 @@ public class ThreadPoolManager
* @param name
* @return The executor service configured for the name.
*/
- public ExecutorService getExecutorService( final String name )
+ public synchronized ExecutorService getExecutorService( final String name )
{
- return getExecutorService(name, loadConfig( PROP_NAME_ROOT + "." + name, defaultConfig ));
+ return getExecutorService(name, loadConfig(PROP_NAME_ROOT + "." + name, defaultConfig));
}
/**
@@ -426,8 +447,11 @@ public class ThreadPoolManager
* @param config The pool configuration
* @return The executor service configured for the name.
*/
- public ExecutorService getExecutorService(final String name, final PoolConfiguration config)
+ public synchronized ExecutorService getExecutorService(final String name, final PoolConfiguration config)
{
+ AtomicInteger useCount = poolUseCounts.computeIfAbsent(name, k -> new AtomicInteger());
+ useCount.getAndIncrement();
+
return pools.computeIfAbsent(name, key -> {
log.debug("Creating pool for name [{0}]", key);
return createPool(config, JCS_THREAD_POOL_MANAGER_PREFIX + key + "-");
@@ -453,12 +477,28 @@ public class ThreadPoolManager
* @param name
* @return The scheduler pool configured for the name.
*/
- public ScheduledExecutorService getSchedulerPool( final String name )
+ public synchronized ScheduledExecutorService getSchedulerPool(final String name)
{
+ return getSchedulerPool(name, loadConfig(PROP_NAME_SCHEDULER_ROOT + "." + name, defaultSchedulerConfig));
+ }
+
+ /**
+ * Returns a scheduler pool by name. If a pool by this name does not exist in the configuration file or
+ * properties, one will be created using the named configuration.
+ * <p>
+ * Pools are lazily created.
+ *
+ * @param name
+ * @param config The pool configuration
+ * @return The scheduler pool configured for the name.
+ */
+ public synchronized ScheduledExecutorService getSchedulerPool(final String name, PoolConfiguration config)
+ {
+ AtomicInteger useCount = schedulerPoolUseCounts.computeIfAbsent(name, k -> new AtomicInteger());
+ useCount.getAndIncrement();
+
return schedulerPools.computeIfAbsent(name, key -> {
log.debug( "Creating scheduler pool for name [{0}]", key );
- final PoolConfiguration config = loadConfig( PROP_NAME_SCHEDULER_ROOT + "." + key,
- defaultSchedulerConfig );
return createSchedulerPool( config, JCS_THREAD_POOL_MANAGER_PREFIX + key + "-");
});
}
diff --git a/commons-jcs4-core/src/test/java/org/apache/commons/jcs4/engine/control/event/ElementEventQueueUnitTest.java b/commons-jcs4-core/src/test/java/org/apache/commons/jcs4/engine/control/event/ElementEventQueueUnitTest.java
deleted file mode 100644
index 96ca2d40..00000000
--- a/commons-jcs4-core/src/test/java/org/apache/commons/jcs4/engine/control/event/ElementEventQueueUnitTest.java
+++ /dev/null
@@ -1,87 +0,0 @@
-package org.apache.commons.jcs4.engine.control.event;
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-
-import java.util.HashSet;
-import java.util.Set;
-
-import org.apache.commons.jcs4.utils.threadpool.ThreadPoolManager;
-import org.junit.jupiter.api.Test;
-
-/** Tests the lifecycle of the element-event worker owned by the queue. */
-class ElementEventQueueUnitTest
-{
- private static final String THREAD_PREFIX = ThreadPoolManager.JCS_THREAD_POOL_MANAGER_PREFIX + ElementEventQueue.POOL_NAME;
-
- @Test
- void testDisposeStopsOwnedWorkerThread()
- throws InterruptedException
- {
- final Set<Long> threadsBefore = eventQueueThreadIds();
- final ElementEventQueue queue = new ElementEventQueue();
- final Thread worker = waitForNewWorker( threadsBefore );
-
- assertNotNull( worker, "The element-event queue did not start its worker" );
-
- queue.dispose();
- worker.join( 2000 );
-
- assertFalse( worker.isAlive(), "The element-event worker is still alive after dispose" );
-
- // Disposal is a lifecycle operation and must be safe when invoked more than once.
- queue.dispose();
- }
-
- private static Set<Long> eventQueueThreadIds()
- {
- final Set<Long> result = new HashSet<>();
- for ( final Thread thread : Thread.getAllStackTraces().keySet() )
- {
- if (thread.getName().startsWith(THREAD_PREFIX))
- {
- result.add( thread.getId() );
- }
- }
- return result;
- }
-
- private static Thread waitForNewWorker( final Set<Long> threadsBefore )
- throws InterruptedException
- {
- final long deadline = System.currentTimeMillis() + 2000;
- do
- {
- for ( final Thread thread : Thread.getAllStackTraces().keySet() )
- {
- if (thread.getName().startsWith(THREAD_PREFIX) && !threadsBefore.contains(thread.getId()))
- {
- return thread;
- }
- }
- Thread.sleep( 10 );
- }
- while (System.currentTimeMillis() < deadline);
-
- return null;
- }
-}
\ No newline at end of file
diff --git a/commons-jcs4-core/src/test/java/org/apache/commons/jcs4/utils/threadpool/ThreadPoolManagerUnitTest.java b/commons-jcs4-core/src/test/java/org/apache/commons/jcs4/utils/threadpool/ThreadPoolManagerUnitTest.java
index 0d022e1a..091550ba 100644
--- a/commons-jcs4-core/src/test/java/org/apache/commons/jcs4/utils/threadpool/ThreadPoolManagerUnitTest.java
+++ b/commons-jcs4-core/src/test/java/org/apache/commons/jcs4/utils/threadpool/ThreadPoolManagerUnitTest.java
@@ -1,5 +1,7 @@
package org.apache.commons.jcs4.utils.threadpool;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -27,6 +29,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import org.apache.commons.jcs4.utils.props.PropertyLoader;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
/**
@@ -35,6 +38,11 @@ import org.junit.jupiter.api.Test;
*/
class ThreadPoolManagerUnitTest
{
+ @AfterEach
+ void dispose()
+ {
+ ThreadPoolManager.getInstance().dispose();
+ }
/**
* Make sure it can load a default cache.ccf file
@@ -85,4 +93,40 @@ class ThreadPoolManagerUnitTest
final ExecutorService pool = mgr.getExecutorService( "aborttest" );
assertNotNull( pool );
}
+
+ /**
+ * Test correct lifecycle
+ * @throws InterruptedException
+ */
+ @Test
+ void testLifecycle() throws InterruptedException
+ {
+ final ThreadPoolManager mgr = ThreadPoolManager.getInstance();
+ assertNotNull( mgr );
+
+ final String poolName1 = "testGetPoolNames1";
+ mgr.getExecutorService( poolName1 );
+
+ final String poolName2 = "testGetPoolNames2";
+ mgr.getExecutorService( poolName2 );
+
+ // simulate shared pool
+ mgr.getExecutorService( poolName2 );
+
+ mgr.disposeExecutorService(poolName1);
+ mgr.disposeExecutorService(poolName2);
+
+ final Set<String> names = mgr.getPoolNames();
+ assertFalse( names.contains( poolName1 ), "Should not have poolName1 in list." );
+ assertTrue( names.contains( poolName2 ), "Should still have poolName2 in list." );
+
+ mgr.disposeExecutorService(poolName2);
+ assertFalse( names.contains( poolName2 ), "Should not have poolName2 in list." );
+
+ Thread.sleep(1000);
+ for ( final Thread thread : Thread.getAllStackTraces().keySet() )
+ {
+ assertFalse(thread.getName().startsWith(ThreadPoolManager.JCS_THREAD_POOL_MANAGER_PREFIX), "No ThreadPoolManager threads should be left " + thread.getName());
+ }
+ }
}