#38886 Streams: one PipeReader loop with owned chunks, hold-not-adopt buffer pins, right-sized native pulls

node-scenarios.mjs — Node-API only, runs unchanged on node and bun:

// Node-API-only scenarios; runs unchanged on node and bun. usage: <runtime> node-scenarios.mjs <name>
import fs from "node:fs"; import { spawn } from "node:child_process"; import http from "node:http"; import { Readable } from "node:stream"; import { once } from "node:events";
const name = process.argv[2]; const MB = 1024 * 1024; const big = process.env.IOBENCH_BIG, html = process.env.IOBENCH_HTML, gz = process.env.IOBENCH_GZ, producer = process.env.IOBENCH_PRODUCER;
let bytes = 0; const t0 = performance.now();
const drain = async s => { for await (const c of s) bytes += c.length; };
const web = s => Readable.toWeb(s);
const serve = async (mk) => { const srv = http.createServer((req, res) => mk(req, res)); srv.listen(0); await once(srv, "listening"); return srv; };
switch (name) {
  case "fs-readstream-1g": await drain(fs.createReadStream(big)); break;                       // node stream, for-await
  case "fs-readstream-web-1g": await drain(web(fs.createReadStream(big))); break;              // → web stream
  case "fs-blob-stream-1g": await drain((await fs.openAsBlob(big)).stream()); break;            // Blob.stream()
  case "fs-readfile-1g": bytes = (await fs.promises.readFile(big)).length; break;
  case "cp-cat-1g": { const p = spawn("cat", [big]); await drain(p.stdout); break; }
  case "cp-cat-web-1g": { const p = spawn("cat", [big]); await drain(web(p.stdout)); break; }
  case "cp-tiny-chunks": { const p = spawn(producer, ["-e", "const l=Buffer.alloc(100,97);l[99]=10;for(let i=0;i<200000;i++)require('fs').writeSync(1,l)"]); await drain(p.stdout); break; }
  case "cp-64k-chunks": { const p = spawn(producer, ["-e", "const b=Buffer.alloc(65536,97);for(let i=0;i<8192;i++)require('fs').writeSync(1,b)"]); await drain(p.stdout); break; }
  case "cp-many-small": { for (let i = 0; i < 300; i++) { const p = spawn("/bin/echo", ["hello"]); for await (const c of p.stdout) bytes += c.length; } break; }
  case "stdin-1g": await drain(process.stdin); break;                                           // cat big | X
  case "http-fs-stream-1g": { const srv = await serve((q, r) => fs.createReadStream(big).pipe(r)); await drain((await fetch(`http://127.0.0.1:${srv.address().port}/`)).body); srv.close(); break; }
  case "http-cp-stdout": { const srv = await serve((q, r) => spawn("cat", [big]).stdout.pipe(r)); await drain((await fetch(`http://127.0.0.1:${srv.address().port}/`)).body); srv.close(); break; }
  case "http-upload-256m": { let got = 0; const srv = await serve(async (q, r) => { for await (const c of q) got += c.length; r.end("ok"); }); await fetch(`http://127.0.0.1:${srv.address().port}/`, { method: "POST", body: web(fs.createReadStream(big, { end: 256 * MB - 1 })), duplex: "half" }); bytes = got; srv.close(); break; }
  case "gzip-64m": await drain(web(fs.createReadStream(html)).pipeThrough(new CompressionStream("gzip"))); break;
  case "gunzip": await drain(web(fs.createReadStream(gz)).pipeThrough(new DecompressionStream("gzip"))); break;
  case "textdecoder-64m": { for await (const s of web(fs.createReadStream(html)).pipeThrough(new TextDecoderStream())) bytes += s.length; break; }
  case "transform-1g": await drain(web(fs.createReadStream(big)).pipeThrough(new TransformStream())); break;
  case "response-text-64m": bytes = (await new Response(web(fs.createReadStream(html))).text()).length; break;
  default: console.error("unknown " + name); process.exit(2);
}
const ms = performance.now() - t0;
console.log(JSON.stringify({ name, bytes, ms: +ms.toFixed(1), mbps: +((bytes / MB) / (ms / 1000)).toFixed(1) }));
process.exit(0);

scenarios.mjs — Bun APIs:

