stream: speed up reads and iteration over default WHATWG streams

benchmark/webstreams/readable-async-iterator.js

@@ -6,23 +6,40 @@ const {
 
 const bench = common.createBenchmark(main, {
   n: [1e5],
+  type: ['normal', 'bytes'],
 });
 
 
-async function main({ n }) {
-  const rs = new ReadableStream({
-    pull: function(controller) {
-      controller.enqueue(1);
-    },
-  });
+async function main({ n, type }) {
+  const rs = type === 'bytes' ?
+    new ReadableStream({
+      type: 'bytes',
+      pull: function(controller) {
+        controller.enqueue(new Uint8Array(1));
+      },
+    }) :
+    new ReadableStream({
+      pull: function(controller) {
+        controller.enqueue(1);
+      },
+    });
 
   let x = 0;
 
   bench.start();
-  for await (const chunk of rs) {
-    x += chunk;
-    if (x > n) {
-      break;
+  if (type === 'bytes') {
+    for await (const chunk of rs) {
+      x += chunk.byteLength;
+      if (x > n) {
+        break;
+      }
+    }
+  } else {
+    for await (const chunk of rs) {
+      x += chunk;
+      if (x > n) {
+        break;
+      }
     }
   }
   // Use x to ensure V8 does not optimize away the loop as a noop.

lib/internal/webstreams/readablestream.js

@@ -501,6 +501,12 @@ class ReadableStream {
       current: undefined,
     };
     let started = false;
+    // A single reusable read request: at most one read is ever in flight
+    // (next() chains through state.current), and the request is consumed
+    // before the next read starts, so only its promise record changes
+    // per read.
+    // eslint-disable-next-line no-use-before-define
+    const readRequest = new ReadableStreamAsyncIteratorReadRequest(reader, state, undefined);
 
     // The nextSteps function is not an async function in order
     // to make it more efficient. Because nextSteps explicitly
@@ -519,8 +525,8 @@ class ReadableStream {
       }
       const promise = PromiseWithResolvers();
 
-      // eslint-disable-next-line no-use-before-define
-      readableStreamDefaultReaderRead(reader, new ReadableStreamAsyncIteratorReadRequest(reader, state, promise));
+      readRequest.promise = promise;
+      readableStreamDefaultReaderRead(reader, readRequest);
       return promise.promise;
     }
 
@@ -574,28 +580,42 @@ class ReadableStream {
         }
         // No read is in flight. Mirror the buffered fast path of
         // ReadableStreamDefaultReader.read(): when data is already queued
-        // in a default controller, resolve immediately without allocating
-        // a read request. The result settles synchronously, so leaving
+        // in the controller, resolve immediately without allocating a
+        // read request. The result settles synchronously, so leaving
         // state.current undefined matches the state the slow path reaches
         // once its read request callbacks have settled.
         const stream = reader[kState].stream;
-        if (!state.done && stream !== undefined) {
+        if (!state.done && stream !== undefined &&
+            stream[kState].state === 'readable') {
           const controller = stream[kState].controller;
-          if (stream[kState].state === 'readable' &&
-              isReadableStreamDefaultController(controller) &&
-              controller[kState].queue.length > 0) {
-            stream[kState].disturbed = true;
-            const chunk = dequeueValue(controller);
-
-            if (controller[kState].closeRequested &&
-                !controller[kState].queue.length) {
-              readableStreamDefaultControllerClearAlgorithms(controller);
-              readableStreamClose(stream);
-            } else {
-              readableStreamDefaultControllerCallPullIfNeeded(controller);
+          if (isReadableStreamDefaultController(controller)) {
+            if (controller[kState].queue.length > 0) {
+              stream[kState].disturbed = true;
+              const chunk = dequeueValue(controller);
+
+              if (controller[kState].closeRequested &&
+                  !controller[kState].queue.length) {
+                readableStreamDefaultControllerClearAlgorithms(controller);
+                readableStreamClose(stream);
+              } else if (!controller[kState].closeRequested &&
+                         controller[kState].started &&
+                         controller[kState].highWaterMark -
+                           controller[kState].queueTotalSize > 0) {
+                // Reduced ShouldCallPull, as in the read() fast path.
+                readableStreamDefaultControllerPull(controller);
+              }
+
+              return PromiseResolve({ done: false, value: chunk });
             }
+          } else if (controller[kState].queueTotalSize > 0) {
+            // Byte controller with buffered data: same shape as above via
+            // the queue-filled arm of the byte controller's pull steps.
+            stream[kState].disturbed = true;
+            return PromiseResolve({
+              done: false,
 
-            return PromiseResolve({ done: false, value: chunk });
+              value: readableByteStreamControllerDequeueChunk(controller),
+            });
           }
         }
         state.current = nextSteps();
@@ -918,24 +938,42 @@ class ReadableStreamDefaultReader {
     const stream = this[kState].stream;
     const controller = stream[kState].controller;
 
-    // Fast path: if data is already buffered in a default controller,
+    // Fast path: if data is already buffered in the controller's queue,
     // return a resolved promise immediately without creating a read request.
     // This is spec-compliant because read() returns a Promise, and
     // Promise.resolve() callbacks still run in the microtask queue.
-    if (stream[kState].state === 'readable' &&
-        isReadableStreamDefaultController(controller) &&
-        controller[kState].queue.length > 0) {
-      stream[kState].disturbed = true;
-      const chunk = dequeueValue(controller);
+    if (stream[kState].state === 'readable') {
+      if (isReadableStreamDefaultController(controller)) {
+        if (controller[kState].queue.length > 0) {
+          stream[kState].disturbed = true;
+          const chunk = dequeueValue(controller);
+
+          if (controller[kState].closeRequested && !controller[kState].queue.length) {
+            readableStreamDefaultControllerClearAlgorithms(controller);
+            readableStreamClose(stream);
+          } else if (!controller[kState].closeRequested &&
+                     controller[kState].started &&
+                     controller[kState].highWaterMark -
+                       controller[kState].queueTotalSize > 0) {
+            // ShouldCallPull reduced to the conditions not already
+            // established on this path: the state is readable and the
+            // queue was non-empty, so no read requests can be parked.
+            readableStreamDefaultControllerPull(controller);
+          }
 
-      if (controller[kState].closeRequested && !controller[kState].queue.length) {
-        readableStreamDefaultControllerClearAlgorithms(controller);
-        readableStreamClose(stream);
-      } else {
-        readableStreamDefaultControllerCallPullIfNeeded(controller);
+          return PromiseResolve({ done: false, value: chunk });
+        }
+      } else if (controller[kState].queueTotalSize > 0) {
+        // Byte controller with buffered data: mirror the queue-filled arm
+        // of its pull steps (which never consults pendingPullIntos) minus
+        // the read request.
+        stream[kState].disturbed = true;
+        return PromiseResolve({
+          done: false,
+
+          value: readableByteStreamControllerDequeueChunk(controller),
+        });
       }
-
-      return PromiseResolve({ done: false, value: chunk });
     }
 
     // Slow path: create request and go through normal flow
@@ -2494,8 +2532,27 @@ function readableStreamDefaultControllerEnqueue(controller, chunk) {
       reader[kState] !== undefined &&
       reader[kType] === 'ReadableStreamDefaultReader' &&
       reader[kState].readRequests.length) {
+    // Fulfilling a read request can run user code synchronously (the
+    // pipeTo read request invokes the sink's write algorithm), so the
+    // full ShouldCallPull predicate has to be re-evaluated afterwards.
     readableStreamFulfillReadRequest(stream, chunk, false);
+  } else if (controllerState.sizeAlgorithm === defaultSizeAlgorithm) {
+    // The default size algorithm always returns 1, so the call and the
+    // size validation in enqueueValueWithSize can be skipped entirely.
+    // No user code runs between the guards at the top of this function
+    // and this point, so ShouldCallPull reduces to the started flag and
+    // the desired size (this branch implies no parked read requests,
+    // ruling out the reader arm of the predicate).
+    ArrayPrototypePush(controllerState.queue, { value: chunk, size: 1 });
+    controllerState.queueTotalSize++;
+    if (controllerState.started &&
+        controllerState.highWaterMark - controllerState.queueTotalSize > 0) {
+      readableStreamDefaultControllerPull(controller);
+    }
+    return;
   } else {
+    // The user-supplied size algorithm may run arbitrary code, so the
+    // full ShouldCallPull predicate has to be re-evaluated afterwards.
     try {
       const chunkSize =
         FunctionPrototypeCall(
@@ -2564,6 +2621,13 @@ function readableStreamDefaultControllerShouldCallPull(controller) {
 function readableStreamDefaultControllerCallPullIfNeeded(controller) {
   if (!readableStreamDefaultControllerShouldCallPull(controller))
     return;
+  readableStreamDefaultControllerPull(controller);
+}
+
+// The ShouldCallPull half of CallPullIfNeeded, split out so that callers
+// that have already established the predicate from state in scope (the
+// enqueue path above) can skip re-running it.
+function readableStreamDefaultControllerPull(controller) {
   if (controller[kState].pulling) {
     controller[kState].pullAgain = true;
     return;
@@ -2624,14 +2688,24 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) {
     if (controller[kState].closeRequested && !queue.length) {
       readableStreamDefaultControllerClearAlgorithms(controller);
       readableStreamClose(stream);
-    } else {
-      readableStreamDefaultControllerCallPullIfNeeded(controller);
+    } else if (!controller[kState].closeRequested &&
+               controller[kState].started &&
+               controller[kState].highWaterMark -
+                 controller[kState].queueTotalSize > 0) {
+      // Reduced ShouldCallPull: the state is known to be readable and the
+      // queue was non-empty, so no read requests can be parked.
+      readableStreamDefaultControllerPull(controller);
     }
     readRequest[kChunk](chunk);
     return;
   }
   readableStreamAddReadRequest(stream, readRequest);
-  readableStreamDefaultControllerCallPullIfNeeded(controller);
+  // Reduced ShouldCallPull: the state is known to be readable, an empty
+  // queue in that state implies close has not been requested (close with
+  // an empty queue closes the stream immediately), and the read request
+  // parked above already satisfies the reader-with-pending-reads arm.
+  if (controller[kState].started)
+    readableStreamDefaultControllerPull(controller);
 }
 
 function setupReadableStreamDefaultController(
@@ -3044,9 +3118,23 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
     }
   }
 
-  if (readableStreamHasDefaultReader(stream)) {
-    readableByteStreamControllerProcessReadRequestsUsingQueue(controller);
-    if (!readableStreamGetNumReadRequests(stream)) {
+  // Single consolidated pass over the reader state. The spec routes this
+  // through HasDefaultReader / ProcessReadRequestsUsingQueue /
+  // GetNumReadRequests / FulfillReadRequest, which would re-run the same
+  // reader brand check and re-load the read request list four times on
+  // this per-chunk path.
+  const { reader } = stream[kState];
+  if (reader !== undefined &&
+      reader[kState] !== undefined &&
+      reader[kType] === 'ReadableStreamDefaultReader') {
+    const { readRequests } = reader[kState];
+    if (readRequests.length && controller[kState].queueTotalSize > 0) {
+      // Only possible when data was enqueued while the stream was not
+      // being read; read requests otherwise never coexist with a
+      // non-empty queue.
+      readableByteStreamControllerProcessReadRequestsUsingQueue(controller);
+    }
+    if (!readRequests.length) {
       readableByteStreamControllerEnqueueChunkToQueue(
         controller,
         transferredBuffer,
@@ -3060,7 +3148,8 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
       }
       const transferredView =
         new Uint8Array(transferredBuffer, byteOffset, byteLength);
-      readableStreamFulfillReadRequest(stream, transferredView, false);
+      const readRequest = ArrayPrototypeShift(readRequests);
+      readRequest[kChunk](transferredView);
     }
   } else if (readableStreamHasBYOBReader(stream)) {
     readableByteStreamControllerEnqueueChunkToQueue(
@@ -3395,22 +3484,28 @@ function readableByteStreamControllerCancelSteps(controller, reason) {
   return result;
 }
 
-function readableByteStreamControllerFillReadRequestFromQueue(controller, readRequest) {
-  const {
-    queue,
-    queueTotalSize,
-  } = controller[kState];
-  assert(queueTotalSize > 0);
+// Dequeues the first chunk of the byte queue as a Uint8Array view,
+// handling queue drain (close-on-empty or pull) before the view is
+// created. This is the [[queueTotalSize]] > 0 arm of the byte
+// controller's pull steps; it is also called directly from the
+// buffered fast paths in ReadableStreamDefaultReader.read() and the
+// async iterator, which resolve with the view without allocating a
+// read request.
+function readableByteStreamControllerDequeueChunk(controller) {
+  assert(controller[kState].queueTotalSize > 0);
   const {
     buffer,
     byteOffset,
     byteLength,
-  } = ArrayPrototypeShift(queue);
+  } = ArrayPrototypeShift(controller[kState].queue);
 
   controller[kState].queueTotalSize -= byteLength;
   readableByteStreamControllerHandleQueueDrain(controller);
-  const view = new Uint8Array(buffer, byteOffset, byteLength);
-  readRequest[kChunk](view);
+  return new Uint8Array(buffer, byteOffset, byteLength);
+}
+
+function readableByteStreamControllerFillReadRequestFromQueue(controller, readRequest) {
+  readRequest[kChunk](readableByteStreamControllerDequeueChunk(controller));
 }
 
 function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) {