github/ebean
4
0
Fork 0
mirror of https://github.com/ebean-orm/ebean.git synced 2024-04-21 10:51:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix for Issue 4 - Allow programmatic shutdown of EbeanServer

Browse Source
This commit is contained in:
Robin Bygrave
2013-04-26 23:21:25 +12:00
parent 384931d92e
commit bcce1a7148
31 changed files with 1397 additions and 2073 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/avaje/ebean/EbeanServer.java
+17
View File
@@ -83,6 +83,23 @@ import com.avaje.ebean.text.json.JsonContext;
*/
public interface EbeanServer {
/**
* Shutdown the EbeanServer.
* <p>
* 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.
* </p>
*
* @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.
src/main/java/com/avaje/ebean/config/DataSourceConfig.java
+23 -1
View File
@@ -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.
* <p>
* This is the expected frequency in which the DataSource should be checked to
* make sure it is healthy and trim idle connections.
* </p>
*/
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.
src/main/java/com/avaje/ebeaninternal/api/SpiEbeanServer.java
+141 -137
View File
@@ -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.
* <p>
* Required for Oracle specific batch handling.
* </p>
*/
public PstmtBatch getPstmtBatch();
/**
* Create an object to represent the current CallStack.
* <p>
* Typically used to identify the origin of queries for Autofetch
* and object graph costing.
* </p>
*/
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.
* <p>
* Required for Oracle specific batch handling.
* </p>
*/
public PstmtBatch getPstmtBatch();
/**
* Return all the descriptors.
*/
public List<BeanDescriptor<?>> getBeanDescriptors();
/**
* Create an object to represent the current CallStack.
* <p>
* Typically used to identify the origin of queries for Autofetch and object
* graph costing.
* </p>
*/
public CallStack createCallStack();
/**
* Return the BeanDescriptor for a given type of bean.
*/
public <T> BeanDescriptor<T> getBeanDescriptor(Class<T> 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<BeanDescriptor<?>> getBeanDescriptors(String tableName);
/**
* Clear the query execution statistics.
*/
public void clearQueryStatistics();
/**
* Process committed changes from another framework.
* <p>
* 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.
* </p>
*/
public void externalModification(TransactionEventTable event);
/**
* Return all the descriptors.
*/
public List<BeanDescriptor<?>> getBeanDescriptors();
/**
* Create a ServerTransaction.
* <p>
* To specify to use the default transaction isolation use a value of -1.
* </p>
*/
public SpiTransaction createServerTransaction(boolean isExplicit, int isolationLevel);
/**
* Return the BeanDescriptor for a given type of bean.
*/
public <T> BeanDescriptor<T> getBeanDescriptor(Class<T> 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<BeanDescriptor<?>> getBeanDescriptors(String tableName);
/**
* Create a ServerTransaction for query purposes.
*/
public SpiTransaction createQueryTransaction();
/**
* Process committed changes from another framework.
* <p>
* 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.
* </p>
*/
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.
* <p>
* To specify to use the default transaction isolation use a value of -1.
* </p>
*/
public SpiTransaction createServerTransaction(boolean isExplicit, int isolationLevel);
/**
* Create a query request object.
*/
public <T> SpiOrmQueryRequest<T> createQueryRequest(BeanDescriptor<T> desc, SpiQuery<T> q, Transaction t);
/**
* Return the current transaction or null if there is no current transaction.
*/
public SpiTransaction getCurrentServerTransaction();
/**
* Compile a query.
*/
public <T> CQuery<T> compileQuery(Query<T> 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.
* <p>
* 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).
* </p>
*/
public <T> List<Object> findIdsWithCopy(Query<T> 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 <T> int findRowCountWithCopy(Query<T> query, Transaction t);
/**
* Create a query request object.
*/
public <T> SpiOrmQueryRequest<T> createQueryRequest(BeanDescriptor<T> desc, SpiQuery<T> q,
Transaction t);
/**
* Load a batch of Associated One Beans.
*/
public void loadBean(LoadBeanRequest loadRequest);
/**
* Compile a query.
*/
public <T> CQuery<T> compileQuery(Query<T> 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.
* <p>
* 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).
* </p>
*/
public <T> List<Object> findIdsWithCopy(Query<T> query, Transaction t);
/**
* Execute the findRowCount query but without copying the query.
*/
public <T> int findRowCountWithCopy(Query<T> 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);
}
src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManager.java
+4 -2
View File
@@ -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);
}
}
src/main/java/com/avaje/ebeaninternal/server/autofetch/DefaultAutoFetchManagerLogging.java
+11 -9
View File
@@ -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 {
src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java
+6 -4
View File
@@ -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();
}
/**
src/main/java/com/avaje/ebeaninternal/server/core/DefaultBackgroundExecutor.java
+2 -2
View File
@@ -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-");
}
src/main/java/com/avaje/ebeaninternal/server/core/DefaultServer.java
+56 -20
View File
@@ -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<SpiEbeanPlugin> spiPlugins = new ArrayList<SpiEbeanPlugin>();
final Iterator<SpiEbeanPlugin> 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.
*/
src/main/java/com/avaje/ebeaninternal/server/core/DefaultServerFactory.java
+13 -18
View File
@@ -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);
}
/**
src/main/java/com/avaje/ebeaninternal/server/core/TraditionalBackgroundExecutor.java
+12 -1
View File
@@ -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();
}
src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundRunnable.java
-147
View File
@@ -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.
* <p>
* This is used to stop jobs firing immediately.
* </p>
*/
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();
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/BackgroundThread.java
-220
View File
@@ -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.
* <p>
* 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.
* </p>
* <p>
* SystemProperties:<br>
*
* <pre><code>
* ## initially sleep for 5 seconds before starting
* backgroundthread.initialsleep=5
* </code></pre>
*
* </p>
*
* @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<BackgroundRunnable> list = new Vector<BackgroundRunnable>();
/**
* 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<BackgroundRunnable> 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<BackgroundRunnable> 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<BackgroundRunnable> it = runnables();
while (it.hasNext()) {
BackgroundRunnable bgr = it.next();
sb.append(bgr);
}
return sb.toString();
}
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/DaemonScheduleThreadPool.java
+52 -51
View File
@@ -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.
* <p>
* This will wait a maximum of 20 seconds before terminating any threads
* still working.
* </p>
*/
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.
* <p>
* This will wait a maximum of 20 seconds before terminating any threads still
* working.
* </p>
*/
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();
}
};
}
src/main/java/com/avaje/ebeaninternal/server/lib/DaemonThreadPool.java
+64 -64
View File
@@ -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<Runnable>(), 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.
* <p>
* This will wait a maximum of 20 seconds before terminating any threads
* still working.
* </p>
*/
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<Runnable>(), 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.
* <p>
* This will wait a maximum of 20 seconds before terminating any threads still
* working.
* </p>
*/
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();
}
};
}
src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownHook.java
-11
View File
@@ -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.
*/
src/main/java/com/avaje/ebeaninternal/server/lib/ShutdownManager.java
+111 -126
View File
@@ -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.
* <p>
* Makes sure all the resources are shutdown properly and in order.
* </p>
@@ -24,19 +25,19 @@ public final class ShutdownManager {
private static final Logger logger = LoggerFactory.getLogger(ShutdownManager.class);
static final Vector<Runnable> runnables = new Vector<Runnable>();
static final List<SpiEbeanServer> servers = new ArrayList<SpiEbeanServer>();
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.
* </p>
*/
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.
* <p>
* This is typically invoked via JVM shutdown hook.
* </p>
*/
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<Driver> 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.
* <p>
* <ul>
* <li>Run the application specific shutdown runnable
* <li>Run any other registered shutdown runnable
* <li>Deregister from the cluster if required
* <li>Shutdown Thread pools
* <li>Shutdown any database connection pools
* </ul>
* This is done when the ebeanServer is shutdown manually.
* </p>
*/
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<Runnable> 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<Driver> 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);
}
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceNotify.java → src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlert.java
+4 -4
View File
@@ -8,20 +8,20 @@ package com.avaje.ebeaninternal.server.lib.sql;
* when these events occur on the DataSource.
* </p>
*/
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);
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceAlertListener.java
-26
View File
@@ -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);
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceGlobalManager.java
-50
View File
@@ -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<DataSourcePool> 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);
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourceManager.java
-233
View File
@@ -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<String,DataSourcePool> dsMap = new Hashtable<String, DataSourcePool>();
/**
* 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<DataSourcePool> 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<DataSourcePool> getPools() {
synchronized(monitor) {
// create a copy of the DataSourcePool's
ArrayList<DataSourcePool> list = new ArrayList<DataSourcePool>();
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<DataSourcePool> 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();
}
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/DataSourcePool.java
+57 -23
View File
@@ -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.
* <p>
* This is the frequency that the heartbeat runnable should be run.
* </p>
*/
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.
* <p>
* This is called by the HeartbeatRunnable which should be scheduled to
* run periodically (every heartbeatFreqSecs seconds actually).
* </p>
*/
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.
* <p>
* This method is protected by synchronisation in calling methods.
* This method is protected by synchronization in calling methods.
* </p>
*/
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.
* </p>
*/
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);
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnection.java
+4 -2
View File
@@ -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) {
src/main/java/com/avaje/ebeaninternal/server/lib/sql/PooledConnectionQueue.java
+4 -4
View File
@@ -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 {
src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleAlerter.java
-107
View File
@@ -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.
* <ul>
* <li>alert.fromuser = the from user name
* <li>alert.fromemail = the from email account
* <li>alert.toemail = comma delimited list of email accounts to email
* <li>alert.mailserver = the smpt server name
* </ul>
*/
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);
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/sql/SimpleDataSourceAlert.java
+109
View File
@@ -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.
* <ul>
* <li>alert.fromuser = the from user name
* <li>alert.fromemail = the from email account
* <li>alert.toemail = comma delimited list of email accounts to email
* <li>alert.mailserver = the smpt server name
* </ul>
*/
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);
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/thread/PooledThread.java
+245 -226
View File
@@ -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.
* <p>
* If the Thread was interrupted then it will be removed from the pool.
* </p>
*/
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.
* <p>
* If the Thread was interrupted then it will be removed from the pool.
* </p>
*/
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;
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPool.java
+433 -404
View File
@@ -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<PooledThread> freeList = new Vector<PooledThread>();
/**
* 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<PooledThread> busyList = new Vector<PooledThread>();
/**
* Flag that the pool should terminate all the threads and stop.
*/
private boolean stopThePool;
/**
* List holding queued work.
*/
private Vector<Work> workOverflowQueue = new Vector<Work>();
/**
* 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<PooledThread> freeList = new Vector<PooledThread>();
/**
* Flag that the pool should terminate all the threads and stop.
*/
private boolean stopThePool;
/**
* List of PooledThread that are busy.
*/
private Vector<PooledThread> busyList = new Vector<PooledThread>();
/**
* Create the ThreadPool.
*/
public ThreadPool(String poolName, boolean isDaemon, Integer threadPriority) {
/**
* List holding queued work.
*/
private Vector<Work> workOverflowQueue = new Vector<Work>();
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.
* <p>
* When the pool is fully busy...
* </p>
* <p>
* addToQueue=true -> work is added to queue, returns false<br>
* addToQueue=false -> work is not done or queued, returns false<br>
* </p>
*
* @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<PooledThread> 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.
* <p>
* When the pool is fully busy...
* </p>
* <p>
* addToQueue=true -> work is added to queue, returns false<br>
* addToQueue=false -> work is not done or queued, returns false<br>
* </p>
*
* @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<PooledThread> 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<PooledThread> stopList = new ArrayList<PooledThread>();
Iterator<PooledThread> it = freeList.iterator();
while (it.hasNext() && numToStop > 0) {
PooledThread thread = (PooledThread) it.next();
if (thread.getLastUsedTime() < usedAfter) {
stopList.add(thread);
numToStop--;
}
}
Iterator<PooledThread> 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.
* <p>
* 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.
* </p>
* <p>
* 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.
*
* <pre>
* <code>
* 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();
* //..
* }
* </code>
* </pre>
*
* </p>
*
* @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<PooledThread> 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<PooledThread> 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<PooledThread> 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<PooledThread> stopList = new ArrayList<PooledThread>();
Iterator<PooledThread> it = freeList.iterator();
while (it.hasNext() && numToStop > 0) {
PooledThread thread = (PooledThread) it.next();
if (thread.getLastUsedTime() < usedAfter) {
stopList.add(thread);
numToStop--;
}
}
Iterator<PooledThread> 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.
* <p>
* 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.
* </p>
* <p>
* 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.
* <pre><code>
* 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();
* //..
* }
* </code></pre>
* </p>
* @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<PooledThread> 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;
}
}
src/main/java/com/avaje/ebeaninternal/server/lib/thread/ThreadPoolManager.java
-177
View File
@@ -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<String,ThreadPool> threadPoolCache = new ConcurrentHashMap<String, ThreadPool>();
/**
* 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.
* <p>
* To change this you can set the threadpool.idletime property:<br>
* <br>
* <b><code>## set threadpool idletime to 120 seconds</code></b><br>
* <b><code>threadpool.idletime=120</code></b><br>
* </p>
*/
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.
* <p>
* Note that the ThreadPools should not be removed by the iterator.
* </p>
*/
public static Iterator<ThreadPool> 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<ThreadPool> 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<ThreadPool> 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;
}
};
src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java
+15 -2
View File
@@ -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() {