stream: skip unobserved 'readable' emission at EOF

When a stream reaches its end via push(null) while nobody is observing
it (no 'readable' listener, not flowing, no pending readable need), do
not schedule the deferred 'readable' emission: it fires into the void
and costs a tick. Record that the end-of-stream notification is owed
instead; attaching a 'readable' listener later emits it, and read()
and resume() reach the end of the stream on their own.

As a side effect, a 'readable' listener attached more than a tick
after an unobserved end now receives the owed 'readable' before 'end',
where it previously received only 'end'.

For an HTTP server this removes one nextTick and one dead emit per
request whose body the handler never reads.

Signed-off-by: Matteo Collina <hello@matteocollina.com>
PR-URL: #65749
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Paolo Insogna <paolo@cowtech.it>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: mcollina

lib/internal/streams/readable.js

@@ -131,6 +131,7 @@ const kHasPaused = 1 << 25;
 const kPaused = 1 << 26;
 const kDataListening = 1 << 27;
 const kEndScheduled = 1 << 28;
+const kEofReadablePending = 1 << 29;
 
 // TODO(benjamingr) it is likely slower to do it this way than with free functions
 function makeBitMapDescriptor(bit) {
@@ -670,7 +671,7 @@ Readable.prototype.read = function(n) {
     state.highWaterMark = computeNewHighWaterMark(n);
 
   if (n !== 0)
-    state[kState] &= ~kEmittedReadable;
+    state[kState] &= ~(kEmittedReadable | kEofReadablePending);
 
   // If we're doing read(0) to trigger a readable event, but we
   // already have a bunch of data in the buffer, then just trigger
@@ -813,7 +814,16 @@ function onEofChunk(stream, state) {
     // If we are sync, wait until next tick to emit the data.
     // Otherwise we risk emitting data in the flow()
     // the readable code triggers during a read() call.
-    emitReadable(stream);
+    if ((state[kState] & (kFlowing | kNeedReadable)) !== 0 ||
+        stream.listenerCount('readable') > 0) {
+      emitReadable(stream);
+    } else {
+      // Nobody is observing the stream: do not schedule the 'readable'
+      // emission at all. A 'readable' listener attached later redeems it
+      // (see Readable.prototype.on), and read() and resume() reach the
+      // end of the stream on their own.
+      state[kState] |= kEofReadablePending;
+    }
   } else {
     // Emit 'readable' now to make sure it gets picked up.
     state[kState] &= ~kNeedReadable;
@@ -1162,8 +1172,17 @@ Readable.prototype.on = function(ev, fn) {
       debug('on readable');
       if (state.length) {
         emitReadable(this);
-      } else if ((state[kState] & kReading) === 0) {
-        process.nextTick(nReadingNextTick, this);
+      } else {
+        if ((state[kState] & kEofReadablePending) !== 0) {
+          // The end-of-stream 'readable' emission was skipped because
+          // nobody was listening when the stream ended (see onEofChunk):
+          // emit it now.
+          state[kState] &= ~kEofReadablePending;
+          emitReadable(this);
+        }
+        if ((state[kState] & kReading) === 0) {
+          process.nextTick(nReadingNextTick, this);
+        }
       }
     }
   }

test/parallel/test-stream-readable-unobserved-eof-readable.js

@@ -0,0 +1,76 @@
+'use strict';
+const common = require('../common');
+const { Readable } = require('stream');
+const assert = require('assert');
+
+// When a stream reaches its end while nobody is observing it, the
+// end-of-stream 'readable' emission is not scheduled. It is emitted
+// later if a 'readable' listener is attached, and read()/resume()
+// still reach 'end' on their own.
+
+{
+  // Listener attached synchronously after push(null) gets 'readable'
+  // and then 'end'.
+  const r = new Readable({ read() {} });
+  r.push(null);
+  let readableEmitted = false;
+  r.on('readable', common.mustCall(() => {
+    readableEmitted = true;
+    assert.strictEqual(r.read(), null);
+  }));
+  r.on('end', common.mustCall(() => {
+    assert.strictEqual(readableEmitted, true);
+  }));
+}
+
+{
+  // Listener attached one macrotask after the unobserved end still gets
+  // the owed 'readable' before 'end'.
+  const r = new Readable({ read() {} });
+  r.push(null);
+  setImmediate(common.mustCall(() => {
+    let readableEmitted = false;
+    r.on('readable', common.mustCall(() => {
+      readableEmitted = true;
+      assert.strictEqual(r.read(), null);
+    }));
+    r.on('end', common.mustCall(() => {
+      assert.strictEqual(readableEmitted, true);
+    }));
+  }));
+}
+
+{
+  // A stream that ends unobserved still emits 'end' when resumed later.
+  const r = new Readable({ read() {} });
+  r.push(null);
+  setImmediate(common.mustCall(() => {
+    r.resume();
+    r.on('end', common.mustCall());
+  }));
+}
+
+{
+  // read() after an unobserved end consumes the owed notification: a
+  // 'readable' listener attached afterwards does not receive it, but
+  // 'end' is still emitted.
+  const r = new Readable({ read() {} });
+  r.push(null);
+  setImmediate(common.mustCall(() => {
+    assert.strictEqual(r.read(), null);
+    r.on('end', common.mustCall());
+  }));
+}
+
+{
+  // A 'data' listener attached after an unobserved end still gets 'end'.
+  const r = new Readable({ read() {} });
+  r.push('x');
+  r.push(null);
+  setImmediate(common.mustCall(() => {
+    r.on('data', common.mustCall((chunk) => {
+      assert.strictEqual(chunk.toString(), 'x');
+    }));
+    r.on('end', common.mustCall());
+  }));
+}