node/lib/internal/webstreams/util.js

'use strict';

const {
  Array,
  ArrayBufferPrototypeGetByteLength,
  ArrayBufferPrototypeGetDetached,
  ArrayBufferPrototypeSlice,
  DataViewPrototypeGetBuffer,
  DataViewPrototypeGetByteLength,
  DataViewPrototypeGetByteOffset,
  FunctionPrototypeCall,
  MathMax,
  NumberIsNaN,
  ObjectFreeze,
  Promise,
  PromisePrototypeThen,
  PromiseReject,
  PromiseResolve,
  Symbol,
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
  Uint8Array,
} = primordials;

const {
  codes: {
    ERR_INVALID_ARG_VALUE,
  },
} = require('internal/errors');

const {
  copyArrayBuffer,
} = internalBinding('buffer');

const {
  inspect,
} = require('util');

const {
  constants: {
    kPending,
  },
  getPromiseDetails,
} = internalBinding('util');

const assert = require('internal/assert');

const {
  isDataView,
} = require('internal/util/types');

const {
  validateFunction,
} = require('internal/validators');

const kState = Symbol('kState');
const kType = Symbol('kType');

const getNonWritablePropertyDescriptor = (value) => {
  return {
    __proto__: null,
    configurable: true,
    value,
  };
};

function extractHighWaterMark(value, defaultHWM) {
  if (value === undefined) return defaultHWM;
  const coercedValue = +value;
  if (NumberIsNaN(coercedValue) ||
      coercedValue < 0)
    throw new ERR_INVALID_ARG_VALUE.RangeError('strategy.highWaterMark', value);
  return coercedValue;
}

// The default size algorithm is never exposed to user code, so a single
// shared function avoids one closure allocation per stream.
const defaultSizeAlgorithm = () => 1;

function extractSizeAlgorithm(size) {
  if (size === undefined) return defaultSizeAlgorithm;
  validateFunction(size, 'strategy.size');
  return size;
}

function customInspect(depth, options, name, data) {
  if (depth < 0)
    return this;

  const opts = {
    ...options,
    depth: options.depth == null ? null : options.depth - 1,
  };

  return `${name} ${inspect(data, opts)}`;
}

// These use the original prototype getters so that user tampering with
// the buffer, byteLength, and byteOffset properties on ArrayBuffer and
// ArrayBufferView's is not observed. They run once or more per chunk on
// every byte-stream path, so they must not go through a reflective get
// (the previous view.constructor.prototype lookup was both slower and
// spoofable via a user-defined .constructor).

function ArrayBufferViewGetBuffer(view) {
  return isDataView(view) ?
    DataViewPrototypeGetBuffer(view) :
    TypedArrayPrototypeGetBuffer(view);
}

function ArrayBufferViewGetByteLength(view) {
  return isDataView(view) ?
    DataViewPrototypeGetByteLength(view) :
    TypedArrayPrototypeGetByteLength(view);
}

function ArrayBufferViewGetByteOffset(view) {
  return isDataView(view) ?
    DataViewPrototypeGetByteOffset(view) :
    TypedArrayPrototypeGetByteOffset(view);
}

function cloneAsUint8Array(view) {
  const buffer = ArrayBufferViewGetBuffer(view);
  const byteOffset = ArrayBufferViewGetByteOffset(view);
  const byteLength = ArrayBufferViewGetByteLength(view);
  return new Uint8Array(
    ArrayBufferPrototypeSlice(buffer, byteOffset, byteOffset + byteLength),
  );
}

function canCopyArrayBuffer(toBuffer, toIndex, fromBuffer, fromIndex, count) {
  return toBuffer !== fromBuffer &&
    !ArrayBufferPrototypeGetDetached(toBuffer) &&
    !ArrayBufferPrototypeGetDetached(fromBuffer) &&
    toIndex + count <= ArrayBufferPrototypeGetByteLength(toBuffer) &&
    fromIndex + count <= ArrayBufferPrototypeGetByteLength(fromBuffer);
}

