Fix Redline native memory growth, atomics, wait/notify and traps

redline/api/src/main/java/run/endive/redline/experimental/api/NativeCodeSerializer.java

@@ -13,7 +13,7 @@
  * <p>Format:
  * <pre>
  *   [4 bytes: magic "CL4J"]
- *   [4 bytes: version (2)]
+ *   [4 bytes: version (3)]
  *   [modified UTF-8: target triple]
  *   [4 bytes: image length]
  *   [N bytes: image]
@@ -30,7 +30,7 @@
 public final class NativeCodeSerializer {
 
     private static final int MAGIC = 0x434C344A; // "CL4J"
-    private static final int VERSION = 2;
+    private static final int VERSION = 3;
 
     private NativeCodeSerializer() {}
 

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

@@ -19,6 +19,7 @@
  *  56     i64    memmovePtr        Pointer to libc memmove
  *  64     i64    memsetPtr         Pointer to libc memset
  *  72     i64    interruptFlag     Non-zero = interrupt requested
+ *  80     i64    memoryPagesPtr    Shared memory: pointer to its current page count (i32)
  * 200     i64    globalsPtr        Pointer to globals buffer
  * 208     i64    memGrowPtr        Upcall stub for memory.grow
  * 216     i32    memoryPages       Current memory page count
@@ -52,6 +53,7 @@ private CtxBuffer() {}
     public static final int MEMMOVE_PTR = 56;
     public static final int MEMSET_PTR = 64;
     public static final int INTERRUPT_FLAG = 72;
+    public static final int MEMORY_PAGES_PTR = 80;
 
     public static final int GLOBALS_PTR = 200;
     public static final int MEM_GROW_PTR = 208;

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

@@ -0,0 +1,215 @@
+package run.endive.redline.experimental.api.internal;
+
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+/** Hardware atomics on a native memory's mapping, via a direct buffer; callers check bounds. */
+public final class NativeAtomics {
+
+    private static final VarHandle INT =
+            MethodHandles.byteBufferViewVarHandle(int[].class, ByteOrder.LITTLE_ENDIAN);
+    private static final VarHandle LONG =
+            MethodHandles.byteBufferViewVarHandle(long[].class, ByteOrder.LITTLE_ENDIAN);
+
+    private static final int ADD = 0;
+    private static final int AND = 1;
+    private static final int OR = 2;
+    private static final int XOR = 3;
+    private static final int XCHG = 4;
+
+    private final ByteBuffer buffer;
+
+    public NativeAtomics(ByteBuffer buffer) {
+        this.buffer = buffer;
+    }
+
+    public int readInt(int addr) {
+        return (int) INT.getVolatile(buffer, addr);
+    }
+
+    public long readLong(int addr) {
+        return (long) LONG.getVolatile(buffer, addr);
+    }
+
+    public short readShort(int addr) {
+        return (short) narrowGet(addr, 16);
+    }
+
+    public byte readByte(int addr) {
+        return (byte) narrowGet(addr, 8);
+    }
+
+    public void writeInt(int addr, int value) {
+        INT.setVolatile(buffer, addr, value);
+    }
+
+    public void writeLong(int addr, long value) {
+        LONG.setVolatile(buffer, addr, value);
+    }
+
+    public void writeShort(int addr, short value) {
+        narrowUpdate(addr, 16, XCHG, value);
+    }
+
+    public void writeByte(int addr, byte value) {
+        narrowUpdate(addr, 8, XCHG, value);
+    }
+
+    public int addInt(int addr, int delta) {
+        return (int) INT.getAndAdd(buffer, addr, delta);
+    }
+
+    public int andInt(int addr, int mask) {
+        return (int) INT.getAndBitwiseAnd(buffer, addr, mask);
+    }
+
+    public int orInt(int addr, int mask) {
+        return (int) INT.getAndBitwiseOr(buffer, addr, mask);
+    }
+
+    public int xorInt(int addr, int mask) {
+        return (int) INT.getAndBitwiseXor(buffer, addr, mask);
+    }
+
+    public int xchgInt(int addr, int value) {
+        return (int) INT.getAndSet(buffer, addr, value);
+    }
+
+    public int cmpxchgInt(int addr, int expected, int replacement) {
+        return (int) INT.compareAndExchange(buffer, addr, expected, replacement);
+    }
+
+    public long addLong(int addr, long delta) {
+        return (long) LONG.getAndAdd(buffer, addr, delta);
+    }
+
+    public long andLong(int addr, long mask) {
+        return (long) LONG.getAndBitwiseAnd(buffer, addr, mask);
+    }
+
+    public long orLong(int addr, long mask) {
+        return (long) LONG.getAndBitwiseOr(buffer, addr, mask);
+    }
+
+    public long xorLong(int addr, long mask) {
+        return (long) LONG.getAndBitwiseXor(buffer, addr, mask);
+    }
+
+    public long xchgLong(int addr, long value) {
+        return (long) LONG.getAndSet(buffer, addr, value);
+    }
+
+    public long cmpxchgLong(int addr, long expected, long replacement) {
+        return (long) LONG.compareAndExchange(buffer, addr, expected, replacement);
+    }
+
+    public short addShort(int addr, short delta) {
+        return (short) narrowUpdate(addr, 16, ADD, delta);
+    }
+
+    public short andShort(int addr, short mask) {
+        return (short) narrowUpdate(addr, 16, AND, mask);
+    }
+
+    public short orShort(int addr, short mask) {
+        return (short) narrowUpdate(addr, 16, OR, mask);
+    }
+
+    public short xorShort(int addr, short mask) {
+        return (short) narrowUpdate(addr, 16, XOR, mask);
+    }
+
+    public short xchgShort(int addr, short value) {
+        return (short) narrowUpdate(addr, 16, XCHG, value);
+    }
+
+    public short cmpxchgShort(int addr, short expected, short replacement) {
+        return (short) narrowCmpxchg(addr, 16, expected, replacement);
+    }
+
+    public byte addByte(int addr, byte delta) {
+        return (byte) narrowUpdate(addr, 8, ADD, delta);
+    }
+
+    public byte andByte(int addr, byte mask) {
+        return (byte) narrowUpdate(addr, 8, AND, mask);
+    }
+
+    public byte orByte(int addr, byte mask) {
+        return (byte) narrowUpdate(addr, 8, OR, mask);
+    }
+
+    public byte xorByte(int addr, byte mask) {
+        return (byte) narrowUpdate(addr, 8, XOR, mask);
+    }
+
+    public byte xchgByte(int addr, byte value) {
+        return (byte) narrowUpdate(addr, 8, XCHG, value);
+    }
+
+    public byte cmpxchgByte(int addr, byte expected, byte replacement) {
+        return (byte) narrowCmpxchg(addr, 8, expected, replacement);
+    }
+
+    private int narrowGet(int addr, int bits) {
+        int shift = shift(addr, bits);
+        return ((int) INT.getVolatile(buffer, addr & ~3) >>> shift) & ((1 << bits) - 1);
+    }
+
+    // Byte and short have no atomic VarHandle modes: CAS the enclosing aligned int.
+    private int narrowUpdate(int addr, int bits, int op, int operand) {
+        int base = addr & ~3;
+        int shift = shift(addr, bits);
+        int mask = ((1 << bits) - 1) << shift;
+        while (true) {
+            int word = (int) INT.getVolatile(buffer, base);
+            int old = (word & mask) >>> shift;
+            int updated = (word & ~mask) | ((apply(op, old, operand) << shift) & mask);
+            if (INT.compareAndSet(buffer, base, word, updated)) {
+                return old;
+            }
+        }
+    }
+
+    private int narrowCmpxchg(int addr, int bits, int expected, int replacement) {
+        int base = addr & ~3;
+        int shift = shift(addr, bits);
+        int valueMask = (1 << bits) - 1;
+        int mask = valueMask << shift;
+        while (true) {
+            int word = (int) INT.getVolatile(buffer, base);
+            int old = (word & mask) >>> shift;
+            if (old != (expected & valueMask)) {
+                return old;
+            }
+            int updated = (word & ~mask) | ((replacement & valueMask) << shift);
+            if (INT.compareAndSet(buffer, base, word, updated)) {
+                return old;
+            }
+        }
+    }
+
+    private static int shift(int addr, int bits) {
+        if ((addr & (bits / 8 - 1)) != 0) {
+            throw new IllegalStateException("Misaligned atomic access at address " + addr);
+        }
+        return (addr & 3) * 8;
+    }
+
+    private static int apply(int op, int old, int operand) {
+        switch (op) {
+            case ADD:
+                return old + operand;
+            case AND:
+                return old & operand;
+            case OR:
+                return old | operand;
+            case XOR:
+                return old ^ operand;
+            default:
+                return operand;
+        }
+    }
+}

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

@@ -0,0 +1,100 @@
+package run.endive.redline.experimental.api.internal;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BooleanSupplier;
+import run.endive.runtime.WasmInterruptedException;
+import run.endive.wasm.WasmEngineException;
+
+/** memory.atomic.wait and notify for a native memory, as ByteArrayMemory implements them. */
+public final class NativeWaiters {
+
+    // Guarded by its own monitor; 0 <= pendingWakeups <= waiterCount
+    private static final class WaitState {
+        int waiterCount;
+        int pendingWakeups;
+        // a waiter only consumes a wakeup minted after it registered
+        long generation;
+    }
+
+    private final Map<Integer, WaitState> waitStates = new ConcurrentHashMap<>();
+    private final boolean shared;
+
+    public NativeWaiters(boolean shared) {
+        this.shared = shared;
+    }
+
+    public Object monitor(int address) {
+        return waitStates.computeIfAbsent(address, k -> new WaitState());
+    }
+
+    /** Waits while {@code condition} holds: 0 woken, 1 not-equal, 2 timed out. */
+    public int waitOn(int address, BooleanSupplier condition, long timeout) {
+        if (!shared) {
+            throw new WasmEngineException("Attempt to wait on a non-shared memory, not supported.");
+        }
+
+        long deadline = (timeout < 0) ? Long.MAX_VALUE : System.nanoTime() + timeout;
+        WaitState state = waitStates.computeIfAbsent(address, k -> new WaitState());
+
+        synchronized (state) {
+            if (!condition.getAsBoolean()) {
+                return 1;
+            }
+
+            state.waiterCount++;
+            final long arrivalGeneration = state.generation;
+            try {
+                while (state.pendingWakeups == 0 || state.generation == arrivalGeneration) {
+                    long remaining = deadline - System.nanoTime();
+                    if (remaining <= 0) {
+                        return 2;
+                    }
+                    long millis = Math.max(remaining / 1_000_000L, 0);
+                    int nanos = Math.max((int) (remaining % 1_000_000L), 0);
+                    try {
+                        state.wait(millis, nanos);
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                        throw new WasmInterruptedException("Thread interrupted");
+                    }
+                }
+                return 0;
+            } finally {
+                // Only a wakeup minted while this thread waited is its to consume.
+                if (state.pendingWakeups > 0 && state.generation != arrivalGeneration) {
+                    state.pendingWakeups--;
+                }
+                state.waiterCount--;
+                assert (0 <= state.pendingWakeups);
+                assert (state.pendingWakeups <= state.waiterCount);
+            }
+        }
+    }
+
+    /** Wakes up to {@code maxThreads} waiters (all if negative); returns how many. */
+    public int notify(int address, int maxThreads) {
+        if (!shared) {
+            return 0;
+        }
+
+        WaitState state = waitStates.get(address);
+        if (state == null) {
+            return 0;
+        }
+
+        synchronized (state) {
+            int actualWaiters = state.waiterCount - state.pendingWakeups;
+            if (actualWaiters == 0) {
+                return 0;
+            }
+            int toWake = maxThreads < 0 ? actualWaiters : Math.min(actualWaiters, maxThreads);
+            // waiters arriving after this notify wait for the next one
+            state.pendingWakeups += toWake;
+            state.generation++;
+            assert (state.pendingWakeups <= state.waiterCount);
+            state.notifyAll();
+            return toWake;
+        }
+    }
+}

redline/compiler/src/main/java/run/endive/redline/experimental/compiler/internal/EmitContext.java

@@ -25,6 +25,7 @@ final class EmitContext {
     final Map<String, Integer> sigRefCache;
     final boolean multiReturn;
     final int[] canonicalTypeMap;
+    final boolean sharedMemory;
 
     EmitContext(
             CraneliftBridge bridge,
@@ -37,7 +38,8 @@ final class EmitContext {
             int ctxPtrVar,
             Map<String, Integer> sigRefCache,
             boolean multiReturn,
-            int[] canonicalTypeMap) {
+            int[] canonicalTypeMap,
+            boolean sharedMemory) {
         this.bridge = bridge;
         this.valueStack = valueStack;
         this.module = module;
@@ -49,6 +51,19 @@ final class EmitContext {
         this.sigRefCache = sigRefCache;
         this.multiReturn = multiReturn;
         this.canonicalTypeMap = canonicalTypeMap;
+        this.sharedMemory = sharedMemory;
+    }
+
+    /** The current page count; a shared memory's lives with the memory, as others may grow it. */
+    int emitLoadMemoryPages() {
+        var b = bridge.exports();
+        int zero = b.emitIconst32(0);
+        int ctxPtr = b.useVar(ctxPtrVar);
+        if (!sharedMemory) {
+            return b.emitLoadI32(ctxPtr, zero, CtxBuffer.MEMORY_PAGES);
+        }
+        int pagesPtr = b.emitLoadI64(ctxPtr, zero, CtxBuffer.MEMORY_PAGES_PTR);
+        return b.emitLoadI32(pagesPtr, zero, 0);
     }
 
     // --- Helpers used by emitters ---