From bcce1a7148fb90256874f03488d2f97c48975dc8 Mon Sep 17 00:00:00 2001 From: Robin Bygrave Date: Fri, 26 Apr 2013 23:21:25 +1200 Subject: [PATCH] Fix for Issue 4 - Allow programmatic shutdown of EbeanServer --- .../java/com/avaje/ebean/EbeanServer.java | 17 + .../avaje/ebean/config/DataSourceConfig.java | 24 +- .../ebeaninternal/api/SpiEbeanServer.java | 278 +++--- .../autofetch/DefaultAutoFetchManager.java | 6 +- .../DefaultAutoFetchManagerLogging.java | 20 +- .../cluster/socket/SocketClusterListener.java | 10 +- .../core/DefaultBackgroundExecutor.java | 4 +- .../server/core/DefaultServer.java | 76 +- .../server/core/DefaultServerFactory.java | 31 +- .../core/TraditionalBackgroundExecutor.java | 13 +- .../server/lib/BackgroundRunnable.java | 147 --- .../server/lib/BackgroundThread.java | 220 ----- .../server/lib/DaemonScheduleThreadPool.java | 103 +-- .../server/lib/DaemonThreadPool.java | 128 +-- .../server/lib/ShutdownHook.java | 11 - .../server/lib/ShutdownManager.java | 237 +++-- ...SourceNotify.java => DataSourceAlert.java} | 8 +- .../lib/sql/DataSourceAlertListener.java | 26 - .../lib/sql/DataSourceGlobalManager.java | 50 -- .../server/lib/sql/DataSourceManager.java | 233 ----- .../server/lib/sql/DataSourcePool.java | 80 +- .../server/lib/sql/PooledConnection.java | 6 +- .../server/lib/sql/PooledConnectionQueue.java | 8 +- .../server/lib/sql/SimpleAlerter.java | 107 --- .../server/lib/sql/SimpleDataSourceAlert.java | 109 +++ .../server/lib/thread/PooledThread.java | 471 +++++----- .../server/lib/thread/ThreadPool.java | 837 +++++++++--------- .../server/lib/thread/ThreadPoolManager.java | 177 ---- .../transaction/TransactionManager.java | 17 +- .../text/json/TestJsonBeanWithTimeZone.java | 14 +- src/test/resources/logback-test.xml | 2 + 31 files changed, 1397 insertions(+), 2073 deletions(-) delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundRunnable.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundThread.java rename src/main/java/com/avaje/ebeaninternal/server/lib/sql/{DataSourceNotify.java => DataSourceAlert.java} (72%) delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlertListener.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceGlobalManager.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceManager.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleAlerter.java create mode 100644 src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleDataSourceAlert.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPoolManager.java diff --git a/src/main/java/com/avaje/ebean/EbeanServer.java b/src/main/java/com/avaje/ebean/EbeanServer.java index cd8252962..ec28b5d14 100644 --- a/src/main/java/com/avaje/ebean/EbeanServer.java +++ b/src/main/java/com/avaje/ebean/EbeanServer.java @@ -83,6 +83,23 @@ import com.avaje.ebean.text.json.JsonContext; */ public interface EbeanServer { + /** + * Shutdown the EbeanServer. + *

+ * If the under underlying DataSource is the EbeanORM implementation then you + * also have the option of shutting down the DataSource and deregistering the + * JDBC driver. + *

+ * + * @param shutdownDataSource + * if true then shutdown the underlying DataSource if it is the EbeanORM + * DataSource implementation. + * @param deregisterDriver + * if true then deregister the JDBC driver if it is the EbeanORM + * DataSource implementation. + */ + public void shutdown(boolean shutdownDataSource, boolean deregisterDriver); + /** * Return the AdminAutofetch which is used to control and configure the * Autofetch service at runtime. diff --git a/src/main/java/com/avaje/ebean/config/DataSourceConfig.java b/src/main/java/com/avaje/ebean/config/DataSourceConfig.java index 8779a1fb1..d6481e28c 100644 --- a/src/main/java/com/avaje/ebean/config/DataSourceConfig.java +++ b/src/main/java/com/avaje/ebean/config/DataSourceConfig.java @@ -33,7 +33,9 @@ public class DataSourceConfig { private int isolationLevel = Transaction.READ_COMMITTED; private String heartbeatSql; - + + private int heartbeatFreqSecs = 30; + private boolean captureStackTrace; private int maxStackTraceSize = 5; @@ -43,6 +45,7 @@ public class DataSourceConfig { private int maxInactiveTimeSecs = 900; private int pstmtCacheSize = 20; + private int cstmtCacheSize = 20; private int waitTimeoutMillis = 1000; @@ -173,6 +176,25 @@ public class DataSourceConfig { this.heartbeatSql = heartbeatSql; } + + /** + * Return the heartbeat frequency in seconds. + *

+ * This is the expected frequency in which the DataSource should be checked to + * make sure it is healthy and trim idle connections. + *

+ */ + public int getHeartbeatFreqSecs() { + return heartbeatFreqSecs; + } + + /** + * Set the expected heartbeat frequency in seconds. + */ + public void setHeartbeatFreqSecs(int heartbeatFreqSecs) { + this.heartbeatFreqSecs = heartbeatFreqSecs; + } + /** * Return true if a stack trace should be captured when obtaining a connection * from the pool. diff --git a/src/main/java/com/avaje/ebeaninternal/api/SpiEbeanServer.java b/src/main/java/com/avaje/ebeaninternal/api/SpiEbeanServer.java index 818231694..98ecd2db9 100644 --- a/src/main/java/com/avaje/ebeaninternal/api/SpiEbeanServer.java +++ b/src/main/java/com/avaje/ebeaninternal/api/SpiEbeanServer.java @@ -23,164 +23,168 @@ import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; * Service Provider extension to EbeanServer. */ public interface SpiEbeanServer extends EbeanServer, BeanLoader, BeanCollectionLoader { + + /** + * For internal use, shutdown of the server invoked by JVM Shutdown. + */ + public void shutdownManaged(); - /** - * Return true if DeleteMissingChildren defaults to true for stateless updates. - */ - public boolean isDefaultDeleteMissingChildren(); + /** + * Return true if DeleteMissingChildren defaults to true for stateless + * updates. + */ + public boolean isDefaultDeleteMissingChildren(); - /** - * Return true if UpdateNullProperties defaults to true for stateless updates. - */ - public boolean isDefaultUpdateNullProperties(); - - /** - * Return the DatabasePlatform for this server. - */ - public DatabasePlatform getDatabasePlatform(); - - /** - * Return a JDBC driver specific handler for batching. - *

- * Required for Oracle specific batch handling. - *

- */ - public PstmtBatch getPstmtBatch(); - - /** - * Create an object to represent the current CallStack. - *

- * Typically used to identify the origin of queries for Autofetch - * and object graph costing. - *

- */ - public CallStack createCallStack(); - - /** - * Return the DDL generator. - */ - public DdlGenerator getDdlGenerator(); + /** + * Return true if UpdateNullProperties defaults to true for stateless updates. + */ + public boolean isDefaultUpdateNullProperties(); - /** - * Return the AutoFetchListener. - */ - public AutoFetchManager getAutoFetchManager(); + /** + * Return the DatabasePlatform for this server. + */ + public DatabasePlatform getDatabasePlatform(); - /** - * Clear the query execution statistics. - */ - public void clearQueryStatistics(); + /** + * Return a JDBC driver specific handler for batching. + *

+ * Required for Oracle specific batch handling. + *

+ */ + public PstmtBatch getPstmtBatch(); - /** - * Return all the descriptors. - */ - public List> getBeanDescriptors(); + /** + * Create an object to represent the current CallStack. + *

+ * Typically used to identify the origin of queries for Autofetch and object + * graph costing. + *

+ */ + public CallStack createCallStack(); - /** - * Return the BeanDescriptor for a given type of bean. - */ - public BeanDescriptor getBeanDescriptor(Class type); + /** + * Return the DDL generator. + */ + public DdlGenerator getDdlGenerator(); - /** - * Return BeanDescriptor using it's unique id. - */ - public BeanDescriptor getBeanDescriptorById(String descriptorId); + /** + * Return the AutoFetchListener. + */ + public AutoFetchManager getAutoFetchManager(); - /** - * Return BeanDescriptors mapped to this table. - */ - public List> getBeanDescriptors(String tableName); + /** + * Clear the query execution statistics. + */ + public void clearQueryStatistics(); - /** - * Process committed changes from another framework. - *

- * This notifies this instance of the framework that beans have been - * committed externally to it. Either by another framework or clustered - * server. It uses this to maintain its cache and text indexes - * appropriately. - *

- */ - public void externalModification(TransactionEventTable event); + /** + * Return all the descriptors. + */ + public List> getBeanDescriptors(); - /** - * Create a ServerTransaction. - *

- * To specify to use the default transaction isolation use a value of -1. - *

- */ - public SpiTransaction createServerTransaction(boolean isExplicit, int isolationLevel); + /** + * Return the BeanDescriptor for a given type of bean. + */ + public BeanDescriptor getBeanDescriptor(Class type); - /** - * Return the current transaction or null if there is no current - * transaction. - */ - public SpiTransaction getCurrentServerTransaction(); + /** + * Return BeanDescriptor using it's unique id. + */ + public BeanDescriptor getBeanDescriptorById(String descriptorId); - /** - * Create a ScopeTrans for a method for the given scope definition. - */ - public ScopeTrans createScopeTrans(TxScope txScope); + /** + * Return BeanDescriptors mapped to this table. + */ + public List> getBeanDescriptors(String tableName); - /** - * Create a ServerTransaction for query purposes. - */ - public SpiTransaction createQueryTransaction(); + /** + * Process committed changes from another framework. + *

+ * This notifies this instance of the framework that beans have been committed + * externally to it. Either by another framework or clustered server. It uses + * this to maintain its cache and text indexes appropriately. + *

+ */ + public void externalModification(TransactionEventTable event); - /** - * An event from another server in the cluster used to notify local - * BeanListeners of remote inserts updates and deletes. - */ - public void remoteTransactionEvent(RemoteTransactionEvent event); + /** + * Create a ServerTransaction. + *

+ * To specify to use the default transaction isolation use a value of -1. + *

+ */ + public SpiTransaction createServerTransaction(boolean isExplicit, int isolationLevel); - - /** - * Create a query request object. - */ - public SpiOrmQueryRequest createQueryRequest(BeanDescriptor desc, SpiQuery q, Transaction t); + /** + * Return the current transaction or null if there is no current transaction. + */ + public SpiTransaction getCurrentServerTransaction(); - /** - * Compile a query. - */ - public CQuery compileQuery(Query query, Transaction t); + /** + * Create a ScopeTrans for a method for the given scope definition. + */ + public ScopeTrans createScopeTrans(TxScope txScope); - /** - * Return the queryEngine for this server. - */ - public CQueryEngine getQueryEngine(); + /** + * Create a ServerTransaction for query purposes. + */ + public SpiTransaction createQueryTransaction(); - /** - * Execute the findId's query but without copying the query. - *

- * Used so that the list of Id's can be made accessible to client code - * before the query has finished (if executing in a background thread). - *

- */ - public List findIdsWithCopy(Query query, Transaction t); + /** + * An event from another server in the cluster used to notify local + * BeanListeners of remote inserts updates and deletes. + */ + public void remoteTransactionEvent(RemoteTransactionEvent event); - /** - * Execute the findRowCount query but without copying the query. - */ - public int findRowCountWithCopy(Query query, Transaction t); + /** + * Create a query request object. + */ + public SpiOrmQueryRequest createQueryRequest(BeanDescriptor desc, SpiQuery q, + Transaction t); - /** - * Load a batch of Associated One Beans. - */ - public void loadBean(LoadBeanRequest loadRequest); + /** + * Compile a query. + */ + public CQuery compileQuery(Query query, Transaction t); - /** - * Lazy load a batch of Many's. - */ - public void loadMany(LoadManyRequest loadRequest); + /** + * Return the queryEngine for this server. + */ + public CQueryEngine getQueryEngine(); - /** - * Return the default batch size for lazy loading. - */ - public int getLazyLoadBatchSize(); - - /** - * Return true if the type is known as an Entity or Xml type - * or a List Set or Map of known bean types. - */ - public boolean isSupportedType(java.lang.reflect.Type genericType); + /** + * Execute the findId's query but without copying the query. + *

+ * Used so that the list of Id's can be made accessible to client code before + * the query has finished (if executing in a background thread). + *

+ */ + public List findIdsWithCopy(Query query, Transaction t); + + /** + * Execute the findRowCount query but without copying the query. + */ + public int findRowCountWithCopy(Query query, Transaction t); + + /** + * Load a batch of Associated One Beans. + */ + public void loadBean(LoadBeanRequest loadRequest); + + /** + * Lazy load a batch of Many's. + */ + public void loadMany(LoadManyRequest loadRequest); + + /** + * Return the default batch size for lazy loading. + */ + public int getLazyLoadBatchSize(); + + /** + * Return true if the type is known as an Entity or Xml type or a List Set or + * Map of known bean types. + */ + public boolean isSupportedType(java.lang.reflect.Type genericType); } diff --git a/src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManager.java b/src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManager.java index 8291ffe3c..11ee8e651 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManager.java +++ b/src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManager.java @@ -92,7 +92,7 @@ public class DefaultAutoFetchManager implements AutoFetchManager, Serializable { public void setOwner(SpiEbeanServer server, ServerConfig serverConfig) { this.server = server; this.logging = new DefaultAutoFetchManagerLogging(serverConfig, this); - + AutofetchConfig autofetchConfig = serverConfig.getAutofetchConfig(); garbageCollectionOnShutdown = autofetchConfig.isGarbageCollectionOnShutdown(); @@ -104,7 +104,6 @@ public class DefaultAutoFetchManager implements AutoFetchManager, Serializable { setProfilingRate(autofetchConfig.getProfilingRate()); - defaultGarbageCollectionWait = (long) autofetchConfig.getGarbageCollectionWait(); // determine the mode to use when Query.setAutoFetch() was @@ -117,6 +116,9 @@ public class DefaultAutoFetchManager implements AutoFetchManager, Serializable { + "] mode[" + mode + "] profiling rate[" + profilingRate + "] min[" + profilingMin + "] base[" + profilingBase + "]"; logging.logInfo(msg, null); + + // Register a periodic update of the profiling informations + this.logging.init(server); } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManagerLogging.java b/src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManagerLogging.java index 1fbe15737..ee869baf9 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManagerLogging.java +++ b/src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManagerLogging.java @@ -1,12 +1,13 @@ package com.avaje.ebeaninternal.server.autofetch; +import java.util.concurrent.TimeUnit; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.avaje.ebean.config.AutofetchConfig; import com.avaje.ebean.config.GlobalProperties; import com.avaje.ebean.config.ServerConfig; -import com.avaje.ebeaninternal.server.lib.BackgroundThread; +import com.avaje.ebeaninternal.api.SpiEbeanServer; import com.avaje.ebeaninternal.server.querydefn.OrmQueryDetail; /** @@ -24,16 +25,17 @@ public class DefaultAutoFetchManagerLogging { private final boolean traceUsageCollection; + private final int updateFreqInSecs; + public DefaultAutoFetchManagerLogging(ServerConfig serverConfig, DefaultAutoFetchManager profileListener) { this.manager = profileListener; - - AutofetchConfig autofetchConfig = serverConfig.getAutofetchConfig(); - - traceUsageCollection = GlobalProperties.getBoolean("ebean.autofetch.traceUsageCollection", false); - - int updateFreqInSecs = autofetchConfig.getProfileUpdateFrequency(); - BackgroundThread.add(updateFreqInSecs, new UpdateProfile()); + this.traceUsageCollection = GlobalProperties.getBoolean("ebean.autofetch.traceUsageCollection", false); + this.updateFreqInSecs = serverConfig.getAutofetchConfig().getProfileUpdateFrequency(); + } + + public void init(SpiEbeanServer ebeanServer) { + ebeanServer.getBackgroundExecutor().executePeriodically(new UpdateProfile(), updateFreqInSecs, TimeUnit.SECONDS); } private final class UpdateProfile implements Runnable { diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java index 0b863b6a9..f5d5197b5 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java @@ -6,11 +6,11 @@ import java.net.ServerSocket; import java.net.Socket; import java.net.SocketException; -import com.avaje.ebeaninternal.server.lib.thread.ThreadPool; -import com.avaje.ebeaninternal.server.lib.thread.ThreadPoolManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.avaje.ebeaninternal.server.lib.thread.ThreadPool; + /** * Serverside multithreaded socket listener. Accepts connections and dispatches * them to an appropriate handler. @@ -70,7 +70,7 @@ class SocketClusterListener implements Runnable { */ public SocketClusterListener(SocketClusterBroadcast owner, int port) { this.owner = owner; - this.threadPool = ThreadPoolManager.getThreadPool("EbeanClusterMember"); + this.threadPool = ThreadPool.createThreadPool("EbeanCluster"); this.port = port; try { @@ -118,8 +118,10 @@ class SocketClusterListener implements Runnable { listenerThread.interrupt(); serverListenSocket.close(); } catch (IOException e) { - logger.error(null, e); + logger.error("Error shutting down listener", e); } + + threadPool.shutdown(); } /** diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java index 2dccea7ae..ece35c94c 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java @@ -28,8 +28,8 @@ public class DefaultBackgroundExecutor implements SpiBackgroundExecutor { * the time in seconds allowed for the pool to shutdown nicely. * After this the pool is forced to shutdown. */ - public DefaultBackgroundExecutor(int mainPoolSize, int schedulePoolSize, long keepAliveSecs,int shutdownWaitSeconds, String namePrefix) { - this.pool = new DaemonThreadPool(mainPoolSize, keepAliveSecs, shutdownWaitSeconds, namePrefix); + public DefaultBackgroundExecutor(int schedulePoolSize, int corePoolSize, long keepAliveSecs,int shutdownWaitSeconds, String namePrefix) { + this.pool = new DaemonThreadPool(corePoolSize, keepAliveSecs, shutdownWaitSeconds, namePrefix); this.schedulePool = new DaemonScheduleThreadPool(schedulePoolSize, shutdownWaitSeconds, namePrefix+"-periodic-"); } diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java index 6b9f03126..5c85680ae 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java @@ -177,6 +177,11 @@ public final class DefaultServer implements SpiEbeanServer { */ private MBeanServer mbeanServer; + /** + * Flag set when the server has shutdown. + */ + private boolean shutdown; + /** * The default batch size for lazy loading beans or collections. */ @@ -236,24 +241,26 @@ public final class DefaultServer implements SpiEbeanServer { this.jsonContext = config.createJsonContext(this); loadAndInitializePlugins(config); - - ShutdownManager.register(new Shutdown()); + + // Register with the JVM Shutdown hook + ShutdownManager.registerEbeanServer(this); } protected void loadAndInitializePlugins(InternalConfiguration config) { + List spiPlugins = new ArrayList(); final Iterator plugins = ServiceLoader.load(SpiEbeanPlugin.class).iterator(); while (plugins.hasNext()) { SpiEbeanPlugin plugin = plugins.next(); - spiPlugins.add(plugin); - plugin.setup(this, this.getDatabasePlatform(), config.getServerConfig()); - if (plugin instanceof DdlGenerator) // backwards compatible + if (plugin instanceof DdlGenerator) { + // backwards compatible ddlGenerator = (DdlGenerator)plugin; + } } ebeanPlugins = Collections.unmodifiableList(spiPlugins); @@ -364,24 +371,53 @@ public final class DefaultServer implements SpiEbeanServer { } } - private final class Shutdown implements Runnable { - public void run() { - try { - if (mbeanServer != null) { - mbeanServer.unregisterMBean(new ObjectName(mbeanName + ",key=AutoFetch")); - } - } catch (Exception e) { - String msg = "Error unregistering Ebean " + mbeanName; - logger.error(msg, e); - } - - // shutdown services - transactionManager.shutdown(); - autoFetchManager.shutdown(); - backgroundExecutor.shutdown(); + /** + * Shutting down via JVM Shutdown hook. + */ + public void shutdownManaged() { + synchronized (this) { + shutdownInternal(true, false); } } + /** + * Shutting down manually. + */ + public void shutdown(boolean shutdownDataSource, boolean deregisterDriver) { + synchronized (this) { + // Unregister from JVM Shutdown hook + ShutdownManager.unregisterEbeanServer(this); + shutdownInternal(shutdownDataSource, deregisterDriver); + } + } + + /** + * Shutdown the services like threads and DataSource. + */ + private void shutdownInternal(boolean shutdownDataSource, boolean deregisterDriver) { + + logger.debug("Shutting down EbeanServer " + getName()); + if (shutdown) { + // Already shutdown + return; + } + try { + if (mbeanServer != null) { + mbeanServer.unregisterMBean(new ObjectName(mbeanName + ",key=AutoFetch")); + } + } catch (Exception e) { + logger.error("Error unregistering Ebean " + mbeanName, e); + } + + // shutdown autofetch profile collection + autoFetchManager.shutdown(); + // shutdown background threads + backgroundExecutor.shutdown(); + // shutdown DataSource (if its an Ebean one) + transactionManager.shutdown(shutdownDataSource, deregisterDriver); + shutdown = true; + } + /** * Return the server name. */ diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java index 3c5f6598c..2bf4dc6fe 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java @@ -32,6 +32,9 @@ import javax.management.MBeanServerFactory; import javax.persistence.PersistenceException; import javax.sql.DataSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import com.avaje.ebean.EbeanServer; import com.avaje.ebean.cache.ServerCacheFactory; import com.avaje.ebean.cache.ServerCacheManager; @@ -51,12 +54,10 @@ import com.avaje.ebeaninternal.server.cluster.ClusterManager; import com.avaje.ebeaninternal.server.jdbc.OraclePstmtBatch; import com.avaje.ebeaninternal.server.jdbc.StandardPstmtDelegate; import com.avaje.ebeaninternal.server.lib.ShutdownManager; -import com.avaje.ebeaninternal.server.lib.sql.DataSourceGlobalManager; +import com.avaje.ebeaninternal.server.lib.sql.DataSourceAlert; import com.avaje.ebeaninternal.server.lib.sql.DataSourcePool; +import com.avaje.ebeaninternal.server.lib.sql.SimpleDataSourceAlert; import com.avaje.ebeaninternal.server.lib.thread.ThreadPool; -import com.avaje.ebeaninternal.server.lib.thread.ThreadPoolManager; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Default Server side implementation of ServerFactory. @@ -128,31 +129,24 @@ public class DefaultServerFactory implements BootupEbeanManager { String namePrefix = "Ebean-" + serverConfig.getName(); - // the size of the pool for executing periodic tasks (such as cache - // flushing) + // the size of the pool for executing periodic tasks (such as cache flushing) int schedulePoolSize = GlobalProperties.getInt("backgroundExecutor.schedulePoolsize", 1); // the side of the main pool for immediate background task execution - int minPoolSize = GlobalProperties.getInt("backgroundExecutor.minPoolSize", 1); + int minPoolSize = GlobalProperties.getInt("backgroundExecutor.minPoolSize", 0); int poolSize = GlobalProperties.getInt("backgroundExecutor.poolsize", 20); int maxPoolSize = GlobalProperties.getInt("backgroundExecutor.maxPoolSize", poolSize); - int idleSecs = GlobalProperties.getInt("backgroundExecutor.idlesecs", 60); + int idleSecs = GlobalProperties.getInt("backgroundExecutor.idlesecs", 120); int shutdownSecs = GlobalProperties.getInt("backgroundExecutor.shutdownSecs", 30); boolean useTrad = GlobalProperties.getBoolean("backgroundExecutor.traditional", true); - if (useTrad) { - // this pool will use Idle seconds between min and max so I think it is - // better - // as it will let the thread count float between the min and max - ThreadPool pool = ThreadPoolManager.getThreadPool(namePrefix); - pool.setMinSize(minPoolSize); - pool.setMaxSize(maxPoolSize); - pool.setMaxIdleTime(idleSecs * 1000); + // this pool will use Idle seconds to maintain the thread count between min and max + ThreadPool pool = new ThreadPool(namePrefix, true, null, minPoolSize, maxPoolSize, idleSecs*1000); return new TraditionalBackgroundExecutor(pool, schedulePoolSize, shutdownSecs, namePrefix); } else { - return new DefaultBackgroundExecutor(poolSize, schedulePoolSize, idleSecs, shutdownSecs, namePrefix); + return new DefaultBackgroundExecutor(schedulePoolSize, maxPoolSize, idleSecs, shutdownSecs, namePrefix); } } @@ -422,7 +416,8 @@ public class DefaultServerFactory implements BootupEbeanManager { dsConfig.setHeartbeatSql(heartbeatSql); } - return DataSourceGlobalManager.getDataSource(config.getName(), dsConfig); + DataSourceAlert notify = new SimpleDataSourceAlert(); + return new DataSourcePool(notify, config.getName(), dsConfig); } /** diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/TraditionalBackgroundExecutor.java b/src/main/java/com/avaje/ebeaninternal/server/core/TraditionalBackgroundExecutor.java index 4710cc57a..a8b3bb2bd 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/TraditionalBackgroundExecutor.java +++ b/src/main/java/com/avaje/ebeaninternal/server/core/TraditionalBackgroundExecutor.java @@ -2,6 +2,9 @@ package com.avaje.ebeaninternal.server.core; import java.util.concurrent.TimeUnit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import com.avaje.ebeaninternal.api.SpiBackgroundExecutor; import com.avaje.ebeaninternal.server.lib.DaemonScheduleThreadPool; import com.avaje.ebeaninternal.server.lib.thread.ThreadPool; @@ -13,6 +16,8 @@ import com.avaje.ebeaninternal.server.lib.thread.ThreadPool; */ public class TraditionalBackgroundExecutor implements SpiBackgroundExecutor { + private static Logger logger = LoggerFactory.getLogger(TraditionalBackgroundExecutor.class); + private final ThreadPool pool; private final DaemonScheduleThreadPool schedulePool; @@ -33,11 +38,17 @@ public class TraditionalBackgroundExecutor implements SpiBackgroundExecutor { } public void executePeriodically(Runnable r, long delay, TimeUnit unit) { + if (logger.isDebugEnabled()) { + logger.debug("Registering for executePeriodically {} delay:{} {}",r, delay, unit); + } schedulePool.scheduleWithFixedDelay(r, delay, delay, unit); } public void shutdown() { - // the pool is shutdown automatically by the ThreadPoolManager + if (logger.isDebugEnabled()) { + logger.debug("Shutting down"); + } + pool.shutdown(); schedulePool.shutdown(); } diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundRunnable.java b/src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundRunnable.java deleted file mode 100644 index cc861633a..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundRunnable.java +++ /dev/null @@ -1,147 +0,0 @@ -package com.avaje.ebeaninternal.server.lib; - -/** - * Wraps a Runnable that is registed with BackgroundThread. - * @see BackgroundThread - */ -public class BackgroundRunnable { - - /** - * The task to run. - */ - Runnable runnable; - - /** - * The frequency to run the task. - */ - int freqInSecs; - - /** - * The number of times the task has run. - */ - int runCount = 0; - - /** - * The total time taken to run the task. - */ - long totalRunTime = 0; - - /** - * The start time the task was started. - */ - long startTimeTemp; - - long startAfter; - - /** - * Used to disable/enable a task. - */ - boolean isActive = true; - - public BackgroundRunnable(Runnable runnable, int freqInSecs){ - this(runnable, freqInSecs, System.currentTimeMillis()+1000*(freqInSecs+10)); - } - - public BackgroundRunnable(Runnable runnable, int freqInSecs, long startAfter){ - this.runnable = runnable; - this.freqInSecs = freqInSecs; - this.startAfter = startAfter; - } - - /** - * Return true if this can be run now. - *

- * This is used to stop jobs firing immediately. - *

- */ - public boolean runNow(long now){ - return now > startAfter; - } - - /** - * Returns true if the task is currently enabled. - */ - public boolean isActive() { - return isActive; - } - - /** - * Set this to false to stop this task from running. - * Useful to temporarily disable a particular task. - */ - public void setActive(boolean isActive) { - this.isActive = isActive; - } - - /** - * Mark the start time of a task run. - */ - protected void runStart() { - startTimeTemp = System.currentTimeMillis(); - } - - /** - * Mark the end time of a task run. - */ - protected void runEnd(){ - runCount++; - long exeTime = System.currentTimeMillis() - startTimeTemp; - totalRunTime = totalRunTime + exeTime; - } - - /** - * Return the number of times this task was run. - */ - public int getRunCount() { - return runCount; - } - - /** - * Return the average time this task takes to run. - */ - public long getAverageRunTime() { - if (runCount == 0){ - return 0; - } - return totalRunTime/runCount; - } - - /** - * Return the frequency in seconds that this task runs. - */ - public int getFreqInSecs() { - return freqInSecs; - } - - /** - * Set the frequency in seconds that this task runs. - */ - public void setFreqInSecs(int freqInSecs) { - this.freqInSecs = freqInSecs; - } - - /** - * Return the underlying runnable. - */ - public Runnable getRunnable() { - return runnable; - } - - /** - * Set the underlying runnable. - */ - public void setRunnable(Runnable runnable) { - this.runnable = runnable; - } - - public String toString() { - StringBuffer sb = new StringBuffer(); - sb.append("["); - sb.append(runnable.getClass().getName()); - sb.append(" freq:").append(freqInSecs); - sb.append(" count:").append(getRunCount()); - sb.append(" avgTime:").append(getAverageRunTime()); - sb.append("]"); - return sb.toString(); - } -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundThread.java b/src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundThread.java deleted file mode 100644 index caba0a7ac..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundThread.java +++ /dev/null @@ -1,220 +0,0 @@ -package com.avaje.ebeaninternal.server.lib; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Iterator; -import java.util.Vector; - -/** - * A general background thread that runs registered tasks periodically. - *

- * Several features such as CacheManager, DataSourceManager, ThreadPoolManager - * require tasks to be undertaken periodically. Instead of each having their own - * background thread they register runnables with this one. - *

- *

- * SystemProperties:
- * - * 


- *   ## initially sleep for 5 seconds before starting 
- *   backgroundthread.initialsleep=5
- *
- * - *

- * - * @see BackgroundRunnable - */ -public final class BackgroundThread { - - private static final Logger logger = LoggerFactory.getLogger(BackgroundThread.class); - - private static final BackgroundThread me = new BackgroundThread(); - - /** - * The list of Runnable tasks. - */ - private Vector list = new Vector(); - - /** - * Used to synchronize the list. - */ - private final Object monitor = new Object(); - - /** - * The underlying background thread. - */ - private final Thread thread; - - /** - * Wakes every second to look for tasks to run. - */ - private long sleepTime = 1000; - - /** - * The number of times a task is run. - */ - private long count; - - /** - * The time it takes to run the tasks. - */ - private long exeTime; - - /** - * Set when shutting down. - */ - private boolean stopped; - - /** - * Used to shutdown nicely. - */ - private Object threadMonitor = new Object(); - - private BackgroundThread() { - - thread = new Thread(new Runner(), "EbeanBackgroundThread"); - thread.setDaemon(true); - thread.start(); - } - - /** - * Register a Runnable to execute every freqInSecs seconds. - */ - public static void add(int freqInSecs, Runnable runnable) { - add(new BackgroundRunnable(runnable, freqInSecs)); - } - - /** - * Register a Runnable to execute every freqInSecs seconds. - */ - public static void add(BackgroundRunnable backgroundRunnable) { - me.addTask(backgroundRunnable); - } - - /** - * Stop the service. - */ - public static void shutdown() { - me.stop(); - } - - /** - * Return the registered BackgroundRunnable objects. - */ - public static Iterator runnables() { - synchronized (me.monitor) { - return me.list.iterator(); - } - } - - private void addTask(BackgroundRunnable backgroundRunnable) { - synchronized (monitor) { - list.add(backgroundRunnable); - } - } - - /** - * Stop the thread nicely. This will wait a maximum of 10 seconds for - * current work to be finished. - */ - private void stop() { - - stopped = true; - synchronized (threadMonitor) { - try { - threadMonitor.wait(10000); - } catch (InterruptedException e) { - ; - } - } - // thread = null; - } - - private class Runner implements Runnable { - - /** - * Run the registered tasks periodically. - */ - public void run() { - - if (ShutdownManager.isStopping()) { - return; - } - - while (!stopped) { - try { - - long actualSleep = sleepTime - exeTime; - if (actualSleep < 0) { - actualSleep = sleepTime; - } - Thread.sleep(actualSleep); - synchronized (monitor) { - runJobs(); - } - - } catch (InterruptedException e) { - logger.error(null, e); - } - } - - // Tell Stop() we have shut ourselves down successfully - synchronized (threadMonitor) { - threadMonitor.notifyAll(); - } - } - - private void runJobs() { - - long startTime = System.currentTimeMillis(); - - // call trim on each cache - Iterator it = list.iterator(); - while (it.hasNext()) { - BackgroundRunnable bgr = (BackgroundRunnable) it.next(); - if (bgr.isActive()) { - - int freqInSecs = bgr.getFreqInSecs(); - - if (count % freqInSecs == 0) { - Runnable runable = bgr.getRunnable(); - if (bgr.runNow(startTime)){ - bgr.runStart(); - if (logger.isTraceEnabled()) { - String msg = count + " BGRunnable running [" - + runable.getClass().getName() + "]"; - logger.trace(msg); - } - - runable.run(); - bgr.runEnd(); - } - } - } - } - exeTime = System.currentTimeMillis() - startTime; - count++; - - if (count == 86400) { - // reset count back to zero every day - count = 0; - } - } - } - - public String toString() { - synchronized (monitor) { - StringBuffer sb = new StringBuffer(); - - Iterator it = runnables(); - while (it.hasNext()) { - BackgroundRunnable bgr = it.next(); - sb.append(bgr); - } - - return sb.toString(); - } - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/DaemonScheduleThreadPool.java b/src/main/java/com/avaje/ebeaninternal/server/lib/DaemonScheduleThreadPool.java index 557ea3ac0..63747071a 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/DaemonScheduleThreadPool.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/DaemonScheduleThreadPool.java @@ -1,10 +1,8 @@ package com.avaje.ebeaninternal.server.lib; - import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import com.avaje.ebeaninternal.api.Monitor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -18,61 +16,64 @@ import org.slf4j.LoggerFactory; */ public final class DaemonScheduleThreadPool extends ScheduledThreadPoolExecutor { - private static final Logger logger = LoggerFactory.getLogger(DaemonScheduleThreadPool.class); + private static final Logger logger = LoggerFactory.getLogger(DaemonScheduleThreadPool.class); - private final Monitor monitor = new Monitor(); + private final String namePrefix; - private int shutdownWaitSeconds; + private int shutdownWaitSeconds; - /** - * Construct the DaemonScheduleThreadPool. - */ - public DaemonScheduleThreadPool(int coreSize, int shutdownWaitSeconds, String namePrefix) { - super(coreSize, new DaemonThreadFactory(namePrefix)); - this.shutdownWaitSeconds = shutdownWaitSeconds; - - // we want to shutdown nicely when either the web application stops. - // Adding the JVM shutdown hook as a safety (and when not run in tomcat) - Runtime.getRuntime().addShutdownHook(new ShutdownHook()); - } + /** + * Construct the DaemonScheduleThreadPool. + */ + public DaemonScheduleThreadPool(int coreSize, int shutdownWaitSeconds, String namePrefix) { - /** - * Shutdown this thread pool nicely if possible. - *

- * This will wait a maximum of 20 seconds before terminating any threads - * still working. - *

- */ - public void shutdown() { - synchronized (monitor) { - if (super.isShutdown()) { - logger.debug("... DaemonScheduleThreadPool already shut down"); - return; - } - try { - logger.debug("DaemonScheduleThreadPool shutting down..."); - super.shutdown(); - if (!super.awaitTermination(shutdownWaitSeconds, TimeUnit.SECONDS)) { - logger.info("ScheduleService shut down timeout exceeded. Terminating running threads."); - super.shutdownNow(); - } + super(coreSize, new DaemonThreadFactory(namePrefix)); + this.namePrefix = namePrefix; + this.shutdownWaitSeconds = shutdownWaitSeconds; + } - } catch (Exception e) { - String msg = "Error during shutdown"; - logger.error(msg, e); - e.printStackTrace(); - } + /** + * Register a shutdown hook with the JVM Runtime. + */ + public void registerShutdownHook() { + Runtime.getRuntime().addShutdownHook(new ShutdownHook()); + } + + /** + * Shutdown this thread pool nicely if possible. + *

+ * This will wait a maximum of 20 seconds before terminating any threads still + * working. + *

+ */ + public void shutdown() { + synchronized (this) { + if (super.isShutdown()) { + logger.debug("DaemonScheduleThreadPool {} already shut down", namePrefix); + return; + } + try { + logger.debug("DaemonScheduleThreadPool {} shutting down...", namePrefix); + super.shutdown(); + if (!super.awaitTermination(shutdownWaitSeconds, TimeUnit.SECONDS)) { + logger.info("DaemonScheduleThreadPool shut down timeout exceeded. Terminating running threads."); + super.shutdownNow(); } + + } catch (Exception e) { + logger.error("Error during shutdown of " + namePrefix, e); + e.printStackTrace(); + } } + } - /** - * Fired by the JVM Runtime shutdown. - */ - private class ShutdownHook extends Thread { - @Override - public void run() { - shutdown(); - } - }; + /** + * Fired by the JVM Runtime shutdown. + */ + private class ShutdownHook extends Thread { + @Override + public void run() { + shutdown(); + } + }; } - diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/DaemonThreadPool.java b/src/main/java/com/avaje/ebeaninternal/server/lib/DaemonThreadPool.java index df07acfc1..a0bcbcdde 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/DaemonThreadPool.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/DaemonThreadPool.java @@ -1,11 +1,9 @@ package com.avaje.ebeaninternal.server.lib; - import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import com.avaje.ebeaninternal.api.Monitor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -16,71 +14,73 @@ import org.slf4j.LoggerFactory; */ public final class DaemonThreadPool extends ThreadPoolExecutor { - private static final Logger logger = LoggerFactory.getLogger(DaemonThreadPool.class); + private static final Logger logger = LoggerFactory.getLogger(DaemonThreadPool.class); - private final Monitor monitor = new Monitor(); - - private final String namePrefix; - - private int shutdownWaitSeconds; - - /** - * Construct the DaemonThreadPool. - * - * @param coreSize - * the core size of the thread pool. - * @param keepAliveSecs - * the time in seconds idle threads are keep alive - * @param shutdownWaitSeconds - * the time in seconds allowed for the pool to shutdown nicely. - * After this the pool is forced to shutdown. - */ - public DaemonThreadPool(int coreSize, long keepAliveSecs, int shutdownWaitSeconds, String namePrefix) { - super(coreSize, coreSize, keepAliveSecs, TimeUnit.SECONDS, new LinkedBlockingQueue(), new DaemonThreadFactory(namePrefix)); - this.shutdownWaitSeconds = shutdownWaitSeconds; - this.namePrefix = namePrefix; - // we want to shutdown nicely when either the web application stops. - // Adding the JVM shutdown hook as a safety (and when not run in tomcat) - Runtime.getRuntime().addShutdownHook(new ShutdownHook()); - } + private final String namePrefix; - /** - * Shutdown this thread pool nicely if possible. - *

- * This will wait a maximum of 20 seconds before terminating any threads - * still working. - *

- */ - public void shutdown() { - synchronized (monitor) { - if (super.isShutdown()) { - logger.debug("... DaemonThreadPool["+namePrefix+"] already shut down"); - return; - } - try { - logger.debug("DaemonThreadPool["+namePrefix+"] shutting down..."); - super.shutdown(); - if (!super.awaitTermination(shutdownWaitSeconds, TimeUnit.SECONDS)) { - logger.info("DaemonThreadPool["+namePrefix+"] shut down timeout exceeded. Terminating running threads."); - super.shutdownNow(); - } + private final int shutdownWaitSeconds; - } catch (Exception e) { - String msg = "Error during shutdown of DaemonThreadPool["+namePrefix+"]"; - logger.error(msg, e); - e.printStackTrace(); - } + /** + * Construct the DaemonThreadPool. + * + * @param coreSize + * the core size of the thread pool. + * @param keepAliveSecs + * the time in seconds idle threads are keep alive + * @param shutdownWaitSeconds + * the time in seconds allowed for the pool to shutdown nicely. After + * this the pool is forced to shutdown. + */ + public DaemonThreadPool(int coreSize, long keepAliveSecs, int shutdownWaitSeconds, String namePrefix) { + + super(coreSize, coreSize, keepAliveSecs, TimeUnit.SECONDS, new LinkedBlockingQueue(), new DaemonThreadFactory(namePrefix)); + allowCoreThreadTimeOut(true); + this.shutdownWaitSeconds = shutdownWaitSeconds; + this.namePrefix = namePrefix; + } + + /** + * Register a shutdown hook with the JVM Runtime. + */ + public void registerShutdownHook() { + Runtime.getRuntime().addShutdownHook(new ShutdownHook()); + } + + /** + * Shutdown this thread pool nicely if possible. + *

+ * This will wait a maximum of 20 seconds before terminating any threads still + * working. + *

+ */ + public void shutdown() { + synchronized (this) { + if (super.isShutdown()) { + logger.debug("DaemonThreadPool[" + namePrefix + "] already shut down"); + return; + } + try { + logger.debug("DaemonThreadPool[" + namePrefix + "] shutting down..."); + super.shutdown(); + if (!super.awaitTermination(shutdownWaitSeconds, TimeUnit.SECONDS)) { + logger.info("DaemonThreadPool[" + namePrefix+ "] shut down timeout exceeded. Terminating running threads."); + super.shutdownNow(); } - } - /** - * Fired by the JVM Runtime shutdown. - */ - private class ShutdownHook extends Thread { - @Override - public void run() { - shutdown(); - } - }; + } catch (Exception e) { + logger.error("Error during shutdown of DaemonThreadPool[" + namePrefix + "]", e); + e.printStackTrace(); + } + } + } + + /** + * Fired by the JVM Runtime shutdown. + */ + private class ShutdownHook extends Thread { + @Override + public void run() { + shutdown(); + } + }; } - diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownHook.java b/src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownHook.java index d3a7e453e..7961434e3 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownHook.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownHook.java @@ -6,23 +6,12 @@ package com.avaje.ebeaninternal.server.lib; * This is the ShutdownHook that gets added to Runtime. * It will try to shutdown the system cleanly when the JVM exits. * It is best to add your own shutdown hooks to StartStop. - * */ class ShutdownHook extends Thread { ShutdownHook() { } -// /** -// * Register this as a Shutdown hook with the Runtime. -// */ -// public static void registerWithRuntime() { -// -// ShutdownHook hook = new ShutdownHook(); -// Runtime.getRuntime().addShutdownHook(hook); -// } - - /** * Fired by the JVM Runtime on shutdown. */ diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownManager.java b/src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownManager.java index ab4fc5b03..9207e70b3 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownManager.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownManager.java @@ -3,19 +3,20 @@ package com.avaje.ebeaninternal.server.lib; import java.sql.Driver; import java.sql.DriverManager; import java.sql.SQLException; +import java.util.ArrayList; import java.util.Enumeration; -import java.util.Vector; +import java.util.List; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import com.avaje.ebean.common.BootupEbeanManager; import com.avaje.ebean.config.GlobalProperties; import com.avaje.ebeaninternal.api.ClassUtil; -import com.avaje.ebeaninternal.server.lib.sql.DataSourceGlobalManager; -import com.avaje.ebeaninternal.server.lib.thread.ThreadPoolManager; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import com.avaje.ebeaninternal.api.SpiEbeanServer; /** - * Manages the shutdown of the Runtime. + * Manages the shutdown of the JVM Runtime. *

* Makes sure all the resources are shutdown properly and in order. *

@@ -24,19 +25,19 @@ public final class ShutdownManager { private static final Logger logger = LoggerFactory.getLogger(ShutdownManager.class); - static final Vector runnables = new Vector(); + static final List servers = new ArrayList(); + + static final ShutdownHook shutdownHook = new ShutdownHook(); static boolean stopping; - static BootupEbeanManager serverFactory; - - static final ShutdownHook shutdownHook = new ShutdownHook(); + static BootupEbeanManager serverFactory; static boolean whyShutdown; static { // Register the Shutdown hook - register(); + registerShutdownHook(); whyShutdown = GlobalProperties.getBoolean("debug.shutdown.why",false); } @@ -49,18 +50,19 @@ public final class ShutdownManager { public static void registerServerFactory(BootupEbeanManager factory){ serverFactory = factory; } + /** * Make sure the ShutdownManager is activated. */ public static void touch() { - + // Do nothing } /** * Return true if the system is in the process of stopping. */ public static boolean isStopping() { - synchronized (runnables) { + synchronized (servers) { return stopping; } } @@ -72,8 +74,8 @@ public final class ShutdownManager { * for that case we need to make sure the shutdown hook is deregistered. *

*/ - private static void deregister() { - synchronized (runnables) { + protected static void deregisterShutdownHook() { + synchronized (servers) { try { Runtime.getRuntime().removeShutdownHook(shutdownHook); } catch (IllegalStateException ex) { @@ -87,8 +89,8 @@ public final class ShutdownManager { /** * Register the shutdown hook with the Runtime. */ - private static void register() { - synchronized (runnables) { + protected static void registerShutdownHook() { + synchronized (servers) { try { Runtime.getRuntime().addShutdownHook(shutdownHook); } catch (IllegalStateException ex) { @@ -98,118 +100,101 @@ public final class ShutdownManager { } } } + + /** + * Shutdown gracefully cleaning up any resources as required. + *

+ * This is typically invoked via JVM shutdown hook. + *

+ */ + public static void shutdown() { + synchronized (servers) { + if (stopping) { + // Already run shutdown... + return; + } + + if (logger.isDebugEnabled()) { + logger.debug("Shutting down"); + } + + if (whyShutdown) { + try { + throw new RuntimeException("debug.shutdown.why=true ..."); + } catch (Throwable e) { + logger.warn("Stacktrace showing why shutdown was fired", e); + } + } + + stopping = true; + + deregisterShutdownHook(); + + String shutdownRunner = GlobalProperties.get("system.shutdown.runnable", null); + if (shutdownRunner != null) { + try { + // A custom runnable executed at the start of shutdown + Runnable r = (Runnable) ClassUtil.newInstance(shutdownRunner); + r.run(); + } catch (Exception e) { + logger.error("Error running custom shutdown runnable", e); + } + } + + if (serverFactory != null) { + // shutdown cluster networking if active + serverFactory.shutdown(); + } + + // shutdown any registered servers that have not + // already been shutdown manually + for (SpiEbeanServer server : servers) { + try { + server.shutdownManaged(); + } catch (Exception ex) { + logger.error("Error executing shutdown runnable", ex); + ex.printStackTrace(); + } + } + + if (GlobalProperties.getBoolean("datasource.deregisterAllDrivers", false)) { + deregisterAllJdbcDrivers(); + } + } + } + + private static void deregisterAllJdbcDrivers() { + // This manually deregisters all JDBC drivers + Enumeration drivers = DriverManager.getDrivers(); + while (drivers.hasMoreElements()) { + Driver driver = drivers.nextElement(); + try { + logger.info("Deregistering jdbc driver: "+driver); + DriverManager.deregisterDriver(driver); + } catch (SQLException e) { + logger.error("Error deregistering driver "+driver, e); + } + } + } /** - * cleanup any resources as Runtime is stopping. + * Register an ebeanServer to be shutdown when the JVM is shutdown. + */ + public static void registerEbeanServer(SpiEbeanServer server) { + synchronized (servers) { + servers.add(server); + } + } + + /** + * Deregister an ebeanServer. *

- *

    - *
  • Run the application specific shutdown runnable - *
  • Run any other registered shutdown runnable - *
  • Deregister from the cluster if required - *
  • Shutdown Thread pools - *
  • Shutdown any database connection pools - *
+ * This is done when the ebeanServer is shutdown manually. *

*/ - public static void shutdown() { - synchronized (runnables) { - if (stopping) { - // Already run shutdown... - return; - } - - if (whyShutdown){ - try { - throw new RuntimeException("debug.shutdown.why=true ..."); - } catch(Throwable e){ - logger.warn("Stacktrace showing why shutdown was fired", e); - } - } - - stopping = true; - //logger.info("Stopping [" + SystemProperties.getContextName() + "]"); - - deregister(); - - // stop the BackgroundThread - BackgroundThread.shutdown(); - - String shutdownRunner = GlobalProperties.get("system.shutdown.runnable", null); - if (shutdownRunner != null) { - try { - Runnable r = (Runnable)ClassUtil.newInstance(shutdownRunner); - r.run(); - } catch (Exception e) { - logger.error(null, e); - } - } - - // shutdown any registered runnable - - Enumeration e = runnables.elements(); - while (e.hasMoreElements()) { - try { - Runnable r = (Runnable) e.nextElement(); - r.run(); - } catch (Exception ex) { - logger.error(null, ex); - ex.printStackTrace(); - } - } - try { - // shutdown order is important! - // CronManager is ok - - if (serverFactory != null){ - serverFactory.shutdown(); - } - - ThreadPoolManager.shutdown(); - - DataSourceGlobalManager.shutdown(); - - boolean dereg = GlobalProperties.getBoolean("datasource.deregisterAllDrivers", false); - if (dereg){ - deregisterAllJdbcDrivers(); - } - - } catch (Exception ex) { - String msg = "Shutdown Exception: "+ ex.getMessage(); - System.err.println(msg); - ex.printStackTrace(); - try { - logger.error(null, ex); - } catch (Exception exc) { - String ms = "Error Logging error to the Log. It may be shutting down."; - System.err.println(ms); - exc.printStackTrace(); - } - } - } - } - - private static void deregisterAllJdbcDrivers() { - // This manually deregisters JDBC driver, which prevents Tomcat 7 from complaining about memory leaks wrto this class - Enumeration drivers = DriverManager.getDrivers(); - while (drivers.hasMoreElements()) { - Driver driver = drivers.nextElement(); - try { - DriverManager.deregisterDriver(driver); - logger.info(String.format("Deregistering jdbc driver: %s", driver)); - } catch (SQLException e) { - logger.error(String.format("Error deregistering driver %s", driver), e); - } - } - } - - /** - * Register a runnable to be executed when the system is shutdown. Note that - * runnables registered here are shutdown before any thread pools or - * DataSource pools are shutdown. - */ - public static void register(Runnable runnable) { - synchronized (runnables) { - runnables.add(runnable); - } - } + public static void unregisterEbeanServer(SpiEbeanServer server) { + synchronized (servers) { + servers.remove(server); + } + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceNotify.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlert.java similarity index 72% rename from src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceNotify.java rename to src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlert.java index c2e12704c..e6626b1e3 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceNotify.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlert.java @@ -8,20 +8,20 @@ package com.avaje.ebeaninternal.server.lib.sql; * when these events occur on the DataSource. *

*/ -public interface DataSourceNotify { +public interface DataSourceAlert { /** * Send an alert to say the dataSource is back up. */ - public void notifyDataSourceUp(String dataSourceName); + public void dataSourceUp(String dataSourceName); /** * Send an alert to say the dataSource is down. */ - public void notifyDataSourceDown(String dataSourceName); + public void dataSourceDown(String dataSourceName); /** * Send an alert to say the dataSource is getting close to its max size. */ - public void notifyWarning(String subject, String msg); + public void dataSourceWarning(String subject, String msg); } diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlertListener.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlertListener.java deleted file mode 100644 index 97655000c..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlertListener.java +++ /dev/null @@ -1,26 +0,0 @@ -package com.avaje.ebeaninternal.server.lib.sql; - - -/** - * Listens for alerting events such as DataSource down. - */ -public interface DataSourceAlertListener { - - /** - * Send an Alert saying the dataSource is down. - */ - public void dataSourceDown(String dataSourceName); - - /** - * Send an Alert saying the dataSource is back up. - */ - public void dataSourceUp(String dataSourceName); - - /** - * Send an Alert saying the dataSource has reached a high - * number of connections. - */ - public void warning(String subject, String msg); - - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceGlobalManager.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceGlobalManager.java deleted file mode 100644 index df84f3e85..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceGlobalManager.java +++ /dev/null @@ -1,50 +0,0 @@ -package com.avaje.ebeaninternal.server.lib.sql; - -import java.util.List; - -import com.avaje.ebean.config.DataSourceConfig; - -/** - * Manages access to named DataSources using singleton scope. - */ -public final class DataSourceGlobalManager { - - private static final DataSourceManager manager = new DataSourceManager(); - - private DataSourceGlobalManager() { - } - - /** - * Return true when the dataSource is shutting down. - */ - public static boolean isShuttingDown() { - return manager.isShuttingDown(); - } - - /** - * Shutdown the dataSources. - */ - public static void shutdown() { - manager.shutdown(); - } - - /** - * Return the list of DataSourcePool's. - */ - public static List getPools() { - return manager.getPools(); - } - - /** - * Return a DataSource pool by its name. - */ - public static DataSourcePool getDataSource(String name) { - return manager.getDataSource(name); - } - - public static DataSourcePool getDataSource(String name, DataSourceConfig dsConfig) { - return manager.getDataSource(name, dsConfig); - } - - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceManager.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceManager.java deleted file mode 100644 index 7321f9bd0..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceManager.java +++ /dev/null @@ -1,233 +0,0 @@ -package com.avaje.ebeaninternal.server.lib.sql; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.Hashtable; -import java.util.Iterator; -import java.util.List; - -import com.avaje.ebean.config.DataSourceConfig; -import com.avaje.ebean.config.GlobalProperties; -import com.avaje.ebeaninternal.api.ClassUtil; -import com.avaje.ebeaninternal.server.lib.BackgroundRunnable; -import com.avaje.ebeaninternal.server.lib.BackgroundThread; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - - -/** - * Manages access to named DataSources. - */ -public class DataSourceManager implements DataSourceNotify { - - private static final Logger logger = LoggerFactory.getLogger(DataSourceManager.class); - - /** - * An alerter that notifies when the database has problems. - */ - private final DataSourceAlertListener alertlistener; - - /** - * Cache of the named DataSources. - */ - private final Hashtable dsMap = new Hashtable(); - - /** - * Monitor for creating dataSources. - */ - private final Object monitor = new Object(); - - /** - * The database checker registered with BackgroundThread. - */ - private final BackgroundRunnable dbChecker; - - /** - * The frequency to test db while it is up. - */ - private final int dbUpFreqInSecs; - - /** - * The frequency to test db while it is down. - */ - private final int dbDownFreqInSecs; - - /** - * Set to true when shutting down. - */ - private boolean shuttingDown; - - private boolean deregisterDriver; - - /** - * Construct with explicit ConfigProperties. - */ - public DataSourceManager() { - - this.alertlistener = createAlertListener(); - - // perform heart beat every 30 seconds by default - this.dbUpFreqInSecs = GlobalProperties.getInt("datasource.heartbeatfreq",30); - this.dbDownFreqInSecs = GlobalProperties.getInt("datasource.deadbeatfreq",10); - this.dbChecker = new BackgroundRunnable(new Checker(), dbUpFreqInSecs); - this.deregisterDriver = GlobalProperties.getBoolean("datasource.deregisterDriver", true); - - try { - BackgroundThread.add(dbChecker); - - } catch (Exception e) { - logger.error(null, e); - } - } - - private DataSourceAlertListener createAlertListener() throws DataSourceException { - - String alertCN = GlobalProperties.get("datasource.alert.class", null); - if (alertCN == null){ - return new SimpleAlerter(); - - } else { - try { - return (DataSourceAlertListener)ClassUtil.newInstance(alertCN, this.getClass()); - - } catch (Exception ex){ - throw new DataSourceException(ex); - } - } - } - - /** - * Send an alert to say the dataSource is back up. - */ - public void notifyDataSourceUp(String dataSourceName){ - - dbChecker.setFreqInSecs(dbUpFreqInSecs); - - if (alertlistener != null){ - alertlistener.dataSourceUp(dataSourceName); - } - } - - /** - * Send an alert to say the dataSource is down. - */ - public void notifyDataSourceDown(String dataSourceName){ - - dbChecker.setFreqInSecs(dbDownFreqInSecs); - - if (alertlistener != null){ - alertlistener.dataSourceDown(dataSourceName); - } - } - - /** - * Send an alert to say the dataSource is getting close to its max size. - */ - public void notifyWarning(String subject, String msg){ - if (alertlistener != null){ - alertlistener.warning(subject, msg); - } - } - - /** - * Return true when the dataSource is shutting down. - */ - public boolean isShuttingDown() { - synchronized(monitor) { - return shuttingDown; - } - } - - /** - * Shutdown the dataSources. - */ - public void shutdown() { - - synchronized(monitor) { - - this.shuttingDown = true; - - Collection values = dsMap.values(); - for (DataSourcePool ds : values) { - try { - ds.shutdown(); - } catch (DataSourceException e) { - // should never be thrown as the DataSources are all created... - logger.error(null, e); - } - } - if (deregisterDriver){ - for (DataSourcePool ds : values) { - ds.deregisterDriver(); - } - } - } - } - - /** - * Return the DataSourcePool's. - */ - public List getPools() { - synchronized(monitor) { - // create a copy of the DataSourcePool's - ArrayList list = new ArrayList(); - list.addAll(dsMap.values()); - return list; - } - } - - /** - * Get the dataSource using the default ConfigProperties. - */ - public DataSourcePool getDataSource(String name) { - return getDataSource(name, null); - } - - - public DataSourcePool getDataSource(String name, DataSourceConfig dsConfig){ - - if (name == null){ - throw new IllegalArgumentException("name not defined"); - } - - synchronized(monitor){ - DataSourcePool pool = dsMap.get(name); - if (pool == null){ - if (dsConfig == null){ - dsConfig = new DataSourceConfig(); - dsConfig.loadSettings(name); - } - pool = new DataSourcePool(this, name, dsConfig); - dsMap.put(name, pool); - } - return pool; - } - } - - /** - * Check that the database is up by performing a simple query. This should - * be done periodically. By default every 30 seconds. - */ - private void checkDataSource() { - - synchronized (monitor) { - if (!isShuttingDown()) { - Iterator it = dsMap.values().iterator(); - while (it.hasNext()) { - DataSourcePool ds = it.next(); - ds.checkDataSource(); - } - } - } - } - - /** - * Runs every dbUpFreqInSecs to make sure dataSource is up. - */ - private final class Checker implements Runnable { - - public void run() { - checkDataSource(); - } - } -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java index bbefd7fa5..c54d5582e 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java @@ -46,7 +46,7 @@ public class DataSourcePool implements DataSource { /** * Used to notify of changes to the DataSource status. */ - private final DataSourceNotify notify; + private final DataSourceAlert notify; /** * Optional listener that can be notified when connections are got from and @@ -73,6 +73,8 @@ public class DataSourcePool implements DataSource { * The sql used to test a connection. */ private final String heartbeatsql; + + private final int heartbeatFreqSecs; /** * The transaction isolation level as per java.sql.Connection. @@ -154,7 +156,9 @@ public class DataSourcePool implements DataSource { */ private long leakTimeMinutes; - public DataSourcePool(DataSourceNotify notify, String name, DataSourceConfig params) { + private final Runnable heartbeatRunnable = new HeartBeatRunnable(); + + public DataSourcePool(DataSourceAlert notify, String name, DataSourceConfig params) { this.notify = notify; this.name = name; @@ -175,7 +179,8 @@ public class DataSourcePool implements DataSource { this.maxConnections = params.getMaxConnections(); this.waitTimeoutMillis = params.getWaitTimeoutMillis(); this.heartbeatsql = params.getHeartbeatSql(); - + this.heartbeatFreqSecs = params.getHeartbeatFreqSecs(); + queue = new PooledConnectionQueue(this); String un = params.getUsername(); @@ -204,7 +209,15 @@ public class DataSourcePool implements DataSource { throw new DataSourceException(ex); } } + + class HeartBeatRunnable implements Runnable { + @Override + public void run() { + checkDataSource(); + } + } + @Override public java.util.logging.Logger getParentLogger() throws SQLFeatureNotSupportedException { throw new SQLFeatureNotSupportedException("We do not support java.util.logging"); @@ -296,7 +309,7 @@ public class DataSourcePool implements DataSource { logger.warn(msg); if (notify != null) { String subject = "DataSourcePool [" + name + "] warning"; - notify.notifyWarning(subject, msg); + notify.dataSourceWarning(subject, msg); } } } @@ -304,10 +317,9 @@ public class DataSourcePool implements DataSource { private void notifyDataSourceIsDown(SQLException ex) { if (!dataSourceDownAlertSent) { - String msg = "FATAL: DataSourcePool [" + name + "] is down!!!"; - logger.error(msg, ex); + logger.error("FATAL: DataSourcePool [" + name + "] is down!!!", ex); if (notify != null) { - notify.notifyDataSourceDown(name); + notify.dataSourceDown(name); } dataSourceDownAlertSent = true; @@ -320,15 +332,14 @@ public class DataSourcePool implements DataSource { private void notifyDataSourceIsUp() { if (dataSourceDownAlertSent) { - String msg = "RESOLVED FATAL: DataSourcePool [" + name + "] is back up!"; - logger.error(msg); + logger.error("RESOLVED FATAL: DataSourcePool [" + name + "] is back up!"); if (notify != null) { - notify.notifyDataSourceUp(name); + notify.dataSourceUp(name); } dataSourceDownAlertSent = false; } else if (!dataSourceUp) { - logger.warn("DataSourcePool [" + name + "] is back up!"); + logger.info("DataSourcePool [" + name + "] is back up!"); } if (!dataSourceUp) { @@ -336,14 +347,36 @@ public class DataSourcePool implements DataSource { reset(); } } + + + /** + * Return the heartbeat frequency in seconds. + *

+ * This is the frequency that the heartbeat runnable should be run. + *

+ */ + public int getHeartbeatFreqSecs() { + return heartbeatFreqSecs; + } + + /** + * Returns the Runnable used to check the dataSource using a heartbeat query. + */ + public Runnable getHeartbeatRunnable() { + return heartbeatRunnable; + } /** * Check the dataSource is up. Trim connections. + *

+ * This is called by the HeartbeatRunnable which should be scheduled to + * run periodically (every heartbeatFreqSecs seconds actually). + *

*/ - protected void checkDataSource() { + public void checkDataSource() { Connection conn = null; try { - // test to see if we can create a new connection... + // Get a connection from the pool and test it conn = getConnection(); testConnection(conn); @@ -356,6 +389,7 @@ public class DataSourcePool implements DataSource { } catch (SQLException ex) { notifyDataSourceIsDown(ex); + } finally { try { if (conn != null) { @@ -506,7 +540,7 @@ public class DataSourcePool implements DataSource { protected boolean validateConnection(PooledConnection conn) { try { if (heartbeatsql == null) { - logger.info("Can not test connection as heartbeatsql is not set"); + logger.debug("Can not test connection as heartbeatsql is not set"); return false; } @@ -514,8 +548,7 @@ public class DataSourcePool implements DataSource { return true; } catch (Exception e) { - String desc = "heartbeatsql test failed on connection[" + conn.getName() + "]"; - logger.warn(desc); + logger.warn("heartbeatsql test failed on connection[" + conn.getName() + "]"); return false; } } @@ -582,7 +615,7 @@ public class DataSourcePool implements DataSource { * Grow the pool by creating a new connection. The connection can either be * added to the available list, or returned. *

- * This method is protected by synchronisation in calling methods. + * This method is protected by synchronization in calling methods. *

*/ protected PooledConnection createConnectionForQueue(int connId) throws SQLException { @@ -661,7 +694,7 @@ public class DataSourcePool implements DataSource { String msg = "Just testing if alert message is sent successfully."; if (notify != null) { - notify.notifyWarning(subject, msg); + notify.dataSourceWarning(subject, msg); } } @@ -674,8 +707,11 @@ public class DataSourcePool implements DataSource { * Connections are not waited on, as that would hang the server. *

*/ - public void shutdown() { + public void shutdown(boolean deregisterDriver) { queue.shutdown(); + if (deregisterDriver){ + deregisterDriver(); + } } /** @@ -804,12 +840,10 @@ public class DataSourcePool implements DataSource { */ public void deregisterDriver() { try { + logger.debug("Deregistered the JDBC driver "+this.databaseDriver); DriverManager.deregisterDriver(DriverManager.getDriver(this.databaseUrl)); - String msg = "Deregistered the JDBC driver "+this.databaseDriver; - logger.debug(msg); } catch (SQLException e) { - String msg = "Error trying to deregister the JDBC driver "+this.databaseDriver; - logger.warn(msg, e); + logger.warn("Error trying to deregister the JDBC driver "+this.databaseDriver, e); } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java index e0074771e..5b29b7fb5 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java @@ -256,11 +256,13 @@ public class PooledConnection extends ConnectionDelegator String msg = "Closing Connection[" + getName() + "]" + " psReuse[" + pstmtHitCounter + "] psCreate[" + pstmtMissCounter + "] psSize[" + pstmtCache.size() + "]"; - logger.info(msg); + logger.debug(msg); try { if (connection.isClosed()) { - logger.warn("Closing Connection[" + getName() + "] that is already closed?"); + // Typically the JDBC Driver has its own JVM shutdown hook and already + // closed the connections in our DataSource pool so making this DEBUG level + logger.debug("Closing Connection[" + getName() + "] that is already closed?"); return; } } catch (SQLException ex) { diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java index 8ef09780a..b7daa3942 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java @@ -273,7 +273,7 @@ public class PooledConnectionQueue { int busySize = registerBusyConnection(c); String msg = "DataSourcePool [" + name + "] grow; id["+c.getName()+"] busy["+busySize+"] max["+maxSize+"]"; - logger.info(msg); + logger.debug(msg); checkForWarningSize(); return c; @@ -333,7 +333,7 @@ public class PooledConnectionQueue { try { doingShutdown = true; Status status = createStatus(); - logger.info("DataSourcePool [" + name + "] shutdown: "+status); + logger.debug("DataSourcePool [" + name + "] shutdown: "+status); closeFreeConnections(true); @@ -422,7 +422,7 @@ public class PooledConnectionQueue { freeList.setShallowCopy(freeListCopy); String msg = "DataSourcePool [" + name + "] trimmed [" + trimedCount + "] inactive connections. New size[" + totalConnections() + "]"; - logger.info(msg); + logger.debug(msg); } return trimedCount; } @@ -436,7 +436,7 @@ public class PooledConnectionQueue { try { while (!freeList.isEmpty()) { PooledConnection c = freeList.remove(); - logger.info("PSTMT Statistics: "+c.getStatistics()); + logger.debug("PSTMT Statistics: "+c.getStatistics()); c.closeConnectionFully(logErrors); } } finally { diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleAlerter.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleAlerter.java deleted file mode 100644 index 6eb587c76..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleAlerter.java +++ /dev/null @@ -1,107 +0,0 @@ -package com.avaje.ebeaninternal.server.lib.sql; - -import com.avaje.ebean.config.GlobalProperties; -import com.avaje.ebeaninternal.server.lib.util.MailEvent; -import com.avaje.ebeaninternal.server.lib.util.MailListener; -import com.avaje.ebeaninternal.server.lib.util.MailMessage; -import com.avaje.ebeaninternal.server.lib.util.MailSender; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * A simple smtp email alert that sends a email message - * on dataSourceDown and dataSourceUp etc. - *
    - *
  • alert.fromuser = the from user name - *
  • alert.fromemail = the from email account - *
  • alert.toemail = comma delimited list of email accounts to email - *
  • alert.mailserver = the smpt server name - *
- */ -public class SimpleAlerter implements DataSourceAlertListener, MailListener { - - private static final Logger logger = LoggerFactory.getLogger(SimpleAlerter.class); - - //boolean sendInBackGround = true; - - /** - * Create a SimpleAlerter. - */ - public SimpleAlerter() { - } - - /** - * If the email failed then log the error. - */ - public void handleEvent(MailEvent event) { - Throwable e = event.getError(); - if (e != null){ - logger.error(null, e); - } - } - - /** - * Send the dataSource down alert. - */ - public void dataSourceDown(String dataSourceName) { - String msg = getSubject(true, dataSourceName); - sendMessage(msg, msg); - } - - /** - * Send the dataSource up alert. - */ - public void dataSourceUp(String dataSourceName) { - String msg = getSubject(false, dataSourceName); - sendMessage(msg, msg); - } - - /** - * Send the warning message. - */ - public void warning(String subject, String msg) { - sendMessage(subject, msg); - } - - private String getSubject(boolean isDown, String dsName) { - String msg = "The DataSource "+dsName; - if (isDown){ - msg += " is DOWN!!"; - } else { - msg += " is UP."; - } - return msg; - } - - private void sendMessage(String subject, String msg){ - - String fromUser = GlobalProperties.get("alert.fromuser", null); - String fromEmail = GlobalProperties.get("alert.fromemail", null); - String mailServerName = GlobalProperties.get("alert.mailserver", null); - String toEmail = GlobalProperties.get("alert.toemail", null); - - if (mailServerName == null){ - //throw new RuntimeException("alert.mailserver not set..."); - return; - } - - MailMessage data = new MailMessage(); - data.setSender(fromUser, fromEmail); - data.addBodyLine(msg); - data.setSubject(subject); - - String[] toList = toEmail.split(","); - if (toList.length==0) { - throw new RuntimeException("alert.toemail has not been set?"); - } - for (int i = 0; i < toList.length; i++) { - data.addRecipient(null, toList[i].trim()); - } - - MailSender sender = new MailSender(mailServerName); - sender.setMailListener(this); - sender.sendInBackground(data); - } - - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleDataSourceAlert.java b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleDataSourceAlert.java new file mode 100644 index 000000000..17ff55537 --- /dev/null +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleDataSourceAlert.java @@ -0,0 +1,109 @@ +package com.avaje.ebeaninternal.server.lib.sql; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.avaje.ebean.config.GlobalProperties; +import com.avaje.ebeaninternal.server.lib.util.MailEvent; +import com.avaje.ebeaninternal.server.lib.util.MailListener; +import com.avaje.ebeaninternal.server.lib.util.MailMessage; +import com.avaje.ebeaninternal.server.lib.util.MailSender; + +/** + * A simple smtp email alert that sends a email message on dataSourceDown and + * dataSourceUp etc. + *
    + *
  • alert.fromuser = the from user name + *
  • alert.fromemail = the from email account + *
  • alert.toemail = comma delimited list of email accounts to email + *
  • alert.mailserver = the smpt server name + *
+ */ +public class SimpleDataSourceAlert implements DataSourceAlert, MailListener { + + private static final Logger logger = LoggerFactory.getLogger(SimpleDataSourceAlert.class); + + // boolean sendInBackGround = true; + + /** + * Create a SimpleAlerter. + */ + public SimpleDataSourceAlert() { + } + + /** + * If the email failed then log the error. + */ + public void handleEvent(MailEvent event) { + Throwable e = event.getError(); + if (e != null) { + logger.error(null, e); + } + } + + /** + * Send the dataSource down alert. + */ + @Override + public void dataSourceDown(String dataSourceName) { + String msg = getSubject(true, dataSourceName); + sendMessage(msg, msg); + } + + /** + * Send the dataSource up alert. + */ + @Override + public void dataSourceUp(String dataSourceName) { + String msg = getSubject(false, dataSourceName); + sendMessage(msg, msg); + } + + /** + * Send the warning message. + */ + @Override + public void dataSourceWarning(String subject, String msg) { + sendMessage(subject, msg); + } + + private String getSubject(boolean isDown, String dsName) { + String msg = "The DataSource " + dsName; + if (isDown) { + msg += " is DOWN!!"; + } else { + msg += " is UP."; + } + return msg; + } + + private void sendMessage(String subject, String msg) { + + String mailServerName = GlobalProperties.get("datasource.alert.mailserver", null); + if (mailServerName == null) { + return; + } + + String fromUser = GlobalProperties.get("datasource.alert.fromuser", null); + String fromEmail = GlobalProperties.get("datasource.alert.fromemail", null); + String toEmail = GlobalProperties.get("datasource.alert.toemail", null); + + MailMessage data = new MailMessage(); + data.setSender(fromUser, fromEmail); + data.addBodyLine(msg); + data.setSubject(subject); + + String[] toList = toEmail.split(","); + if (toList.length == 0) { + throw new RuntimeException("alert.toemail has not been set?"); + } + for (int i = 0; i < toList.length; i++) { + data.addRecipient(null, toList[i].trim()); + } + + MailSender sender = new MailSender(mailServerName); + sender.setMailListener(this); + sender.sendInBackground(data); + } + +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/thread/PooledThread.java b/src/main/java/com/avaje/ebeaninternal/server/lib/thread/PooledThread.java index 79aaffa41..201231152 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/thread/PooledThread.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/thread/PooledThread.java @@ -9,251 +9,270 @@ import org.slf4j.LoggerFactory; */ public class PooledThread implements Runnable { - private static final Logger logger = LoggerFactory.getLogger(PooledThread.class); - - /** - * Create the PooledThread. - */ - protected PooledThread(ThreadPool threadPool, String name, boolean isDaemon, - Integer threadPriority) { + private static final Logger logger = LoggerFactory.getLogger(PooledThread.class); - this.name = name; - this.threadPool = threadPool; - this.lastUsedTime = System.currentTimeMillis(); + /** + * Flag to indicate that the thread was interrupted. + */ + private boolean wasInterrupted; - thread = new Thread(this, name); - thread.setDaemon(isDaemon); + /** + * The time the thread was last used. + */ + private long lastUsedTime; - if (threadPriority != null) { - thread.setPriority(threadPriority.intValue()); - } - //thread.start(); + /** + * The work to run + */ + private Work work; + + /** + * Set to indicate the thread is stopping. + */ + private boolean isStopping; + + /** + * Set when the thread has stopped. + */ + private boolean isStopped; + /** + * The background thread. + */ + private Thread thread; + + /** + * The pool this worker belongs to. + */ + private ThreadPool threadPool; + + /** + * The name of the Thread + */ + private String name; + + /** + * The thread synchronization object. + */ + private Object threadMonitor = new Object(); + + /** + * The monitor for work notification. + */ + private Object workMonitor = new Object(); + + /** + * Total number of work performed. + */ + private int totalWorkCount; + + /** + * Total work execution time. + */ + private long totalWorkExecutionTime; + + /** + * Create the PooledThread. + */ + protected PooledThread(ThreadPool threadPool, String name, boolean isDaemon, Integer threadPriority) { + + this.name = name; + this.threadPool = threadPool; + this.lastUsedTime = System.currentTimeMillis(); + + thread = new Thread(this, name); + thread.setDaemon(isDaemon); + + if (threadPriority != null) { + thread.setPriority(threadPriority.intValue()); } - - protected void start() { - thread.start(); - } - - /** - * Assign work to this thread. The thread will notify the listener when it - * has finished the work. - */ - public boolean assignWork(Work work) { - synchronized (workMonitor) { - this.work = work; - workMonitor.notifyAll(); - } - return true; + } + + public String toString() { + return name; + } + + protected void start() { + thread.start(); + } + + /** + * Assign work to this thread. The thread will notify the listener when it has + * finished the work. + */ + public boolean assignWork(Work work) { + synchronized (workMonitor) { + this.work = work; + workMonitor.notifyAll(); } + return true; + } - /** - * process any assigned work until stopped or interrupted. - */ - public void run() { - // process assigned work until we receive a shutdown signal - synchronized (workMonitor) { - while (!isStopping) { - try { - if (work == null) { - workMonitor.wait(); - } - } catch (InterruptedException e) { - } - doTheWork(); - } - } - - // Tell stop() we have shut ourselves down successfully - synchronized (threadMonitor) { - threadMonitor.notifyAll(); - } - //Log.debug("PooledThread [" + getName() + "] finished "); - isStopped = true; - } - - /** - * Actually do the work and gather the appropriate measures. - */ - private void doTheWork() { - if (isStopping){ - return; - } - - long startTime = System.currentTimeMillis(); - if (work == null) { - // probably shutting down the thread - - } else { - try { - work.setStartTime(startTime); - work.getRunnable().run(); - - } catch (Throwable ex) { - logger.error(null, ex); - - if (wasInterrupted) { - this.isStopping = true; - threadPool.removeThread(this); - logger.info("PooledThread [" + name + "] removed due to interrupt"); - try { - thread.interrupt(); - } catch (Exception e){ - String msg = "Error interrupting PooledThead["+name+"]"; - logger.error(msg, e); - } - return; - } - } - } - lastUsedTime = System.currentTimeMillis(); - totalWorkCount++; - totalWorkExecutionTime = totalWorkExecutionTime + lastUsedTime - startTime; - this.work = null; - threadPool.returnThread(this); - - } - - /** - * Try to interrupt the thread. - *

- * If the Thread was interrupted then it will be removed from the pool. - *

- */ - public void interrupt() { - - // set a flag so doTheWork knows that it was interrupted - // and removes rather than returns - wasInterrupted = true; + /** + * process any assigned work until stopped or interrupted. + */ + public void run() { + // process assigned work until we receive a shutdown signal + synchronized (workMonitor) { + while (!isStopping) { try { - thread.interrupt(); - - } catch (SecurityException ex) { - wasInterrupted = false; - throw ex; + if (work == null) { + workMonitor.wait(); + } + } catch (InterruptedException e) { } + doTheWork(); + } } - - /** - * Returns true if the thread has finished. - */ - public boolean isStopped() { - return isStopped; + + // Tell stop() we have shut ourselves down successfully + synchronized (threadMonitor) { + threadMonitor.notifyAll(); } - - /** - * Stop the thread relatively nicely. It will wait a maximum of 10 seconds - * for it to complete any existing work. - */ - protected void stop() { - isStopping = true; + if (logger.isTraceEnabled()) { + logger.trace("PooledThread [" + getName() + "] finished "); + } + isStopped = true; + } + + /** + * Actually do the work and gather the appropriate measures. + */ + private void doTheWork() { + if (isStopping) { + return; + } + + long startTime = System.currentTimeMillis(); + if (work == null) { + // probably shutting down the thread + + } else { + try { + if (logger.isTraceEnabled()) { + logger.trace("start work "+work); + } - synchronized (threadMonitor) { - - assignWork(null); - //trace("stop assigned null work..."); - try { - threadMonitor.wait(10000); - } catch (InterruptedException e) { - ; - } - + work.setStartTime(startTime); + work.getRunnable().run(); + + if (logger.isTraceEnabled()) { + logger.trace("finished work "+work); } + + } catch (Throwable ex) { + logger.error(null, ex); + + if (wasInterrupted) { + this.isStopping = true; + threadPool.removeThread(this); + if (logger.isInfoEnabled()) { + logger.info("PooledThread [" + name + "] removed due to interrupt"); + } + try { + thread.interrupt(); + } catch (Exception e) { + logger.error("Error interrupting PooledThead[" + name + "]", e); + } + return; + } + } + } + lastUsedTime = System.currentTimeMillis(); + totalWorkCount++; + totalWorkExecutionTime = totalWorkExecutionTime + lastUsedTime - startTime; + this.work = null; + threadPool.returnThread(this); + + } + + /** + * Try to interrupt the thread. + *

+ * If the Thread was interrupted then it will be removed from the pool. + *

+ */ + public void interrupt() { + + // set a flag so doTheWork knows that it was interrupted + // and removes rather than returns + wasInterrupted = true; + try { + if (logger.isTraceEnabled()) { + logger.trace("interrupt()"); + } + thread.interrupt(); + + } catch (SecurityException ex) { + wasInterrupted = false; + throw ex; + } + } + + /** + * Returns true if the thread has finished. + */ + public boolean isStopped() { + return isStopped; + } + + /** + * Stop the thread relatively nicely. It will wait a maximum of 10 seconds for + * it to complete any existing work. + */ + protected void stop() { + isStopping = true; + + synchronized (threadMonitor) { + + assignWork(null); + if (logger.isTraceEnabled()) { + logger.trace("stopping thread ["+name+"]"); + } + try { + threadMonitor.wait(10000); + } catch (InterruptedException e) { + ; + } - thread = null; - threadPool.removeThread(this); } - /** - * return the name of the thread. - */ - public String getName() { - return name; - } - /** - * Returns the currently executing work, otherwise null. - */ - public Work getWork() { - return work; - } + thread = null; + threadPool.removeThread(this); + } - /** - * The total number of jobs this thread has run. - */ - public int getTotalWorkCount() { - return totalWorkCount; - } + /** + * return the name of the thread. + */ + public String getName() { + return name; + } - /** - * The total time for performing all assigned work. - */ - public long getTotalWorkExecutionTime() { - return totalWorkExecutionTime; - } - - /** - * Returns the time this thread was last used. - */ - public long getLastUsedTime() { - return lastUsedTime; - } + /** + * Returns the currently executing work, otherwise null. + */ + public Work getWork() { + return work; + } - /** - * Flag to indicate that the thread was interrupted. - */ - private boolean wasInterrupted = false; + /** + * The total number of jobs this thread has run. + */ + public int getTotalWorkCount() { + return totalWorkCount; + } - /** - * The time the thread was last used. - */ - private long lastUsedTime; + /** + * The total time for performing all assigned work. + */ + public long getTotalWorkExecutionTime() { + return totalWorkExecutionTime; + } - /** - * The work to run - */ - private Work work = null; - - /** - * Set to indicate the thread is stopping. - */ - private boolean isStopping = false; - - /** - * Set when the thread has stopped. - */ - private boolean isStopped = false; - /** - * The background thread. - */ - private Thread thread = null; - - /** - * The pool this worker belongs to. - */ - private ThreadPool threadPool; - - /** - * The name of the Thread - */ - private String name = null; - - /** - * The thread synchronization object. - */ - private Object threadMonitor = new Object(); - - /** - * The monitor for work notification. - */ - private Object workMonitor = new Object(); - - /** - * Total number of work performed. - */ - private int totalWorkCount = 0; - - /** - * Total work execution time. - */ - private long totalWorkExecutionTime = 0; + /** + * Returns the time this thread was last used. + */ + public long getLastUsedTime() { + return lastUsedTime; + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPool.java b/src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPool.java index 43822b687..76b123c12 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPool.java +++ b/src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPool.java @@ -3,6 +3,8 @@ package com.avaje.ebeaninternal.server.lib.thread; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.avaje.ebean.config.GlobalProperties; + import java.util.ArrayList; import java.util.Iterator; import java.util.Vector; @@ -17,457 +19,484 @@ import java.util.Vector; */ public class ThreadPool { - private static final Logger logger = LoggerFactory.getLogger(ThreadPool.class); - - /** - * The max idle time used to trim idle threads from the pool. - */ - private long maxIdleTime; + private static final Logger logger = LoggerFactory.getLogger(ThreadPool.class); - /** - * The name of the pool - */ - private String poolName; + /** + * The max idle time used to trim idle threads from the pool. + */ + private long maxIdleTime; - /** - * The initial pool size. - */ - private int minSize; + /** + * The name of the pool + */ + private String poolName; - /** - * Whether or not the threads are going to be Daemon threads. - */ - private boolean isDaemon; + /** + * The initial pool size. + */ + private int minSize; - /** - * Flag to indicate that the pool is being shutdown. - */ - private boolean isStopping = false; + /** + * Whether or not the threads are going to be Daemon threads. + */ + private boolean isDaemon; - /** - * The priority or the threads. Can be null, in which case the threads have - * the default priority. - */ - private Integer threadPriority; + /** + * The priority or the threads. Can be null, in which case the threads have + * the default priority. + */ + private Integer threadPriority; - /** - * Incrementing int for thread name. NB: currentThreadCount will go up and - * down as the pool grows and shrinks. - */ - private int uniqueThreadID; + /** + * Incrementing int for thread name. NB: currentThreadCount will go up and + * down as the pool grows and shrinks. + */ + private int uniqueThreadID; - /** - * List of PooledThread that are free for work. - */ - private Vector freeList = new Vector(); + /** + * The maximum number of threads to grow to. Hitting this limit will have + * performance ramifications. + */ + private int maxSize = 100; - /** - * List of PooledThread that are busy. - */ - private Vector busyList = new Vector(); + /** + * Flag that the pool should terminate all the threads and stop. + */ + private boolean stopThePool; - /** - * List holding queued work. - */ - private Vector workOverflowQueue = new Vector(); + /** + * Flag to indicate that the pool is being shutdown. + */ + private boolean isStopping; - /** - * The maximum number of threads to grow to. Hitting this limit will have - * performance ramifications. - */ - private int maxSize = 100; + /** + * List of PooledThread that are free for work. + */ + private Vector freeList = new Vector(); - /** - * Flag that the pool should terminate all the threads and stop. - */ - private boolean stopThePool; + /** + * List of PooledThread that are busy. + */ + private Vector busyList = new Vector(); - - /** - * Create the ThreadPool. - */ - public ThreadPool(String poolName, boolean isDaemon, Integer threadPriority) { + /** + * List holding queued work. + */ + private Vector workOverflowQueue = new Vector(); - this.poolName = poolName; - this.stopThePool = false; - this.isDaemon = isDaemon; - this.threadPriority = threadPriority; + /** + * Create and return a ThreadPool reading configuration from GlobalProperties. + */ + public static ThreadPool createThreadPool(String poolName) { + + int min = GlobalProperties.getInt("threadpool." + poolName + ".min", 0); + int max = GlobalProperties.getInt("threadpool." + poolName + ".max", 100); + + int defaultIdleTime = GlobalProperties.getInt("threadpool.idletime", 60); + long idle = 1000 * GlobalProperties.getInt("threadpool." + poolName + ".idletime", defaultIdleTime); + + Integer priority = null; + String threadPriority = GlobalProperties.get("threadpool." + poolName + ".priority", null); + if (threadPriority != null) { + priority = new Integer(threadPriority); } - /** - * Return true if the pool is shutting down. - */ - public boolean isStopping() { - return isStopping; + return new ThreadPool(poolName, true, priority, min, max, idle); + } + + public ThreadPool(String poolName, boolean isDaemon, Integer threadPriority, int minSize, int maxSize, long maxIdleMillis) { + this(poolName, isDaemon, threadPriority); + this.minSize = minSize; + this.maxSize = maxSize; + this.maxIdleTime = maxIdleMillis; + } + + /** + * Create the ThreadPool. + */ + public ThreadPool(String poolName, boolean isDaemon, Integer threadPriority) { + + this.poolName = poolName; + this.isDaemon = isDaemon; + this.threadPriority = threadPriority; + } + + /** + * Return true if the pool is shutting down. + */ + public boolean isStopping() { + synchronized (freeList) { + return isStopping; + } + } + + /** + * Return the name of the thread pool. + */ + public String getName() { + return poolName; + } + + /** + * Set the minimum size the pool should try to maintain. + */ + public void setMinSize(int minSize) { + if (minSize > 0) { + if (minSize > maxSize) { + this.maxSize = minSize; + } + this.minSize = minSize; + maintainPoolSize(); + } + } + + /** + * Return the minimum size the pool should maintain. + */ + public int getMinSize() { + return minSize; + } + + /** + * Set the maximum size the pool should grow to. + */ + public void setMaxSize(int maxSize) { + if (maxSize > 0) { + if (minSize > maxSize) { + minSize = maxSize; + } + this.maxSize = maxSize; + maintainPoolSize(); + } + } + + /** + * Return the maximum size this pool can grow to. + */ + public int getMaxSize() { + return maxSize; + } + + /** + * Return the total number of busy and free threads in the pool. + */ + public int size() { + return busyList.size() + freeList.size(); + } + + /** + * Return the number of currently busy threads. + */ + public int getBusyCount() { + return busyList.size(); + } + + /** + * Assign a task to the thread pool, specifing the options to wait or queue + * the task if the pool is fully busy and can't grow. + *

+ * When the pool is fully busy... + *

+ *

+ * addToQueue=true -> work is added to queue, returns false
+ * addToQueue=false -> work is not done or queued, returns false
+ *

+ * + * @param work + * the runnable work to do. + * @param addToQueueIfFull + * If the pool is maxed out and this is true then it queues the + * Runnable. + */ + public boolean assign(Runnable work, boolean addToQueueIfFull) { + + if (stopThePool) { + throw new RuntimeException("Pool is stopping... no more work please."); } - /** - * Return the name of the thread pool. - */ - public String getName() { - return poolName; - } + Work runWork = new Work(work); - /** - * Set the minimum size the pool should try to maintain. - */ - public void setMinSize(int minSize) { - if (minSize > 0) { - if (minSize > maxSize) { - this.maxSize = minSize; - } - this.minSize = minSize; - maintainPoolSize(); + // get the next available thread in the pool (block) + PooledThread thread = getNextAvailableThread(); + if (thread != null) { + // assign the work to that thread + busyList.add(thread); + thread.assignWork(runWork); + return true; + + } else { + if (addToQueueIfFull) { + runWork.setEnterQueueTime(System.currentTimeMillis()); + workOverflowQueue.add(runWork); + } + return false; + } + } + + /** + * Remove the thread from the pool. The thread should be stopped before it is + * removed. + */ + protected void removeThread(PooledThread thread) { + synchronized (freeList) { + busyList.remove(thread); + freeList.remove(thread); + freeList.notify(); + + if (logger.isTraceEnabled()) { + logger.trace("PooledThread stopped [" + getName() + "]"); + } + } + } + + /** + * fired when a Thread from the pool has finished, and can be put back into + * the pool. + */ + protected void returnThread(PooledThread thread) { + + synchronized (freeList) { + + // deregister from the busyList + busyList.remove(thread); + + if (!workOverflowQueue.isEmpty()) { + // get the first bit of work off the queue + Work queuedWork = (Work) workOverflowQueue.remove(0); + + // work out the queue time and counts etc + queuedWork.setExitQueueTime(System.currentTimeMillis()); + busyList.add(thread); + thread.assignWork(queuedWork); + + } else { + if (logger.isTraceEnabled()) { + logger.trace("returnThread - add to freeList [" + thread + "]"); } + // put the thread back onto the available list + freeList.add(thread); + // tell shutdown() one has returned + freeList.notify(); + } } + } - /** - * Return the minimum size the pool should maintain. - */ - public int getMinSize() { - return minSize; + /** + * Get the next available thread. Block until thread is available. NB: The + * dispatcher is blocked but work can still be assigned to the dispatcher in a + * non-blocking way + */ + private PooledThread getNextAvailableThread() { + + synchronized (freeList) { + if (!freeList.isEmpty()) { + return (PooledThread) freeList.remove(0); + } + if (size() < maxSize) { + return growPool(true); + } + return null; } + } - /** - * Set the maximum size the pool should grow to. - */ - public void setMaxSize(int maxSize) { - if (maxSize > 0) { - if (minSize > maxSize) { - minSize = maxSize; - } - this.maxSize = maxSize; - maintainPoolSize(); - } + /** + * Return an Iterator of PooledThread that are currently running. You should + * only use this for display. Use the getPooledThread() or interrupt() methods + * to interrupt a particular thread. + * + * @return an Iterator of busy PooledThread's. + */ + public Iterator getBusyThreads() { + synchronized (freeList) { + return busyList.iterator(); } + } + + /** + * Shutdown the threadpool stopping all the threads. This will wait until any + * busy threads have finished their assigned work. + */ + public void shutdown() { + + synchronized (freeList) { + if (isStopping) { + logger.debug("already shutting down"); + } + isStopping = true; - /** - * Return the maximum size this pool can grow to. - */ - public int getMaxSize() { - return maxSize; - } + if (logger.isDebugEnabled()) { + logger.debug("ThreadPool [" + poolName + "] Shutting down; threadCount[" + size()+ "] busyCount[" + getBusyCount() + "]"); + } + stopThePool = true; - /** - * Return the total number of busy and free threads in the pool. - */ - public int size() { - return busyList.size() + freeList.size(); - } + while (!freeList.isEmpty()) { + PooledThread thread = (PooledThread) freeList.remove(0); + thread.stop(); + } - /** - * Return the number of currently busy threads. - */ - public int getBusyCount() { - return busyList.size(); - } + try { + while (getBusyCount() > 0) { - /** - * Assign a task to the thread pool, specifing the options to wait or queue - * the task if the pool is fully busy and can't grow. - *

- * When the pool is fully busy... - *

- *

- * addToQueue=true -> work is added to queue, returns false
- * addToQueue=false -> work is not done or queued, returns false
- *

- * - * @param work the runnable work to do. - * @param addToQueueIfFull If the pool is maxed out and this is true then it - * queues the Runnable. - */ - public boolean assign(Runnable work, boolean addToQueueIfFull) { + String msg = "ThreadPool [" + poolName + "] has [" + getBusyCount()+ "] busy threads, waiting for those to finish."; + logger.info(msg); - if (stopThePool) { - throw new RuntimeException("Pool is stopping... no more work please."); + Iterator busyThreads = getBusyThreads(); + while (busyThreads.hasNext()) { + PooledThread busyThread = (PooledThread) busyThreads.next(); + String busymsg = "Busy thread [" + busyThread.getName() + "] work["+ busyThread.getWork() + "]"; + logger.info(busymsg); + } + + freeList.wait(); + PooledThread thread = (PooledThread) freeList.remove(0); + logger.debug("wait finished on thread[" + thread.getName() + "]"); + if (thread != null) { + thread.stop(); + } } - Work runWork = new Work(work); - - // get the next available thread in the pool (block) - PooledThread thread = getNextAvailableThread(); - if (thread != null) { - // assign the work to that thread - busyList.add(thread); - thread.assignWork(runWork); - return true; - - } else { - if (addToQueueIfFull) { - runWork.setEnterQueueTime(System.currentTimeMillis()); - workOverflowQueue.add(runWork); - } - return false; - } + } catch (InterruptedException e) { + logger.error("Error during threadpool shutdown", e); + } } + } - /** - * Remove the thread from the pool. The thread should be stopped before it - * is removed. - */ - protected void removeThread(PooledThread thread) { - synchronized (freeList) { - busyList.remove(thread); - freeList.remove(thread); - freeList.notify(); - - // if (ThreadPoolManager.getDebugLevel()>0){ - //Log.debug("PooledThread stopped [" + getName() + "]"); - // } + /** + * Trim or grow the pool leaving at least min free. + */ + protected void maintainPoolSize() { + synchronized (freeList) { + if (isStopping) { + // don't bother as the pool is shutting down + return; + } + + int numToStop = size() - minSize; + if (numToStop > 0) { + // should trim idle threads as we are over the minSize + long usedAfter = System.currentTimeMillis() - maxIdleTime; + ArrayList stopList = new ArrayList(); + Iterator it = freeList.iterator(); + while (it.hasNext() && numToStop > 0) { + PooledThread thread = (PooledThread) it.next(); + if (thread.getLastUsedTime() < usedAfter) { + stopList.add(thread); + numToStop--; + } } + Iterator stopIt = stopList.iterator(); + while (stopIt.hasNext()) { + PooledThread thread = (PooledThread) stopIt.next(); + if (logger.isDebugEnabled()) { + logger.debug("trimming pool - stopping thread "+thread); + } + thread.stop(); + } + } + int numToAdd = minSize - size(); + if (numToAdd > 0) { + // should add some more to the pool + for (int i = 0; i < numToAdd; i++) { + growPool(false); + } + } } + } - /** - * fired when a Thread from the pool has finished, and can be put back into - * the pool. - */ - protected void returnThread(PooledThread thread) { - - synchronized (freeList) { - - // deregister from the busyList - busyList.remove(thread); - - if (!workOverflowQueue.isEmpty()) { - // get the first bit of work off the queue - Work queuedWork = (Work) workOverflowQueue.remove(0); - - // work out the queue time and counts etc - queuedWork.setExitQueueTime(System.currentTimeMillis()); - busyList.add(thread); - thread.assignWork(queuedWork); - - } else { - // put the thread back onto the available list - freeList.add(thread); - // tell shutdown() one has returned - freeList.notify(); - } - } + /** + * Interrupt a named thread that is currently busy. + *

+ * Returns the thread that was interrupted or null if the thread was not + * found. If the thread was interrupted then it will automatically be stopped + * and removed from the pool. + *

+ *

+ * Note that it may take some time to actually interrupt the thread so an + * immediate test to see if the thread stopped will probably be wrong. + * + * 

+   * 
+   * ThreadPool test = ThreadPoolManager.getThreadPool("test");
+   * PooledThread pt = test.interrupt("test.1");
+   * if (pt == null) {
+   *      // the thread was not found, perhaps finished?
+   * } else {
+   *      // give interrupt a little time to execute
+   *      Thread.sleep(1000);
+   *      boolean hasStopped = pt.isStopped();
+   *      //..
+   * }
+   * 
+   *
+ * + *

+ * + * @return the thread that was interrupted + */ + public PooledThread interrupt(String threadName) { + PooledThread thread = getBusyThread(threadName); + if (thread != null) { + thread.interrupt(); + return thread; } + return null; + } - - /** - * Get the next available thread. Block until thread is available. NB: The - * dispatcher is blocked but work can still be assigned to the dispatcher in - * a non-blocking way - */ - private PooledThread getNextAvailableThread() { - - synchronized (freeList) { - if (!freeList.isEmpty()) { - return (PooledThread) freeList.remove(0); - } - if (size() < maxSize) { - return growPool(true); - } - return null; + /** + * Find a thread using its name from the busy list. Returns null if the thread + * is not found in the busy list. + */ + public PooledThread getBusyThread(String threadName) { + synchronized (freeList) { + Iterator it = getBusyThreads(); + while (it.hasNext()) { + PooledThread pt = (PooledThread) it.next(); + if (pt.getName().equals(threadName)) { + return pt; } + } + return null; } + } - /** - * Return an Iterator of PooledThread that are currently running. You should - * only use this for display. Use the getPooledThread() or interrupt() - * methods to interrupt a particular thread. - * - * @return an Iterator of busy PooledThread's. - */ - public Iterator getBusyThreads() { - synchronized (freeList) { - return busyList.iterator(); - } - } + /** + * Grow the pool with the option of either putting it on the available list, + * or returning it. + */ + private PooledThread growPool(boolean andReturn) { - /** - * Shutdown the threadpool stopping all the threads. This will - * wait until any busy threads have finished their assigned work. - */ - protected void shutdown() { + synchronized (freeList) { - synchronized (freeList) { - isStopping = true; - - int size = size(); - - if (size > 0){ - String msg = null; - msg = "ThreadPool [" + poolName + "] Shutting down; threadCount[" + size() - + "] busyCount[" + getBusyCount() + "]"; - - logger.info(msg); - } + String threadName = poolName + "." + uniqueThreadID++; + PooledThread bgw = new PooledThread(this, threadName, isDaemon, threadPriority); + bgw.start(); - stopThePool = true; - - while (!freeList.isEmpty()) { - PooledThread thread = (PooledThread) freeList.remove(0); - thread.stop(); - } - - try { - while (getBusyCount() > 0) { - // synchronized (freeList) { - String msg = "ThreadPool [" + poolName + "] has [" + getBusyCount() - + "] busy threads, waiting for those to finish."; - logger.info(msg); - - Iterator busyThreads = getBusyThreads(); - while (busyThreads.hasNext()) { - PooledThread busyThread = (PooledThread) busyThreads.next(); - - String threadName = busyThread.getName(); - Work work = busyThread.getWork(); - - String busymsg = "Busy thread [" + threadName + "] work[" + work + "]"; - logger.info(busymsg); - } - // trace("wait for a busy thread to be put back into - // freeList"); - freeList.wait(); - PooledThread thread = (PooledThread) freeList.remove(0); - // trace("wait finished...now shut it down [" + - // thread.getName() + "]"); - if (thread != null) { - thread.stop(); - } - } - - - } catch (InterruptedException e) { - logger.error(null, e); - } - } - } - - /** - * Trim or grow the pool leaving at least min free. - */ - protected void maintainPoolSize() { - synchronized (freeList) { - if (isStopping) { - // don't bother as the pool is shutting down - return; - } - - int numToStop = size() - minSize; - if (numToStop > 0) { - // should trim idle threads as we are over the minSize - long usedAfter = System.currentTimeMillis() - maxIdleTime; - ArrayList stopList = new ArrayList(); - Iterator it = freeList.iterator(); - while (it.hasNext() && numToStop > 0) { - PooledThread thread = (PooledThread) it.next(); - if (thread.getLastUsedTime() < usedAfter) { - stopList.add(thread); - numToStop--; - } - } - Iterator stopIt = stopList.iterator(); - while (stopIt.hasNext()) { - PooledThread thread = (PooledThread) stopIt.next(); - thread.stop(); - } - } - int numToAdd = minSize - size(); - if (numToAdd > 0) { - // should add some more to the pool - for (int i = 0; i < numToAdd; i++) { - growPool(false); - } - } - } - } - - /** - * Interrupt a named thread that is currently busy. - *

- * Returns the thread that was interrupted or null if the thread - * was not found. If the thread was interrupted then it will - * automatically be stopped and removed from the pool. - *

- *

- * Note that it may take some time to actually interrupt the thread so - * an immediate test to see if the thread stopped will probably be wrong. - * 


-     * ThreadPool test = ThreadPoolManager.getThreadPool("test");
-     * PooledThread pt = test.interrupt("test.1");
-     * if (pt == null) {
-     *      // the thread was not found, perhaps finished?
-     * } else {
-     *      // give interrupt a little time to execute
-     *      Thread.sleep(1000);
-     *      boolean hasStopped = pt.isStopped();
-     *      //..
-     * }
-     *
- *

- * @return the thread that was interrupted - */ - public PooledThread interrupt(String threadName) { - PooledThread thread = getBusyThread(threadName); - if (thread != null) { - thread.interrupt(); - return thread; - } + if (logger.isDebugEnabled()) { + logger.debug("ThreadPool grow created [" + threadName + "] size[" + size() + "]"); + } + if (andReturn) { + return bgw; + } else { + freeList.add(bgw); return null; + } } + } - /** - * Find a thread using its name from the busy list. Returns null if the - * thread is not found in the busy list. - */ - public PooledThread getBusyThread(String threadName) { - synchronized (freeList) { - Iterator it = getBusyThreads(); - while (it.hasNext()) { - PooledThread pt = (PooledThread) it.next(); - if (pt.getName().equals(threadName)) { - return pt; - } - } - return null; - } - } + /** + * Return the maximum amount of time in millis that Threads can be idle before + * they are trimmed. + */ + public long getMaxIdleTime() { + return maxIdleTime; + } - /** - * Grow the pool with the option of either putting it on the available list, - * or returning it. - */ - private PooledThread growPool(boolean andReturn) { + /** + * Set the maxiumium amount of time in millis that Threads can be idle before + * they are trimed. + */ + public void setMaxIdleTime(long maxIdleTime) { + this.maxIdleTime = maxIdleTime; + } - synchronized (freeList) { - - String threadName = poolName + "." + uniqueThreadID++; - PooledThread bgw = new PooledThread(this, threadName, isDaemon, threadPriority); - bgw.start(); - - if (logger.isDebugEnabled()) { - logger.debug("ThreadPool grow created [" + threadName + "] size[" + size() + "]"); - } - if (andReturn) { - return bgw; - } else { - freeList.add(bgw); - return null; - } - } - } - - /** - * Return the maximum amount of time in millis that Threads can be idle - * before they are trimmed. - */ - public long getMaxIdleTime() { - return maxIdleTime; - } - - /** - * Set the maxiumium amount of time in millis that Threads can be idle - * before they are trimed. - */ - public void setMaxIdleTime(long maxIdleTime) { - this.maxIdleTime = maxIdleTime; - } - } diff --git a/src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPoolManager.java b/src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPoolManager.java deleted file mode 100644 index 55c600153..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPoolManager.java +++ /dev/null @@ -1,177 +0,0 @@ -package com.avaje.ebeaninternal.server.lib.thread; - -import java.util.Iterator; -import java.util.concurrent.ConcurrentHashMap; - -import com.avaje.ebean.config.GlobalProperties; -import com.avaje.ebeaninternal.server.lib.BackgroundThread; - -/** - * Singleton that manages a list of named ThreadPools. - */ -public class ThreadPoolManager implements Runnable { - - private static final class Single { - private static final ThreadPoolManager me = new ThreadPoolManager(); - } - - private static int debugLevel = 0; - - /** - * set when the pools are being shutdown. - */ - private boolean isShuttingDown = false; - - /** - * Holds all the thread pools. - */ - private ConcurrentHashMap threadPoolCache = new ConcurrentHashMap(); - - /** - * The default time threads are idle before they are stopped and removed. - * This can occur when the pool grows larger than the min size and then goes idle - * for some time. - */ - private long defaultIdleTime; - - private ThreadPoolManager() { - initialise(); - } - - private void initialise() { - - debugLevel = GlobalProperties.getInt("threadpool.debugLevel", 0); - - defaultIdleTime = 1000 * GlobalProperties.getInt("threadpool.idletime", 60); - - int freqIsSecs = GlobalProperties.getInt("threadpool.sleeptime", 30); - - BackgroundThread.add(freqIsSecs, this); - } - - /** - * Set the debug level. - */ - public static void setDebugLevel(int level) { - debugLevel = level; - } - - /** - * Return the debug level. - */ - public static int getDebugLevel() { - return debugLevel; - } - - /** - * Periodically maintains the pool size. Stops threads that have - * been idle for too long and ensures the minimum number of threads. - *

- * To change this you can set the threadpool.idletime property:
- *
- * ## set threadpool idletime to 120 seconds
- * threadpool.idletime=120
- *

- */ - public void run() { - if (!isShuttingDown) { - maintainPoolSize(); - } - } - - /** - * Return the named thread pool. - */ - public static ThreadPool getThreadPool(String poolName) { - return Single.me.getPool(poolName); - } - - /** - * Return the named ThreadPool. If the ThreadPool doesn't exist it will be - * created. - */ - private ThreadPool getPool(String poolName) { - synchronized (this) { - ThreadPool threadPool = (ThreadPool) threadPoolCache.get(poolName); - if (threadPool == null) { - threadPool = createThreadPool(poolName); - threadPoolCache.put(poolName, threadPool); - } - return threadPool; - } - } - - /** - * Returns an iterator of ThreadPools. - *

- * Note that the ThreadPools should not be removed by the iterator. - *

- */ - public static Iterator pools() { - return Single.me.threadPoolCache.values().iterator(); - } - - /** - * Maintain the size of all the thread pools. - * Trims down to minimum size threads that have been idle for a while. - * Adds threads if it is short of the minimum size. - */ - private void maintainPoolSize() { - if (isShuttingDown){ - return; - } - synchronized (this) { - - Iterator e = pools(); - while (e.hasNext()) { - ThreadPool pool = (ThreadPool) e.next(); - pool.maintainPoolSize(); - } - } - } - - /** - * Shutdown all the ThreadPools nicely. - * This will wait for all currently runnable and queued work to finish. - */ - public static void shutdown() { - Single.me.shutdownPools(); - } - - private void shutdownPools() { - synchronized (this) { - isShuttingDown = true; - Iterator i = pools(); - while (i.hasNext()) { - ThreadPool pool = (ThreadPool) i.next(); - pool.shutdown(); - } - } - } - - private ThreadPool createThreadPool(String poolName) { - - int min = GlobalProperties.getInt("threadpool." + poolName + ".min", 0); - int max = GlobalProperties.getInt("threadpool." + poolName + ".max", 100); - - long idle = 1000 * GlobalProperties.getInt("threadpool." + poolName + ".idletime", -1); - if (idle < 0) { - idle = defaultIdleTime; - } - - boolean isDaemon = true; - Integer priority = null; - String threadPriority = GlobalProperties.get("threadpool." + poolName + ".priority", null); - if (threadPriority != null) { - priority = new Integer(threadPriority); - } - - ThreadPool newPool = new ThreadPool(poolName, isDaemon, priority); - newPool.setMaxSize(max); - newPool.setMinSize(min); - newPool.setMaxIdleTime(idle); - - return newPool; - } - -}; diff --git a/src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java b/src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java index bf638fadd..23e7ef0bf 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java +++ b/src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java @@ -3,6 +3,7 @@ package com.avaje.ebeaninternal.server.transaction; import java.sql.Connection; import java.sql.SQLException; import java.util.List; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import javax.persistence.PersistenceException; @@ -23,6 +24,7 @@ import com.avaje.ebeaninternal.api.TransactionEventTable.TableIUD; import com.avaje.ebeaninternal.server.cluster.ClusterManager; import com.avaje.ebeaninternal.server.core.BootupClasses; import com.avaje.ebeaninternal.server.deploy.BeanDescriptorManager; +import com.avaje.ebeaninternal.server.lib.sql.DataSourcePool; /** * Manages transactions. @@ -136,10 +138,21 @@ public class TransactionManager { String value = GlobalProperties.get("transaction.onqueryonly", "ROLLBACK").toUpperCase().trim(); this.onQueryOnly = getOnQueryOnly(value, dataSource); + + initialiseHeartbeat(); } - public void shutdown() { - // Nothing to do + private void initialiseHeartbeat() { + if (dataSource instanceof DataSourcePool) { + DataSourcePool ds = (DataSourcePool)dataSource; + backgroundExecutor.executePeriodically(ds.getHeartbeatRunnable(), ds.getHeartbeatFreqSecs(), TimeUnit.SECONDS); + } + } + + public void shutdown(boolean shutdownDataSource, boolean deregisterDriver) { + if (shutdownDataSource && (dataSource instanceof DataSourcePool)) { + ((DataSourcePool)dataSource).shutdown(deregisterDriver); + } } public BeanDescriptorManager getBeanDescriptorManager() { diff --git a/src/test/java/com/avaje/tests/text/json/TestJsonBeanWithTimeZone.java b/src/test/java/com/avaje/tests/text/json/TestJsonBeanWithTimeZone.java index a1b0894dd..02c32302b 100644 --- a/src/test/java/com/avaje/tests/text/json/TestJsonBeanWithTimeZone.java +++ b/src/test/java/com/avaje/tests/text/json/TestJsonBeanWithTimeZone.java @@ -2,14 +2,20 @@ package com.avaje.tests.text.json; import java.util.TimeZone; +import junit.framework.TestCase; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import com.avaje.ebean.Ebean; +import com.avaje.ebean.EbeanServer; import com.avaje.ebean.text.json.JsonContext; import com.avaje.tests.model.basic.BeanWithTimeZone; -import junit.framework.TestCase; - public class TestJsonBeanWithTimeZone extends TestCase { + private static final Logger logger = LoggerFactory.getLogger(TestJsonBeanWithTimeZone.class); + public void testSimple() { TimeZone defaultTimeZone = TimeZone.getDefault(); @@ -37,6 +43,10 @@ public class TestJsonBeanWithTimeZone extends TestCase { assertEquals(bean.getTimezone(), bean3.getTimezone()); + // EbeanServer server = Ebean.getServer(null); + //server.shutdown(); + //logger.info("shudown server manually, JVM shutdown hook fires next"); + } } diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml index 68108fe96..002bdf8e4 100644 --- a/src/test/resources/logback-test.xml +++ b/src/test/resources/logback-test.xml @@ -33,6 +33,8 @@ + +