Using query fix: engine thread pool (checked for dead-locks)

Author: lvca

engine/src/main/java/com/arcadedb/query/QueryEngineManager.java

@@ -36,12 +36,14 @@ public class QueryEngineManager {
 
   private QueryEngineManager() {
     final int maxThreads = Math.max(2, Runtime.getRuntime().availableProcessors());
-    executorService = new ThreadPoolExecutor(0, maxThreads, 60L, TimeUnit.SECONDS,
+    final ThreadPoolExecutor pool = new ThreadPoolExecutor(maxThreads, maxThreads, 60L, TimeUnit.SECONDS,
         new LinkedBlockingQueue<>(), r -> {
       final Thread t = new Thread(r, "ArcadeDB-QueryWorker");
       t.setDaemon(true);
       return t;
     });
+    pool.allowCoreThreadTimeOut(true);
+    executorService = pool;
 
     // REGISTER ALL THE SUPPORTED LANGUAGE FROM POLYGLOT ENGINE
     for (final String language : PolyglotQueryEngine.PolyglotQueryEngineFactory.getSupportedLanguages())

engine/src/main/java/com/arcadedb/query/opencypher/executor/operators/GAVFusedChainOperator.java

@@ -32,13 +32,17 @@
 import com.arcadedb.query.sql.executor.ResultSet;
 import com.arcadedb.utility.LongLongHashMap;
 
+import com.arcadedb.query.QueryEngineManager;
+
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.NoSuchElementException;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 
 /**
  * Fused multi-hop GAV traversal operator — zero intermediate object allocation.
@@ -177,7 +181,6 @@ else if (sourceObj instanceof Vertex)
     // Parallel DFS: each thread processes a chunk of source vertices with its own stack
     @SuppressWarnings("unchecked")
     final List<Result>[] threadResults = new List[Math.min(parallelism, Math.max(1, (totalSources + chunkSize - 1) / chunkSize))];
-    final AtomicReference<Throwable> firstError = new AtomicReference<>();
 
     final int threadCount;
     if (totalSources < 8192) {
@@ -186,8 +189,9 @@ else if (sourceObj instanceof Vertex)
       traverseChunk(sourceNodeIds, 0, totalSources, hopViews, chainLength, outputNames, db, context, threadResults[0]);
       threadCount = 1;
     } else {
-      // Parallel execution
-      final Thread[] threads = new Thread[threadResults.length];
+      // Parallel execution using shared query worker pool
+      final ExecutorService executor = QueryEngineManager.getInstance().getExecutorService();
+      final Future<?>[] futures = new Future<?>[threadResults.length];
       int launched = 0;
       for (int t = 0; t < threadResults.length; t++) {
         final int start = t * chunkSize;
@@ -196,29 +200,21 @@ else if (sourceObj instanceof Vertex)
           break;
         threadResults[t] = new ArrayList<>();
         final int threadIdx = t;
-        threads[t] = new Thread(() -> {
-          try {
-            traverseChunk(sourceNodeIds, start, end, hopViews, chainLength, outputNames, db, context, threadResults[threadIdx]);
-          } catch (final Throwable e) {
-            firstError.compareAndSet(null, e);
-          }
-        });
-        threads[t].setDaemon(true);
-        threads[t].setName("gav-chain-" + t);
-        threads[t].start();
+        futures[t] = executor.submit(() ->
+            traverseChunk(sourceNodeIds, start, end, hopViews, chainLength, outputNames, db, context, threadResults[threadIdx]));
         launched++;
       }
-      // Wait for all threads
+      // Wait for all tasks
       for (int t = 0; t < launched; t++) {
         try {
-          threads[t].join();
+          futures[t].get();
         } catch (final InterruptedException e) {
           Thread.currentThread().interrupt();
           break;
+        } catch (final ExecutionException e) {
+          throw new RuntimeException("Parallel GAV traversal failed", e.getCause());
         }
       }
-      if (firstError.get() != null)
-        throw new RuntimeException("Parallel GAV traversal failed", (Exception) firstError.get());
       threadCount = launched;
     }
 
@@ -275,15 +271,14 @@ private ResultSet executeWithFusedAggregation(final int[] sourceNodeIds, final i
     final int numThreads = Math.min(parallelism, Math.max(1, (totalSources + chunkSize - 1) / chunkSize));
     @SuppressWarnings("unchecked")
     final LongLongHashMap[] threadMaps = new LongLongHashMap[numThreads];
-    final AtomicReference<Throwable> firstError = new AtomicReference<>();
-
     if (totalSources < 8192) {
       // Single-threaded
       threadMaps[0] = new LongLongHashMap();
       aggregateChunk(sourceNodeIds, 0, totalSources, hopViews, chainLength, groupKeySlots, db, context, threadMaps[0]);
     } else {
-      // Parallel
-      final Thread[] threads = new Thread[numThreads];
+      // Parallel using shared query worker pool
+      final ExecutorService executor = QueryEngineManager.getInstance().getExecutorService();
+      final Future<?>[] futures = new Future<?>[numThreads];
       int launched = 0;
       for (int t = 0; t < numThreads; t++) {
         final int start = t * chunkSize;
@@ -292,28 +287,20 @@ private ResultSet executeWithFusedAggregation(final int[] sourceNodeIds, final i
           break;
         threadMaps[t] = new LongLongHashMap();
         final int threadIdx = t;
-        threads[t] = new Thread(() -> {
-          try {
-            aggregateChunk(sourceNodeIds, start, end, hopViews, chainLength, groupKeySlots, db, context, threadMaps[threadIdx]);
-          } catch (final Throwable e) {
-            firstError.compareAndSet(null, e);
-          }
-        });
-        threads[t].setDaemon(true);
-        threads[t].setName("gav-agg-" + t);
-        threads[t].start();
+        futures[t] = executor.submit(() ->
+            aggregateChunk(sourceNodeIds, start, end, hopViews, chainLength, groupKeySlots, db, context, threadMaps[threadIdx]));
         launched++;
       }
       for (int t = 0; t < launched; t++) {
         try {
-          threads[t].join();
+          futures[t].get();
         } catch (final InterruptedException e) {
           Thread.currentThread().interrupt();
           break;
+        } catch (final ExecutionException e) {
+          throw new RuntimeException("Parallel GAV aggregation failed", e.getCause());
         }
       }
-      if (firstError.get() != null)
-        throw new RuntimeException("Parallel GAV aggregation failed", (Exception) firstError.get());
     }
 
     // Merge thread-local maps (zero boxing — primitive long operations)

engine/src/main/java/com/arcadedb/query/opencypher/executor/steps/PartitionedTriangleOp.java

@@ -27,12 +27,17 @@
 import com.arcadedb.schema.DocumentType;
 import com.arcadedb.schema.VertexType;
 
+import com.arcadedb.query.QueryEngineManager;
+
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 
 /**
  * Count operator for country-partitioned triangle patterns (Q3).
@@ -78,22 +83,24 @@ public long execute(final GraphTraversalProvider provider, final Database db) {
     if (nodeCount < 1000) {
       partialCounts[0] = countRange(knowsView, nbrs, personPartition, 0, nodeCount);
     } else {
-      final Thread[] threads = new Thread[threadCount];
+      final ExecutorService executor = QueryEngineManager.getInstance().getExecutorService();
+      final Future<?>[] futures = new Future<?>[threadCount];
       final int chunkSize = (nodeCount + threadCount - 1) / threadCount;
       for (int t = 0; t < threadCount; t++) {
         final int start = t * chunkSize;
         final int end = Math.min(start + chunkSize, nodeCount);
         final int threadIdx = t;
-        threads[t] = new Thread(() ->
+        futures[t] = executor.submit(() ->
             partialCounts[threadIdx] = countRange(knowsView, nbrs, personPartition, start, end));
-        threads[t].start();
       }
-      for (final Thread thread : threads) {
+      for (final Future<?> future : futures) {
         try {
-          thread.join();
+          future.get();
         } catch (final InterruptedException e) {
           Thread.currentThread().interrupt();
           break;
+        } catch (final ExecutionException e) {
+          throw new RuntimeException("Parallel triangle counting failed", e.getCause());
         }
       }
     }

engine/src/main/java/com/arcadedb/query/sql/executor/FetchFromTypeExecutionStep.java

@@ -279,8 +279,10 @@ private ResultSet syncPullParallel(final CommandContext context, final int nReco
         scanFutures.add(future);
       }
 
-      // Background thread to signal completion
-      scanExecutor.submit(() -> {
+      // Lightweight daemon thread to signal completion — must NOT run on the shared
+      // query pool to avoid thread starvation (this thread blocks on f.get() waiting
+      // for scan tasks that themselves run on the pool).
+      final Thread completionThread = new Thread(() -> {
         for (final Future<?> f : scanFutures) {
           try {
             f.get();
@@ -289,7 +291,9 @@ private ResultSet syncPullParallel(final CommandContext context, final int nReco
           }
         }
         parallelScanComplete = true;
-      });
+      }, "ArcadeDB-ScanCompletion");
+      completionThread.setDaemon(true);
+      completionThread.start();
     }
 
     return new ResultSet() {