From f909bdf6c85d6cf26ed440bcd87e1c063d56b8b2 Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Wed, 28 Oct 2026 23:34:34 -0500 Subject: [PATCH 1/2] Make fewer syscalls per fs thread-pool completion and pool-thread write - A worker writes the completion wake byte only when the pipe has none. The loop reads it, with the queue lock held, when it takes the last waiting completion. The pipe holds a byte exactly while completions wait, so a burst of completions costs one write and one read instead of one write per completion and a read loop per dispatch. - The fs pool's threads keep SIGPIPE and SIGXFSZ blocked for their lifetime. scr_fs_write_once checks once per thread whether both are blocked; on such a thread a write makes no signal-mask calls, and a write that fails with EPIPE or EFBIG consumes the signal it raised. Other threads keep the per-write guard. fs-write-sync.test.ts covers EPIPE and EFBIG from write streams, whose writes run on the pool's threads, and matches Node's errors. --- packages/runtime/src/scr_async.c | 42 +++++++++++++++++------- packages/runtime/src/scr_lib.c | 43 +++++++++++++++++++++++++ tests/fixtures/fs-write-sync-sigpipe.ts | 20 ++++++++++-- tests/harness/fs-write-sync.test.ts | 18 ++++++++--- 4 files changed, 105 insertions(+), 18 deletions(-) diff --git a/packages/runtime/src/scr_async.c b/packages/runtime/src/scr_async.c index 0225269f4..92e95b8b7 100644 --- a/packages/runtime/src/scr_async.c +++ b/packages/runtime/src/scr_async.c @@ -3423,20 +3423,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; @@ -3530,28 +3535,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 } @@ -3661,6 +3667,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; } @@ -3759,13 +3777,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--; diff --git a/packages/runtime/src/scr_lib.c b/packages/runtime/src/scr_lib.c index ad0c613ee..edc2849c7 100644 --- a/packages/runtime/src/scr_lib.c +++ b/packages/runtime/src/scr_lib.c @@ -4101,6 +4101,49 @@ 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 thread and + * computed once. */ + static SCR_TL signed char thread_blocks = -1; + if (thread_blocks < 0) { + sigset_t current; + bool blocks = pthread_sigmask(SIG_BLOCK, NULL, ¤t) == 0; +#ifdef SIGPIPE + blocks = blocks && sigismember(¤t, SIGPIPE) == 1; +#endif +#ifdef SIGXFSZ + blocks = blocks && sigismember(¤t, 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) diff --git a/tests/fixtures/fs-write-sync-sigpipe.ts b/tests/fixtures/fs-write-sync-sigpipe.ts index e5fca1d71..1cbafb8b3 100644 --- a/tests/fixtures/fs-write-sync-sigpipe.ts +++ b/tests/fixtures/fs-write-sync-sigpipe.ts @@ -5,12 +5,28 @@ 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") { 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-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 { diff --git a/tests/harness/fs-write-sync.test.ts b/tests/harness/fs-write-sync.test.ts index bd277cd1c..a9ed08d98 100644 --- a/tests/harness/fs-write-sync.test.ts +++ b/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,14 @@ 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 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"); }); }, ); From 50aa42883f303e53343ac177211f0068a00e7a44 Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Thu, 29 Oct 2026 00:18:11 -0500 Subject: [PATCH 2/2] Cache the pool-thread signal check per OS thread scr_fs_write_once cached whether the calling thread blocks SIGPIPE and SIGXFSZ in an SCR_TL variable. SCR_TL is empty in builds without thread instances, so a pool thread's answer applied to the main thread too, and a later writeSync to a closed pipe ended the process with SIGPIPE instead of throwing EPIPE. The cache is _Thread_local now. The fs-write-sync fixture covers a write stream followed by writeSync on a closed stdout. --- packages/runtime/src/scr_lib.c | 7 ++++--- tests/fixtures/fs-write-sync-sigpipe.ts | 16 +++++++++++++++- tests/harness/fs-write-sync.test.ts | 4 ++++ 3 files changed, 23 insertions(+), 4 deletions(-) diff --git a/packages/runtime/src/scr_lib.c b/packages/runtime/src/scr_lib.c index edc2849c7..9ef71a223 100644 --- a/packages/runtime/src/scr_lib.c +++ b/packages/runtime/src/scr_lib.c @@ -4103,9 +4103,10 @@ static ssize_t scr_fs_write_once(int fd, const void *data, size_t length, #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 thread and - * computed once. */ - static SCR_TL signed char thread_blocks = -1; + * 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, ¤t) == 0; diff --git a/tests/fixtures/fs-write-sync-sigpipe.ts b/tests/fixtures/fs-write-sync-sigpipe.ts index 1cbafb8b3..24c0012b4 100644 --- a/tests/fixtures/fs-write-sync-sigpipe.ts +++ b/tests/fixtures/fs-write-sync-sigpipe.ts @@ -5,7 +5,7 @@ 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" || mode === "stream-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); } @@ -17,6 +17,20 @@ if (mode === "stream-sigpipe") { 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); diff --git a/tests/harness/fs-write-sync.test.ts b/tests/harness/fs-write-sync.test.ts index a9ed08d98..1c8b6c41e 100644 --- a/tests/harness/fs-write-sync.test.ts +++ b/tests/harness/fs-write-sync.test.ts @@ -87,6 +87,10 @@ describe.skipIf(process.platform === "win32")( 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");