Merged feature/thread-pool-improvements into dev

qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/ActionHelper.java

@@ -37,6 +37,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.model.session.QSession;
 import com.kingsrook.qqq.backend.core.modules.authentication.QAuthenticationModuleDispatcher;
 import com.kingsrook.qqq.backend.core.modules.authentication.QAuthenticationModuleInterface;
+import com.kingsrook.qqq.backend.core.utils.PrefixedDefaultThreadFactory;
 
 
 /*******************************************************************************
@@ -55,7 +56,7 @@ public class ActionHelper
    /////////////////////////////////////////////////////////////////////////////
    private static Integer         CORE_THREADS    = 8;
    private static Integer         MAX_THREADS     = 500;
-   private static ExecutorService executorService = new ThreadPoolExecutor(CORE_THREADS, MAX_THREADS, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());
+   private static ExecutorService executorService = new ThreadPoolExecutor(CORE_THREADS, MAX_THREADS, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), new PrefixedDefaultThreadFactory(ActionHelper.class));
 
 
 

qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/async/AsyncJobManager.java

@@ -42,6 +42,7 @@ import com.kingsrook.qqq.backend.core.state.InMemoryStateProvider;
 import com.kingsrook.qqq.backend.core.state.StateProviderInterface;
 import com.kingsrook.qqq.backend.core.state.StateType;
 import com.kingsrook.qqq.backend.core.state.UUIDAndTypeStateKey;
+import com.kingsrook.qqq.backend.core.utils.PrefixedDefaultThreadFactory;
 import com.kingsrook.qqq.backend.core.utils.StringUtils;
 import org.apache.logging.log4j.Level;
 import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
@@ -65,7 +66,7 @@ public class AsyncJobManager
    /////////////////////////////////////////////////////////////////////////////
    private static Integer         CORE_THREADS    = 8;
    private static Integer         MAX_THREADS     = 500;
-   private static ExecutorService executorService = new ThreadPoolExecutor(CORE_THREADS, MAX_THREADS, 60L, TimeUnit.SECONDS, new SynchronousQueue<>());
+   private static ExecutorService executorService = new ThreadPoolExecutor(CORE_THREADS, MAX_THREADS, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), new PrefixedDefaultThreadFactory(AsyncJobManager.class));
 
 
    private String forcedJobUUID = null;

qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/tables/helpers/ActionTimeoutHelper.java

@@ -23,8 +23,10 @@ package com.kingsrook.qqq.backend.core.actions.tables.helpers;
 
 
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import com.kingsrook.qqq.backend.core.utils.PrefixedDefaultThreadFactory;
 
 
 /*******************************************************************************
@@ -50,6 +52,9 @@ public class ActionTimeoutHelper
 
    private boolean didTimeout = false;
 
+   private static Integer                  CORE_THREADS             = 10;
+   private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(CORE_THREADS, new PrefixedDefaultThreadFactory(ActionTimeoutHelper.class));
+
 
 
    /*******************************************************************************
@@ -75,7 +80,7 @@ public class ActionTimeoutHelper
          return;
       }
 
-      future = Executors.newSingleThreadScheduledExecutor().schedule(() ->
+      future = scheduledExecutorService.schedule(() ->
       {
          didTimeout = true;
          runnable.run();

qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/tables/helpers/QueryStatManager.java

@@ -57,6 +57,7 @@ import com.kingsrook.qqq.backend.core.model.session.QSession;
 import com.kingsrook.qqq.backend.core.model.tables.QQQTable;
 import com.kingsrook.qqq.backend.core.model.tables.QQQTablesMetaDataProvider;
 import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
+import com.kingsrook.qqq.backend.core.utils.PrefixedDefaultThreadFactory;
 import com.kingsrook.qqq.backend.core.utils.StringUtils;
 import com.kingsrook.qqq.backend.core.utils.collections.MapBuilder;
 import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
@@ -176,7 +177,7 @@ public class QueryStatManager
       active = true;
       queryStats = new ArrayList<>();
 
-      executorService = Executors.newSingleThreadScheduledExecutor();
+      executorService = Executors.newSingleThreadScheduledExecutor(new PrefixedDefaultThreadFactory(this));
       executorService.scheduleAtFixedRate(new QueryStatManagerInsertJob(), jobInitialDelay, jobPeriodSeconds, TimeUnit.SECONDS);
    }
 

qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/scheduler/simple/StandardScheduledExecutor.java

@@ -29,6 +29,7 @@ import java.util.function.Supplier;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.utils.PrefixedDefaultThreadFactory;
 
 
 /*******************************************************************************
@@ -95,7 +96,7 @@ public class StandardScheduledExecutor
       }
 
       LOG.info("Starting [" + name + "]");
-      service = Executors.newSingleThreadScheduledExecutor();
+      service = Executors.newSingleThreadScheduledExecutor(new PrefixedDefaultThreadFactory(this));
       service.scheduleWithFixedDelay(getRunnable(), initialDelayMillis, delayMillis, TimeUnit.MILLISECONDS);
       runningState = RunningState.RUNNING;
       return (true);

qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/state/InMemoryStateProvider.java

@@ -32,6 +32,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.utils.PrefixedDefaultThreadFactory;
 import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
 
 
@@ -62,7 +63,7 @@ public class InMemoryStateProvider implements StateProviderInterface
       ///////////////////////////////////////////////////////////
       // Start a single thread executor to handle the cleaning //
       ///////////////////////////////////////////////////////////
-      ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+      ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(new PrefixedDefaultThreadFactory(this));
       executorService.scheduleAtFixedRate(new InMemoryStateProvider.InMemoryStateProviderCleanJob(), jobInitialDelay, jobPeriodSeconds, TimeUnit.SECONDS);
    }
 

qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/utils/PrefixedDefaultThreadFactory.java

@@ -0,0 +1,99 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2024.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package com.kingsrook.qqq.backend.core.utils;
+
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
+
+/*******************************************************************************
+ ** ThreadFactory implementation that puts a common prefix on all threads.
+ **
+ ** Makes it so that, instead of having 100s of pool-x-thread-y names that are
+ ** hard to tell apart, they can have a prefix:  MyService-pool-x-thread-y, vs
+ ** YourThing-pool-x-thread-y.
+ **
+ ** You can put '-' at the end of your threadNamePrefix (constructor arg) or
+ ** you can omit it, either way, we'll make it look like shown above.
+ *******************************************************************************/
+public class PrefixedDefaultThreadFactory implements ThreadFactory
+{
+   private final String        threadNamePrefix;
+   private final ThreadFactory threadFactory;
+
+
+
+   /*******************************************************************************
+    ** Constructor
+    **
+    *******************************************************************************/
+   public PrefixedDefaultThreadFactory(String threadNamePrefix)
+   {
+      if(StringUtils.hasContent(threadNamePrefix))
+      {
+         this.threadNamePrefix = threadNamePrefix.replaceAll("-+$", "") + "-";
+      }
+      else
+      {
+         this.threadNamePrefix = "";
+      }
+
+      threadFactory = Executors.defaultThreadFactory();
+   }
+
+
+
+   /*******************************************************************************
+    ** Constructor
+    **
+    *******************************************************************************/
+   public PrefixedDefaultThreadFactory(Class<?> callerClass)
+   {
+      this(callerClass != null ? callerClass.getSimpleName() : "");
+   }
+
+
+
+   /*******************************************************************************
+    ** Constructor
+    **
+    *******************************************************************************/
+   public PrefixedDefaultThreadFactory(Object caller)
+   {
+      this(caller != null ? caller.getClass() : null);
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   @Override
+   public Thread newThread(Runnable r)
+   {
+      Thread thread = threadFactory.newThread(r);
+      thread.setName(threadNamePrefix + thread.getName());
+      return (thread);
+   }
+
+}

qqq-backend-core/src/test/java/com/kingsrook/qqq/backend/core/actions/tables/helpers/ActionTimeoutHelperTest.java

@@ -22,10 +22,19 @@
 package com.kingsrook.qqq.backend.core.actions.tables.helpers;
 
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import com.kingsrook.qqq.backend.core.BaseTest;
 import com.kingsrook.qqq.backend.core.utils.SleepUtils;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -35,7 +44,18 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
  *******************************************************************************/
 class ActionTimeoutHelperTest extends BaseTest
 {
-   boolean didCancel = false;
+   private static AtomicInteger cancelCount;
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   @BeforeEach
+   void beforeEach()
+   {
+      cancelCount = new AtomicInteger(0);
+   }
 
 
 
@@ -45,11 +65,10 @@ class ActionTimeoutHelperTest extends BaseTest
    @Test
    void testTimesOut()
    {
-      didCancel = false;
       ActionTimeoutHelper actionTimeoutHelper = new ActionTimeoutHelper(10, TimeUnit.MILLISECONDS, () -> doCancel());
       actionTimeoutHelper.start();
       SleepUtils.sleep(50, TimeUnit.MILLISECONDS);
-      assertTrue(didCancel);
+      assertEquals(1, cancelCount.get());
       assertTrue(actionTimeoutHelper.getDidTimeout());
    }
 
@@ -61,25 +80,58 @@ class ActionTimeoutHelperTest extends BaseTest
    @Test
    void testGetsCancelled()
    {
-      didCancel = false;
       ActionTimeoutHelper actionTimeoutHelper = new ActionTimeoutHelper(100, TimeUnit.MILLISECONDS, () -> doCancel());
       actionTimeoutHelper.start();
       SleepUtils.sleep(10, TimeUnit.MILLISECONDS);
       actionTimeoutHelper.cancel();
-      assertFalse(didCancel);
+      assertEquals(0, cancelCount.get());
       SleepUtils.sleep(200, TimeUnit.MILLISECONDS);
-      assertFalse(didCancel);
+      assertEquals(0, cancelCount.get());
       assertFalse(actionTimeoutHelper.getDidTimeout());
    }
 
 
 
+   /*******************************************************************************
+    ** goal here is - confirm that we can have more threads running at same time
+    ** than we have threads allocated to the ActionTimeoutHelper's thread pool,
+    ** and they should all still get cancelled.
+    *******************************************************************************/
+   @Test
+   void testManyThreads() throws InterruptedException, ExecutionException
+   {
+      int N = 50;
+
+      ExecutorService executorService = Executors.newCachedThreadPool();
+      List<Future<?>> futureList      = new ArrayList<>();
+
+      for(int i = 0; i < N; i++)
+      {
+         System.out.println("Submitting: " + i);
+         futureList.add(executorService.submit(() ->
+         {
+            ActionTimeoutHelper actionTimeoutHelper = new ActionTimeoutHelper(10, TimeUnit.MILLISECONDS, () -> doCancel());
+            actionTimeoutHelper.start();
+            SleepUtils.sleep(1, TimeUnit.SECONDS);
+         }));
+      }
+
+      for(Future<?> future : futureList)
+      {
+         future.get();
+      }
+
+      assertEquals(N, cancelCount.get());
+   }
+
+
+
    /*******************************************************************************
     **
     *******************************************************************************/
    private void doCancel()
    {
-      didCancel = true;
+      cancelCount.getAndIncrement();
    }
 
 }

qqq-backend-core/src/test/java/com/kingsrook/qqq/backend/core/utils/PrefixedDefaultThreadFactoryTest.java

@@ -0,0 +1,65 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2024.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package com.kingsrook.qqq.backend.core.utils;
+
+
+import com.kingsrook.qqq.backend.core.BaseTest;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Unit test for PrefixedDefaultThreadFactory 
+ *******************************************************************************/
+class PrefixedDefaultThreadFactoryTest extends BaseTest
+{
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   @Test
+   void test()
+   {
+      assertThat(new PrefixedDefaultThreadFactory("ItsMe").newThread(this::noop).getName()).startsWith("ItsMe-pool");
+      assertThat(new PrefixedDefaultThreadFactory("ItsMe-").newThread(this::noop).getName()).startsWith("ItsMe-pool");
+      assertThat(new PrefixedDefaultThreadFactory("ItsMe--").newThread(this::noop).getName()).startsWith("ItsMe-pool");
+      assertThat(new PrefixedDefaultThreadFactory((String) null).newThread(this::noop).getName()).startsWith("pool");
+      assertThat(new PrefixedDefaultThreadFactory((Class<?>) null).newThread(this::noop).getName()).startsWith("pool");
+      assertThat(new PrefixedDefaultThreadFactory((Object) null).newThread(this::noop).getName()).startsWith("pool");
+      assertThat(new PrefixedDefaultThreadFactory("").newThread(this::noop).getName()).startsWith("pool");
+      assertThat(new PrefixedDefaultThreadFactory(" ").newThread(this::noop).getName()).startsWith("pool");
+      assertThat(new PrefixedDefaultThreadFactory(this).newThread(this::noop).getName()).startsWith(getClass().getSimpleName() + "-pool");
+      assertThat(new PrefixedDefaultThreadFactory(InsertAction.class).newThread(this::noop).getName()).startsWith(InsertAction.class.getSimpleName() + "-pool");
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   void noop()
+   {
+      System.out.println("noop");
+   }
+
+}