streams: 5% throughput gain when sending small chunks

Improves the performance when moving small buffers by 5%, and it adds a benchmark to avoid regression in that area. In all other cases it is equally performant to current master. Full performance results available at: https://gist.github.com/mcollina/717c35ad07d15710b6b9. PR-URL: #4354 Reviewed-By: James M Snell <[email protected]>

Author: mcollina

benchmark/net/net-c2s-cork.js

@@ -0,0 +1,96 @@
+// test the speed of .pipe() with sockets
+
+varcommon=require('../common.js');
+varPORT=common.PORT;
+
+varbench=common.createBenchmark(main,{
+len: [4,8,16,32,64,128,512,1024],
+type: ['buf'],
+dur: [5],
+});
+
+vardur;
+varlen;
+vartype;
+varchunk;
+varencoding;
+
+functionmain(conf){
+dur=+conf.dur;
+len=+conf.len;
+type=conf.type;
+
+switch(type){
+case'buf':
+chunk=newBuffer(len);
+chunk.fill('x');
+break;
+case'utf':
+encoding='utf8';
+chunk=newArray(len/2+1).join('ü');
+break;
+case'asc':
+encoding='ascii';
+chunk=newArray(len+1).join('x');
+break;
+default:
+thrownewError('invalid type: '+type);
+break;
+}
+
+server();
+}
+
+varnet=require('net');
+
+functionWriter(){
+this.received=0;
+this.writable=true;
+}
+
+Writer.prototype.write=function(chunk,encoding,cb){
+this.received+=chunk.length;
+
+if(typeofencoding==='function')
+encoding();
+elseif(typeofcb==='function')
+cb();
+
+returntrue;
+};
+
+// doesn't matter, never emits anything.
+Writer.prototype.on=function(){};
+Writer.prototype.once=function(){};
+Writer.prototype.emit=function(){};
+
+functionserver(){
+varwriter=newWriter();
+
+// the actual benchmark.
+varserver=net.createServer(function(socket){
+socket.pipe(writer);
+});
+
+server.listen(PORT,function(){
+varsocket=net.connect(PORT);
+socket.on('connect',function(){
+bench.start();
+
+socket.on('drain',send)
+send()
+
+setTimeout(function(){
+varbytes=writer.received;
+vargbits=(bytes*8)/(1024*1024*1024);
+bench.end(gbits);
+},dur*1000);
+
+functionsend(){
+socket.cork();
+while(socket.write(chunk,encoding)){}
+socket.uncork();
+}
+});
+});
+}

lib/_stream_writable.js

@@ -108,6 +108,14 @@ function WritableState(options, stream){
 
 // True if the error was already emitted and should not be thrown again
 this.errorEmitted=false;
+
+// count buffered requests
+this.bufferedRequestCount=0;
+
+// create the two objects needed to store the corked requests
+// they are not a linked list, as no new elements are inserted in there
+this.corkedRequestsFree=newCorkedRequest(this);
+this.corkedRequestsFree.next=newCorkedRequest(this);
 }
 
 WritableState.prototype.getBuffer=functionwritableStateGetBuffer(){
@@ -274,6 +282,7 @@ function writeOrBuffer(stream, state, chunk, encoding, cb){
 }else{
 state.bufferedRequest=state.lastBufferedRequest;
 }
+state.bufferedRequestCount+=1;
 }else{
 doWrite(stream,state,false,len,chunk,encoding,cb);
 }
@@ -357,34 +366,33 @@ function onwriteDrain(stream, state){
 }
 }
 
-
 // if there's something in the buffer waiting, then process it
 functionclearBuffer(stream,state){
 state.bufferProcessing=true;
 varentry=state.bufferedRequest;
 
 if(stream._writev&&entry&&entry.next){
 // Fast case, write everything using _writev()
-varbuffer=[];
-varcbs=[];
+varl=state.bufferedRequestCount;
+varbuffer=newArray(l);
+varholder=state.corkedRequestsFree;
+holder.entry=entry;
+
+varcount=0;
 while(entry){
-cbs.push(entry.callback);
-buffer.push(entry);
+buffer[count]=entry;
 entry=entry.next;
+count+=1;
 }
 
-// count the one we are adding, as well.
-// TODO(isaacs) clean this up
+doWrite(stream,state,true,state.length,buffer,'',holder.finish);
+
+// doWrite is always async, defer these to save a bit of time
+// as the hot path ends with doWrite
 state.pendingcb++;
 state.lastBufferedRequest=null;
-doWrite(stream,state,true,state.length,buffer,'',function(err){
-for(vari=0;i<cbs.length;i++){
-state.pendingcb--;
-cbs[i](err);
-}
-});
-
-// Clear buffer
+state.corkedRequestsFree=holder.next;
+holder.next=null;
 }else{
 // Slow case, write chunks one-by-one
 while(entry){
@@ -407,6 +415,8 @@ function clearBuffer(stream, state){
 if(entry===null)
 state.lastBufferedRequest=null;
 }
+
+state.bufferedRequestCount=0;
 state.bufferedRequest=entry;
 state.bufferProcessing=false;
 }
@@ -485,3 +495,26 @@ function endWritable(stream, state, cb){
 state.ended=true;
 stream.writable=false;
 }
+
+// It seems a linked list but it is not
+// there will be only 2 of these for each stream
+functionCorkedRequest(state){
+this.next=null;
+this.entry=null;
+
+this.finish=(err)=>{
+varentry=this.entry;
+this.entry=null;
+while(entry){
+varcb=entry.callback;
+state.pendingcb--;
+cb(err);
+entry=entry.next;
+}
+if(state.corkedRequestsFree){
+state.corkedRequestsFree.next=this;
+}else{
+state.corkedRequestsFree=this;
+}
+};
+}