Use SimpleQueue in WritableStream implementation

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 382 matching lines @@
       }
 
       if (stream[_controller] !== undefined) {
         throw new TypeError(streamErrors.illegalConstructor);
       }
 
       this[_controlledReadableStream] = stream;
 
       this[_underlyingSource] = underlyingSource;
 
-      this[_queue] = new Queue();
+      this[_queue] = new binding.SimpleQueue();
       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 Queue();
+    controller[_queue] = new binding.SimpleQueue();
 
     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 Queue();
+      this[_readRequests] = new binding.SimpleQueue();
     }
 
     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 Queue();
+    controller[_queue] = new binding.SimpleQueue();
     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) {
       reader[_readRequests].forEach(request => v8.rejectPromise(request, e));
-      reader[_readRequests] = new Queue();
+      reader[_readRequests] = new binding.SimpleQueue();
     }
 
     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) {
       reader[_readRequests].forEach(request =>
           v8.resolvePromise(request, CreateIterResultObject(undefined, true)));
-      reader[_readRequests] = new Queue();
+      reader[_readRequests] = new binding.SimpleQueue();
     }
 
     v8.resolvePromise(reader[_closedPromise], undefined);
   }
 
   function ReadableStreamDefaultControllerGetDesiredSize(controller) {
     const queueSize = GetTotalQueueSize(controller);
     return controller[_strategyHWM] - queueSize;
   }
 
@@ skipping 248 matching lines @@
         const cancelResult = ReadableStreamCancel(stream, compositeReason);
         v8.resolvePromise(promise, cancelResult);
       }
 
       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.
-  const QUEUE_MAX_ARRAY_SIZE = 16384;
-  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 === QUEUE_MAX_ARRAY_SIZE) {
-        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 === QUEUE_MAX_ARRAY_SIZE);
-        // assert(this.front.next !== undefined);
-        this.front = this.front.next;
-        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 === QUEUE_MAX_ARRAY_SIZE);
-          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 101 matching lines @@
       (underlyingSource, strategy) => {
         return new ReadableStream(
             underlyingSource, strategy, createWithExternalControllerSentinel);
       };
 
   // Temporary exports while pipeTo() and pipeThrough() are behind flags
   binding.ReadableStream_prototype_pipeThrough =
       ReadableStream_prototype_pipeThrough;
   binding.ReadableStream_prototype_pipeTo = ReadableStream_prototype_pipeTo;
 });