webstreams: rewrite ReadableStream, WritableStream, and TransformStream in C++ (zero JS builtins)

.gitignore

@@ -198,3 +198,7 @@ src/runtime/bake/generated.ts
 # them on disk; keep ignored so they don't show as untracked).
 src/jsc/bindings/GeneratedJS2Native.zig
 src/jsc/bindings/GeneratedBindings.zig
+
+# Web Streams rewrite working notes (design docs, spec transcription, review logs).
+# Kept locally for the ongoing work; not part of the source tree.
+/specs/

bench/snippets/webstreams-consumers.mjs

@@ -0,0 +1,100 @@
+// Consumer x chunk-shape matrix for JS-sourced ReadableStreams (MB/s, best of RUNS).
+// Every shape totals ~8 MiB so numbers are comparable across rows.
+const RUNS = 5;
+const MB = 1024 * 1024;
+const TOTAL = 8 * MB;
+
+const binary = size => {
+  const chunk = new Uint8Array(size).fill(120);
+  const count = TOTAL / size;
+  return () => {
+    let i = 0;
+    return new ReadableStream({
+      pull(c) {
+        if (i++ < count) c.enqueue(chunk);
+        else c.close();
+      },
+    });
+  };
+};
+const text = size => {
+  const chunk = "x".repeat(size);
+  const count = TOTAL / size;
+  return () => {
+    let i = 0;
+    return new ReadableStream({
+      pull(c) {
+        if (i++ < count) c.enqueue(chunk);
+        else c.close();
+      },
+    });
+  };
+};
+const mixed = size => {
+  const textChunk = "y".repeat(size);
+  const binaryChunk = new Uint8Array(size).fill(121);
+  const count = TOTAL / size;
+  return () => {
+    let i = 0;
+    return new ReadableStream({
+      pull(c) {
+        if (i < count) (c.enqueue(i % 2 ? textChunk : binaryChunk), i++);
+        else c.close();
+      },
+    });
+  };
+};
+
+const shapes = {
+  "binary 64KiB x128": binary(64 * 1024),
+  "binary 1KiB x8192": binary(1024),
+  "text 64KiB x128": text(64 * 1024),
+  "text 1KiB x8192": text(1024),
+  "mixed text/bytes 64KiB x128": mixed(64 * 1024),
+  "one 8MiB chunk": (() => {
+    const chunk = new Uint8Array(TOTAL).fill(122);
+    return () =>
+      new ReadableStream({
+        start(c) {
+          c.enqueue(chunk);
+          c.close();
+        },
+      });
+  })(),
+};
+
+const consumers = {
+  "toText": s => Bun.readableStreamToText(s),
+  "toArrayBuffer": s => Bun.readableStreamToArrayBuffer(s),
+  "toBytes": s => Bun.readableStreamToBytes(s),
+  "toArray": s => Bun.readableStreamToArray(s),
+  "toBlob": async s => (await Bun.readableStreamToBlob(s)).size,
+  "Response.text": s => new Response(s).text(),
+  "Response.arrayBuffer": s => new Response(s).arrayBuffer(),
+  "for await": async s => {
+    let n = 0;
+    for await (const c of s) n += c.length;
+    return n;
+  },
+};
+
+const table = {};
+for (const [shapeName, make] of Object.entries(shapes)) {
+  const row = (table[shapeName] = {});
+  for (const [consumerName, consume] of Object.entries(consumers)) {
+    await consume(make()); // warmup + validity
+    let best = Infinity;
+    for (let i = 0; i < RUNS; i++) {
+      const t0 = performance.now();
+      await consume(make());
+      best = Math.min(best, performance.now() - t0);
+    }
+    row[consumerName] = Math.round(TOTAL / MB / (best / 1000));
+  }
+}
+const consumerNames = Object.keys(consumers);
+const version = typeof Bun !== "undefined" ? `bun ${Bun.revision.slice(0, 9)}` : `node ${process.version}`;
+console.log(`# webstreams consumers (MB/s) — ${version} — ${TOTAL / MB} MiB per pass, best of ${RUNS}`);
+console.log(["shape".padEnd(28), ...consumerNames.map(n => n.padStart(14))].join(""));
+for (const [shapeName, row] of Object.entries(table))
+  console.log([shapeName.padEnd(28), ...consumerNames.map(n => String(row[n]).padStart(14))].join(""));

bench/snippets/webstreams-memory.mjs

