Skip to content

Runtime: fix lost wakeup from late-arriving atomicWaitBroken wait notify #1315

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
andreas-karlsson wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Runtime: fix lost wakeup from late-arriving atomicWaitBroken wait notify #1315

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f2166bb
Runtime: add regression test for atomicWait/atomicNotify late-arrival…
andreas-karlsson Jul 17, 2026
da5699e
Runtime: prevent late-arriving atomicWait from stealing a pending notify
andreas-karlsson Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 19 additions & 6 deletions runtime/src/main/java/com/dylibso/chicory/runtime/ByteArrayMemory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,14 @@ private static final class WaitState {
int waiterCount;
// The number of waiting threads that have been scheduled to wake up
int pendingWakeups;
// Bumped by every notify that issues wakeups. A waiter may only consume
// a pending wakeup minted while it was waiting (generation moved since
// it registered). Without this, a newly-arrived waiter could steal a
// wakeup meant for an already-parked thread. That thread would then
// re-check, find none, and sleep forever (lost wakeup).
// Only ever compared with ==/!=, never ordered, so wraparound is not
// a practical concern.
long generation;
}

private final Map<Integer, WaitState> waitStates;
Expand Down Expand Up @@ -146,9 +154,10 @@ private int waitOn(int address, BooleanSupplier condition, long timeout) {
}

