diff --git a/benchmark/streams/writable-manywrites.js b/benchmark/streams/writable-manywrites.js index 6fcb07e849d615..0ed38d0357a438 100644 --- a/benchmark/streams/writable-manywrites.js +++ b/benchmark/streams/writable-manywrites.js @@ -4,14 +4,19 @@ const common = require('../common'); const Writable = require('stream').Writable; const bench = common.createBenchmark(main, { - n: [2e6] + n: [2e6], + sync: ['yes', 'no'] }); -function main({ n }) { +function main({ n, sync }) { const b = Buffer.allocUnsafe(1024); const s = new Writable(); + sync = sync === 'yes'; s._write = function(chunk, encoding, cb) { - cb(); + if (sync) + cb(); + else + process.nextTick(cb); }; bench.start(); diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js index 5be90bb81591a2..9bea209e257d2a 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); } diff --git a/test/parallel/test-stream-writable-samecb-singletick.js b/test/parallel/test-stream-writable-samecb-singletick.js new file mode 100644 index 00000000000000..e7dfa648797821 --- /dev/null +++ b/test/parallel/test-stream-writable-samecb-singletick.js @@ -0,0 +1,30 @@ +'use strict'; +const common = require('../common'); +const { Console } = require('console'); +const { Writable } = require('stream'); +const async_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 + +const checkTickCreated = common.mustCall(); + +async_hooks.createHook({ + init(id, type, triggerId, resoure) { + if (type === 'TickObject') checkTickCreated(); + } +}).enable(); + +const console = new Console(new Writable({ + write: common.mustCall((chunk, encoding, cb) => { + cb(); + }, 100) +})); + +for (let i = 0; i < 100; i++) + console.log(i);