Bun 1.4

// End-to-end pipelines between native stream types (fetch body, DecompressionStream, TextDecoderStream,
// file streams via node:stream Readable/Writable.toWeb, child_process pipes). Portable: Bun, Node, Deno.
// Run: <runtime> native-pipeline.mjs --scenario=prep                       (writes the 64 MiB fixture files once)
//      <runtime> native-pipeline.mjs --scenario=download-gunzip-decode --server=http://127.0.0.1:39872
//      <runtime> native-pipeline.mjs --scenario=file-gzip-upload --server=http://127.0.0.1:39872
//      <runtime> native-pipeline.mjs --scenario=file-decode-encode-file
//      <runtime> native-pipeline.mjs --scenario=spawn-passthrough --server=http://127.0.0.1:39872
// Server: `bun run serve-body.mjs --gzip`. 64 MiB payloads/files, 4 KiB chunks end to end. Wrap in /usr/bin/time -v for peak RSS.
import fs from "node:fs";
import { Readable, Writable } from "node:stream";
import { spawn } from "node:child_process";

const MB = 1024 * 1024;
const CHUNK = 4096;
const BYTES = 64 * MB;
const argv = globalThis.process?.argv ?? [];
const arg = (k, d) =>
  argv.find((a) => a.startsWith(`--${k}=`))?.slice(k.length + 3) ?? d;
const scenario = arg("scenario", "prep");
const server = arg("server");
const dir = arg("dir", "/tmp/native-pipeline");
const JSON_FILE = `${dir}/json-64mb.txt`;
const UTF8_FILE = `${dir}/utf8-64mb.txt`;
const OUT_FILE = `${dir}/out-${scenario}.txt`;

const jsonTemplate = new TextEncoder()
  .encode(
    JSON.stringify({
      messages: Array.from({ length: 700 }, (_, i) => ({
        id: i,
        role: i % 2 ? "assistant" : "user",
        ts: 1700000000 + i,
        body: "the quick brown fox jumps over the lazy dog " + i,
      })),
    }),
  )
  .slice(0, CHUNK);
const utf8Text = (
  "hello world \u{1F30A} stream ✨ café naïve 中文 " + "x".repeat(40)
).repeat(2000);
const utf8Template = new TextEncoder().encode(utf8Text).slice(0, CHUNK);
// Keep the UTF-8 fixture valid at 64 KiB chunk boundaries: cut the template at a char boundary.
const utf8Chunk = (() => {
  let end = utf8Template.byteLength;
  while ((utf8Template[end - 1] & 0xc0) === 0x80) end--;
  if (end < utf8Template.byteLength) end--; // drop the lead byte of the truncated char too
  return utf8Template.slice(0, end);
})();

const countBytes = async (rs) => {
  let n = 0;
  for await (const v of rs)
    n += typeof v === "string" ? v.length : v.byteLength;
  return n;
};

async function prep() {
  fs.mkdirSync(dir, { recursive: true });
  const write = (path, template, varyByte) => {
    if (fs.existsSync(path) && fs.statSync(path).size === BYTES) return;
    const fd = fs.openSync(path, "w");
    let written = 0,
      i = 0;
    const buf = new Uint8Array(template.byteLength);
    while (written < BYTES) {
      buf.set(template);
      if (varyByte) buf[0] = 32 + (i++ & 63);
      fs.writeSync(fd, buf);
      written += buf.byteLength;
    }
    fs.closeSync(fd);
    console.log(`wrote ${path} (${(written / MB).toFixed(0)} MiB)`);
  };
  write(JSON_FILE, jsonTemplate, true);
  write(UTF8_FILE, utf8Chunk, false);
}

