New observer for each binding expression

Author: JaroslavTulach

engine/runtime-instrument-common/src/main/java/org/enso/interpreter/service/ExecutionCallbacks.java

@@ -6,6 +6,7 @@
 import java.util.Map;
 import java.util.UUID;
 import java.util.function.Consumer;
+import org.enso.common.CachePreferences;
 import org.enso.interpreter.instrument.ExpressionExecutionState;
 import org.enso.interpreter.instrument.MethodCallsCache;
 import org.enso.interpreter.instrument.OneshotExpression;
@@ -39,7 +40,7 @@ final class ExecutionCallbacks implements IdExecutionService.Callbacks {
   private final Consumer<ExpressionValue> onComputedCallback;
   private final Consumer<ExpressionCall> functionCallCallback;
   private final Consumer<ExecutedVisualization> onExecutedVisualizationCallback;
-  private final ExecutionProgressObserver progressObserver = new ExecutionProgressObserver();
+  private ExecutionProgressObserver progressObserver;
 
   /**
    * Creates callbacks instance.
@@ -94,22 +95,37 @@ public Object findCachedResult(IdExecutionService.Info info) {
       callOnCachedCallback(nodeId, result);
       return result;
     } else {
-      callOnNotCachedCallback(nodeId, -1.0);
-      progressObserver.startComputation(
-          nodeId,
-          (progress) -> {
-            callOnNotCachedCallback(nodeId, progress);
-          });
+      if (cache.getPreferences().get(nodeId) == CachePreferences.Kind.BINDING_EXPRESSION) {
+        refreshObserver(
+            ExecutionProgressObserver.startComputation(
+                nodeId,
+                (progress) -> {
+                  callOnNotCachedCallback(nodeId, progress);
+                }));
+      }
     }
 
     return null;
   }
 
+  private void refreshObserver(ExecutionProgressObserver newObserverOrNull) {
+    var o = progressObserver;
+    if (o != null) {
+      o.finishComputation();
+    }
+    this.progressObserver = newObserverOrNull;
+  }
+
   @Override
   public void updateCachedResult(IdExecutionService.Info info) {
     Object result = info.getResult();
     TypeInfo resultType = typeOf(result);
     UUID nodeId = info.getId();
+
+    if (progressObserver instanceof ExecutionProgressObserver o && nodeId.equals(o.nodeId())) {
+      refreshObserver(null);
+    }
+
     TypeInfo cachedType = cache.getType(nodeId);
     FunctionCallInfo call = functionCallInfoById(nodeId);
     FunctionCallInfo cachedCall = cache.getCall(nodeId);
@@ -187,9 +203,7 @@ private void callOnCachedCallback(UUID nodeId, Object result) {
 
   @CompilerDirectives.TruffleBoundary
   private void callOnNotCachedCallback(UUID nodeId, double amount) {
-    ExpressionValue expressionValue =
-        new ExpressionValue(nodeId, null, null, null, null, null, null, false);
-
+    var expressionValue = ExpressionValue.progress(nodeId, amount);
     onCachedCallback.accept(expressionValue);
   }
 

engine/runtime-instrument-common/src/main/java/org/enso/interpreter/service/ExecutionProgressObserver.java

@@ -9,14 +9,47 @@
 final class ExecutionProgressObserver implements Consumer<ObservedMessage> {
   private static final Logger PROGRESS = LoggerFactory.getLogger("Standard.Base.Logging.Progress");
 
-  private AutoCloseable handle = ObservedMessage.observe(PROGRESS, this);
+  private final UUID nodeId;
+  private final Thread thread;
+  private final AutoCloseable handle;
+  private final Consumer<Double> consumer;
 
-  ExecutionProgressObserver() {}
+  ExecutionProgressObserver(UUID nodeId, Consumer<Double> c) {
+    this.nodeId = nodeId;
+    this.handle = ObservedMessage.observe(PROGRESS, this);
+    this.thread = Thread.currentThread();
+    this.consumer = c;
+    // indeterminate computation has just started
+    c.accept(-1.0);
+    System.err.println("Observing for " + nodeId);
+  }
+
+  UUID nodeId() {
+    return nodeId;
+  }
 
-  final void startComputation(UUID nodeId, Consumer<Double> c) {}
+  static ExecutionProgressObserver startComputation(UUID nodeId, Consumer<Double> c) {
+    return new ExecutionProgressObserver(nodeId, c);
+  }
 
   @Override
   public void accept(ObservedMessage t) {
-    System.err.println("seeing " + t.getMessage());
+    if (Thread.currentThread() == thread) {
+      System.err.println("  seeing " + t.getMessage() + " for " + nodeId);
+    }
+  }
+
+  final void finishComputation() {
+    System.err.println("Finished computing " + nodeId);
+    try {
+      handle.close();
+    } catch (Exception ex) {
+      throw raise(RuntimeException.class, ex);
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static <T extends Exception> T raise(Class<T> aClass, Exception ex) throws T {
+    throw (T) ex;
   }
 }

engine/runtime-instrument-common/src/main/java/org/enso/interpreter/service/ExecutionService.java

@@ -704,6 +704,14 @@ public ExpressionValue(
       this.wasCached = wasCached;
     }
 
+    static ExpressionValue progress(UUID nodeId, double amount) {
+      return new ExpressionValue(nodeId, amount, null, null, null, null, null, false);
+    }
+
+    public boolean isProgressUpdate() {
+      return profilingInfo == null;
+    }
+
     @Override
     public String toString() {
       String profilingInfo = Arrays.toString(this.profilingInfo);

engine/runtime-instrument-common/src/main/scala/org/enso/interpreter/instrument/job/ProgramExecutionSupport.scala

@@ -447,10 +447,10 @@ object ProgramExecutionSupport {
     value: ExpressionValue
   )(implicit ctx: RuntimeContext): Unit = {
     val expressionId = value.getExpressionId
-    if (value.getValue == null && value.getProfilingInfo() == null) {
+    if (value.isProgressUpdate()) {
       val p = Api.ExpressionUpdate.Payload.Pending(
         None,
-        Some(-1.0)
+        Some(value.getValue().asInstanceOf[Double])
       )
       ctx.endpoint.sendToClient(
         Api.Response(