stream: reduce overhead of transfer

PR-URL: nodejs#50107
Reviewed-By: Yagiz Nizipli <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Stephen Belanger <[email protected]>

benchmark/webstreams/js_transfer.js

@@ -0,0 +1,52 @@
+'use strict';
+
+const common = require('../common.js');
+
+const { MessageChannel } = require('worker_threads');
+const { WritableStream, TransformStream, ReadableStream } = require('stream/web');
+
+const bench = common.createBenchmark(main, {
+  payload: ['WritableStream', 'ReadableStream', 'TransformStream'],
+  n: [1e4],
+});
+
+function main({ n, payload: payloadType }) {
+  let createPayload;
+  let messages = 0;
+
+  switch (payloadType) {
+    case 'WritableStream':
+      createPayload = () => new WritableStream();
+      break;
+    case 'ReadableStream':
+      createPayload = () => new ReadableStream();
+      break;
+    case 'TransformStream':
+      createPayload = () => new TransformStream();
+      break;
+    default:
+      throw new Error('Unsupported payload type');
+  }
+
+  const { port1, port2 } = new MessageChannel();
+
+  port2.onmessage = onMessage;
+
+  function onMessage() {
+    if (messages++ === n) {
+      bench.end(n);
+      port1.close();
+    } else {
+      send();
+    }
+  }
+
+  function send() {
+    const stream = createPayload();
+
+    port1.postMessage(stream, [stream]);
+  }
+
+  bench.start();
+  send();
+}

lib/internal/webstreams/readablestream.js

