diff --git a/packages/runtime/src/scr_async.c b/packages/runtime/src/scr_async.c index 9e35c673b..8f540d3d6 100644 --- a/packages/runtime/src/scr_async.c +++ b/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--; diff --git a/packages/runtime/src/scr_lib.c b/packages/runtime/src/scr_lib.c index ad0c613ee..9ef71a223 100644 --- a/packages/runtime/src/scr_lib.c +++ b/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, ¤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..24c0012b4 100644 --- a/tests/fixtures/fs-write-sync-sigpipe.ts +++ b/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 { diff --git a/tests/harness/fs-write-sync.test.ts b/tests/harness/fs-write-sync.test.ts index bd277cd1c..1c8b6c41e 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,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"); }); }, );