Repository: nodejs/node
File: lib/internal/webstreams/readablestream.js (JavaScript)

Code before:

  function watchClosed(stream, promise, action) {
    if (stream[kState].state === 'closed')
      action();
    else
      PromisePrototypeThen(promise, action, () => {});
  }

  async function step() {
    if (shuttingDown)
      return true;

    await writer[kState].ready.promise;

    const controller = source[kState].controller;

    // Fast path: batch reads when data is buffered in a default controller.
    // This avoids creating PipeToReadableStreamReadRequest objects and
    // reduces promise allocation overhead.
    if (source[kState].state === 'readable' &&
        isReadableStreamDefaultController(controller) &&
        controller[kState].queue.length > 0) {
      let batchCount = 0;
      const hwm = controller[kState].highWaterMark || 1;

      while (controller[kState].queue.length > 0 && batchCount < hwm) {
        if (shuttingDown) return true;

        source[kState].disturbed = true;
        const chunk = dequeueValue(controller);

        if (controller[kState].closeRequested && !controller[kState].queue.length) {
          readableStreamDefaultControllerClearAlgorithms(controller);
          readableStreamClose(source);
        }

        // Write the chunk - we're already in a separate microtask from enqueue
        // because we awaited writer[kState].ready.promise above
        state.currentWrite = writableStreamDefaultWriterWrite(writer, chunk);
        setPromiseHandled(state.currentWrite);

        batchCount++;

        // Check backpressure after each write
        if (writer[kState].ready.promise !== undefined &&
            dest[kState].state === 'writable') {
          const desiredSize = writer.desiredSize;
          if (desiredSize !== null && desiredSize <= 0) {
            // Backpressure - stop batch and wait for ready
            break;
          }

Review comment (performance):
Comparing `batchCount` to `hwm` is only valid if `controller[kState].sizeAlgorithm` is the default size algorithm (i.e. each chunk has size 1). I don't think we can assume that?

We should add a test that uses a custom size algorithm.

Code after:

  function watchClosed(stream, promise, action) {
    if (stream[kState].state === 'closed')
      action();
    else
      PromisePrototypeThen(promise, action, () => {});
  }

  async function step() {
    if (shuttingDown)
      return true;

    await writer[kState].ready.promise;

    const controller = source[kState].controller;

    // Fast path: batch reads when data is buffered in a default controller.
    // This avoids creating PipeToReadableStreamReadRequest objects and
    // reduces promise allocation overhead.
    if (source[kState].state === 'readable' &&
        isReadableStreamDefaultController(controller) &&
        controller[kState].queue.length > 0) {

      while (controller[kState].queue.length > 0) {
        if (shuttingDown) return true;

        source[kState].disturbed = true;
        const chunk = dequeueValue(controller);

        if (controller[kState].closeRequested && !controller[kState].queue.length) {
          readableStreamDefaultControllerClearAlgorithms(controller);
          readableStreamClose(source);
        }

        // Write the chunk - we're already in a separate microtask from enqueue
        // because we awaited writer[kState].ready.promise above
        state.currentWrite = writableStreamDefaultWriterWrite(writer, chunk);
        setPromiseHandled(state.currentWrite);

        // Check backpressure after each write
        if (dest[kState].state === 'writable') {
          const desiredSize = writer.desiredSize;
          if (desiredSize !== null && desiredSize <= 0) {
            // Backpressure - stop batch and wait for ready
            break;
          }
        }
      }

      // Trigger pull if needed after batch
      if (source[kState].state === 'readable' &&