stream: speed up reads and iteration over default WHATWG streams

Skip the size algorithm call and its result validation when the size
algorithm is the default one, and stop re-running the full
ShouldCallPull predicate at per-chunk call sites where its inputs are
already established.

readable-async-iterator type=normal: +16.6% (***)
readable-read-buffered: +13.9% to +27.5% (**/***)
pipe-to: +3.7% to +6.0% (15/16 configs significant)

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

lib/internal/webstreams/readablestream.js

@@ -599,8 +599,12 @@ class ReadableStream {
                   !controller[kState].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, as in the read() fast path.
+                readableStreamDefaultControllerPull(controller);
               }
 
               return PromiseResolve({ done: false, value: chunk });
@@ -949,8 +953,14 @@ class ReadableStreamDefaultReader {
           if (controller[kState].closeRequested && !controller[kState].queue.length) {
             readableStreamDefaultControllerClearAlgorithms(controller);
             readableStreamClose(stream);
-          } else {
-            readableStreamDefaultControllerCallPullIfNeeded(controller);
+          } 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);
           }
 
           return PromiseResolve({ done: false, value: chunk });
@@ -2524,13 +2534,28 @@ 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 internal default size algorithm is never observable by user
     // code, always returns 1, and cannot throw: enqueue with the
-    // constant instead of calling it.
-    enqueueValueWithSize(controller, chunk, 1);
+    // constant size instead of calling it or validating the result.
+    // 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).
+    materializeQueue(controllerState).pushPair(chunk, 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(
@@ -2599,6 +2624,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;
@@ -2659,14 +2691,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(