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(). Only one read is ever in flight, so a
single reused read request delivers chunks through
readableStreamDefaultReaderRead() instead, forwarding each chunk in a
microtask to keep the previous 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 (the
same contract pipeTo uses) now dispatches chunks directly and settles
the node callback, with failures delivered in a microtask because the
callback can destroy the stream while the writable machinery is
mid-transition.

Also add benchmark/webstreams/adapters.js; the suite had no rows for
the adapter paths.

                                       confidence improvement accuracy (*)   (**)  (***)
webstreams/adapters.js kind='readable-from-web' n=100000    *      5.07 %       ±4.84% ±6.44%  ±8.39%
webstreams/adapters.js kind='readable-to-web' n=100000             0.92 %       ±6.25% ±8.32% ±10.83%
webstreams/adapters.js kind='writable-from-web' n=100000  ***     39.27 %       ±7.19% ±9.58% ±12.48%
webstreams/adapters.js kind='writable-to-web' n=100000            -1.74 %       ±6.71% ±8.94% ±11.63%

Signed-off-by: Matteo Collina <hello@matteocollina.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;
+  }
+}