function isBrandCheck(brand) {
  return (value) => {
    return value != null &&
           value[kState] !== undefined &&
           value[kType] === brand;
  };
}

// Backing store for the spec's [[queue]]: a power-of-two ring buffer
// instead of a plain array. Entries are pushed at the tail and consumed
// at the head, so a plain array either moves every element or forces the
// engine to re-linearize on each shift, and the default controllers would
// additionally have to allocate a { value, size } wrapper object per
// chunk just to keep the pair together. Default readable/writable
// controller queues store each entry as (value, size) in two consecutive
// slots via the *Pair methods; the readable byte controller queue stores
// its chunk descriptor records in single slots via push/shift/peek. A
// given instance only ever uses one of the two access patterns, so
// head/tail stay aligned to the entry stride.
class Queue {
  constructor(listLength = 8) {
    this.head = 0;
    this.tail = 0;
    // Number of logical entries currently in the queue: (value, size)
    // pairs for default controller queues, descriptor records for byte
    // controller queues.
    this.length = 0;
    this.capacityMask = listLength - 1;
    this.list = new Array(listLength);
    this.dequeuedSize = 0;
  }

  // Single-slot entries (readable byte controller chunk records).

  push(entry) {
    if (this.length === this.list.length)
      this.grow();
    const tail = this.tail;
    this.list[tail] = entry;
    this.tail = (tail + 1) & this.capacityMask;
    this.length++;
  }

  shift() {
    const head = this.head;
    const list = this.list;
    const entry = list[head];
    list[head] = undefined;
    this.head = (head + 1) & this.capacityMask;
    if (--this.length === 0)
      this.rewind();
    return entry;
  }

  peek() {
    return this.list[this.head];
  }

  // Two-slot (value, size) entries (default controller queues). The
  // stride is always 2 and capacities are even, so `tail + 1`/`head + 1`
  // never need to wrap.

  pushPair(value, size) {
    if (this.length * 2 === this.list.length)
      this.grow();
    const tail = this.tail;
    const list = this.list;
    list[tail] = value;
    list[tail + 1] = size;
    this.tail = (tail + 2) & this.capacityMask;
    this.length++;
  }

  // Returns the dequeued value; the size of the same entry is left in
  // `this.dequeuedSize` so that callers can update [[queueTotalSize]]
  // without a per-entry wrapper object having to exist.
  shiftPair() {
    const head = this.head;
    const list = this.list;
    const value = list[head];
    this.dequeuedSize = list[head + 1];
    list[head] = undefined;
    list[head + 1] = undefined;
    this.head = (head + 2) & this.capacityMask;
    if (--this.length === 0)
      this.rewind();
    return value;
  }

  peekPairValue() {
    return this.list[this.head];
  }

  // The ring is completely full (the tail has caught up with the head, so
  // the next push would overwrite the oldest entry): double the capacity,
  // re-linearizing from the head so index arithmetic stays trivial.
  // Growing before the push rather than after it lets the initial 8-slot
  // ring hold four (value, size) pairs without reallocating.
  grow() {
    const list = this.list;
    const capacity = list.length;
    const head = this.head;
    if (head !== 0) {
      const relinearized = new Array(capacity * 2);
      let n = 0;
      for (let i = head; i < capacity; i++)
        relinearized[n++] = list[i];
      for (let i = 0; i < head; i++)
        relinearized[n++] = list[i];
      this.list = relinearized;
      this.head = 0;
    } else {
      list.length = capacity * 2;
    }
    this.tail = capacity;
    this.capacityMask = (capacity * 2) - 1;
  }

  // The queue just became empty: restart at slot 0 so shallow queues net
  // sequential slot access, and drop the enlarged backing store after a
  // large burst has fully drained.
  rewind() {
    this.head = 0;
    this.tail = 0;
    if (this.list.length > 1024) {
      this.list.length = 8;
      this.capacityMask = 0b111;
    }
  }
}

