Merge pull request #2583 from ebean-orm/feature/2582-bg-executor

#2582 - Fix for regression, BackgroundExecutor is bounded (since 12.6.2)

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.
+ * <p>
+ * 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;
+  }
+
+  <T> Future<T> submit(Callable<T> task) {
+    return service.submit(task);
+  }
+
+  Future<?> submit(Runnable task) {
+    return service.submit(task);
+  }
+
+  /**
+   * Shutdown this thread pool nicely if possible.
+   * <p>
+   * 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();
+    }
+  }
+
+}

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 <T> Future<T> submit(Callable<T> 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 <V> ScheduledFuture<V> schedule(Callable<V> 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();
   }
 
 }

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<Future<?>> 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<Future<?>> 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);
+    }
+  }
+
+}