stream: drain WHATWG stream queues from a head index

lib/internal/webstreams/readablestream.js

@@ -119,6 +119,7 @@ const {
   nonOpCancel,
   nonOpPull,
   nonOpStart,
+  queueLength,
   rejectedHandledRecord,
   resetQueue,
   resolvedRecord,
@@ -589,16 +590,16 @@ class ReadableStream {
             stream[kState].state === 'readable') {
           const controller = stream[kState].controller;
           if (isReadableStreamDefaultController(controller)) {
-            if (controller[kState].queue.length > 0) {
+            if (queueLength(controller) > 0) {
               stream[kState].disturbed = true;
               const chunk = dequeueValue(controller);
 
               if (controller[kState].closeRequested &&
-                  !controller[kState].queue.length) {
+                  queueLength(controller) === 0) {
                 readableStreamDefaultControllerClearAlgorithms(controller);
                 readableStreamClose(stream);
               } else {
-                readableStreamDefaultControllerCallPullIfNeeded(controller);
+                readableStreamDefaultControllerPullIfNeededAfterDequeue(controller);
               }
 
               return PromiseResolve({ done: false, value: chunk });
@@ -940,15 +941,16 @@ class ReadableStreamDefaultReader {
     // Promise.resolve() callbacks still run in the microtask queue.
     if (stream[kState].state === 'readable') {
       if (isReadableStreamDefaultController(controller)) {
-        if (controller[kState].queue.length > 0) {
+        if (queueLength(controller) > 0) {
           stream[kState].disturbed = true;
           const chunk = dequeueValue(controller);
 
-          if (controller[kState].closeRequested && !controller[kState].queue.length) {
+          if (controller[kState].closeRequested &&
+              queueLength(controller) === 0) {
             readableStreamDefaultControllerClearAlgorithms(controller);
             readableStreamClose(stream);
           } else {
-            readableStreamDefaultControllerCallPullIfNeeded(controller);
+            readableStreamDefaultControllerPullIfNeededAfterDequeue(controller);
           }
 
           return PromiseResolve({ done: false, value: chunk });
@@ -1640,14 +1642,15 @@ function readableStreamPipeTo(
     // reduces promise allocation overhead.
     if (source[kState].state === 'readable' &&
         isReadableStreamDefaultController(controller) &&
-        controller[kState].queue.length > 0) {
+        queueLength(controller) > 0) {
 
-      while (controller[kState].queue.length > 0) {
+      while (queueLength(controller) > 0) {
         if (shuttingDown) return true;
 
         const chunk = dequeueValue(controller);
 
-        if (controller[kState].closeRequested && !controller[kState].queue.length) {
+        if (controller[kState].closeRequested &&
+            queueLength(controller) === 0) {
           readableStreamDefaultControllerClearAlgorithms(controller);
           readableStreamClose(source);
         }
@@ -2501,7 +2504,7 @@ function readableStreamDefaultControllerClose(controller) {
   if (!readableStreamDefaultControllerCanCloseOrEnqueue(controller))
     return;
   controller[kState].closeRequested = true;
-  if (!controller[kState].queue.length) {
+  if (queueLength(controller) === 0) {
     readableStreamDefaultControllerClearAlgorithms(controller);
     readableStreamClose(controller[kState].stream);
   }
@@ -2522,8 +2525,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(
@@ -2592,6 +2614,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;
@@ -2618,6 +2647,19 @@ function readableStreamDefaultControllerCallPullIfNeeded(controller) {
     controller[kState].pullRejected);
 }
 
+// ShouldCallPull reduced for a caller that has just dequeued a chunk from a
+// readable default controller: the state is known to be readable and the
+// queue was non-empty, so no read requests can be parked and only the
+// close / started / desired-size arms of the predicate remain.
+function readableStreamDefaultControllerPullIfNeededAfterDequeue(controller) {
+  const state = controller[kState];
+  if (!state.closeRequested &&
+      state.started &&
+      state.highWaterMark - state.queueTotalSize > 0) {
+    readableStreamDefaultControllerPull(controller);
+  }
+}
+
 function readableStreamDefaultControllerClearAlgorithms(controller) {
   controller[kState].pullAlgorithm = undefined;
   controller[kState].cancelAlgorithm = undefined;
@@ -2643,23 +2685,25 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) {
 }
 
 function readableStreamDefaultControllerPullSteps(controller, readRequest) {
-  const {
-    stream,
-    queue,
-  } = controller[kState];
-  if (queue.length) {
+  const stream = controller[kState].stream;
+  if (queueLength(controller) > 0) {
     const chunk = dequeueValue(controller);
-    if (controller[kState].closeRequested && !queue.length) {
+    if (controller[kState].closeRequested && queueLength(controller) === 0) {
       readableStreamDefaultControllerClearAlgorithms(controller);
       readableStreamClose(stream);
     } else {
-      readableStreamDefaultControllerCallPullIfNeeded(controller);
+      readableStreamDefaultControllerPullIfNeededAfterDequeue(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(
@@ -2681,6 +2725,7 @@ function setupReadableStreamDefaultController(
     pullFulfilled: undefined,
     pullRejected: undefined,
     queue: [],
+    queueHead: 0,
     queueTotalSize: 0,
     started: false,
     sizeAlgorithm,
@@ -3548,6 +3593,7 @@ function setupReadableByteStreamController(
     started: false,
     stream,
     queue: [],
+    queueHead: 0,
     queueTotalSize: 0,
     highWaterMark,
     pullAlgorithm,

lib/internal/webstreams/util.js

@@ -5,7 +5,7 @@ const {
   ArrayBufferPrototypeGetDetached,
   ArrayBufferPrototypeSlice,
   ArrayPrototypePush,
-  ArrayPrototypeShift,
+  ArrayPrototypeSlice,
   AsyncIteratorPrototype,
   DataViewPrototypeGetBuffer,
   DataViewPrototypeGetByteLength,
@@ -44,8 +44,6 @@ const {
   getPromiseDetails,
 } = internalBinding('util');
 
-const assert = require('internal/assert');
-
 const {
   isDataView,
 } = require('internal/util/types');
@@ -157,27 +155,50 @@ function isBrandCheck(brand) {
 // single time and don't assert the existence of the queue fields (both
 // are unconditionally initialized during controller setup and only ever
 // replaced wholesale).
+//
+// The queue is drained from a moving head index instead of with
+// ArrayPrototypeShift: shifting is O(queue length), so draining a buffered
+// queue one chunk at a time is O(n^2). Reading through `queueHead` makes
+// each dequeue O(1); the consumed prefix is dropped in amortized O(1) once
+// it grows to at least half of the backing array (and past a small floor,
+// so a short oscillating queue never pays a per-dequeue array mutation).
+// Callers must use queueLength(), not `queue.length`, for the number of
+// live entries.
+function queueLength(controller) {
+  const state = controller[kState];
+  return state.queue.length - state.queueHead;
+}
+
 function dequeueValue(controller) {
   const state = controller[kState];
-  assert(state.queue.length);
+  const queue = state.queue;
+  const head = state.queueHead;
   const {
     value,
     size,
-  } = ArrayPrototypeShift(state.queue);
+  } = queue[head];
+  queue[head] = undefined;
+  const newHead = head + 1;
   state.queueTotalSize = MathMax(0, state.queueTotalSize - size);
+  if (newHead >= 32 && newHead * 2 >= queue.length) {
+    state.queue = ArrayPrototypeSlice(queue, newHead);
+    state.queueHead = 0;
+  } else {
+    state.queueHead = newHead;
+  }
   return value;
 }
 
 function resetQueue(controller) {
   const state = controller[kState];
   state.queue = [];
+  state.queueHead = 0;
   state.queueTotalSize = 0;
 }
 
 function peekQueueValue(controller) {
   const state = controller[kState];
-  assert(state.queue.length);
-  return state.queue[0].value;
+  return state.queue[state.queueHead].value;
 }
 
 function enqueueValueWithSize(controller, value, size) {
@@ -293,6 +314,7 @@ module.exports = {
   nonOpStart,
   nonOpWrite,
   peekQueueValue,
+  queueLength,
   rejectedHandledRecord,
   resetQueue,
   resolvedRecord,

lib/internal/webstreams/writablestream.js

@@ -74,6 +74,7 @@ const {
   nonOpStart,
   nonOpWrite,
   peekQueueValue,
+  queueLength,
   rejectedHandledRecord,
   resetQueue,
   resolvedRecord,
@@ -1143,12 +1144,11 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
 function writableStreamDefaultControllerProcessClose(controller) {
   const {
     closeAlgorithm,
-    queue,
     stream,
   } = controller[kState];
   writableStreamMarkCloseRequestInFlight(stream);
   dequeueValue(controller);
-  assert(!queue.length);
+  assert(queueLength(controller) === 0);
   const sinkClosePromise = closeAlgorithm();
   writableStreamDefaultControllerClearAlgorithms(controller);
   PromisePrototypeThen(
@@ -1223,7 +1223,6 @@ function writableStreamDefaultControllerGetBackpressure(controller) {
 
 function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) {
   const {
-    queue,
     started,
     stream,
   } = controller[kState];
@@ -1235,7 +1234,7 @@ function writableStreamDefaultControllerAdvanceQueueIfNeeded(controller) {
     return;
   }
 
-  if (!queue.length)
+  if (queueLength(controller) === 0)
     return;
 
   const value = peekQueueValue(controller);
@@ -1294,6 +1293,7 @@ function setupWritableStreamDefaultController(
     closeAlgorithm,
     highWaterMark,
     queue: [],
+    queueHead: 0,
     queueTotalSize: 0,
     abortController: new AbortController(),
     sizeAlgorithm,