EventDrivenUpdateGraph: Allow two concurrent threads to safely requestRefresh()

Author: rcaudy

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

@@ -629,7 +629,7 @@ private void flushTerminalNotifications() {
             runNotification(notificationForThisThread);
         }
 
-        // We can not proceed until all of the terminal notifications have executed.
+        // We can not proceed until all the terminal notifications have executed.
         notificationProcessor.doAllWork();
     }
 
@@ -836,6 +836,10 @@ private void computeStatsAndLogCycle(final long cycleTimeNanos) {
         }
     }
 
+    void reportLockWaitNanos(final long lockWaitNanos) {
+        currentCycleLockWaitTotalNanos += lockWaitNanos;
+    }
+
     /**
      * Is the provided cycle time on budget?
      *
@@ -907,7 +911,7 @@ void refreshAllTables() {
     private void doRefresh(@NotNull final Runnable refreshFunction) {
         final long lockStartTimeNanos = System.nanoTime();
         exclusiveLock().doLocked(() -> {
-            currentCycleLockWaitTotalNanos += System.nanoTime() - lockStartTimeNanos;
+            reportLockWaitNanos(System.nanoTime() - lockStartTimeNanos);
             if (!running) {
                 return;
             }

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

@@ -54,13 +54,23 @@ public int parallelismFactor() {
     @Override
     public void requestRefresh() {
         maybeStart();
-        // do the work to refresh everything, on this thread
-        isUpdateThread.set(true);
-        try (final SafeCloseable ignored = ExecutionContext.newBuilder().setUpdateGraph(this).build().open()) {
-            refreshAllTables();
-        } finally {
-            isUpdateThread.remove();
-        }
+        // 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);
+            }
+            try (final SafeCloseable ignored = ExecutionContext.newBuilder().setUpdateGraph(this).build().open()) {
+                refreshAllTables();
+            } finally {
+                if (!wasUpdateThread) {
+                    isUpdateThread.remove();
+                }
+            }
+        });
         final long nowNanos = System.nanoTime();
         synchronized (this) {
             maybeFlushUpdatePerformance(nowNanos, nowNanos);

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

@@ -22,10 +22,12 @@
 import io.deephaven.util.SafeCloseable;
 import io.deephaven.util.annotations.ReflexiveUse;
 import junit.framework.TestCase;
+import org.apache.commons.lang3.mutable.MutableInt;
 import org.junit.*;
 
 import java.nio.file.Path;
 import java.util.Collections;
+import java.util.concurrent.*;
 
 import static io.deephaven.engine.context.TestExecutionContext.OPERATION_INITIALIZATION;
 import static io.deephaven.engine.util.TableTools.*;
@@ -165,6 +167,40 @@ public void testSimpleModify() {
         }
     }
 
+    @Test
+    public void testRefreshRace() throws ExecutionException, InterruptedException, TimeoutException {
+        final EventDrivenUpdateGraph eventDrivenUpdateGraph = EventDrivenUpdateGraph.newBuilder("TestEDUG").build();
+
+        final MutableInt sourceRefreshCount = new MutableInt(0);
+        final Runnable sleepingSource = () -> {
+            try {
+                Thread.sleep(100);
+                sourceRefreshCount.increment();
+            } catch (InterruptedException e) {
+                Assert.fail("Interrupted while sleeping");
+            }
+        };
+        eventDrivenUpdateGraph.addSource(sleepingSource);
+
+        final int numConcurrentRefreshes = 10;
+        final Future<?>[] refreshFutures = new Future[numConcurrentRefreshes];
+        final ExecutorService executor = Executors.newFixedThreadPool(numConcurrentRefreshes);
+        try {
+            for (int cri = 0; cri < numConcurrentRefreshes; ++cri) {
+                refreshFutures[cri] = executor.submit(eventDrivenUpdateGraph::requestRefresh);
+                Thread.sleep(10);
+            }
+            for (final Future<?> refreshFuture : refreshFutures) {
+                refreshFuture.get();// 1, TimeUnit.SECONDS);
+            }
+        } finally {
+            executor.shutdown();
+            Assert.assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS));
+        }
+
+        Assert.assertEquals(numConcurrentRefreshes, sourceRefreshCount.intValue());
+    }
+
     @Test
     public void testUpdatePerformanceTracker() {
         final Table upt = UpdatePerformanceTracker.getQueryTable();