fix: DH-20516: Add implicit barriers to serial Selectables. (#7252) (#7255)

Author: cpwright

docs/groovy/conceptual/query-engine/parallelization.md

@@ -85,7 +85,10 @@ The [`ConcurrencyControl`](https://docs.deephaven.io/core/javadoc/io/deephaven/a
 To explicitly mark a `Selectable` or `Filter` as stateful, use the `withSerial` method.
 
 - A serial `Filter` cannot be reordered with respect to other `Filter`s. Every input row to a stateful `Filter` is evaluated in order.
-- When a `Selectable` is serial, then every row for that column is evaluated in order. For `Selectable`s, no additional ordering between expressions is imposed. As with every `select` or `update` call, if column B references column A, then the necessary inputs to column B from column A are evaluated before column B is evaluated. To impose further ordering constraints, use barriers.
+- When a `Selectable` is serial, then every row for that column is evaluated in order.
+- For `Selectable`s, additional ordering constraints are controlled by the value of the `QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS`. This is set by the property `QueryTable.serialSelectImplicitBarriers` (defaulting to the value of `QueryTable.statelessSelectByDefault`).
+- If `QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS` is false, no additional ordering between expressions is imposed. As with every `select` or `update` call, if column B references column A, then the necessary inputs to column B from column A are evaluated before column B is evaluated. To impose further ordering constraints, use barriers.
+- If `QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS` is true, then a serial `Selectable` is an absolute barrier with respect to all other serial `Selectable`s. This prohibits serial `Selectable`s from being evaluated concurrently, permitting them to access global state. `Selectable`s that are not serial may be reordered with respect to a serial selectable.
 
 `Filter`s and `Selectable`s may declare a _barrier_. A barrier is an opaque object (compared using reference equality) that is used to mark a particular `Filter` or `Selectable`. Subsequent `Filter`s or `Selectable`s may respect a previously declared barrier. If a `Filter` respects a barrier, that `Filter` cannot begin evaluation until the `Filter` that declares the barrier has been completely evaluated. Similarly, if a `Selectable` respects a barrier, it cannot begin evaluation until the `Selectable` that declared the barrier has been completely evaluated.
 

engine/table/src/main/java/io/deephaven/engine/table/impl/QueryTable.java

@@ -317,6 +317,14 @@ public interface MemoizableOperation<T extends DynamicNode & NotificationStepRec
     public static boolean STATELESS_SELECT_BY_DEFAULT =
             Configuration.getInstance().getBooleanWithDefault("QueryTable.statelessSelectByDefault", false);
 
+    /**
+     * If set to true, then stateful SelectColumns form implicit barriers. If set to false, then StatefulSelectColumns
+     * do not form implicit barriers.
+     */
+    public static boolean SERIAL_SELECT_IMPLICIT_BARRIERS =
+            Configuration.getInstance().getBooleanWithDefault("QueryTable.serialSelectImplicitBarriers",
+                    STATELESS_SELECT_BY_DEFAULT);
+
     private static final AtomicReferenceFieldUpdater<QueryTable, ModifiedColumnSet> MODIFIED_COLUMN_SET_UPDATER =
             AtomicReferenceFieldUpdater.newUpdater(QueryTable.class, ModifiedColumnSet.class, "modifiedColumnSet");
     private static final AtomicReferenceFieldUpdater<QueryTable, Map> CACHED_OPERATIONS_UPDATER =

engine/table/src/main/java/io/deephaven/engine/table/impl/select/analyzers/SelectAndViewAnalyzer.java

@@ -158,11 +158,12 @@ public static AnalyzerContext createContext(
         compilationProcessor.compile();
 
         final Map<Object, Integer> barrierToLayerIndex = new IdentityHashMap<>();
+        final List<Object> implicitSerialBarriers = new ArrayList<>();
 
         // Second pass builds the analyzer and destination columns
         final HashMap<String, ColumnSource<?>> resultAlias = new HashMap<>();
         for (int columnIndex = 0; columnIndex < context.processedCols.size(); ++columnIndex) {
-            final SelectColumn sc = context.processedCols.get(columnIndex);
+            SelectColumn sc = context.processedCols.get(columnIndex);
 
             // if this select column depends on result column then its updates must happen in result-key-space
             // note: if flatResult is true then we are not preserving any parent columns
@@ -172,6 +173,17 @@ public static AnalyzerContext createContext(
 
             sc.initInputs(rowSet, useResultKeySpace ? context.allSourcesInResultKeySpace : context.allSources);
 
+            if (QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS && !sc.isStateless()) {
+                final Object implicitBarrier = new ImplicitBarrier(sc.getName(), columnIndex);
+                if (!implicitSerialBarriers.isEmpty()) {
+                    sc = sc.withRespectedBarriers(implicitSerialBarriers.toArray())
+                            .withDeclaredBarriers(implicitBarrier);
+                } else {
+                    sc = sc.withDeclaredBarriers(implicitBarrier);
+                }
+                implicitSerialBarriers.add(implicitBarrier);
+            }
+
             // TODO (deephaven-core#5760): If layers may define more than one column, we'll need to fix resultAlias.
             // new columns shadow known aliases
             resultAlias.remove(sc.getName());
@@ -882,6 +894,24 @@ public void applyUpdate(
         scheduler.tryToKickOffWork();
     }
 
+    /**
+     * A class used as an implicit barrier for serial selectables, with a useful toString.
+     */
+    private static class ImplicitBarrier {
+        private final String name;
+        private final int fci;
+
+        public ImplicitBarrier(String name, int fci) {
+            this.name = name;
+            this.fci = fci;
+        }
+
+        @Override
+        public String toString() {
+            return "Implicit barrier for Serial column " + name + " (index " + fci + ")";
+        }
+    }
+
     private class UpdateScheduler {
         private final ReentrantLock runLock = new ReentrantLock();
 

engine/table/src/test/java/io/deephaven/engine/table/impl/QueryTableSelectBarrierTest.java

@@ -19,7 +19,6 @@
 import io.deephaven.engine.util.TableTools;
 import io.deephaven.util.SafeCloseable;
 import io.deephaven.util.thread.ThreadInitializationFactory;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 
@@ -73,6 +72,7 @@ private void testBarrierSelectColumn() {
                 final SafeCloseable ignored3 = threadPool::shutdown) {
             QueryTable.FORCE_PARALLEL_SELECT_AND_UPDATE = true;
             QueryTable.STATELESS_SELECT_BY_DEFAULT = false;
+            QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS = false;
 
             final SelectColumn sa =
                     SelectColumnFactory.getExpression("A=slow_a.getAndIncrementSlow(0, new int[]{ 0, 1 })");
@@ -156,6 +156,7 @@ private void testBarrierSelectable() {
                 final SafeCloseable ignored3 = threadPool::shutdown) {
             QueryTable.FORCE_PARALLEL_SELECT_AND_UPDATE = true;
             QueryTable.STATELESS_SELECT_BY_DEFAULT = true;
+            QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS = false;
 
             final Selectable sa =
                     Selectable.of(ColumnName.of("A"), Method.of(ColumnName.of("AtomicInt"), "getAndIncrementSlow",
@@ -227,6 +228,79 @@ private void testBarrierSelectable() {
         }
     }
 
+    // if you would like to convince yourself that we are reliably producing out-of-order conditions as appropriate
+    @Test
+    public void testRepeatedImplicitBarriers() {
+        for (int ii = 0; ii < REPEATS_FOR_CONFIDENCE; ++ii) {
+            System.out.println("Repetition " + ii);
+            try (final SafeCloseable ignored = LivenessScopeStack.open()) {
+                testImplicitBarriers();
+            }
+        }
+    }
+
+    private void testImplicitBarriers() {
+        final int segments = 10;
+        final int size = Math.toIntExact(QueryTable.MINIMUM_PARALLEL_SELECT_ROWS * segments);
+        final SlowedAtomicInteger slow_a = new SlowedAtomicInteger(0, 2);
+        QueryScope.addParam("slow_a", slow_a);
+        QueryScope.addParam("size", size);
+        final Table x = TableTools.emptyTable(size)
+                .updateView("AtomicInt=slow_a")
+                .updateView("SecondChecks=new int[]{0}")
+                .updateView("FirstChecks=new int[]{1}");
+
+        // use segments times the number of columns we are evaluating threads, so we can start off each of the
+        // individual blocks of data at once to maximize "chaos" in
+        // terms of the atomic integers being mixed up
+        final OperationInitializationThreadPool threadPool =
+                new OperationInitializationThreadPool(ThreadInitializationFactory.NO_OP, segments * 4);
+        final ExecutionContext executionContext = ExecutionContext.getContext().withOperationInitializer(threadPool);
+        try (final SafeCloseable ignored = executionContext.open();
+                final SafeCloseable ignored2 = new SaveQueryTableOptions();
+                final SafeCloseable ignored3 = threadPool::shutdown) {
+            QueryTable.FORCE_PARALLEL_SELECT_AND_UPDATE = true;
+            QueryTable.STATELESS_SELECT_BY_DEFAULT = true;
+            QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS = true;
+
+            final Selectable sa =
+                    Selectable.of(ColumnName.of("A"), Method.of(ColumnName.of("AtomicInt"), "getAndIncrementSlow",
+                            Literal.of(0), ColumnName.of("FirstChecks")));
+            final Selectable sb =
+                    Selectable.of(ColumnName.of("B"), Method.of(ColumnName.of("AtomicInt"), "getAndIncrementSlow",
+                            Literal.of(1), ColumnName.of("SecondChecks")));
+
+            final Selectable sc = Selectable.of(ColumnName.of("C"), Literal.of(1));
+
+            slow_a.reset(0, -1);
+            System.out.println(Instant.now() + ": serial");
+            final Table y = x.update(List.of(sa.withSerial(), sb.withSerial()));
+
+            final Table expected =
+                    x.updateView(List.of(SelectColumn.ofStateless(SelectColumnFactory.getExpression("A=i"),
+                            SelectColumnFactory.getExpression("B=size + i"))));
+            assertTableEquals(expected, y);
+
+            // now do the same thing with an extra column that is not stateless (so we certainly parallelize)
+            slow_a.reset(0, -1);
+            System.out.println(Instant.now() + ": serial + const");
+            final Table y2 = x.update(List.of(sa.withSerial(), sb.withSerial(), sc));
+            assertTableEquals(expected.update("C=1"), y2);
+
+            // now turn off the implicit barriers
+            QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS = false;
+            ((QueryTable) x).clearMemoizedResults();
+            slow_a.reset(0, 1);
+            slow_a.setMinThreads(1);
+            final Table y3 = x.update(List.of(sa.withSerial(), sb.withSerial(), sc));
+            checkTotalSum(size, y3);
+            checkMixedColumns(y3, true);
+            assertFalse(isOutOfOrder(y3, "A"));
+            assertFalse(isOutOfOrder(y3, "B"));
+            System.out.println(Instant.now() + ": no barrier");
+        }
+    }
+
     // if you would like to convince yourself that we are reliably producing out-of-order conditions as appropriate
     @Test
     public void testRepeatedBarrierAcrossShift() {
@@ -254,6 +328,7 @@ private void testBarrierAcrossShift() {
                 final SafeCloseable ignored3 = threadPool::shutdown) {
             QueryTable.FORCE_PARALLEL_SELECT_AND_UPDATE = true;
             QueryTable.STATELESS_SELECT_BY_DEFAULT = false;
+            QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS = false;
 
             final SelectColumn sa =
                     SelectColumnFactory.getExpression("A=slow_a.getAndIncrementSlow(0, new int[] { 0, 1, 0, 1 })");
@@ -363,6 +438,7 @@ private void testBarrierAliases() {
                 final SafeCloseable ignored3 = threadPool::shutdown) {
             QueryTable.FORCE_PARALLEL_SELECT_AND_UPDATE = true;
             QueryTable.STATELESS_SELECT_BY_DEFAULT = false;
+            QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS = false;
 
             final SelectColumn sa = SelectColumnFactory
                     .getExpression("A=slow_a.getAndIncrementSlow(0, new int[] { 0, 1 })");
@@ -654,11 +730,13 @@ private static boolean isOutOfOrder(Table z, final String column) {
     private static class SaveQueryTableOptions implements SafeCloseable {
         final boolean oldForceParallel = QueryTable.FORCE_PARALLEL_SELECT_AND_UPDATE;
         final boolean oldStateless = QueryTable.STATELESS_SELECT_BY_DEFAULT;
+        final boolean oldSerialBarriers = QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS;
 
         @Override
         public void close() {
             QueryTable.FORCE_PARALLEL_SELECT_AND_UPDATE = oldForceParallel;
             QueryTable.STATELESS_SELECT_BY_DEFAULT = oldStateless;
+            QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS = oldSerialBarriers;
         }
     }
 
@@ -680,6 +758,7 @@ public static class SlowedAtomicInteger {
         private final List<Set<Thread>> columnThreads;
         int minChecks;
         int maxChecks;
+        int minThreads = 2;
 
         /**
          *
@@ -702,7 +781,7 @@ public static class SlowedAtomicInteger {
         /**
          * Do the get and increment on the underlying atomic integer, possibly sleeping to cause data to be interspersed
          * if there are not declared barriers.
-         * 
+         *
          * @param columnNumber the column number we are updating
          * @param checkThreads an array of column numbers that we should verify had at least two threads access them. -1
          *        indicates that we are not checking this position. The {@link #reset(int, int)} call enables us to
@@ -733,7 +812,7 @@ private void mabyeWaitABit(int columnNumber, int[] checkThreads) {
                 if (checkThreads[ii] < 0) {
                     continue;
                 }
-                sleepRequired = sleepRequired || columnThreads.get(checkThread).size() < 2;
+                sleepRequired = sleepRequired || columnThreads.get(checkThread).size() < minThreads;
             }
 
             if (sleepRequired) {
@@ -751,6 +830,10 @@ void reset(final int minChecks, final int maxChecks) {
             this.maxChecks = maxChecks;
             columnThreads.forEach(Set::clear);
         }
+
+        void setMinThreads(int minThreads) {
+            this.minThreads = minThreads;
+        }
     }
 }
 

table-api/src/main/java/io/deephaven/api/ConcurrencyControl.java

@@ -37,9 +37,24 @@ public interface ConcurrencyControl<T> {
      * parts of a filter are executed out of order relative to this serial wrapper.
      * </p>
      * <p>
-     * For selectables, no additional ordering between expressions is imposed. As with every select or update call, if
-     * column B references column A, then the necessary inputs from column A are evaluated before column B is evaluated.
-     * To impose further ordering constraints, use barriers.
+     * For selectables, additional ordering constraints are controlled by the value of the
+     * {@code QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS}. This is set by the property
+     * {@code QueryTable.serialSelectImplicitBarriers} (defaulting to the value of
+     * {@code QueryTable.statelessSelectByDefault}).
+     * </p>
+     *
+     * <p>
+     * If {@code QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS} is false, then no additional ordering between selectable
+     * expressions is imposed. As with every select or update call, if column B references column A, then the necessary
+     * inputs from column A are evaluated before column B is evaluated. To impose further ordering constraints, use
+     * barriers.
+     * </p>
+     *
+     * <p>
+     * If {@code QueryTable.SERIAL_SELECT_IMPLICIT_BARRIERS} is true, then a serial selectable is an absolute barrier
+     * with respect to all other serial selectables. This prohibits serial selectables from being evaluated
+     * concurrently, permitting them to access global state. Selectables that are not serial may be reordered with
+     * respect to a serial selectable.
      * </p>
      * </li>
      * </ul>