state.waiterCount++;
final long arrivalGeneration = state.generation;
try {

while (state.pendingWakeups == 0) {
while (state.pendingWakeups == 0 || state.generation == arrivalGeneration) {
long remaining = deadline - System.nanoTime();
if (remaining <= 0) {
return 2; // timeout
Expand All @@ -165,11 +174,12 @@ private int waitOn(int address, BooleanSupplier condition, long timeout) {
return 0; // woken
} finally {

if (state.pendingWakeups > 0) {
// any thread leaving the try block is correct to consume
// a pending wakeup IF available:
if (state.pendingWakeups > 0 && state.generation != arrivalGeneration) {
// any thread leaving the try block is correct to consume a
// pending wakeup IF one was minted while it was waiting:
// ret 0 - woken thread correctly consumes a wakeup
// ret 2 - timeout can only occur with pendingWakeups == 0 so will not consume
// ret 2 - timeout cannot coincide with a consumable wakeup
// (the while loop would have exited instead)
// throw - isn't part of the wasm runtime but is semantically correct to consume
state.pendingWakeups--;
}
Expand Down Expand Up @@ -220,8 +230,11 @@ public int notify(int address, int maxThreads) {
toWake = Math.min(actualWaiters, maxThreads);
}

// Add pending wakeups - waiters will consume these
// Add pending wakeups - waiters will consume these. The generation
// bump marks every currently-registered waiter as eligible; threads
// arriving later must wait for the next notify.
state.pendingWakeups += toWake;
state.generation++;
assert (state.pendingWakeups <= state.waiterCount);

// It's always safe to notify all. In the wait routine we consume pendingWakeups
Expand Down
25 changes: 19 additions & 6 deletions runtime/src/main/java/com/dylibso/chicory/runtime/ByteBufferMemory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,14 @@ private static final class WaitState {
int waiterCount;
// The number of waiting threads that have been scheduled to wake up
int pendingWakeups;
// Bumped by every notify that issues wakeups. A waiter may only consume
// a pending wakeup minted while it was waiting (generation moved since
// it registered). Without this, a newly-arrived waiter could steal a
// wakeup meant for an already-parked thread. That thread would then
// re-check, find none, and sleep forever (lost wakeup).
// Only ever compared with ==/!=, never ordered, so wraparound is not
// a practical concern.
long generation;
}

private final Map<Integer, WaitState> waitStates;
Expand Down Expand Up @@ -117,9 +125,10 @@ private int waitOn(int address, BooleanSupplier condition, long timeout) {
}

state.waiterCount++;
final long arrivalGeneration = state.generation;
try {

while (state.pendingWakeups == 0) {
while (state.pendingWakeups == 0 || state.generation == arrivalGeneration) {
long remaining = deadline - System.nanoTime();
if (remaining <= 0) {
return 2; // timeout
Expand All @@ -136,11 +145,12 @@ private int waitOn(int address, BooleanSupplier condition, long timeout) {
return 0; // woken
} finally {

if (state.pendingWakeups > 0) {
// any thread leaving the try block is correct to consume
// a pending wakeup IF available:
if (state.pendingWakeups > 0 && state.generation != arrivalGeneration) {
// any thread leaving the try block is correct to consume a
// pending wakeup IF one was minted while it was waiting:
// ret 0 - woken thread correctly consumes a wakeup
// ret 2 - timeout can only occur with pendingWakeups == 0 so will not consume
// ret 2 - timeout cannot coincide with a consumable wakeup
// (the while loop would have exited instead)
// throw - isn't part of the wasm runtime but is semantically correct to consume
state.pendingWakeups--;
}
Expand Down Expand Up @@ -191,8 +201,11 @@ public int notify(int address, int maxThreads) {
toWake = Math.min(actualWaiters, maxThreads);
}

// Add pending wakeups - waiters will consume these
// Add pending wakeups - waiters will consume these. The generation
// bump marks every currently-registered waiter as eligible; threads
// arriving later must wait for the next notify.
state.pendingWakeups += toWake;
state.generation++;
assert (state.pendingWakeups <= state.waiterCount);

// It's always safe to notify all. In the wait routine we consume pendingWakeups
Expand Down
127 changes: 127 additions & 0 deletions runtime/src/test/java/com/dylibso/chicory/runtime/MemoryStressTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
package com.dylibso.chicory.runtime;

import static org.junit.jupiter.api.Assertions.assertFalse;

import com.dylibso.chicory.wasm.types.MemoryLimits;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Stream;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;

public class MemoryStressTest {

static final class RwLock {
private static final int WRITER_BIT = 0x80000000;
private static final int READER_MASK = 0x7FFFFFFF;

private final Memory memory;
private final int addr;

public RwLock(Memory memory, int addr) {
this.memory = memory;
this.addr = addr;
}

public void lock() {
while (true) {
int s = memory.atomicReadInt(addr);
if ((s & WRITER_BIT) != 0) {
memory.atomicWait(addr, s, -1L);
continue;
}
if (memory.atomicCmpxchgInt(addr, s, s | WRITER_BIT) == s) {
break;
}
}
while (true) {
int s = memory.atomicReadInt(addr);
if ((s & READER_MASK) == 0) {
return;
}
memory.atomicWait(addr, s, -1L);
}
}

public void unlock() {
memory.atomicWriteInt(addr, 0);
memory.atomicNotify(addr, -1);
}

public void lockShared() {
while (true) {
int s = memory.atomicReadInt(addr);
if ((s & WRITER_BIT) != 0) {
memory.atomicWait(addr, s, -1L);
continue;
}
if (memory.atomicCmpxchgInt(addr, s, s + 1) == s) {
return;
}
}
}

public void unlockShared() {
int prev = memory.atomicAddInt(addr, -1);
if ((prev & READER_MASK) == 1 && (prev & WRITER_BIT) != 0) {
memory.atomicNotify(addr, -1);
}
}
}

private static Stream<Arguments> sharedMemoryImplementations() {
return Stream.of(
Arguments.of(
"ByteArrayMemory",
(Supplier<Memory>) () -> new ByteArrayMemory(new MemoryLimits(1, 1, true))),
Arguments.of(
"ByteBufferMemory",
(Supplier<Memory>)
() -> new ByteBufferMemory(new MemoryLimits(1, 1, true))));
}

@ParameterizedTest(name = "{0}")
@MethodSource("sharedMemoryImplementations")
public void rwLockSurvivesContention(String name, Supplier<Memory> memorySupplier)
throws InterruptedException {
final RwLock lock = new RwLock(memorySupplier.get(), 0);
final long deadline = System.currentTimeMillis() + 3_000;

final List<Thread> threads = new ArrayList<Thread>();
for (int r = 0; r < 6; r++) {
threads.add(
new Thread(
() -> {
while (System.currentTimeMillis() < deadline) {
lock.lockShared();
lock.unlockShared();
}
},
"reader-" + r));
}
threads.add(
new Thread(
() -> {
while (System.currentTimeMillis() < deadline) {
lock.lock();
lock.unlock();
}
},
"writer"));

for (Thread t : threads) {
t.setDaemon(true);
t.start();
}
final long joinDeadline = deadline + 1000;
for (Thread t : threads) {
final long remaining = Long.max(10, joinDeadline - System.currentTimeMillis());
t.join(remaining);
}

final boolean anyThreadBlocked = threads.stream().anyMatch(Thread::isAlive);
assertFalse(anyThreadBlocked);
}
}
51 changes: 51 additions & 0 deletions runtime/src/test/java/com/dylibso/chicory/runtime/MemoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
package com.dylibso.chicory.runtime;

import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;

import com.dylibso.chicory.wasm.types.MemoryLimits;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import java.util.stream.Stream;
Expand Down Expand Up @@ -165,4 +167,53 @@ public void nonOverlappingCopyCrossPage(String name, Supplier<Memory> memorySupp
// Source should be unchanged
assertArrayEquals(data, memory.readBytes(100, data.length));
}

/**
* This test verifies a permit stealing bug that existed in the memory implementations.
* An atomicWait could avoid waiting if it found a "notify permit". This is wrong by definition
* since atomicNotify can only wake threads in a waiting state.
*
* The test creates a thread that calls (A) atomicWait, while the main thread calls atomicNotify and
* then (B) atomicWait. If the notify call returns one, it means it should have woken (A) since
* (B) has not yet happened. This means that (B) must timeout since there are no other notify calls
* involved.
*/
@ParameterizedTest(name = "{0}")
@MethodSource("growableMemoryImplementations")
public void notifyWakesParkedThreadNotLateArrival(String name, Supplier<Memory> memorySupplier)
throws Exception {
final int addr = 0;
final int sentinel = 42;
var memory = memorySupplier.get();

for (int round = 0; round < 200; round++) {
memory.atomicWriteInt(addr, sentinel);

var parkedReady = new CountDownLatch(1);

var parked =
new Thread(
() -> {
parkedReady.countDown();
memory.atomicWait(addr, sentinel, 100_000_000L);
});
parked.setDaemon(true);
parked.start();
parkedReady.await();
// This sleep gives the other thread a higher chance of reaching the atomicWait,
// but setting it too high makes a passing test slow (200 x sleep)
Thread.sleep(1);

int woken = memory.atomicNotify(addr, 1);
if (woken == 0) {
// we didn't wake up the other thread. This case is not interesting, retry
continue;
}

// Late arrival: this wait MUST timeout as we have only woken ONE thread.
final int result = memory.atomicWait(addr, sentinel, 0L);

assertEquals(2, result, "round " + round + ": main thread should have timed out (2)");
}
}
}
Loading