From 102c2f079d978d5ebd5d63777407394e64aada57 Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Thu, 4 Jan 2018 18:06:56 +0100 Subject: [PATCH] stream: always defer 'readable' with nextTick Fixes: https://github.com/nodejs/node/issues/3203 --- lib/_stream_readable.js | 10 ++-- test/parallel/test-net-end-close.js | 12 ++-- ...tream-pipe-await-drain-push-while-write.js | 32 ++++------- .../test-stream-readable-emittedReadable.js | 17 +++--- .../test-stream-readable-needReadable.js | 23 ++++---- ...stream-readable-object-multi-push-async.js | 56 +++++++++++++++++++ ...est-stream-readable-reading-readingMore.js | 12 ++-- test/parallel/test-stream2-transform.js | 13 +++-- 8 files changed, 118 insertions(+), 57 deletions(-) create mode 100644 test/parallel/test-stream-readable-object-multi-push-async.js diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js index 21598efa65f254..12fd15b0e6eb66 100644 --- a/lib/_stream_readable.js +++ b/lib/_stream_readable.js @@ -508,16 +508,15 @@ function emitReadable(stream) { if (!state.emittedReadable) { debug('emitReadable', state.flowing); state.emittedReadable = true; - if (state.sync) - process.nextTick(emitReadable_, stream); - else - emitReadable_(stream); + process.nextTick(emitReadable_, stream); } } function emitReadable_(stream) { + var state = stream._readableState; debug('emit readable'); stream.emit('readable'); + state.needReadable = !state.flowing && !state.ended; flow(stream); } @@ -644,6 +643,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) { debug('ondata'); increasedAwaitDrain = false; var ret = dest.write(chunk); + debug('dest.write', ret); if (false === ret && !increasedAwaitDrain) { // If the user unpiped during `dest.write()`, it is possible // to get stuck in a permanently paused state if that write @@ -824,8 +824,8 @@ function resume(stream, state) { } function resume_(stream, state) { + debug('resume', state.reading); if (!state.reading) { - debug('resume read 0'); stream.read(0); } diff --git a/test/parallel/test-net-end-close.js b/test/parallel/test-net-end-close.js index 44a539a3e800a6..31c150e09c09af 100644 --- a/test/parallel/test-net-end-close.js +++ b/test/parallel/test-net-end-close.js @@ -8,9 +8,9 @@ const uv = process.binding('uv'); const s = new net.Socket({ handle: { readStart: function() { - process.nextTick(() => this.onread(uv.UV_EOF, null)); + setImmediate(() => this.onread(uv.UV_EOF, null)); }, - close: (cb) => process.nextTick(cb) + close: (cb) => setImmediate(cb) }, writable: false }); @@ -18,8 +18,12 @@ assert.strictEqual(s, s.resume()); const events = []; -s.on('end', () => events.push('end')); -s.on('close', () => events.push('close')); +s.on('end', () => { + events.push('end'); +}); +s.on('close', () => { + events.push('close'); +}); process.on('exit', () => { assert.deepStrictEqual(events, [ 'end', 'close' ]); diff --git a/test/parallel/test-stream-pipe-await-drain-push-while-write.js b/test/parallel/test-stream-pipe-await-drain-push-while-write.js index adefad70adc20c..263e6b6801f68b 100644 --- a/test/parallel/test-stream-pipe-await-drain-push-while-write.js +++ b/test/parallel/test-stream-pipe-await-drain-push-while-write.js @@ -3,32 +3,24 @@ const common = require('../common'); const stream = require('stream'); const assert = require('assert'); -const awaitDrainStates = [ - 1, // after first chunk before callback - 1, // after second chunk before callback - 0 // resolving chunk pushed after first chunk, awaitDrain is decreased -]; - -// A writable stream which pushes data onto the stream which pipes into it, -// but only the first time it's written to. Since it's not paused at this time, -// a second write will occur. If the pipe increases awaitDrain twice, we'll -// never get subsequent chunks because 'drain' is only emitted once. const writable = new stream.Writable({ write: common.mustCall(function(chunk, encoding, cb) { - if (chunk.length === 32 * 1024) { // first chunk - const beforePush = readable._readableState.awaitDrain; - readable.push(Buffer.alloc(34 * 1024)); // above hwm - // We should check if awaitDrain counter is increased. - const afterPush = readable._readableState.awaitDrain; - assert.strictEqual(afterPush - beforePush, 1, - 'Counter is not increased for awaitDrain'); - } - assert.strictEqual( - awaitDrainStates.shift(), readable._readableState.awaitDrain, + 0, 'State variable awaitDrain is not correct.' ); + + if (chunk.length === 32 * 1024) { // first chunk + readable.push(Buffer.alloc(34 * 1024)); // above hwm + // We should check if awaitDrain counter is increased in the next + // tick, because awaitDrain is incremented after this method finished + process.nextTick(() => { + assert.strictEqual(readable._readableState.awaitDrain, 1, + 'Counter is not increased for awaitDrain'); + }); + } + cb(); }, 3) }); diff --git a/test/parallel/test-stream-readable-emittedReadable.js b/test/parallel/test-stream-readable-emittedReadable.js index 65b6b5b15a5895..5b9affc59fbbf2 100644 --- a/test/parallel/test-stream-readable-emittedReadable.js +++ b/test/parallel/test-stream-readable-emittedReadable.js @@ -10,30 +10,33 @@ const readable = new Readable({ // Initialized to false. assert.strictEqual(readable._readableState.emittedReadable, false); +const expected = [Buffer.from('foobar'), Buffer.from('quo'), null]; readable.on('readable', common.mustCall(() => { // emittedReadable should be true when the readable event is emitted assert.strictEqual(readable._readableState.emittedReadable, true); - readable.read(); + assert.deepStrictEqual(readable.read(), expected.shift()); // emittedReadable is reset to false during read() assert.strictEqual(readable._readableState.emittedReadable, false); -}, 4)); +}, 3)); // When the first readable listener is just attached, // emittedReadable should be false assert.strictEqual(readable._readableState.emittedReadable, false); -// Each one of these should trigger a readable event. +// These trigger a single 'readable', as things are batched up process.nextTick(common.mustCall(() => { readable.push('foo'); })); process.nextTick(common.mustCall(() => { readable.push('bar'); })); -process.nextTick(common.mustCall(() => { + +// these triggers two readable events +setImmediate(common.mustCall(() => { readable.push('quo'); -})); -process.nextTick(common.mustCall(() => { - readable.push(null); + process.nextTick(common.mustCall(() => { + readable.push(null); + })); })); const noRead = new Readable({ diff --git a/test/parallel/test-stream-readable-needReadable.js b/test/parallel/test-stream-readable-needReadable.js index be397dc5dc5f74..7058e123f07823 100644 --- a/test/parallel/test-stream-readable-needReadable.js +++ b/test/parallel/test-stream-readable-needReadable.js @@ -38,7 +38,7 @@ asyncReadable.on('readable', common.mustCall(() => { // then we need to notify the reader on future changes. assert.strictEqual(asyncReadable._readableState.needReadable, true); } -}, 3)); +}, 2)); process.nextTick(common.mustCall(() => { asyncReadable.push('foooo'); @@ -46,8 +46,9 @@ process.nextTick(common.mustCall(() => { process.nextTick(common.mustCall(() => { asyncReadable.push('bar'); })); -process.nextTick(common.mustCall(() => { +setImmediate(common.mustCall(() => { asyncReadable.push(null); + assert.strictEqual(asyncReadable._readableState.needReadable, false); })); const flowing = new Readable({ @@ -84,13 +85,13 @@ slowProducer.on('readable', common.mustCall(() => { process.nextTick(common.mustCall(() => { slowProducer.push('foo'); -})); -process.nextTick(common.mustCall(() => { - slowProducer.push('foo'); -})); -process.nextTick(common.mustCall(() => { - slowProducer.push('foo'); -})); -process.nextTick(common.mustCall(() => { - slowProducer.push(null); + process.nextTick(common.mustCall(() => { + slowProducer.push('foo'); + process.nextTick(common.mustCall(() => { + slowProducer.push('foo'); + process.nextTick(common.mustCall(() => { + slowProducer.push(null); + })); + })); + })); })); diff --git a/test/parallel/test-stream-readable-object-multi-push-async.js b/test/parallel/test-stream-readable-object-multi-push-async.js new file mode 100644 index 00000000000000..1a684add2ef590 --- /dev/null +++ b/test/parallel/test-stream-readable-object-multi-push-async.js @@ -0,0 +1,56 @@ +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { Readable } = require('stream'); + +const MAX = 42; +const BATCH = 10; + +const readable = new Readable({ + objectMode: true, + read: common.mustCall(function() { + console.log('>> READ'); + fetchData((err, data) => { + if (err) { + this.destroy(err); + return; + } + + if (data.length === 0) { + console.log('pushing null'); + this.push(null); + return; + } + + console.log('pushing'); + data.forEach((d) => this.push(d)); + }); + }, Math.floor(MAX / BATCH) + 2) +}); + +let i = 0; +function fetchData(cb) { + if (i > MAX) { + setTimeout(cb, 10, null, []); + } else { + const array = []; + const max = i + BATCH; + for (; i < max; i++) { + array.push(i); + } + setTimeout(cb, 10, null, array); + } +} + +readable.on('readable', () => { + let data; + console.log('readable emitted'); + while (data = readable.read()) { + console.log(data); + } +}); + +readable.on('end', () => { + assert.strictEqual(i, (Math.floor(MAX / BATCH) + 1) * BATCH); +}); diff --git a/test/parallel/test-stream-readable-reading-readingMore.js b/test/parallel/test-stream-readable-reading-readingMore.js index e31d2dd921ce5b..638ea48efec4f6 100644 --- a/test/parallel/test-stream-readable-reading-readingMore.js +++ b/test/parallel/test-stream-readable-reading-readingMore.js @@ -29,15 +29,19 @@ function onStreamEnd() { assert.strictEqual(state.reading, false); } +const expected = [ + true, // stream is not ended + false // stream is ended +]; + readable.on('readable', common.mustCall(() => { - // 'readable' always gets called before 'end' - // since 'end' hasn't been emitted, more data could be incoming - assert.strictEqual(state.readingMore, true); + assert.strictEqual(state.readingMore, expected.shift()); // if the stream has ended, we shouldn't be reading assert.strictEqual(state.ended, !state.reading); - if (readable.read() === null) // reached end of stream + const data = readable.read(); + if (data === null) // reached end of stream process.nextTick(common.mustCall(onStreamEnd, 1)); }, 2)); diff --git a/test/parallel/test-stream2-transform.js b/test/parallel/test-stream2-transform.js index d0859265428b81..7a5aa22d55879e 100644 --- a/test/parallel/test-stream2-transform.js +++ b/test/parallel/test-stream2-transform.js @@ -306,25 +306,26 @@ const Transform = require('_stream_transform'); pt.write(Buffer.from('foog')); pt.write(Buffer.from('bark')); - assert.strictEqual(emits, 1); + assert.strictEqual(emits, 0); assert.strictEqual(pt.read(5).toString(), 'foogb'); assert.strictEqual(String(pt.read(5)), 'null'); + assert.strictEqual(emits, 0); pt.write(Buffer.from('bazy')); pt.write(Buffer.from('kuel')); - assert.strictEqual(emits, 2); + assert.strictEqual(emits, 0); assert.strictEqual(pt.read(5).toString(), 'arkba'); assert.strictEqual(pt.read(5).toString(), 'zykue'); assert.strictEqual(pt.read(5), null); pt.end(); - assert.strictEqual(emits, 3); + assert.strictEqual(emits, 0); assert.strictEqual(pt.read(5).toString(), 'l'); assert.strictEqual(pt.read(5), null); - assert.strictEqual(emits, 3); + assert.strictEqual(emits, 0); } { @@ -338,7 +339,7 @@ const Transform = require('_stream_transform'); pt.write(Buffer.from('foog')); pt.write(Buffer.from('bark')); - assert.strictEqual(emits, 1); + assert.strictEqual(emits, 0); assert.strictEqual(pt.read(5).toString(), 'foogb'); assert.strictEqual(pt.read(5), null); @@ -352,7 +353,7 @@ const Transform = require('_stream_transform'); pt.once('readable', common.mustCall(function() { assert.strictEqual(pt.read(5).toString(), 'l'); assert.strictEqual(pt.read(5), null); - assert.strictEqual(emits, 4); + assert.strictEqual(emits, 3); })); pt.end(); }));