// usage: bun scenarios.mjs <name>   — prints JSON {name, bytes, ms, mbps}
const name = process.argv[2];
const MB = 1024 * 1024;
const t0 = performance.now();
let bytes = 0;
const done = extra => { const ms = performance.now() - t0; console.log(JSON.stringify({ name, bytes, ms: +ms.toFixed(1), mbps: +((bytes / MB) / (ms / 1000)).toFixed(1), ...extra })); process.exit(0); };

async function drain(stream) { for await (const c of stream) bytes += c.length; }
async function drainReader(stream) { const r = stream.getReader(); for (;;) { const { done: d, value } = await r.read(); if (d) break; bytes += value.length; } }

const big = process.env.IOBENCH_BIG;      // 1 GiB file
const small = process.env.IOBENCH_SMALL;  // 64 KiB file
switch (name) {
  case "file-stream-1g": await drain(Bun.file(big).stream()); break;                       // for-await over a big file
  case "file-reader-1g": await drainReader(Bun.file(big).stream()); break;                 // reader.read() loop
  case "file-text-1g": bytes = (await Bun.file(big).arrayBuffer()).byteLength; break;      // whole-file read
  case "file-small-x2000": for (let i = 0; i < 2000; i++) bytes += (await Bun.file(small).arrayBuffer()).byteLength; break;
  case "file-small-stream-x2000": for (let i = 0; i < 2000; i++) await drain(Bun.file(small).stream()); break;
  case "spawn-cat-1g": { const p = Bun.spawn(["cat", big], { stdout: "pipe" }); await drain(p.stdout); await p.exited; break; }
  case "spawn-cat-1g-text": { const p = Bun.spawn(["cat", big], { stdout: "pipe" }); bytes = (await p.stdout.bytes()).length; await p.exited; break; }
  case "spawn-tiny-chunks": { // child writes 200k × 100-byte lines with a flush between each
      const p = Bun.spawn([process.execPath, "-e", "const l=Buffer.alloc(100,97);l[99]=10;for(let i=0;i<200000;i++)require('fs').writeSync(1,l)"], { stdout: "pipe" }); await drain(p.stdout); await p.exited; break; }
  case "spawn-64k-chunks": { const p = Bun.spawn([process.execPath, "-e", "const b=Buffer.alloc(65536,97);for(let i=0;i<8192;i++)require('fs').writeSync(1,b)"], { stdout: "pipe" }); await drain(p.stdout); await p.exited; break; }
  case "spawn-many-small": { for (let i = 0; i < 300; i++) { const p = Bun.spawn(["/bin/echo", "hello"], { stdout: "pipe" }); bytes += (await p.stdout.text()).length; await p.exited; } break; }
  case "spawnsync-many-small": { for (let i = 0; i < 300; i++) bytes += Bun.spawnSync(["/bin/echo", "hello"]).stdout.length; break; }
  case "stdin-pipe-1g": await drain(Bun.stdin.stream()); break;                           // run as: cat big | bun scenarios.mjs stdin-pipe-1g
  case "spawn-stdin-stream-256m": { // parent streams a file into child's stdin, child counts
      const p = Bun.spawn([process.execPath, "-e", "let n=0;for await(const c of Bun.stdin.stream())n+=c.length;console.log(n)"], { stdin: Bun.file(big).slice(0, 256 * MB), stdout: "pipe" }); bytes = +(await p.stdout.text()); await p.exited; break; }
  case "http-serve-file-1g": { // serve Bun.file, fetch it back over loopback
      const s = Bun.serve({ port: 0, fetch: () => new Response(Bun.file(big)) }); await drain((await fetch(s.url)).body); s.stop(true); break; }
  case "http-serve-spawn-stdout": { // proxy a child's stdout as the response body
      const s = Bun.serve({ port: 0, fetch: () => new Response(Bun.spawn(["cat", big], { stdout: "pipe" }).stdout) }); await drain((await fetch(s.url)).body); s.stop(true); break; }
  case "http-fetch-to-spawn-stdin": { // fetch body → child stdin
      const s = Bun.serve({ port: 0, fetch: () => new Response(Bun.file(big).slice(0, 256 * MB)) });
      const p = Bun.spawn([process.execPath, "-e", "let n=0;for await(const c of Bun.stdin.stream())n+=c.length;console.log(n)"], { stdin: (await fetch(s.url)).body, stdout: "pipe" }); bytes = +(await p.stdout.text()); await p.exited; s.stop(true); break; }
  case "htmlrewriter-file-64m": { const r = new HTMLRewriter().on("p", { element(e) { e.setAttribute("x", "1"); } }).transform(new Response(Bun.file(process.env.IOBENCH_HTML))); bytes = (await r.arrayBuffer()).byteLength; break; }
  case "file-gzip-stream-64m": { // File → CompressionStream → count
      await drain(Bun.file(process.env.IOBENCH_HTML).stream().pipeThrough(new CompressionStream("gzip"))); break; }
  case "file-gunzip-stream": { // gz File → DecompressionStream → count
      await drain(Bun.file(process.env.IOBENCH_GZ).stream().pipeThrough(new DecompressionStream("gzip"))); break; }
  case "file-textdecoder-lines-256m": { // File → TextDecoderStream → split lines in JS
      let lines = 0; for await (const s of Bun.file(process.env.IOBENCH_HTML).stream().pipeThrough(new TextDecoderStream())) { bytes += s.length; lines += s.split("<").length; } break; }
  case "file-transform-1g": { // File → identity TransformStream → count
      await drain(Bun.file(big).stream().pipeThrough(new TransformStream())); break; }
  case "file-response-text-256m": { bytes = (await new Response(Bun.file(big).slice(0, 256 * MB)).text()).length; break; }
  case "file-tee-1g": { const [a, b] = Bun.file(big).stream().tee(); await Promise.all([drain(a), drain(b)]); break; }
  case "http-serve-file-gzip-64m": { // Bun.serve: File → CompressionStream response; client: DecompressionStream
      const s = Bun.serve({ port: 0, fetch: () => new Response(Bun.file(process.env.IOBENCH_HTML).stream().pipeThrough(new CompressionStream("gzip")), { headers: { "content-encoding": "identity" } }) });
      await drain((await fetch(s.url)).body.pipeThrough(new DecompressionStream("gzip"))); s.stop(true); break; }
  case "http-serve-file-range-x200": { // many 1 MiB ranged responses off one big file
      const s = Bun.serve({ port: 0, fetch: req => { const i = +new URL(req.url).pathname.slice(1); return new Response(Bun.file(big).slice(i * MB, (i + 1) * MB)); } });
      for (let i = 0; i < 200; i++) bytes += (await (await fetch(`${s.url}${i}`)).arrayBuffer()).byteLength; s.stop(true); break; }
  case "http-upload-file-256m": { // fetch(body: Bun.file) → server drains req.body
      let got = 0; const s = Bun.serve({ port: 0, maxRequestBodySize: 1e12, async fetch(req) { for await (const c of req.body) got += c.length; return new Response("ok"); } });
      await fetch(s.url, { method: "POST", body: Bun.file(big).slice(0, 256 * MB) }); bytes = got; s.stop(true); break; }
  case "spawn-stdout-to-file-1g": { // child stdout → Bun.write(file)
      const p = Bun.spawn(["cat", big], { stdout: "pipe" }); bytes = await Bun.write(process.env.IOBENCH_OUT, new Response(p.stdout)); await p.exited; break; }
  case "spawn-pipeline-gzip": { // cat big | gzip -1 (child) → parent DecompressionStream
      const p = Bun.spawn(["sh", "-c", `gzip -1 -c ${process.env.IOBENCH_HTML}`], { stdout: "pipe" }); await drain(p.stdout.pipeThrough(new DecompressionStream("gzip"))); await p.exited; break; }
  case "spawn-stderr-interleaved": { // child alternates 4 KiB writes to stdout/stderr, parent drains both
      const p = Bun.spawn([process.execPath, "-e", "const b=Buffer.alloc(4096,97);for(let i=0;i<20000;i++){require('fs').writeSync(1,b);require('fs').writeSync(2,b)}"], { stdout: "pipe", stderr: "pipe" });
      const [a, b] = await Promise.all([p.stdout.bytes(), p.stderr.bytes()]); bytes = a.length + b.length; await p.exited; break; }
  default: console.error("unknown scenario " + name); process.exit(2);
}
done({ rssMB: +(process.memoryUsage.rss() / MB).toFixed(1) });

