stream: add kStreamBase marker for internal pipe optimization

benchmark/webstreams/internal-pipe.js

@@ -0,0 +1,106 @@
+'use strict';
+const common = require('../common.js');
+const fsp = require('fs/promises');
+const path = require('path');
+const os = require('os');
+const { pipeline } = require('stream/promises');
+const {
+  ReadableStream,
+  WritableStream,
+} = require('node:stream/web');
+
+const bench = common.createBenchmark(main, {
+  type: [
+    'node-streams',
+    'webstream-js',
+    'webstream-file-read',
+  ],
+  size: [1024, 16384, 65536],
+  n: [1e4, 1e5],
+});
+
+async function main({ type, size, n }) {
+  const chunk = Buffer.alloc(size, 'x');
+  const totalBytes = size * n;
+
+  switch (type) {
+    case 'node-streams': {
+      // Baseline: Node.js streams
+      let received = 0;
+      const readable = new (require('stream').Readable)({
+        read() {
+          for (let i = 0; i < 100 && received < n; i++) {
+            this.push(chunk);
+            received++;
+          }
+          if (received >= n) this.push(null);
+        },
+      });
+
+      const writable = new (require('stream').Writable)({
+        write(data, enc, cb) { cb(); },
+      });
+
+      bench.start();
+      await pipeline(readable, writable);
+      bench.end(totalBytes);
+      break;
+    }
+
+    case 'webstream-js': {
+      // Web streams with pure JS source/sink
+      let sent = 0;
+      const rs = new ReadableStream({
+        pull(controller) {
+          if (sent++ < n) {
+            controller.enqueue(chunk);
+          } else {
+            controller.close();
+          }
+        },
+      });
+
+      const ws = new WritableStream({
+        write() {},
+        close() { bench.end(totalBytes); },
+      });
+
+      bench.start();
+      await rs.pipeTo(ws);
+      break;
+    }
+
+    case 'webstream-file-read': {
+      // Create a temporary file with test data
+      const tmpDir = os.tmpdir();
+      const tmpFile = path.join(tmpDir, `bench-webstream-${process.pid}.tmp`);
+
+      // Write test data to file
+      const fd = await fsp.open(tmpFile, 'w');
+      for (let i = 0; i < n; i++) {
+        await fd.write(chunk);
+      }
+      await fd.close();
+
+      // Read using readableWebStream
+      const readFd = await fsp.open(tmpFile, 'r');
+      const rs = readFd.readableWebStream({ type: 'bytes' });
+
+      const ws = new WritableStream({
+        write() {},
+        close() {
+          bench.end(totalBytes);
+          // Cleanup
+          readFd.close().then(() => fsp.unlink(tmpFile));
+        },
+      });
+
+      bench.start();
+      await rs.pipeTo(ws);
+      break;
+    }
+
+    default:
+      throw new Error(`Unknown type: ${type}`);
+  }
+}

lib/internal/webstreams/adapters.js

@@ -35,6 +35,10 @@ const {
   ByteLengthQueuingStrategy,
 } = require('internal/webstreams/queuingstrategies');
 
