Be told how a statement is getting on, and stop it

README.md

@@ -150,6 +150,21 @@ A million rows, three columns, the same laptop:
 
 The first two lines are the whole point. Half a millisecond to make a million rows queryable against six hundred to write them down, and the three lines after that say the query is not paying for it afterwards.
 
+## Watching a statement, and stopping it
+
+Three ways, and they are the same mechanism seen from different sides. `conn.interrupt()` is safe from another thread and stops whatever is running. `conn.rowsRead()` is safe from another thread too and says how far it has got. And a progress callback is both of those without the thread that would otherwise have to do the polling:
+
+```java
+long deadline = 30_000;
+conn.onProgress(Duration.ofMillis(250), (rows, millis) -> millis < deadline);
+```
+
+Answering false stops the statement exactly as `interrupt()` would, so a timeout is a one line watcher and a progress bar is the same watcher with a repaint in it. The arrangement belongs to the connection and covers every statement after it, so it is set once when the connection is opened rather than around each query.
+
+The callback runs on a thread of the library's, one per statement, never two at once and never after the statement it belongs to has returned. Two things follow. Whatever it touches has to be usable from another thread, so a counter a progress bar reads should be an `AtomicLong` rather than a field. And it must not call back into the library on the connection it is reporting on, because that connection is inside the executor.
+
+This is the other of the two places a pointer to Java code goes the other way. An exception crossing an upcall would take the JVM down, so a watcher that throws is logged and answered as though it had asked for the statement to stop, which is the reading that loses least: a callback that threw is a program that has stopped wanting the answer.
+
 ## How it binds
 
 The Foreign Function and Memory API is the primary path. The downcall handles are written by hand against `zu.h` rather than generated with `jextract`, because the C ABI here is around seventy functions with a stable shape, and a hand-written layer is where the interesting decisions live: which calls are `Linker.Option.critical` because they are short pure accessors, where the out-parameter scratch space comes from so that a query does not allocate, and how a `zu_error` becomes a typed Java exception exactly once. There is no native code in this repository beyond `libzu` itself.

zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java

@@ -68,6 +68,7 @@ final class Abi {
   final MethodHandle connClose;
   final MethodHandle connInterrupt;
   final MethodHandle connRowsRead;
+  final MethodHandle connSetProgress;
   final MethodHandle connInTransaction;
   final MethodHandle begin;
   final MethodHandle commit;
@@ -201,6 +202,8 @@ final class Abi {
     connClose = h("zu_conn_close", FunctionDescriptor.ofVoid(ADDRESS));
     connInterrupt = h("zu_conn_interrupt", FunctionDescriptor.of(JAVA_INT, ADDRESS));
     connRowsRead = h("zu_conn_rows_read", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS));
+    connSetProgress = h("zu_conn_set_progress", FunctionDescriptor.of(
+        JAVA_INT, ADDRESS, ADDRESS, ADDRESS, JAVA_LONG));
     connInTransaction = h("zu_conn_in_transaction", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS));
     begin = h("zu_begin", FunctionDescriptor.of(JAVA_INT, ADDRESS, JAVA_INT, ADDRESS));
     commit = h("zu_commit", FunctionDescriptor.of(JAVA_INT, ADDRESS, ADDRESS));

zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java