run3.sh:

#!/bin/bash
# usage: ./run3.sh <node> <bunBase> <bunPR>
set -u; N=$1; A=$2; B=$3; D=${IOBENCH_DIR:-/tmp/iobench-data}
export IOBENCH_BIG=$D/big IOBENCH_HTML=$D/page.html IOBENCH_GZ=$D/big.gz IOBENCH_PRODUCER=$B
TIME=$(command -v gtime || echo /usr/bin/time)
S="fs-readstream-1g fs-readstream-web-1g fs-blob-stream-1g fs-readfile-1g cp-cat-1g cp-cat-web-1g cp-tiny-chunks cp-64k-chunks cp-many-small stdin-1g http-fs-stream-1g http-cp-stdout http-upload-256m gzip-64m gunzip textdecoder-64m transform-1g response-text-64m"
printf "%-22s | %8s %7s %5s | %8s %7s %5s | %8s %7s %5s\n" scenario "node ms" MB/s RSS "base ms" MB/s RSS "PR ms" MB/s RSS
for s in $S; do row=""; for bin in $N $A $B; do
  if [ $s = stdin-1g ]; then out=$( { timeout 300 $TIME -v sh -c "cat $D/big | $bin node-scenarios.mjs $s"; } 2>&1 ); else out=$( { timeout 300 $TIME -v $bin node-scenarios.mjs $s; } 2>&1 ); fi
  json=$(echo "$out" | grep '^{' | tail -1); rss=$(echo "$out" | awk '/Maximum resident/ {printf "%.0f", $6/1024}')
  ms=$(echo "$json" | sed -nE 's/.*"ms":([0-9.]+).*/\1/p'); mbps=$(echo "$json" | sed -nE 's/.*"mbps":([0-9.]+).*/\1/p')
  row="$row | $(printf "%8s %7s %5s" "${ms:--}" "${mbps:--}" "${rss:--}")"; done
  printf "%-22s%s\n" $s "$row"; done