@@ -0,0 +1,122 @@
+// Web Streams memory: (1) retained RSS per live instance, (2) peak/settled RSS for
+// streaming workloads. Run with any JS runtime; extra heap counts print under Bun.
+const N = 100_000;
+const gc = globalThis.Bun?.gc ?? globalThis.gc;
+if (!gc) throw new Error("This benchmark needs a GC hook: run with Bun, or node --expose-gc.");
+const rss = () => process.memoryUsage.rss();
+const MB = 1024 * 1024;
+const fmt = n => (n / MB).toFixed(1).padStart(8) + " MB";
+
+console.log(`# per-instance retained RSS (n=${N} live instances)`);
+const keep = [];
+function perInstance(label, make) {
+  gc(true);
+  const before = rss();
+  const held = new Array(N);
+  for (let i = 0; i < N; i++) held[i] = make();
+  gc(true);
+  const perObject = (rss() - before) / N;
+  console.log(`${label.padEnd(40)} ${perObject.toFixed(0).padStart(6)} bytes/instance`);
+  keep.push(held);
+}
+perInstance("new ReadableStream({pull(){}})", () => new ReadableStream({ pull() {} }));
+perInstance("new ReadableStream() + getReader()", () => new ReadableStream({ pull() {} }).getReader());
+perInstance("new WritableStream({write(){}})", () => new WritableStream({ write() {} }));
+perInstance("new TransformStream()", () => new TransformStream());
+keep.length = 0;
+gc(true);
+
+console.log(`\n# workload RSS (peak over baseline during the run, settled after gc)`);
+const CHUNK = new Uint8Array(64 * 1024).fill(120);
+async function workload(label, fn) {
+  gc(true);
+  const before = rss();
+  let peak = before;
+  const timer = setInterval(() => {
+    peak = Math.max(peak, rss());
+  }, 5);
+  // Whatever `fn` returns is kept alive until after the settled measurement, so
+  // "N live objects" workloads measure retention rather than post-return garbage.
+  const keepAlive = await fn();
+  clearInterval(timer);
+  peak = Math.max(peak, rss());
+  gc(true);
+  const settled = rss();
+  console.log(`${label.padEnd(46)} peak ${fmt(peak - before)}   settled ${fmt(settled - before)}`);
+  return keepAlive;
+}
+const source = n => {
+  let i = 0;
+  return new ReadableStream({
+    pull(c) {
+      if (i++ < n) c.enqueue(CHUNK);
+      else c.close();
+    },
+  });
+};
+await workload("pipeTo 512 MiB (64 KiB chunks)", async () => {
+  let n = 0;
+  await source(8192).pipeTo(
+    new WritableStream({
+      write(c) {
+        n += c.length;
+      },
+    }),
+  );
+});
+await workload("for await 512 MiB", async () => {
+  let n = 0;
+  for await (const c of source(8192)) n += c.length;
+});
+await workload("Response(stream 256 MiB).arrayBuffer()", async () => {
+  (await new Response(source(4096)).arrayBuffer()).byteLength;
+});
+await workload("Response(stream 256 MiB of text).text()", async () => {
+  let i = 0;
+  const text = "x".repeat(64 * 1024);
+  const rs = new ReadableStream({
+    pull(c) {
+      if (i++ < 4096) c.enqueue(text);
+      else c.close();
+    },
+  });
+  (await new Response(rs).text()).length;
+});
+{
+  const held = await workload("10k live TransformStream chains (held)", async () => {
+    const chains = new Array(10_000);
+    for (let i = 0; i < chains.length; i++) {
+      const ts = new TransformStream();
+      chains[i] = [ts, ts.readable.getReader(), ts.writable.getWriter()];
+    }
+    gc(true);
+    return chains;
+  });
+  held.length = 0;
+}
+await workload("2k concurrent pipeThrough pipes (1 MiB each)", async () => {
+  const pipes = [];
+  for (let i = 0; i < 2000; i++) {
+    let k = 0;
+    const rs = new ReadableStream({
+      pull(c) {
+        if (k++ < 16) c.enqueue(CHUNK);
+        else c.close();
+      },
+    });
+    pipes.push(rs.pipeThrough(new TransformStream()).pipeTo(new WritableStream({ write() {} })));
+  }
+  await Promise.all(pipes);
+});
+
+if (typeof Bun !== "undefined") {
+  gc(true);
+  const { heapStats } = await import("bun:jsc");
+  const counts = heapStats().objectTypeCounts;
+  const interesting = Object.entries(counts)
+    .filter(([k]) => /Stream|Reader|Writer|Controller|Request|Promise|Function/i.test(k))
+    .sort((a, b) => b[1] - a[1])
+    .slice(0, 16);
+  console.log("\n# heapStats().objectTypeCounts after the workloads (top stream-related):");
+  for (const [k, v] of interesting) console.log(`  ${k}: ${v}`);
+}