// Controllers start out with (and are reset to) this shared immutable
// empty queue, so constructing a stream never allocates queue storage;
// a real Queue is materialized by the enqueue paths on first use. All
// dequeue/peek paths are guarded by `.length` (or the equivalent
// [[queueTotalSize]]) checks, so they can never observe the sentinel in
// a mutating way; it never stores entries, so it gets a zero-length
// backing list.
const kEmptyQueue = ObjectFreeze(new Queue(0));

function materializeQueue(state) {
  const queue = state.queue;
  if (queue === kEmptyQueue)
    return state.queue = new Queue();
  return queue;
}

// The queue helpers below run once per chunk on the hot paths of every
// default readable/writable stream, so they load the controller state a
// single time and don't assert the existence of the queue fields (both
// are unconditionally initialized during controller setup and only ever
// replaced wholesale).
function dequeueValue(controller) {
  const state = controller[kState];
  const queue = state.queue;
  assert(queue.length);
  const value = queue.shiftPair();
  state.queueTotalSize = MathMax(0, state.queueTotalSize - queue.dequeuedSize);
  return value;
}

function resetQueue(controller) {
  const state = controller[kState];
  state.queue = kEmptyQueue;
  state.queueTotalSize = 0;
}

function peekQueueValue(controller) {
  const state = controller[kState];
  assert(state.queue.length);
  return state.queue.peekPairValue();
}

function enqueueValueWithSize(controller, value, size) {
  const state = controller[kState];
  const coercedSize = +size;
  if (NumberIsNaN(coercedSize) ||
      coercedSize < 0 ||
      coercedSize === Infinity) {
    throw new ERR_INVALID_ARG_VALUE.RangeError('size', size);
  }
  materializeQueue(state).pushPair(value, coercedSize);
  state.queueTotalSize += coercedSize;
}

// Arity-specialized variants of the promise-callback wrapper. The generic
// rest-parameter + ReflectApply form allocated an arguments array on every
// invocation; these run on per-chunk hot paths (pull/write/transform), so
// each known call-site arity gets its own wrapper. The exact number of
// arguments passed through to the user callback is observable and must be
// preserved.
function createPromiseCallbackNoParams(name, fn, thisArg) {
  validateFunction(fn, name);
  return async () => FunctionPrototypeCall(fn, thisArg);
}

// Raw variants that skip the async wrapper's implicit result promise.
// Consumers of a raw callback invoke it inside try/catch and route the
// result through thenAlgorithmResult() below.
function createRawCallback1Param(name, fn, thisArg) {
  validateFunction(fn, name);
  return (arg) => FunctionPrototypeCall(fn, thisArg, arg);
}

