stream: speed up flowing pipe of buffers

flow() pulls one already-buffered chunk and calls _read() for the
next one on every iteration. That goes through the general read()
path, which updates a holey buffer array and then pulls the chunk
back out.

While a synchronous byte-mode flow is in progress, keep that
prefetched chunk on the readable state and emit it directly.
_read() of the next chunk still runs before 'data', and a nested
read() moves the chunk back onto the buffer.

benchmark/streams/pipe.js is about 77% faster (15 runs).
pipe-object-mode, readable-readall, and readable-bigread stay
within noise.

Assisted-by: a closed-source coding agent
Signed-off-by: Yagiz Nizipli <yagiz@nizipli.com>
PR-URL: #66182
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
Reviewed-By: Zeyu "Alex" Yang <himself65@outlook.com>

Author: anonrig

lib/internal/streams/destroy.js

@@ -182,6 +182,7 @@ function undestroy() {
     r.errored = null;
     r.errorEmitted = false;
     r.reading = false;
+    r.fastChunk = null;
     r.ended = r.readable === false;
     r.endEmitted = r.readable === false;
   }

lib/internal/streams/readable.js

@@ -139,6 +139,9 @@ const kPaused = 1 << 26;
 const kDataListening = 1 << 27;
 const kEndScheduled = 1 << 28;
 const kEofReadablePending = 1 << 29;
