stream: speed up flowing pipe of buffers

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,12 @@ 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;
+  // The sole pipe() 'data' listener, when there is exactly one destination.
+  // flowSync() writes buffers straight to that destination.
+  this.pipeOnData = null;
+  this.pipePause = null;
   this.pipes = [];
 
   // Should close be emitted on destroy. Defaults to true.
@@ -400,9 +409,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 +630,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 +698,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.
@@ -953,6 +989,16 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   }
 
   state.pipes.push(dest);
+  // Only a single pipe destination can skip emit('data') and call write()
+  // directly. A second destination, or an extra 'data' listener, must go
+  // through emit so every listener still runs.
+  if (state.pipes.length === 1) {
+    state.pipeOnData = ondata;
+    state.pipePause = pause;
+  } else {
+    state.pipeOnData = null;
+    state.pipePause = null;
+  }
   debug('pipe count=%d opts=%j', state.pipes.length, pipeOpts);
 
   const doEnd = (!pipeOpts || pipeOpts.end !== false) &&
@@ -1143,6 +1189,8 @@ Readable.prototype.unpipe = function(dest) {
     // remove all.
     const dests = state.pipes;
     state.pipes = [];
+    state.pipeOnData = null;
+    state.pipePause = null;
     this.pause();
 
     for (let i = 0; i < dests.length; i++)
@@ -1156,6 +1204,8 @@ Readable.prototype.unpipe = function(dest) {
     return this;
 
   state.pipes.splice(index, 1);
+  state.pipeOnData = null;
+  state.pipePause = null;
   if (state.pipes.length === 0)
     this.pause();
 
@@ -1336,7 +1386,115 @@ 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;
+
+let writeKnownBuffer;
+
+// Pipe's only listener is ondata(), which calls dest.write(). Skip emit
+// and the general write() checks for a single Buffer in that steady state.
+function deliverFlowChunk(stream, state, chunk) {
+  const pipeOnData = state.pipeOnData;
+  const events = stream._events;
+  if (pipeOnData !== null && events !== undefined && events.data === pipeOnData) {
+    writeKnownBuffer ??= require('internal/streams/writable').writeKnownBuffer;
+    const dest = state.pipes[0];
+    let ret;
+    try {
+      ret = writeKnownBuffer(dest, chunk);
+    } catch (error) {
+      dest.destroy(error);
+      return;
+    }
+    if (ret === undefined)
+      stream.emit('data', chunk);
+    else if (ret === false && state.pipePause !== null)
+      state.pipePause();
+    return;
+  }
+  stream.emit('data', chunk);
+}
+
+// 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;
+      deliverFlowChunk(stream, state, 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.
@@ -1724,7 +1882,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;
     },
   },
 

lib/internal/streams/writable.js

@@ -586,6 +586,43 @@ function writeOrBuffer(stream, state, chunk, encoding, callback) {
   return ret && (state[kState] & (kDestroyed | kErrored)) === 0;
 }
 
+// Steady state of a flowing pipe into a byte-mode Writable: one Buffer,
+// nothing queued, and no user write callback. Returns undefined when the
+// caller must use write() instead. Otherwise the same boolean as write().
+const kWriteFlowBlock = kObjectMode | kDestroyed | kErrored | kSync |
+  kEnding | kFinished | kWriting | kCorked | kBuffered | kEnded |
+  kNeedDrain | kWriteCb | kExpectWriteCb | kBufferProcessing |
+  kFinalCalled | kPrefinished | kOnFinished | kErrorEmitted;
+
+function writeKnownBuffer(stream, chunk) {
+  const state = stream._writableState;
+  if (state == null || state.length !== 0 || !(chunk instanceof Buffer))
+    return undefined;
+
+  const bits = state[kState];
+  if ((bits & kConstructed) === 0 || (bits & kWriteFlowBlock) !== 0)
+    return undefined;
+
+  const len = chunk.length;
+  state.pendingcb++;
+  state.length = len;
+  state.writelen = len;
+  state[kState] = bits | kWriting | kSync | kExpectWriteCb;
+  stream._write(chunk, 'buffer', state.onwrite);
+  state[kState] &= ~kSync;
+
+  const ret = state.length < state.highWaterMark || state.length === 0;
+  if (!ret)
+    state[kState] |= kNeedDrain;
+  return ret && (state[kState] & (kDestroyed | kErrored)) === 0;
+}
+
+ObjectDefineProperty(Writable, 'writeKnownBuffer', {
+  __proto__: null,
+  value: writeKnownBuffer,
+  enumerable: false,
+});
+
 function doWrite(stream, state, writev, len, chunk, encoding, cb) {
   state.writelen = len;
   if (cb !== nop) {