stream: prototype for new stream implementation

Author: jasnell

benchmark/fs/bench-filehandle-pull-vs-webstream.js

@@ -0,0 +1,196 @@
+// Compare FileHandle.createReadStream() vs readableWebStream() vs pull()
+// reading a large file through two transforms: uppercase then gzip compress.
+'use strict';
+
+const common = require('../common.js');
+const fs = require('fs');
+const zlib = require('zlib');
+const { Transform, Writable, pipeline } = require('stream');
+
+const tmpdir = require('../../test/common/tmpdir');
+tmpdir.refresh();
+const filename = tmpdir.resolve(`.removeme-benchmark-garbage-${process.pid}`);
+
+const bench = common.createBenchmark(main, {
+  api: ['classic', 'webstream', 'pull'],
+  filesize: [1024 * 1024, 16 * 1024 * 1024, 64 * 1024 * 1024],
+  n: [5],
+});
+
+function main({ api, filesize, n }) {
+  // Create the fixture file with repeating lowercase ASCII
+  const chunk = Buffer.alloc(Math.min(filesize, 64 * 1024), 'abcdefghij');
+  const fd = fs.openSync(filename, 'w');
+  let remaining = filesize;
+  while (remaining > 0) {
+    const toWrite = Math.min(remaining, chunk.length);
+    fs.writeSync(fd, chunk, 0, toWrite);
+    remaining -= toWrite;
+  }
+  fs.closeSync(fd);
+
+  if (api === 'classic') {
+    benchClassic(n, filesize).then(() => cleanup());
+  } else if (api === 'webstream') {
+    benchWebStream(n, filesize).then(() => cleanup());
+  } else {
+    benchPull(n, filesize).then(() => cleanup());
+  }
+}
+
+function cleanup() {
+  try { fs.unlinkSync(filename); } catch { /* ignore */ }
+}
+
+// ---------------------------------------------------------------------------
+// Classic streams path: createReadStream -> Transform (upper) -> createGzip
+// ---------------------------------------------------------------------------
+async function benchClassic(n, filesize) {
+  // Warm up
+  await runClassic();
+
+  bench.start();
+  let totalBytes = 0;
+  for (let i = 0; i < n; i++) {
+    totalBytes += await runClassic();
+  }
+  bench.end(totalBytes / (1024 * 1024));
+}
+
+function runClassic() {
+  return new Promise((resolve, reject) => {
+    const rs = fs.createReadStream(filename);
+
+    // Transform 1: uppercase
+    const upper = new Transform({
+      transform(chunk, encoding, callback) {
+        const buf = Buffer.allocUnsafe(chunk.length);
+        for (let i = 0; i < chunk.length; i++) {
+          const b = chunk[i];
+          buf[i] = (b >= 0x61 && b <= 0x7a) ? b - 0x20 : b;
+        }
+        callback(null, buf);
+      },
+    });
+
+    // Transform 2: gzip
+    const gz = zlib.createGzip();
+
+    // Sink: count compressed bytes
+    let totalBytes = 0;
+    const sink = new Writable({
+      write(chunk, encoding, callback) {
+        totalBytes += chunk.length;
+        callback();
+      },
+    });
+
+    pipeline(rs, upper, gz, sink, (err) => {
+      if (err) reject(err);
+      else resolve(totalBytes);
+    });
+  });
+}
+
+// ---------------------------------------------------------------------------
+// WebStream path: readableWebStream -> TransformStream (upper) -> CompressionStream
+// ---------------------------------------------------------------------------
+async function benchWebStream(n, filesize) {
+  // Warm up
+  await runWebStream();
+
+  bench.start();
+  let totalBytes = 0;
+  for (let i = 0; i < n; i++) {
+    totalBytes += await runWebStream();
+  }
+  bench.end(totalBytes / (1024 * 1024));
+}
+
+async function runWebStream() {
+  const fh = await fs.promises.open(filename, 'r');
+  try {
+    const rs = fh.readableWebStream();
+
+    // Transform 1: uppercase
+    const upper = new TransformStream({
+      transform(chunk, controller) {
+        const buf = new Uint8Array(chunk.length);
+        for (let i = 0; i < chunk.length; i++) {
+          const b = chunk[i];
+          // a-z (0x61-0x7a) -> A-Z (0x41-0x5a)
+          buf[i] = (b >= 0x61 && b <= 0x7a) ? b - 0x20 : b;
+        }
+        controller.enqueue(buf);
+      },
+    });
+
+    // Transform 2: gzip via CompressionStream
+    const compress = new CompressionStream('gzip');
+
+    const output = rs.pipeThrough(upper).pipeThrough(compress);
+    const reader = output.getReader();
+
+    let totalBytes = 0;
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      totalBytes += value.byteLength;
+    }
+    return totalBytes;
+  } finally {
+    await fh.close();
+  }
+}
+
+// ---------------------------------------------------------------------------
+// New streams path: pull() with uppercase transform + gzip transform
+// ---------------------------------------------------------------------------
+async function benchPull(n, filesize) {
+  const { pull, compressGzip } = require('stream/new');
+
+  // Warm up
+  await runPull(pull, compressGzip);
+
+  bench.start();
+  let totalBytes = 0;
+  for (let i = 0; i < n; i++) {
+    totalBytes += await runPull(pull, compressGzip);
+  }
+  bench.end(totalBytes / (1024 * 1024));
+}
+
+async function runPull(pull, compressGzip) {
+  const fh = await fs.promises.open(filename, 'r');
+  try {
+    // Stateless transform: uppercase each chunk in the batch
+    const upper = (chunks) => {
+      if (chunks === null) return null;
+      const out = new Array(chunks.length);
+      for (let j = 0; j < chunks.length; j++) {
+        const src = chunks[j];
+        const buf = new Uint8Array(src.length);
+        for (let i = 0; i < src.length; i++) {
+          const b = src[i];
+          buf[i] = (b >= 0x61 && b <= 0x7a) ? b - 0x20 : b;
+        }
+        out[j] = buf;
+      }
+      return out;
+    };
+
+    const readable = fh.pull(upper, compressGzip());
+
+    // Count bytes symmetrically with the classic path (no final
+    // concatenation into a single buffer).
+    let totalBytes = 0;
+    for await (const chunks of readable) {
+      for (let i = 0; i < chunks.length; i++) {
+        totalBytes += chunks[i].byteLength;
+      }
+    }
+    return totalBytes;
+  } finally {
+    await fh.close();
+  }
+}