const scenarios = {
  // fetch(gzip body) -> DecompressionStream -> TextDecoderStream -> for await (count chars). MB/s over decompressed bytes.
  "download-gunzip-decode": async () => {
    const res = await fetch(`${server}/gzip`);
    const expected = +res.headers.get("x-uncompressed-length");
    const chars = await countBytes(
      res.body
        .pipeThrough(new DecompressionStream("gzip"))
        .pipeThrough(new TextDecoderStream()),
    );
    if (chars !== expected)
      throw new Error(
        `decoded ${chars} chars, expected ${expected} (ASCII payload)`,
      );
    return expected;
  },
  // fs.createReadStream(64 MiB, 4 KiB reads) -> CompressionStream -> fetch POST body; server returns bytes received. MB/s over input bytes.
  "file-gzip-upload": async () => {
    const body = Readable.toWeb(
      fs.createReadStream(JSON_FILE, { highWaterMark: CHUNK }),
    ).pipeThrough(new CompressionStream("gzip"));
    const res = await fetch(`${server}/upload`, {
      method: "POST",
      body,
      duplex: "half",
    });
    const received = +(await res.text());
    if (!(received > 0 && received < BYTES))
      throw new Error(`server received ${received} bytes`);
    return fs.statSync(JSON_FILE).size;
  },
  // fs.createReadStream(64 MiB utf-8, 4 KiB reads) -> TextDecoderStream -> TextEncoderStream -> fs.createWriteStream. MB/s over file bytes.
  "file-decode-encode-file": async () => {
    await Readable.toWeb(
      fs.createReadStream(UTF8_FILE, { highWaterMark: CHUNK }),
    )
      .pipeThrough(new TextDecoderStream())
      .pipeThrough(new TextEncoderStream())
      .pipeTo(Writable.toWeb(fs.createWriteStream(OUT_FILE)));
    const n = fs.statSync(OUT_FILE).size;
    if (n !== fs.statSync(UTF8_FILE).size) throw new Error(`wrote ${n} bytes`);
    fs.unlinkSync(OUT_FILE);
    return n;
  },
  // fetch(64 MiB body in 4 KiB chunks).body -> cat stdin ; cat stdout -> for await. MB/s over body bytes.
  "spawn-passthrough": async () => {
    const child = spawn("cat", [], { stdio: ["pipe", "pipe", "inherit"] });
    const res = await fetch(`${server}/?bytes=${BYTES}&chunk=${CHUNK}`);
    const [, n] = await Promise.all([
      res.body.pipeTo(Writable.toWeb(child.stdin)),
      countBytes(Readable.toWeb(child.stdout)),
    ]);
    await new Promise((r) => child.on("close", r));
    if (n !== BYTES) throw new Error(`got ${n} bytes from cat`);
    return n;
  },
};

if (scenario === "prep") {
  await prep();
} else {
  const fn = scenarios[scenario];
  if (!fn)
    throw new Error(
      `unknown --scenario=${scenario}; prep | ${Object.keys(scenarios).join(
        " | ",
      )}`,
    );
  if (scenario !== "file-decode-encode-file" && !server)
    throw new Error("--server=URL required (bun run serve-body.mjs --gzip)");
  const t0 = performance.now();
  const bytes = await fn();
  const ms = performance.now() - t0;
  console.log(
    `${scenario.padEnd(26)} ${(bytes / MB / (ms / 1000))
      .toFixed(0)
      .padStart(6)} MB/s  ${ms.toFixed(0).padStart(6)} ms  ${(
      bytes / MB
    ).toFixed(0)} MiB`,
  );
}

// Streaming-body server for streams-throughput.mjs --scenario=fetch and native-pipeline.mjs.
// Run: bun run serve-body.mjs [--gzip]   (listens on 127.0.0.1:39872)
//   GET  /?bytes=N&chunk=C   fresh C-byte chunks (default 65536), N bytes total
//   GET  /gzip                64 MiB of JSON-like text gzip-compressed once at startup (--gzip), served in 4 KiB chunks,
//                             no content-encoding header (the client decompresses explicitly)
//   POST /upload     drains the request body, responds with the byte count
const MB = 1024 * 1024;
const CHUNK = 64 * 1024;
const argv = process.argv;
const GZIP_BYTES = 64 * MB;
const GZIP_CHUNK = 4096;

const jsonTemplate = new TextEncoder()
  .encode(
    JSON.stringify({
      messages: Array.from({ length: 700 }, (_, i) => ({
        id: i,
        role: i % 2 ? "assistant" : "user",
        ts: 1700000000 + i,
        body: "the quick brown fox jumps over the lazy dog " + i,
      })),
    }),
  )
  .slice(0, CHUNK);
