stream: avoid drain for sync streams

Previously a sync writable receiving chunks larger than highwatermark would unecessarily ping pong needDrain. PR-URL: #32887 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: James M Snell <[email protected]>

Author: ronag

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]
 });
 
-functionmain({ n, sync, writev, callback }){
-constb=Buffer.allocUnsafe(1024);
+functionmain({ n, sync, writev, callback, len}){
+constb=Buffer.allocUnsafe(len);
 consts=newWritable();
 sync=sync==='yes';
 

lib/_stream_writable.js

@@ -337,11 +337,6 @@ function writeOrBuffer(stream, state, chunk, encoding, callback){
 
 state.length+=len;
 
-constret=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){
 state.buffered.push({ chunk, encoding, callback });
 if(state.allBuffers&&encoding!=='buffer'){
@@ -359,6 +354,12 @@ function writeOrBuffer(stream, state, chunk, encoding, callback){
 state.sync=false;
 }
 
+constret=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.
 returnret&&!state.errored&&!state.destroyed;

test/parallel/test-stream-big-packet.js

@@ -36,7 +36,11 @@ class TestStream extends stream.Transform{
 }
 }
 
-consts1=newstream.PassThrough();
+consts1=newstream.Transform({
+transform(chunk,encoding,cb){
+process.nextTick(cb,null,chunk);
+}
+});
 consts2=newstream.PassThrough();
 consts3=newTestStream();
 s1.pipe(s3);

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);
 }
 });
 

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)
 });
 

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

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);

test/parallel/test-stream2-finish-pipe.js

@@ -30,7 +30,7 @@ r._read = function(size){
 
 constw=newstream.Writable();
 w._write=function(data,encoding,cb){
-cb(null);
+process.nextTick(cb,null);
 };
 
 r.pipe(w);