Barrage Client: fix race conditions in subscription and static-table …

…logic (deephaven#4630) Co-authored-by: Ryan Caudy <[email protected]>
kosak · Oct 17, 2023 · cbaf19b · cbaf19b
1 parent 186891c
commit cbaf19b
Show file tree

Hide file tree

Showing 14 changed files with 524 additions and 376 deletions.
diff --git a/...nsions/barrage/src/main/java/io/deephaven/extensions/barrage/table/BarrageBlinkTable.java b/...nsions/barrage/src/main/java/io/deephaven/extensions/barrage/table/BarrageBlinkTable.java
@@ -48,8 +48,8 @@ protected BarrageBlinkTable(
             final LinkedHashMap<String, ColumnSource<?>> columns,
             final WritableColumnSource<?>[] writableSources,
             final Map<String, Object> attributes,
-            final long initialViewPortRows) {
-        super(registrar, notificationQueue, executorService, columns, writableSources, attributes, initialViewPortRows);
+            @Nullable final ViewportChangedCallback vpCallback) {
+        super(registrar, notificationQueue, executorService, columns, writableSources, attributes, vpCallback);
     }
 
     private void processUpdate(final BarrageMessage update) {
@@ -73,9 +73,7 @@ private void processUpdate(final BarrageMessage update) {
         }
 
         if (update.isSnapshot) {
-            serverViewport = update.snapshotRowSet == null ? null : update.snapshotRowSet.copy();
-            serverReverseViewport = update.snapshotRowSetIsReversed;
-            serverColumns = update.snapshotColumns == null ? null : (BitSet) update.snapshotColumns.clone();
+            updateServerViewport(update.snapshotRowSet, update.snapshotColumns, update.snapshotRowSetIsReversed);
         }
 
         if (update.shifted.nonempty()) {
@@ -127,10 +125,6 @@ private void processUpdate(final BarrageMessage update) {
         }
     }
 
-    private boolean isSubscribedColumn(int i) {
-        return serverColumns == null || serverColumns.get(i);
-    }
-
     private void ensureCapacity(long size) {
         if (capacity < size) {
             capacity = Long.highestOneBit(Math.max(size * 2, 8));

diff --git a/...s/barrage/src/main/java/io/deephaven/extensions/barrage/table/BarrageRedirectedTable.java b/...s/barrage/src/main/java/io/deephaven/extensions/barrage/table/BarrageRedirectedTable.java
@@ -55,8 +55,8 @@ protected BarrageRedirectedTable(final UpdateSourceRegistrar registrar,
             final WritableColumnSource<?>[] writableSources,
             final WritableRowRedirection rowRedirection,
             final Map<String, Object> attributes,
-            final long initialViewPortRows) {
-        super(registrar, notificationQueue, executorService, columns, writableSources, attributes, initialViewPortRows);
+            @Nullable final ViewportChangedCallback vpCallback) {
+        super(registrar, notificationQueue, executorService, columns, writableSources, attributes, vpCallback);
         this.rowRedirection = rowRedirection;
     }
 
@@ -90,16 +90,17 @@ private UpdateCoalescer processUpdate(final BarrageMessage update, final UpdateC
         }
 
         if (update.isSnapshot) {
-            serverViewport = update.snapshotRowSet == null ? null : update.snapshotRowSet.copy();
-            serverReverseViewport = update.snapshotRowSetIsReversed;
-            serverColumns = update.snapshotColumns == null ? null : (BitSet) update.snapshotColumns.clone();
+            updateServerViewport(update.snapshotRowSet, update.snapshotColumns, update.snapshotRowSetIsReversed);
         }
 
         // make sure that these RowSet updates make some sense compared with each other, and our current view of the
         // table
         final WritableRowSet currentRowSet = getRowSet().writableCast();
         final boolean mightBeInitialSnapshot = currentRowSet.isEmpty() && update.isSnapshot;
 
+        final RowSet serverViewport = getServerViewport();
+        final boolean serverReverseViewport = getServerReverseViewport();
+
         try (final RowSet currRowsFromPrev = currentRowSet.copy();
                 final WritableRowSet populatedRows =
                         (serverViewport != null
@@ -249,10 +250,6 @@ private UpdateCoalescer processUpdate(final BarrageMessage update, final UpdateC
         }
     }
 
-    private boolean isSubscribedColumn(int i) {
-        return serverColumns == null || serverColumns.get(i);
-    }
-
     private RowSet getFreeRows(long size) {
         if (size <= 0) {
             return RowSetFactory.empty();