doc/api/fs.md

@@ -377,6 +377,61 @@ added: v10.0.0
 
 * Type: {number} The numeric file descriptor managed by the {FileHandle} object.
 
+#### `filehandle.pull([...transforms][, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `...transforms` {Function|Object} Optional transforms to apply via
+  [`stream/new pull()`][].
+* `options` {Object}
+  * `signal` {AbortSignal}
+  * `autoClose` {boolean} Close the file handle when the stream ends.
+    **Default:** `false`.
+* Returns: {AsyncIterable\<Uint8Array\[]>}
+
+Return the file contents as an async iterable using the
+[`node:stream/new`][] pull model. Reads are performed in 64 KB chunks.
+If transforms are provided, they are applied via [`stream/new pull()`][].
+
+The file handle is locked while the iterable is being consumed and unlocked
+when iteration completes.
+
+```mjs
+import { open } from 'node:fs/promises';
+import { text, compressGzip } from 'node:stream/new';
+
+const fh = await open('input.txt', 'r');
+
+// Read as text
+console.log(await text(fh.pull({ autoClose: true })));
+
+// Read with compression
+const fh2 = await open('input.txt', 'r');
+const compressed = fh2.pull(compressGzip(), { autoClose: true });
+```
+
+```cjs
+const { open } = require('node:fs/promises');
+const { text, compressGzip } = require('node:stream/new');
+
+async function run() {
+  const fh = await open('input.txt', 'r');
+
+  // Read as text
+  console.log(await text(fh.pull({ autoClose: true })));
+
+  // Read with compression
+  const fh2 = await open('input.txt', 'r');
+  const compressed = fh2.pull(compressGzip(), { autoClose: true });
+}
+
+run().catch(console.error);
+```
+
 #### `filehandle.read(buffer, offset, length, position)`
 
 <!-- YAML
@@ -859,6 +914,55 @@ On Linux, positional writes don't work when the file is opened in append mode.
 The kernel ignores the position argument and always appends the data to
 the end of the file.
 
+#### `filehandle.writer([options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `options` {Object}
+  * `autoClose` {boolean} Close the file handle when the writer ends.
+    **Default:** `false`.
+  * `start` {number} Byte offset to start writing at. **Default:** current
+    position (append).
+* Returns: {Object}
+  * `write(chunk)` {Function} Returns {Promise\<void>}.
+  * `writev(chunks)` {Function} Returns {Promise\<void>}. Uses scatter/gather
+    I/O via a single `writev()` syscall.
+  * `end()` {Function} Returns {Promise\<number>} total bytes written.
+  * `abort(reason)` {Function} Returns {Promise\<void>}.
+
+Return a [`node:stream/new`][] writer backed by this file handle.
+
+The writer supports `Symbol.asyncDispose`, so it can be used with
+`await using`.
+
+```mjs
+import { open } from 'node:fs/promises';
+import { from, pipeTo, compressGzip } from 'node:stream/new';
+
+const fh = await open('output.gz', 'w');
+const w = fh.writer({ autoClose: true });
+await pipeTo(from('Hello!'), compressGzip(), w);
+await w.end();
+```
+
+```cjs
+const { open } = require('node:fs/promises');
+const { from, pipeTo, compressGzip } = require('node:stream/new');
+
+async function run() {
+  const fh = await open('output.gz', 'w');
+  const w = fh.writer({ autoClose: true });
+  await pipeTo(from('Hello!'), compressGzip(), w);
+  await w.end();
+}
+
+run().catch(console.error);
+```
+
 #### `filehandle[Symbol.asyncDispose]()`
 
 <!-- YAML
@@ -8779,6 +8883,8 @@ the file contents.
 [`inotify(7)`]: https://man7.org/linux/man-pages/man7/inotify.7.html
 [`kqueue(2)`]: https://www.freebsd.org/cgi/man.cgi?query=kqueue&sektion=2
 [`minimatch`]: https://github.com/isaacs/minimatch
+[`node:stream/new`]: stream_new.md
+[`stream/new pull()`]: stream_new.md#pullsource-transforms-options
 [`util.promisify()`]: util.md#utilpromisifyoriginal
 [bigints]: https://tc39.github.io/proposal-bigint
 [caveats]: #caveats

doc/api/index.md

@@ -43,6 +43,7 @@
 * [Modules: Packages](packages.md)
 * [Modules: TypeScript](typescript.md)
 * [Net](net.md)
+* [New Streams API](stream_new.md)
 * [OS](os.md)
 * [Path](path.md)
 * [Performance hooks](perf_hooks.md)