stream: apply more perf improvements to stream/new

Author: jasnell

lib/internal/streams/new/broadcast.js

@@ -8,12 +8,9 @@
 
 const {
   ArrayIsArray,
-  ArrayPrototypeIndexOf,
   ArrayPrototypeMap,
   ArrayPrototypePush,
-  ArrayPrototypeShift,
   ArrayPrototypeSlice,
-  ArrayPrototypeSplice,
   Error,
   MathMax,
   Promise,
@@ -54,6 +51,10 @@ const {
   allUint8Array,
 } = require('internal/streams/new/utils');
 
+const {
+  RingBuffer,
+} = require('internal/streams/new/ringbuffer');
+
 const encoder = new TextEncoder();
 
 // Cached resolved promise to avoid allocating a new one on every sync fast-path.
@@ -97,7 +98,7 @@ function parsePushArgs(args) {
 
 class BroadcastImpl {
   constructor(options) {
-    this._buffer = [];
+    this._buffer = new RingBuffer();
     this._bufferStart = 0;
     this._consumers = new SafeSet();
     this._ended = false;
@@ -157,7 +158,7 @@ class BroadcastImpl {
 
             const bufferIndex = state.cursor - self._bufferStart;
             if (bufferIndex < self._buffer.length) {
-              const chunk = self._buffer[bufferIndex];
+              const chunk = self._buffer.get(bufferIndex);
               state.cursor++;
               self._tryTrimBuffer();
               return { __proto__: null, done: false, value: chunk };
@@ -245,7 +246,7 @@ class BroadcastImpl {
         case 'block':
           return false;
         case 'drop-oldest':
-          ArrayPrototypeShift(this._buffer);
+          this._buffer.shift();
           this._bufferStart++;
           for (const consumer of this._consumers) {
             if (consumer.cursor < this._bufferStart) {
@@ -258,7 +259,7 @@ class BroadcastImpl {
       }
     }
 
-    ArrayPrototypePush(this._buffer, chunk);
+    this._buffer.push(chunk);
     this._notifyConsumers();
     return true;
   }
@@ -271,7 +272,7 @@ class BroadcastImpl {
       if (consumer.resolve) {
         const bufferIndex = consumer.cursor - this._bufferStart;
         if (bufferIndex < this._buffer.length) {
-          const chunk = this._buffer[bufferIndex];
+          const chunk = this._buffer.get(bufferIndex);
           consumer.cursor++;
           consumer.resolve({ done: false, value: chunk });
         } else {
@@ -330,7 +331,7 @@ class BroadcastImpl {
     const minCursor = this._getMinCursor();
     const trimCount = minCursor - this._bufferStart;
     if (trimCount > 0) {
-      ArrayPrototypeSplice(this._buffer, 0, trimCount);
+      this._buffer.trimFront(trimCount);
       this._bufferStart = minCursor;
 
       if (this._onBufferDrained &&
@@ -345,7 +346,7 @@ class BroadcastImpl {
       if (consumer.resolve) {
         const bufferIndex = consumer.cursor - this._bufferStart;
         if (bufferIndex < this._buffer.length) {
-          const chunk = this._buffer[bufferIndex];
+          const chunk = this._buffer.get(bufferIndex);
           consumer.cursor++;
           const resolve = consumer.resolve;
           consumer.resolve = null;
@@ -368,7 +369,7 @@ class BroadcastWriter {
     this._totalBytes = 0;
     this._closed = false;
     this._aborted = false;
-    this._pendingWrites = [];
+    this._pendingWrites = new RingBuffer();
     this._pendingDrains = [];
 
     this._broadcast._onBufferDrained = () => {
@@ -544,14 +545,14 @@ class BroadcastWriter {
   _createPendingWrite(chunk, signal) {
     return new Promise((resolve, reject) => {
       const entry = { chunk, resolve, reject };
-      ArrayPrototypePush(this._pendingWrites, entry);
+      this._pendingWrites.push(entry);
 
       if (!signal) return;
 
       const onAbort = () => {
         // Remove from queue so it doesn't occupy a slot
-        const idx = ArrayPrototypeIndexOf(this._pendingWrites, entry);
-        if (idx !== -1) ArrayPrototypeSplice(this._pendingWrites, idx, 1);
+        const idx = this._pendingWrites.indexOf(entry);
+        if (idx !== -1) this._pendingWrites.removeAt(idx);
         reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError'));
       };
 
@@ -573,7 +574,7 @@ class BroadcastWriter {
 
   _resolvePendingWrites() {
     while (this._pendingWrites.length > 0 && this._broadcast._canWrite()) {
-      const pending = ArrayPrototypeShift(this._pendingWrites);
+      const pending = this._pendingWrites.shift();
       if (this._broadcast._write(pending.chunk)) {
         for (let i = 0; i < pending.chunk.length; i++) {
           this._totalBytes += pending.chunk[i].byteLength;
@@ -587,10 +588,8 @@ class BroadcastWriter {
   }
 
   _rejectPendingWrites(error) {
-    const writes = this._pendingWrites;
-    this._pendingWrites = [];
-    for (let i = 0; i < writes.length; i++) {
-      writes[i].reject(error);
+    while (this._pendingWrites.length > 0) {
+      this._pendingWrites.shift().reject(error);
     }
   }
 

lib/internal/streams/new/push.js

@@ -6,11 +6,8 @@
 // with built-in backpressure.
 
 const {
-  ArrayPrototypeIndexOf,
   ArrayPrototypePush,
-  ArrayPrototypeShift,
   ArrayPrototypeSlice,
-  ArrayPrototypeSplice,
   Error,
   MathMax,
   Promise,
@@ -38,6 +35,10 @@ const {
   pull: pullWithTransforms,
 } = require('internal/streams/new/pull');
 
+const {
+  RingBuffer,
+} = require('internal/streams/new/ringbuffer');
+
 // Cached resolved promise to avoid allocating a new one on every sync fast-path.
 const kResolvedPromise = PromiseResolve();
 
@@ -48,11 +49,11 @@ const kResolvedPromise = PromiseResolve();
 class PushQueue {
   constructor(options = {}) {
     /** Buffered chunks (each slot is from one write/writev call) */
-    this._slots = [];
+    this._slots = new RingBuffer();
     /** Pending writes waiting for buffer space */
-    this._pendingWrites = [];
+    this._pendingWrites = new RingBuffer();
     /** Pending reads waiting for data */
-    this._pendingReads = [];
+    this._pendingReads = new RingBuffer();
     /** Pending drains waiting for backpressure to clear */
     this._pendingDrains = [];
     /** Writer state: 'open' | 'closed' | 'errored' */
@@ -134,7 +135,7 @@ class PushQueue {
           return false;
         case 'drop-oldest':
           if (this._slots.length > 0) {
-            ArrayPrototypeShift(this._slots);
+            this._slots.shift();
           }
           break;
         case 'drop-newest':
@@ -146,7 +147,7 @@ class PushQueue {
       }
     }
 
-    ArrayPrototypePush(this._slots, chunks);
+    this._slots.push(chunks);
     for (let i = 0; i < chunks.length; i++) {
       this._bytesWritten += chunks[i].byteLength;
     }
@@ -209,14 +210,14 @@ class PushQueue {
   _createPendingWrite(chunks, signal) {
     return new Promise((resolve, reject) => {
       const entry = { chunks, resolve, reject };
-      ArrayPrototypePush(this._pendingWrites, entry);
+      this._pendingWrites.push(entry);
 
       if (!signal) return;
 
       const onAbort = () => {
         // Remove from queue so it doesn't occupy a slot
-        const idx = ArrayPrototypeIndexOf(this._pendingWrites, entry);
-        if (idx !== -1) ArrayPrototypeSplice(this._pendingWrites, idx, 1);
+        const idx = this._pendingWrites.indexOf(entry);
+        if (idx !== -1) this._pendingWrites.removeAt(idx);
         reject(signal.reason ?? lazyDOMException('Aborted', 'AbortError'));
       };
 
@@ -301,7 +302,7 @@ class PushQueue {
     }
 
     return new Promise((resolve, reject) => {
-      ArrayPrototypePush(this._pendingReads, { resolve, reject });
+      this._pendingReads.push({ resolve, reject });
     });
   }
 
@@ -331,27 +332,27 @@ class PushQueue {
   _drain() {
     const result = [];
     for (let i = 0; i < this._slots.length; i++) {
-      const slot = this._slots[i];
+      const slot = this._slots.get(i);
       for (let j = 0; j < slot.length; j++) {
         ArrayPrototypePush(result, slot[j]);
       }
     }
-    this._slots = [];
+    this._slots.clear();
     return result;
   }
 
   _resolvePendingReads() {
     while (this._pendingReads.length > 0) {
       if (this._slots.length > 0) {
-        const pending = ArrayPrototypeShift(this._pendingReads);
+        const pending = this._pendingReads.shift();
         const result = this._drain();
         this._resolvePendingWrites();
         pending.resolve({ value: result, done: false });
       } else if (this._writerState === 'closed') {
-        const pending = ArrayPrototypeShift(this._pendingReads);
+        const pending = this._pendingReads.shift();
         pending.resolve({ value: undefined, done: true });
       } else if (this._writerState === 'errored' && this._error) {
-        const pending = ArrayPrototypeShift(this._pendingReads);
+        const pending = this._pendingReads.shift();
         pending.reject(this._error);
       } else {
         break;
@@ -362,8 +363,8 @@ class PushQueue {
   _resolvePendingWrites() {
     while (this._pendingWrites.length > 0 &&
            this._slots.length < this._highWaterMark) {
-      const pending = ArrayPrototypeShift(this._pendingWrites);
-      ArrayPrototypePush(this._slots, pending.chunks);
+      const pending = this._pendingWrites.shift();
+      this._slots.push(pending.chunks);
       for (let i = 0; i < pending.chunks.length; i++) {
         this._bytesWritten += pending.chunks[i].byteLength;
       }
@@ -392,18 +393,14 @@ class PushQueue {
   }
 
   _rejectPendingReads(error) {
-    const reads = this._pendingReads;
-    this._pendingReads = [];
-    for (let i = 0; i < reads.length; i++) {
-      reads[i].reject(error);
+    while (this._pendingReads.length > 0) {
+      this._pendingReads.shift().reject(error);
     }
   }
 
   _rejectPendingWrites(error) {
-    const writes = this._pendingWrites;
-    this._pendingWrites = [];
-    for (let i = 0; i < writes.length; i++) {
-      writes[i].reject(error);
+    while (this._pendingWrites.length > 0) {
+      this._pendingWrites.shift().reject(error);
     }
   }