Return freed memory sooner: chunked shared writes, a bounded string spare, idle trims (#2518)

packages/compiler/test/ts7/baselines/order-parity/tests/corpus/shared-bytes-stdio-write.ts.json

@@ -0,0 +1,6 @@
+{
+ "order": [
+  "<repo>/tests/corpus/shared-bytes-stdio-write.ts"
+ ],
+ "diags": []
+}

packages/runtime/src/scr_alloc.c

@@ -1624,10 +1624,30 @@ static void scr_sa_group_starting(void) {
  * waits. Only that thread writes it. */
 static _Atomic(unsigned long) scr_sa_loop_turns;
 
+/* The carved bytes of the main thread's arena at its last idle trim. */
+static size_t scr_sa_idle_carved;
+
+/* The main thread returns its free pages when its loop waits with nothing
+ * ready, as a parking thread does (scr_sa_trim), once it has carved at least
+ * SCR_SA_TRIM_MIN more since its last trim: memory it freed (a large teardown
+ * before it starts workers, say) then stops counting while the workers
+ * allocate their own. Measuring the carved bytes costs one pass over the
+ * classes; the trim itself only runs when the frontier has grown. */
+static void scr_sa_idle_trim(void) {
+  if (scr_sa_arena < 0) return;
+  size_t carved = scr_sa_carved();
+  if (carved - scr_sa_idle_carved < SCR_SA_TRIM_MIN) return;
+  scr_sa_trim();
+  scr_sa_idle_carved = scr_sa_carved();
+}
+
 void scr_sa_loop_waiting(bool waiting) {
   if (scr_sa_worker) return;
   unsigned long turns = atomic_load_explicit(&scr_sa_loop_turns, memory_order_relaxed);
-  if ((bool)(turns & 1) != waiting) atomic_store_explicit(&scr_sa_loop_turns, turns + 1, memory_order_relaxed);
+  if ((bool)(turns & 1) != waiting) {
+    atomic_store_explicit(&scr_sa_loop_turns, turns + 1, memory_order_relaxed);
+    if (waiting) scr_sa_idle_trim();
+  }
 }
 
 /* Waits out the delay first, and stops if a worker starts meanwhile: the

packages/runtime/src/scr_alloc.test.c

@@ -923,6 +923,49 @@ static void check_discard_waits_for_loop(void) {
   free(discard_blocks);
 }
 
+/* The thread that is not a worker returns its free pages when its loop
+ * waits with nothing ready, once it has carved SCR_SA_TRIM_MIN more since
+ * its last trim; blocks allocated afterwards come from the returned range,
+ * zeroed. The test's thread plays the loop's. IDLE_SIZE is a class the
+ * other checks leave unused. */
+enum { IDLE_BLOCKS = 1 << 20, IDLE_KEEP = 4096, IDLE_SIZE = 88 };
+static void check_idle_trim(void) {
+  unsigned char **blocks = malloc(IDLE_BLOCKS * sizeof *blocks);
+  unsigned char *kept[IDLE_BLOCKS / IDLE_KEEP];
+  assert(blocks);
+  uintptr_t lo = UINTPTR_MAX, hi = 0;
+  for (size_t i = 0; i < IDLE_BLOCKS; i++) {
+    blocks[i] = scr_mem_alloc(IDLE_SIZE);
+    assert(blocks[i] && owned(blocks[i]));
+    fill(blocks[i], IDLE_SIZE, (unsigned char)i);
+    if ((uintptr_t)blocks[i] < lo) lo = (uintptr_t)blocks[i];
+    if ((uintptr_t)blocks[i] + IDLE_SIZE > hi) hi = (uintptr_t)blocks[i] + IDLE_SIZE;
+  }
+  for (size_t i = 0; i < IDLE_BLOCKS; i++) {
+    if (i % IDLE_KEEP == 0) kept[i / IDLE_KEEP] = blocks[i];
+    else scr_mem_free(blocks[i]);
+  }
+  size_t total = (hi - lo) / (size_t)sysconf(_SC_PAGESIZE);
+  assert(resident_pages(lo, hi) > total / 2);
+  scr_sa_loop_waiting(true);
+  scr_sa_loop_waiting(false);
+  /* One page per kept block stays, give or take the pages at run ends. */
+  assert(resident_pages(lo, hi) < total / 8);
+  for (size_t k = 0; k < IDLE_BLOCKS / IDLE_KEEP; k++) verify(kept[k], IDLE_SIZE, (unsigned char)(k * IDLE_KEEP));
+  size_t outside = 0;
+  for (size_t i = 0; i < IDLE_BLOCKS / 2; i++) {
+    unsigned char *q = scr_mem_calloc(IDLE_SIZE);
+    assert(q && owned(q));
+    for (size_t k = 0; k < IDLE_SIZE; k++) assert(q[k] == 0);
+    if ((uintptr_t)q < lo || (uintptr_t)q >= hi) outside++;
+    blocks[i] = q;
+  }
+  assert(outside < 256);
+  for (size_t i = 0; i < IDLE_BLOCKS / 2; i++) scr_mem_free(blocks[i]);
+  for (size_t k = 0; k < IDLE_BLOCKS / IDLE_KEEP; k++) scr_mem_free(kept[k]);
+  free(blocks);
+}
+
 static void check_discard(void) {
   discard_blocks = malloc(DISCARD_BLOCKS * sizeof *discard_blocks);
   assert(discard_blocks);
@@ -1172,6 +1215,7 @@ int main(void) {
   check_discard();
   check_discard_delay_grows();
   check_discard_waits_for_loop();
+  check_idle_trim();
 #endif
   printf("allocator checks passed: size-class workers\n");
 #else

packages/runtime/src/scr_bytes_io.c

@@ -1249,16 +1249,46 @@ ScrBytes *scr_crypto_random_bytes(double n) {
 
 /* ── process.stdout/stderr.write(buf) ──────────────────────────────────── */
 
+/* Where another thread can write shared bytes (where SCR_BYTES_SNAPSHOT
+ * copies them), they are read with scr_shared_load, so no shared raw memory
+ * reaches the system call. Written to the stream itself, they go through a
+ * bounded local buffer, so a large shared view is not copied whole first.
+ * With a write hook (the program reads process.stdout or process.stderr as
+ * a value, and may have replaced its write), the view is copied whole: a
+ * replaced write receives it in one call, as in Node, whatever it returns. */
+static bool scr_stdio_write_bytes(int fd, const ScrBytes *b) {
+  size_t n = b->len * scr_bytes_elem_size(b->elem);
+#if defined(SCR_WORKERS) || (defined(SCR_LIB) && defined(SCR_THREAD_INSTANCES))
+  if (!b->shared) return scr_stdio_write(fd, b->data, n);
+  if (scr_stdio_write_hook) {
+    unsigned char *copy = scr_mem_alloc(n ? n : 1);
+    scr_shared_load(copy, b->data, n);
+    bool ok = scr_stdio_write(fd, copy, n);
+    scr_mem_free(copy);
+    return ok;
+  }
+  enum { CHUNK = 64 * 1024 };
+  unsigned char *chunk = scr_mem_alloc(n < CHUNK ? (n ? n : 1) : CHUNK);
+  for (size_t off = 0; off < n; off += CHUNK) {
+    size_t k = n - off < CHUNK ? n - off : CHUNK;
+    scr_shared_load(chunk, b->data + off, k);
+    (void)scr_stdio_write_raw(fd, chunk, k);
+  }
+  scr_mem_free(chunk);
+  return true;
+#else
+  return scr_stdio_write(fd, b->data, n);
+#endif
+}
+
 bool scr_process_stdout_write_bytes(const ScrBytes *b, const ScrStr *encoding) {
-  SCR_BYTES_SNAPSHOT(b);
   (void)encoding;
-  return scr_stdio_write(1, b->data, b->len * scr_bytes_elem_size(b->elem));
+  return scr_stdio_write_bytes(1, b);
 }
 
 bool scr_process_stderr_write_bytes(const ScrBytes *b, const ScrStr *encoding) {
-  SCR_BYTES_SNAPSHOT(b);
   (void)encoding;
-  return scr_stdio_write(2, b->data, b->len * scr_bytes_elem_size(b->elem));
+  return scr_stdio_write_bytes(2, b);
 }
 
 /* Buffer.from over checked-native input. Strings use the existing codec;

packages/runtime/src/scr_string.c

@@ -432,9 +432,13 @@ static ScrStr *scr_str_new_ascii(const char *bytes, size_t len) {
 #ifndef SCR_RC_AUDIT
 static SCR_TL ScrStr *scr_str_spare;
 #endif
+/* The largest block the spare keeps. A larger one stays resident until a
+ * string of about its size comes along, which may be never: a program that
+ * replaces a 50 MB input with its edited copy would otherwise hold both. */
+#define SCR_STR_SPARE_MAX ((size_t)1 << 20)
 
 /* A spare-block reuse must not waste grossly (cap <= 4x the need) and only
- * sizable blocks are worth stashing (>= 512). */
+ * sizable blocks are worth stashing (512 bytes to SCR_STR_SPARE_MAX). */
 static ScrStr *scr_str_take_spare(size_t len) {
 #ifndef SCR_RC_AUDIT
   ScrStr *s = scr_str_spare;
@@ -483,7 +487,7 @@ void scr_str_release(ScrStr *s) {
     scr_live_strings--;
 #endif
 #ifndef SCR_RC_AUDIT
-    if (s->cap >= 512) {
+    if (s->cap >= 512 && s->cap <= SCR_STR_SPARE_MAX) {
       ScrStr *old = scr_str_spare;
       scr_str_spare = s;
       if (!old) return;

tests/corpus/shared-bytes-stdio-write.ts

@@ -0,0 +1,40 @@
+// process.stdout.write and process.stderr.write of views into a SharedArrayBuffer that a worker
+// filled: a view larger than one write chunk (64 KiB), at a byte offset, one shorter than a chunk,
+// and an empty one.
+import { Worker, isMainThread, parentPort, workerData } from "node:worker_threads";
+
+type Fill = { buffer: SharedArrayBuffer };
+
+if (isMainThread) {
+  const sab = new SharedArrayBuffer(200_000);
+  const worker = new Worker(new URL(import.meta.url), { workerData: { buffer: sab } });
+  worker.on("message", () => {
+    process.stdout.write(new Uint8Array(sab, 1000, 150_000));
+    process.stdout.write("\n");
+    process.stdout.write(new Uint8Array(sab, 0, 0));
+
+    process.stdout.write(new Uint8Array(sab, 160_000, 20_000));
+    process.stdout.write("\n");
+
+    process.stderr.write(new Uint8Array(sab, 63, 70_000));
+    process.stderr.write("\n");
+
+    // A replaced write receives the whole view in one call, even when it
+    // returns false.
+    const out = process.stdout as unknown as Record<string, unknown>;
+    const original = out["write"];
+    const sizes: number[] = [];
+    out["write"] = (chunk: unknown): boolean => {
+      sizes.push(typeof chunk === "string" ? chunk.length : (chunk as Uint8Array).length);
+      return false;
+    };
+    const accepted = process.stdout.write(new Uint8Array(sab, 1000, 150_000));
+    out["write"] = original;
+    console.log("replaced", accepted, sizes.join(","));
+    console.log("done");
+  });
+} else {
+  const bytes = new Uint8Array((workerData as Fill).buffer);
+  for (let i = 0; i < bytes.length; i++) bytes[i] = i % 64 === 63 ? 10 : 65 + (i % 26);
+  parentPort!.postMessage("filled");
+}

tests/harness/corpus-costs/shared-bytes-stdio-write.ts.json

@@ -0,0 +1 @@
+{ "plain": 7025, "san": 74291 }