stream: skip write() checks in flowing pipe

pipe() installs one 'data' listener that calls dest.write() for
every chunk. That repeats encoding, mode, and end checks that stay
the same for a synchronous buffer write.

When that listener is still the only one, hand the Buffer to the
same synchronous write path without those checks. A second
listener, a non-buffer chunk, or a busy writable still goes through
emit('data').

On top of the flowing-read fast path, benchmark/streams/pipe.js is
about 31% faster (20 runs). Object-mode pipe and readable-readall
stay within noise.

Assisted-by: a closed-source coding agent
Signed-off-by: Yagiz Nizipli <yagiz@nizipli.com>
PR-URL: https://github.com/nodejs/node/pull/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>

lib/internal/streams/readable.js

@@ -301,6 +301,10 @@ function ReadableState(options, stream, isDuplex) {
   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.
@@ -985,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) &&
@@ -1175,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++)
@@ -1188,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();
 
@@ -1380,6 +1398,32 @@ 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) {
@@ -1425,7 +1469,7 @@ function flowSync(stream, state) {
 
     if ((state[kState] & (kErrorEmitted | kCloseEmitted)) === 0) {
       state[kState] |= kDataEmitted;
-      stream.emit('data', current);
+      deliverFlowChunk(stream, state, current);
     }
 
     // Nested read() moved fastChunk into the buffer and may have refilled.

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) {