stream: fast path single-destination pipe

Avoid EventEmitter dispatch for the readable pipe data handler when it
is the only data listener. Fall back to normal data event emission when
additional data listeners are present to preserve listener ordering.

Track the pipe data handler lazily so non-piped readable streams do not
pay an extra ReadableState property initialization or data listener
identity check.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5

Author: trivikr

lib/internal/streams/readable.js

@@ -111,6 +111,7 @@ const kErroredValue = Symbol('kErroredValue');
 const kDefaultEncodingValue = Symbol('kDefaultEncodingValue');
 const kDecoderValue = Symbol('kDecoderValue');
 const kEncodingValue = Symbol('kEncodingValue');
+const kPipeData = Symbol('kPipeData');
 
 const kEnded = 1 << 9;
 const kEndEmitted = 1 << 10;
@@ -289,6 +290,7 @@ function ReadableState(options, stream, isDuplex) {
   this.bufferIndex = 0;
   this.length = 0;
   this.pipes = [];
+  this[kPipeData] = null;
 
   // Should close be emitted on destroy. Defaults to true.
   if (options && options.emitClose === false) this[kState] &= ~kEmitClose;
@@ -564,8 +566,7 @@ function addChunk(stream, state, chunk, addToFront) {
       state.awaitDrainWriters = null;
     }
 
-    state[kState] |= kDataEmitted;
-    stream.emit('data', chunk);
+    emitData(stream, state, chunk);
   } else {
     // Update the buffer info.
     state.length += (state[kState] & kObjectMode) !== 0 ? 1 : chunk.length;
@@ -790,13 +791,21 @@ Readable.prototype.read = function(n) {
   }
 
   if (ret !== null && (state[kState] & (kErrorEmitted | kCloseEmitted)) === 0) {
-    state[kState] |= kDataEmitted;
-    this.emit('data', ret);
+    emitData(this, state, ret);
   }
 
   return ret;
 };
 
+function emitData(stream, state, chunk) {
+  state[kState] |= kDataEmitted;
+  if (stream._events.data === state[kPipeData]) {
+    state[kPipeData](chunk);
+  } else {
+    stream.emit('data', chunk);
+  }
+}
+
 function onEofChunk(stream, state) {
   debug('onEofChunk');
   if ((state[kState] & kEnded) !== 0) return;
@@ -976,6 +985,8 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     src.removeListener('end', onend);
     src.removeListener('end', unpipe);
     src.removeListener('data', ondata);
+    if (state[kPipeData] === ondata)
+      state[kPipeData] = null;
 
     cleanedUp = true;
 
@@ -1016,6 +1027,8 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   }
 
   src.on('data', ondata);
+  if (src._events.data === ondata)
+    state[kPipeData] = ondata;
   function ondata(chunk) {
     debug('ondata');
     try {

test/parallel/test-stream2-basic.js

@@ -146,6 +146,54 @@ class TestWriter extends EE {
   r.pipe(w);
 }
 
+{
+  // Verify data listener ordering around pipe.
+  function makeWritable(events) {
+    return new W({
+      write(chunk, enc, cb) {
+        events.push('write');
+        cb();
+      },
+    });
+  }
+
+  {
+    const events = [];
+    const r = R.from(['x']);
+    const w = makeWritable(events);
+
+    r.on('data', () => events.push('data'));
+    r.pipe(w);
+    w.on('finish', common.mustCall(() => {
+      assert.deepStrictEqual(events, ['data', 'write']);
+    }));
+  }
+
+  {
+    const events = [];
+    const r = R.from(['x']);
+    const w = makeWritable(events);
+
+    r.pipe(w);
+    r.on('data', () => events.push('data'));
+    w.on('finish', common.mustCall(() => {
+      assert.deepStrictEqual(events, ['write', 'data']);
+    }));
+  }
+
+  {
+    const events = [];
+    const r = R.from(['x']);
+    const w = makeWritable(events);
+
+    r.pipe(w);
+    r.prependListener('data', () => events.push('data'));
+    w.on('finish', common.mustCall(() => {
+      assert.deepStrictEqual(events, ['data', 'write']);
+    }));
+  }
+}
+
 
 [1, 2, 3, 4, 5, 6, 7, 8, 9].forEach(function(SPLIT) {
   // Verify unpipe