Redline shared watchdog

redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java

@@ -0,0 +1,180 @@
+package run.endive.redline.experimental.api.internal;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/** Raises {@link CtxBuffer#INTERRUPT_FLAG} for watched calls whose thread is interrupted. */
+public final class InterruptWatchdog {
+
+    private static final Logger LOG = Logger.getLogger(InterruptWatchdog.class.getName());
+
+    private static final long POLL_INTERVAL_NANOS =
+            millisProperty("endive.redline.interruptPollMillis", 100);
+
+    // how long the poller waits without calls before it exits
+    private static final long IDLE_EXIT_NANOS =
+            millisProperty("endive.redline.interruptIdleMillis", 60_000);
+
+    // STATE holds the RUNNING and IDLE flags of the poller plus CALL per watched call
+    private static final int RUNNING = 1;
+
+    private static final int IDLE = 2;
+
+    private static final int CALL = 4;
+
+    private static final AtomicInteger STATE = new AtomicInteger();
+
+    private static final Set<Registration> ACTIVE = ConcurrentHashMap.newKeySet();
+
+    private static volatile Thread poller;
+
+    private InterruptWatchdog() {}
+
+    /** Raises the interrupt flag in a machine's context buffer. */
+    @FunctionalInterface
+    public interface InterruptSink {
+        void requestInterrupt();
+    }
+
+    /** Watches the current thread until the returned handle is passed to {@link #exit}. */
+    public static Registration enter(InterruptSink sink) {
+        var registration = new Registration(Thread.currentThread(), sink);
+        ACTIVE.add(registration);
+        int state = STATE.getAndAdd(CALL);
+        try {
+            if ((state & RUNNING) == 0) {
+                ensurePoller();
+            } else if ((state & IDLE) != 0 && (STATE.getAndUpdate(s -> s & ~IDLE) & IDLE) != 0) {
+                LockSupport.unpark(poller);
+            }
+        } catch (RuntimeException | Error e) {
+            exit(registration);
+            throw e;
+        }
+        return registration;
+    }
+
+    /** Stops watching; once this returns the poller can no longer raise the flag. */
+    public static void exit(Registration registration) {
+        registration.deactivate();
+        ACTIVE.remove(registration);
+        STATE.getAndAdd(-CALL);
+    }
+
+    private static long millisProperty(String name, long defaultMillis) {
+        return TimeUnit.MILLISECONDS.toNanos(Math.max(1, Long.getLong(name, defaultMillis)));
+    }
+
+    @SuppressWarnings("ThreadPriorityCheck") // not the priority of whichever caller starts it
+    private static void ensurePoller() {
+        if ((STATE.getAndUpdate(s -> s | RUNNING) & RUNNING) != 0) {
+            return;
+        }
+        try {
+            // no thread locals or class loader from the caller either
+            var thread =
+                    new Thread(
+                            null,
+                            InterruptWatchdog::pollLoop,
+                            "endive-redline-interrupt",
+                            0,
+                            false);
+            thread.setDaemon(true);
+            thread.setContextClassLoader(null);
+            thread.setPriority(Thread.NORM_PRIORITY);
+            poller = thread;
+            thread.start();
+        } catch (RuntimeException | Error e) {
+            STATE.getAndAdd(-RUNNING);
+            throw e;
+        }
+    }
+
+    private static void pollLoop() {
+        boolean retired = false;
+        try {
+            while (true) {
+                // an interrupt status would make every park return at once
+                Thread.interrupted();
+                if (STATE.get() < CALL && idleUntilExit()) {
+                    retired = true;
+                    return;
+                }
+                pollAll();
+                LockSupport.parkNanos(POLL_INTERVAL_NANOS);
+            }
+        } finally {
+            if (!retired && STATE.updateAndGet(s -> s & ~(RUNNING | IDLE)) >= CALL) {
+                // died on an error: hand the watched calls to a new poller
+                ensurePoller();
+            }
+        }
+    }
+
+    // parks until a call clears IDLE or the idle time passes; true if the poller should exit
+    private static boolean idleUntilExit() {
+        if (!STATE.compareAndSet(RUNNING, RUNNING | IDLE)) {
+            return false;
+        }
+        long deadline = System.nanoTime() + IDLE_EXIT_NANOS;
+        long left;
+        while ((STATE.get() & IDLE) != 0 && (left = deadline - System.nanoTime()) > 0) {
+            LockSupport.parkNanos(left);
+            Thread.interrupted();
+        }
+        if (STATE.compareAndSet(RUNNING | IDLE, 0)) {
+            return true;
+        }
+        STATE.getAndUpdate(s -> s & ~IDLE);
+        return false;
+    }
+
+    private static void pollAll() {
+        for (Iterator<Registration> it = ACTIVE.iterator(); it.hasNext(); ) {
+            var registration = it.next();
+            try {
+                registration.poll();
+            } catch (RuntimeException | Error e) {
+                // a broken sink: stop watching that call rather than every call
+                it.remove();
+                LOG.log(Level.WARNING, "Stopped watching a call whose interrupt flag failed", e);
+            }
+        }
+    }
+
+    /** One watched call. */
+    public static final class Registration {
+
+        private final Thread caller;
+        private final InterruptSink sink;
+        private boolean active = true;
+
+        private Registration(Thread caller, InterruptSink sink) {
+            this.caller = caller;
+            this.sink = sink;
+        }
+
+        // the lock pairs with deactivate(), so the flag is never raised after exit()
+        private void poll() {
+            if (caller.isInterrupted()) {
+                synchronized (this) {
+                    if (active) {
+                        sink.requestInterrupt();
+                    }
+                }
+            }
+        }
+
+        private void deactivate() {
+            synchronized (this) {
+                active = false;
+            }
+        }
+    }
+}

