stream: avoid drain for sync streams

ronag · ronag · commit abc1d0b642c2 · 2020-04-16T21:59:10.000+02:00
Previously a sync writable receiving chunks
larger than highwatermark would unecessarily
ping pong needDrain.
diff --git a/benchmark/streams/writable-manywrites.js b/benchmark/streams/writable-manywrites.js
@@ -7,11 +7,12 @@ const bench = common.createBenchmark(main, {
   n: [2e6],
   sync: ['yes', 'no'],
   writev: ['yes', 'no'],
-  callback: ['yes', 'no']
+  callback: ['yes', 'no'],
+  len: [1024, 32 * 1024]
 });
 
-function main({ n, sync, writev, callback }) {
-  const b = Buffer.allocUnsafe(1024);
+function main({ n, sync, writev, callback, len }) {
+  const b = Buffer.allocUnsafe(len);
   const s = new Writable();
   sync = sync === 'yes';
 
diff --git a/lib/_stream_writable.js b/lib/_stream_writable.js
@@ -344,11 +344,6 @@ function writeOrBuffer(stream, state, chunk, encoding, cb) {
 
   state.length += len;
 
-  const ret = state.length < state.highWaterMark;
-  // We must ensure that previous needDrain will not be reset to false.
-  if (!ret)
-    state.needDrain = true;
-
   if (state.writing || state.corked || state.errored) {
     const last = state.lastBufferedRequest;
     state.lastBufferedRequest = {
@@ -367,6 +362,12 @@ function writeOrBuffer(stream, state, chunk, encoding, cb) {
     doWrite(stream, state, false, len, chunk, encoding, cb);
   }
 
+  const ret = state.length < state.highWaterMark;
+
+  // We must ensure that previous needDrain will not be reset to false.
+  if (!ret)
+    state.needDrain = true;
+
   // Return false if errored or destroyed in order to break
   // any synchronous while(stream.write(data)) loops.
   return ret && !state.errored && !state.destroyed;
diff --git a/test/parallel/test-stream-big-packet.js b/test/parallel/test-stream-big-packet.js
@@ -36,7 +36,11 @@ class TestStream extends stream.Transform {
   }
 }
 
-const s1 = new stream.PassThrough();
+const s1 = new stream.Transform({
+  transform(chunk, encoding, cb) {
+    process.nextTick(cb, null, chunk);
+  }
+});
 const s2 = new stream.PassThrough();
 const s3 = new TestStream();
 s1.pipe(s3);
diff --git a/test/parallel/test-stream-catch-rejections.js b/test/parallel/test-stream-catch-rejections.js
@@ -30,7 +30,7 @@ const assert = require('assert');
     captureRejections: true,
     highWaterMark: 1,
     write(chunk, enc, cb) {
-      cb();
+      process.nextTick(cb);
     }
   });
 
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
@@ -19,7 +19,7 @@ const writable = new stream.Writable({
       });
     }
 
-    cb();
+    process.nextTick(cb);
   }, 3)
 });
 
diff --git a/test/parallel/test-stream-pipe-await-drain.js b/test/parallel/test-stream-pipe-await-drain.js
@@ -19,7 +19,7 @@ reader._read = () => {};
 
 writer1._write = common.mustCall(function(chunk, encoding, cb) {
   this.emit('chunk-received');
-  cb();
+  process.nextTick(cb);
 }, 1);
 
 writer1.once('chunk-received', () => {
@@ -42,7 +42,7 @@ writer2._write = common.mustCall((chunk, encoding, cb) => {
     reader._readableState.awaitDrainWriters.size,
     1,
     'awaitDrain should be 1 after first push, actual is ' +
-    reader._readableState.awaitDrainWriters
+    reader._readableState.awaitDrainWriters.size
   );
   // Not calling cb here to "simulate" slow stream.
   // This should be called exactly once, since the first .write() call
@@ -54,7 +54,7 @@ writer3._write = common.mustCall((chunk, encoding, cb) => {
     reader._readableState.awaitDrainWriters.size,
     2,
     'awaitDrain should be 2 after second push, actual is ' +
-    reader._readableState.awaitDrainWriters
+    reader._readableState.awaitDrainWriters.size
   );
   // Not calling cb here to "simulate" slow stream.
   // This should be called exactly once, since the first .write() call
diff --git a/test/parallel/test-stream-writable-needdrain-state.js b/test/parallel/test-stream-writable-needdrain-state.js
@@ -10,8 +10,10 @@ const transform = new stream.Transform({
 });
 
 function _transform(chunk, encoding, cb) {
-  assert.strictEqual(transform._writableState.needDrain, true);
-  cb();
+  process.nextTick(() => {
+    assert.strictEqual(transform._writableState.needDrain, true);
+    cb();
+  });
 }
 
 assert.strictEqual(transform._writableState.needDrain, false);

Original file line number	Diff line number	Diff line change
`@@ -36,7 +36,11 @@ class TestStream extends stream.Transform {`
`36`	`36`	`}`
`37`	`37`	`}`
`38`	`38`
`39`		`-const s1 = new stream.PassThrough();`
	`39`	`+const s1 = new stream.Transform({`
	`40`	`+ transform(chunk, encoding, cb) {`
	`41`	`+ process.nextTick(cb, null, chunk);`
	`42`	`+ }`
	`43`	`+});`
`40`	`44`	`const s2 = new stream.PassThrough();`
`41`	`45`	`const s3 = new TestStream();`
`42`	`46`	`s1.pipe(s3);`
Original file line number	Diff line number	Diff line change
`@@ -30,7 +30,7 @@ const assert = require('assert');`
`30`	`30`	`captureRejections: true,`
`31`	`31`	`highWaterMark: 1,`
`32`	`32`	`write(chunk, enc, cb) {`
`33`		`- cb();`
	`33`	`+ process.nextTick(cb);`
`34`	`34`	`}`
`35`	`35`	`});`
`36`	`36`
Original file line number	Diff line number	Diff line change
`@@ -19,7 +19,7 @@ const writable = new stream.Writable({`
`19`	`19`	`});`
`20`	`20`	`}`
`21`	`21`
`22`		`- cb();`
	`22`	`+ process.nextTick(cb);`
`23`	`23`	`}, 3)`
`24`	`24`	`});`
`25`	`25`