stream: improve tee perf by reduce ReflectConstruct usages

also added more webstream creation benchmarks PR-URL: #49546 Backport-PR-URL: #49620 Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
nodejs · Sep 13, 2023 · 8aac95d · 8aac95d
1 parent db4ce8a
commit 8aac95d
Show file tree

Hide file tree

Showing 2 changed files with 96 additions and 35 deletions.
diff --git a/benchmark/webstreams/creation.js b/benchmark/webstreams/creation.js
@@ -2,47 +2,99 @@
 const common = require('../common.js');
 const {
   ReadableStream,
+  ReadableStreamDefaultReader,
+  ReadableStreamBYOBReader,
   TransformStream,
   WritableStream,
 } = require('node:stream/web');
 const assert = require('assert');
 
 const bench = common.createBenchmark(main, {
   n: [50e3],
-  kind: ['ReadableStream', 'TransformStream', 'WritableStream'],
+  kind: [
+    'ReadableStream',
+    'TransformStream',
+    'WritableStream',
+
+    'ReadableStreamDefaultReader',
+    'ReadableStreamBYOBReader',
+
+    'ReadableStream.tee',
+  ],
 });
 
-let rs, ws, ts;
+let readableStream;
+let transformStream;
+let writableStream;
+let readableStreamDefaultReader;
+let readableStreamBYOBReader;
+let teeResult;
 
 function main({ n, kind }) {
   switch (kind) {
     case 'ReadableStream':
       bench.start();
       for (let i = 0; i < n; ++i)
-        rs = new ReadableStream();
+        readableStream = new ReadableStream();
       bench.end(n);
 
       // Avoid V8 deadcode (elimination)
-      assert.ok(rs);
+      assert.ok(readableStream);
       break;
     case 'WritableStream':
       bench.start();
       for (let i = 0; i < n; ++i)
-        ws = new WritableStream();
+        writableStream = new WritableStream();
       bench.end(n);
 
       // Avoid V8 deadcode (elimination)
-      assert.ok(ws);
+      assert.ok(writableStream);
       break;
     case 'TransformStream':
       bench.start();
       for (let i = 0; i < n; ++i)
-        ts = new TransformStream();
+        transformStream = new TransformStream();
+      bench.end(n);
+
+      // Avoid V8 deadcode (elimination)
+      assert.ok(transformStream);
+      break;
+    case 'ReadableStreamDefaultReader': {
+      const readers = Array.from({ length: n }, () => new ReadableStream());
+
+      bench.start();
+      for (let i = 0; i < n; ++i)
+        readableStreamDefaultReader = new ReadableStreamDefaultReader(readers[i]);
+      bench.end(n);
+
+      // Avoid V8 deadcode (elimination)
+      assert.ok(readableStreamDefaultReader);
+      break;
+    }
+    case 'ReadableStreamBYOBReader': {
+      const readers = Array.from({ length: n }, () => new ReadableStream({ type: 'bytes' }));
+
+      bench.start();
+      for (let i = 0; i < n; ++i)
+        readableStreamBYOBReader = new ReadableStreamBYOBReader(readers[i]);
+      bench.end(n);
+
+      // Avoid V8 deadcode (elimination)
+      assert.ok(readableStreamBYOBReader);
+      break;
+    }
+    case 'ReadableStream.tee': {
+      const streams = Array.from({ length: n }, () => new ReadableStream());
+
+      bench.start();
+      for (let i = 0; i < n; ++i)
+        teeResult = streams[i].tee();
       bench.end(n);
 
       // Avoid V8 deadcode (elimination)
-      assert.ok(ts);
+      assert.ok(teeResult);
       break;
+    }
     default:
       throw new Error('Invalid kind');
   }

diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
@@ -1200,34 +1200,43 @@ ObjectDefineProperties(ReadableByteStreamController.prototype, {
   [SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableByteStreamController.name),
 });
 
+function TeeReadableStream(start, pull, cancel) {
+  this[kType] = 'ReadableStream';
+  this[kState] = {
+    disturbed: false,
+    state: 'readable',
+    storedError: undefined,
+    stream: undefined,
+    transfer: {
+      writable: undefined,
+      port: undefined,
+      promise: undefined,
+    },
+  };
+  this[kIsClosedPromise] = createDeferredPromise();
+  setupReadableStreamDefaultControllerFromSource(
+    this,
+    ObjectCreate(null, {
+      start: { __proto__: null, value: start },
+      pull: { __proto__: null, value: pull },
+      cancel: { __proto__: null, value: cancel },
+    }),
+    1,
+    () => 1);
+
+
+  return makeTransferable(this);
+}
+
+ObjectSetPrototypeOf(TeeReadableStream.prototype, ReadableStream.prototype);
+ObjectSetPrototypeOf(TeeReadableStream, ReadableStream);
+
 function createTeeReadableStream(start, pull, cancel) {
-  return ReflectConstruct(
-    function() {
-      this[kType] = 'ReadableStream';
-      this[kState] = {
-        disturbed: false,
-        state: 'readable',
-        storedError: undefined,
-        stream: undefined,
-        transfer: {
-          writable: undefined,
-          port: undefined,
-          promise: undefined,
-        },
-      };
-      this[kIsClosedPromise] = createDeferredPromise();
-      setupReadableStreamDefaultControllerFromSource(
-        this,
-        ObjectCreate(null, {
-          start: { __proto__: null, value: start },
-          pull: { __proto__: null, value: pull },
-          cancel: { __proto__: null, value: cancel },
-        }),
-        1,
-        () => 1);
-      return makeTransferable(this);
-    }, [], ReadableStream,
-  );
+  const tee = new TeeReadableStream(start, pull, cancel);
+
+  // For spec compliance the Tee must be a ReadableStream
+  tee.constructor = ReadableStream;
+  return tee;
 }
 
 const isReadableStream =