run.sh:

#!/bin/bash
# usage: ./run.sh <binA> <binB>   (labels A=first B=second). Needs IOBENCH_DIR with big/small/html files.
set -u
A=$1; B=$2; D=${IOBENCH_DIR:-/tmp/iobench-data}
mkdir -p $D
[ -f $D/big ] || head -c $((1024*1024*1024)) /dev/urandom > $D/big
[ -f $D/small ] || head -c 65536 /dev/urandom > $D/small
[ -f $D/page.html ] || $A -e "require('fs').writeFileSync('$D/page.html', Buffer.alloc(64*1024*1024, '<p>'+'a'.repeat(1000)+'</p>'))"
[ -f $D/big.gz ] || gzip -1 -c $D/page.html > $D/big.gz
export IOBENCH_BIG=$D/big IOBENCH_SMALL=$D/small IOBENCH_HTML=$D/page.html IOBENCH_GZ=$D/big.gz IOBENCH_OUT=$D/out.bin
TIME=$(command -v gtime || command -v /usr/bin/time)
S="${SCENARIOS:-file-stream-1g file-reader-1g file-text-1g file-small-x2000 file-small-stream-x2000 spawn-cat-1g spawn-cat-1g-text spawn-tiny-chunks spawn-64k-chunks spawn-many-small spawnsync-many-small stdin-pipe-1g spawn-stdin-stream-256m http-serve-file-1g http-serve-spawn-stdout http-fetch-to-spawn-stdin htmlrewriter-file-64m file-gzip-stream-64m file-gunzip-stream file-textdecoder-lines-256m file-transform-1g file-response-text-256m file-tee-1g http-serve-file-gzip-64m http-serve-file-range-x200 http-upload-file-256m spawn-stdout-to-file-1g spawn-pipeline-gzip spawn-stderr-interleaved}"
printf "%-28s | %10s %8s %8s | %10s %8s %8s\n" scenario "A ms" "A MB/s" "A maxRSS" "B ms" "B MB/s" "B maxRSS"
for s in $S; do
  for bin in $A $B; do
    if [ $s = stdin-pipe-1g ]; then out=$( { $TIME -v sh -c "cat $D/big | $bin scenarios.mjs $s"; } 2>&1 ); else out=$( { $TIME -v $bin scenarios.mjs $s; } 2>&1 ); fi
    json=$(echo "$out" | grep '^{' | tail -1); rss=$(echo "$out" | awk '/Maximum resident/ {printf "%.0f", $6/1024}')
    ms=$(echo "$json" | sed -E 's/.*"ms":([0-9.]+).*/\1/'); mbps=$(echo "$json" | sed -E 's/.*"mbps":([0-9.]+).*/\1/')
    eval "R_$( [ $bin = $A ] && echo A || echo B )=\"$ms $mbps $rss\""
  done
  printf "%-28s | %10s %8s %8s | %10s %8s %8s\n" $s $R_A $R_B
done