+const {
+  kStreamBase,
+} = require('internal/webstreams/util');
+
 const {
   Writable,
   Readable,
@@ -946,7 +950,7 @@ function newWritableStreamFromStreamBase(streamBase, strategy) {
     return promise.promise;
   }
 
-  return new WritableStream({
+  const stream = new WritableStream({
     write(chunk, controller) {
       current = current !== undefined ?
         PromisePrototypeThen(
@@ -967,6 +971,10 @@ function newWritableStreamFromStreamBase(streamBase, strategy) {
       return promise.promise;
     },
   }, strategy);
+
+  stream[kStreamBase] = streamBase;
+
+  return stream;
 }
 
 /**
@@ -1017,7 +1025,7 @@ function newReadableStreamFromStreamBase(streamBase, strategy, options = kEmptyO
     }
   };
 
-  return new ReadableStream({
+  const stream = new ReadableStream({
     start(c) { controller = c; },
 
     pull() {
@@ -1040,6 +1048,10 @@ function newReadableStreamFromStreamBase(streamBase, strategy, options = kEmptyO
       return promise.promise;
     },
   }, strategy);
+
+  stream[kStreamBase] = streamBase;
+
+  return stream;
 }
 
 module.exports = {

lib/internal/webstreams/readablestream.js

@@ -31,6 +31,7 @@ const {
 
 const {
   AbortError,
+  ErrnoException,
   codes: {
     ERR_ILLEGAL_CONSTRUCTOR,
     ERR_INVALID_ARG_TYPE,
@@ -45,6 +46,17 @@ const {
   DOMException,
 } = internalBinding('messaging');
 
+const {
+  StreamPipe,
+} = internalBinding('stream_pipe');
+
+const {
+  kReadBytesOrError,
+  streamBaseState,
+} = internalBinding('stream_wrap');
+
+const { UV_EOF } = internalBinding('uv');
+
 const {
   isArrayBufferView,
   isDataView,
@@ -114,6 +126,7 @@ const {
   iteratorNext,
   kType,
   kState,
+  kStreamBase,
 } = require('internal/webstreams/util');
 
 const {
@@ -125,6 +138,7 @@ const {
   isWritableStreamDefaultWriter,
 
   writableStreamAbort,
+  writableStreamClose,
   writableStreamCloseQueuedOrInFlight,
   writableStreamDefaultWriterCloseWithErrorPropagation,
   writableStreamDefaultWriterRelease,
@@ -1369,6 +1383,83 @@ function readableStreamPipeTo(
   preventCancel,
   signal) {
 
+  const sourceStreamBase = source?.[kStreamBase];
+  const destStreamBase = dest?.[kStreamBase];
+
+  if (sourceStreamBase !== undefined &&
+      destStreamBase !== undefined &&
+      signal === undefined &&
+      !preventClose &&
+      !preventAbort &&
+      !preventCancel) {
+    // Native C++ StreamPipe path for internal-to-internal piping.
+    // Ref: https://github.com/nodejs/performance/issues/134
+    const promise = PromiseWithResolvers();
+
+    source[kState].disturbed = true;
+
+    let pipeError = null;
+    let isComplete = false;
+    const originalSourceOnread = sourceStreamBase.onread;
+
+    sourceStreamBase.onread = (arrayBuffer) => {
+      const nread = streamBaseState[kReadBytesOrError];
+      if (nread < 0 && nread !== UV_EOF) {
+        pipeError = new ErrnoException(nread, 'read');
+      }
+      if (originalSourceOnread) {
+        return originalSourceOnread(arrayBuffer);
+      }
+    };
+
+    function finalize(error) {
+      if (isComplete) return;
+      isComplete = true;
+      sourceStreamBase.onread = originalSourceOnread;
+
+      if (error) {
+        if (source[kState].state === 'readable') {
+          readableStreamError(source, error);
+        }
+        if (dest[kState].state === 'writable') {
+          writableStreamAbort(dest, error);
+        }
+        promise.reject(error);
+      } else {
+        if (source[kState].state === 'readable') {
+          readableStreamClose(source);
+        }
+        if (dest[kState].state === 'writable' &&
+            !writableStreamCloseQueuedOrInFlight(dest)) {
+          PromisePrototypeThen(
+            writableStreamClose(dest),
+            () => promise.resolve(),
+            (err) => promise.reject(err),
+          );
+        } else {
+          promise.resolve();
+        }
+      }
+    }
+
+    try {
+      const pipe = new StreamPipe(sourceStreamBase, destStreamBase);
+      pipe.onunpipe = () => {
+        if (pipeError) {
+          finalize(pipeError);
+        }
+      };
+      pipe.oncomplete = () => finalize(pipeError);
+      pipe.start();
+    } catch (error) {
+      sourceStreamBase.onread = originalSourceOnread;
+      promise.reject(error);
+    }
+
+    return promise.promise;
+  }
+
+  // Use JS-based piping
   let reader;
   let writer;
   let disposable;

lib/internal/webstreams/util.js

@@ -49,6 +49,7 @@ const {
 
 const kState = Symbol('kState');
 const kType = Symbol('kType');
+const kStreamBase = Symbol('kStreamBase');
 
 const AsyncIterator = {
   __proto__: AsyncIteratorPrototype,
@@ -296,4 +297,5 @@ module.exports = {
   iteratorNext,
   kType,
   kState,
+  kStreamBase,
 };