@@ -46,10 +46,27 @@ final class FfmBinding implements ZuBinding {
 
   private final Abi abi;
 
+  /**
+   * The progress stub each connection is watching through, which is the one
+   * thing this binding has to remember about a connection.
+   *
+   * <p>It lives here rather than on the {@code Connection} because the stub is
+   * an FFM type and the API module never names one. Keyed by the handle, which
+   * is safe because the entry goes when the connection closes and not a moment
+   * later, so an address the allocator hands out again cannot find an old one.
+   */
+  private final java.util.Map<Long, Watch> watches = new java.util.concurrent.ConcurrentHashMap<>();
+
   FfmBinding(Abi abi) {
     this.abi = abi;
   }
 
+  private static void spend(Watch watch) {
+    if (watch != null) {
+      watch.spend();
+    }
+  }
+
   @Override
   public String version() {
     try {
@@ -139,6 +156,10 @@ public void connClose(long conn) {
       abi.connClose.invokeExact(ptr(conn));
     } catch (Throwable t) {
       throw fail("zu_conn_close", t);
+    } finally {
+      // The connection is gone, so nothing will call its progress stub again,
+      // and a later connection could be handed this very address.
+      spend(watches.remove(conn));
     }
   }
 
@@ -164,6 +185,44 @@ public long connRowsRead(long conn) {
     }
   }
 
+  @Override
+  public void connSetProgress(long conn, dev.zudb.Progress watcher, long intervalMillis) {
+    if (watcher == null) {
+      try {
+        int st =
+            (int)
+                abi.connSetProgress.invokeExact(
+                    ptr(conn), MemorySegment.NULL, MemorySegment.NULL, 0L);
+        check("zu_conn_set_progress", st, null);
+      } catch (Throwable t) {
+        throw fail("zu_conn_set_progress", t);
+      }
+      spend(watches.remove(conn));
+      return;
+    }
+    Watch watch = Watch.of(watcher);
+    boolean set = false;
+    try {
+      int st =
+          (int)
+              abi.connSetProgress.invokeExact(
+                  ptr(conn), watch.stub(), MemorySegment.NULL, intervalMillis);
+      check("zu_conn_set_progress", st, null);
+      set = true;
+    } catch (Throwable t) {
+      throw fail("zu_conn_set_progress", t);
+    } finally {
+      if (!set) {
+        watch.spend();
+      }
+    }
+    // Only now, because until the call returned the old arrangement was still
+    // the live one. The connection is not running a statement while it is in
+    // this call, so the callback the previous stub belongs to is not in flight
+    // either.
+    spend(watches.put(conn, watch));
+  }
+
   @Override
   public boolean connInTransaction(long conn) {
     Scratch s = Scratch.get();

zudb-ffm/src/main/java/dev/zudb/ffm/Release.java

@@ -4,33 +4,19 @@
 
 import java.lang.System.Logger;
 import java.lang.System.Logger.Level;
-import java.lang.foreign.Arena;
 import java.lang.foreign.FunctionDescriptor;
-import java.lang.foreign.Linker;
 import java.lang.foreign.MemorySegment;
 import java.lang.invoke.MethodHandle;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.MethodType;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
 
 /**
- * The one place this binding hands the engine a pointer to Java code rather
- * than the other way round.
+ * What a frame's release callback is on this side.
  *
- * <p>A frame's release callback is how a host learns the engine has finished
- * with the buffers it lent: it runs once, on a thread of the library's, after
- * the last statement reading the frame ends. Reaching Java from there needs an
- * upcall stub, which is executable memory with a lifetime of its own.
- *
- * <p>That lifetime is the awkward part. The stub has to outlive the frame,
- * because the callback is the last thing to happen and may happen after the
- * frame was freed, so it cannot hang off the frame. It cannot free itself
- * either: closing the arena a stub lives in from inside a call through that
- * same stub is closing the ground you are standing on. So a spent arena goes
- * on a queue and the next stub to be made closes it, which costs nothing, needs
- * no thread of ours, and bounds the outstanding stubs at the number of frames
- * whose callbacks have not run yet.
+ * <p>It is how a host learns the engine has finished with the buffers it lent:
+ * it runs once, on a thread of the library's, after the last statement reading
+ * the frame ends, which is neither the unregister that preceded it nor the
+ * free.
  */
 final class Release {
 
@@ -40,40 +26,23 @@ final class Release {
 
   private static final MethodHandle RUN = run();
 
-  /** Arenas whose callback has been and gone, waiting for somebody else to close them. */
-  private static final Queue<Arena> SPENT = new ConcurrentLinkedQueue<>();
-
   private final Runnable body;
-  private final Arena arena;
+  private volatile Upcall upcall;
 
-  private Release(Runnable body, Arena arena) {
+  private Release(Runnable body) {
     this.body = body;
-    this.arena = arena;
   }
 
-  private MemorySegment stub;
-
   /**
    * A {@code void (*)(void *)} the engine can call, bound to this runnable.
    *
    * @param body what to run when the engine is finished with the frame
-   * @return the release, which frees itself by way of the queue above
+   * @return the release
    */
-  @SuppressWarnings("restricted")
   static Release of(Runnable body) {
-    sweep();
-    // Shared rather than confined: the callback arrives on a thread of the
-    // library's, and a confined arena would refuse the call it is there to
-    // serve.
-    Arena arena = Arena.ofShared();
-    try {
-      Release release = new Release(body, arena);
-      release.stub = Linker.nativeLinker().upcallStub(RUN.bindTo(release), DESCRIPTOR, arena);
-      return release;
-    } catch (RuntimeException | Error e) {
-      arena.close();
-      throw e;
-    }
+    Release release = new Release(body);
+    release.upcall = Upcall.of(RUN.bindTo(release), DESCRIPTOR);
+    return release;
   }
 
   /**
@@ -82,15 +51,15 @@ static Release of(Runnable body) {
    * @return the stub
    */
   MemorySegment stub() {
-    return stub;
+    return upcall.stub();
   }
 
   /**
    * Says the callback will never be called, for the frame that failed to be
    * made at all, so that the stub goes the way a spent one goes.
    */
   void abandon() {
-    SPENT.add(arena);
+    upcall.spend();
   }
 
   /**
@@ -112,21 +81,7 @@ private void run(MemorySegment owner) {
     } catch (Throwable t) {
       LOG.log(Level.ERROR, "a zu frame release callback threw, which nothing above it can see", t);
     } finally {
-      SPENT.add(arena);
-    }
-  }
-
-  /** Closes the arenas of every callback that has already run. */
-  private static void sweep() {
-    for (Arena arena = SPENT.poll(); arena != null; arena = SPENT.poll()) {
-      try {
-        arena.close();
-      } catch (RuntimeException e) {
-        // A shared arena refuses to close while a thread is still inside a
-        // call into it. Put it back and let the next frame try.
-        SPENT.add(arena);
-        return;
-      }
+      upcall.spend();
     }
   }
 

zudb-ffm/src/main/java/dev/zudb/ffm/Upcall.java

@@ -0,0 +1,88 @@
+package dev.zudb.ffm;
+
+import java.lang.foreign.Arena;
+import java.lang.foreign.FunctionDescriptor;
+import java.lang.foreign.Linker;
+import java.lang.foreign.MemorySegment;
+import java.lang.invoke.MethodHandle;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * A pointer to Java code the engine can call, which is the direction
+ * everything else in this binding does not go.
+ *
+ * <p>An upcall stub is executable memory with a lifetime of its own, and that
+ * lifetime is the awkward part. A stub has to outlive whatever holds it,
+ * because the callback is the last thing to happen and may happen after the
+ * thing that arranged it has gone. It cannot free itself either: closing the
+ * arena a stub lives in from inside a call through that same stub is closing
+ * the ground you are standing on.
+ *
+ * <p>So a spent arena goes on a queue and the next stub to be made closes it.
+ * That costs nothing, needs no thread of ours, and bounds the outstanding
+ * stubs at the number of callbacks that have not been spent yet.
+ */
+final class Upcall {
+
+  /** Arenas nothing will call again, waiting for somebody else to close them. */
+  private static final Queue<Arena> SPENT = new ConcurrentLinkedQueue<>();
+
+  private final Arena arena;
+  private final MemorySegment stub;
+
+  private Upcall(Arena arena, MemorySegment stub) {
+    this.arena = arena;
+    this.stub = stub;
+  }
+
+  /**
+   * Binds a method handle as a function pointer.
+   *
+   * @param target what to call, already bound to whatever it is called on
+   * @param descriptor the C signature
+   * @return the stub, which is freed by way of {@link #spend()} and the queue
+   */
+  @SuppressWarnings("restricted")
+  static Upcall of(MethodHandle target, FunctionDescriptor descriptor) {
+    sweep();
+    // Shared rather than confined: the callback arrives on a thread of the
+    // library's, and a confined arena would refuse the call it is there to
+    // serve.
+    Arena arena = Arena.ofShared();
+    try {
+      return new Upcall(arena, Linker.nativeLinker().upcallStub(target, descriptor, arena));
+    } catch (RuntimeException | Error e) {
+      arena.close();
+      throw e;
+    }
+  }
+
+  /**
+   * The function pointer.
+   *
+   * @return the stub
+   */
+  MemorySegment stub() {
+    return stub;
+  }
+
+  /** Says nothing will call this again, so that the memory can go back. */
+  void spend() {
+    SPENT.add(arena);
+  }
+
+  /** Closes the arenas of every stub already spent. */
+  private static void sweep() {
+    for (Arena arena = SPENT.poll(); arena != null; arena = SPENT.poll()) {
+      try {
+        arena.close();
+      } catch (RuntimeException e) {
+        // A shared arena refuses to close while a thread is still inside a
+        // call into it. Put it back and let the next one try.
+        SPENT.add(arena);
+        return;
+      }
+    }
+  }
+}