stream: avoid per-chunk promises in webstream adapters

Readable.fromWeb() and the read side of Duplex.fromWeb() allocated a
promise, a read-result object, and two reaction closures for every
chunk through reader.read(); a single reused read request now delivers
chunks through readableStreamDefaultReaderRead() instead, forwarding
each chunk in a microtask to preserve the delivery order relative to
errors and destroy.

Writable.fromWeb() and the write side of Duplex.fromWeb() paid two
derived promises off writer.ready plus the writer.write() promise and a
fresh closure pair per chunk; a single shared write request now
dispatches chunks directly and settles the node callback, with failures
delivered in a microtask.

Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #65548
Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>
Reviewed-By: Paolo Insogna <paolo@cowtech.it>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: mcollina

benchmark/webstreams/adapters.js

@@ -0,0 +1,106 @@
+'use strict';
+const common = require('../common.js');
+const {
+  Readable,
+  Writable,
+} = require('node:stream');
+const {
+  ReadableStream,
+  WritableStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  n: [1e5],
+  kind: [
+    'readable-to-web',
+    'readable-from-web',
+    'writable-to-web',
+    'writable-from-web',
+  ],
+});
+
+async function readableToWeb(n) {
+  const chunk = Buffer.alloc(1024);
+  let i = 0;
+  const streamReadable = new Readable({
+    read() {
+      if (i++ < n)
+        this.push(chunk);
+      else
+        this.push(null);
+    },
+  });
+  const reader = Readable.toWeb(streamReadable).getReader();
+  bench.start();
+  while (!(await reader.read()).done);
+  bench.end(n);
+}
+
+function readableFromWeb(n) {
+  const chunk = Buffer.alloc(1024);
+  let i = 0;
+  const readableStream = new ReadableStream({
+    pull(controller) {
+      if (i++ < n)
+        controller.enqueue(chunk);
+      else
+        controller.close();
+    },
+  });
+  const streamReadable = Readable.fromWeb(readableStream);
+  bench.start();
+  streamReadable.on('data', () => {});
+  streamReadable.on('end', () => bench.end(n));
+}
+
+async function writableToWeb(n) {
+  const chunk = Buffer.alloc(1024);
+  const streamWritable = new Writable({
+    write(chunk, encoding, callback) {
+      callback();
+    },
+  });
+  const writer = Writable.toWeb(streamWritable).getWriter();
+  bench.start();
+  for (let i = 0; i < n; i++)
+    await writer.write(chunk);
+  await writer.close();
+  bench.end(n);
+}
+
+function writableFromWeb(n) {
+  const chunk = Buffer.alloc(1024);
+  const writableStream = new WritableStream({
+    write() {},
+  });
+  const streamWritable = Writable.fromWeb(writableStream);
+  bench.start();
+  let i = 0;
+  function writeLoop() {
+    while (i++ < n) {
+      if (!streamWritable.write(chunk)) {
+        streamWritable.once('drain', writeLoop);
+        return;
+      }
+    }
+    streamWritable.end(() => bench.end(n));
+  }
+  writeLoop();
+}
+
+function main({ n, kind }) {
+  switch (kind) {
+    case 'readable-to-web':
+      readableToWeb(n);
+      break;
+    case 'readable-from-web':
+      readableFromWeb(n);
+      break;
+    case 'writable-to-web':
+      writableToWeb(n);
+      break;
+    case 'writable-from-web':
+      writableFromWeb(n);
+      break;
+  }
+}