Use a Deque to avoid StackOverflow in ImmediateJobScheduler

deephaven · Jan 26, 2024 · 389006e · 389006e
1 parent cf4a0fd
commit 389006e
Show file tree

Hide file tree

Showing 7 changed files with 51 additions and 17 deletions.
diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/InitialFilterExecution.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/InitialFilterExecution.java
@@ -38,7 +38,7 @@ class InitialFilterExecution extends AbstractFilterExecution {
         if (permitParallelization) {
             jobScheduler = new OperationInitializerJobScheduler();
         } else {
-            jobScheduler = ImmediateJobScheduler.INSTANCE;
+            jobScheduler = new ImmediateJobScheduler();
         }
     }
 

diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/QueryTable.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/QueryTable.java
@@ -1479,7 +1479,7 @@ this, mode, columns, rowSet, getModifiedColumnSetForUpdates(), publishTheseSourc
                             && analyzer.allowCrossColumnParallelization()) {
                         jobScheduler = new OperationInitializerJobScheduler();
                     } else {
-                        jobScheduler = ImmediateJobScheduler.INSTANCE;
+                        jobScheduler = new ImmediateJobScheduler();
                     }
 
                     final QueryTable resultTable;

diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/SelectOrUpdateListener.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/SelectOrUpdateListener.java
@@ -88,7 +88,7 @@ public void onUpdate(final TableUpdate upstream) {
         if (enableParallelUpdate) {
             jobScheduler = new UpdateGraphJobScheduler(getUpdateGraph());
         } else {
-            jobScheduler = ImmediateJobScheduler.INSTANCE;
+            jobScheduler = new ImmediateJobScheduler();
         }
 
         analyzer.applyUpdate(acquiredUpdate, toClear, updateHelper, jobScheduler, this,

diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/WhereListener.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/WhereListener.java
@@ -260,7 +260,7 @@ private ListenerFilterExecution(
             if (permitParallelization) {
                 jobScheduler = new UpdateGraphJobScheduler(getUpdateGraph());
             } else {
-                jobScheduler = ImmediateJobScheduler.INSTANCE;
+                jobScheduler = new ImmediateJobScheduler();
             }
         }
 

diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/rangejoin/RangeJoinOperation.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/rangejoin/RangeJoinOperation.java
@@ -255,7 +255,7 @@ public Result<QueryTable> initialize(final boolean usePrev, final long beforeClo
         if (ExecutionContext.getContext().getOperationInitializer().canParallelize()) {
             jobScheduler = new OperationInitializerJobScheduler();
         } else {
-            jobScheduler = ImmediateJobScheduler.INSTANCE;
+            jobScheduler = new ImmediateJobScheduler();
         }
 
         final ExecutionContext executionContext = ExecutionContext.newBuilder()

diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/UpdateBy.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/UpdateBy.java
@@ -302,7 +302,7 @@ class PhasedUpdateProcessor implements LogOutputAppendable {
                 if (ExecutionContext.getContext().getOperationInitializer().canParallelize()) {
                     jobScheduler = new OperationInitializerJobScheduler();
                 } else {
-                    jobScheduler = ImmediateJobScheduler.INSTANCE;
+                    jobScheduler = new ImmediateJobScheduler();
                 }
                 executionContext = ExecutionContext.newBuilder()
                         .markSystemic().build();
@@ -331,7 +331,7 @@ class PhasedUpdateProcessor implements LogOutputAppendable {
                 if (source.getUpdateGraph().parallelismFactor() > 1) {
                     jobScheduler = new UpdateGraphJobScheduler(source.getUpdateGraph());
                 } else {
-                    jobScheduler = ImmediateJobScheduler.INSTANCE;
+                    jobScheduler = new ImmediateJobScheduler();
                 }
                 executionContext = ExecutionContext.newBuilder()
                         .setUpdateGraph(result().getUpdateGraph())

diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/util/ImmediateJobScheduler.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/util/ImmediateJobScheduler.java
@@ -1,32 +1,66 @@
 package io.deephaven.engine.table.impl.util;
 
+import com.google.common.collect.Queues;
 import io.deephaven.base.log.LogOutputAppendable;
 import io.deephaven.engine.context.ExecutionContext;
 import io.deephaven.engine.table.impl.perf.BasePerformanceEntry;
 import io.deephaven.io.log.impl.LogOutputStringImpl;
 import io.deephaven.util.SafeCloseable;
 import io.deephaven.util.process.ProcessEnvironment;
 
+import java.util.Deque;
 import java.util.function.Consumer;
 
 public class ImmediateJobScheduler implements JobScheduler {
-    public static final ImmediateJobScheduler INSTANCE = new ImmediateJobScheduler();
+
+    private Thread processingThread;
+    private final Deque<Runnable> pendingJobs = Queues.newArrayDeque();
 
     @Override
     public void submit(
             final ExecutionContext executionContext,
             final Runnable runnable,
             final LogOutputAppendable description,
             final Consumer<Exception> onError) {
-        // We do not need to install the update context since we are not changing thread contexts.
-        try (SafeCloseable ignored = executionContext != null ? executionContext.open() : null) {
-            runnable.run();
-        } catch (Exception e) {
-            onError.accept(e);
-        } catch (Error e) {
-            final String logMessage = new LogOutputStringImpl().append(description).append(" Error").toString();
-            ProcessEnvironment.getGlobalFatalErrorReporter().report(logMessage, e);
-            throw e;
+
+        final boolean alreadyRunning;
+        synchronized (this) {
+            alreadyRunning = processingThread != null;
+
+            if (alreadyRunning && processingThread != Thread.currentThread()) {
+                throw new IllegalCallerException("An unexpected thread submitted a job to this job scheduler.");
+            } else if (!alreadyRunning) {
+                processingThread = Thread.currentThread();
+            }
+
+            pendingJobs.add(() -> {
+                // We do not need to install the update context since we are not changing thread contexts.
+                try (SafeCloseable ignored = executionContext != null ? executionContext.open() : null) {
+                    runnable.run();
+                } catch (Exception e) {
+                    onError.accept(e);
+                } catch (Error e) {
+                    // we're aborting this job; might as well clean up
+                    pendingJobs.clear();
+                    processingThread = null;
+
+                    final String logMessage = new LogOutputStringImpl().append(description).append(" Error").toString();
+                    ProcessEnvironment.getGlobalFatalErrorReporter().report(logMessage, e);
+                    throw e;
+                }
+            });
+        }
+
+        if (alreadyRunning) {
+            return;
+        }
+
+        // it's my job to drain the queue
+        while (!pendingJobs.isEmpty()) {
+            pendingJobs.pop().run();
+        }
+        synchronized (this) {
+            processingThread = null;
         }
     }