stream: improve performance for sync write finishes

Improve performance and reduce memory usage when a writable stream
is written to with the same callback (which is the most common case)
and when the write operation finishes synchronously (which is also
often the case).

                                                         confidence improvement accuracy (*)    (**)   (***)
    streams/writable-manywrites.js sync='no' n=2000000                  0.99 %       ±3.20%  ±4.28%  ±5.61%
    streams/writable-manywrites.js sync='yes' n=2000000        ***    710.69 %      ±19.65% ±26.47% ±35.09%

Refs: https://github.com/nodejs/node/issues/18013
Refs: https://github.com/nodejs/node/issues/18367

PR-URL: https://github.com/nodejs/node/pull/30710
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

1 files changed, 28 insertions, 5 deletions

diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
index cf498e7d93..ab168fafe5 100644
--- a/lib/_stream_writable.js
+++ b/lib/_stream_writable.js
@@ -142,6 +142,10 @@ function WritableState(options, stream, isDuplex) {
   // The amount that is being written when _write is called.
   this.writelen = 0;
 
+  // Storage for data passed to the afterWrite() callback in case of
+  // synchronous _write() completion.
+  this.afterWriteTickInfo = null;
+
   this.bufferedRequest = null;
   this.lastBufferedRequest = null;
 
@@ -498,22 +502,41 @@ function onwrite(stream, er) {
     }
 
     if (sync) {
-      process.nextTick(afterWrite, stream, state, cb);
+      // It is a common case that the callback passed to .write() is always
+      // the same. In that case, we do not schedule a new nextTick(), but rather
+      // just increase a counter, to improve performance and avoid memory
+      // allocations.
+      if (state.afterWriteTickInfo !== null &&
+          state.afterWriteTickInfo.cb === cb) {
+        state.afterWriteTickInfo.count++;
+      } else {
+        state.afterWriteTickInfo = { count: 1, cb, stream, state };
+        process.nextTick(afterWriteTick, state.afterWriteTickInfo);
+      }
     } else {
-      afterWrite(stream, state, cb);
+      afterWrite(stream, state, 1, cb);
     }
   }
 }
 
-function afterWrite(stream, state, cb) {
+function afterWriteTick({ stream, state, count, cb }) {
+  state.afterWriteTickInfo = null;
+  return afterWrite(stream, state, count, cb);
+}
+
+function afterWrite(stream, state, count, cb) {
   const needDrain = !state.ending && !stream.destroyed && state.length === 0 &&
     state.needDrain;
   if (needDrain) {
     state.needDrain = false;
     stream.emit('drain');
   }
-  state.pendingcb--;
-  cb();
+
+  while (count-- > 0) {
+    state.pendingcb--;
+    cb();
+  }
+
   finishMaybe(stream, state);
 }