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 ee410efe4a15b777ad28f5e43d282c2e9363edff Author: Thomas Vandahl <[email protected]> AuthorDate: Tue Sep 1 13:56:20 2026 +0200 Improve thread pool handling --- .../remote/AbstractRemoteAuxiliaryCache.java | 6 + .../jcs4/engine/control/CompositeCacheManager.java | 9 +- .../engine/control/event/ElementEventQueue.java | 25 +- .../jcs4/utils/discovery/UDPDiscoveryReceiver.java | 8 +- .../jcs4/utils/threadpool/ThreadPoolManager.java | 398 ++++++++++++--------- 5 files changed, 251 insertions(+), 195 deletions(-) diff --git a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/auxiliary/remote/AbstractRemoteAuxiliaryCache.java b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/auxiliary/remote/AbstractRemoteAuxiliaryCache.java index 392a7e14..387acec5 100644 --- a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/auxiliary/remote/AbstractRemoteAuxiliaryCache.java +++ b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/auxiliary/remote/AbstractRemoteAuxiliaryCache.java @@ -356,6 +356,12 @@ public abstract class AbstractRemoteAuxiliaryCache<K, V> { getRemoteCacheListener().dispose(); } + + if (usePoolForGet) + { + usePoolForGet = false; + ThreadPoolManager.getInstance().disposeExecutorService(getAuxiliaryCacheAttributes().getThreadPoolName()); + } } catch ( final IOException ex ) { 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 85498462..3c387566 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 @@ -28,7 +28,6 @@ import java.util.Properties; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicInteger; @@ -56,7 +55,6 @@ import org.apache.commons.jcs4.engine.stats.CacheStats; import org.apache.commons.jcs4.engine.stats.behavior.ICacheStats; import org.apache.commons.jcs4.log.Log; import org.apache.commons.jcs4.utils.config.OptionConverter; -import org.apache.commons.jcs4.utils.threadpool.DaemonThreadFactory; import org.apache.commons.jcs4.utils.threadpool.ThreadPoolManager; import org.apache.commons.jcs4.utils.timing.ElapsedTimer; @@ -382,6 +380,8 @@ public class CompositeCacheManager final ThreadPoolManager poolMgr = ThreadPoolManager.getInstance(); log.debug( "ThreadPoolManager = {0}", poolMgr); + this.scheduledExecutor = ThreadPoolManager.getInstance().getSchedulerPool("default"); + // Create event queue this.elementEventQueue = new ElementEventQueue(); @@ -666,9 +666,6 @@ public class CompositeCacheManager log.error( "Could not register shutdown hook.", e ); } - this.scheduledExecutor = Executors.newScheduledThreadPool(4, - new DaemonThreadFactory("JCS-Scheduler-", Thread.MIN_PRIORITY)); - // Register JMX bean if (!isJMXRegistered && jmxName != null) { @@ -844,7 +841,7 @@ public class CompositeCacheManager auxiliaryFactoryRegistry.clear(); // shutdown all scheduled jobs - this.scheduledExecutor.shutdownNow(); + ThreadPoolManager.getInstance().disposeSchedulerPool("default"); // shutdown all thread pools ThreadPoolManager.getInstance().dispose(); 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 29432870..a95e4351 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 @@ -22,7 +22,7 @@ package org.apache.commons.jcs4.engine.control.event; import java.io.IOException; import java.time.Duration; import java.util.concurrent.ExecutorService; -import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.RejectedExecutionException; import org.apache.commons.jcs4.engine.control.event.behavior.IElementEvent; import org.apache.commons.jcs4.engine.control.event.behavior.IElementEventHandler; @@ -43,9 +43,6 @@ public class ElementEventQueue /** The logger */ private static final Log log = Log.getLog( ElementEventQueue.class ); - /** Shutdown or not */ - private final AtomicBoolean destroyed = new AtomicBoolean(); - /** The worker thread pool. */ private final ExecutorService queueProcessor; @@ -71,15 +68,23 @@ public class ElementEventQueue throws IOException { - log.debug("Adding Event Handler to QUEUE, !destroyed = {0}", !destroyed.get()); + log.debug("Adding Event Handler to QUEUE"); - if (destroyed.get()) + if (queueProcessor.isShutdown()) { log.warn("Event submitted to disposed element event queue {0}", event); } else { - queueProcessor.execute(() -> hand.handleElementEvent(event)); + try + { + queueProcessor.execute(() -> hand.handleElementEvent(event)); + } + catch (RejectedExecutionException e) + { + log.warn("Event execution rejected {0}", event, e); + } + } } @@ -89,9 +94,7 @@ public class ElementEventQueue @Override public void dispose() { - if (destroyed.compareAndSet(false, true)) - { - log.info( "Element event queue destroyed: {0}", this ); - } + 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/discovery/UDPDiscoveryReceiver.java b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/discovery/UDPDiscoveryReceiver.java index eb217107..4894c280 100644 --- a/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/discovery/UDPDiscoveryReceiver.java +++ b/commons-jcs4-core/src/main/java/org/apache/commons/jcs4/utils/discovery/UDPDiscoveryReceiver.java @@ -74,6 +74,9 @@ public class UDPDiscoveryReceiver /** The processor */ private final ExecutorService pooledExecutor; + /** The processor name */ + private final String poolName; + /** Number of messages received. For debugging and testing. */ private final AtomicInteger cnt = new AtomicInteger(); @@ -104,8 +107,10 @@ public class UDPDiscoveryReceiver { setService(service); + this.poolName = "UDPDiscoveryReceiver-" + multicastAddress.getHostAddress() + + (multicastInterfaceString == null ? "" : "@" + multicastInterfaceString) + ":" + multicastPort; // create a small thread pool to handle a barrage - this.pooledExecutor = ThreadPoolManager.getInstance().getExecutorService("UDPDiscoveryReceiver", + this.pooledExecutor = ThreadPoolManager.getInstance().getExecutorService(poolName, new PoolConfiguration(false, 0, maxPoolSize, maxPoolSize, Duration.ZERO, WhenBlockedPolicy.DISCARDOLDEST, maxPoolSize, Thread.MIN_PRIORITY)); @@ -333,6 +338,7 @@ public class UDPDiscoveryReceiver selector.close(); multicastGroupKey.drop(); multicastChannel.close(); + ThreadPoolManager.getInstance().disposeExecutorService(this.poolName); } catch ( final IOException e ) { 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 3be4c30b..63d4fb4a 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 @@ -37,6 +37,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.jcs4.log.Log; import org.apache.commons.jcs4.utils.config.ConfigurationBuilder; +import org.apache.commons.jcs4.utils.threadpool.PoolConfiguration.WhenBlockedPolicy; /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file @@ -110,162 +111,6 @@ public class ThreadPoolManager */ private static volatile Properties props; - /** - * Dispose of the instance of the ThreadPoolManger and shut down all thread pools - */ - public synchronized void dispose() - { - for (final Iterator<Map.Entry<String, ExecutorService>> i = - pools.entrySet().iterator(); i.hasNext();) - { - final Map.Entry<String, ExecutorService> entry = i.next(); - try - { - entry.getValue().shutdownNow(); - } - catch (final Throwable t) - { - log.warn("Failed to close pool {0}", entry.getKey(), t); - } - i.remove(); - poolUseCounts.remove(entry.getKey()); - } - - for (final Iterator<Map.Entry<String, ScheduledExecutorService>> i = - schedulerPools.entrySet().iterator(); i.hasNext();) - { - final Map.Entry<String, ScheduledExecutorService> entry = i.next(); - try - { - entry.getValue().shutdownNow(); - } - catch (final Throwable t) - { - log.warn("Failed to close pool {0}", entry.getKey(), t); - } - i.remove(); - schedulerPoolUseCounts.remove(entry.getKey()); - } - } - - /** - * Dispose of a thread pool - * - * @param poolName the name of the pool - */ - public synchronized void disposeExecutorService(String poolName) - { - disposeExecutorService(poolName, Duration.ZERO); - } - - /** - * Dispose of a thread pool - * - * @param poolName the name of the pool - * @param wait Duration to wait for termination - */ - public synchronized void disposeExecutorService(String poolName, Duration wait) - { - AtomicInteger useCount = poolUseCounts.computeIfAbsent(poolName, k -> new AtomicInteger()); - if (useCount.decrementAndGet() == 0) - { - poolUseCounts.remove(poolName, useCount); - ExecutorService pool = pools.remove(poolName); - if (pool == null) - { - log.warn("Failed to close non-existing pool {0}", poolName); - } - else - { - try - { - if (wait == null || wait.isZero()) - { - pool.shutdownNow(); - } - else - { - 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); - } - } - } - } - - /** - * Dispose of a scheduler thread pool - * - * @param poolName the name of the pool - */ - public synchronized void disposeSchedulerPool(String poolName) - { - disposeSchedulerPool(poolName, Duration.ZERO); - } - - /** - * Dispose of a scheduler thread pool - * - * @param poolName the name of the pool - * @param wait Duration to wait for termination - */ - public synchronized void disposeSchedulerPool(String poolName, Duration wait) - { - AtomicInteger useCount = schedulerPoolUseCounts.computeIfAbsent(poolName, k -> new AtomicInteger()); - if (useCount.decrementAndGet() == 0) - { - schedulerPoolUseCounts.remove(poolName, useCount); - ExecutorService pool = schedulerPools.remove(poolName); - if (pool == null) - { - log.warn("Failed to close non-existing pool {0}", poolName); - } - else - { - try - { - if (wait == null || wait.isZero()) - { - pool.shutdownNow(); - } - else - { - 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); - } - } - } - } - /** * Returns a configured instance of the ThreadPoolManger To specify a configuration file or * Properties object to use call the appropriate setter prior to calling getInstance. @@ -284,14 +129,14 @@ public class ThreadPoolManager * @param defaultPoolConfiguration The default configuration * @return PoolConfiguration */ - private static PoolConfiguration loadConfig( final String root, final PoolConfiguration defaultPoolConfiguration ) + private static PoolConfiguration loadConfig(final String root, final PoolConfiguration defaultPoolConfiguration) { final PoolConfiguration config = ConfigurationBuilder .create(PoolConfiguration.class, defaultPoolConfiguration) .fromProperties(props, root) .build(); - log.debug( "{0} PoolConfiguration = {1}", root, config ); + log.debug("{0} PoolConfiguration = {1}", root, config); return config; } @@ -302,7 +147,7 @@ public class ThreadPoolManager * * @param props The props to set. */ - public static void setProps( final Properties props ) + public static void setProps(final Properties props) { ThreadPoolManager.props = props; } @@ -352,7 +197,8 @@ public class ThreadPoolManager // set initial default and then override if new settings are available defaultConfig = loadConfig(DEFAULT_PROP_NAME_ROOT, PoolConfiguration.defaults()); - defaultSchedulerConfig = loadConfig(DEFAULT_PROP_NAME_SCHEDULER_ROOT, PoolConfiguration.defaults()); + defaultSchedulerConfig = loadConfig(DEFAULT_PROP_NAME_SCHEDULER_ROOT, + new PoolConfiguration(false, 0, 4, 4, Duration.ZERO, WhenBlockedPolicy.DISCARDOLDEST, 4, Thread.MIN_PRIORITY)); } /** @@ -362,7 +208,7 @@ public class ThreadPoolManager * @param threadNamePrefix prefix for the thread names of the pool * @return A ThreadPool wrapper */ - private ExecutorService createPool( final PoolConfiguration config, final String threadNamePrefix) + private ExecutorService createPool(final PoolConfiguration config, final String threadNamePrefix) { BlockingQueue<Runnable> queue = null; if ( config.useBoundary() ) @@ -432,7 +278,7 @@ public class ThreadPoolManager * @param name * @return The executor service configured for the name. */ - public synchronized ExecutorService getExecutorService( final String name ) + public ExecutorService getExecutorService(final String name) { return getExecutorService(name, loadConfig(PROP_NAME_ROOT + "." + name, defaultConfig)); } @@ -447,15 +293,20 @@ public class ThreadPoolManager * @param config The pool configuration * @return The executor service configured for the name. */ - public synchronized ExecutorService getExecutorService(final String name, final PoolConfiguration config) + public ExecutorService getExecutorService(final String name, final PoolConfiguration config) { - AtomicInteger useCount = poolUseCounts.computeIfAbsent(name, k -> new AtomicInteger()); - useCount.getAndIncrement(); + synchronized (pools) + { + ExecutorService pool = pools.computeIfAbsent(name, key -> { + log.debug("Creating pool for name [{0}]", key); + return createPool(config, JCS_THREAD_POOL_MANAGER_PREFIX + key + "-"); + }); - return pools.computeIfAbsent(name, key -> { - log.debug("Creating pool for name [{0}]", key); - return createPool(config, JCS_THREAD_POOL_MANAGER_PREFIX + key + "-"); - }); + AtomicInteger useCount = poolUseCounts.computeIfAbsent(name, k -> new AtomicInteger()); + useCount.getAndIncrement(); + + return pool; + } } /** @@ -477,7 +328,7 @@ public class ThreadPoolManager * @param name * @return The scheduler pool configured for the name. */ - public synchronized ScheduledExecutorService getSchedulerPool(final String name) + public ScheduledExecutorService getSchedulerPool(final String name) { return getSchedulerPool(name, loadConfig(PROP_NAME_SCHEDULER_ROOT + "." + name, defaultSchedulerConfig)); } @@ -492,14 +343,207 @@ public class ThreadPoolManager * @param config The pool configuration * @return The scheduler pool configured for the name. */ - public synchronized ScheduledExecutorService getSchedulerPool(final String name, PoolConfiguration config) + public ScheduledExecutorService getSchedulerPool(final String name, PoolConfiguration config) + { + synchronized (schedulerPools) + { + ScheduledExecutorService pool = schedulerPools.computeIfAbsent(name, key -> { + log.debug( "Creating scheduler pool for name [{0}]", key ); + return createSchedulerPool(config, JCS_THREAD_POOL_MANAGER_PREFIX + key + "-"); + }); + + AtomicInteger useCount = schedulerPoolUseCounts.computeIfAbsent(name, k -> new AtomicInteger()); + useCount.getAndIncrement(); + + return pool; + } + } + + /** + * Dispose of the instance of the ThreadPoolManger and shut down all thread pools + */ + public void dispose() + { + synchronized (pools) + { + for (final Iterator<Map.Entry<String, ExecutorService>> i = + pools.entrySet().iterator(); i.hasNext();) + { + final Map.Entry<String, ExecutorService> entry = i.next(); + try + { + entry.getValue().shutdownNow(); + } + catch (final Throwable t) + { + log.warn("Failed to close pool {0}", entry.getKey(), t); + } + i.remove(); + poolUseCounts.remove(entry.getKey()); + } + } + + synchronized (schedulerPools) + { + for (final Iterator<Map.Entry<String, ScheduledExecutorService>> i = + schedulerPools.entrySet().iterator(); i.hasNext();) + { + final Map.Entry<String, ScheduledExecutorService> entry = i.next(); + try + { + entry.getValue().shutdownNow(); + } + catch (final Throwable t) + { + log.warn("Failed to close pool {0}", entry.getKey(), t); + } + i.remove(); + schedulerPoolUseCounts.remove(entry.getKey()); + } + } + } + + /** + * Dispose of a thread pool + * + * @param poolName the name of the pool + */ + public void disposeExecutorService(String poolName) + { + disposeExecutorService(poolName, Duration.ZERO); + } + + /** + * Dispose of a thread pool + * + * @param poolName the name of the pool + * @param wait Duration to wait for termination + */ + public void disposeExecutorService(String poolName, Duration wait) + { + ExecutorService pool = null; + + synchronized (pools) + { + AtomicInteger useCount = poolUseCounts.get(poolName); + if (useCount == null) + { + log.warn("No useCount exists for pool {0}", poolName); + } + else if (useCount.decrementAndGet() <= 0) + { + poolUseCounts.remove(poolName, useCount); + pool = pools.remove(poolName); + if (pool == null) + { + log.warn("Failed to close non-existing pool {0}", poolName); + } + else + { + try + { + if (wait == null || wait.isZero()) + { + pool.shutdownNow(); + } + else + { + pool.shutdown(); + } + } + catch (final Throwable t) + { + log.warn("Failed to close pool {0}", poolName, t); + } + } + } + } + + if (pool != null && wait != null && !wait.isZero()) + { + try + { + if (!pool.awaitTermination(wait.toMillis(), TimeUnit.MILLISECONDS)) + { + log.info( "No longer waiting for pool {0} to terminate", poolName); + } + } + catch (final InterruptedException e) + { + // ignore + } + } + } + + /** + * Dispose of a scheduler thread pool + * + * @param poolName the name of the pool + */ + public void disposeSchedulerPool(String poolName) + { + disposeSchedulerPool(poolName, Duration.ZERO); + } + + /** + * Dispose of a scheduler thread pool + * + * @param poolName the name of the pool + * @param wait Duration to wait for termination + */ + public void disposeSchedulerPool(String poolName, Duration wait) { - AtomicInteger useCount = schedulerPoolUseCounts.computeIfAbsent(name, k -> new AtomicInteger()); - useCount.getAndIncrement(); + ExecutorService pool = null; - return schedulerPools.computeIfAbsent(name, key -> { - log.debug( "Creating scheduler pool for name [{0}]", key ); - return createSchedulerPool( config, JCS_THREAD_POOL_MANAGER_PREFIX + key + "-"); - }); + synchronized (schedulerPools) + { + AtomicInteger useCount = schedulerPoolUseCounts.get(poolName); + if (useCount == null) + { + log.warn("No useCount exists for pool {0}", poolName); + } + else if (useCount.decrementAndGet() == 0) + { + schedulerPoolUseCounts.remove(poolName, useCount); + pool = schedulerPools.remove(poolName); + if (pool == null) + { + log.warn("Failed to close non-existing pool {0}", poolName); + } + else + { + try + { + if (wait == null || wait.isZero()) + { + pool.shutdownNow(); + } + else + { + pool.shutdown(); + } + } + catch (final Throwable t) + { + log.warn("Failed to close pool {0}", poolName, t); + } + } + } + } + + if (pool != null && wait != null && !wait.isZero()) + { + try + { + if (!pool.awaitTermination(wait.toMillis(), TimeUnit.MILLISECONDS)) + { + log.info( "No longer waiting for pool {0} to terminate", poolName); + } + } + catch (final InterruptedException e) + { + // ignore + } + } } }
