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 <[email protected]> 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()); + } + } }