redline/runner-jffi-tests/pom.xml

@@ -78,6 +78,16 @@
 
   <build>
     <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <systemPropertyVariables>
+            <!-- let the interrupt watchdog retire and restart between tests -->
+            <endive.redline.interruptIdleMillis>200</endive.redline.interruptIdleMillis>
+          </systemPropertyVariables>
+        </configuration>
+      </plugin>
       <plugin>
         <groupId>run.endive</groupId>
         <artifactId>test-gen-plugin</artifactId>

redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java

@@ -1,24 +1,20 @@
 package run.endive.redline.experimental.runner.jffi.internal;
 
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 
+import java.util.List;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import run.endive.corpus.CorpusResources;
-import run.endive.redline.experimental.api.internal.RedlineTarget;
-import run.endive.redline.experimental.compiler.internal.NativeCompiler;
-import run.endive.redline.experimental.runner.jffi.JffiNativeMachineFactory;
 import run.endive.runtime.HostFunction;
 import run.endive.runtime.ImportValues;
+import run.endive.testing.NativeInstanceBuilder;
 import run.endive.wasm.Parser;
 import run.endive.wasm.types.FunctionType;
 
-/**
- * The watchdog raises the interrupt flag from another thread, so it can land after
- * the call it was meant to stop has passed its last check. The flag must not then
- * sit in the context and stop a later call that nobody interrupted.
- */
+/** An interrupt the host handled during a call must not stop a later call. */
 public class InterruptFlagTest {
 
     @AfterEach
@@ -28,45 +24,80 @@ public void clearInterruptStatus() {
     }
 
     @Test
-    public void aFlagRaisedMidCallDoesNotStopTheNextCall() {
+    public void anInterruptHandledByTheHostDoesNotStopTheNextCall() {
         var module =
                 Parser.parse(CorpusResources.getResource("compiled/interrupt-midcall.wat.wasm"));
-
-        var machineRef = new JffiNativeMachine[1];
         var imports =
                 ImportValues.builder()
                         .addFunction(
                                 new HostFunction(
                                         "host",
                                         "raiseFlag",
-                                        FunctionType.of(java.util.List.of(), java.util.List.of()),
+                                        FunctionType.of(List.of(), List.of()),
                                         (inst, args) -> {
-                                            machineRef[0].requestInterrupt();
+                                            interruptAndHandle();
                                             return null;
                                         }))
                         .build();
 
         try (var instance =
-                JffiNativeMachineFactory.builder(module)
-                        .withImportValues(imports)
-                        .withCompilerFunction(
-                                m ->
-                                        NativeCompiler.compileAll(
-                                                RedlineTarget.detectHost().orElseThrow().triple(),
-                                                m))
-                        .build()) {
-            machineRef[0] = (JffiNativeMachine) instance.getMachine();
-
-            // Returns normally: the entry check ran before the flag was raised.
+                NativeInstanceBuilder.builder(module).withImportValues(imports).build()) {
+            // nothing polls after the host function, so this returns normally
             instance.export("callHost").apply();
 
             assertEquals(
                     42,
                     (int) instance.export("answer").apply()[0],
-                    "a flag left over from the previous call must not stop this one");
-            assertFalse(
-                    Thread.currentThread().isInterrupted(),
-                    "no interrupt happened, so the caller must not be left interrupted");
+                    "an interrupt handled during the previous call must not stop this one");
+            assertFalse(Thread.currentThread().isInterrupted());
         }
     }
+
+    @Test
+    public void anInterruptHandledInANestedCallDoesNotStopTheOuterCall() {
+        var module =
+                Parser.parse(CorpusResources.getResource("compiled/reentrant-interrupt.wat.wasm"));
+        var noParams = FunctionType.of(List.of(), List.of());
+        var imports =
+                ImportValues.builder()
+                        .addFunction(
+                                new HostFunction(
+                                        "host",
+                                        "reenter",
+                                        noParams,
+                                        (inst, args) -> {
+                                            inst.export("raise").apply();
+                                            return null;
+                                        }))
+                        .addFunction(
+                                new HostFunction(
+                                        "host",
+                                        "raiseFlag",
+                                        noParams,
+                                        (inst, args) -> {
+                                            interruptAndHandle();
+                                            return null;
+                                        }))
+                        .addFunction(
+                                new HostFunction("host", "tick", noParams, (inst, args) -> null))
+                        .build();
+
+        try (var instance =
+                NativeInstanceBuilder.builder(module).withImportValues(imports).build()) {
+            assertEquals(
+                    1000,
+                    (int) instance.export("run").apply()[0],
+                    "an interrupt handled in the nested call must not stop the outer one");
+        }
+    }
+
+    // interrupts this thread, gives the watchdog time to see it, then handles it as a host would
+    private static void interruptAndHandle() {
+        Thread.currentThread().interrupt();
+        long end = System.nanoTime() + MILLISECONDS.toNanos(500);
+        while (System.nanoTime() < end) {
+            Thread.onSpinWait();
+        }
+        Thread.interrupted();
+    }
 }