Read and write cached regular files on the loop thread in fs streams (#2540)

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/8530-fs-stream-cached-positions.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/8530-fs-stream-cached-positions.ts"
+ ],
+ "diags": []
+}

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

packages/runtime/src/scr_fs_stream.c

@@ -19,6 +19,11 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#if defined(__linux__) && !defined(__ANDROID__)
+#include <sys/resource.h>
+#include <sys/stat.h>
+#include <sys/uio.h>
+#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);
 }
 

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

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 });
+      });
+    });
+  });

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 });
+  });

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 <dlfcn.h>
+#include <sys/uio.h>
+
+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);
+}

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);
+  });
+});