diff --git a/ebean-core/src/main/java/io/ebeaninternal/server/executor/DaemonExecutorService.java b/ebean-core/src/main/java/io/ebeaninternal/server/executor/DaemonExecutorService.java new file mode 100644 index 000000000..af23a086d --- /dev/null +++ b/ebean-core/src/main/java/io/ebeaninternal/server/executor/DaemonExecutorService.java @@ -0,0 +1,67 @@ +package io.ebeaninternal.server.executor; + + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.*; +import java.util.concurrent.locks.ReentrantLock; + +/** + * A "CachedThreadPool" based on Daemon threads. + *

+ * The Threads are created as needed and once idle live for 60 seconds. + */ +final class DaemonExecutorService { + + private static final Logger logger = LoggerFactory.getLogger(DaemonExecutorService.class); + + private final ReentrantLock lock = new ReentrantLock(false); + private final String namePrefix; + private final int shutdownWaitSeconds; + private final ExecutorService service; + + DaemonExecutorService(int shutdownWaitSeconds, String namePrefix) { + this.service = Executors.newCachedThreadPool(new DaemonThreadFactory(namePrefix)); + this.shutdownWaitSeconds = shutdownWaitSeconds; + this.namePrefix = namePrefix; + } + + Future submit(Callable task) { + return service.submit(task); + } + + Future submit(Runnable task) { + return service.submit(task); + } + + /** + * Shutdown this thread pool nicely if possible. + *

+ * This will wait a maximum of 20 seconds before terminating any threads still working. + */ + void shutdown() { + lock.lock(); + try { + 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(); + } + } finally { + lock.unlock(); + } + } + +} diff --git a/ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java b/ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java index 7090a63e4..d04348ba8 100644 --- a/ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java +++ b/ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java @@ -5,11 +5,7 @@ import io.ebeaninternal.api.SpiBackgroundExecutor; import org.slf4j.MDC; import java.util.Map; -import java.util.concurrent.Callable; -import java.util.concurrent.Future; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; /** * The default implementation of the BackgroundExecutor. @@ -17,13 +13,15 @@ import java.util.concurrent.TimeUnit; @NonNullApi public final class DefaultBackgroundExecutor implements SpiBackgroundExecutor { - private final ScheduledExecutorService executor; + private final ScheduledExecutorService schedulePool; + private final DaemonExecutorService pool; /** * Construct the default implementation of BackgroundExecutor. */ public DefaultBackgroundExecutor(int schedulePoolSize, int shutdownWaitSeconds, String namePrefix) { - this.executor = new DaemonScheduleThreadPool(schedulePoolSize, shutdownWaitSeconds, namePrefix); + this.schedulePool = new DaemonScheduleThreadPool(schedulePoolSize, shutdownWaitSeconds, namePrefix + "-periodic-"); + this.pool = new DaemonExecutorService(shutdownWaitSeconds, namePrefix); } /** @@ -66,7 +64,7 @@ public final class DefaultBackgroundExecutor implements SpiBackgroundExecutor { @Override public Future submit(Callable task) { - return executor.submit(wrapMDC(task)); + return pool.submit(wrapMDC(task)); } /** @@ -74,7 +72,7 @@ public final class DefaultBackgroundExecutor implements SpiBackgroundExecutor { */ @Override public Future submit(Runnable task) { - return executor.submit(wrapMDC(task)); + return pool.submit(wrapMDC(task)); } @Override @@ -84,37 +82,38 @@ public final class DefaultBackgroundExecutor implements SpiBackgroundExecutor { @Override public void executePeriodically(Runnable task, long delay, TimeUnit unit) { - executor.scheduleWithFixedDelay(wrapMDC(task), delay, delay, unit); + schedulePool.scheduleWithFixedDelay(wrapMDC(task), delay, delay, unit); } @Override public void executePeriodically(Runnable task, long initialDelay, long delay, TimeUnit unit) { - executor.scheduleWithFixedDelay(wrapMDC(task), initialDelay, delay, unit); + schedulePool.scheduleWithFixedDelay(wrapMDC(task), initialDelay, delay, unit); } @Override public ScheduledFuture scheduleWithFixedDelay(Runnable task, long initialDelay, long delay, TimeUnit unit) { - return executor.scheduleWithFixedDelay(wrapMDC(task), initialDelay, delay, unit); + return schedulePool.scheduleWithFixedDelay(wrapMDC(task), initialDelay, delay, unit); } @Override public ScheduledFuture scheduleAtFixedRate(Runnable task, long initialDelay, long delay, TimeUnit unit) { - return executor.scheduleAtFixedRate(wrapMDC(task), initialDelay, delay, unit); + return schedulePool.scheduleAtFixedRate(wrapMDC(task), initialDelay, delay, unit); } @Override public ScheduledFuture schedule(Runnable task, long delay, TimeUnit unit) { - return executor.schedule(wrapMDC(task), delay, unit); + return schedulePool.schedule(wrapMDC(task), delay, unit); } @Override public ScheduledFuture schedule(Callable task, long delay, TimeUnit unit) { - return executor.schedule(wrapMDC(task), delay, unit); + return schedulePool.schedule(wrapMDC(task), delay, unit); } @Override public void shutdown() { - executor.shutdown(); + schedulePool.shutdown(); + pool.shutdown(); } } diff --git a/ebean-core/src/test/java/io/ebeaninternal/server/executor/DaemonExecutorServiceTest.java b/ebean-core/src/test/java/io/ebeaninternal/server/executor/DaemonExecutorServiceTest.java new file mode 100644 index 000000000..72ad74223 --- /dev/null +++ b/ebean-core/src/test/java/io/ebeaninternal/server/executor/DaemonExecutorServiceTest.java @@ -0,0 +1,56 @@ +package io.ebeaninternal.server.executor; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Future; + +import static org.assertj.core.api.Assertions.assertThat; + +class DaemonExecutorServiceTest { + + private final int count = 10; + private final int waitMillis = 100; + + @Test + void submit() throws Exception { + DaemonExecutorService des = new DaemonExecutorService(5, "junk"); + long start = System.currentTimeMillis(); + List> futures = new ArrayList<>(); + for (int i = 0; i < count; i++) { + futures.add(des.submit(this::doStuff)); + } + for (Future f: futures) { + f.get(); + } + long exeMillis = System.currentTimeMillis() - start; + assertThat(exeMillis).isLessThan(count * waitMillis); + des.shutdown(); + } + + @Test + void submit_via_DefaultBackgroundExecutor() throws Exception { + DefaultBackgroundExecutor des = new DefaultBackgroundExecutor(1, 5, "junk"); + long start = System.currentTimeMillis(); + List> futures = new ArrayList<>(); + for (int i = 0; i < count; i++) { + futures.add(des.submit(this::doStuff)); + } + for (Future f: futures) { + f.get(); + } + long exeMillis = System.currentTimeMillis() - start; + assertThat(exeMillis).isLessThan(count * waitMillis); + des.shutdown(); + } + + private void doStuff() { + try { + Thread.sleep(waitMillis); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + +}