Make fewer syscalls per fs thread-pool completion and pool-thread write (#2519)

packages/runtime/src/scr_async.c

@@ -3491,20 +3491,25 @@ ScrPromise *scr_fsp_rename(ScrStr *oldpath, ScrStr *newpath) {
  * platform mutex publishes each result before the loop removes it from the
  * completion queue. */
 #if !defined(SCR_LIB) && !defined(__wasi__)
-/* The completion wake: a worker signals it once per published completion,
+/* The completion wake: a worker signals it when it publishes a completion,
  * so the loop's sleep ends for the completion — a pipe the poll(2) sleep
- * watches on POSIX, an auto-reset event the Windows sleep waits on. */
+ * watches on POSIX, an auto-reset event the Windows sleep waits on. On
+ * POSIX, `posted` (guarded by the queue lock) records that the pipe holds
+ * a byte: a worker writes one only when none is there, and the loop reads
+ * it when it takes the last completion, so the pipe holds a byte exactly
+ * while completions wait. */
 typedef struct ScrFsRenameWake {
 #ifdef _WIN32
   HANDLE event; /* NULL until created */
 #else
   int fds[2]; /* -1 until created */
+  bool posted;
 #endif
 } ScrFsRenameWake;
 #ifdef _WIN32
 #define SCR_FS_RENAME_WAKE_NONE {NULL}
 #else
-#define SCR_FS_RENAME_WAKE_NONE {{-1, -1}}
+#define SCR_FS_RENAME_WAKE_NONE {{-1, -1}, false}
 #endif
 #ifdef SCR_WORKERS
 typedef struct ScrFsRenameOwner ScrFsRenameOwner;
@@ -3598,28 +3603,29 @@ static void scr_fs_rename_wake_open(ScrFsRenameWake *wake) {
 #endif
 }
 
+/* Called with the queue lock held. */
 static void scr_fs_rename_wake_post(ScrFsRenameWake *wake) {
 #ifdef _WIN32
   if (wake->event != NULL) (void)SetEvent(wake->event);
 #else
-  if (wake->fds[1] >= 0) {
+  if (wake->fds[1] >= 0 && !wake->posted) {
     char byte = 1;
-    ssize_t n = write(wake->fds[1], &byte, 1); /* a full pipe already wakes the loop */
-    (void)n;
+    wake->posted = write(wake->fds[1], &byte, 1) == 1;
   }
 #endif
 }
 
-/* POSIX: empty the pipe before taking the queue, so a completion published
- * afterwards leaves its byte for the next poll. The Windows event resets
- * when a wait consumes it. */
+/* POSIX, with the queue lock held: the loop took the last completion, so
+ * the pipe's byte is read and the next completion writes a new one. The
+ * Windows event resets when a wait consumes it. */
 static void scr_fs_rename_wake_drain(ScrFsRenameWake *wake) {
 #ifdef _WIN32
   (void)wake;
 #else
-  if (wake->fds[0] < 0) return;
+  if (wake->fds[0] < 0 || !wake->posted) return;
+  wake->posted = false;
   char buf[64];
-  while (read(wake->fds[0], buf, sizeof buf) > 0) {
+  while (read(wake->fds[0], buf, sizeof buf) == (ssize_t)sizeof buf) {
   }
 #endif
 }
@@ -3729,6 +3735,18 @@ static DWORD WINAPI scr_fs_rename_worker(LPVOID payload) {
 #else
 static void *scr_fs_rename_worker(void *payload) {
   (void)payload;
+  /* A pool thread keeps SIGPIPE and SIGXFSZ blocked: its file writes then
+   * report EPIPE and EFBIG without changing the signal mask per write
+   * (scr_fs_write_once). */
+  sigset_t write_set;
+  sigemptyset(&write_set);
+#ifdef SIGPIPE
+  sigaddset(&write_set, SIGPIPE);
+#endif
+#ifdef SIGXFSZ
+  sigaddset(&write_set, SIGXFSZ);
+#endif
+  (void)pthread_sigmask(SIG_BLOCK, &write_set, NULL);
   scr_fs_rename_worker_loop();
   return NULL;
 }
@@ -3827,13 +3845,13 @@ static bool scr_fs_renames_dispatch(void) {
 #ifdef SCR_WORKERS
   if (!scr_fs_rename_owner) return false;
 #endif
-  scr_fs_rename_wake_drain(SCR_FS_RENAME_WAKE(scr_fs_rename_owner));
   scr_fs_rename_lock_enter();
   ScrFsRenameOp *op = scr_fs_rename_done;
   if (op != NULL) {
     scr_fs_rename_done = op->next;
     if (scr_fs_rename_done == NULL) scr_fs_rename_done_tail = &scr_fs_rename_done;
   }
+  if (scr_fs_rename_done == NULL) scr_fs_rename_wake_drain(SCR_FS_RENAME_WAKE(scr_fs_rename_owner));
   scr_fs_rename_lock_leave();
   if (op == NULL) return false;
   scr_fs_rename_pending_count--;

packages/runtime/src/scr_lib.c

@@ -4101,6 +4101,50 @@ static ssize_t scr_fs_write_once(int fd, const void *data, size_t length,
 #ifdef SIGXFSZ
   sigaddset(&write_set, SIGXFSZ);
 #endif
+  /* A thread that keeps both signals blocked (the fs pool's threads) writes
+   * without changing its mask; a write that fails with EPIPE or EFBIG then
+   * consumes the signal it left pending. The answer is per OS thread and
+   * computed once: _Thread_local, not SCR_TL, which is empty in builds
+   * without thread instances while the pool threads still exist. */
+  static _Thread_local signed char thread_blocks = -1;
+  if (thread_blocks < 0) {
+    sigset_t current;
+    bool blocks = pthread_sigmask(SIG_BLOCK, NULL, &current) == 0;
+#ifdef SIGPIPE
+    blocks = blocks && sigismember(&current, SIGPIPE) == 1;
+#endif
+#ifdef SIGXFSZ
+    blocks = blocks && sigismember(&current, SIGXFSZ) == 1;
+#endif
+    thread_blocks = blocks ? 1 : 0;
+  }
+  if (thread_blocks == 1) {
+    ssize_t n = positioned
+      ? scr_fs_pwrite(fd, data, length, position)
+      : write(fd, data, length);
+    if (n < 0) {
+      int write_errno = errno;
+      int raised = 0;
+#ifdef SIGPIPE
+      if (write_errno == EPIPE) raised = SIGPIPE;
+#endif
+#ifdef SIGXFSZ
+      if (write_errno == EFBIG) raised = SIGXFSZ;
+#endif
+      if (raised != 0 && sigpending(&pending) == 0 && sigismember(&pending, raised) == 1) {
+        sigset_t raised_set;
+        sigemptyset(&raised_set);
+        sigaddset(&raised_set, raised);
+        int caught = 0;
+        int wait_err;
+        do {
+          wait_err = sigwait(&raised_set, &caught);
+        } while (wait_err == EINTR);
+      }
+      errno = write_errno;
+    }
+    return n;
+  }
   if (pthread_sigmask(SIG_BLOCK, &write_set, &old_set) != 0) {
     return positioned
       ? scr_fs_pwrite(fd, data, length, position)

tests/fixtures/fs-write-sync-sigpipe.ts

@@ -5,12 +5,42 @@ import * as os from "node:os";
 import * as path from "node:path";
 
 const mode = process.argv.length > 2 ? process.argv[2] : "sigpipe";
-if (mode === "sigpipe") {
+if (mode === "sigpipe" || mode === "stream-sigpipe" || mode === "stream-then-sync-sigpipe") {
   const sleeper = new Int32Array(new SharedArrayBuffer(4));
   Atomics.wait(sleeper, 0, 0, 100);
 }
 
-if (mode === "eagain") {
+if (mode === "stream-sigpipe") {
+  // a write stream's writes run on the fs pool's threads
+  const out = fs.createWriteStream("", { fd: 1 });
+  out.on("error", (err: NodeJS.ErrnoException) => {
+    console.error("stream caught:", err.name, err.code, err.message);
+  });
+  out.write("probe\n");
+} else if (mode === "stream-then-sync-sigpipe") {
+  // a pool thread writes first; the calling thread's writeSync after it
+  // keeps its own guard
+  const scratch = path.join(os.tmpdir(), `scr-write-then-sync-${process.pid}`);
+  const file = fs.createWriteStream(scratch);
+  file.end("pool\n", () => {
+    fs.unlinkSync(scratch);
+    try {
+      fs.writeSync(1, "probe\n");
+    } catch (e) {
+      const err = e as NodeJS.ErrnoException;
+      console.error("sync caught:", err.name, err.code);
+    }
+  });
+} else if (mode === "stream-efbig") {
+  const scratch = path.join(os.tmpdir(), `scr-write-stream-${process.pid}`);
+  const out = fs.createWriteStream(scratch);
+  out.on("error", (err: NodeJS.ErrnoException) => {
+    console.error("stream caught:", err.name, err.code, err.message);
+    fs.unlinkSync(scratch);
+  });
+  out.write(Buffer.alloc(512));
+  out.write(Buffer.alloc(4096));
+} else if (mode === "eagain") {
   const data = Buffer.alloc(65536);
   for (;;) {
     try {

tests/harness/fs-write-sync.test.ts

@@ -1,7 +1,9 @@
-/* fs.writeSync contracts that need harness control over the native process's
- * descriptors. The ordinary byte/error behavior stays in differential corpus
- * 2685; this test controls stdout's pipe state, nonblocking mode, and the
- * process file-size limit to pin the signal-backed descriptor errors. */
+/* fs.writeSync and write-stream contracts that need harness control over the
+ * native process's descriptors. The ordinary byte/error behavior stays in
+ * differential corpus 2685; this test controls stdout's pipe state,
+ * nonblocking mode, and the process file-size limit to pin the signal-backed
+ * descriptor errors, on the calling thread (writeSync) and on the fs pool's
+ * threads (write streams). */
 import { execFileSync, spawn } from "node:child_process";
 import { join } from "node:path";
 import { describe, expect, test } from "vitest";
@@ -80,6 +82,18 @@ describe.skipIf(process.platform === "win32")(
           "positioned first: 512\n" +
           "positioned caught: Error EFBIG EFBIG: file too large, write\n",
       );
+
+      const streamPipe = await run(result.binaryPath, ["stream-sigpipe"], true);
+      expect(streamPipe.outcome).toEqual({ code: 0, signal: null });
+      expect(streamPipe.stderr).toBe("stream caught: Error EPIPE EPIPE: broken pipe, write\n");
+
+      const thenSync = await run(result.binaryPath, ["stream-then-sync-sigpipe"], true);
+      expect(thenSync.outcome).toEqual({ code: 0, signal: null });
+      expect(thenSync.stderr).toBe("sync caught: Error EPIPE\n");
+
+      const streamLimit = await run(launcher, ["rlimit", result.binaryPath, "stream-efbig"]);
+      expect(streamLimit.outcome).toEqual({ code: 0, signal: null });
+      expect(streamLimit.stderr).toBe("stream caught: Error EFBIG EFBIG: file too large, write\n");
     });
   },
 );