function createRawCallback2Params(name, fn, thisArg) {
  validateFunction(fn, name);
  return (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
}

// A single shared, forever-resolved promise used to enqueue a reaction at
// the next microtask checkpoint without allocating a fresh promise.
const kResolvedPromise = PromiseResolve();

// Returned by an internal algorithm to signal that it parked the
// operation and takes responsibility for delivering the fulfilled (or
// rejected) continuation itself later, instead of settling a promise
// (see the transform stream source pull algorithm).
const kParkedAlgorithmResult = Symbol('kParkedAlgorithmResult');

// Start algorithm of internal streams whose creator delivers the
// post-start step itself (see transformStreamStart): the controller setup
// leaves `started` unset when the start result is this sentinel.
const kDeferredStart = Symbol('kDeferredStart');

function deferredStartAlgorithm() {
  return kDeferredStart;
}

// Wires the (possibly non-thenable) result of an underlying algorithm
// callback to its fulfilled/rejected continuations. A non-thenable result
// means fulfillment is guaranteed and no then() lookup is observable, so
// the fulfillment step is enqueued directly at the exact microtask
// position the coerced promise's reaction would have had, skipping the
// per-chunk promise allocation. For thenable results PromiseResolve()
// matches the spec's "a promise resolved with" conversion (identity for
// native promises).
function thenAlgorithmResult(result, onFulfilled, onRejected) {
  if (result === kParkedAlgorithmResult)
    return;
  if (result === null ||
      (typeof result !== 'object' && typeof result !== 'function')) {
    PromisePrototypeThen(kResolvedPromise, onFulfilled);
  } else {
    PromisePrototypeThen(PromiseResolve(result), onFulfilled, onRejected);
  }
}

function createPromiseCallback1Param(name, fn, thisArg) {
  validateFunction(fn, name);
  return async (arg) => FunctionPrototypeCall(fn, thisArg, arg);
}

function createPromiseCallback2Params(name, fn, thisArg) {
  validateFunction(fn, name);
  return async (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
}

function isPromisePending(promise) {
  if (promise === undefined) return false;
  const details = getPromiseDetails(promise);
  return details?.[0] === kPending;
}

// Shared shapes for lazily-materialized { promise, resolve, reject }
// records whose settlement is already known.
function resolvedRecord() {
  return {
    promise: PromiseResolve(),
    resolve: undefined,
    reject: undefined,
  };
}

function rejectedHandledRecord(error) {
  const record = {
    promise: PromiseReject(error),
    resolve: undefined,
    reject: undefined,
  };
  setPromiseHandled(record.promise);
  return record;
}

// A single shared, forever-pending promise carried by the duck-typed
// promise records that pipeTo and tee install on their internal reader
// and writer (see ClosedPromiseHook in readablestream.js): the probes on
// the erroring/release paths see a pending promise, and setPromiseHandled
// skips it so that no reaction ever accumulates on it.
const kPendingPromise = new Promise(() => {});

function setPromiseHandled(promise) {
  if (promise === kPendingPromise)
    return;
  // Alternatively, we could use the native API
  // MarkAsHandled, but this avoids the extra boundary cross
  // and is hopefully faster at the cost of an extra Promise
  // allocation.
  PromisePrototypeThen(promise, undefined, () => {});
}

async function nonOpFlush() {}

// Shared non-op for the start/pull/write algorithm callbacks, which all
// follow the raw-callback contract (see createRawCallback*): the
// non-thenable return takes the allocation-free fast path in
// thenAlgorithmResult().
function nonOpCallback() {}

async function nonOpCancel() {}

let transfer;
function lazyTransfer() {
  if (transfer === undefined)
    transfer = require('internal/webstreams/transfer');
  return transfer;
}

module.exports = {
  ArrayBufferViewGetBuffer,
  ArrayBufferViewGetByteLength,
  ArrayBufferViewGetByteOffset,
  Queue,
  canCopyArrayBuffer,
  cloneAsUint8Array,
  copyArrayBuffer,
  createPromiseCallbackNoParams,
  createPromiseCallback1Param,
  createPromiseCallback2Params,
  createRawCallback1Param,
  createRawCallback2Params,
  customInspect,
  defaultSizeAlgorithm,
  deferredStartAlgorithm,
  dequeueValue,
  enqueueValueWithSize,
  extractHighWaterMark,
  extractSizeAlgorithm,
  getNonWritablePropertyDescriptor,
  isBrandCheck,
  isPromisePending,
  kDeferredStart,
  kEmptyQueue,
  kParkedAlgorithmResult,
  kPendingPromise,
  kResolvedPromise,
  kState,
  kType,
  lazyTransfer,
  materializeQueue,
  nonOpCallback,
  nonOpCancel,
  nonOpFlush,

  peekQueueValue,
  rejectedHandledRecord,
  resetQueue,
  resolvedRecord,
  setPromiseHandled,
  thenAlgorithmResult,
};