diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8530-fs-stream-cached-positions.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8530-fs-stream-cached-positions.ts.json new file mode 100644 index 000000000..9fd2b0d20 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8530-fs-stream-cached-positions.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/8530-fs-stream-cached-positions.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_async.c b/packages/runtime/src/scr_async.c index 8f540d3d6..872026b97 100644 --- a/packages/runtime/src/scr_async.c +++ b/packages/runtime/src/scr_async.c @@ -3540,15 +3540,25 @@ struct ScrFsRenameOwner { ScrFsRenameOp **tail; size_t pending; ScrFsRenameWake wake; + ScrFsRenameOp *next_turn; + ScrFsRenameOp **next_turn_tail; }; static SCR_TL ScrFsRenameOwner *scr_fs_rename_owner; #define scr_fs_rename_done (scr_fs_rename_owner->done) #define scr_fs_rename_done_tail (scr_fs_rename_owner->tail) #define scr_fs_rename_pending_count (scr_fs_rename_owner->pending) +#define scr_fs_rename_next_turn (scr_fs_rename_owner->next_turn) +#define scr_fs_rename_next_turn_tail (scr_fs_rename_owner->next_turn_tail) #else static ScrFsRenameOp *scr_fs_rename_done = NULL; static ScrFsRenameOp **scr_fs_rename_done_tail = &scr_fs_rename_done; static size_t scr_fs_rename_pending_count = 0; +/* Jobs the main thread completed itself (scr_work_submit_done): they join + * the done queue at the loop's next sleep step (scr_fs_renames_promote), so + * their completions wait one loop turn as a pool job's do. Main thread + * only, so no lock. */ +static ScrFsRenameOp *scr_fs_rename_next_turn = NULL; +static ScrFsRenameOp **scr_fs_rename_next_turn_tail = &scr_fs_rename_next_turn; static ScrFsRenameWake scr_fs_rename_wake_global = SCR_FS_RENAME_WAKE_NONE; #endif static size_t scr_fs_rename_worker_count = 0; @@ -3659,6 +3669,13 @@ static void scr_fs_rename_op_release(ScrFsRenameOp *op) { #ifdef SCR_WORKERS static void scr_fs_rename_context_cleanup(void) { ScrFsRenameOwner *owner = scr_fs_rename_owner; + /* Jobs completed on this thread need no wait. */ + while (owner->next_turn) { + ScrFsRenameOp *op = owner->next_turn; + owner->next_turn = op->next; + owner->pending--; + scr_fs_rename_op_release(op); + } /* Outstanding native work may still borrow payload storage. Wait for its * publication before reclaiming it, without invoking script callbacks * during context teardown. Other script contexts keep using the pool. */ @@ -3682,6 +3699,7 @@ static ScrFsRenameOwner *scr_fs_rename_context(void) { ScrFsRenameOwner *owner = calloc(1, sizeof(*owner)); if (!owner) scr_oom(); owner->tail = &owner->done; + owner->next_turn_tail = &owner->next_turn; owner->wake = (ScrFsRenameWake)SCR_FS_RENAME_WAKE_NONE; if (scr_atexit(scr_fs_rename_context_cleanup) != 0) { free(owner); @@ -3800,9 +3818,33 @@ static void scr_fs_renames_shutdown(void) { scr_fs_rename_op_release(done); done = next; } + while (scr_fs_rename_next_turn != NULL) { + ScrFsRenameOp *next = scr_fs_rename_next_turn->next; + scr_fs_rename_op_release(scr_fs_rename_next_turn); + scr_fs_rename_next_turn = next; + } + scr_fs_rename_next_turn_tail = &scr_fs_rename_next_turn; #endif } +/* At the loop's sleep step: the jobs this thread completed during the turn + * join the done queue, so the next turn delivers them, one per checkpoint + * as pool completions are. Answers whether any joined (the sleep is then + * skipped). */ +static bool scr_fs_renames_promote(void) { +#ifdef SCR_WORKERS + if (!scr_fs_rename_owner) return false; +#endif + if (scr_fs_rename_next_turn == NULL) return false; + scr_fs_rename_lock_enter(); + *scr_fs_rename_done_tail = scr_fs_rename_next_turn; + scr_fs_rename_done_tail = scr_fs_rename_next_turn_tail; + scr_fs_rename_lock_leave(); + scr_fs_rename_next_turn = NULL; + scr_fs_rename_next_turn_tail = &scr_fs_rename_next_turn; + return true; +} + static bool scr_fs_renames_pending(void) { #ifdef SCR_WORKERS if (!scr_fs_rename_owner) return false; @@ -3966,6 +4008,7 @@ static size_t scr_fs_rename_pending_count = 0; static bool scr_fs_renames_pending(void) { return scr_fs_rename_pending_count != 0; } static int scr_fs_renames_pollfd(void) { return -1; } static bool scr_fs_renames_can_wait(void) { return false; } +static bool scr_fs_renames_promote(void) { return false; } static bool scr_fs_renames_dispatch(void) { ScrFsRenameOp *op = scr_fs_rename_done; @@ -4005,6 +4048,7 @@ static bool scr_fs_renames_pending(void) { return false; } static bool scr_fs_renames_dispatch(void) { return false; } static int scr_fs_renames_pollfd(void) { return -1; } static bool scr_fs_renames_can_wait(void) { return false; } +static bool scr_fs_renames_promote(void) { return false; } #endif void scr_fs_rename_async(ScrStr *oldpath, ScrStr *newpath, @@ -4069,6 +4113,44 @@ void scr_work_submit(void *payload, ScrWorkFn work, ScrWorkFn after, #endif } +#if defined(__wasi__) +static void scr_work_already_done(void *payload) { (void)payload; } +#endif + +void scr_work_submit_done(void *payload, ScrWorkFn after, ScrWorkFn destroy) { +#if defined(SCR_LIB) + (void)after; + destroy(payload); + scr_trap("scriptc: native worker jobs are not supported by this target\n"); +#else + ScrFsRenameOp *op = calloc(1, sizeof *op); + if (!op) scr_trap("scriptc: out of memory\n"); + op->generic = true; + op->payload = payload; + op->after = after; + op->destroy = destroy; +#if defined(__wasi__) + op->work = &scr_work_already_done; /* the dispatch runs work, then after */ + *scr_fs_rename_done_tail = op; + scr_fs_rename_done_tail = &op->next; +#else + op->ran = true; +#ifdef SCR_WORKERS + op->owner = scr_fs_rename_context(); +#endif + if (!scr_fs_rename_shutdown_registered) { + if (atexit(scr_fs_renames_shutdown) != 0) { + scr_trap("scriptc: could not register worker cleanup\n"); + } + scr_fs_rename_shutdown_registered = true; + } + *scr_fs_rename_next_turn_tail = op; + scr_fs_rename_next_turn_tail = &op->next; +#endif + scr_fs_rename_pending_count++; +#endif +} + void scr_fs_rename_thunk0(ScrClosure *cb, ScrError *err) { (void)err; ((void (*)(ScrClosure *))cb->fn)(cb); @@ -5238,11 +5320,13 @@ void scr_loop_hosted_io_state(ScrNodeIoState *st) { * end of the turn that queued them and run in the next one, as the loop * pass delivers them; Node's own fs requests take a turn or more too. */ if (st->phase == SCR_NODE_IO_CHECK) scr_loop_deferred_ready = scr_loop_deferred_count; + /* jobs this thread completed during the turn: delivered next turn */ + bool promoted = st->phase == SCR_NODE_IO_CHECK && scr_fs_renames_promote(); st->alive = kids_reffed || kids_failed || events || net || dgram || watch || ffi || stream || renames || crypto || deferred || workers; /* Queued deferred callbacks keep the poll from sleeping (their work is * done; only the delivery waits for the next turn). */ - st->ready = stream || crypto || deferred || kids_failed || + st->ready = stream || crypto || deferred || promoted || kids_failed || (scr_loop_children != NULL && scr_loop_children->ready()); #ifdef SCR_WORKERS if (scr_workers_ready_fn && scr_workers_ready_fn()) st->ready = 1; @@ -5699,6 +5783,9 @@ static bool scr_loop_run_pass(ScrPromise *top_level, bool first_checkpoint) { due = now; scr_loop_deferred_ready = scr_loop_deferred_count; } + /* Jobs this thread completed during the turn (scr_work_submit_done) + * join the done queue: the next turn delivers them, no sleep. */ + if (scr_fs_renames_promote()) due = now; /* Child polling can drain a pipe into the IPC queue after its dispatch * station has run. Completed sends and disconnects can also become ready * during that turn. No fd will wake us for this userspace work: return to diff --git a/packages/runtime/src/scr_fs_stream.c b/packages/runtime/src/scr_fs_stream.c index 940068a7b..ae53f5d4e 100644 --- a/packages/runtime/src/scr_fs_stream.c +++ b/packages/runtime/src/scr_fs_stream.c @@ -19,6 +19,11 @@ #include #include #include +#if defined(__linux__) && !defined(__ANDROID__) +#include +#include +#include +#endif #define SCR_FSS_MAX_SAFE 9007199254740991.0 @@ -39,6 +44,10 @@ typedef struct ScrFsStream { ScrError *destroy_err; /* owned: the waiting _destroy's error */ void *owner; /* a FileHandle the stream reads or writes, or NULL */ const ScrFsStreamOwnerOps *owner_ops; + /* the descriptor scr_fss_try_inline last checked (0 before the first + * check) and whether its reads and writes may run on the loop thread */ + int inline_fd; + bool inline_ok; } ScrFsStream; static void scr_fss_free(void *p) { @@ -77,6 +86,7 @@ typedef struct ScrFssOp { size_t off, len; double pos; /* < 0: the current offset */ int64_t result; /* answer or -errno */ + size_t got; /* READ: bytes the loop thread already read into buf */ int retries; /* WRITE: zero-byte retries */ ScrError *carry; /* FSYNC/CLOSE: the error to pass on (+1) */ /* The AsyncLocalStorage context the step was started in (+1): its @@ -88,7 +98,13 @@ static void scr_fss_work(void *p) { ScrFssOp *op = p; switch (op->kind) { case SCR_FSS_OPEN: op->result = scr_fs_raw_open(op->path, op->oflags, op->mode); break; - case SCR_FSS_READ: op->result = scr_fs_raw_read(op->fd, op->buf, op->len, op->pos); break; + case SCR_FSS_READ: { + /* the rest of a read the loop thread began (scr_fss_try_inline) */ + int64_t r = scr_fs_raw_read(op->fd, op->buf + op->got, op->len - op->got, + op->pos >= 0 ? op->pos + (double)op->got : -1); + op->result = op->got == 0 ? r : r < 0 ? (int64_t)op->got : (int64_t)op->got + r; + break; + } case SCR_FSS_WRITE: op->result = scr_fs_raw_write(op->fd, op->chunk->data + op->off, op->len, op->pos); break; @@ -127,7 +143,67 @@ static void scr_fss_after_in_context(void *p) { scr_als_dispatch_leave(prev); } +#if defined(__linux__) && !defined(__ANDROID__) && defined(RWF_NOWAIT) +/* Without a file-size limit a regular file's write cannot raise SIGXFSZ on + * the loop thread (scr_fs_write_once's guard is then not needed). */ +static bool scr_fss_size_unlimited(void) { + static int unlimited = -1; + if (unlimited < 0) { + struct rlimit lim; + unlimited = getrlimit(RLIMIT_FSIZE, &lim) == 0 && lim.rlim_cur == RLIM_INFINITY; + } + return unlimited == 1; +} + +/* A read or write of a regular file whose data the page cache holds runs on + * the loop thread: RWF_NOWAIT answers EAGAIN instead of blocking. The + * completion still waits for the next loop turn (scr_work_submit_done), as + * a pool job's does. A read the cache holds only part of finishes on the + * pool, so the chunk is the one a single read() returns; a partial write + * completes with the bytes written, and the rest is submitted again, as + * after any partial write. Answers whether the step completed here. */ +static bool scr_fss_try_inline(ScrFssOp *op) { + if (op->kind != SCR_FSS_READ && op->kind != SCR_FSS_WRITE) return false; + ScrFsStream *f = scr_fss_of(op->s); + if (f->inline_fd != op->fd + 1) { + struct stat st; + f->inline_fd = op->fd + 1; + f->inline_ok = fstat(op->fd, &st) == 0 && S_ISREG(st.st_mode); + } + if (!f->inline_ok || op->len == 0) return false; + if (op->kind == SCR_FSS_WRITE && !scr_fss_size_unlimited()) return false; + off_t at = op->pos >= 0 ? (off_t)op->pos : -1; + ssize_t n; + if (op->kind == SCR_FSS_READ) { + struct iovec iov = {op->buf, op->len}; + n = preadv2(op->fd, &iov, 1, at, RWF_NOWAIT); + } else { + struct iovec iov = {op->chunk->data + op->off, op->len}; + n = pwritev2(op->fd, &iov, 1, at, RWF_NOWAIT); + } + if (n < 0) { + /* a kernel or file system without RWF_NOWAIT: the pool from now on */ + if (errno == EOPNOTSUPP || errno == EINVAL || errno == ENOSYS) f->inline_ok = false; + return false; + } + /* A zero is not taken as EOF here: some kernels (Linux 5.9 and 5.10) + * answer 0 instead of EAGAIN for an uncached RWF_NOWAIT read, so the + * pool's read decides, as it does for a zero-byte write. */ + if (n == 0) return false; + if (op->kind == SCR_FSS_READ && n > 0 && (size_t)n < op->len) { + op->got = (size_t)n; + return false; + } + op->result = n; + scr_work_submit_done(op, &scr_fss_after_in_context, &scr_fss_op_destroy); + return true; +} +#endif + static void scr_fss_submit(ScrFssOp *op) { +#if defined(__linux__) && !defined(__ANDROID__) && defined(RWF_NOWAIT) + if (scr_fss_try_inline(op)) return; +#endif scr_work_submit(op, &scr_fss_work, &scr_fss_after_in_context, &scr_fss_op_destroy); } diff --git a/packages/runtime/src/scr_lib.h b/packages/runtime/src/scr_lib.h index bc0184573..daaf9aae0 100644 --- a/packages/runtime/src/scr_lib.h +++ b/packages/runtime/src/scr_lib.h @@ -575,6 +575,10 @@ int scr_fs_raw_close(int fd); typedef void (*ScrWorkFn)(void *payload); void scr_work_submit(void *payload, ScrWorkFn work, ScrWorkFn after, ScrWorkFn destroy); +/* A job whose work already ran on the main runtime thread: `after` runs on + * the next loop turn, as a pool job's completion does, and `destroy` + * follows it. */ +void scr_work_submit_done(void *payload, ScrWorkFn after, ScrWorkFn destroy); /* fs.rename: the syscall is submitted to a native worker immediately and * its error-first callback fires on a later event-loop turn through an diff --git a/tests/corpus/8530-fs-stream-cached-positions.ts b/tests/corpus/8530-fs-stream-cached-positions.ts new file mode 100644 index 000000000..1361085d3 --- /dev/null +++ b/tests/corpus/8530-fs-stream-cached-positions.ts @@ -0,0 +1,36 @@ +// fs streams over a cached regular file: chunk sizes of a positioned read +// with a small highWaterMark, a write stream with a start position into an +// existing file, and an append stream, checked against the file contents. +import { createReadStream, createWriteStream, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const dir = mkdtempSync(join(tmpdir(), "scr-fs-cached-")); +const file = join(dir, "data.bin"); +const bytes = Buffer.alloc(100000); +for (let i = 0; i < bytes.length; i++) bytes[i] = (i * 7) & 0xff; +writeFileSync(file, bytes); + +const sizes: number[] = []; +let sum = 0; +createReadStream(file, { start: 1000, end: 50999, highWaterMark: 4096 }) + .on("data", (c: Buffer) => { + sizes.push(c.length); + for (const b of c) sum = (sum + b) % 65521; + }) + .on("end", () => { + console.log(`read chunks=${sizes.length} first=${sizes[0]} last=${sizes[sizes.length - 1]} sum=${sum}`); + const out = createWriteStream(file, { flags: "r+", start: 10 }); + for (let i = 0; i < 50; i++) out.write(Buffer.alloc(1000, i)); + out.end(() => { + const after = readFileSync(file); + console.log(`positioned len=${after.length} at10=${after[10]} at1009=${after[1009]} at1010=${after[1010]} at50009=${after[50009]} at50010=${after[50010]}`); + const app = createWriteStream(file, { flags: "a" }); + app.write("tail-1;"); + app.end("tail-2", () => { + const final = readFileSync(file); + console.log(`append len=${final.length} end=${final.subarray(final.length - 13).toString()}`); + rmSync(dir, { recursive: true }); + }); + }); + }); diff --git a/tests/fixtures/fs-stream-nowait-read.ts b/tests/fixtures/fs-stream-nowait-read.ts new file mode 100644 index 000000000..5e9fd3739 --- /dev/null +++ b/tests/fixtures/fs-stream-nowait-read.ts @@ -0,0 +1,23 @@ +// Reads a file it writes first through a ReadStream and prints the byte +// count and a checksum (fs-stream-nowait.test.ts runs it with a preadv2 +// shim). +import { createReadStream, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const dir = mkdtempSync(join(tmpdir(), "scr-nowait-")); +const file = join(dir, "data.bin"); +const data = Buffer.alloc(300000); +for (let i = 0; i < data.length; i++) data[i] = (i * 13) & 0xff; +writeFileSync(file, data); +let bytes = 0; +let sum = 0; +createReadStream(file, { highWaterMark: 16384 }) + .on("data", (c: Buffer) => { + bytes += c.length; + for (const b of c) sum = (sum + b) % 65521; + }) + .on("end", () => { + console.log(`bytes=${bytes} sum=${sum}`); + rmSync(dir, { recursive: true }); + }); diff --git a/tests/fixtures/fs-stream-nowait-zero.c b/tests/fixtures/fs-stream-nowait-zero.c new file mode 100644 index 000000000..69626997f --- /dev/null +++ b/tests/fixtures/fs-stream-nowait-zero.c @@ -0,0 +1,13 @@ +/* LD_PRELOAD shim for fs-stream-nowait.test.ts: preadv2 with RWF_NOWAIT + * answers 0, as Linux 5.9 and 5.10 do for an uncached RWF_NOWAIT read; + * other calls go to the real preadv2. */ +#define _GNU_SOURCE +#include +#include + +ssize_t preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) { + if (flags & RWF_NOWAIT) return 0; + ssize_t (*real)(int, const struct iovec *, int, off_t, int) = + (ssize_t (*)(int, const struct iovec *, int, off_t, int))dlsym(RTLD_NEXT, "preadv2"); + return real(fd, iov, iovcnt, offset, flags); +} diff --git a/tests/harness/fs-stream-nowait.test.ts b/tests/harness/fs-stream-nowait.test.ts new file mode 100644 index 000000000..a6ba9e074 --- /dev/null +++ b/tests/harness/fs-stream-nowait.test.ts @@ -0,0 +1,44 @@ +/* fs streams read a regular file with RWF_NOWAIT on the loop thread first + * (scr_fs_stream.c). Linux 5.9 and 5.10 answer 0 instead of EAGAIN for + * an uncached RWF_NOWAIT read; a preloaded preadv2 shim reproduces that + * answer, and the stream must still read the whole file, as Node does. */ +import { execFileSync, spawnSync } from "node:child_process"; +import { join } from "node:path"; +import { describe, expect, test } from "vitest"; +import { compile } from "@scriptc/compiler"; + +const repoRoot = join(import.meta.dirname, "../.."); +const cacheDir = join(repoRoot, "node_modules/.cache/scriptc-tests"); +const sanitize = process.env["SCRIPTC_SAN"] === "1"; + +describe.skipIf(process.platform !== "linux" || sanitize)("fs stream RWF_NOWAIT reads", () => { + test("a zero from an RWF_NOWAIT read is not taken as EOF", async () => { + const outDir = join(cacheDir, "fs-stream-nowait"); + const result = await compile(join(repoRoot, "tests/fixtures/fs-stream-nowait-read.ts"), { + outPath: join(outDir, "read"), + outDir, + backend: "llvm", + }); + if (!result.ok) { + throw new Error(result.diagnostics.map((d) => `${d.code}: ${d.message}`).join("\n")); + } + const shim = join(outDir, "nowait-zero.so"); + execFileSync("clang", [ + "-shared", + "-fPIC", + join(repoRoot, "tests/fixtures/fs-stream-nowait-zero.c"), + "-o", + shim, + "-ldl", + ]); + const expected = "bytes=300000 sum=50681\n"; + const plain = spawnSync(result.binaryPath, [], { encoding: "utf8" }); + expect(plain.stdout).toBe(expected); + const shimmed = spawnSync(result.binaryPath, [], { + encoding: "utf8", + env: { ...process.env, LD_PRELOAD: shim }, + }); + expect(shimmed.status).toBe(0); + expect(shimmed.stdout).toBe(expected); + }); +});