single: a woken fiber waits for the waker to park; writev/readv block

Every system call bumps the wake counter, and in the single-thread build
each wake resolved the parked fibers at once. Their continuations then
ran ahead of the waker's own return from the call (its syscall is a
promise too), so a writer and its reader traded places on every write:
`yes | head -c 20M | wc -c` made 48.9k parks and 29.3k wakes. Sched now
keeps the woken in a ready list until the running fiber parks, calls
sched_yield, or the event loop turns (posted on every wake, so a wake
from outside any fiber still goes at once). The same pipeline: 2528
parks, 1809 wakes.

Filling the pipe exposed a kernel bug: writev to a full blocking pipe
answered EAGAIN, as readv on an empty one did, because blocking_op knew
only write and read. coreutils' head writes with writev and died
"Resource temporarily unavailable". Both now park on their first
non-empty vector and finish with a short count (kernel test
vectored_io_on_a_blocked_pipe_parks).

check-glue reads the `single.on ? new WebAssembly.Suspending((...) =>`
spelling of an import.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_781bad2ec05143db9067c86b4b2678e2

crates/gex-kernel/src/syscalls.rs

@@ -2257,6 +2257,49 @@ impl Kernel {
                     len: a[2].min(1 << 24) as usize,
                 })
             }
+            // `writev` and `readv` on a stream block as `write` and `read` do,
+            // on their first non-empty vector: the parked call is finished
+            // with that vector alone, a short count the caller already loops
+            // on. Without this a full pipe answered `writev` with `EAGAIN` on
+            // a blocking descriptor (coreutils' `head` writes with it, and
+            // died "Resource temporarily unavailable" once its reader was
+            // slower than it).
+            nr::WRITEV => {
+                let file = self.fds.get(a[0] as i32).ok()?;
+                if !file.write_is_atomic() {
+                    return None;
+                }
+                let (buf, len) = Self::read_iovecs(m, a[1], a[2], self.compat)
+                    .ok()?
+                    .into_iter()
+                    .find(|v| v.1 > 0)?;
+                match file.handle.write_room() {
+                    Some(room) if len <= room => return None,
+                    Some(_) => {}
+                    None if !file.write_would_block() => return None,
+                    None => {}
+                }
+                Some(BlockOp::WriteStream {
+                    handle: Arc::clone(&file.handle),
+                    buf,
+                    len: len.min(1 << 24) as usize,
+                })
+            }
+            nr::READV => {
+                let file = self.fds.get(a[0] as i32).ok()?;
+                if !file.would_block() {
+                    return None;
+                }
+                let (buf, len) = Self::read_iovecs(m, a[1], a[2], self.compat)
+                    .ok()?
+                    .into_iter()
+                    .find(|v| v.1 > 0)?;
+                Some(BlockOp::ReadStream {
+                    handle: Arc::clone(&file.handle),
+                    buf,
+                    len: len.min(1 << 24) as usize,
+                })
+            }
             nr::ACCEPT | nr::ACCEPT4 => {
                 let file = self.fds.get(a[0] as i32).ok()?;
                 let socket: Arc<dyn crate::vfs::FileHandle> = Arc::clone(&file.handle);
@@ -7533,6 +7576,36 @@ mod tests {
         }
     }
 
