Prevent shutdown of shared thread pool (#760) (#2531)

Fixes #760

Signed-off-by: Jan Vybíral <jan.vybiral1@gmail.com>

bundles/org.openhab.core/src/main/java/org/openhab/core/common/ThreadPoolManager.java

@@ -12,17 +12,25 @@
  */
 package org.openhab.core.common;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
 import java.util.WeakHashMap;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.openhab.core.internal.common.WrappedScheduledExecutorService;
 import org.osgi.framework.Constants;
@@ -133,7 +141,7 @@ public class ThreadPoolManager {
             }
         }
         if (pool instanceof ScheduledExecutorService) {
-            return (ScheduledExecutorService) pool;
+            return new UnstoppableScheduledExecutorService(poolName, (ScheduledExecutorService) pool);
         } else {
             throw new IllegalArgumentException("Pool " + poolName + " is not a scheduled pool!");
         }
@@ -162,7 +170,17 @@ public class ThreadPoolManager {
                 }
             }
         }
-        return pool;
+        return new UnstoppableExecutorService<>(poolName, pool);
+    }
+
+    static ThreadPoolExecutor getPoolUnwrapped(String poolName) {
+        UnstoppableExecutorService<?> ret = (UnstoppableExecutorService<?>) getPool(poolName);
+        return (ThreadPoolExecutor) ret.getDelegate();
+    }
+
+    static ThreadPoolExecutor getScheduledPoolUnwrapped(String poolName) {
+        UnstoppableExecutorService<?> ret = (UnstoppableScheduledExecutorService) getScheduledPool(poolName);
+        return (ThreadPoolExecutor) ret.getDelegate();
     }
 
     protected static int getConfig(String poolName) {
@@ -173,4 +191,120 @@ public class ThreadPoolManager {
     public static Set<String> getPoolNames() {
         return new HashSet<>(pools.keySet());
     }
+
+    static class UnstoppableExecutorService<T extends ExecutorService> implements ExecutorService {
+
+        protected final Logger logger = LoggerFactory.getLogger(getClass());
+        protected final T delegate;
+        protected final String threadPoolName;
+
+        private UnstoppableExecutorService(String threadPoolName, T delegate) {
+            this.threadPoolName = threadPoolName;
+            this.delegate = delegate;
+        }
+
+        @Override
+        public void shutdown() {
+            logger.warn("shutdown() invoked on a shared thread pool '{}'. This is a bug, please submit a bug report",
+                    threadPoolName, new IllegalStateException());
+        }
+
+        @Override
+        public List<Runnable> shutdownNow() {
+            logger.warn("shutdownNow() invoked on a shared thread pool '{}'. This is a bug, please submit a bug report",
+                    threadPoolName, new IllegalStateException());
+            return Collections.emptyList();
+        }
+
+        @Override
+        public boolean isShutdown() {
+            return delegate.isShutdown();
+        }
+
+        @Override
+        public boolean isTerminated() {
+            return delegate.isTerminated();
+        }
+
+        @Override
+        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+            return delegate.awaitTermination(timeout, unit);
+        }
+
+        @Override
+        public <T> Future<T> submit(Callable<T> task) {
+            return delegate.submit(task);
+        }
+
+        @Override
+        public <T> Future<T> submit(Runnable task, T result) {
+            return delegate.submit(task, result);
+        }
+
+        @Override
+        public Future<?> submit(Runnable task) {
+            return delegate.submit(task);
+        }
+
+        @Override
+        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
+            return delegate.invokeAll(tasks);
+        }
+
+        @Override
+        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+                throws InterruptedException {
+            return delegate.invokeAll(tasks, timeout, unit);
+        }
+
+        @Override
+        public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
+                throws InterruptedException, ExecutionException {
+            return delegate.invokeAny(tasks);
+        }
+
+        @Override
+        public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+                throws InterruptedException, ExecutionException, TimeoutException {
+            return delegate.invokeAny(tasks, timeout, unit);
+        }
+
+        @Override
+        public void execute(Runnable command) {
+            delegate.execute(command);
+        }
+
+        T getDelegate() {
+            return delegate;
+        }
+    }
+
+    static class UnstoppableScheduledExecutorService extends UnstoppableExecutorService<ScheduledExecutorService>
+            implements ScheduledExecutorService {
+
+        private UnstoppableScheduledExecutorService(String threadPoolName, ScheduledExecutorService delegate) {
+            super(threadPoolName, delegate);
+        }
+
+        @Override
+        public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+            return delegate.schedule(command, delay, unit);
+        }
+
+        @Override
+        public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+            return delegate.schedule(callable, delay, unit);
+        }
+
+        @Override
+        public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
+            return delegate.scheduleAtFixedRate(command, initialDelay, period, unit);
+        }
+
+        @Override
+        public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay,
+                TimeUnit unit) {
+            return delegate.scheduleWithFixedDelay(command, initialDelay, delay, unit);
+        }
+    }
 }

bundles/org.openhab.core/src/test/java/org/openhab/core/common/ThreadPoolManagerTest.java

