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/test_runner/reporter/junit.js

@@ -5,9 +5,6 @@ const {
   ArrayPrototypeMap,
   ArrayPrototypePush,
   ArrayPrototypeSome,
-  Date,
-  DateNow,
-  DatePrototypeToISOString,
   NumberPrototypeToFixed,
   ObjectEntries,
   RegExpPrototypeSymbolReplace,
@@ -115,11 +112,6 @@ module.exports = async function* junitReporter(source) {
           currentTest.attrs.tests = nonCommentChildren.length;
           currentTest.attrs.failures = ArrayPrototypeFilter(currentTest.children, isFailure).length;
           currentTest.attrs.skipped = ArrayPrototypeFilter(currentTest.children, isSkipped).length;
-          // A suite's `test:start` is emitted lazily (when its first subtest
-          // reports), so derive the start time from the end minus the measured
-          // duration rather than stamping the (late) test:start moment.
-          currentTest.attrs.timestamp =
-            DatePrototypeToISOString(new Date(DateNow() - event.data.details.duration_ms));
           currentTest.attrs.hostname = HOSTNAME;
         } else {
           currentTest.tag = 'testcase';

lib/internal/webstreams/adapters.js

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

lib/internal/webstreams/readablestream.js

@@ -32,6 +32,7 @@
 
 const {
   AbortError,
+  ErrnoException,
   codes: {
     ERR_ARG_NOT_ITERABLE,
     ERR_ILLEGAL_CONSTRUCTOR,
@@ -51,6 +52,13 @@
   markPromiseAsHandled,
 } = internalBinding('util');
 
+const {
+  kReadBytesOrError,
+  streamBaseState,
+} = internalBinding('stream_wrap');
+
+const { UV_EOF } = internalBinding('uv');
+
 const {
   isArrayBufferView,
   isDataView,
@@ -123,6 +131,7 @@
   resetQueue,
   resolvedRecord,
   setPromiseHandled,
+  kStreamBase,
 } = require('internal/webstreams/util');
 
 const {
@@ -134,6 +143,7 @@
   isWritableStreamDefaultWriter,
 
   writableStreamAbort,
+  writableStreamClose,
   writableStreamCloseQueuedOrInFlight,
   writableStreamDefaultWriterCloseWithErrorPropagation,
   writableStreamDefaultWriterRelease,
@@ -1462,6 +1472,83 @@
   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

@@ -47,6 +47,7 @@ const {
 
 const kState = Symbol('kState');
 const kType = Symbol('kType');
+const kStreamBase = Symbol('kStreamBase');
 
 const AsyncIterator = {
   __proto__: AsyncIteratorPrototype,
@@ -280,4 +281,5 @@ module.exports = {
   resetQueue,
   resolvedRecord,
   setPromiseHandled,
+  kStreamBase,
 };

test/common/assertSnapshot.js

@@ -226,7 +226,6 @@ function replaceJunitDuration(str) {
     .replaceAll(/time="[0-9.]+"/g, 'time="*"')
     .replaceAll(/duration_ms [0-9.]+/g, 'duration_ms *')
     .replaceAll(`hostname="${hostname()}"`, 'hostname="HOSTNAME"')
-    .replaceAll(/timestamp="[^"]*"/g, 'timestamp="*"')
     .replaceAll(/file="[^"]*"/g, 'file="*"');
 }
 

test/fixtures/test-runner/output/junit_reporter.snapshot

@@ -128,7 +128,7 @@ true !== false
 	<testcase name="immediate throw - passes but warns" time="*" classname="test" file="*"/>
 	<testcase name="immediate reject - passes but warns" time="*" classname="test" file="*"/>
 	<testcase name="immediate resolve pass" time="*" classname="test" file="*"/>
-	<testsuite name="subtest sync throw fail" time="*" disabled="0" errors="0" tests="1" failures="1" skipped="0" timestamp="*" hostname="HOSTNAME">
+	<testsuite name="subtest sync throw fail" time="*" disabled="0" errors="0" tests="1" failures="1" skipped="0" hostname="HOSTNAME">
 		<testcase name="+sync throw fail" time="*" classname="test" file="*" failure="thrown from subtest sync throw fail">
 			<failure type="testCodeFailure" message="thrown from subtest sync throw fail">
 Error [ERR_TEST_FAILURE]: thrown from subtest sync throw fail
@@ -151,15 +151,15 @@ Error [ERR_TEST_FAILURE]: thrown from subtest sync throw fail
 [Error [ERR_TEST_FAILURE]: Symbol(thrown symbol from sync throw non-error fail)] { code: 'ERR_TEST_FAILURE', failureType: 'testCodeFailure', cause: Symbol(thrown symbol from sync throw non-error fail) }
 		</failure>
 	</testcase>
-	<testsuite name="level 0a" time="*" disabled="0" errors="0" tests="4" failures="0" skipped="0" timestamp="*" hostname="HOSTNAME">
+	<testsuite name="level 0a" time="*" disabled="0" errors="0" tests="4" failures="0" skipped="0" hostname="HOSTNAME">
 		<testcase name="level 1a" time="*" classname="test" file="*"/>
 		<testcase name="level 1b" time="*" classname="test" file="*"/>
 		<testcase name="level 1c" time="*" classname="test" file="*"/>
 		<testcase name="level 1d" time="*" classname="test" file="*"/>
 	</testsuite>
-	<testsuite name="top level" time="*" disabled="0" errors="0" tests="2" failures="0" skipped="0" timestamp="*" hostname="HOSTNAME">
+	<testsuite name="top level" time="*" disabled="0" errors="0" tests="2" failures="0" skipped="0" hostname="HOSTNAME">
 		<testcase name="+long running" time="*" classname="test" file="*"/>
-		<testsuite name="+short running" time="*" disabled="0" errors="0" tests="1" failures="0" skipped="0" timestamp="*" hostname="HOSTNAME">
+		<testsuite name="+short running" time="*" disabled="0" errors="0" tests="1" failures="0" skipped="0" hostname="HOSTNAME">
 			<testcase name="++short running" time="*" classname="test" file="*"/>
 		</testsuite>
 	</testsuite>
@@ -266,7 +266,7 @@ Error [ERR_TEST_FAILURE]: thrown from callback async throw
 		</failure>
 	</testcase>
 	<testcase name="callback async throw after done" time="*" classname="test" file="*"/>
-	<testsuite name="only is set on subtests but not in only mode" time="*" disabled="0" errors="0" tests="3" failures="0" skipped="0" timestamp="*" hostname="HOSTNAME">
+	<testsuite name="only is set on subtests but not in only mode" time="*" disabled="0" errors="0" tests="3" failures="0" skipped="0" hostname="HOSTNAME">
 		<testcase name="running subtest 1" time="*" classname="test" file="*"/>
 		<testcase name="running subtest 3" time="*" classname="test" file="*"/>
 		<testcase name="running subtest 4" time="*" classname="test" file="*"/>
@@ -288,7 +288,7 @@ Error [ERR_TEST_FAILURE]: thrown from callback async throw
 }
 		</failure>
 	</testcase>
-	<testsuite name="subtest sync throw fails" time="*" disabled="0" errors="0" tests="2" failures="2" skipped="0" timestamp="*" hostname="HOSTNAME">
+	<testsuite name="subtest sync throw fails" time="*" disabled="0" errors="0" tests="2" failures="2" skipped="0" hostname="HOSTNAME">
 		<testcase name="sync throw fails at first" time="*" classname="test" file="*" failure="thrown from subtest sync throw fails at first">
 			<failure type="testCodeFailure" message="thrown from subtest sync throw fails at first">
 Error [ERR_TEST_FAILURE]: thrown from subtest sync throw fails at first

test/parallel/test-runner-reporters.js

@@ -200,12 +200,6 @@ describe('node:test reporters', { concurrency: true }, () => {
     assert.strictEqual(child.stdout.toString(), '');
     const fileContents = fs.readFileSync(file, 'utf8');
     assert.match(fileContents, /<testsuite .*name="nested".*tests="2".*failures="1".*skipped="0".*>/);
-    // The exact timestamp format is intentionally not pinned here (still under
-    // discussion); assert only that the value is present and a real date.
-    const { 1: timestamp } = fileContents.match(/<testsuite [^>]*timestamp="([^"]+)"/) ?? [];
-    assert.ok(timestamp, 'testsuite should have a timestamp attribute');
-    assert.match(timestamp, /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}/);
-    assert.ok(!Number.isNaN(Date.parse(timestamp)), `expected a valid date, got ${timestamp}`);
     assert.match(fileContents, /<testcase .*name="failing".*>\s*<failure .*type="testCodeFailure".*message="error".*>/);
     assert.match(fileContents, /<testcase .*name="ok".*classname="test".*\/>/);
     assert.match(fileContents, /<testcase .*name="top level".*classname="test".*\/>/);