stream: bound pipeTo sync source normalization

[trivikr]

Avoid materializing all chunks produced while normalizing a single non-fast-path value in the sync iterable pipeTo fast path. Write normalized chunks incrementally in FROM_BATCH_SIZE batches instead, reducing peak memory and latency for nested iterable values.

[trivikr]

Example micro benchmark

Code

// Measures latency to the first write from pipeTo()'s sync iterable fast path
// when a source yields a nested iterable value.
'use strict';

const common = require('../common.js');

const bench = common.createBenchmark(main, {
  chunks: [256, 4096, 65536],
  chunkSize: [16],
  n: [100],
}, {
  flags: ['--experimental-stream-iter'],
  test: {
    chunks: 256,
    chunkSize: 16,
    n: 1,
  },
});

const kFirstWrite = new Error('first write');

function main({ chunks, chunkSize, n }) {
  const { pipeTo } = require('stream/iter');
  const writer = {
    write() {
      throw new Error('unexpected async write');
    },
    writeSync() {
      throw new Error('unexpected single-chunk write');
    },
    writev() {
      throw new Error('unexpected async writev');
    },
    writevSync() {
      throw kFirstWrite;
    },
  };

  function* nested() {
    for (let i = 0; i < chunks; i++) {
      yield new Uint8Array(chunkSize);
    }
  }

  function* source() {
    yield nested();
  }

  (async () => {
    bench.start();
    for (let i = 0; i < n; i++) {
      try {
        await pipeTo(source(), writer);
      } catch (err) {
        if (err !== kFirstWrite) {
          throw err;
        }
      }
    }
    bench.end(n);
  })();
}

Results

                                                                          confidence improvement accuracy (*)     (**)    (***)
streams/iter-pipeto-nested-first-write.js n=100 chunkSize=16 chunks=256          ***     51.85 %       ±5.06%   ±6.79%   ±8.95%
streams/iter-pipeto-nested-first-write.js n=100 chunkSize=16 chunks=4096         ***   1294.53 %      ±64.91%  ±87.48% ±116.13%
streams/iter-pipeto-nested-first-write.js n=100 chunkSize=16 chunks=65536        ***  27658.47 %     ±398.62% ±537.23% ±713.23%