From b3ba4c63fd98cd0ecba12ea361cb77fb2e6eea17 Mon Sep 17 00:00:00 2001 From: Tam Nguyen Duc <1218621+tamnd@users.noreply.github.com> Date: Thu, 20 Aug 2026 09:07:53 +0700 Subject: [PATCH] Be told how a statement is getting on, and stop it A progress callback is the connection's interrupt and its rows read together, without the thread that would otherwise have to do the polling. It is set once on a connection and covers every statement after it, and a watcher answering false stops the statement exactly as an interrupt would, so a timeout is one line and a progress bar is the same line with a repaint in it. The stub the engine calls is the same machinery the frame release callback needed, so that machinery moved out into Upcall: an arena and a function pointer, spent onto a queue that the next stub to be made drains, because a stub cannot close the arena it is standing in. The frame release is now one user of that and the progress watcher the other. A progress stub outlives the call that set it, so the provider holds one per connection and frees it when the arrangement is replaced, taken back, or the connection closes. Freeing it at those three moments is safe because all three are uses of the connection, and a connection in a call of ours is not running a statement. Nothing may be thrown out of an upcall, so a watcher that throws is logged and answered as a stop. That is the reading that loses least: a callback that threw is a program that has stopped wanting the answer, and running on would only mean throwing the answer away later. Ten tests, over a nested loop of nine million pairs, because a scan of ten million rows in a frame finishes in ten milliseconds and there is nothing there to report on. --- README.md | 15 ++ zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java | 3 + .../main/java/dev/zudb/ffm/FfmBinding.java | 59 +++++ .../src/main/java/dev/zudb/ffm/Release.java | 73 ++---- .../src/main/java/dev/zudb/ffm/Upcall.java | 88 +++++++ .../src/main/java/dev/zudb/ffm/Watch.java | 96 ++++++++ .../test/java/dev/zudb/ffm/ProgressTest.java | 225 ++++++++++++++++++ zudb/src/main/java/dev/zudb/Connection.java | 38 +++ zudb/src/main/java/dev/zudb/Progress.java | 37 +++ .../src/main/java/dev/zudb/spi/ZuBinding.java | 18 ++ 10 files changed, 593 insertions(+), 59 deletions(-) create mode 100644 zudb-ffm/src/main/java/dev/zudb/ffm/Upcall.java create mode 100644 zudb-ffm/src/main/java/dev/zudb/ffm/Watch.java create mode 100644 zudb-ffm/src/test/java/dev/zudb/ffm/ProgressTest.java create mode 100644 zudb/src/main/java/dev/zudb/Progress.java diff --git a/README.md b/README.md index fa5af6f..a46e123 100644 --- a/README.md +++ b/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. diff --git a/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java b/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java index d2e7e39..b1469c9 100644 --- a/zudb-ffm/src/main/java/dev/zudb/ffm/Abi.java +++ b/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)); diff --git a/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java b/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java index d4cb887..8ab8368 100644 --- a/zudb-ffm/src/main/java/dev/zudb/ffm/FfmBinding.java +++ b/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. + * + *

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 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(); diff --git a/zudb-ffm/src/main/java/dev/zudb/ffm/Release.java b/zudb-ffm/src/main/java/dev/zudb/ffm/Release.java index 8b7ba03..9464665 100644 --- a/zudb-ffm/src/main/java/dev/zudb/ffm/Release.java +++ b/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. * - *

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. - * - *

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. + *

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 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,7 +51,7 @@ static Release of(Runnable body) { * @return the stub */ MemorySegment stub() { - return stub; + return upcall.stub(); } /** @@ -90,7 +59,7 @@ MemorySegment stub() { * 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(); } } diff --git a/zudb-ffm/src/main/java/dev/zudb/ffm/Upcall.java b/zudb-ffm/src/main/java/dev/zudb/ffm/Upcall.java new file mode 100644 index 0000000..4958b8c --- /dev/null +++ b/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. + * + *

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. + * + *

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 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; + } + } + } +} diff --git a/zudb-ffm/src/main/java/dev/zudb/ffm/Watch.java b/zudb-ffm/src/main/java/dev/zudb/ffm/Watch.java new file mode 100644 index 0000000..0d90641 --- /dev/null +++ b/zudb-ffm/src/main/java/dev/zudb/ffm/Watch.java @@ -0,0 +1,96 @@ +package dev.zudb.ffm; + +import static java.lang.foreign.ValueLayout.ADDRESS; +import static java.lang.foreign.ValueLayout.JAVA_INT; +import static java.lang.foreign.ValueLayout.JAVA_LONG; + +import dev.zudb.Progress; +import java.lang.System.Logger; +import java.lang.System.Logger.Level; +import java.lang.foreign.FunctionDescriptor; +import java.lang.foreign.MemorySegment; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; + +/** What a progress callback is on this side. */ +final class Watch { + + private static final Logger LOG = System.getLogger("dev.zudb"); + + private static final FunctionDescriptor DESCRIPTOR = + FunctionDescriptor.of(JAVA_INT, ADDRESS, JAVA_LONG, JAVA_LONG); + + private static final MethodHandle AT = at(); + + private final Progress body; + private volatile Upcall upcall; + + private Watch(Progress body) { + this.body = body; + } + + /** + * A {@code zu_progress_fn} bound to this watcher. + * + * @param body what to call + * @return the watch + */ + static Watch of(Progress body) { + Watch watch = new Watch(body); + watch.upcall = Upcall.of(AT.bindTo(watch), DESCRIPTOR); + return watch; + } + + /** + * The function pointer. + * + * @return the stub + */ + MemorySegment stub() { + return upcall.stub(); + } + + /** Says nothing will call this again, which is what taking the arrangement back means. */ + void spend() { + upcall.spend(); + } + + /** + * Called by the engine. + * + *

Nothing may be thrown out of an upcall, so a watcher that throws is + * logged and answered as though it had asked for the statement to stop. That + * is the reading that loses least: a progress callback that threw is a + * program that has stopped wanting the answer, and letting the statement run + * on would only mean throwing the answer away later. + * + * @param userData the pointer passed at creation, which this binding does + * not use because the watcher is already bound to this stub + * @param rows how many rows have been read + * @param millis how long the statement has been running + * @return 1 to let it go on, 0 to stop it + */ + @SuppressWarnings("unused") + private int at(MemorySegment userData, long rows, long millis) { + try { + return body.at(rows, millis) ? 1 : 0; + } catch (Throwable t) { + LOG.log(Level.ERROR, "a zu progress callback threw, so the statement it watched is stopping", t); + return 0; + } + } + + private static MethodHandle at() { + try { + return MethodHandles.lookup() + .findVirtual( + Watch.class, + "at", + MethodType.methodType( + int.class, MemorySegment.class, long.class, long.class)); + } catch (ReflectiveOperationException e) { + throw new ExceptionInInitializerError(e); + } + } +} diff --git a/zudb-ffm/src/test/java/dev/zudb/ffm/ProgressTest.java b/zudb-ffm/src/test/java/dev/zudb/ffm/ProgressTest.java new file mode 100644 index 0000000..6f62d6c --- /dev/null +++ b/zudb-ffm/src/test/java/dev/zudb/ffm/ProgressTest.java @@ -0,0 +1,225 @@ +package dev.zudb.ffm; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.zudb.Connection; +import dev.zudb.Database; +import dev.zudb.Frame; +import dev.zudb.Progress; +import dev.zudb.ZuException; +import dev.zudb.ZuInterruptedException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.LongBuffer; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * Being told how a statement is getting on, and telling it to stop. + * + *

Every test here needs a statement that runs long enough to be reported + * on, which frames make harder rather than easier: a scan of ten million rows + * is ten milliseconds. So what is watched here is a nested loop over three + * thousand rows against themselves, which is a third of a second and is not + * something the planner can fold into a count. + */ +class ProgressTest { + + private static final int ROWS = 3_000; + + /** + * Nine million pairs, which is what there is to watch. + * + *

A plain scan of a frame is too fast to be reported on at all, which is + * a nice problem to have and an awkward one to write a test against. A pair + * of patterns with a predicate over them is a nested loop the planner cannot + * fold into a count, and three thousand rows of it is a third of a second. + */ + private static final String SLOW = + "MATCH (a:Person), (b:Person) WHERE a.id < b.id RETURN count(*)"; + + private static Database db; + private static Connection conn; + private static Frame frame; + + @BeforeAll + static void engine() { + Libzu.require(); + LongBuffer ids = + ByteBuffer.allocateDirect(ROWS * Long.BYTES).order(ByteOrder.nativeOrder()).asLongBuffer(); + for (int i = 0; i < ROWS; i++) { + ids.put(i); + } + ids.flip(); + db = Database.memory(); + conn = db.connect(); + frame = Frame.of("Person", ROWS); + frame.column("id", ids); + conn.register(frame); + } + + @AfterAll + static void done() { + conn.close(); + frame.close(); + db.close(); + } + + @BeforeEach + void quiet() { + conn.clearProgress(); + } + + @Test + void theWatcherIsToldHowFarTheStatementHasGot() { + List calls = new CopyOnWriteArrayList<>(); + conn.onProgress( + Duration.ofMillis(1), + (rows, millis) -> { + calls.add(new long[] {rows, millis}); + return true; + }); + conn.query(SLOW).close(); + assertFalse(calls.isEmpty(), "a third of a second went by without a word"); + long rows = 0; + long millis = 0; + for (long[] call : calls) { + assertTrue(call[0] >= rows, "the rows read went backwards"); + assertTrue(call[1] >= millis, "the clock went backwards"); + rows = call[0]; + millis = call[1]; + } + assertTrue(rows > 0, "the rows read never moved off nought"); + } + + @Test + void theWatcherRunsOnAThreadOfTheLibrarys() { + Thread asked = Thread.currentThread(); + List threads = new CopyOnWriteArrayList<>(); + conn.onProgress( + Duration.ofMillis(1), + (rows, millis) -> { + threads.add(Thread.currentThread()); + return true; + }); + conn.query(SLOW).close(); + assertFalse(threads.isEmpty()); + for (Thread thread : threads) { + assertFalse(thread == asked, "the callback ran on the thread that asked for the statement"); + } + } + + @Test + void aWatcherThatSaysNoStopsTheStatement() { + conn.onProgress(Duration.ofMillis(1), (rows, millis) -> false); + assertThrows(ZuInterruptedException.class, () -> conn.query(SLOW).close()); + } + + @Test + void theConnectionRunsTheNextStatementNormallyAfterOneWasStopped() { + conn.onProgress(Duration.ofMillis(1), (rows, millis) -> false); + assertThrows(ZuInterruptedException.class, () -> conn.query(SLOW).close()); + conn.clearProgress(); + assertEquals(1L, one("RETURN 1 AS v")); + } + + @Test + void aWatcherThatThrowsStopsTheStatementRatherThanTheJvm() { + // An exception crossing an upcall would take the JVM down, so the binding + // catches it, logs it and answers as though the watcher had asked for the + // statement to stop. The one thrown here carries no stack trace, because + // printing one takes a fair share of the scan being watched and a test + // should not be timing the logger. + conn.onProgress(Duration.ofMillis(1), (rows, millis) -> { + throw new Quiet(); + }); + assertThrows(ZuInterruptedException.class, () -> conn.query(SLOW).close()); + } + + @Test + void takingTheArrangementBackStopsTheCalls() { + AtomicInteger calls = new AtomicInteger(); + conn.onProgress( + Duration.ofMillis(1), + (rows, millis) -> { + calls.incrementAndGet(); + return true; + }); + conn.query(SLOW).close(); + assertTrue(calls.get() > 0); + conn.clearProgress(); + int seen = calls.get(); + conn.query(SLOW).close(); + assertEquals(seen, calls.get(), "the watcher was called after it was taken back"); + } + + @Test + void anArrangementIsReplacedRatherThanAddedTo() { + AtomicInteger first = new AtomicInteger(); + AtomicInteger second = new AtomicInteger(); + conn.onProgress(Duration.ofMillis(1), counting(first)); + conn.onProgress(Duration.ofMillis(1), counting(second)); + conn.query(SLOW).close(); + assertEquals(0, first.get(), "the watcher that was replaced was called anyway"); + assertTrue(second.get() > 0); + } + + @Test + void anIntervalOfNothingIsRefused() { + assertThrows( + ZuException.class, () -> conn.onProgress(Duration.ZERO, (rows, millis) -> true)); + } + + @Test + void anIntervalLongerThanTheStatementMeansNothingIsSaidBeforeItEnds() { + AtomicInteger calls = new AtomicInteger(); + conn.onProgress(Duration.ofSeconds(30), counting(calls)); + assertEquals(1L, one("RETURN 1 AS v")); + assertEquals(0, calls.get()); + } + + @Test + void aConnectionThatCloseWithAWatcherOnItLetsGoOfIt() { + // Nothing to assert but that this does not crash: the stub outlives the + // connection and something has to free it. + try (Connection other = db.connect()) { + other.onProgress(Duration.ofMillis(1), (rows, millis) -> true); + } + Connection another = db.connect(); + another.onProgress(Duration.ofMillis(5), (rows, millis) -> true); + another.close(); + assertEquals(1L, one("RETURN 1 AS v")); + } + + private static Progress counting(AtomicInteger calls) { + return (rows, millis) -> { + calls.incrementAndGet(); + return true; + }; + } + + /** An exception that costs nothing to log. */ + private static final class Quiet extends RuntimeException { + + private static final long serialVersionUID = 1L; + + Quiet() { + super("the program has stopped wanting this answer", null, false, false); + } + } + + private static long one(String statement) { + try (var r = conn.query(statement)) { + return r.row(0).getLong(0); + } + } +} diff --git a/zudb/src/main/java/dev/zudb/Connection.java b/zudb/src/main/java/dev/zudb/Connection.java index 0d9cc10..3790914 100644 --- a/zudb/src/main/java/dev/zudb/Connection.java +++ b/zudb/src/main/java/dev/zudb/Connection.java @@ -1,6 +1,7 @@ package dev.zudb; import dev.zudb.spi.ZuBinding; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicLong; @@ -197,6 +198,43 @@ public long rowsRead() { return zu.connRowsRead(open()); } + /** + * Asks to be called back every so often while a statement runs, with how + * far it has got and whether it should go on. + * + *

This is {@link #rowsRead()} without the thread that would have to do + * the polling. The arrangement belongs to the connection and covers every + * statement after it, and a statement already running keeps the one it + * started with, so this is set once when the connection is opened rather + * than around each query. + * + *

A watcher answering false stops the statement exactly as + * {@link #interrupt()} would, which is what a timeout is: + * + * 

{@code
+   * long deadline = 30_000;
+   * conn.onProgress(Duration.ofMillis(250), (rows, millis) -> millis < deadline);
+   * }
+ * + *

The callback runs on a thread of the library's, so read + * {@link Progress} before writing one. + * + * @param every how often to be called, which the engine refuses at zero + * because a period of nothing is not a period + * @param watcher what to call + */ + public void onProgress(Duration every, Progress watcher) { + zu.connSetProgress(open(), watcher, every.toMillis()); + } + + /** + * Takes the arrangement back, after which nothing is called and the next + * statement runs as it would have. + */ + public void clearProgress() { + zu.connSetProgress(open(), null, 0); + } + /** * Starts a transaction. * diff --git a/zudb/src/main/java/dev/zudb/Progress.java b/zudb/src/main/java/dev/zudb/Progress.java new file mode 100644 index 0000000..086b87a --- /dev/null +++ b/zudb/src/main/java/dev/zudb/Progress.java @@ -0,0 +1,37 @@ +package dev.zudb; + +/** + * Called while a statement runs, to say how far it has got and to be asked + * whether it should go on. + * + *

This is {@link Connection#rowsRead()} the other way round: a poll wants a + * thread of its own to do the polling, and this one is called for you. It is + * what a progress bar, a query timeout and a server that has to answer within + * a deadline are all made of, and the deadline case is why the call has an + * answer at all. + * + *

It 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 from that. Whatever the callback touches has to be usable from + * another thread, so a counter a progress bar reads should be an + * {@code AtomicLong} rather than a field. And a callback must not call back + * into the library on the connection it is reporting on, because that + * connection is inside the executor and would answer + * {@link ZuConcurrentException} at best. + */ +@FunctionalInterface +public interface Progress { + + /** + * How far the running statement has got. + * + * @param rows how many rows it has read out of storage, which is rows read + * rather than rows answered because the statement a user is waiting on + * is exactly the one reading a hundred million rows to answer one + * @param millis how long it has been running + * @return true to let it go on, false to stop it, which ends it exactly as + * {@link Connection#interrupt()} would and raises + * {@link ZuInterruptedException} at the caller + */ + boolean at(long rows, long millis); +} diff --git a/zudb/src/main/java/dev/zudb/spi/ZuBinding.java b/zudb/src/main/java/dev/zudb/spi/ZuBinding.java index 7db49f3..2322a91 100644 --- a/zudb/src/main/java/dev/zudb/spi/ZuBinding.java +++ b/zudb/src/main/java/dev/zudb/spi/ZuBinding.java @@ -1,6 +1,7 @@ package dev.zudb.spi; import dev.zudb.Diagnostic; +import dev.zudb.Progress; import java.nio.Buffer; import java.nio.ByteBuffer; import java.nio.DoubleBuffer; @@ -159,6 +160,23 @@ public interface ZuBinding { */ long connRowsRead(long conn); + /** + * Asks to be called back every so often while a statement runs. + * + *

The arrangement belongs to the connection and covers every statement + * after it, and a statement already running keeps the one it started with. + * A provider owns whatever it had to build to make the callback reachable + * from a thread of the library's, and frees it when the arrangement is + * replaced, taken back, or the connection closes. + * + * @param conn the connection + * @param watcher what to call, or null to take the arrangement back + * @param intervalMillis how often, which the engine refuses at zero because + * a period of nothing is not a period, and which is ignored for a null + * watcher + */ + void connSetProgress(long conn, Progress watcher, long intervalMillis); + /** * Whether a transaction is running, which no statement answers and every * host offering a block needs.