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: #18013 Refs: #18367 PR-URL: #30710 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: James M Snell <[email protected]>

Author: addaleax

benchmark/streams/writable-manywrites.js

@@ -4,14 +4,19 @@ const common = require('../common');
 constWritable=require('stream').Writable;
 
 constbench=common.createBenchmark(main,{
-n: [2e6]
+n: [2e6],
+sync: ['yes','no']
 });
 
-functionmain({ n }){
+functionmain({ n, sync}){
 constb=Buffer.allocUnsafe(1024);
 consts=newWritable();
+sync=sync==='yes';
 s._write=function(chunk,encoding,cb){
-cb();
+if(sync)
+cb();
+else
+process.nextTick(cb);
 };
 
 bench.start();

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);
 }
 }
 }
 
-functionafterWrite(stream,state,cb){
+functionafterWriteTick({ stream, state, count, cb }){
+state.afterWriteTickInfo=null;
+returnafterWrite(stream,state,count,cb);
+}
+
+functionafterWrite(stream,state,count,cb){
 constneedDrain=!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);
 }
 

test/parallel/test-stream-writable-samecb-singletick.js

@@ -0,0 +1,30 @@
+'use strict';
+constcommon=require('../common');
+const{ Console }=require('console');
+const{ Writable }=require('stream');
+constasync_hooks=require('async_hooks');
+
+// Make sure that repeated calls to console.log(), and by extension
+// stream.write() for the underlying stream, allocate exactly 1 tick object.
+// At the time of writing, that is enough to ensure a flat memory profile
+// from repeated console.log() calls, rather than having callbacks pile up
+// over time, assuming that data can be written synchronously.
+// Refs: https://github.com/nodejs/node/issues/18013
+// Refs: https://github.com/nodejs/node/issues/18367
+
+constcheckTickCreated=common.mustCall();
+
+async_hooks.createHook({
+init(id,type,triggerId,resoure){
+if(type==='TickObject')checkTickCreated();
+}
+}).enable();
+
+constconsole=newConsole(newWritable({
+write: common.mustCall((chunk,encoding,cb)=>{
+cb();
+},100)
+}));
+
+for(leti=0;i<100;i++)
+console.log(i);