const jsonSource = (total) => {
  const count = Math.ceil(total / CHUNK);
  let i = 0;
  return new ReadableStream({
    pull(c) {
      if (i < count) {
        const b = new Uint8Array(CHUNK);
        b.set(jsonTemplate);
        b[0] = 32 + (i++ & 63);
        c.enqueue(b);
      } else c.close();
    },
  });
};
let gzipped = null;
if (argv.includes("--gzip")) {
  const t0 = performance.now();
  gzipped = new Uint8Array(
    await new Response(
      jsonSource(GZIP_BYTES).pipeThrough(new CompressionStream("gzip")),
    ).arrayBuffer(),
  );
  console.log(
    `pre-compressed ${GZIP_BYTES / MB} MiB -> ${(
      gzipped.byteLength / MB
    ).toFixed(1)} MiB gzip in ${(performance.now() - t0).toFixed(0)} ms`,
  );
}

Bun.serve({
  port: 39872,
  hostname: "127.0.0.1",
  idleTimeout: 255,
  maxRequestBodySize: 8 * 1024 * MB,
  async fetch(req) {
    const url = new URL(req.url);
    if (req.method === "POST" && url.pathname === "/upload") {
      let n = 0;
      for await (const c of req.body) n += c.byteLength;
      return new Response(String(n));
    }
    if (url.pathname === "/gzip") {
      if (!gzipped) return new Response("start with --gzip", { status: 500 });
      let off = 0;
      const body = new ReadableStream({
        pull(c) {
          if (off < gzipped.byteLength) {
            c.enqueue(
              gzipped.slice(
                off,
                Math.min(off + GZIP_CHUNK, gzipped.byteLength),
              ),
            );
            off += GZIP_CHUNK;
          } else c.close();
        },
      });
      return new Response(body, {
        headers: {
          "content-type": "application/gzip",
          "x-uncompressed-length": String(GZIP_BYTES),
        },
      });
    }
    const total = +url.searchParams.get("bytes");
    const chunk = +(url.searchParams.get("chunk") ?? CHUNK);
    const count = Math.ceil(total / chunk);
    let i = 0;
    const body = new ReadableStream({
      pull(c) {
        if (i < count) c.enqueue(new Uint8Array(chunk).fill(i++ & 0xff));
        else c.close();
      },
    });
    return new Response(body, { headers: { "content-length": String(total) } });
  },
});
console.log("listening on http://127.0.0.1:39872");

// Response.clone() and ReadableStream.tee() with fresh 64 KiB buffers. Peak RSS (via /usr/bin/time -v) is the point.
// Run: bun run response-clone.mjs --scenario=clone-both --bytes=67108864
//      node response-clone.mjs --scenario=clone-chain --depth=100 --bytes=104857600
//      deno run -A response-clone.mjs --scenario=tee --bytes=2147483648
// clone-both:  res.clone(), then read both bodies concurrently.
// clone-only:  res.clone(), read only the clone; the original is never read.
// clone-chain: clone a streaming Response N times, read only the last clone.
// tee:         split a stream and drain both branches concurrently. MB/s is over the source bytes.
const MB = 1024 * 1024;
const CHUNK = 64 * 1024;
const argv = globalThis.process?.argv ?? [];
const arg = (k, d) =>
  argv.find((a) => a.startsWith(`--${k}=`))?.slice(k.length + 3) ?? d;
const scenario = arg("scenario", "clone-chain");
const DEPTH = +arg("depth", 100);
const BYTES = +arg(
  "bytes",
  { "tee": 2048 * MB, "clone-chain": 100 * MB }[scenario] ?? 1024 * MB,
);

const freshSource = (total) => {
  const count = Math.ceil(total / CHUNK);
  let i = 0;
  return new ReadableStream({
    pull(c) {
      if (i < count) c.enqueue(new Uint8Array(CHUNK).fill(i++ & 0xff));
      else c.close();
    },
  });
};
const drain = async (rs) => {
  const r = rs.getReader();
  let n = 0;
  for (;;) {
    const { done, value } = await r.read();
    if (done) return n;
    n += value.byteLength;
  }
};

