我在汇编脚本中实现了一个同步屏障（注意我也在一个穷举式并行模拟器中对它进行了建模，结果在3个线程下没有发现无效状态）。下面是屏障代码：

@external("env", "w")
declare const w: i32;

const SREG_W:       usize = 0;   // i32 : number of workers
const SREG_CNT:     usize = 4;   // i32 : barrier counter  (atomic)
const SREG_EPOCH:   usize = 8;   // i32 : barrier epoch    (atomic)
const SREG_CMD:     usize = 12;  // i32 : command (-1 = stop)
const SREG_W0_ITER: usize = 16;  // i32 : w0 current iteration (atomic, debug)
const READY_BASE:   usize = 256; // i32[W] : result slot per worker

const CMD_TEST_BARRIER_PING: i32 = 0;

function W(): i32 {
  return atomic.load<i32>(SREG_W);
}

function barrierSync(): void {
  const total = W();
  const epoch = atomic.load<i32>(SREG_EPOCH);
  const prev  = i32.atomic.rmw.add(SREG_CNT, 1);
  const count = prev + 1;

  if (count < total) {
    while (atomic.wait<i32>(SREG_EPOCH, epoch, -1) === AtomicWaitResult.TIMED_OUT) {}
  } else {
    atomic.store<i32>(SREG_CNT, 0);
    atomic.store<i32>(SREG_EPOCH, ~epoch);
    atomic.notify(SREG_EPOCH, i32.MAX_VALUE);
  }
}

我的测试是在屏障上进行一万次迭代的循环；如果某个线程的迭代次数相对于工作线程0的当前迭代多出大约2次或更多，就中断。

最多到4 个线程时，我很难重现这个错误。从5个线程开始，大约90% 的情况会被触发。

下面是一个完整的MRE的链接，你只需要 pnpm i && pnpm build && pnpm serve 即可：https://github.com/hl037/barrier-bug-atomics

解决方案

设想两条线程如下交错执行：

Thread 1                        Thread 2

// thread 1 first call to barrierSync()
const epoch = atomic.load<i32>(SREG_EPOCH);    // 0
const prev  = i32.atomic.rmw.add(SREG_CNT, 1); // 0
const count = prev + 1;                        // 1

if (count < total) {                           // true

                                // thread 2 first call barrierSync()
                                const epoch = atomic.load<i32>(SREG_EPOCH);    // 0
                                const prev  = i32.atomic.rmw.add(SREG_CNT, 1); // 1
                                const count = prev + 1;                        // 2

                                if (count < total) {                           // false
                                } else {
                                    atomic.store<i32>(SREG_CNT, 0);
                                    atomic.store<i32>(SREG_EPOCH, ~epoch);     // SREG_EPOCH -> -1
                                // thread 2 is delayed

while (atomic.wait<i32>(SREG_EPOCH, epoch, -1) // NOT_EQUAL
      === AtomicWaitResult.TIMED_OUT) {}       // loop terminates

// thread 1 second call to barrierSync()

const epoch = atomic.load<i32>(SREG_EPOCH);    // -1
const prev  = i32.atomic.rmw.add(SREG_CNT, 1); // 0
const count = prev + 1;                        // 1

if (count < total) {                           // true
while (atomic.wait<i32>(SREG_EPOCH, epoch, -1) // waiting...

                                    // thread 2 resumes
                                    atomic.notify(SREG_EPOCH, i32.MAX_VALUE);

// thread 1 wakes up
      === AtomicWaitResult.TIMED_OUT) {}       // OK
// loop terminates and thread 1 returns from barrierSync(),
// even though thread 2 has not reached its second barrier.

不能以从 atomic.wait() 返回的 OK 就认为该值真的改变了。

C++ std::atomic::wait() 在唤醒时会重新检查其变量，如果仍然相等就重新进入休眠状态。看起来wasm的 atomic.wait() 似乎没有这种行为。

所以我会把

    while (atomic.wait<i32>(SREG_EPOCH, epoch, -1) === AtomicWaitResult.TIMED_OUT) {}

改成

    while (atomic.wait<i32>(SREG_EPOCH, epoch, -1) !== AtomicWaitResult.NOT_EQUAL) {}

这样，如果你因为通知而唤醒（atomic.wait() 返回 OK），你会再次调用 atomic.wait()。如果 epoch 实际上已经改变，则第二次调用将立即返回并得到 NOT_EQUAL，循环将退出。如果没有改变，你将再次进入睡眠；对 SREG_EPOCH 的下一次写入将紧随其后的是另一次通知，因此你将在那个时刻再次被唤醒。

这恰恰是你在一个允许虚假唤醒的模型中所期望的行为，其中 wait() 允许虚假唤醒。wasm规范本身似乎并不直接允许虚假唤醒，但上面描述的情形（在完成所需写入之后、通知之前加载并调用 wait()）具有类似的效果。