@@ -17,6 +17,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.*;
 
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -34,7 +35,7 @@ public class ThreadPoolManagerTest {
 
     @Test
     public void testGetScheduledPool() {
-        ThreadPoolExecutor result = (ThreadPoolExecutor) ThreadPoolManager.getScheduledPool("test1");
+        ThreadPoolExecutor result = ThreadPoolManager.getScheduledPoolUnwrapped("test1");
 
         assertThat(result, instanceOf(ScheduledExecutorService.class));
 
@@ -45,7 +46,7 @@ public class ThreadPoolManagerTest {
 
     @Test
     public void testGetCachedPool() {
-        ExecutorService result = ThreadPoolManager.getPool("test2");
+        ExecutorService result = ThreadPoolManager.getPoolUnwrapped("test2");
 
         assertThat(result, instanceOf(ExecutorService.class));
 
@@ -60,7 +61,7 @@ public class ThreadPoolManagerTest {
     public void testGetConfiguredScheduledPool() {
         ThreadPoolManager tpm = new ThreadPoolManager();
         tpm.modified(Map.of("test3", "5"));
-        ThreadPoolExecutor result = (ThreadPoolExecutor) ThreadPoolManager.getScheduledPool("test3");
+        ThreadPoolExecutor result = ThreadPoolManager.getScheduledPoolUnwrapped("test3");
 
         assertThat(result, instanceOf(ScheduledExecutorService.class));
         assertEquals(5, result.getCorePoolSize());
@@ -70,14 +71,14 @@ public class ThreadPoolManagerTest {
     public void testGetConfiguredCachedPool() {
         ThreadPoolManager tpm = new ThreadPoolManager();
         tpm.modified(Map.of("test4", "4"));
-        ThreadPoolExecutor result = (ThreadPoolExecutor) ThreadPoolManager.getPool("test4");
+        ThreadPoolExecutor result = ThreadPoolManager.getPoolUnwrapped("test4");
 
         assertEquals(4, result.getMaximumPoolSize());
     }
 
     @Test
     public void testReconfiguringScheduledPool() {
-        ThreadPoolExecutor result = (ThreadPoolExecutor) ThreadPoolManager.getScheduledPool("test5");
+        ThreadPoolExecutor result = ThreadPoolManager.getScheduledPoolUnwrapped("test5");
         assertEquals(ThreadPoolManager.DEFAULT_THREAD_POOL_SIZE, result.getCorePoolSize());
 
         ThreadPoolManager tpm = new ThreadPoolManager();
@@ -88,7 +89,7 @@ public class ThreadPoolManagerTest {
 
     @Test
     public void testReconfiguringCachedPool() {
-        ThreadPoolExecutor result = (ThreadPoolExecutor) ThreadPoolManager.getPool("test6");
+        ThreadPoolExecutor result = ThreadPoolManager.getPoolUnwrapped("test6");
         assertEquals(ThreadPoolManager.DEFAULT_THREAD_POOL_SIZE, result.getMaximumPoolSize());
 
         ThreadPoolManager tpm = new ThreadPoolManager();
@@ -99,4 +100,38 @@ public class ThreadPoolManagerTest {
         tpm.modified(Map.of("test6", "3"));
         assertEquals(3, result.getMaximumPoolSize());
     }
+
+    @Test
+    public void testGetPoolShutdown() throws InterruptedException {
+        checkThreadPoolWorks("Test");
+        ThreadPoolManager.getPool("Test").shutdown();
+        checkThreadPoolWorks("Test");
+        ThreadPoolManager.getPool("Test2").shutdownNow();
+        checkThreadPoolWorks("Test2");
+    }
+
+    @Test
+    public void testGetScheduledPoolShutdown() throws InterruptedException {
+        checkScheduledPoolWorks("Test2");
+        ThreadPoolManager.getScheduledPool("Test2").shutdown();
+        checkScheduledPoolWorks("Test2");
+        ThreadPoolManager.getScheduledPool("Test3").shutdownNow();
+        checkScheduledPoolWorks("Test3");
+    }
+
+    private void checkThreadPoolWorks(String poolName) throws InterruptedException {
+        ExecutorService threadPool = ThreadPoolManager.getPool(poolName);
+        CountDownLatch cdl = new CountDownLatch(1);
+        threadPool.execute(cdl::countDown);
+        assertTrue(cdl.await(1, TimeUnit.SECONDS), "Checking if thread pool " + poolName + " works");
+        assertFalse(threadPool.isShutdown(), "Checking if thread pool is not shut down");
+    }
+
+    private void checkScheduledPoolWorks(String poolName) throws InterruptedException {
+        ScheduledExecutorService threadPool = ThreadPoolManager.getScheduledPool(poolName);
+        CountDownLatch cdl = new CountDownLatch(1);
+        threadPool.schedule(cdl::countDown, 100, TimeUnit.MILLISECONDS);
+        assertTrue(cdl.await(1, TimeUnit.SECONDS), "Checking if thread pool " + poolName + " works");
+        assertFalse(threadPool.isShutdown(), "Checking if thread pool is not shut down");
+    }
 }