EventDrivenUpdateGraph: Prevent requestRefresh from update threads (no nested refreshes allowed)

Author: rcaudy

engine/table/src/main/java/io/deephaven/engine/updategraph/impl/EventDrivenUpdateGraph.java

@@ -53,22 +53,20 @@ public int parallelismFactor() {
      */
     @Override
     public void requestRefresh() {
+        if (isUpdateThread.get()) {
+            throw new IllegalStateException("Cannot request a refresh from an update thread");
+        }
         maybeStart();
         // Do the work to refresh everything, driven by this thread. Note that we acquire the lock "early" in order to
         // avoid any inconsistencies w.r.t. assumptions about clock, lock, and update-thread state.
         final long lockStartTimeNanos = System.nanoTime();
         exclusiveLock().doLocked(() -> {
             reportLockWaitNanos(System.nanoTime() - lockStartTimeNanos);
-            final boolean wasUpdateThread = isUpdateThread.get();
-            if (!wasUpdateThread) {
-                isUpdateThread.set(true);
-            }
+            isUpdateThread.set(true);
             try (final SafeCloseable ignored = ExecutionContext.newBuilder().setUpdateGraph(this).build().open()) {
                 refreshAllTables();
             } finally {
-                if (!wasUpdateThread) {
-                    isUpdateThread.remove();
-                }
+                isUpdateThread.remove();
             }
         });
         final long nowNanos = System.nanoTime();

engine/table/src/test/java/io/deephaven/engine/updategraph/impl/TestEventDrivenUpdateGraph.java

@@ -17,6 +17,7 @@
 import io.deephaven.engine.table.impl.perf.UpdatePerformanceTracker;
 import io.deephaven.engine.table.impl.sources.LongSingleValueSource;
 import io.deephaven.engine.testutil.TstUtils;
+import io.deephaven.engine.updategraph.TerminalNotification;
 import io.deephaven.engine.updategraph.UpdateGraph;
 import io.deephaven.engine.util.TableTools;
 import io.deephaven.util.SafeCloseable;
@@ -26,7 +27,9 @@
 import org.junit.*;
 
 import java.nio.file.Path;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.*;
 
 import static io.deephaven.engine.context.TestExecutionContext.OPERATION_INITIALIZATION;
@@ -170,6 +173,7 @@ public void testSimpleModify() {
     @Test
     public void testRefreshRace() throws ExecutionException, InterruptedException, TimeoutException {
         final EventDrivenUpdateGraph eventDrivenUpdateGraph = EventDrivenUpdateGraph.newBuilder("TestEDUG").build();
+        final List<Runnable> retainedReferences = new ArrayList<>();
 
         final MutableInt sourceRefreshCount = new MutableInt(0);
         final Runnable sleepingSource = () -> {
@@ -180,6 +184,7 @@ public void testRefreshRace() throws ExecutionException, InterruptedException, T
                 Assert.fail("Interrupted while sleeping");
             }
         };
+        retainedReferences.add(sleepingSource);
         eventDrivenUpdateGraph.addSource(sleepingSource);
 
         final int numConcurrentRefreshes = 10;
@@ -199,6 +204,21 @@ public void testRefreshRace() throws ExecutionException, InterruptedException, T
         }
 
         Assert.assertEquals(numConcurrentRefreshes, sourceRefreshCount.intValue());
+        Assert.assertEquals(sleepingSource, retainedReferences.get(0));
+    }
+
+    @Test
+    public void testIllegalRefresh() {
+        final EventDrivenUpdateGraph eventDrivenUpdateGraph = EventDrivenUpdateGraph.newBuilder("TestEDUG").build();
+
+        eventDrivenUpdateGraph.addNotification(new TerminalNotification() {
+            @Override
+            public void run() {
+                Assert.assertThrows(IllegalStateException.class, eventDrivenUpdateGraph::requestRefresh);
+            }
+        });
+
+        eventDrivenUpdateGraph.requestRefresh();
     }
 
     @Test