From 911c2e2a2511f362c790203f7d2f00f624fa4575 Mon Sep 17 00:00:00 2001 From: andreatp Date: Tue, 22 Sep 2026 18:09:29 +0100 Subject: [PATCH 1/5] fix(redline): share one interrupt watchdog instead of starting a thread per call --- .../api/internal/InterruptWatchdog.java | 130 ++++++++++++++++++ .../internal/WatchdogRegistrationTest.java | 118 ++++++++++++++++ .../jffi/internal/JffiNativeMachine.java | 67 +++++---- .../internal/WatchdogRegistrationTest.java | 118 ++++++++++++++++ .../runner/internal/NativeMachine.java | 43 +++--- 5 files changed, 418 insertions(+), 58 deletions(-) create mode 100644 redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java create mode 100644 redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java create mode 100644 redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java diff --git a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java new file mode 100644 index 000000000..7400821b7 --- /dev/null +++ b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java @@ -0,0 +1,130 @@ +package run.endive.redline.experimental.api.internal; + +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; + +/** + * Translates {@link Thread#interrupt()} on a thread that is inside native code into a write + * of {@link CtxBuffer#INTERRUPT_FLAG}, which compiled code polls at function entry and at + * loop headers. + * + *

A poller is unavoidable: the JVM offers no callback on interrupt, and a compute loop + * that never calls a host function reaches no other safepoint. What is avoidable is paying + * for a fresh {@link Thread} on every call, which is what the runners used to do — a host + * function calling back into an export paid it again for the nested call. + * + *

So there is one daemon poller for the whole JVM. It parks while no call is in flight, + * so an idle process carries no cost, and it wakes on the first registration. + * + *

Registrations are per call, not per thread: a host function may call into a + * different machine, and that machine's call is outermost as far as it is + * concerned, so both have to be watched. + */ +public final class InterruptWatchdog { + + /** How often the poller re-reads the interrupt status of every in-flight caller. */ + private static final long POLL_INTERVAL_NANOS = 1_000_000L; + + private static final Set ACTIVE = ConcurrentHashMap.newKeySet(); + + private static final AtomicReference POLLER = new AtomicReference<>(); + + private InterruptWatchdog() {} + + /** Raises the interrupt flag in a machine's context buffer. */ + @FunctionalInterface + public interface InterruptSink { + void requestInterrupt(); + } + + /** + * Starts watching {@code caller} for the duration of one call. The returned handle must + * be passed to {@link #exit(Registration)} in a {@code finally}. + */ + public static Registration enter(Thread caller, InterruptSink sink) { + var registration = new Registration(caller, sink); + ACTIVE.add(registration); + LockSupport.unpark(poller()); + return registration; + } + + /** + * Stops watching. On return the poller is guaranteed not to be part-way through raising + * the flag for this registration, so the caller may clear the flag without racing it. + */ + public static void exit(Registration registration) { + registration.deactivate(); + ACTIVE.remove(registration); + } + + /** Visible for testing: how many calls are currently being watched. */ + public static int activeCount() { + return ACTIVE.size(); + } + + private static Thread poller() { + Thread existing = POLLER.get(); + if (existing != null) { + return existing; + } + synchronized (InterruptWatchdog.class) { + existing = POLLER.get(); + if (existing != null) { + return existing; + } + var thread = new Thread(InterruptWatchdog::pollLoop, "endive-redline-interrupt"); + thread.setDaemon(true); + POLLER.set(thread); + thread.start(); + return thread; + } + } + + private static void pollLoop() { + while (true) { + if (ACTIVE.isEmpty()) { + // A registration that lands between the check and the park still wakes us: + // unpark leaves a permit behind, so park returns at once. + LockSupport.park(); + continue; + } + for (Registration registration : ACTIVE) { + registration.poll(); + } + LockSupport.parkNanos(POLL_INTERVAL_NANOS); + } + } + + /** One in-flight call. Identity-based equality, so a thread may hold several. */ + 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; + } + + // Synchronized against deactivate() so that once exit() returns, no poller thread + // can still be about to raise the flag. Without it the flag could be set just after + // the call cleared it, and the next call on that machine would trap for an + // interrupt nobody requested. + private void poll() { + synchronized (this) { + if (active && caller.isInterrupted()) { + sink.requestInterrupt(); + } + } + } + + private void deactivate() { + synchronized (this) { + active = false; + } + } + } +} diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java new file mode 100644 index 000000000..0c7602c05 --- /dev/null +++ b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java @@ -0,0 +1,118 @@ +package run.endive.redline.experimental.runner.jffi.internal; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.lang.management.ManagementFactory; +import java.util.List; +import java.util.function.IntConsumer; +import org.junit.jupiter.api.Test; +import run.endive.corpus.CorpusResources; +import run.endive.redline.experimental.api.internal.InterruptWatchdog; +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.runtime.Instance; +import run.endive.wasm.Parser; +import run.endive.wasm.types.FunctionType; + +/** + * Interrupting a call means raising a flag that compiled code polls, and the JVM offers no + * callback on {@link Thread#interrupt()} to raise it from. The runners used to answer that + * with a fresh thread per call, which a host function calling back into an export paid + * again for every nested level. + * + *

These two tests pin the replacement from both ends: one registration per outermost + * call however deep the nesting goes, and no threads started as a result of calling. + */ +public class WatchdogRegistrationTest { + + /** Nesting per call. Well clear of the reentrant stack guard, which fires near 115. */ + private static final int DEPTH = 20; + + private static final int ROUNDS = 20; + + @Test + public void nestingAddsNoRegistrations() { + int[] deepest = {0}; + withReentrantInstance( + instance -> { + instance.export("recurse").apply(); + }, + level -> deepest[0] = Math.max(deepest[0], InterruptWatchdog.activeCount())); + + assertEquals( + 1, + deepest[0], + "a nested call runs on the same thread inside the same watched window, so it" + + " must reuse the outermost call's registration"); + } + + @Test + public void callingStartsNoThreads() { + var threads = ManagementFactory.getThreadMXBean(); + long[] started = {0}; + + withReentrantInstance( + instance -> { + // One call first, so the shared poller is already running by the time + // the counter is read. + instance.export("recurse").apply(); + + long before = threads.getTotalStartedThreadCount(); + for (int i = 0; i < ROUNDS; i++) { + instance.export("recurse").apply(); + } + started[0] = threads.getTotalStartedThreadCount() - before; + }, + level -> {}); + + // Generous, because the JIT may start compiler threads while this runs. The + // behaviour being excluded started one thread per call, so ROUNDS * (DEPTH + 1). + assertTrue( + started[0] < ROUNDS, + "calling must not start a thread per call, but " + + started[0] + + " threads started across " + + (ROUNDS * (DEPTH + 1)) + + " calls"); + } + + private static void withReentrantInstance( + java.util.function.Consumer body, IntConsumer atEachLevel) { + var module = + Parser.parse(CorpusResources.getResource("compiled/reentrant-recursion.wat.wasm")); + + int[] depth = {0}; + var imports = + ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + FunctionType.of(List.of(), List.of()), + (Instance inst, long... args) -> { + atEachLevel.accept(depth[0]); + if (depth[0]++ < DEPTH) { + inst.export("recurse").apply(); + } + depth[0] = 0; + return null; + })) + .build(); + + try (var instance = + JffiNativeMachineFactory.builder(module) + .withImportValues(imports) + .withCompilerFunction( + m -> + NativeCompiler.compileAll( + RedlineTarget.detectHost().orElseThrow().triple(), + m)) + .build()) { + body.accept(instance); + } + } +} diff --git a/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java b/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java index 4adfb8d2b..8ea697219 100644 --- a/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java +++ b/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java @@ -17,6 +17,7 @@ import java.util.List; import java.util.Map; import run.endive.redline.experimental.api.internal.CtxBuffer; +import run.endive.redline.experimental.api.internal.InterruptWatchdog; import run.endive.redline.experimental.api.internal.RedlineTarget; import run.endive.redline.experimental.api.internal.TypeMapUtils; import run.endive.redline.experimental.bridge.internal.CraneliftBridge; @@ -87,6 +88,11 @@ public final class JffiNativeMachine implements Machine { private final Instance instance; private final CallContext[] entryTrampolineCallCtxs; // entry trampoline CallContext per func private final long[] entryTrampolineAddrs; // entry trampoline native addr per func + + // Only the >6-native-arg path needs a Function, and it is a pure value holder over + // (address, context) whose dispose() is a no-op, so it is built once per function + // rather than on every call. + private final Function[] entryTrampolineFunctions; private final FunctionType[] funcTypes; // wasm FunctionType per func private final long codeRegionAddr; private final int codeRegionOsPages; @@ -139,6 +145,7 @@ public JffiNativeMachine( int totalFuncs = numImports + module.codeSection().functionBodyCount(); this.entryTrampolineCallCtxs = new CallContext[totalFuncs]; this.entryTrampolineAddrs = new long[totalFuncs]; + this.entryTrampolineFunctions = new Function[totalFuncs]; this.funcTypes = new FunctionType[totalFuncs]; this.importHandles = new Closure.Handle[numImports]; @@ -376,6 +383,10 @@ public JffiNativeMachine( entryTrampolineAddrs[funcId] = entryTrampolinePtrs.get(funcTypesByBody[i]); entryTrampolineCallCtxs[funcId] = createEntryTrampolineCallContext(funcTypesByBody[i]); + entryTrampolineFunctions[funcId] = + new Function( + entryTrampolineAddrs[funcId], + entryTrampolineCallCtxs[funcId]); } } } @@ -941,6 +952,7 @@ private static WasmEngineException trapException(int trapCode) { private long invokeViaEntryTrampoline( CallContext trampolineCallCtx, long trampolineAddr, + Function trampolineFunction, FunctionType funcType, long funcAddr, long memBase, @@ -978,25 +990,17 @@ private long invokeViaEntryTrampoline( default: // >6 args: use HeapInvocationBuffer return invokeViaBufferWithTrampoline( - trampolineCallCtx, - trampolineAddr, - funcType, - funcAddr, - memBase, - ctxPtr, - wasmArgs); + trampolineFunction, funcType, funcAddr, memBase, ctxPtr, wasmArgs); } } private long invokeViaBufferWithTrampoline( - CallContext trampolineCallCtx, - long trampolineAddr, + Function func, FunctionType funcType, long funcAddr, long memBase, long ctxPtr, long[] wasmArgs) { - var func = new Function(trampolineAddr, trampolineCallCtx); var buffer = new HeapInvocationBuffer(func); buffer.putAddress(funcAddr); // funcPtr (first arg to entry trampoline) buffer.putAddress(memBase); @@ -1080,42 +1084,37 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { throw new TrapException("interrupted"); } - Thread caller = Thread.currentThread(); - Thread watchdog = - new Thread( - () -> { - // Keeps raising rather than returning after the - // first: a nested call clears the flag when it - // finishes, and the outer call still needs it. - while (!Thread.currentThread().isInterrupted()) { - if (caller.isInterrupted()) { - requestInterrupt(); - } - try { - Thread.sleep(1); - } catch (InterruptedException e) { - return; - } - } - }); - watchdog.setDaemon(true); - watchdog.start(); + // Only the outermost call registers. A nested call runs on the same + // thread, inside the same watched window, so watching it again would + // buy nothing — and re-entry through a host function is common enough + // that doing so once dominated the cost of the call itself. + InterruptWatchdog.Registration watchdog = + outermostCall + ? InterruptWatchdog.enter( + Thread.currentThread(), this::requestInterrupt) + : null; long result; try { result = invokeViaEntryTrampoline( trampolineCallCtx, trampolineAddr, + entryTrampolineFunctions[funcId], funcType, funcAddr, cachedMemBase, ctxBufferAddr, args); } finally { - watchdog.interrupt(); - // The flag only ever means "stop this call". Left set it would - // trap the next one on a thread nobody interrupted. - clearInterrupt(); + if (watchdog != null) { + // Deregister before clearing: exit() guarantees the poller is + // not part-way through raising the flag, so the clear below + // cannot be undone behind our back. + InterruptWatchdog.exit(watchdog); + // The flag only ever means "stop this call". Left set it would + // trap the next one on a thread nobody interrupted. + clearInterrupt(); + } } // Check for exceptions from upcall stubs first — a host function diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java new file mode 100644 index 000000000..8b7c5c2ff --- /dev/null +++ b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java @@ -0,0 +1,118 @@ +package run.endive.redline.experimental.runner.internal; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.lang.management.ManagementFactory; +import java.util.List; +import java.util.function.IntConsumer; +import org.junit.jupiter.api.Test; +import run.endive.corpus.CorpusResources; +import run.endive.redline.experimental.api.internal.InterruptWatchdog; +import run.endive.redline.experimental.api.internal.RedlineTarget; +import run.endive.redline.experimental.compiler.internal.NativeCompiler; +import run.endive.redline.experimental.runner.NativeMachineFactory; +import run.endive.runtime.HostFunction; +import run.endive.runtime.ImportValues; +import run.endive.runtime.Instance; +import run.endive.wasm.Parser; +import run.endive.wasm.types.FunctionType; + +/** + * Interrupting a call means raising a flag that compiled code polls, and the JVM offers no + * callback on {@link Thread#interrupt()} to raise it from. The runners used to answer that + * with a fresh thread per call, which a host function calling back into an export paid + * again for every nested level. + * + *

These two tests pin the replacement from both ends: one registration per outermost + * call however deep the nesting goes, and no threads started as a result of calling. + */ +public class WatchdogRegistrationTest { + + /** Nesting per call. Well clear of the reentrant stack guard, which fires near 115. */ + private static final int DEPTH = 20; + + private static final int ROUNDS = 20; + + @Test + public void nestingAddsNoRegistrations() { + int[] deepest = {0}; + withReentrantInstance( + instance -> { + instance.export("recurse").apply(); + }, + level -> deepest[0] = Math.max(deepest[0], InterruptWatchdog.activeCount())); + + assertEquals( + 1, + deepest[0], + "a nested call runs on the same thread inside the same watched window, so it" + + " must reuse the outermost call's registration"); + } + + @Test + public void callingStartsNoThreads() { + var threads = ManagementFactory.getThreadMXBean(); + long[] started = {0}; + + withReentrantInstance( + instance -> { + // One call first, so the shared poller is already running by the time + // the counter is read. + instance.export("recurse").apply(); + + long before = threads.getTotalStartedThreadCount(); + for (int i = 0; i < ROUNDS; i++) { + instance.export("recurse").apply(); + } + started[0] = threads.getTotalStartedThreadCount() - before; + }, + level -> {}); + + // Generous, because the JIT may start compiler threads while this runs. The + // behaviour being excluded started one thread per call, so ROUNDS * (DEPTH + 1). + assertTrue( + started[0] < ROUNDS, + "calling must not start a thread per call, but " + + started[0] + + " threads started across " + + (ROUNDS * (DEPTH + 1)) + + " calls"); + } + + private static void withReentrantInstance( + java.util.function.Consumer body, IntConsumer atEachLevel) { + var module = + Parser.parse(CorpusResources.getResource("compiled/reentrant-recursion.wat.wasm")); + + int[] depth = {0}; + var imports = + ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + FunctionType.of(List.of(), List.of()), + (Instance inst, long... args) -> { + atEachLevel.accept(depth[0]); + if (depth[0]++ < DEPTH) { + inst.export("recurse").apply(); + } + depth[0] = 0; + return null; + })) + .build(); + + try (var instance = + NativeMachineFactory.builder(module) + .withImportValues(imports) + .withCompilerFunction( + m -> + NativeCompiler.compileAll( + RedlineTarget.detectHost().orElseThrow().triple(), + m)) + .build()) { + body.accept(instance); + } + } +} diff --git a/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java b/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java index b9da6235b..c5b29d8c9 100644 --- a/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java +++ b/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java @@ -14,6 +14,7 @@ import java.util.Map; import java.util.function.Function; import run.endive.redline.experimental.api.internal.CtxBuffer; +import run.endive.redline.experimental.api.internal.InterruptWatchdog; import run.endive.redline.experimental.api.internal.RedlineTarget; import run.endive.redline.experimental.api.internal.TypeMapUtils; import run.endive.redline.experimental.bridge.internal.CraneliftBridge; @@ -1083,34 +1084,28 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { throw new TrapException("interrupted"); } - Thread caller = Thread.currentThread(); - Thread watchdog = - new Thread( - () -> { - // Keeps raising rather than returning after the - // first: a nested call clears the flag when it - // finishes, and the outer call still needs it. - while (!Thread.currentThread().isInterrupted()) { - if (caller.isInterrupted()) { - requestInterrupt(); - } - try { - Thread.sleep(1); - } catch (InterruptedException e) { - return; - } - } - }); - watchdog.setDaemon(true); - watchdog.start(); + // Only the outermost call registers. A nested call runs on the same + // thread, inside the same watched window, so watching it again would + // buy nothing — and re-entry through a host function is common enough + // that doing so once dominated the cost of the call itself. + InterruptWatchdog.Registration watchdog = + outermostCall + ? InterruptWatchdog.enter( + Thread.currentThread(), this::requestInterrupt) + : null; long result; try { result = (long) handle.invokeExact(cachedMemBase, ctxBuffer, args); } finally { - watchdog.interrupt(); - // The flag only ever means "stop this call". Left set it would - // trap the next one on a thread nobody interrupted. - clearInterrupt(); + if (watchdog != null) { + // Deregister before clearing: exit() guarantees the poller is + // not part-way through raising the flag, so the clear below + // cannot be undone behind our back. + InterruptWatchdog.exit(watchdog); + // The flag only ever means "stop this call". Left set it would + // trap the next one on a thread nobody interrupted. + clearInterrupt(); + } } // Check for exceptions from upcall stubs first — a host function From c194b8bf924e0eb3f7d0f24975cc1b9bb209d37c Mon Sep 17 00:00:00 2001 From: andreatp Date: Tue, 29 Sep 2026 10:24:02 +0100 Subject: [PATCH 2/5] Harden the shared redline interrupt watchdog --- .../api/internal/InterruptWatchdog.java | 78 +++++++++--------- .../jffi/internal/InterruptFlagTest.java | 49 ++++++++++- .../jffi/internal/InterruptionTest.java | 35 ++++++++ .../internal/WatchdogRegistrationTest.java | 64 ++++---------- .../jffi/internal/JffiNativeMachine.java | 27 ++---- .../runner/internal/InterruptFlagTest.java | 49 ++++++++++- .../runner/internal/InterruptionTest.java | 35 ++++++++ .../internal/WatchdogRegistrationTest.java | 64 ++++---------- .../runner/internal/NativeMachine.java | 23 ++---- .../compiled/reentrant-interrupt.wat.wasm | Bin 0 -> 177 bytes .../resources/wat/reentrant-interrupt.wat | 19 +++++ 11 files changed, 274 insertions(+), 169 deletions(-) create mode 100644 wasm-corpus/src/main/resources/compiled/reentrant-interrupt.wat.wasm create mode 100644 wasm-corpus/src/main/resources/wat/reentrant-interrupt.wat diff --git a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java index 7400821b7..eb7bf577b 100644 --- a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java +++ b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java @@ -1,36 +1,26 @@ package run.endive.redline.experimental.api.internal; +import java.util.Iterator; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; /** - * Translates {@link Thread#interrupt()} on a thread that is inside native code into a write - * of {@link CtxBuffer#INTERRUPT_FLAG}, which compiled code polls at function entry and at - * loop headers. - * - *

A poller is unavoidable: the JVM offers no callback on interrupt, and a compute loop - * that never calls a host function reaches no other safepoint. What is avoidable is paying - * for a fresh {@link Thread} on every call, which is what the runners used to do — a host - * function calling back into an export paid it again for the nested call. - * - *

So there is one daemon poller for the whole JVM. It parks while no call is in flight, - * so an idle process carries no cost, and it wakes on the first registration. - * - *

Registrations are per call, not per thread: a host function may call into a - * different machine, and that machine's call is outermost as far as it is - * concerned, so both have to be watched. + * Raises {@link CtxBuffer#INTERRUPT_FLAG} for calls whose thread is interrupted, from a single + * daemon poller shared by all machines. */ public final class InterruptWatchdog { - /** How often the poller re-reads the interrupt status of every in-flight caller. */ private static final long POLL_INTERVAL_NANOS = 1_000_000L; private static final Set ACTIVE = ConcurrentHashMap.newKeySet(); private static final AtomicReference POLLER = new AtomicReference<>(); + // true while the poller is parked with nothing to watch + private static volatile boolean idle; + private InterruptWatchdog() {} /** Raises the interrupt flag in a machine's context buffer. */ @@ -39,27 +29,24 @@ public interface InterruptSink { void requestInterrupt(); } - /** - * Starts watching {@code caller} for the duration of one call. The returned handle must - * be passed to {@link #exit(Registration)} in a {@code finally}. - */ + /** Watches {@code caller} until the returned handle is passed to {@link #exit}. */ public static Registration enter(Thread caller, InterruptSink sink) { + var poller = poller(); var registration = new Registration(caller, sink); ACTIVE.add(registration); - LockSupport.unpark(poller()); + if (idle) { + LockSupport.unpark(poller); + } return registration; } - /** - * Stops watching. On return the poller is guaranteed not to be part-way through raising - * the flag for this registration, so the caller may clear the flag without racing it. - */ + /** Stops watching; once this returns the poller can no longer raise the flag for it. */ public static void exit(Registration registration) { registration.deactivate(); ACTIVE.remove(registration); } - /** Visible for testing: how many calls are currently being watched. */ + /** Visible for testing. */ public static int activeCount() { return ACTIVE.size(); } @@ -74,30 +61,48 @@ private static Thread poller() { if (existing != null) { return existing; } - var thread = new Thread(InterruptWatchdog::pollLoop, "endive-redline-interrupt"); + // no thread locals or class loader from whichever caller starts it + var thread = + new Thread( + null, + InterruptWatchdog::pollLoop, + "endive-redline-interrupt", + 0, + false); thread.setDaemon(true); - POLLER.set(thread); + thread.setContextClassLoader(null); thread.start(); + POLLER.set(thread); return thread; } } private static void pollLoop() { while (true) { + // an interrupt status would make every park return at once + Thread.interrupted(); if (ACTIVE.isEmpty()) { - // A registration that lands between the check and the park still wakes us: - // unpark leaves a permit behind, so park returns at once. - LockSupport.park(); + idle = true; + // re-check, enter() may have read idle before it was set + if (ACTIVE.isEmpty()) { + LockSupport.park(); + } + idle = false; continue; } - for (Registration registration : ACTIVE) { - registration.poll(); + for (Iterator it = ACTIVE.iterator(); it.hasNext(); ) { + try { + it.next().poll(); + } catch (RuntimeException e) { + // a failing sink must not stop the poller for everyone else + it.remove(); + } } LockSupport.parkNanos(POLL_INTERVAL_NANOS); } } - /** One in-flight call. Identity-based equality, so a thread may hold several. */ + /** One in-flight call. */ public static final class Registration { private final Thread caller; @@ -109,10 +114,7 @@ private Registration(Thread caller, InterruptSink sink) { this.sink = sink; } - // Synchronized against deactivate() so that once exit() returns, no poller thread - // can still be about to raise the flag. Without it the flag could be set just after - // the call cleared it, and the next call on that machine would trap for an - // interrupt nobody requested. + // synchronized with deactivate(), so the flag is never raised after exit() private void poll() { synchronized (this) { if (active && caller.isInterrupted()) { diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java index 55c8fdf3c..327001459 100644 --- a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java +++ b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java @@ -3,6 +3,7 @@ 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; @@ -39,7 +40,7 @@ public void aFlagRaisedMidCallDoesNotStopTheNextCall() { 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(); return null; @@ -69,4 +70,50 @@ public void aFlagRaisedMidCallDoesNotStopTheNextCall() { "no interrupt happened, so the caller must not be left interrupted"); } } + + @Test + public void aFlagRaisedInANestedCallDoesNotStopTheOuterCall() { + var module = + Parser.parse(CorpusResources.getResource("compiled/reentrant-interrupt.wat.wasm")); + + var machineRef = new JffiNativeMachine[1]; + var imports = + ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + FunctionType.of(List.of(), List.of()), + (inst, args) -> { + inst.export("raise").apply(); + return null; + })) + .addFunction( + new HostFunction( + "host", + "raiseFlag", + FunctionType.of(List.of(), List.of()), + (inst, args) -> { + machineRef[0].requestInterrupt(); + 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(); + + assertEquals( + 1000, + (int) instance.export("run").apply()[0], + "a flag left over from the nested call must not stop the outer one"); + } + } } diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java index fff74104e..949febccc 100644 --- a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java +++ b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java @@ -5,15 +5,19 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; 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.runtime.Instance; import run.endive.wasm.Parser; import run.endive.wasm.WasmEngineException; +import run.endive.wasm.types.FunctionType; public class InterruptionTest { @@ -33,6 +37,32 @@ public void shouldInterruptCallViaThread() throws InterruptedException { } } + @Test + public void shouldInterruptNestedLoopViaThread() throws InterruptedException { + var imports = + ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + FunctionType.of(List.of(), List.of()), + (inst, args) -> { + inst.export("spin").apply(); + return null; + })) + .addFunction( + new HostFunction( + "host", + "raiseFlag", + FunctionType.of(List.of(), List.of()), + (inst, args) -> null)) + .build(); + try (var instance = buildInstance("compiled/reentrant-interrupt.wat.wasm", imports)) { + var function = instance.export("run"); + assertThreadInterruption(function::apply); + } + } + private static void assertThreadInterruption(Runnable function) throws InterruptedException { AtomicBoolean interrupted = new AtomicBoolean(); Thread thread = @@ -52,8 +82,13 @@ private static void assertThreadInterruption(Runnable function) throws Interrupt } private static Instance buildInstance(String resource) { + return buildInstance(resource, ImportValues.builder().build()); + } + + private static Instance buildInstance(String resource, ImportValues imports) { var module = Parser.parse(CorpusResources.getResource(resource)); return JffiNativeMachineFactory.builder(module) + .withImportValues(imports) .withCompilerFunction( m -> NativeCompiler.compileAll( diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java index 0c7602c05..1f20fb6bb 100644 --- a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java +++ b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java @@ -5,49 +5,35 @@ import java.lang.management.ManagementFactory; import java.util.List; -import java.util.function.IntConsumer; +import java.util.function.Consumer; import org.junit.jupiter.api.Test; import run.endive.corpus.CorpusResources; import run.endive.redline.experimental.api.internal.InterruptWatchdog; -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.runtime.Instance; +import run.endive.testing.NativeInstanceBuilder; import run.endive.wasm.Parser; import run.endive.wasm.types.FunctionType; -/** - * Interrupting a call means raising a flag that compiled code polls, and the JVM offers no - * callback on {@link Thread#interrupt()} to raise it from. The runners used to answer that - * with a fresh thread per call, which a host function calling back into an export paid - * again for every nested level. - * - *

These two tests pin the replacement from both ends: one registration per outermost - * call however deep the nesting goes, and no threads started as a result of calling. - */ +/** Calls are watched by the shared poller, one registration per outermost call. */ public class WatchdogRegistrationTest { - /** Nesting per call. Well clear of the reentrant stack guard, which fires near 115. */ + // well below the reentrant stack guard, which fires near 115 private static final int DEPTH = 20; private static final int ROUNDS = 20; @Test - public void nestingAddsNoRegistrations() { + public void nestingRegistersOnlyTheOutermostCall() { + int before = InterruptWatchdog.activeCount(); int[] deepest = {0}; withReentrantInstance( - instance -> { - instance.export("recurse").apply(); - }, - level -> deepest[0] = Math.max(deepest[0], InterruptWatchdog.activeCount())); + instance -> instance.export("recurse").apply(), + () -> deepest[0] = Math.max(deepest[0], InterruptWatchdog.activeCount())); - assertEquals( - 1, - deepest[0], - "a nested call runs on the same thread inside the same watched window, so it" - + " must reuse the outermost call's registration"); + assertEquals(1, deepest[0] - before); + assertEquals(before, InterruptWatchdog.activeCount()); } @Test @@ -57,8 +43,7 @@ public void callingStartsNoThreads() { withReentrantInstance( instance -> { - // One call first, so the shared poller is already running by the time - // the counter is read. + // starts the shared poller instance.export("recurse").apply(); long before = threads.getTotalStartedThreadCount(); @@ -67,21 +52,13 @@ public void callingStartsNoThreads() { } started[0] = threads.getTotalStartedThreadCount() - before; }, - level -> {}); + () -> {}); - // Generous, because the JIT may start compiler threads while this runs. The - // behaviour being excluded started one thread per call, so ROUNDS * (DEPTH + 1). - assertTrue( - started[0] < ROUNDS, - "calling must not start a thread per call, but " - + started[0] - + " threads started across " - + (ROUNDS * (DEPTH + 1)) - + " calls"); + // loose bound: JIT compiler threads may start meanwhile + assertTrue(started[0] < ROUNDS, started[0] + " threads started"); } - private static void withReentrantInstance( - java.util.function.Consumer body, IntConsumer atEachLevel) { + private static void withReentrantInstance(Consumer body, Runnable atEachLevel) { var module = Parser.parse(CorpusResources.getResource("compiled/reentrant-recursion.wat.wasm")); @@ -94,7 +71,7 @@ private static void withReentrantInstance( "reenter", FunctionType.of(List.of(), List.of()), (Instance inst, long... args) -> { - atEachLevel.accept(depth[0]); + atEachLevel.run(); if (depth[0]++ < DEPTH) { inst.export("recurse").apply(); } @@ -104,14 +81,7 @@ private static void withReentrantInstance( .build(); try (var instance = - JffiNativeMachineFactory.builder(module) - .withImportValues(imports) - .withCompilerFunction( - m -> - NativeCompiler.compileAll( - RedlineTarget.detectHost().orElseThrow().triple(), - m)) - .build()) { + NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { body.accept(instance); } } diff --git a/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java b/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java index 8ea697219..795260496 100644 --- a/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java +++ b/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java @@ -42,7 +42,7 @@ * return: Wasm return value * */ -public final class JffiNativeMachine implements Machine { +public final class JffiNativeMachine implements Machine, InterruptWatchdog.InterruptSink { private static final int CTX_SIZE = CtxBuffer.CTX_SIZE; private static final MemoryIO MEM = MemoryIO.getInstance(); @@ -89,9 +89,7 @@ public final class JffiNativeMachine implements Machine { private final CallContext[] entryTrampolineCallCtxs; // entry trampoline CallContext per func private final long[] entryTrampolineAddrs; // entry trampoline native addr per func - // Only the >6-native-arg path needs a Function, and it is a pure value holder over - // (address, context) whose dispose() is a no-op, so it is built once per function - // rather than on every call. + // built once per function instead of per call, for the >6-native-arg path private final Function[] entryTrampolineFunctions; private final FunctionType[] funcTypes; // wasm FunctionType per func private final long codeRegionAddr; @@ -1080,19 +1078,13 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { MEM.putInt(ctxBufferAddr + CtxBuffer.MEMORY_PAGES, mem.pages()); } - if (Thread.interrupted()) { + if (Thread.currentThread().isInterrupted()) { throw new TrapException("interrupted"); } - // Only the outermost call registers. A nested call runs on the same - // thread, inside the same watched window, so watching it again would - // buy nothing — and re-entry through a host function is common enough - // that doing so once dominated the cost of the call itself. + // nested calls run inside the outermost call's watch InterruptWatchdog.Registration watchdog = - outermostCall - ? InterruptWatchdog.enter( - Thread.currentThread(), this::requestInterrupt) - : null; + outermostCall ? InterruptWatchdog.enter(Thread.currentThread(), this) : null; long result; try { result = @@ -1107,14 +1099,10 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { args); } finally { if (watchdog != null) { - // Deregister before clearing: exit() guarantees the poller is - // not part-way through raising the flag, so the clear below - // cannot be undone behind our back. + // exit first, so the poller cannot raise the flag after the clear InterruptWatchdog.exit(watchdog); - // The flag only ever means "stop this call". Left set it would - // trap the next one on a thread nobody interrupted. - clearInterrupt(); } + clearInterrupt(); } // Check for exceptions from upcall stubs first — a host function @@ -1169,6 +1157,7 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { } } + @Override public void requestInterrupt() { CHECKED_MEM.putLong(ctxBufferAddr + CtxBuffer.INTERRUPT_FLAG, 1L); } diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java index d81115e90..b186cd845 100644 --- a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java +++ b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java @@ -3,6 +3,7 @@ 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; @@ -39,7 +40,7 @@ public void aFlagRaisedMidCallDoesNotStopTheNextCall() { 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(); return null; @@ -69,4 +70,50 @@ public void aFlagRaisedMidCallDoesNotStopTheNextCall() { "no interrupt happened, so the caller must not be left interrupted"); } } + + @Test + public void aFlagRaisedInANestedCallDoesNotStopTheOuterCall() { + var module = + Parser.parse(CorpusResources.getResource("compiled/reentrant-interrupt.wat.wasm")); + + var machineRef = new NativeMachine[1]; + var imports = + ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + FunctionType.of(List.of(), List.of()), + (inst, args) -> { + inst.export("raise").apply(); + return null; + })) + .addFunction( + new HostFunction( + "host", + "raiseFlag", + FunctionType.of(List.of(), List.of()), + (inst, args) -> { + machineRef[0].requestInterrupt(); + return null; + })) + .build(); + + try (var instance = + NativeMachineFactory.builder(module) + .withImportValues(imports) + .withCompilerFunction( + m -> + NativeCompiler.compileAll( + RedlineTarget.detectHost().orElseThrow().triple(), + m)) + .build()) { + machineRef[0] = (NativeMachine) instance.getMachine(); + + assertEquals( + 1000, + (int) instance.export("run").apply()[0], + "a flag left over from the nested call must not stop the outer one"); + } + } } diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java index 755f10d0b..5bd340d76 100644 --- a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java +++ b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java @@ -5,15 +5,19 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; 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.NativeMachineFactory; +import run.endive.runtime.HostFunction; +import run.endive.runtime.ImportValues; import run.endive.runtime.Instance; import run.endive.wasm.Parser; import run.endive.wasm.WasmEngineException; +import run.endive.wasm.types.FunctionType; public class InterruptionTest { @@ -33,6 +37,32 @@ public void shouldInterruptCallViaThread() throws InterruptedException { } } + @Test + public void shouldInterruptNestedLoopViaThread() throws InterruptedException { + var imports = + ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + FunctionType.of(List.of(), List.of()), + (inst, args) -> { + inst.export("spin").apply(); + return null; + })) + .addFunction( + new HostFunction( + "host", + "raiseFlag", + FunctionType.of(List.of(), List.of()), + (inst, args) -> null)) + .build(); + try (var instance = buildInstance("compiled/reentrant-interrupt.wat.wasm", imports)) { + var function = instance.export("run"); + assertThreadInterruption(function::apply); + } + } + private static void assertThreadInterruption(Runnable function) throws InterruptedException { AtomicBoolean interrupted = new AtomicBoolean(); Thread thread = @@ -52,8 +82,13 @@ private static void assertThreadInterruption(Runnable function) throws Interrupt } private static Instance buildInstance(String resource) { + return buildInstance(resource, ImportValues.builder().build()); + } + + private static Instance buildInstance(String resource, ImportValues imports) { var module = Parser.parse(CorpusResources.getResource(resource)); return NativeMachineFactory.builder(module) + .withImportValues(imports) .withCompilerFunction( m -> NativeCompiler.compileAll( diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java index 8b7c5c2ff..fdeaaa205 100644 --- a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java +++ b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java @@ -5,49 +5,35 @@ import java.lang.management.ManagementFactory; import java.util.List; -import java.util.function.IntConsumer; +import java.util.function.Consumer; import org.junit.jupiter.api.Test; import run.endive.corpus.CorpusResources; import run.endive.redline.experimental.api.internal.InterruptWatchdog; -import run.endive.redline.experimental.api.internal.RedlineTarget; -import run.endive.redline.experimental.compiler.internal.NativeCompiler; -import run.endive.redline.experimental.runner.NativeMachineFactory; import run.endive.runtime.HostFunction; import run.endive.runtime.ImportValues; import run.endive.runtime.Instance; +import run.endive.testing.NativeInstanceBuilder; import run.endive.wasm.Parser; import run.endive.wasm.types.FunctionType; -/** - * Interrupting a call means raising a flag that compiled code polls, and the JVM offers no - * callback on {@link Thread#interrupt()} to raise it from. The runners used to answer that - * with a fresh thread per call, which a host function calling back into an export paid - * again for every nested level. - * - *

These two tests pin the replacement from both ends: one registration per outermost - * call however deep the nesting goes, and no threads started as a result of calling. - */ +/** Calls are watched by the shared poller, one registration per outermost call. */ public class WatchdogRegistrationTest { - /** Nesting per call. Well clear of the reentrant stack guard, which fires near 115. */ + // well below the reentrant stack guard, which fires near 115 private static final int DEPTH = 20; private static final int ROUNDS = 20; @Test - public void nestingAddsNoRegistrations() { + public void nestingRegistersOnlyTheOutermostCall() { + int before = InterruptWatchdog.activeCount(); int[] deepest = {0}; withReentrantInstance( - instance -> { - instance.export("recurse").apply(); - }, - level -> deepest[0] = Math.max(deepest[0], InterruptWatchdog.activeCount())); + instance -> instance.export("recurse").apply(), + () -> deepest[0] = Math.max(deepest[0], InterruptWatchdog.activeCount())); - assertEquals( - 1, - deepest[0], - "a nested call runs on the same thread inside the same watched window, so it" - + " must reuse the outermost call's registration"); + assertEquals(1, deepest[0] - before); + assertEquals(before, InterruptWatchdog.activeCount()); } @Test @@ -57,8 +43,7 @@ public void callingStartsNoThreads() { withReentrantInstance( instance -> { - // One call first, so the shared poller is already running by the time - // the counter is read. + // starts the shared poller instance.export("recurse").apply(); long before = threads.getTotalStartedThreadCount(); @@ -67,21 +52,13 @@ public void callingStartsNoThreads() { } started[0] = threads.getTotalStartedThreadCount() - before; }, - level -> {}); + () -> {}); - // Generous, because the JIT may start compiler threads while this runs. The - // behaviour being excluded started one thread per call, so ROUNDS * (DEPTH + 1). - assertTrue( - started[0] < ROUNDS, - "calling must not start a thread per call, but " - + started[0] - + " threads started across " - + (ROUNDS * (DEPTH + 1)) - + " calls"); + // loose bound: JIT compiler threads may start meanwhile + assertTrue(started[0] < ROUNDS, started[0] + " threads started"); } - private static void withReentrantInstance( - java.util.function.Consumer body, IntConsumer atEachLevel) { + private static void withReentrantInstance(Consumer body, Runnable atEachLevel) { var module = Parser.parse(CorpusResources.getResource("compiled/reentrant-recursion.wat.wasm")); @@ -94,7 +71,7 @@ private static void withReentrantInstance( "reenter", FunctionType.of(List.of(), List.of()), (Instance inst, long... args) -> { - atEachLevel.accept(depth[0]); + atEachLevel.run(); if (depth[0]++ < DEPTH) { inst.export("recurse").apply(); } @@ -104,14 +81,7 @@ private static void withReentrantInstance( .build(); try (var instance = - NativeMachineFactory.builder(module) - .withImportValues(imports) - .withCompilerFunction( - m -> - NativeCompiler.compileAll( - RedlineTarget.detectHost().orElseThrow().triple(), - m)) - .build()) { + NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { body.accept(instance); } } diff --git a/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java b/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java index c5b29d8c9..1793fc86a 100644 --- a/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java +++ b/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java @@ -41,7 +41,7 @@ * *

See {@link CtxBuffer} for the full layout definition. */ -public final class NativeMachine implements Machine { +public final class NativeMachine implements Machine, InterruptWatchdog.InterruptSink { private static final int CTX_SIZE = CtxBuffer.CTX_SIZE; @@ -1080,32 +1080,22 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { ctxBuffer.set(ValueLayout.JAVA_INT, CtxBuffer.MEMORY_PAGES, mem.pages()); } - if (Thread.interrupted()) { + if (Thread.currentThread().isInterrupted()) { throw new TrapException("interrupted"); } - // Only the outermost call registers. A nested call runs on the same - // thread, inside the same watched window, so watching it again would - // buy nothing — and re-entry through a host function is common enough - // that doing so once dominated the cost of the call itself. + // nested calls run inside the outermost call's watch InterruptWatchdog.Registration watchdog = - outermostCall - ? InterruptWatchdog.enter( - Thread.currentThread(), this::requestInterrupt) - : null; + outermostCall ? InterruptWatchdog.enter(Thread.currentThread(), this) : null; long result; try { result = (long) handle.invokeExact(cachedMemBase, ctxBuffer, args); } finally { if (watchdog != null) { - // Deregister before clearing: exit() guarantees the poller is - // not part-way through raising the flag, so the clear below - // cannot be undone behind our back. + // exit first, so the poller cannot raise the flag after the clear InterruptWatchdog.exit(watchdog); - // The flag only ever means "stop this call". Left set it would - // trap the next one on a thread nobody interrupted. - clearInterrupt(); } + clearInterrupt(); } // Check for exceptions from upcall stubs first — a host function @@ -1159,6 +1149,7 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { } } + @Override public void requestInterrupt() { ctxBuffer.set(ValueLayout.JAVA_LONG, CtxBuffer.INTERRUPT_FLAG, 1L); } diff --git a/wasm-corpus/src/main/resources/compiled/reentrant-interrupt.wat.wasm b/wasm-corpus/src/main/resources/compiled/reentrant-interrupt.wat.wasm new file mode 100644 index 0000000000000000000000000000000000000000..9a1d4e7b4dff316fba565aa5915c137d959f1841 GIT binary patch literal 177 zcmYL?I}XAy7=!&kAV5Is0s~V!GI9f0khn!NL{QWQl$5#G Date: Thu, 1 Oct 2026 13:31:49 +0100 Subject: [PATCH 3/5] Retire the shared redline interrupt poller when idle and make it failure-safe --- .../api/internal/InterruptWatchdog.java | 160 ++++++++++++------ .../api/internal/InterruptWatchdogTest.java | 159 +++++++++++++++++ .../jffi/internal/InterruptFlagTest.java | 28 +-- .../jffi/internal/InterruptionTest.java | 131 +++++++++----- .../jffi/internal/JffiNativeMachine.java | 39 ++--- .../runner/internal/InterruptFlagTest.java | 28 +-- .../runner/internal/InterruptionTest.java | 131 +++++++++----- .../runner/internal/NativeMachine.java | 5 +- .../compiled/reentrant-interrupt.wat.wasm | Bin 177 -> 197 bytes .../resources/wat/reentrant-interrupt.wat | 5 +- 10 files changed, 477 insertions(+), 209 deletions(-) create mode 100644 redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java diff --git a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java index eb7bf577b..74a0fc199 100644 --- a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java +++ b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java @@ -3,23 +3,35 @@ import java.util.Iterator; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.LockSupport; +import java.util.logging.Level; +import java.util.logging.Logger; -/** - * Raises {@link CtxBuffer#INTERRUPT_FLAG} for calls whose thread is interrupted, from a single - * daemon poller shared by all machines. - */ +/** Raises {@link CtxBuffer#INTERRUPT_FLAG} for watched calls whose thread is interrupted. */ public final class InterruptWatchdog { - private static final long POLL_INTERVAL_NANOS = 1_000_000L; + private static final Logger LOG = Logger.getLogger(InterruptWatchdog.class.getName()); - private static final Set ACTIVE = ConcurrentHashMap.newKeySet(); + private static final long POLL_INTERVAL_NANOS = TimeUnit.MILLISECONDS.toNanos(1); + + // how long the poller waits without calls before it exits; tests shorten it + static final AtomicLong IDLE_EXIT_NANOS = new AtomicLong(TimeUnit.MINUTES.toNanos(1)); + + // 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 AtomicReference POLLER = new AtomicReference<>(); + private static final int CALL = 4; + + private static final AtomicInteger STATE = new AtomicInteger(); + + private static final Set ACTIVE = ConcurrentHashMap.newKeySet(); - // true while the poller is parked with nothing to watch - private static volatile boolean idle; + private static volatile Thread poller; private InterruptWatchdog() {} @@ -29,21 +41,29 @@ public interface InterruptSink { void requestInterrupt(); } - /** Watches {@code caller} until the returned handle is passed to {@link #exit}. */ - public static Registration enter(Thread caller, InterruptSink sink) { - var poller = poller(); - var registration = new Registration(caller, sink); + /** 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); - if (idle) { - LockSupport.unpark(poller); + 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 for it. */ + /** 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); } /** Visible for testing. */ @@ -51,17 +71,21 @@ public static int activeCount() { return ACTIVE.size(); } - private static Thread poller() { - Thread existing = POLLER.get(); - if (existing != null) { - return existing; + static boolean pollerRunning() { + return (STATE.get() & RUNNING) != 0; + } + + static boolean pollerIdle() { + return (STATE.get() & IDLE) != 0; + } + + @SuppressWarnings("ThreadPriorityCheck") // not the priority of whichever caller starts it + private static void ensurePoller() { + if ((STATE.getAndUpdate(s -> s | RUNNING) & RUNNING) != 0) { + return; } - synchronized (InterruptWatchdog.class) { - existing = POLLER.get(); - if (existing != null) { - return existing; - } - // no thread locals or class loader from whichever caller starts it + try { + // no thread locals or class loader from the caller either var thread = new Thread( null, @@ -71,38 +95,68 @@ private static Thread poller() { false); thread.setDaemon(true); thread.setContextClassLoader(null); + thread.setPriority(Thread.NORM_PRIORITY); + poller = thread; thread.start(); - POLLER.set(thread); - return thread; + } catch (RuntimeException | Error e) { + STATE.getAndAdd(-RUNNING); + throw e; } } private static void pollLoop() { - while (true) { - // an interrupt status would make every park return at once - Thread.interrupted(); - if (ACTIVE.isEmpty()) { - idle = true; - // re-check, enter() may have read idle before it was set - if (ACTIVE.isEmpty()) { - LockSupport.park(); + 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; } - idle = false; - continue; + pollAll(); + LockSupport.parkNanos(POLL_INTERVAL_NANOS); } - for (Iterator it = ACTIVE.iterator(); it.hasNext(); ) { - try { - it.next().poll(); - } catch (RuntimeException e) { - // a failing sink must not stop the poller for everyone else - it.remove(); - } + } finally { + if (!retired && STATE.updateAndGet(s -> s & ~(RUNNING | IDLE)) >= CALL) { + // died on an error: hand the watched calls to a new poller + ensurePoller(); } - LockSupport.parkNanos(POLL_INTERVAL_NANOS); } } - /** One in-flight call. */ + // 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.get(); + 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 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; @@ -114,11 +168,13 @@ private Registration(Thread caller, InterruptSink sink) { this.sink = sink; } - // synchronized with deactivate(), so the flag is never raised after exit() + // the lock pairs with deactivate(), so the flag is never raised after exit() private void poll() { - synchronized (this) { - if (active && caller.isInterrupted()) { - sink.requestInterrupt(); + if (caller.isInterrupted()) { + synchronized (this) { + if (active) { + sink.requestInterrupt(); + } } } } diff --git a/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java b/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java new file mode 100644 index 000000000..17790b56f --- /dev/null +++ b/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java @@ -0,0 +1,159 @@ +package run.endive.redline.experimental.api.internal; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; +import java.util.function.BooleanSupplier; +import java.util.logging.Handler; +import java.util.logging.LogRecord; +import java.util.logging.Logger; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +public class InterruptWatchdogTest { + + @AfterEach + public void clearInterruptStatus() { + Thread.interrupted(); + } + + @Test + public void raisesTheFlagOfAnInterruptedCallerUntilExit() { + var flag = new Flag(); + var registration = InterruptWatchdog.enter(flag); + try { + Thread.currentThread().interrupt(); + awaitTrue(() -> flag.raised); + } finally { + InterruptWatchdog.exit(registration); + } + + flag.raised = false; + // the caller is still interrupted, but no longer watched + long end = System.nanoTime() + MILLISECONDS.toNanos(50); + while (System.nanoTime() < end) { + LockSupport.parkNanos(MILLISECONDS.toNanos(1)); + } + assertFalse(flag.raised); + } + + @Test + public void aFailingSinkIsDroppedAndLogged() { + var failure = new IllegalStateException("closed"); + InterruptWatchdog.InterruptSink broken = + () -> { + throw failure; + }; + var logged = new AtomicReference(); + var log = Logger.getLogger(InterruptWatchdog.class.getName()); + var handler = + new Handler() { + @Override + public void publish(LogRecord record) { + logged.compareAndSet(null, record.getThrown()); + } + + @Override + public void flush() {} + + @Override + public void close() {} + }; + log.addHandler(handler); + log.setUseParentHandlers(false); + int before = InterruptWatchdog.activeCount(); + var brokenRegistration = InterruptWatchdog.enter(broken); + var flag = new Flag(); + var registration = InterruptWatchdog.enter(flag); + try { + Thread.currentThread().interrupt(); + awaitTrue(() -> flag.raised && logged.get() != null); + assertSame(failure, logged.get()); + assertEquals(before + 1, InterruptWatchdog.activeCount(), "broken call dropped"); + } finally { + InterruptWatchdog.exit(registration); + InterruptWatchdog.exit(brokenRegistration); + log.removeHandler(handler); + log.setUseParentHandlers(true); + } + } + + @Test + public void aCallWakesAnIdlePoller() { + var flag = new Flag(); + InterruptWatchdog.exit(InterruptWatchdog.enter(flag)); + awaitTrue(InterruptWatchdog::pollerIdle); + + var registration = InterruptWatchdog.enter(flag); + try { + long start = System.nanoTime(); + Thread.currentThread().interrupt(); + awaitTrue(() -> flag.raised); + // well before the idle poller would wake up on its own + assertTrue(System.nanoTime() - start < MILLISECONDS.toNanos(500)); + } finally { + InterruptWatchdog.exit(registration); + } + } + + @Test + public void thePollerExitsWhenIdleAndRestartsOnTheNextCall() { + long idleExit = InterruptWatchdog.IDLE_EXIT_NANOS.getAndSet(MILLISECONDS.toNanos(100)); + try { + var flag = new Flag(); + var first = new AtomicReference(); + var registration = InterruptWatchdog.enter(flag); + try { + awaitTrue(() -> first.updateAndGet(t -> pollerThread()) != null); + } finally { + InterruptWatchdog.exit(registration); + } + + awaitTrue(() -> !InterruptWatchdog.pollerRunning() && !first.get().isAlive()); + + registration = InterruptWatchdog.enter(flag); + try { + // the old poller is dead, so this one was started by the call above + assertNotNull(pollerThread()); + Thread.currentThread().interrupt(); + awaitTrue(() -> flag.raised); + } finally { + InterruptWatchdog.exit(registration); + } + } finally { + InterruptWatchdog.IDLE_EXIT_NANOS.set(idleExit); + } + } + + private static Thread pollerThread() { + return Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().equals("endive-redline-interrupt") && t.isAlive()) + .findFirst() + .orElse(null); + } + + // parkNanos, as the caller is often interrupted on purpose and cannot sleep + private static void awaitTrue(BooleanSupplier condition) { + long deadline = System.nanoTime() + SECONDS.toNanos(10); + while (!condition.getAsBoolean()) { + assertTrue(System.nanoTime() < deadline, "timed out"); + LockSupport.parkNanos(MILLISECONDS.toNanos(1)); + } + } + + private static final class Flag implements InterruptWatchdog.InterruptSink { + volatile boolean raised; + + @Override + public void requestInterrupt() { + raised = true; + } + } +} diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java index 327001459..9e6d14534 100644 --- a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java +++ b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java @@ -7,11 +7,9 @@ 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; @@ -48,14 +46,7 @@ public void aFlagRaisedMidCallDoesNotStopTheNextCall() { .build(); try (var instance = - JffiNativeMachineFactory.builder(module) - .withImportValues(imports) - .withCompilerFunction( - m -> - NativeCompiler.compileAll( - RedlineTarget.detectHost().orElseThrow().triple(), - m)) - .build()) { + NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { machineRef[0] = (JffiNativeMachine) instance.getMachine(); // Returns normally: the entry check ran before the flag was raised. @@ -97,17 +88,16 @@ public void aFlagRaisedInANestedCallDoesNotStopTheOuterCall() { machineRef[0].requestInterrupt(); return null; })) + .addFunction( + new HostFunction( + "host", + "tick", + FunctionType.of(List.of(), List.of()), + (inst, args) -> null)) .build(); try (var instance = - JffiNativeMachineFactory.builder(module) - .withImportValues(imports) - .withCompilerFunction( - m -> - NativeCompiler.compileAll( - RedlineTarget.detectHost().orElseThrow().triple(), - m)) - .build()) { + NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { machineRef[0] = (JffiNativeMachine) instance.getMachine(); assertEquals( diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java index 949febccc..cb6e1e148 100644 --- a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java +++ b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java @@ -2,19 +2,22 @@ import static java.util.concurrent.TimeUnit.SECONDS; 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 static org.junit.jupiter.api.Assertions.fail; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; 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.runtime.Instance; +import run.endive.testing.NativeInstanceBuilder; import run.endive.wasm.Parser; import run.endive.wasm.WasmEngineException; import run.endive.wasm.types.FunctionType; @@ -23,48 +26,51 @@ public class InterruptionTest { @Test public void shouldInterruptLoopViaThread() throws InterruptedException { - try (var instance = buildInstance("compiled/infinite-loop.c.wasm")) { - var function = instance.export("run"); - assertThreadInterruption(function::apply); - } + var instance = + buildInstance("compiled/infinite-loop.c.wasm", ImportValues.builder().build()); + var function = instance.export("run"); + assertThreadInterruption(function::apply, null, instance); } @Test public void shouldInterruptCallViaThread() throws InterruptedException { - try (var instance = buildInstance("compiled/power.c.wasm")) { - var function = instance.export("run"); - assertThreadInterruption(() -> function.apply(100)); - } + var instance = buildInstance("compiled/power.c.wasm", ImportValues.builder().build()); + var function = instance.export("run"); + assertThreadInterruption(() -> function.apply(100), null, instance); } @Test public void shouldInterruptNestedLoopViaThread() throws InterruptedException { - var imports = - ImportValues.builder() - .addFunction( - new HostFunction( - "host", - "reenter", - FunctionType.of(List.of(), List.of()), - (inst, args) -> { - inst.export("spin").apply(); - return null; - })) - .addFunction( - new HostFunction( - "host", - "raiseFlag", - FunctionType.of(List.of(), List.of()), - (inst, args) -> null)) - .build(); - try (var instance = buildInstance("compiled/reentrant-interrupt.wat.wasm", imports)) { - var function = instance.export("run"); - assertThreadInterruption(function::apply); - } + var running = new CountDownLatch(1); + var instance = + buildInstance( + "compiled/reentrant-interrupt.wat.wasm", + reentrantImports(inst -> inst.export("spin").apply(), running)); + var function = instance.export("run"); + assertThreadInterruption(function::apply, running, instance); + } + + @Test + public void shouldInterruptLoopInAnotherMachineViaThread() throws InterruptedException { + var running = new CountDownLatch(1); + var inner = + buildInstance( + "compiled/reentrant-interrupt.wat.wasm", + reentrantImports(inst -> {}, running)); + var outer = + buildInstance( + "compiled/reentrant-interrupt.wat.wasm", + reentrantImports(inst -> inner.export("spin").apply(), running)); + var function = outer.export("run"); + assertThreadInterruption(function::apply, running, outer, inner); } - private static void assertThreadInterruption(Runnable function) throws InterruptedException { + // interrupts once `running` is released, or after 100ms without one + private static void assertThreadInterruption( + Runnable function, CountDownLatch running, Instance... instances) + throws InterruptedException { AtomicBoolean interrupted = new AtomicBoolean(); + var failure = new AtomicReference(); Thread thread = new Thread( () -> { @@ -73,26 +79,63 @@ private static void assertThreadInterruption(Runnable function) throws Interrupt interrupted.set(true); }); thread.setDaemon(true); + thread.setUncaughtExceptionHandler( + (t, e) -> { + failure.set(e); + if (running != null) { + running.countDown(); + } + }); thread.start(); - Thread.sleep(100); + boolean started = true; + if (running == null) { + Thread.sleep(100); + } else { + started = running.await(10, SECONDS); + } thread.interrupt(); SECONDS.timedJoin(thread, 10); + if (failure.get() != null) { + fail("the call failed", failure.get()); + } + assertTrue(started, "the call never reached its loop"); + // a call still running uses the instances' memory, so they are leaked, not closed + assertFalse(thread.isAlive(), "the call was not interrupted"); + for (var instance : instances) { + instance.close(); + } assertTrue(interrupted.get()); } - private static Instance buildInstance(String resource) { - return buildInstance(resource, ImportValues.builder().build()); + private static ImportValues reentrantImports( + Consumer reenter, CountDownLatch running) { + var noParams = FunctionType.of(List.of(), List.of()); + return ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + noParams, + (inst, args) -> { + reenter.accept(inst); + return null; + })) + .addFunction(new HostFunction("host", "raiseFlag", noParams, (inst, args) -> null)) + .addFunction( + new HostFunction( + "host", + "tick", + noParams, + (inst, args) -> { + running.countDown(); + return null; + })) + .build(); } private static Instance buildInstance(String resource, ImportValues imports) { var module = Parser.parse(CorpusResources.getResource(resource)); - return JffiNativeMachineFactory.builder(module) - .withImportValues(imports) - .withCompilerFunction( - m -> - NativeCompiler.compileAll( - RedlineTarget.detectHost().orElseThrow().triple(), m)) - .build(); + return NativeInstanceBuilder.builder(module).withImportValues(imports).build(); } } diff --git a/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java b/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java index 795260496..950aa80e5 100644 --- a/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java +++ b/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java @@ -86,11 +86,7 @@ public final class JffiNativeMachine implements Machine, InterruptWatchdog.Inter } private final Instance instance; - private final CallContext[] entryTrampolineCallCtxs; // entry trampoline CallContext per func - private final long[] entryTrampolineAddrs; // entry trampoline native addr per func - - // built once per function instead of per call, for the >6-native-arg path - private final Function[] entryTrampolineFunctions; + private final Function[] entryTrampolines; // entry trampoline per func private final FunctionType[] funcTypes; // wasm FunctionType per func private final long codeRegionAddr; private final int codeRegionOsPages; @@ -141,9 +137,7 @@ public JffiNativeMachine( .FUNCTION) .count(); int totalFuncs = numImports + module.codeSection().functionBodyCount(); - this.entryTrampolineCallCtxs = new CallContext[totalFuncs]; - this.entryTrampolineAddrs = new long[totalFuncs]; - this.entryTrampolineFunctions = new Function[totalFuncs]; + this.entryTrampolines = new Function[totalFuncs]; this.funcTypes = new FunctionType[totalFuncs]; this.importHandles = new Closure.Handle[numImports]; @@ -378,13 +372,10 @@ public JffiNativeMachine( for (int i = 0; i < compiledCode.length; i++) { if (compiledCode[i] != null) { int funcId = numImports + i; - entryTrampolineAddrs[funcId] = entryTrampolinePtrs.get(funcTypesByBody[i]); - entryTrampolineCallCtxs[funcId] = - createEntryTrampolineCallContext(funcTypesByBody[i]); - entryTrampolineFunctions[funcId] = + entryTrampolines[funcId] = new Function( - entryTrampolineAddrs[funcId], - entryTrampolineCallCtxs[funcId]); + entryTrampolinePtrs.get(funcTypesByBody[i]), + createEntryTrampolineCallContext(funcTypesByBody[i])); } } } @@ -948,9 +939,7 @@ private static WasmEngineException trapException(int trapCode) { // --- Native function invocation --- private long invokeViaEntryTrampoline( - CallContext trampolineCallCtx, - long trampolineAddr, - Function trampolineFunction, + Function trampoline, FunctionType funcType, long funcAddr, long memBase, @@ -958,6 +947,8 @@ private long invokeViaEntryTrampoline( long[] wasmArgs) { // nativeArgCount = funcPtr + memBase + ctxPtr + wasm params int nativeArgCount = 3 + wasmArgs.length; + CallContext trampolineCallCtx = trampoline.getCallContext(); + long trampolineAddr = trampoline.getFunctionAddress(); switch (nativeArgCount) { case 3: @@ -988,7 +979,7 @@ private long invokeViaEntryTrampoline( default: // >6 args: use HeapInvocationBuffer return invokeViaBufferWithTrampoline( - trampolineFunction, funcType, funcAddr, memBase, ctxPtr, wasmArgs); + trampoline, funcType, funcAddr, memBase, ctxPtr, wasmArgs); } } @@ -1044,8 +1035,6 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { var funcType = funcTypes[funcId]; long funcAddr = MEM.getLong(funcTableAddr + (long) funcId * 8); - long trampolineAddr = entryTrampolineAddrs[funcId]; - var trampolineCallCtx = entryTrampolineCallCtxs[funcId]; try { boolean outermostCall = callDepth++ == 0; @@ -1084,14 +1073,12 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { // nested calls run inside the outermost call's watch InterruptWatchdog.Registration watchdog = - outermostCall ? InterruptWatchdog.enter(Thread.currentThread(), this) : null; + outermostCall ? InterruptWatchdog.enter(this) : null; long result; try { result = invokeViaEntryTrampoline( - trampolineCallCtx, - trampolineAddr, - entryTrampolineFunctions[funcId], + entryTrampolines[funcId], funcType, funcAddr, cachedMemBase, @@ -1121,7 +1108,6 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { if (trapCode != 0) { MEM.putInt(ctxBufferAddr + CtxBuffer.TRAP_CODE, 0); if (trapCode == CtxBuffer.TRAP_INTERRUPTED) { - CHECKED_MEM.putLong(ctxBufferAddr + CtxBuffer.INTERRUPT_FLAG, 0L); Thread.currentThread().interrupt(); } throw trapException(trapCode); @@ -1152,7 +1138,6 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { // Prevent the JIT from considering this machine unreachable during // the native call, which would let GC collect and close() free // native memory while code is executing. - // (ctxBuffer, funcTypesArray, code region) while code is executing. Reference.reachabilityFence(this); } } @@ -1162,7 +1147,7 @@ public void requestInterrupt() { CHECKED_MEM.putLong(ctxBufferAddr + CtxBuffer.INTERRUPT_FLAG, 1L); } - public void clearInterrupt() { + private void clearInterrupt() { CHECKED_MEM.putLong(ctxBufferAddr + CtxBuffer.INTERRUPT_FLAG, 0L); } diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java index b186cd845..d9ebd4ee6 100644 --- a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java +++ b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java @@ -7,11 +7,9 @@ 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.NativeMachineFactory; 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; @@ -48,14 +46,7 @@ public void aFlagRaisedMidCallDoesNotStopTheNextCall() { .build(); try (var instance = - NativeMachineFactory.builder(module) - .withImportValues(imports) - .withCompilerFunction( - m -> - NativeCompiler.compileAll( - RedlineTarget.detectHost().orElseThrow().triple(), - m)) - .build()) { + NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { machineRef[0] = (NativeMachine) instance.getMachine(); // Returns normally: the entry check ran before the flag was raised. @@ -97,17 +88,16 @@ public void aFlagRaisedInANestedCallDoesNotStopTheOuterCall() { machineRef[0].requestInterrupt(); return null; })) + .addFunction( + new HostFunction( + "host", + "tick", + FunctionType.of(List.of(), List.of()), + (inst, args) -> null)) .build(); try (var instance = - NativeMachineFactory.builder(module) - .withImportValues(imports) - .withCompilerFunction( - m -> - NativeCompiler.compileAll( - RedlineTarget.detectHost().orElseThrow().triple(), - m)) - .build()) { + NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { machineRef[0] = (NativeMachine) instance.getMachine(); assertEquals( diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java index 5bd340d76..aad760a64 100644 --- a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java +++ b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java @@ -2,19 +2,22 @@ import static java.util.concurrent.TimeUnit.SECONDS; 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 static org.junit.jupiter.api.Assertions.fail; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; 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.NativeMachineFactory; import run.endive.runtime.HostFunction; import run.endive.runtime.ImportValues; import run.endive.runtime.Instance; +import run.endive.testing.NativeInstanceBuilder; import run.endive.wasm.Parser; import run.endive.wasm.WasmEngineException; import run.endive.wasm.types.FunctionType; @@ -23,48 +26,51 @@ public class InterruptionTest { @Test public void shouldInterruptLoopViaThread() throws InterruptedException { - try (var instance = buildInstance("compiled/infinite-loop.c.wasm")) { - var function = instance.export("run"); - assertThreadInterruption(function::apply); - } + var instance = + buildInstance("compiled/infinite-loop.c.wasm", ImportValues.builder().build()); + var function = instance.export("run"); + assertThreadInterruption(function::apply, null, instance); } @Test public void shouldInterruptCallViaThread() throws InterruptedException { - try (var instance = buildInstance("compiled/power.c.wasm")) { - var function = instance.export("run"); - assertThreadInterruption(() -> function.apply(100)); - } + var instance = buildInstance("compiled/power.c.wasm", ImportValues.builder().build()); + var function = instance.export("run"); + assertThreadInterruption(() -> function.apply(100), null, instance); } @Test public void shouldInterruptNestedLoopViaThread() throws InterruptedException { - var imports = - ImportValues.builder() - .addFunction( - new HostFunction( - "host", - "reenter", - FunctionType.of(List.of(), List.of()), - (inst, args) -> { - inst.export("spin").apply(); - return null; - })) - .addFunction( - new HostFunction( - "host", - "raiseFlag", - FunctionType.of(List.of(), List.of()), - (inst, args) -> null)) - .build(); - try (var instance = buildInstance("compiled/reentrant-interrupt.wat.wasm", imports)) { - var function = instance.export("run"); - assertThreadInterruption(function::apply); - } + var running = new CountDownLatch(1); + var instance = + buildInstance( + "compiled/reentrant-interrupt.wat.wasm", + reentrantImports(inst -> inst.export("spin").apply(), running)); + var function = instance.export("run"); + assertThreadInterruption(function::apply, running, instance); + } + + @Test + public void shouldInterruptLoopInAnotherMachineViaThread() throws InterruptedException { + var running = new CountDownLatch(1); + var inner = + buildInstance( + "compiled/reentrant-interrupt.wat.wasm", + reentrantImports(inst -> {}, running)); + var outer = + buildInstance( + "compiled/reentrant-interrupt.wat.wasm", + reentrantImports(inst -> inner.export("spin").apply(), running)); + var function = outer.export("run"); + assertThreadInterruption(function::apply, running, outer, inner); } - private static void assertThreadInterruption(Runnable function) throws InterruptedException { + // interrupts once `running` is released, or after 100ms without one + private static void assertThreadInterruption( + Runnable function, CountDownLatch running, Instance... instances) + throws InterruptedException { AtomicBoolean interrupted = new AtomicBoolean(); + var failure = new AtomicReference(); Thread thread = new Thread( () -> { @@ -73,26 +79,63 @@ private static void assertThreadInterruption(Runnable function) throws Interrupt interrupted.set(true); }); thread.setDaemon(true); + thread.setUncaughtExceptionHandler( + (t, e) -> { + failure.set(e); + if (running != null) { + running.countDown(); + } + }); thread.start(); - Thread.sleep(100); + boolean started = true; + if (running == null) { + Thread.sleep(100); + } else { + started = running.await(10, SECONDS); + } thread.interrupt(); SECONDS.timedJoin(thread, 10); + if (failure.get() != null) { + fail("the call failed", failure.get()); + } + assertTrue(started, "the call never reached its loop"); + // a call still running uses the instances' memory, so they are leaked, not closed + assertFalse(thread.isAlive(), "the call was not interrupted"); + for (var instance : instances) { + instance.close(); + } assertTrue(interrupted.get()); } - private static Instance buildInstance(String resource) { - return buildInstance(resource, ImportValues.builder().build()); + private static ImportValues reentrantImports( + Consumer reenter, CountDownLatch running) { + var noParams = FunctionType.of(List.of(), List.of()); + return ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + noParams, + (inst, args) -> { + reenter.accept(inst); + return null; + })) + .addFunction(new HostFunction("host", "raiseFlag", noParams, (inst, args) -> null)) + .addFunction( + new HostFunction( + "host", + "tick", + noParams, + (inst, args) -> { + running.countDown(); + return null; + })) + .build(); } private static Instance buildInstance(String resource, ImportValues imports) { var module = Parser.parse(CorpusResources.getResource(resource)); - return NativeMachineFactory.builder(module) - .withImportValues(imports) - .withCompilerFunction( - m -> - NativeCompiler.compileAll( - RedlineTarget.detectHost().orElseThrow().triple(), m)) - .build(); + return NativeInstanceBuilder.builder(module).withImportValues(imports).build(); } } diff --git a/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java b/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java index 1793fc86a..0f26b8e70 100644 --- a/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java +++ b/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java @@ -1086,7 +1086,7 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { // nested calls run inside the outermost call's watch InterruptWatchdog.Registration watchdog = - outermostCall ? InterruptWatchdog.enter(Thread.currentThread(), this) : null; + outermostCall ? InterruptWatchdog.enter(this) : null; long result; try { result = (long) handle.invokeExact(cachedMemBase, ctxBuffer, args); @@ -1114,7 +1114,6 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { if (trapCode != 0) { ctxBuffer.set(ValueLayout.JAVA_INT, CtxBuffer.TRAP_CODE, 0); if (trapCode == CtxBuffer.TRAP_INTERRUPTED) { - ctxBuffer.set(ValueLayout.JAVA_LONG, CtxBuffer.INTERRUPT_FLAG, 0L); Thread.currentThread().interrupt(); } throw trapException(trapCode); @@ -1154,7 +1153,7 @@ public void requestInterrupt() { ctxBuffer.set(ValueLayout.JAVA_LONG, CtxBuffer.INTERRUPT_FLAG, 1L); } - public void clearInterrupt() { + private void clearInterrupt() { ctxBuffer.set(ValueLayout.JAVA_LONG, CtxBuffer.INTERRUPT_FLAG, 0L); } diff --git a/wasm-corpus/src/main/resources/compiled/reentrant-interrupt.wat.wasm b/wasm-corpus/src/main/resources/compiled/reentrant-interrupt.wat.wasm index 9a1d4e7b4dff316fba565aa5915c137d959f1841..623e23197373e38a9e88a01a01aa21c65b99e4e8 100644 GIT binary patch delta 145 zcmW-ZF%H5o5Cr#pXF@OqO=xJK25MR)Bz{023t3>17-XB8H{>Zif`5^unQdn0OFyb$ zQ%eA7s%oGHebk=2YQ2;DGiAyC`T$tCcZwx*dOA9_=QwTS@G zh6i8(y}0K&b)BgEdN_#GX92IO<>nA=_gtX6<81voKrVa4h?5qZoDw5c_ww*c+y8JK P?Z#_j0%eK3c+>@7g@_hl diff --git a/wasm-corpus/src/main/resources/wat/reentrant-interrupt.wat b/wasm-corpus/src/main/resources/wat/reentrant-interrupt.wat index 0bfe09120..ef7ff7fb7 100644 --- a/wasm-corpus/src/main/resources/wat/reentrant-interrupt.wat +++ b/wasm-corpus/src/main/resources/wat/reentrant-interrupt.wat @@ -2,6 +2,7 @@ (module (import "host" "reenter" (func $reenter)) (import "host" "raiseFlag" (func $raiseFlag)) + (import "host" "tick" (func $tick)) (func (export "run") (result i32) (local $i i32) @@ -15,5 +16,7 @@ (call $raiseFlag)) (func (export "spin") - (loop $l (br $l))) + (loop $l + (call $tick) + (br $l))) ) From 46849417a527d96562866599bd267cc0d82bad35 Mon Sep 17 00:00:00 2001 From: andreatp Date: Fri, 2 Oct 2026 11:04:55 +0100 Subject: [PATCH 4/5] Poll for interrupts every 100 ms, overridable by a system property --- .../redline/experimental/api/internal/InterruptWatchdog.java | 4 +++- .../experimental/api/internal/InterruptWatchdogTest.java | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java index 74a0fc199..4b191a5a4 100644 --- a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java +++ b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java @@ -15,7 +15,9 @@ public final class InterruptWatchdog { private static final Logger LOG = Logger.getLogger(InterruptWatchdog.class.getName()); - private static final long POLL_INTERVAL_NANOS = TimeUnit.MILLISECONDS.toNanos(1); + static final long POLL_INTERVAL_NANOS = + TimeUnit.MILLISECONDS.toNanos( + Math.max(1, Long.getLong("endive.redline.interruptPollMillis", 100))); // how long the poller waits without calls before it exits; tests shorten it static final AtomicLong IDLE_EXIT_NANOS = new AtomicLong(TimeUnit.MINUTES.toNanos(1)); diff --git a/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java b/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java index 17790b56f..0e5c63128 100644 --- a/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java +++ b/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java @@ -37,7 +37,7 @@ public void raisesTheFlagOfAnInterruptedCallerUntilExit() { flag.raised = false; // the caller is still interrupted, but no longer watched - long end = System.nanoTime() + MILLISECONDS.toNanos(50); + long end = System.nanoTime() + 3 * InterruptWatchdog.POLL_INTERVAL_NANOS; while (System.nanoTime() < end) { LockSupport.parkNanos(MILLISECONDS.toNanos(1)); } From f9dc3ca58a0df90968e081e9ef1f4220388d4b61 Mon Sep 17 00:00:00 2001 From: andreatp Date: Fri, 2 Oct 2026 12:37:42 +0100 Subject: [PATCH 5/5] Interrupt redline through the Thread API only and throw WasmInterruptedException --- .../api/internal/InterruptWatchdog.java | 26 +-- .../api/internal/InterruptWatchdogTest.java | 159 ------------------ redline/runner-jffi-tests/pom.xml | 10 ++ .../jffi/internal/InterruptFlagTest.java | 54 +++--- .../jffi/internal/InterruptionTest.java | 61 ++++++- .../internal/WatchdogRegistrationTest.java | 88 ---------- .../jffi/internal/JffiNativeMachine.java | 13 +- redline/runner-tests/pom.xml | 10 ++ .../runner/internal/InterruptFlagTest.java | 54 +++--- .../runner/internal/InterruptionTest.java | 61 ++++++- .../internal/WatchdogRegistrationTest.java | 88 ---------- .../runner/internal/NativeMachine.java | 13 +- 12 files changed, 204 insertions(+), 433 deletions(-) delete mode 100644 redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java delete mode 100644 redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java delete mode 100644 redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java diff --git a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java index 4b191a5a4..330c54e87 100644 --- a/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java +++ b/redline/api/src/main/java/run/endive/redline/experimental/api/internal/InterruptWatchdog.java @@ -5,7 +5,6 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.LockSupport; import java.util.logging.Level; import java.util.logging.Logger; @@ -15,12 +14,12 @@ public final class InterruptWatchdog { private static final Logger LOG = Logger.getLogger(InterruptWatchdog.class.getName()); - static final long POLL_INTERVAL_NANOS = - TimeUnit.MILLISECONDS.toNanos( - Math.max(1, Long.getLong("endive.redline.interruptPollMillis", 100))); + private static final long POLL_INTERVAL_NANOS = + millisProperty("endive.redline.interruptPollMillis", 100); - // how long the poller waits without calls before it exits; tests shorten it - static final AtomicLong IDLE_EXIT_NANOS = new AtomicLong(TimeUnit.MINUTES.toNanos(1)); + // 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; @@ -68,17 +67,8 @@ public static void exit(Registration registration) { STATE.getAndAdd(-CALL); } - /** Visible for testing. */ - public static int activeCount() { - return ACTIVE.size(); - } - - static boolean pollerRunning() { - return (STATE.get() & RUNNING) != 0; - } - - static boolean pollerIdle() { - return (STATE.get() & IDLE) != 0; + 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 @@ -132,7 +122,7 @@ private static boolean idleUntilExit() { if (!STATE.compareAndSet(RUNNING, RUNNING | IDLE)) { return false; } - long deadline = System.nanoTime() + IDLE_EXIT_NANOS.get(); + long deadline = System.nanoTime() + IDLE_EXIT_NANOS; long left; while ((STATE.get() & IDLE) != 0 && (left = deadline - System.nanoTime()) > 0) { LockSupport.parkNanos(left); diff --git a/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java b/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java deleted file mode 100644 index 0e5c63128..000000000 --- a/redline/api/src/test/java/run/endive/redline/experimental/api/internal/InterruptWatchdogTest.java +++ /dev/null @@ -1,159 +0,0 @@ -package run.endive.redline.experimental.api.internal; - -import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static java.util.concurrent.TimeUnit.SECONDS; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertSame; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import java.util.concurrent.atomic.AtomicReference; -import java.util.concurrent.locks.LockSupport; -import java.util.function.BooleanSupplier; -import java.util.logging.Handler; -import java.util.logging.LogRecord; -import java.util.logging.Logger; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Test; - -public class InterruptWatchdogTest { - - @AfterEach - public void clearInterruptStatus() { - Thread.interrupted(); - } - - @Test - public void raisesTheFlagOfAnInterruptedCallerUntilExit() { - var flag = new Flag(); - var registration = InterruptWatchdog.enter(flag); - try { - Thread.currentThread().interrupt(); - awaitTrue(() -> flag.raised); - } finally { - InterruptWatchdog.exit(registration); - } - - flag.raised = false; - // the caller is still interrupted, but no longer watched - long end = System.nanoTime() + 3 * InterruptWatchdog.POLL_INTERVAL_NANOS; - while (System.nanoTime() < end) { - LockSupport.parkNanos(MILLISECONDS.toNanos(1)); - } - assertFalse(flag.raised); - } - - @Test - public void aFailingSinkIsDroppedAndLogged() { - var failure = new IllegalStateException("closed"); - InterruptWatchdog.InterruptSink broken = - () -> { - throw failure; - }; - var logged = new AtomicReference(); - var log = Logger.getLogger(InterruptWatchdog.class.getName()); - var handler = - new Handler() { - @Override - public void publish(LogRecord record) { - logged.compareAndSet(null, record.getThrown()); - } - - @Override - public void flush() {} - - @Override - public void close() {} - }; - log.addHandler(handler); - log.setUseParentHandlers(false); - int before = InterruptWatchdog.activeCount(); - var brokenRegistration = InterruptWatchdog.enter(broken); - var flag = new Flag(); - var registration = InterruptWatchdog.enter(flag); - try { - Thread.currentThread().interrupt(); - awaitTrue(() -> flag.raised && logged.get() != null); - assertSame(failure, logged.get()); - assertEquals(before + 1, InterruptWatchdog.activeCount(), "broken call dropped"); - } finally { - InterruptWatchdog.exit(registration); - InterruptWatchdog.exit(brokenRegistration); - log.removeHandler(handler); - log.setUseParentHandlers(true); - } - } - - @Test - public void aCallWakesAnIdlePoller() { - var flag = new Flag(); - InterruptWatchdog.exit(InterruptWatchdog.enter(flag)); - awaitTrue(InterruptWatchdog::pollerIdle); - - var registration = InterruptWatchdog.enter(flag); - try { - long start = System.nanoTime(); - Thread.currentThread().interrupt(); - awaitTrue(() -> flag.raised); - // well before the idle poller would wake up on its own - assertTrue(System.nanoTime() - start < MILLISECONDS.toNanos(500)); - } finally { - InterruptWatchdog.exit(registration); - } - } - - @Test - public void thePollerExitsWhenIdleAndRestartsOnTheNextCall() { - long idleExit = InterruptWatchdog.IDLE_EXIT_NANOS.getAndSet(MILLISECONDS.toNanos(100)); - try { - var flag = new Flag(); - var first = new AtomicReference(); - var registration = InterruptWatchdog.enter(flag); - try { - awaitTrue(() -> first.updateAndGet(t -> pollerThread()) != null); - } finally { - InterruptWatchdog.exit(registration); - } - - awaitTrue(() -> !InterruptWatchdog.pollerRunning() && !first.get().isAlive()); - - registration = InterruptWatchdog.enter(flag); - try { - // the old poller is dead, so this one was started by the call above - assertNotNull(pollerThread()); - Thread.currentThread().interrupt(); - awaitTrue(() -> flag.raised); - } finally { - InterruptWatchdog.exit(registration); - } - } finally { - InterruptWatchdog.IDLE_EXIT_NANOS.set(idleExit); - } - } - - private static Thread pollerThread() { - return Thread.getAllStackTraces().keySet().stream() - .filter(t -> t.getName().equals("endive-redline-interrupt") && t.isAlive()) - .findFirst() - .orElse(null); - } - - // parkNanos, as the caller is often interrupted on purpose and cannot sleep - private static void awaitTrue(BooleanSupplier condition) { - long deadline = System.nanoTime() + SECONDS.toNanos(10); - while (!condition.getAsBoolean()) { - assertTrue(System.nanoTime() < deadline, "timed out"); - LockSupport.parkNanos(MILLISECONDS.toNanos(1)); - } - } - - private static final class Flag implements InterruptWatchdog.InterruptSink { - volatile boolean raised; - - @Override - public void requestInterrupt() { - raised = true; - } - } -} diff --git a/redline/runner-jffi-tests/pom.xml b/redline/runner-jffi-tests/pom.xml index e84f52570..6e7ce7dce 100644 --- a/redline/runner-jffi-tests/pom.xml +++ b/redline/runner-jffi-tests/pom.xml @@ -78,6 +78,16 @@ + + org.apache.maven.plugins + maven-surefire-plugin + + + + 200 + + + run.endive test-gen-plugin diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java index 9e6d14534..00f2e9af2 100644 --- a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java +++ b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptFlagTest.java @@ -1,5 +1,6 @@ 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; @@ -13,11 +14,7 @@ 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 @@ -27,11 +24,9 @@ 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( @@ -40,41 +35,36 @@ public void aFlagRaisedMidCallDoesNotStopTheNextCall() { "raiseFlag", FunctionType.of(List.of(), List.of()), (inst, args) -> { - machineRef[0].requestInterrupt(); + interruptAndHandle(); return null; })) .build(); try (var instance = NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { - machineRef[0] = (JffiNativeMachine) instance.getMachine(); - - // Returns normally: the entry check ran before the flag was raised. + // 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 aFlagRaisedInANestedCallDoesNotStopTheOuterCall() { + public void anInterruptHandledInANestedCallDoesNotStopTheOuterCall() { var module = Parser.parse(CorpusResources.getResource("compiled/reentrant-interrupt.wat.wasm")); - - var machineRef = new JffiNativeMachine[1]; + var noParams = FunctionType.of(List.of(), List.of()); var imports = ImportValues.builder() .addFunction( new HostFunction( "host", "reenter", - FunctionType.of(List.of(), List.of()), + noParams, (inst, args) -> { inst.export("raise").apply(); return null; @@ -83,27 +73,31 @@ public void aFlagRaisedInANestedCallDoesNotStopTheOuterCall() { new HostFunction( "host", "raiseFlag", - FunctionType.of(List.of(), List.of()), + noParams, (inst, args) -> { - machineRef[0].requestInterrupt(); + interruptAndHandle(); return null; })) .addFunction( - new HostFunction( - "host", - "tick", - FunctionType.of(List.of(), List.of()), - (inst, args) -> null)) + new HostFunction("host", "tick", noParams, (inst, args) -> null)) .build(); try (var instance = NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { - machineRef[0] = (JffiNativeMachine) instance.getMachine(); - assertEquals( 1000, (int) instance.export("run").apply()[0], - "a flag left over from the nested call must not stop the outer one"); + "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(); } } diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java index cb6e1e148..9d1b62e17 100644 --- a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java +++ b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/InterruptionTest.java @@ -1,12 +1,12 @@ package run.endive.redline.experimental.runner.jffi.internal; import static java.util.concurrent.TimeUnit.SECONDS; -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 static org.junit.jupiter.api.Assertions.fail; +import java.lang.management.ManagementFactory; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicBoolean; @@ -17,13 +17,18 @@ import run.endive.runtime.HostFunction; import run.endive.runtime.ImportValues; import run.endive.runtime.Instance; +import run.endive.runtime.WasmInterruptedException; import run.endive.testing.NativeInstanceBuilder; import run.endive.wasm.Parser; -import run.endive.wasm.WasmEngineException; import run.endive.wasm.types.FunctionType; public class InterruptionTest { + // well below the reentrant stack guard + private static final int DEPTH = 20; + + private static final int ROUNDS = 20; + @Test public void shouldInterruptLoopViaThread() throws InterruptedException { var instance = @@ -65,6 +70,55 @@ public void shouldInterruptLoopInAnotherMachineViaThread() throws InterruptedExc assertThreadInterruption(function::apply, running, outer, inner); } + @Test + public void shouldInterruptAfterTheWatchdogWentIdle() throws InterruptedException { + var running = new CountDownLatch(1); + var instance = + buildInstance( + "compiled/reentrant-interrupt.wat.wasm", + reentrantImports(inst -> inst.export("spin").apply(), running)); + instance.export("raise").apply(); + // longer than the watchdog's idle time in this module's test configuration + Thread.sleep(1000); + var function = instance.export("run"); + assertThreadInterruption(function::apply, running, instance); + } + + @Test + public void callingStartsNoThreads() { + var threads = ManagementFactory.getThreadMXBean(); + int[] depth = {0}; + var imports = + ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + FunctionType.of(List.of(), List.of()), + (inst, args) -> { + if (depth[0]++ < DEPTH) { + inst.export("recurse").apply(); + } + depth[0] = 0; + return null; + })) + .build(); + try (var instance = buildInstance("compiled/reentrant-recursion.wat.wasm", imports)) { + var recurse = instance.export("recurse"); + // starts the shared watchdog + recurse.apply(); + + long before = threads.getTotalStartedThreadCount(); + for (int i = 0; i < ROUNDS; i++) { + recurse.apply(); + } + long started = threads.getTotalStartedThreadCount() - before; + + // loose bound: JIT compiler threads may start meanwhile + assertTrue(started < ROUNDS, started + " threads started"); + } + } + // interrupts once `running` is released, or after 100ms without one private static void assertThreadInterruption( Runnable function, CountDownLatch running, Instance... instances) @@ -74,8 +128,7 @@ private static void assertThreadInterruption( Thread thread = new Thread( () -> { - var e = assertThrows(WasmEngineException.class, function::run); - assertEquals("interrupted", e.getMessage()); + assertThrows(WasmInterruptedException.class, function::run); interrupted.set(true); }); thread.setDaemon(true); diff --git a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java b/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java deleted file mode 100644 index 1f20fb6bb..000000000 --- a/redline/runner-jffi-tests/src/test/java/run/endive/redline/experimental/runner/jffi/internal/WatchdogRegistrationTest.java +++ /dev/null @@ -1,88 +0,0 @@ -package run.endive.redline.experimental.runner.jffi.internal; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import java.lang.management.ManagementFactory; -import java.util.List; -import java.util.function.Consumer; -import org.junit.jupiter.api.Test; -import run.endive.corpus.CorpusResources; -import run.endive.redline.experimental.api.internal.InterruptWatchdog; -import run.endive.runtime.HostFunction; -import run.endive.runtime.ImportValues; -import run.endive.runtime.Instance; -import run.endive.testing.NativeInstanceBuilder; -import run.endive.wasm.Parser; -import run.endive.wasm.types.FunctionType; - -/** Calls are watched by the shared poller, one registration per outermost call. */ -public class WatchdogRegistrationTest { - - // well below the reentrant stack guard, which fires near 115 - private static final int DEPTH = 20; - - private static final int ROUNDS = 20; - - @Test - public void nestingRegistersOnlyTheOutermostCall() { - int before = InterruptWatchdog.activeCount(); - int[] deepest = {0}; - withReentrantInstance( - instance -> instance.export("recurse").apply(), - () -> deepest[0] = Math.max(deepest[0], InterruptWatchdog.activeCount())); - - assertEquals(1, deepest[0] - before); - assertEquals(before, InterruptWatchdog.activeCount()); - } - - @Test - public void callingStartsNoThreads() { - var threads = ManagementFactory.getThreadMXBean(); - long[] started = {0}; - - withReentrantInstance( - instance -> { - // starts the shared poller - instance.export("recurse").apply(); - - long before = threads.getTotalStartedThreadCount(); - for (int i = 0; i < ROUNDS; i++) { - instance.export("recurse").apply(); - } - started[0] = threads.getTotalStartedThreadCount() - before; - }, - () -> {}); - - // loose bound: JIT compiler threads may start meanwhile - assertTrue(started[0] < ROUNDS, started[0] + " threads started"); - } - - private static void withReentrantInstance(Consumer body, Runnable atEachLevel) { - var module = - Parser.parse(CorpusResources.getResource("compiled/reentrant-recursion.wat.wasm")); - - int[] depth = {0}; - var imports = - ImportValues.builder() - .addFunction( - new HostFunction( - "host", - "reenter", - FunctionType.of(List.of(), List.of()), - (Instance inst, long... args) -> { - atEachLevel.run(); - if (depth[0]++ < DEPTH) { - inst.export("recurse").apply(); - } - depth[0] = 0; - return null; - })) - .build(); - - try (var instance = - NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { - body.accept(instance); - } - } -} diff --git a/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java b/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java index 950aa80e5..d5d570744 100644 --- a/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java +++ b/redline/runner-jffi/src/main/java/run/endive/redline/experimental/runner/jffi/internal/JffiNativeMachine.java @@ -24,6 +24,7 @@ import run.endive.runtime.Instance; import run.endive.runtime.Machine; import run.endive.runtime.TrapException; +import run.endive.runtime.WasmInterruptedException; import run.endive.runtime.WasmRuntimeException; import run.endive.wasm.WasmEngineException; import run.endive.wasm.types.FunctionType; @@ -42,7 +43,7 @@ * return: Wasm return value * */ -public final class JffiNativeMachine implements Machine, InterruptWatchdog.InterruptSink { +public final class JffiNativeMachine implements Machine { private static final int CTX_SIZE = CtxBuffer.CTX_SIZE; private static final MemoryIO MEM = MemoryIO.getInstance(); @@ -113,6 +114,7 @@ public final class JffiNativeMachine implements Machine, InterruptWatchdog.Inter private JffiNativeMemory nativeMemory; private volatile Throwable pendingException; private int callDepth; + private final InterruptWatchdog.InterruptSink interruptFlag = this::raiseInterruptFlag; // Keep closure handles alive to prevent GC private final Closure.Handle trampolineHandle; @@ -931,7 +933,7 @@ private static WasmEngineException trapException(int trapCode) { return new TrapException("unaligned atomic"); } if (trapCode == CtxBuffer.TRAP_INTERRUPTED) { - return new TrapException("interrupted"); + return new WasmInterruptedException("Thread interrupted"); } return new WasmEngineException("trap: unknown code " + trapCode); } @@ -1068,12 +1070,12 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { } if (Thread.currentThread().isInterrupted()) { - throw new TrapException("interrupted"); + throw new WasmInterruptedException("Thread interrupted"); } // nested calls run inside the outermost call's watch InterruptWatchdog.Registration watchdog = - outermostCall ? InterruptWatchdog.enter(this) : null; + outermostCall ? InterruptWatchdog.enter(interruptFlag) : null; long result; try { result = @@ -1142,8 +1144,7 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { } } - @Override - public void requestInterrupt() { + private void raiseInterruptFlag() { CHECKED_MEM.putLong(ctxBufferAddr + CtxBuffer.INTERRUPT_FLAG, 1L); } diff --git a/redline/runner-tests/pom.xml b/redline/runner-tests/pom.xml index 95a6b06be..399f1ddac 100644 --- a/redline/runner-tests/pom.xml +++ b/redline/runner-tests/pom.xml @@ -68,6 +68,16 @@ + + org.apache.maven.plugins + maven-surefire-plugin + + + + 200 + + + run.endive test-gen-plugin diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java index d9ebd4ee6..a77f9112b 100644 --- a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java +++ b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptFlagTest.java @@ -1,5 +1,6 @@ package run.endive.redline.experimental.runner.internal; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -13,11 +14,7 @@ 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 @@ -27,11 +24,9 @@ public void clearInterruptStatus() { } @Test - public void aFlagRaisedMidCallDoesNotStopTheNextCall() { + public void anInterruptHandledByTheHostDoesNotStopTheNextCall() { var module = Parser.parse(CorpusResources.getResource("compiled/interrupt-midcall.wat.wasm")); - - var machineRef = new NativeMachine[1]; var imports = ImportValues.builder() .addFunction( @@ -40,41 +35,36 @@ public void aFlagRaisedMidCallDoesNotStopTheNextCall() { "raiseFlag", FunctionType.of(List.of(), List.of()), (inst, args) -> { - machineRef[0].requestInterrupt(); + interruptAndHandle(); return null; })) .build(); try (var instance = NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { - machineRef[0] = (NativeMachine) instance.getMachine(); - - // Returns normally: the entry check ran before the flag was raised. + // 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 aFlagRaisedInANestedCallDoesNotStopTheOuterCall() { + public void anInterruptHandledInANestedCallDoesNotStopTheOuterCall() { var module = Parser.parse(CorpusResources.getResource("compiled/reentrant-interrupt.wat.wasm")); - - var machineRef = new NativeMachine[1]; + var noParams = FunctionType.of(List.of(), List.of()); var imports = ImportValues.builder() .addFunction( new HostFunction( "host", "reenter", - FunctionType.of(List.of(), List.of()), + noParams, (inst, args) -> { inst.export("raise").apply(); return null; @@ -83,27 +73,31 @@ public void aFlagRaisedInANestedCallDoesNotStopTheOuterCall() { new HostFunction( "host", "raiseFlag", - FunctionType.of(List.of(), List.of()), + noParams, (inst, args) -> { - machineRef[0].requestInterrupt(); + interruptAndHandle(); return null; })) .addFunction( - new HostFunction( - "host", - "tick", - FunctionType.of(List.of(), List.of()), - (inst, args) -> null)) + new HostFunction("host", "tick", noParams, (inst, args) -> null)) .build(); try (var instance = NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { - machineRef[0] = (NativeMachine) instance.getMachine(); - assertEquals( 1000, (int) instance.export("run").apply()[0], - "a flag left over from the nested call must not stop the outer one"); + "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(); } } diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java index aad760a64..d6ef31798 100644 --- a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java +++ b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/InterruptionTest.java @@ -1,12 +1,12 @@ package run.endive.redline.experimental.runner.internal; import static java.util.concurrent.TimeUnit.SECONDS; -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 static org.junit.jupiter.api.Assertions.fail; +import java.lang.management.ManagementFactory; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicBoolean; @@ -17,13 +17,18 @@ import run.endive.runtime.HostFunction; import run.endive.runtime.ImportValues; import run.endive.runtime.Instance; +import run.endive.runtime.WasmInterruptedException; import run.endive.testing.NativeInstanceBuilder; import run.endive.wasm.Parser; -import run.endive.wasm.WasmEngineException; import run.endive.wasm.types.FunctionType; public class InterruptionTest { + // well below the reentrant stack guard + private static final int DEPTH = 20; + + private static final int ROUNDS = 20; + @Test public void shouldInterruptLoopViaThread() throws InterruptedException { var instance = @@ -65,6 +70,55 @@ public void shouldInterruptLoopInAnotherMachineViaThread() throws InterruptedExc assertThreadInterruption(function::apply, running, outer, inner); } + @Test + public void shouldInterruptAfterTheWatchdogWentIdle() throws InterruptedException { + var running = new CountDownLatch(1); + var instance = + buildInstance( + "compiled/reentrant-interrupt.wat.wasm", + reentrantImports(inst -> inst.export("spin").apply(), running)); + instance.export("raise").apply(); + // longer than the watchdog's idle time in this module's test configuration + Thread.sleep(1000); + var function = instance.export("run"); + assertThreadInterruption(function::apply, running, instance); + } + + @Test + public void callingStartsNoThreads() { + var threads = ManagementFactory.getThreadMXBean(); + int[] depth = {0}; + var imports = + ImportValues.builder() + .addFunction( + new HostFunction( + "host", + "reenter", + FunctionType.of(List.of(), List.of()), + (inst, args) -> { + if (depth[0]++ < DEPTH) { + inst.export("recurse").apply(); + } + depth[0] = 0; + return null; + })) + .build(); + try (var instance = buildInstance("compiled/reentrant-recursion.wat.wasm", imports)) { + var recurse = instance.export("recurse"); + // starts the shared watchdog + recurse.apply(); + + long before = threads.getTotalStartedThreadCount(); + for (int i = 0; i < ROUNDS; i++) { + recurse.apply(); + } + long started = threads.getTotalStartedThreadCount() - before; + + // loose bound: JIT compiler threads may start meanwhile + assertTrue(started < ROUNDS, started + " threads started"); + } + } + // interrupts once `running` is released, or after 100ms without one private static void assertThreadInterruption( Runnable function, CountDownLatch running, Instance... instances) @@ -74,8 +128,7 @@ private static void assertThreadInterruption( Thread thread = new Thread( () -> { - var e = assertThrows(WasmEngineException.class, function::run); - assertEquals("interrupted", e.getMessage()); + assertThrows(WasmInterruptedException.class, function::run); interrupted.set(true); }); thread.setDaemon(true); diff --git a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java b/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java deleted file mode 100644 index fdeaaa205..000000000 --- a/redline/runner-tests/src/test/java/run/endive/redline/experimental/runner/internal/WatchdogRegistrationTest.java +++ /dev/null @@ -1,88 +0,0 @@ -package run.endive.redline.experimental.runner.internal; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import java.lang.management.ManagementFactory; -import java.util.List; -import java.util.function.Consumer; -import org.junit.jupiter.api.Test; -import run.endive.corpus.CorpusResources; -import run.endive.redline.experimental.api.internal.InterruptWatchdog; -import run.endive.runtime.HostFunction; -import run.endive.runtime.ImportValues; -import run.endive.runtime.Instance; -import run.endive.testing.NativeInstanceBuilder; -import run.endive.wasm.Parser; -import run.endive.wasm.types.FunctionType; - -/** Calls are watched by the shared poller, one registration per outermost call. */ -public class WatchdogRegistrationTest { - - // well below the reentrant stack guard, which fires near 115 - private static final int DEPTH = 20; - - private static final int ROUNDS = 20; - - @Test - public void nestingRegistersOnlyTheOutermostCall() { - int before = InterruptWatchdog.activeCount(); - int[] deepest = {0}; - withReentrantInstance( - instance -> instance.export("recurse").apply(), - () -> deepest[0] = Math.max(deepest[0], InterruptWatchdog.activeCount())); - - assertEquals(1, deepest[0] - before); - assertEquals(before, InterruptWatchdog.activeCount()); - } - - @Test - public void callingStartsNoThreads() { - var threads = ManagementFactory.getThreadMXBean(); - long[] started = {0}; - - withReentrantInstance( - instance -> { - // starts the shared poller - instance.export("recurse").apply(); - - long before = threads.getTotalStartedThreadCount(); - for (int i = 0; i < ROUNDS; i++) { - instance.export("recurse").apply(); - } - started[0] = threads.getTotalStartedThreadCount() - before; - }, - () -> {}); - - // loose bound: JIT compiler threads may start meanwhile - assertTrue(started[0] < ROUNDS, started[0] + " threads started"); - } - - private static void withReentrantInstance(Consumer body, Runnable atEachLevel) { - var module = - Parser.parse(CorpusResources.getResource("compiled/reentrant-recursion.wat.wasm")); - - int[] depth = {0}; - var imports = - ImportValues.builder() - .addFunction( - new HostFunction( - "host", - "reenter", - FunctionType.of(List.of(), List.of()), - (Instance inst, long... args) -> { - atEachLevel.run(); - if (depth[0]++ < DEPTH) { - inst.export("recurse").apply(); - } - depth[0] = 0; - return null; - })) - .build(); - - try (var instance = - NativeInstanceBuilder.builder(module).withImportValues(imports).build()) { - body.accept(instance); - } - } -} diff --git a/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java b/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java index 0f26b8e70..7051207fe 100644 --- a/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java +++ b/redline/runner/src/main/java/run/endive/redline/experimental/runner/internal/NativeMachine.java @@ -21,6 +21,7 @@ import run.endive.runtime.Instance; import run.endive.runtime.Machine; import run.endive.runtime.TrapException; +import run.endive.runtime.WasmInterruptedException; import run.endive.runtime.WasmRuntimeException; import run.endive.wasm.WasmEngineException; import run.endive.wasm.types.FunctionType; @@ -41,7 +42,7 @@ * *

See {@link CtxBuffer} for the full layout definition. */ -public final class NativeMachine implements Machine, InterruptWatchdog.InterruptSink { +public final class NativeMachine implements Machine { private static final int CTX_SIZE = CtxBuffer.CTX_SIZE; @@ -103,6 +104,7 @@ public final class NativeMachine implements Machine, InterruptWatchdog.Interrupt private NativeMemory nativeMemory; private volatile Throwable pendingException; private int callDepth; + private final InterruptWatchdog.InterruptSink interruptFlag = this::raiseInterruptFlag; private boolean ownsMemory; private boolean closed; @@ -973,7 +975,7 @@ private static WasmEngineException trapException(int trapCode) { case CtxBuffer.TRAP_INDIRECT_CALL_TYPE_MISMATCH -> new TrapException("indirect call type mismatch"); case CtxBuffer.TRAP_UNALIGNED_ATOMIC -> new TrapException("unaligned atomic"); - case CtxBuffer.TRAP_INTERRUPTED -> new TrapException("interrupted"); + case CtxBuffer.TRAP_INTERRUPTED -> new WasmInterruptedException("Thread interrupted"); default -> new WasmEngineException("trap: unknown code " + trapCode); }; } @@ -1081,12 +1083,12 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { } if (Thread.currentThread().isInterrupted()) { - throw new TrapException("interrupted"); + throw new WasmInterruptedException("Thread interrupted"); } // nested calls run inside the outermost call's watch InterruptWatchdog.Registration watchdog = - outermostCall ? InterruptWatchdog.enter(this) : null; + outermostCall ? InterruptWatchdog.enter(interruptFlag) : null; long result; try { result = (long) handle.invokeExact(cachedMemBase, ctxBuffer, args); @@ -1148,8 +1150,7 @@ public long[] call(int funcId, long[] args) throws WasmEngineException { } } - @Override - public void requestInterrupt() { + private void raiseInterruptFlag() { ctxBuffer.set(ValueLayout.JAVA_LONG, CtxBuffer.INTERRUPT_FLAG, 1L); }