const scenarios = {
  "clone-both": async () => {
    const res = new Response(freshSource(BYTES));
    const c = res.clone();
    const [a, b] = await Promise.all([res.arrayBuffer(), c.arrayBuffer()]);
    if (a.byteLength !== b.byteLength) throw new Error("clone mismatch");
    return a.byteLength;
  },
  "clone-only": async () => {
    const res = new Response(freshSource(BYTES));
    const c = res.clone();
    return (await c.arrayBuffer()).byteLength;
  },
  "clone-chain": async () => {
    let cur = new Response(freshSource(BYTES));
    const chain = [cur];
    for (let i = 0; i < DEPTH; i++) chain.push((cur = cur.clone()));
    return (await chain.at(-1).arrayBuffer()).byteLength;
  },
  "tee": async () => {
    const [a, b] = freshSource(BYTES).tee();
    const [x, y] = await Promise.all([drain(a), drain(b)]);
    if (x !== y) throw new Error("branch mismatch");
    return x;
  },
};
const fn = scenarios[scenario];
if (!fn)
  throw new Error(
    `unknown --scenario=${scenario}; clone-both | clone-only | clone-chain | tee`,
  );

const rss0 = globalThis.process?.memoryUsage?.().rss ?? 0;
const t0 = performance.now();
const got = await fn();
const ms = performance.now() - t0;
if (got !== BYTES)
  throw new Error(`${scenario}: read ${got} bytes, expected ${BYTES}`);
const rssDelta = ((globalThis.process?.memoryUsage?.().rss ?? 0) - rss0) / MB;
console.log(
  `${scenario.padEnd(12)} ${(BYTES / MB / (ms / 1000))
    .toFixed(0)
    .padStart(6)} MB/s  ${ms.toFixed(0).padStart(6)} ms  ${
    BYTES / MB
  } MiB  rss +${rssDelta.toFixed(0)} MB`,
);

// CompressionStream / DecompressionStream throughput on JSON-like text, generated in fresh 64 KiB chunks.
// Run: bun run compression-stream.mjs --scenario=compress --format=gzip --bytes=1073741824
//      node compression-stream.mjs --scenario=decompress --format=deflate
//      deno run -A compression-stream.mjs --scenario=compress
// MB/s is over uncompressed bytes. `decompress` compresses the input first (untimed), then times the inflate.
const MB = 1024 * 1024;
const CHUNK = 64 * 1024;
const argv = globalThis.process?.argv ?? [];
const arg = (k, d) =>
  argv.find((a) => a.startsWith(`--${k}=`))?.slice(k.length + 3) ?? d;
const scenario = arg("scenario", "compress");
const format = arg("format", "gzip");
const BYTES = +arg("bytes", 1024 * MB);

const template = new TextEncoder()
  .encode(
    JSON.stringify({
      messages: Array.from({ length: 700 }, (_, i) => ({
        id: i,
        role: i % 2 ? "assistant" : "user",
        ts: 1700000000 + i,
        body: "the quick brown fox jumps over the lazy dog " + i,
      })),
    }),
  )
  .slice(0, CHUNK);

const jsonSource = (total) => {
  const count = Math.ceil(total / CHUNK);
  let i = 0;
  return new ReadableStream({
    pull(c) {
      if (i < count) {
        const b = new Uint8Array(CHUNK);
        b.set(template);
        b[0] = 32 + (i++ & 63);
        c.enqueue(b);
      } else c.close();
    },
  });
};
const drain = async (rs) => {
  const r = rs.getReader();
  let n = 0;
  for (;;) {
    const { done, value } = await r.read();
    if (done) return n;
    n += value.byteLength;
  }
};
const collect = async (rs) => {
  const parts = [];
  const r = rs.getReader();
  for (;;) {
    const { done, value } = await r.read();
    if (done) break;
    parts.push(value);
  }
  const out = new Uint8Array(parts.reduce((n, p) => n + p.byteLength, 0));
  let off = 0;
  for (const p of parts) out.set(p, off), (off += p.byteLength);
  return out;
};
const chunked = (buf) =>
  new ReadableStream({
    start(c) {
      for (let i = 0; i < buf.byteLength; i += CHUNK)
        c.enqueue(buf.subarray(i, Math.min(i + CHUNK, buf.byteLength)));
      c.close();
    },
  });

