Fix ReadableStream scalability issue

third_party/WebKit/Source/core/streams/ReadableStream.js

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 (function(global, binding, v8) {
   'use strict';
 
   const _reader = v8.createPrivateSymbol('[[reader]]');
   const _storedError = v8.createPrivateSymbol('[[storedError]]');
   const _controller = v8.createPrivateSymbol('[[controller]]');
@@ skipping 177 matching lines @@
       }
 
       if (stream[_controller] !== undefined) {
         throw new TypeError(streamErrors.illegalConstructor);
       }
 
       this[_controlledReadableStream] = stream;
 
       this[_underlyingSource] = underlyingSource;
 
-      this[_queue] = new v8.InternalPackedArray();
+      this[_queue] = new Queue();
       this[_totalQueuedSize] = 0;
 
       this[_readableStreamDefaultControllerBits] = 0b0;
       if (isExternallyControlled === true) {
         this[_readableStreamDefaultControllerBits] |= EXTERNALLY_CONTROLLED;
       }
 
       const normalizedStrategy =
           ValidateAndNormalizeQueuingStrategy(size, highWaterMark);
       this[_strategySize] = normalizedStrategy.size;
@@ skipping 80 matching lines @@
       }
       if (state === STATE_CLOSED) {
         throw new TypeError(errErrorClosedStream);
       }
 
       return ReadableStreamDefaultControllerError(this, e);
     }
   }
 
   function ReadableStreamDefaultControllerCancel(controller, reason) {
-    controller[_queue] = new v8.InternalPackedArray();
+    controller[_queue] = new Queue();
 
     const underlyingSource = controller[_underlyingSource];
     return PromiseCallOrNoop(underlyingSource, 'cancel', reason, 'underlyingSource.cancel');
   }
 
   function ReadableStreamDefaultControllerPull(controller) {
     const stream = controller[_controlledReadableStream];
 
     if (controller[_queue].length > 0) {
       const chunk = DequeueValue(controller);
@@ skipping 23 matching lines @@
     constructor(stream) {
       if (IsReadableStream(stream) === false) {
         throw new TypeError(errReaderConstructorBadArgument);
       }
       if (IsReadableStreamLocked(stream) === true) {
         throw new TypeError(errReaderConstructorStreamAlreadyLocked);
       }
 
       ReadableStreamReaderGenericInitialize(this, stream);
 
-      this[_readRequests] = new v8.InternalPackedArray();
+      this[_readRequests] = new Queue();
     }
 
     get closed() {
       if (IsReadableStreamDefaultReader(this) === false) {
         return Promise_reject(new TypeError(streamErrors.illegalInvocation));
       }
 
       return this[_closedPromise];
     }
 
@@ skipping 122 matching lines @@
   function ReadableStreamGetState(stream) {
     return (stream[_readableStreamBits] & STATE_MASK) >> STATE_BITS_OFFSET;
   }
 
   function ReadableStreamSetState(stream, state) {
     stream[_readableStreamBits] = (stream[_readableStreamBits] & ~STATE_MASK) |
         (state << STATE_BITS_OFFSET);
   }
 
   function ReadableStreamDefaultControllerError(controller, e) {
-    controller[_queue] = new v8.InternalPackedArray();
+    controller[_queue] = new Queue();
     const stream = controller[_controlledReadableStream];
     ReadableStreamError(stream, e);
   }
 
   function ReadableStreamError(stream, e) {
     stream[_storedError] = e;
     ReadableStreamSetState(stream, STATE_ERRORED);
 
     const reader = stream[_reader];
     if (reader === undefined) {
       return undefined;
     }
 
     if (IsReadableStreamDefaultReader(reader) === true) {
-      const readRequests = reader[_readRequests];
-      for (let i = 0; i < readRequests.length; i++) {
-        v8.rejectPromise(readRequests[i], e);
-      }
-      reader[_readRequests] = new v8.InternalPackedArray();
+      reader[_readRequests].forEach(request => v8.rejectPromise(request, e));
+      reader[_readRequests] = new Queue();
     }
 
     v8.rejectPromise(reader[_closedPromise], e);
     v8.markPromiseAsHandled(reader[_closedPromise]);
   }
 
   function ReadableStreamClose(stream) {
     ReadableStreamSetState(stream, STATE_CLOSED);
 
     const reader = stream[_reader];
     if (reader === undefined) {
       return undefined;
     }
 
     if (IsReadableStreamDefaultReader(reader) === true) {
-      const readRequests = reader[_readRequests];
-      for (let i = 0; i < readRequests.length; i++) {
-        v8.resolvePromise(
-            readRequests[i], CreateIterResultObject(undefined, true));
-      }
-      reader[_readRequests] = new v8.InternalPackedArray();
+      reader[_readRequests].forEach(request =>
+          v8.resolvePromise(request, CreateIterResultObject(undefined, true)));
+      reader[_readRequests] = new Queue();
     }
 
     v8.resolvePromise(reader[_closedPromise], undefined);
   }
 
   function ReadableStreamDefaultControllerGetDesiredSize(controller) {
     const queueSize = GetTotalQueueSize(controller);
     return controller[_strategyHWM] - queueSize;
   }
 
@@ skipping 252 matching lines @@
       return promise;
     }
   }
 
   //
   // Queue-with-sizes
   // Modified from taking the queue (as in the spec) to taking the stream, so we
   // can modify the queue size alongside.
   //
 
