stream: use lazy registration for drain for fast destinations
PR-URL: https://github.com/nodejs/node/pull/29095 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Rich Trott <rtrott@gmail.com>
This commit is contained in:
parent
4111c57f7c
commit
7195cd6fb3
@ -676,12 +676,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|||||||
dest.end();
|
dest.end();
|
||||||
}
|
}
|
||||||
|
|
||||||
// When the dest drains, it reduces the awaitDrain counter
|
let ondrain;
|
||||||
// on the source. This would be more elegant with a .once()
|
|
||||||
// handler in flow(), but adding and removing repeatedly is
|
|
||||||
// too slow.
|
|
||||||
const ondrain = pipeOnDrain(src);
|
|
||||||
dest.on('drain', ondrain);
|
|
||||||
|
|
||||||
var cleanedUp = false;
|
var cleanedUp = false;
|
||||||
function cleanup() {
|
function cleanup() {
|
||||||
@ -689,7 +684,9 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|||||||
// Cleanup event handlers once the pipe is broken
|
// Cleanup event handlers once the pipe is broken
|
||||||
dest.removeListener('close', onclose);
|
dest.removeListener('close', onclose);
|
||||||
dest.removeListener('finish', onfinish);
|
dest.removeListener('finish', onfinish);
|
||||||
dest.removeListener('drain', ondrain);
|
if (ondrain) {
|
||||||
|
dest.removeListener('drain', ondrain);
|
||||||
|
}
|
||||||
dest.removeListener('error', onerror);
|
dest.removeListener('error', onerror);
|
||||||
dest.removeListener('unpipe', onunpipe);
|
dest.removeListener('unpipe', onunpipe);
|
||||||
src.removeListener('end', onend);
|
src.removeListener('end', onend);
|
||||||
@ -703,7 +700,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|||||||
// flowing again.
|
// flowing again.
|
||||||
// So, if this is awaiting a drain, then we just call it now.
|
// So, if this is awaiting a drain, then we just call it now.
|
||||||
// If we don't know, then assume that we are waiting for one.
|
// If we don't know, then assume that we are waiting for one.
|
||||||
if (state.awaitDrain &&
|
if (ondrain && state.awaitDrain &&
|
||||||
(!dest._writableState || dest._writableState.needDrain))
|
(!dest._writableState || dest._writableState.needDrain))
|
||||||
ondrain();
|
ondrain();
|
||||||
}
|
}
|
||||||
@ -722,6 +719,14 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
|
|||||||
debug('false write response, pause', state.awaitDrain);
|
debug('false write response, pause', state.awaitDrain);
|
||||||
state.awaitDrain++;
|
state.awaitDrain++;
|
||||||
}
|
}
|
||||||
|
if (!ondrain) {
|
||||||
|
// When the dest drains, it reduces the awaitDrain counter
|
||||||
|
// on the source. This would be more elegant with a .once()
|
||||||
|
// handler in flow(), but adding and removing repeatedly is
|
||||||
|
// too slow.
|
||||||
|
ondrain = pipeOnDrain(src);
|
||||||
|
dest.on('drain', ondrain);
|
||||||
|
}
|
||||||
src.pause();
|
src.pause();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -1,5 +1,6 @@
|
|||||||
'use strict';
|
'use strict';
|
||||||
const common = require('../common');
|
const common = require('../common');
|
||||||
|
const assert = require('assert');
|
||||||
const { Readable, Writable, PassThrough } = require('stream');
|
const { Readable, Writable, PassThrough } = require('stream');
|
||||||
|
|
||||||
{
|
{
|
||||||
@ -65,3 +66,25 @@ const { Readable, Writable, PassThrough } = require('stream');
|
|||||||
wrapper.resume();
|
wrapper.resume();
|
||||||
wrapper.on('end', common.mustCall());
|
wrapper.on('end', common.mustCall());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
{
|
||||||
|
// Only register drain if there is backpressure.
|
||||||
|
const rs = new Readable({ read() {} });
|
||||||
|
|
||||||
|
const pt = rs
|
||||||
|
.pipe(new PassThrough({ objectMode: true, highWaterMark: 2 }));
|
||||||
|
assert.strictEqual(pt.listenerCount('drain'), 0);
|
||||||
|
pt.on('finish', () => {
|
||||||
|
assert.strictEqual(pt.listenerCount('drain'), 0);
|
||||||
|
});
|
||||||
|
|
||||||
|
rs.push('asd');
|
||||||
|
assert.strictEqual(pt.listenerCount('drain'), 0);
|
||||||
|
|
||||||
|
process.nextTick(() => {
|
||||||
|
rs.push('asd');
|
||||||
|
assert.strictEqual(pt.listenerCount('drain'), 0);
|
||||||
|
rs.push(null);
|
||||||
|
assert.strictEqual(pt.listenerCount('drain'), 0);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
@ -52,11 +52,4 @@ function drain() {
|
|||||||
|
|
||||||
w.end = common.mustCall();
|
w.end = common.mustCall();
|
||||||
|
|
||||||
// Just for kicks, let's mess with the drain count.
|
|
||||||
// This verifies that even if it gets negative in the
|
|
||||||
// pipe() cleanup function, we'll still function properly.
|
|
||||||
r.on('readable', function() {
|
|
||||||
w.emit('drain');
|
|
||||||
});
|
|
||||||
|
|
||||||
r.pipe(w);
|
r.pipe(w);
|
||||||
|
Loading…
x
Reference in New Issue
Block a user