bench/snippets/webstreams-tee.mjs

@@ -0,0 +1,107 @@
+// tee()/clone() throughput and memory: plain ReadableStream.tee, fetch Response.clone,
+// and Bun.serve Request.clone. Reports MB/s over the total bytes moved plus peak/settled
+// RSS growth for each scenario (best-of-RUNS for time; max for memory).
+const RUNS = 3;
+const MB = 1024 * 1024;
+const CHUNK = new Uint8Array(64 * 1024).fill(120);
+const gc = globalThis.Bun?.gc ?? globalThis.gc;
+if (!gc) throw new Error("This benchmark needs a GC hook: run with Bun, or node --expose-gc.");
+const rss = () => process.memoryUsage.rss();
+const fmt = n => (n / MB).toFixed(1).padStart(7) + " MB";
+
+const source = totalBytes => {
+  const count = Math.ceil(totalBytes / CHUNK.length);
+  let i = 0;
+  return new ReadableStream({
+    pull(c) {
+      if (i++ < count) c.enqueue(CHUNK);
+      else c.close();
+    },
+  });
+};
+const drain = async rs => {
+  const r = rs.getReader();
+  let n = 0;
+  while (true) {
+    const { done, value } = await r.read();
+    if (done) return n;
+    n += value.length;
+  }
+};
+
+async function bench(label, totalBytes, fn) {
+  await fn(); // warmup
+  let best = Infinity;
+  let peak = 0;
+  for (let i = 0; i < RUNS; i++) {
+    gc(true);
+    const before = rss();
+    let localPeak = before;
+    const timer = setInterval(() => {
+      localPeak = Math.max(localPeak, rss());
+    }, 5);
+    const t0 = performance.now();
+    await fn();
+    const elapsed = performance.now() - t0;
+    clearInterval(timer);
+    localPeak = Math.max(localPeak, rss());
+    best = Math.min(best, elapsed);
+    peak = Math.max(peak, localPeak - before);
+  }
+  const mbps = totalBytes / MB / (best / 1000);
+  console.log(`${label.padEnd(46)} ${mbps.toFixed(0).padStart(7)} MB/s   peak RSS +${fmt(peak)}`);
+}
+
+const TOTAL = 128 * MB;
+await bench("tee(): both branches drained concurrently", TOTAL * 2, async () => {
+  const [a, b] = source(TOTAL).tee();
+  await Promise.all([drain(a), drain(b)]);
+});
+await bench("tee(): branch B read only after A finishes", TOTAL * 2, async () => {
+  const [a, b] = source(TOTAL).tee();
+  await drain(a);
+  await drain(b);
+});
+await bench("tee(): read A, cancel B", TOTAL, async () => {
+  const [a, b] = source(TOTAL).tee();
+  const done = drain(a);
+  await b.cancel();
+  await done;
+});
+
+if (typeof Bun !== "undefined") {
+  const BODY_BYTES = 64 * MB;
+  await using server = Bun.serve({
+    port: 0,
+    async fetch(req) {
+      const url = new URL(req.url);
+      if (url.pathname === "/stream") return new Response(source(BODY_BYTES));
+      if (url.pathname === "/clone-echo") {
+        // Request.clone(): consume the body twice server-side.
+        const clone = req.clone();
+        const [a, b] = await Promise.all([req.arrayBuffer(), clone.arrayBuffer()]);
+        return new Response(String(a.byteLength + b.byteLength));
+      }
+      return new Response("nope", { status: 404 });
+    },
+  });
+  const base = `http://localhost:${server.port}`;
+
+  await bench("fetch(stream).clone(): read both bodies", BODY_BYTES * 2, async () => {
+    const response = await fetch(`${base}/stream`);
+    const clone = response.clone();
+    await Promise.all([response.arrayBuffer(), clone.arrayBuffer()]);
+  });
+  await bench("fetch(stream).clone(): read one, cancel clone", BODY_BYTES, async () => {
+    const response = await fetch(`${base}/stream`);
+    const clone = response.clone();
+    const read = response.arrayBuffer();
+    await clone.body.cancel();
+    await read;
+  });
+  const upload = new Uint8Array(32 * MB).fill(7);
+  await bench("Bun.serve: req.clone(), read both bodies", upload.length * 2, async () => {
+    const res = await fetch(`${base}/clone-echo`, { method: "POST", body: upload });
+    if ((await res.text()) !== String(upload.length * 2)) throw new Error("bad echo");
+  });
+}