+    /// `writev` to a full blocking pipe parks on its first non-empty vector,
+    /// as `write` does, and `readv` on an empty one parks as `read` does --
+    /// they had answered `EAGAIN` on a blocking descriptor.
+    #[test]
+    fn vectored_io_on_a_blocked_pipe_parks() {
+        let (mut mem, mut k) = setup();
+        let p = Arc::new(Pipe::new());
+        let end = |write| OpenFile::new(Arc::new(crate::file::PipeEnd { pipe: Arc::clone(&p), write }), "pipe".into(), 0, !write, write);
+        let rfd = k.fds.insert(end(false), false).unwrap() as u64;
+        let wfd = k.fds.insert(end(true), false).unwrap() as u64;
+        // Two vectors, the first empty: 0x3000 (len 0) and 0x3100 (len 100).
+        let iov = 0x2000usize;
+        for (i, (b, l)) in [(0x3000u64, 0u64), (0x3100, 100)].into_iter().enumerate() {
+            mem[iov + 16 * i..iov + 16 * i + 8].copy_from_slice(&b.to_le_bytes());
+            mem[iov + 16 * i + 8..iov + 16 * i + 16].copy_from_slice(&l.to_le_bytes());
+        }
+        let op = |k: &mut Kernel, mem: &mut Vec<u8>, n| k.blocking_op(&GuestMem::new(mem), n, [if n == nr::READV { rfd } else { wfd }, iov as u64, 2, 0, 0, 0]);
+        match op(&mut k, &mut mem, nr::READV) {
+            Some(BlockOp::ReadStream { buf: 0x3100, len: 100, .. }) => {}
+            other => panic!("readv on an empty pipe: {other:?}"),
+        }
+        assert!(op(&mut k, &mut mem, nr::WRITEV).is_none(), "writev to an empty pipe goes straight through");
+        p.write(&vec![0u8; crate::pipe::PIPE_BUF]).unwrap();
+        match op(&mut k, &mut mem, nr::WRITEV) {
+            Some(BlockOp::WriteStream { buf: 0x3100, len: 100, .. }) => {}
+            other => panic!("writev to a full pipe: {other:?}"),
+        }
+        assert!(op(&mut k, &mut mem, nr::READV).is_none(), "readv on a full pipe goes straight through");
+    }
+
     #[test]
     fn clock_nanosleep_honours_absolute_deadlines() {
         let (mut mem, mut k) = setup();

shims/browser/js/check-glue.mjs

@@ -124,8 +124,10 @@ function suppliedArities(source) {
   }
   const body = source.slice(at, end);
   const out = new Map();
-  // `name: (a, b) => ...` and `name: () => ...`, at any indentation.
-  for (const m of body.matchAll(/([A-Za-z_][A-Za-z0-9_]*)\s*:\s*\(([^)]*)\)\s*=>/g)) {
+  // `name: (a, b) => ...` and `name: () => ...`, at any indentation, and the
+  // single-thread build's `name: single.on ? new WebAssembly.Suspending((a, b)
+  // => ...) : (a, b) => ...`, read by its first arrow (both take the same).
+  for (const m of body.matchAll(/([A-Za-z_][A-Za-z0-9_]*)\s*:\s*(?:single\.on\s*\?\s*new WebAssembly\.Suspending\(\s*)?\(([^)]*)\)\s*=>/g)) {
     const params = m[2].trim();
     out.set(m[1], params ? params.split(',').length : 0);
   }

shims/browser/js/gex.mjs

@@ -1086,8 +1086,9 @@ export class Gex {
           : (ctx, target, ret, spWant) => this.#callOut(ctx >>> 0, target, ret, spWant),
         // An AOT-only fallback whose target this Worker has just installed
         // (cgen.rs `gex_ftab_fallback`): called as a function body.
+        // In the single build, on a stack of its own, so the body (and guest
+        // code it calls) can block.
         ftab_call: single.on
-          // On a stack of its own, so the body (and guest code it calls) can block.
           ? new WebAssembly.Suspending((idx, ctx, depth, need, ret) => {
             const f = this.#cgen.ftab.get(tIdx(idx >>> 0));
             return this.#hardNested() ? f(ctx, depth, need, ret) : this.#promisingOf(f)(ctx, depth, need, ret);
@@ -4574,7 +4575,9 @@ export class Gex {
         this.#where = 'syscall-done';
         after?.();
         // A preemption point as well: a program that only makes system
-        // calls gives the other fibers a turn here.
+        // calls gives the other fibers a turn here, and `sched_yield` gives
+        // the woken one at once (single.mjs `Sched#settle`).
+        if (e.gex_last_syscall() === (globalThis.GEX_ARCH === 'x86' ? 24 : 124)) return sched.yieldNow();
         return sched.maybeYield();
       });
     });

shims/browser/js/single.mjs

@@ -193,12 +193,15 @@ export class Sched {
   #futex = new Map();
   /** Resolutions held back for a macrotask (see `#resume`). */
   #held = [];
+  /** Woken waiters not yet resumed: they run when the running fiber parks,
+   *  or at the next turn of the event loop (see `#settle`). */
+  #ready = [];
   #turnPosted = false;
   #mc = null;
   /** When the event loop last had a turn. */
   #lastTurn = performance.now();
   /** Counters, for the doc and for a stuck run. */
-  stats = { parks: 0, futexParks: 0, wakes: 0, yields: 0, macroTurns: 0, timeouts: 0 };
+  stats = { parks: 0, futexParks: 0, wakes: 0, yields: 0, macroTurns: 0, timeouts: 0, flushes: 0 };
 
   constructor() {
     if (typeof MessageChannel === 'function') {
@@ -217,7 +220,10 @@ export class Sched {
     this.#lastTurn = performance.now();
     this.stats.macroTurns++;
     const held = this.#held;
+    const ready = this.#ready;
     this.#held = [];
+    this.#ready = [];
+    for (const f of ready) f();
     for (const f of held) f();
   }
 
@@ -260,7 +266,25 @@ export class Sched {
       s?.delete(w);
       if (s && !s.size) this.#futex.delete(w.futex);
     }
-    this.#resume(() => w.resolve(why));
+    // Not resumed yet. A wake comes from the running fiber's system call
+    // (every one bumps the counter), and resolving here would queue the
+    // woken fiber's microtask ahead of the waker's own return from that call:
+    // a writer and its reader would trade places on every write instead of
+    // the writer filling the pipe first. So the woken wait until the running
+    // fiber parks (`#flush`), or until every fiber is blocked on something
+    // else and the event loop turns -- the posted turn. A wake from outside
+    // any fiber (a timer, the network, a key) gets that turn at once.
+    this.#ready.push(() => w.resolve(why));
+    this.#postTurn();
+  }
+
+  /** The running fiber is about to suspend: the woken go now. */
+  #flush() {
+    if (!this.#ready.length) return;
+    this.stats.flushes++;
+    const ready = this.#ready;
+    this.#ready = [];
+    for (const f of ready) this.#resume(f);
   }
 
   #park(futex, ns, fiber) {
@@ -277,6 +301,7 @@ export class Sched {
       if (Number.isFinite(ms) && ms < 2 ** 31) {
         w.timer = setTimeout(() => { this.stats.timeouts++; this.#settle(w, 2); }, ms);
       }
+      this.#flush();
     });
   }
 
@@ -329,6 +354,18 @@ export class Sched {
     });
   }
 
+  /**
+   * The running fiber asked to give way (`sched_yield`): the woken go first,
+   * then it. Without this a guest spinning on `sched_yield` for a fiber it
+   * just woke would wait out the slice.
+   */
+  yieldNow() {
+    if (!this.#ready.length) return this.maybeYield();
+    this.stats.yields++;
+    this.#flush();
+    return Promise.resolve(0);
+  }
+
   /** How many fibers are parked, for a census. */
   get parked() {
     let n = this.#global.size;