streams: make pipe data fast path state lazy

Author: trivikr

lib/internal/streams/readable.js

@@ -132,6 +132,7 @@ const kFlowing = 1 << 24;
 const kHasPaused = 1 << 25;
 const kPaused = 1 << 26;
 const kDataListening = 1 << 27;
+const kHasPipeData = 1 << 28;
 
 // TODO(benjamingr) it is likely slower to do it this way than with free functions
 function makeBitMapDescriptor(bit) {
@@ -290,7 +291,6 @@ 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;
@@ -799,7 +799,8 @@ Readable.prototype.read = function(n) {
 
 function emitData(stream, state, chunk) {
   state[kState] |= kDataEmitted;
-  if (stream._events.data === state[kPipeData]) {
+  if ((state[kState] & kHasPipeData) !== 0 &&
+      stream._events.data === state[kPipeData]) {
     state[kPipeData](chunk);
   } else {
     stream.emit('data', chunk);
@@ -986,7 +987,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     src.removeListener('end', unpipe);
     src.removeListener('data', ondata);
     if (state[kPipeData] === ondata)
-      state[kPipeData] = null;
+      state[kState] &= ~kHasPipeData;
 
     cleanedUp = true;
 
@@ -1027,8 +1028,10 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   }
 
   src.on('data', ondata);
-  if (src._events.data === ondata)
+  if (src._events.data === ondata) {
     state[kPipeData] = ondata;
+    state[kState] |= kHasPipeData;
+  }
   function ondata(chunk) {
     debug('ondata');
     try {