Deliver progress information to the IDE

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

@@ -12,16 +12,15 @@ final class ExecutionProgressObserver implements Consumer<ObservedMessage> {
   private final UUID nodeId;
   private final Thread thread;
   private final AutoCloseable handle;
-  private final Consumer<Double> consumer;
+  private final ProgressAggregator aggregate;
 
   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
+    this.aggregate = new ProgressAggregator(c);
+    // start by notifying indeterminate computation
     c.accept(-1.0);
-    System.err.println("Observing for " + nodeId);
   }
 
   UUID nodeId() {
@@ -35,12 +34,29 @@ static ExecutionProgressObserver startComputation(UUID nodeId, Consumer<Double>
   @Override
   public void accept(ObservedMessage t) {
     if (Thread.currentThread() == thread) {
-      System.err.println("  seeing " + t.getMessage() + " for " + nodeId);
+      switch (t.getMessage()) {
+        case "INIT {}:{}@{}" -> {
+          if (t.getArguments().size() >= 3
+              && t.getArguments().get(1) instanceof String msg
+              && t.getArguments().get(2) instanceof Number max) {
+            var key = t.getArguments().get(0);
+            aggregate.create(key, max.longValue());
+          }
+        }
+        case "ADVANCE {}+{}" -> {
+          if (t.getArguments().size() >= 2 && t.getArguments().get(1) instanceof Number by) {
+            var key = t.getArguments().get(0);
+            aggregate.advanceBy(key, by.longValue());
+          }
+        }
+        default -> {
+          System.err.println("  seeing " + t.getMessage() + " for " + nodeId);
+        }
+      }
     }
   }
 
   final void finishComputation() {
-    System.err.println("Finished computing " + nodeId);
     try {
       handle.close();
     } catch (Exception ex) {

engine/runtime-integration-tests/src/test/java/org/enso/interpreter/runtime/progress/ProgressAggregator.java → engine/runtime-instrument-common/src/main/java/org/enso/interpreter/service/ProgressAggregator.java

@@ -1,14 +1,21 @@
-package org.enso.interpreter.runtime.progress;
+package org.enso.interpreter.service;
 
 import java.util.ArrayDeque;
 import java.util.Deque;
+import java.util.Map;
+import java.util.WeakHashMap;
 import java.util.function.Consumer;
 
 /**
  * Example of a progress aggregator able to nest multiple Progress and compute % of aggregated
  * progress.
  */
 public final class ProgressAggregator {
+  /**
+   * @GuardedBy("this")
+   */
+  private final Map<Object, Progress> map = new WeakHashMap<>();
+
   private final Consumer<Double> updateStatus;
   private final Deque<Progress> stack = new ArrayDeque<>();
   private double current;
@@ -24,13 +31,12 @@ public ProgressAggregator(Consumer<Double> updateStatus) {
   }
 
   /**
-   * Starts new progress in the current aggregator's stack.Nests the current progress in the
+   * Starts new progress in the current aggregator's stack. Nests the current progress in the
    * currently executing step of current progress.
    *
    * @param max maximum number of steps the progress can "advance to"
-   * @return new instance of a progress to {@link Progress#advance}
    */
-  public synchronized Progress create(long max) {
+  public synchronized void create(Object key, long max) {
     Progress p;
     if (stack.isEmpty()) {
       p = new Progress(max, 0.0, 1.0);
@@ -40,11 +46,24 @@ public synchronized Progress create(long max) {
       p = new Progress(max, current, current + previous.singleStep());
     }
     stack.addFirst(p);
-    return p;
+    map.put(key, p);
+  }
+
+  public void closeProgress(Object key) {
+    if (findBy(key) instanceof Progress p) {
+      p.advance(p.max);
+      stack.remove(p);
+    }
+  }
+
+  public void advanceBy(Object key, long steps) {
+    if (findBy(key) instanceof Progress p) {
+      p.advance(steps);
+    }
   }
 
-  private synchronized void closeProgress(Progress p) {
-    stack.remove(p);
+  private synchronized Progress findBy(Object key) {
+    return map.get(key);
   }
 
   private void advanceTo(double now) {

engine/runtime-integration-tests/src/test/java/org/enso/interpreter/runtime/progress/ProgressAggregatorTest.java → engine/runtime-instrument-common/src/test/java/org/enso/interpreter/service/ProgressAggregatorTest.java

@@ -1,4 +1,4 @@
-package org.enso.interpreter.runtime.progress;
+package org.enso.interpreter.service;
 
 import static org.junit.Assert.assertEquals;
 
@@ -13,7 +13,7 @@ public ProgressAggregatorTest() {}
   @Test
   public void threeSteps() {
     var agg = new ProgressAggregator(this);
-    var three = agg.create(3);
+    var three = create(agg, 3);
     three.advance(1);
     assertEquals("1/3", 0.333, current, 0.001);
     three.advance(1);
@@ -25,7 +25,7 @@ public void threeSteps() {
   @Test
   public void closeFinishesItAll() {
     var agg = new ProgressAggregator(this);
-    var ten = agg.create(10);
+    var ten = create(agg, 10);
     ten.advance(1);
     assertEquals("1/10", 0.1, current, 0.001);
     ten.close();
@@ -35,8 +35,8 @@ public void closeFinishesItAll() {
   @Test
   public void halfAndThird() {
     var agg = new ProgressAggregator(this);
-    var half = agg.create(2);
-    var firstThird = agg.create(3);
+    var half = create(agg, 2);
+    var firstThird = create(agg, 3);
     firstThird.advance(1);
     assertEquals("1/6", 1.0 / 6.0, current, 0.001);
     firstThird.advance(1);
@@ -46,7 +46,7 @@ public void halfAndThird() {
     half.advance(1);
     assertEquals("Topmost 1/2", 0.5, current, 0.001);
 
-    var secondThird = agg.create(3);
+    var secondThird = create(agg, 3);
     secondThird.advance(2);
     assertEquals("5/6", 5.0 / 6.0, current, 0.001);
 
@@ -59,4 +59,26 @@ public void halfAndThird() {
   public void accept(Double t) {
     this.current = t;
   }
+
+  private Handle create(ProgressAggregator agg, long max) {
+    var h = new Handle(agg);
+    agg.create(h, max);
+    return h;
+  }
+
+  private final class Handle {
+    private final ProgressAggregator agg;
+
+    Handle(ProgressAggregator agg) {
+      this.agg = agg;
+    }
+
+    final void advance(long n) {
+      agg.advanceBy(this, n);
+    }
+
+    final void close() {
+      agg.closeProgress(this);
+    }
+  }
 }