#775 - Change DefaultBackgroundExecutor to use JDK cached thread pool (rather than configure min/max size etc)

This commit is contained in:
Robin Bygrave
2016-07-15 20:48:41 +12:00
parent 474c2d5595
commit e0eff149b5
9 changed files with 141 additions and 174 deletions
@@ -384,9 +384,6 @@ public class ServerConfig {
// configuration for the background executor service (thread pool)
private int backgroundExecutorSchedulePoolSize = 1;
private int backgroundExecutorCorePoolSize = 1;
private int backgroundExecutorMaxPoolSize = 8;
private int backgroundExecutorIdleSecs = 60;
private int backgroundExecutorShutdownSecs = 30;
// defaults for the L2 bean caching
@@ -1070,48 +1067,6 @@ public class ServerConfig {
this.backgroundExecutorSchedulePoolSize = backgroundExecutorSchedulePoolSize;
}
/**
* Return the Background executor core pool size.
*/
public int getBackgroundExecutorCorePoolSize() {
return backgroundExecutorCorePoolSize;
}
/**
* Set the Background executor core pool size.
*/
public void setBackgroundExecutorCorePoolSize(int backgroundExecutorCorePoolSize) {
this.backgroundExecutorCorePoolSize = backgroundExecutorCorePoolSize;
}
/**
* Return the Background executor max pool size.
*/
public int getBackgroundExecutorMaxPoolSize() {
return backgroundExecutorMaxPoolSize;
}
/**
* Set the Background executor max pool size.
*/
public void setBackgroundExecutorMaxPoolSize(int backgroundExecutorMaxPoolSize) {
this.backgroundExecutorMaxPoolSize = backgroundExecutorMaxPoolSize;
}
/**
* Return the Background executor idle seconds.
*/
public int getBackgroundExecutorIdleSecs() {
return backgroundExecutorIdleSecs;
}
/**
* Set the Background executor idle seconds.
*/
public void setBackgroundExecutorIdleSecs(int backgroundExecutorIdleSecs) {
this.backgroundExecutorIdleSecs = backgroundExecutorIdleSecs;
}
/**
* Return the Background executor shutdown seconds. This is the time allowed for the pool to shutdown nicely
* before it is forced shutdown.
@@ -2399,6 +2354,8 @@ public class ServerConfig {
autoCommitMode = p.getBoolean("autoCommitMode", autoCommitMode);
useJtaTransactionManager = p.getBoolean("useJtaTransactionManager", useJtaTransactionManager);
backgroundExecutorSchedulePoolSize = p.getInt("backgroundExecutorSchedulePoolSize", backgroundExecutorSchedulePoolSize);
backgroundExecutorShutdownSecs = p.getInt("backgroundExecutorShutdownSecs", backgroundExecutorShutdownSecs);
disableClasspathSearch = p.getBoolean("disableClasspathSearch", disableClasspathSearch);
currentUserProvider = createInstance(p, CurrentUserProvider.class, "currentUserProvider", currentUserProvider);
databasePlatform = createInstance(p, DatabasePlatform.class, "databasePlatform", databasePlatform);
@@ -1,35 +1,25 @@
package com.avaje.ebeaninternal.server.core;
import java.util.concurrent.TimeUnit;
import com.avaje.ebeaninternal.api.SpiBackgroundExecutor;
import com.avaje.ebeaninternal.server.lib.DaemonExecutorService;
import com.avaje.ebeaninternal.server.lib.DaemonScheduleThreadPool;
import com.avaje.ebeaninternal.server.lib.DaemonThreadPool;
import java.util.concurrent.TimeUnit;
/**
* The default implementation of the BackgroundExecutor.
*/
public class DefaultBackgroundExecutor implements SpiBackgroundExecutor {
private final DaemonThreadPool pool;
private final DaemonScheduleThreadPool schedulePool;
private final DaemonExecutorService pool;
/**
* Construct the default implementation of BackgroundExecutor.
*
* @param corePoolSize
* the core size of the thread pool.
* @param maximumPoolSize
* the maximum pool size before jobs are queued
* @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 DefaultBackgroundExecutor(int schedulePoolSize, int corePoolSize, int maximumPoolSize, long keepAliveSecs,int shutdownWaitSeconds, String namePrefix) {
this.pool = new DaemonThreadPool(corePoolSize, maximumPoolSize, keepAliveSecs, shutdownWaitSeconds, namePrefix);
public DefaultBackgroundExecutor(int schedulePoolSize, int shutdownWaitSeconds, String namePrefix) {
this.pool = new DaemonExecutorService(shutdownWaitSeconds, namePrefix);
this.schedulePool = new DaemonScheduleThreadPool(schedulePoolSize, shutdownWaitSeconds, namePrefix+"-periodic-");
}
@@ -78,14 +78,10 @@ public class DefaultContainer implements SpiContainer {
private SpiBackgroundExecutor createBackgroundExecutor(ServerConfig serverConfig) {
String namePrefix = "ebean-" + serverConfig.getName();
int schedulePoolSize = serverConfig.getBackgroundExecutorSchedulePoolSize();
int corePoolSize = serverConfig.getBackgroundExecutorCorePoolSize();
int maxPoolSize = serverConfig.getBackgroundExecutorMaxPoolSize();
int idleSecs = serverConfig.getBackgroundExecutorIdleSecs();
int shutdownSecs = serverConfig.getBackgroundExecutorShutdownSecs();
return new DefaultBackgroundExecutor(schedulePoolSize, corePoolSize, maxPoolSize, idleSecs, shutdownSecs, namePrefix);
return new DefaultBackgroundExecutor(schedulePoolSize, shutdownSecs, namePrefix);
}
/**
@@ -0,0 +1,72 @@
package com.avaje.ebeaninternal.server.lib;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* A "CachedThreadPool" based on Daemon threads.
* <p>
* The Threads are created as needed and once idle live for 60 seconds.
*/
public final class DaemonExecutorService {
private static final Logger logger = LoggerFactory.getLogger(DaemonExecutorService.class);
private final String namePrefix;
private final int shutdownWaitSeconds;
private final ExecutorService service;
/**
* Construct the DaemonThreadPool.
*
* @param shutdownWaitSeconds the time in seconds allowed for the pool to shutdown nicely. After
* this the pool is forced to shutdown.
*/
public DaemonExecutorService(int shutdownWaitSeconds, String namePrefix) {
this.service = Executors.newCachedThreadPool(new DaemonThreadFactory(namePrefix));
this.shutdownWaitSeconds = shutdownWaitSeconds;
this.namePrefix = namePrefix;
}
/**
* Execute the Runnable.
*/
public void execute(Runnable runnable) {
service.execute(runnable);
}
/**
* 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 (service.isShutdown()) {
logger.debug("DaemonExecutorService[{}] already shut down", namePrefix);
return;
}
try {
logger.debug("DaemonExecutorService[{}] shutting down...", namePrefix);
service.shutdown();
if (!service.awaitTermination(shutdownWaitSeconds, TimeUnit.SECONDS)) {
logger.info("DaemonExecutorService[{}] shut down timeout exceeded. Terminating running threads.", namePrefix);
service.shutdownNow();
}
} catch (Exception e) {
logger.error("Error during shutdown of DaemonThreadPool[" + namePrefix + "]", e);
e.printStackTrace();
}
}
}
}
@@ -11,8 +11,6 @@ import org.slf4j.LoggerFactory;
* <p>
* Uses Daemon threads and hooks into shutdown event.
* </p>
*
* @author rbygrave
*/
public final class DaemonScheduleThreadPool extends ScheduledThreadPoolExecutor {
@@ -26,19 +24,11 @@ public final class DaemonScheduleThreadPool extends ScheduledThreadPoolExecutor
* Construct the DaemonScheduleThreadPool.
*/
public DaemonScheduleThreadPool(int coreSize, int shutdownWaitSeconds, String namePrefix) {
super(coreSize, new DaemonThreadFactory(namePrefix));
this.namePrefix = namePrefix;
this.shutdownWaitSeconds = shutdownWaitSeconds;
}
/**
* Register a shutdown hook with the JVM Runtime.
*/
public void registerShutdownHook() {
Runtime.getRuntime().addShutdownHook(new ShutdownHook());
}
/**
* Shutdown this thread pool nicely if possible.
* <p>
@@ -66,14 +56,4 @@ public final class DaemonScheduleThreadPool extends ScheduledThreadPoolExecutor
}
}
}
/**
* Fired by the JVM Runtime shutdown.
*/
private class ShutdownHook extends Thread {
@Override
public void run() {
shutdown();
}
}
}
@@ -1,86 +0,0 @@
package com.avaje.ebeaninternal.server.lib;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* The Thread Pool based on Daemon threads.
*
* @author rbygrave
*/
public final class DaemonThreadPool extends ThreadPoolExecutor {
private static final Logger logger = LoggerFactory.getLogger(DaemonThreadPool.class);
private final String namePrefix;
private final 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, int maximumPoolSize, long keepAliveSecs, int shutdownWaitSeconds, String namePrefix) {
super(coreSize, maximumPoolSize, 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();
}
} 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();
}
}
}
@@ -32,6 +32,8 @@ public class ServerConfigTest {
props.setProperty("dbuuid","binary");
props.setProperty("jdbcFetchSizeFindEach", "42");
props.setProperty("jdbcFetchSizeFindList", "43");
props.setProperty("backgroundExecutorShutdownSecs", "98");
props.setProperty("backgroundExecutorSchedulePoolSize", "4");
serverConfig.loadFromProperties(props);
@@ -40,6 +42,8 @@ public class ServerConfigTest {
assertEquals(ServerConfig.DbUuid.BINARY, serverConfig.getDbUuid());
assertEquals(42, serverConfig.getJdbcFetchSizeFindEach());
assertEquals(43, serverConfig.getJdbcFetchSizeFindList());
assertEquals(4, serverConfig.getBackgroundExecutorSchedulePoolSize());
assertEquals(98, serverConfig.getBackgroundExecutorShutdownSecs());
serverConfig.setPersistBatch(PersistBatch.NONE);
serverConfig.setPersistBatchOnCascade(PersistBatch.NONE);
@@ -0,0 +1,54 @@
package com.avaje.ebeaninternal.server.core;
import org.junit.Ignore;
import org.junit.Test;
public class DefaultBackgroundExecutorTest {
@Test @Ignore
public void shutdown_when_running_expect_waitAndNiceShutdown() throws Exception {
DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 20, "test");
es.execute(new RunFor(3000,"a"));
es.execute(new RunFor(3000,"b"));
es.execute(new RunFor(3000,"c"));
es.shutdown();
}
@Test @Ignore
public void shutdown_when_rougeRunnable_expect_InterruptedException() throws Exception {
DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test");
es.execute(new RunFor(300000,"a"));
es.execute(new RunFor(3000,"b"));
es.execute(new RunFor(3000,"c"));
es.shutdown();
}
class RunFor implements Runnable {
final long wait;
final String id;
RunFor(long wait, String id) {
this.wait = wait;
this.id = id;
}
@Override
public void run() {
try {
System.out.println("start " + id);
Thread.sleep(wait);
System.out.println("done " + id);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
@@ -22,7 +22,7 @@ public class TestDataSourceMaxWithEntity extends BaseTestCase {
EbeanServer server = Ebean.getServer(null);
DefaultBackgroundExecutor bg = new DefaultBackgroundExecutor(1, 1, 2, 180, 30, "testDs");
DefaultBackgroundExecutor bg = new DefaultBackgroundExecutor(1, 30, "testDs");
try {
for (int i = 0; i < 12; i++) {