@@ -17,7 +17,6 @@ const {
   PromisePrototypeThen,
   PromiseResolve,
   PromiseReject,
-  ReflectConstruct,
   SafePromiseAll,
   Symbol,
   SymbolAsyncIterator,
@@ -642,26 +641,37 @@ ObjectDefineProperties(ReadableStream.prototype, {
   [SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableStream.name),
 });
 
-function TransferredReadableStream() {
-  return ReflectConstruct(
-    function() {
-      markTransferMode(this, false, true);
-      this[kType] = 'ReadableStream';
-      this[kState] = {
-        disturbed: false,
-        state: 'readable',
-        storedError: undefined,
-        stream: undefined,
-        transfer: {
-          writable: undefined,
-          port: undefined,
-          promise: undefined,
-        },
-      };
-      this[kIsClosedPromise] = createDeferredPromise();
+function InternalTransferredReadableStream() {
+  markTransferMode(this, false, true);
+  this[kType] = 'ReadableStream';
+  this[kState] = {
+    disturbed: false,
+    reader: undefined,
+    state: 'readable',
+    storedError: undefined,
+    stream: undefined,
+    transfer: {
+      writable: undefined,
+      port1: undefined,
+      port2: undefined,
+      promise: undefined,
     },
-    [], ReadableStream);
+  };
+
+  this[kIsClosedPromise] = createDeferredPromise();
 }
+
+ObjectSetPrototypeOf(InternalTransferredReadableStream.prototype, ReadableStream.prototype);
+ObjectSetPrototypeOf(InternalTransferredReadableStream, ReadableStream);
+
+function TransferredReadableStream() {
+  const stream = new InternalTransferredReadableStream();
+
+  stream.constructor = ReadableStream;
+
+  return stream;
+}
+
 TransferredReadableStream.prototype[kDeserialize] = () => {};
 
 class ReadableStreamBYOBRequest {

lib/internal/webstreams/transformstream.js

@@ -4,9 +4,9 @@ const {
   FunctionPrototypeBind,
   FunctionPrototypeCall,
   ObjectDefineProperties,
+  ObjectSetPrototypeOf,
   PromisePrototypeThen,
   PromiseResolve,
-  ReflectConstruct,
   SymbolToStringTag,
   Symbol,
 } = primordials;
@@ -247,25 +247,33 @@ ObjectDefineProperties(TransformStream.prototype, {
   [SymbolToStringTag]: getNonWritablePropertyDescriptor(TransformStream.name),
 });
 
-function TransferredTransformStream() {
-  return ReflectConstruct(
-    function() {
-      markTransferMode(this, false, true);
-      this[kType] = 'TransformStream';
-      this[kState] = {
-        readable: undefined,
-        writable: undefined,
-        backpressure: undefined,
-        backpressureChange: {
-          promise: undefined,
-          resolve: undefined,
-          reject: undefined,
-        },
-        controller: undefined,
-      };
+function InternalTransferredTransformStream() {
+  markTransferMode(this, false, true);
+  this[kType] = 'TransformStream';
+  this[kState] = {
+    readable: undefined,
+    writable: undefined,
+    backpressure: undefined,
+    backpressureChange: {
+      promise: undefined,
+      resolve: undefined,
+      reject: undefined,
     },
-    [], TransformStream);
+    controller: undefined,
+  };
 }
+
+ObjectSetPrototypeOf(InternalTransferredTransformStream.prototype, TransformStream.prototype);
+ObjectSetPrototypeOf(InternalTransferredTransformStream, TransformStream);
+
+function TransferredTransformStream() {
+  const stream = new InternalTransferredTransformStream();
+
+  stream.constructor = TransformStream;
+
+  return stream;
+}
+
 TransferredTransformStream.prototype[kDeserialize] = () => {};
 
 class TransformStreamDefaultController {

lib/internal/webstreams/writablestream.js

@@ -6,10 +6,10 @@ const {
   FunctionPrototypeBind,
   FunctionPrototypeCall,
   ObjectDefineProperties,
+  ObjectSetPrototypeOf,
   PromisePrototypeThen,
   PromiseResolve,
   PromiseReject,
-  ReflectConstruct,
   Symbol,
   SymbolToStringTag,
 } = primordials;
@@ -326,55 +326,63 @@ ObjectDefineProperties(WritableStream.prototype, {
   [SymbolToStringTag]: getNonWritablePropertyDescriptor(WritableStream.name),
 });
 
-function TransferredWritableStream() {
-  return ReflectConstruct(
-    function() {
-      markTransferMode(this, false, true);
-      this[kType] = 'WritableStream';
-      this[kState] = {
-        close: createDeferredPromise(),
-        closeRequest: {
-          promise: undefined,
-          resolve: undefined,
-          reject: undefined,
-        },
-        inFlightWriteRequest: {
-          promise: undefined,
-          resolve: undefined,
-          reject: undefined,
-        },
-        inFlightCloseRequest: {
-          promise: undefined,
-          resolve: undefined,
-          reject: undefined,
-        },
-        pendingAbortRequest: {
-          abort: {
-            promise: undefined,
-            resolve: undefined,
-            reject: undefined,
-          },
-          reason: undefined,
-          wasAlreadyErroring: false,
-        },
-        backpressure: false,
-        controller: undefined,
-        state: 'writable',
-        storedError: undefined,
-        writeRequests: [],
-        writer: undefined,
-        transfer: {
-          promise: undefined,
-          port1: undefined,
-          port2: undefined,
-          readable: undefined,
-        },
-      };
-      this[kIsClosedPromise] = createDeferredPromise();
-      this[kControllerErrorFunction] = () => {};
+function InternalTransferredWritableStream() {
+  markTransferMode(this, false, true);
+  this[kType] = 'WritableStream';
+  this[kState] = {
+    close: createDeferredPromise(),
+    closeRequest: {
+      promise: undefined,
+      resolve: undefined,
+      reject: undefined,
+    },
+    inFlightWriteRequest: {
+      promise: undefined,
+      resolve: undefined,
+      reject: undefined,
     },
-    [], WritableStream);
+    inFlightCloseRequest: {
+      promise: undefined,
+      resolve: undefined,
+      reject: undefined,
+    },
+    pendingAbortRequest: {
+      abort: {
+        promise: undefined,
+        resolve: undefined,
+        reject: undefined,
+      },
+      reason: undefined,
+      wasAlreadyErroring: false,
+    },
+    backpressure: false,
+    controller: undefined,
+    state: 'writable',
+    storedError: undefined,
+    writeRequests: [],
+    writer: undefined,
+    transfer: {
+      readable: undefined,
+      port1: undefined,
+      port2: undefined,
+      promise: undefined,
+    },
+  };
+
+  this[kIsClosedPromise] = createDeferredPromise();
 }
+
+ObjectSetPrototypeOf(InternalTransferredWritableStream.prototype, WritableStream.prototype);
+ObjectSetPrototypeOf(InternalTransferredWritableStream, WritableStream);
+
+function TransferredWritableStream() {
+  const stream = new InternalTransferredWritableStream();
+
+  stream.constructor = WritableStream;
+
+  return stream;
+}
+
 TransferredWritableStream.prototype[kDeserialize] = () => {};
 
 class WritableStreamDefaultWriter {