let run;
if (scenario === "compress")
  run = () =>
    drain(jsonSource(BYTES).pipeThrough(new CompressionStream(format)));
else if (scenario === "decompress") {
  const compressed = await collect(
    jsonSource(BYTES).pipeThrough(new CompressionStream(format)),
  );
  run = () =>
    drain(chunked(compressed).pipeThrough(new DecompressionStream(format)));
} else throw new Error(`unknown --scenario=${scenario}; compress | decompress`);

const t0 = performance.now();
const got = await run();
const ms = performance.now() - t0;
if (scenario === "decompress" && got !== BYTES)
  throw new Error(`inflated ${got} bytes, expected ${BYTES}`);
console.log(
  `${scenario} (${format})`.padEnd(22) +
    ` ${(BYTES / MB / (ms / 1000)).toFixed(0).padStart(6)} MB/s  ${ms
      .toFixed(0)
      .padStart(6)} ms  ${BYTES / MB} MiB`,
);

// TextEncoderStream / TextDecoderStream throughput on mixed multi-byte UTF-8, fresh 64 KiB chunks.
// Run: bun run text-encoder-stream.mjs --scenario=encode --bytes=1073741824
//      node text-encoder-stream.mjs --scenario=decode
//      deno run -A text-encoder-stream.mjs --scenario=encode
// MB/s is over UTF-8 bytes (encoder output / decoder input).
const MB = 1024 * 1024;
const CHUNK = 64 * 1024;
const argv = globalThis.process?.argv ?? [];
const arg = (k, d) =>
  argv.find((a) => a.startsWith(`--${k}=`))?.slice(k.length + 3) ?? d;
const scenario = arg("scenario", "encode");
const BYTES = +arg("bytes", 1024 * MB);

const text = (
  "hello world \u{1F30A} stream ✨ café naïve 中文 " + "x".repeat(40)
).repeat(2000);
const utf8Template = new TextEncoder().encode(text).slice(0, CHUNK);
const stringChunk = text.slice(0, CHUNK);
const stringChunkBytes = new TextEncoder().encode(stringChunk).byteLength;

const stringSource = (total) => {
  const count = Math.ceil(total / stringChunkBytes);
  let i = 0;
  return new ReadableStream({
    pull(c) {
      if (i < count)
        c.enqueue(String(i++ & 0xffff).padStart(5, "0") + stringChunk.slice(5));
      else c.close();
    },
  });
};
const bytesSource = (total) => {
  const count = Math.ceil(total / CHUNK);
  let i = 0;
  return new ReadableStream({
    pull(c) {
      if (i < count) {
        const b = new Uint8Array(CHUNK);
        b.set(utf8Template);
        b[0] = 32 + (i++ & 63);
        c.enqueue(b);
      } else c.close();
    },
  });
};
const drain = async (rs) => {
  const r = rs.getReader();
  let n = 0;
  for (;;) {
    const { done, value } = await r.read();
    if (done) return n;
    n += typeof value === "string" ? value.length : value.byteLength;
  }
};

let run, expected;
if (scenario === "encode") {
  const count = Math.ceil(BYTES / stringChunkBytes);
  expected = count * stringChunkBytes;
  run = () => drain(stringSource(BYTES).pipeThrough(new TextEncoderStream()));
} else if (scenario === "decode") {
  expected = null; // output is chars, not bytes
  run = () => drain(bytesSource(BYTES).pipeThrough(new TextDecoderStream()));
} else throw new Error(`unknown --scenario=${scenario}; encode | decode`);

const t0 = performance.now();
const got = await run();
const ms = performance.now() - t0;
if (expected !== null && got !== expected)
  throw new Error(`encoded ${got} bytes, expected ${expected}`);
const bytes = expected ?? Math.ceil(BYTES / CHUNK) * CHUNK;
console.log(
  `${scenario.padEnd(22)} ${(bytes / MB / (ms / 1000))
    .toFixed(0)
    .padStart(6)} MB/s  ${ms.toFixed(0).padStart(6)} ms  ${(bytes / MB).toFixed(
    0,
  )} MiB`,
);