+// Set only while flowSync() is inside _read(). push() then keeps the
+// chunk on state.fastChunk instead of the buffer array.
+const kFastPush = 1 << 30;
 
 // TODO(benjamingr) it is likely slower to do it this way than with free functions
 function makeBitMapDescriptor(bit) {
@@ -296,6 +299,8 @@ function ReadableState(options, stream, isDuplex) {
   this.buffer = [];
   this.bufferIndex = 0;
   this.length = 0;
+  // Chunk prefetched by flowSync(), kept off the buffer array.
+  this.fastChunk = null;
   this.pipes = [];
 
   // Should close be emitted on destroy. Defaults to true.
@@ -400,9 +405,29 @@ Readable.prototype[SymbolAsyncDispose] = async function() {
 // similar to how Writable.write() returns true if you should
 // write() some more.
 Readable.prototype.push = function(chunk, encoding) {
+  const state = this._readableState;
+
+  // flowSync() is inside _read() and wants a single buffer parked on
+  // fastChunk. A second push, a string, or EOF drops back to the buffer.
+  if ((state[kState] & kFastPush) !== 0 && encoding == null &&
+      state.fastChunk == null && chunk instanceof Buffer && chunk.length > 0) {
+    state.fastChunk = chunk;
+    state.length = chunk.length;
+    state[kState] &= ~kReading;
+    return chunk.length < state.highWaterMark;
+  }
+  if ((state[kState] & kFastPush) !== 0) {
+    state[kState] &= ~kFastPush;
+    if (state.fastChunk != null) {
+      const first = state.fastChunk;
+      state.fastChunk = null;
+      state.length = 0;
+      readableAddChunkPushByteMode(this, state, first);
+    }
+  }
+
   debug('push', chunk);
 
-  const state = this._readableState;
   return (state[kState] & kObjectMode) === 0 ?
     readableAddChunkPushByteMode(this, state, chunk, encoding) :
     readableAddChunkPushObjectMode(this, state, chunk, encoding);
@@ -601,6 +626,8 @@ Readable.prototype.isPaused = function() {
 // Backwards compatibility.
 Readable.prototype.setEncoding = function(enc) {
   const state = this._readableState;
+  if (state.fastChunk != null)
+    materializeFastChunk(state);
 
   const decoder = new StringDecoder(enc);
   state.decoder = decoder;
@@ -667,6 +694,11 @@ function howMuchToRead(n, state) {
 
 // You can override either this method, or the async _read(n) below.
 Readable.prototype.read = function(n) {
+  // A nested read() during flowSync()'s 'data' event must see the
+  // prefetched chunk. Null for every read that is not inside that loop.
+  if (this._readableState.fastChunk != null)
+    materializeFastChunk(this._readableState);
+
   debug('read', n);
   // Same as parseInt(undefined, 10), however V8 7.3 performance regressed
   // in this scenario, so we are doing it manually.
@@ -1336,9 +1368,91 @@ Readable.prototype.pause = function() {
 function flow(stream) {
   const state = stream._readableState;
   debug('flow');
+  // Byte-mode pipe sits in read() to pull one already-buffered chunk and
+  // refill. That read is most of the per-chunk cost. flowSync() keeps the
+  // same prefetch order without the buffer array or the general read path.
+  if (flowSync(stream, state))
+    return;
   while ((state[kState] & kFlowing) !== 0 && stream.read() !== null);
 }
 
+const kFastFlowNeed = kConstructed | kFlowing | kDataListening;
+const kFastFlowBlock = kObjectMode | kDecoder | kEnded | kDestroyed |
+  kErrored | kPaused | kReading | kSync;
+
+// Returns true when this call owned the flowing loop, including any
+// fallback to read() after the fast path stops.
+function flowSync(stream, state) {
+  const bits = state[kState];
+  if ((bits & kFastFlowNeed) !== kFastFlowNeed ||
+      (bits & kFastFlowBlock) !== 0 ||
+      !(state.highWaterMark > 0) ||
+      state.fastChunk != null) {
+    return false;
+  }
+
+  if (state.length !== 0) {
+    const buf = state.buffer;
+    const idx = state.bufferIndex;
+    const chunk = buf[idx];
+    // Only the one-chunk prefetch left by read(0) / the previous read.
+    if (buf.length !== idx + 1 || chunk == null || chunk.length !== state.length)
+      return false;
+    buf.length = 0;
+    state.bufferIndex = 0;
+    state.fastChunk = chunk;
+  }
+
+  while ((state[kState] & kFlowing) !== 0) {
+    const current = state.fastChunk;
+    if (current == null)
+      break;
+    if ((state[kState] & kFastFlowBlock) !== 0)
+      break;
+
+    // _read() of the next chunk runs before 'data', matching read().
+    state.fastChunk = null;
+    state.length = 0;
+    state[kState] |= kReading | kSync | kFastPush;
+    try {
+      stream._read(state.highWaterMark);
+    } catch (err) {
+      state[kState] &= ~(kSync | kFastPush);
+      errorOrDestroy(stream, err);
+      break;
+    }
+    state[kState] &= ~(kSync | kFastPush);
+
+    if ((state[kState] & (kErrorEmitted | kCloseEmitted)) === 0) {
+      state[kState] |= kDataEmitted;
+      stream.emit('data', current);
+    }
+
+    // Nested read() moved fastChunk into the buffer and may have refilled.
+    if (state.fastChunk == null && state.length !== 0)
+      break;
+  }
+
+  if (state.fastChunk != null)
+    materializeFastChunk(state);
+
+  while ((state[kState] & kFlowing) !== 0 && stream.read() !== null);
+  return true;
+}
+
+// Put fastChunk at the head of the buffer. state.length already counts it.
+function materializeFastChunk(state) {
+  const chunk = state.fastChunk;
+  if (chunk == null)
+    return;
+  state.fastChunk = null;
+  if (state.bufferIndex > 0) {
+    state.buffer[--state.bufferIndex] = chunk;
+  } else {
+    state.buffer.unshift(chunk);
+  }
+}
+
 // Wrap an old-style stream as the async data source.
 // This is *not* part of the readable stream interface.
 // It is an ugly unfortunate mess of history.
@@ -1724,7 +1838,16 @@ ObjectDefineProperties(Readable.prototype, {
     __proto__: null,
     enumerable: false,
     get: function() {
-      return this._readableState?.buffer;
+      const state = this._readableState;
+      if (state == null)
+        return undefined;
+      if (state.fastChunk == null)
+        return state.buffer;
+      if (state.bufferIndex === state.buffer.length)
+        return [state.fastChunk];
+      const out = state.buffer.slice(state.bufferIndex);
+      out.push(state.fastChunk);
+      return out;
     },
   },