Create a GrpcApiApplication That Exposes a Blink Table of Export Stat…

…e Changes
nbauernfeind · Sep 5, 2023 · dc4299e · dc4299e
1 parent d3b2147
commit dc4299e
Show file tree

Hide file tree

Showing 25 changed files with 378 additions and 72 deletions.
diff --git a/application-mode/build.gradle b/application-mode/build.gradle
@@ -4,7 +4,9 @@ plugins {
 }
 
 dependencies {
+    implementation project(':Base')
     implementation project(':Integrations')
+    implementation project(':engine-updategraph')
 
     Classpaths.inheritImmutables(project)
 

diff --git a/application-mode/src/main/java/io/deephaven/appmode/ApplicationState.java b/application-mode/src/main/java/io/deephaven/appmode/ApplicationState.java
@@ -11,7 +11,11 @@
 import java.util.Objects;
 import java.util.ServiceLoader;
 
-public class ApplicationState {
+import io.deephaven.engine.liveness.LivenessReferent;
+import io.deephaven.engine.liveness.LivenessScope;
+import io.deephaven.engine.updategraph.DynamicNode;
+
+public class ApplicationState extends LivenessScope {
 
     public interface Factory {
 
@@ -75,12 +79,17 @@ public synchronized <T> void setField(String name, T value, String description)
     }
 
     public synchronized void setField(Field<?> field) {
-        Field<?> oldField = fields.remove(field.name());
-        if (oldField != null) {
-            listener.onRemoveField(this, field);
+        // manage the new value before release the old value
+        final Object newValue = field.value();
+        if ((newValue instanceof LivenessReferent) && DynamicNode.notDynamicOrIsRefreshing(newValue)) {
+            manage((LivenessReferent) newValue);
         }
-        listener.onNewField(this, field);
+
+        // remove and release the old value
+        removeField(field.name());
+
         fields.put(field.name(), field);
+        listener.onNewField(this, field);
     }
 
     public synchronized void setFields(Field<?>... fields) {
@@ -97,6 +106,10 @@ public synchronized void removeField(String name) {
         Field<?> field = fields.remove(name);
         if (field != null) {
             listener.onRemoveField(this, field);
+            Object oldValue = field.value();
+            if ((oldValue instanceof LivenessReferent) && DynamicNode.notDynamicOrIsRefreshing(oldValue)) {
+                unmanage((LivenessReferent) oldValue);
+            }
         }
     }
 

diff --git a/engine/updategraph/build.gradle b/engine/updategraph/build.gradle
@@ -6,6 +6,7 @@ plugins {
 description 'Engine Update Graph: Core utilities for maintaining a DAG for update processing'
 
 dependencies {
+    api project(':Util')
     api project(':qst')
 
     implementation project(':engine-chunk')

diff --git a/plugin/gc-app/src/main/java/io/deephaven/app/GcApplication.java b/plugin/gc-app/src/main/java/io/deephaven/app/GcApplication.java
@@ -4,12 +4,9 @@
 import io.deephaven.appmode.ApplicationState;
 import io.deephaven.appmode.ApplicationState.Listener;
 import io.deephaven.engine.context.ExecutionContext;
-import io.deephaven.engine.liveness.LivenessScope;
-import io.deephaven.engine.liveness.LivenessScopeStack;
 import io.deephaven.engine.table.Table;
 import io.deephaven.engine.table.impl.sources.ring.RingTableTools;
 import io.deephaven.stream.StreamToBlinkTableAdapter;
-import io.deephaven.util.SafeCloseable;
 
 import javax.management.ListenerNotFoundException;
 import javax.management.Notification;
@@ -46,14 +43,12 @@ public final class GcApplication implements ApplicationState.Factory, Notificati
     private static final String POOLS = "pools";
     private static final String POOLS_STATS = "pools_stats";
 
-    private static final String ENABLED = "io.deephaven.app.GcApplication.enabled";
-    private static final String NOTIFICATION_INFO_ENABLED = "io.deephaven.app.GcApplication.notification_info.enabled";
-    private static final String NOTIFICATION_INFO_STATS_ENABLED =
-            "io.deephaven.app.GcApplication.notification_info_stats.enabled";
-    private static final String NOTIFICATION_INFO_RING_ENABLED =
-            "io.deephaven.app.GcApplication.notification_info_ring.enabled";
-    private static final String POOLS_ENABLED = "io.deephaven.app.GcApplication.pools.enabled";
-    private static final String POOLS_STATS_ENABLED = "io.deephaven.app.GcApplication.pools_stats.enabled";
+    private static final String ENABLED = APP_ID + ".enabled";
+    private static final String NOTIFICATION_INFO_ENABLED = APP_ID + ".notification_info.enabled";
+    private static final String NOTIFICATION_INFO_STATS_ENABLED = APP_ID + ".notification_info_stats.enabled";
+    private static final String NOTIFICATION_INFO_RING_ENABLED = APP_ID + ".notification_info_ring.enabled";
+    private static final String POOLS_ENABLED = APP_ID + ".pools.enabled";
+    private static final String POOLS_STATS_ENABLED = APP_ID + ".pools_stats.enabled";
 
     /**
      * Looks up the system property {@value ENABLED}, defaults to {@code false}.
@@ -113,8 +108,6 @@ public static boolean poolStatsEnabled() {
 
     private GcNotificationPublisher notificationInfoPublisher;
     private GcPoolsPublisher poolsPublisher;
-    @SuppressWarnings("FieldCanBeLocal")
-    private LivenessScope scope;
 
     @Override
     public void handleNotification(Notification notification, Object handback) {
@@ -146,14 +139,11 @@ public ApplicationState create(Listener listener) {
         if (!notificationInfoEnabled() && !poolsEnabled()) {
             return state;
         }
-        scope = new LivenessScope();
-        try (final SafeCloseable ignored = LivenessScopeStack.open(scope, false)) {
-            if (notificationInfoEnabled) {
-                setNotificationInfo(state);
-            }
-            if (poolsEnabled) {
-                setPools(state);
-            }
+        if (notificationInfoEnabled) {
+            setNotificationInfo(state);
+        }
+        if (poolsEnabled) {
+            setPools(state);
         }
         install();
         return state;

diff --git a/server/src/main/java/io/deephaven/server/appmode/ApplicationInjector.java b/server/src/main/java/io/deephaven/server/appmode/ApplicationInjector.java
@@ -85,7 +85,7 @@ private void loadApplication(final Path applicationDir, final ApplicationConfig
         log.info().append("Starting application '").append(config.toString()).append('\'').endl();
 
         final ApplicationState app;
-        try (final SafeCloseable ignored = LivenessScopeStack.open();
+        try (final SafeCloseable ignored1 = LivenessScopeStack.open();
                 final SafeCloseable ignored2 = scriptSessionProvider.get().getExecutionContext().open()) {
             app = ApplicationFactory.create(applicationDir, config, scriptSessionProvider.get(), applicationListener);
         }

diff --git a/server/src/main/java/io/deephaven/server/appmode/ApplicationTicketResolver.java b/server/src/main/java/io/deephaven/server/appmode/ApplicationTicketResolver.java
@@ -84,7 +84,7 @@ private <T> SessionState.ExportObject<T> resolve(final AppFieldId id, final Stri
         }
         Object value = authorization.transform(field.value());
         // noinspection unchecked
-        return SessionState.wrapAsExport((T) value);
+        return SessionState.wrapAsExport((T) value, "ApplicationTicketResolver#resolve");
     }
 
     @Override
@@ -113,7 +113,7 @@ public SessionState.ExportObject<Flight.FlightInfo> flightInfoFor(
             }
         }
 
-        return SessionState.wrapAsExport(info);
+        return SessionState.wrapAsExport(info, "ApplicationTicketResolver#flightInfoFor");
     }
 
     @Override

diff --git a/server/src/main/java/io/deephaven/server/appmode/ApplicationsModule.java b/server/src/main/java/io/deephaven/server/appmode/ApplicationsModule.java
@@ -8,6 +8,7 @@
 import dagger.multibindings.IntoSet;
 import io.deephaven.app.GcApplication;
 import io.deephaven.appmode.ApplicationState;
+import io.deephaven.server.grpc_api_app.GrpcApiApplication;
 
 @Module
 public interface ApplicationsModule {
@@ -17,4 +18,10 @@ public interface ApplicationsModule {
     static ApplicationState.Factory providesGcApplication() {
         return new GcApplication();
     }
+
+    @Provides
+    @IntoSet
+    static ApplicationState.Factory providesGrpcApiApplication() {
+        return new GrpcApiApplication();
+    }
 }
diff --git a/server/src/main/java/io/deephaven/server/arrow/ArrowFlightUtil.java b/server/src/main/java/io/deephaven/server/arrow/ArrowFlightUtil.java
@@ -53,8 +53,6 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
 import java.util.*;
 
 import static io.deephaven.extensions.barrage.util.BarrageUtil.DEFAULT_SNAPSHOT_DESER_OPTIONS;
@@ -80,6 +78,7 @@ public static void DoGetCustom(
 
         final long queueStartTm = System.nanoTime();
         session.nonExport()
+                .description("FlightService#DoGet")
                 .require(export)
                 .onError(observer)
                 .submit(() -> {
@@ -154,6 +153,7 @@ public void onNext(final InputStream request) {
                     flightDescriptor = mi.descriptor;
                     resultExportBuilder = ticketRouter
                             .<Table>publish(session, mi.descriptor, "Flight.Descriptor", null)
+                            .description("ArrowFlight#DoPut")
                             .onError(observer);
                 }
             }
@@ -463,6 +463,7 @@ public void handleMessage(@NotNull final BarrageProtoUtil.MessageInfo message) {
 
                     final long queueStartTm = System.nanoTime();
                     session.nonExport()
+                            .description("FlightService#DoExchange(snapshot)")
                             .require(parent)
                             .onError(listener)
                             .submit(() -> {
@@ -563,6 +564,7 @@ public void handleMessage(@NotNull final MessageInfo message) {
 
                     synchronized (this) {
                         onExportResolvedContinuation = session.nonExport()
+                                .description("FlightService#DoExchange(subscription)")
                                 .require(parent)
                                 .onErrorHandler(DoExchangeMarshaller.this::onError)
                                 .submit(() -> onExportResolved(parent));

diff --git a/server/src/main/java/io/deephaven/server/arrow/FlightServiceGrpcImpl.java b/server/src/main/java/io/deephaven/server/arrow/FlightServiceGrpcImpl.java
@@ -175,6 +175,7 @@ public void getFlightInfo(
 
         if (session != null) {
             session.nonExport()
+                    .description("FlightService#getFlightInfo")
                     .require(export)
                     .onError(responseObserver)
                     .submit(() -> {
@@ -209,6 +210,7 @@ public void getSchema(
 
         if (session != null) {
             session.nonExport()
+                    .description("FlightService#getSchema")
                     .require(export)
                     .onError(responseObserver)
                     .submit(() -> {

diff --git a/server/src/main/java/io/deephaven/server/console/ConsoleServiceGrpcImpl.java b/server/src/main/java/io/deephaven/server/console/ConsoleServiceGrpcImpl.java
@@ -128,6 +128,7 @@ public void startConsole(
         }
 
         session.newExport(request.getResultId(), "resultId")
+                .description("ConsoleService#startConsole")
                 .onError(responseObserver)
                 .submit(() -> {
                     final ScriptSession scriptSession = new DelegatingScriptSession(scriptSessionProvider.get());
@@ -167,25 +168,33 @@ public void executeCommand(
         SessionState.ExportObject<ScriptSession> exportedConsole =
                 ticketRouter.resolve(session, consoleId, "consoleId");
         session.nonExport()
+                .description("ConsoleService#executeCommand")
                 .requiresSerialQueue()
                 .require(exportedConsole)
                 .onError(responseObserver)
                 .submit(() -> {
-                    ScriptSession scriptSession = exportedConsole.get();
-                    ScriptSession.Changes changes = scriptSession.evaluateScript(request.getCode());
-                    ExecuteCommandResponse.Builder diff = ExecuteCommandResponse.newBuilder();
-                    FieldsChangeUpdate.Builder fieldChanges = FieldsChangeUpdate.newBuilder();
-                    changes.created.entrySet()
-                            .forEach(entry -> fieldChanges.addCreated(makeVariableDefinition(entry)));
-                    changes.updated.entrySet()
-                            .forEach(entry -> fieldChanges.addUpdated(makeVariableDefinition(entry)));
-                    changes.removed.entrySet()
-                            .forEach(entry -> fieldChanges.addRemoved(makeVariableDefinition(entry)));
-                    if (changes.error != null) {
-                        diff.setErrorMessage(Throwables.getStackTraceAsString(changes.error));
-                        log.error().append("Error running script: ").append(changes.error).endl();
+                    // since we are on the serial queue, we can safely stash our session id someplace accessible
+                    SessionService.CURRENT_SESSION_ID = session.getSessionId();
+
+                    try {
+                        ScriptSession scriptSession = exportedConsole.get();
+                        ScriptSession.Changes changes = scriptSession.evaluateScript(request.getCode());
+                        ExecuteCommandResponse.Builder diff = ExecuteCommandResponse.newBuilder();
+                        FieldsChangeUpdate.Builder fieldChanges = FieldsChangeUpdate.newBuilder();
+                        changes.created.entrySet()
+                                .forEach(entry -> fieldChanges.addCreated(makeVariableDefinition(entry)));
+                        changes.updated.entrySet()
+                                .forEach(entry -> fieldChanges.addUpdated(makeVariableDefinition(entry)));
+                        changes.removed.entrySet()
+                                .forEach(entry -> fieldChanges.addRemoved(makeVariableDefinition(entry)));
+                        if (changes.error != null) {
+                            diff.setErrorMessage(Throwables.getStackTraceAsString(changes.error));
+                            log.error().append("Error running script: ").append(changes.error).endl();
+                        }
+                        safelyComplete(responseObserver, diff.setChanges(fieldChanges).build());
+                    } finally {
+                        SessionService.CURRENT_SESSION_ID = "NO SESSION IN USE";
                     }
-                    safelyComplete(responseObserver, diff.setChanges(fieldChanges).build());
                 });
     }
 
@@ -244,6 +253,7 @@ public void bindTableToVariable(
         final SessionState.ExportObject<ScriptSession> exportedConsole;
 
         ExportBuilder<?> exportBuilder = session.nonExport()
+                .description("ConsoleService#bindTableToVariable")
                 .requiresSerialQueue()
                 .onError(responseObserver);
 

diff --git a/server/src/main/java/io/deephaven/server/console/ScopeTicketResolver.java b/server/src/main/java/io/deephaven/server/console/ScopeTicketResolver.java
@@ -8,6 +8,7 @@
 import io.deephaven.base.string.EncodingInfo;
 import io.deephaven.engine.context.QueryScope;
 import io.deephaven.engine.liveness.LivenessReferent;
+import io.deephaven.engine.liveness.LivenessStateException;
 import io.deephaven.engine.table.Table;
 import io.deephaven.engine.updategraph.DynamicNode;
 import io.deephaven.engine.util.ScriptSession;
@@ -75,7 +76,7 @@ public SessionState.ExportObject<Flight.FlightInfo> flightInfoFor(
                             "Could not resolve '" + logId + "': no variable exists with name '" + scopeName + "'");
                 });
 
-        return SessionState.wrapAsExport(flightInfo);
+        return SessionState.wrapAsExport(flightInfo, "ScopeTicketResolver#flightInfoFor");
     }
 
     @Override
@@ -106,6 +107,7 @@ private <T> SessionState.ExportObject<T> resolve(
         // fetch the variable from the scope right now
         T export = gss.getExecutionContext().getUpdateGraph().sharedLock().computeLocked(() -> {
             T scopeVar = null;
+            // note we do not need to synchronize on gss here because it cannot change while holding the ugp lock
             try {
                 // noinspection unchecked
                 scopeVar = (T) gss.unwrapObject(gss.getVariable(scopeName));
@@ -119,9 +121,20 @@ private <T> SessionState.ExportObject<T> resolve(
         if (export == null) {
             return SessionState.wrapAsFailedExport(Exceptions.statusRuntimeException(Code.FAILED_PRECONDITION,
                     "Could not resolve '" + logId + "': no variable exists with name '" + scopeName + "'"));
+        } else if (export instanceof SessionState.ExportObject) {
+            // noinspection unchecked
+            final SessionState.ExportObject<T> asExportObject = (SessionState.ExportObject<T>) export;
+            // ensure the result is live until the caller uses it
+            try {
+                asExportObject.manageWithCurrentScope();
+                return asExportObject;
+            } catch (LivenessStateException ignored) {
+                return SessionState.wrapAsFailedExport(Exceptions.statusRuntimeException(Code.FAILED_PRECONDITION,
+                        "Could not resolve '" + logId + "': no variable exists with name '" + scopeName + "'"));
+            }
         }
 
-        return SessionState.wrapAsExport(export);
+        return SessionState.wrapAsExport(export, "ScopeTicketResolver#resolve");
     }
 
     @Override
@@ -153,11 +166,16 @@ private <T> SessionState.ExportBuilder<T> publish(
         final SessionState.ExportObject<T> resultExport = resultBuilder.getExport();
         final SessionState.ExportBuilder<T> publishTask = session.nonExport();
 
+        // if we receive requests to read from this variable before the client has finished publishing, we will
+        // give the user the result export object to wait on as a dependency.
+        final ScriptSession gss = scriptSessionProvider.get();
+        gss.setVariable(varName, resultExport);
+
         publishTask
+                .description("ScopeTicketResolver#publish(" + varName + ")")
                 .requiresSerialQueue()
                 .require(resultExport)
                 .submit(() -> {
-                    final ScriptSession gss = scriptSessionProvider.get();
                     T value = resultExport.get();
                     if (value instanceof LivenessReferent && DynamicNode.notDynamicOrIsRefreshing(value)) {
                         gss.manage((LivenessReferent) value);
@@ -168,6 +186,8 @@ private <T> SessionState.ExportBuilder<T> publish(
                     }
                 });
 
+        publishTask.setStateToPublishing();
+        resultBuilder.setStateToPublishing();
         return resultBuilder;
     }