+  // Simple queue structure. Avoids scalability issues with using
+  // InternalPackedArray directly by using multiple arrays
+  // in a linked list and keeping the array size bounded.
+  class Queue {
+    constructor() {
+      this.front = {
+        elements: new v8.InternalPackedArray(),
+        next: undefined,
+      };
+      this.back = this.front;
+      // The cursor is used to avoid calling InternalPackedArray.shift().
+      this.cursor = 0;
+      this.size = 0;
+    }
+
+    get length() {
+      return this.size;
+    }
+
+    push(element) {
+      ++this.size;
+      if (this.back.elements.length === 16384) {

[yhirano, 2017/01/24 11:14:13]

[optional] Having a constant might be good given that we are using it in assert
comments.

[Adam Rice, 2017/01/24 12:33:02]

I didn't do this because I wasn't sure that v8 would perform constant folding.
But that was just a premature optimisation.

Done.

+        const oldBack = this.back;
+        this.back = {
+          elements: new v8.InternalPackedArray(),
+          next: undefined,
+        };
+        oldBack.next = this.back;
+      }
+      this.back.elements.push(element);
+    }
+
+    shift() {
+      // assert(this.size > 0);
+      --this.size;
+      if (this.front.elements.length === this.cursor) {
+        // assert(this.cursor === MAX_ARRAY_LENGTH);
+        if (this.front.next !== undefined) {

[yhirano, 2017/01/24 11:14:13]

this.front.next === undefined && this.front.elements.length === this.cursor
means the queue is empty, which must not happen. Maybe assert(this.front.next
!== undefined) is enough?

[Adam Rice, 2017/01/24 12:33:02]

You are right. Done.

+          this.front = this.front.next;
+        } else {
+          // assert(this.front === this.back);
+          // assert(this.back.elements === MAX_ARRAY_LENGTH);
+          this.front.elements.length = 0;
+        }
+        this.cursor = 0;
+      }
+      const element = this.front.elements[this.cursor];
+      // Permit shifted element to be garbage collected.
+      this.front.elements[this.cursor] = undefined;
+      ++this.cursor;
+
+      return element;
+    }
+
+    forEach(callback) {
+      let i = this.cursor;
+      let node = this.front;
+      let elements = node.elements;
+      while (i !== elements.length || node.next !== undefined) {
+        if (i === elements.length) {
+          // assert(node.next !== undefined);
+          // assert(i === MAX_ARRAY_LENGTH);
+          node = node.next;
+          elements = node.elements;
+          i = 0;
+        }
+        callback(elements[i]);
+        ++i;
+      }
+    }
+  }
+
   function DequeueValue(controller) {
     const result = controller[_queue].shift();
     controller[_totalQueuedSize] -= result.size;
     return result.value;
   }
 
   function EnqueueValueWithSize(controller, value, size) {
     size = Number(size);
     if (Number_isNaN(size) || size === +Infinity || size < 0) {
       throw new RangeError(streamErrors.invalidSize);
@@ skipping 96 matching lines @@
   binding.ReadableStreamDefaultControllerGetDesiredSize = ReadableStreamDefaultControllerGetDesiredSize;
   binding.ReadableStreamDefaultControllerEnqueue = ReadableStreamDefaultControllerEnqueue;
   binding.ReadableStreamDefaultControllerError = ReadableStreamDefaultControllerError;
 
   binding.createReadableStreamWithExternalController =
       (underlyingSource, strategy) => {
         return new ReadableStream(
             underlyingSource, strategy, createWithExternalControllerSentinel);
       };
 });