streams: support unlimited synchronous cork/uncork cycles

net streams can request multiple chunks to be written in a synchronous fashion. If this is combined with cork/uncork, en error is currently thrown because of a regression introduced in: 89aeab9 (#4354). Fixes: #6154 PR-URL: #6164 Reviewed-By: Benjamin Gruenbaum <[email protected]> Reviewed-By: Mathias Buus <[email protected]> Reviewed-By: James M Snell <[email protected]>

Author: mcollina

lib/_stream_writable.js

@@ -112,10 +112,9 @@ function WritableState(options, stream){
 // 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
+// allocate the first CorkedRequest, there is always
+// one allocated and free to use, and we maintain at most two
 this.corkedRequestsFree=newCorkedRequest(this);
-this.corkedRequestsFree.next=newCorkedRequest(this);
 }
 
 WritableState.prototype.getBuffer=functionwritableStateGetBuffer(){
@@ -387,12 +386,16 @@ function clearBuffer(stream, state){
 
 doWrite(stream,state,true,state.length,buffer,'',holder.finish);
 
-// doWrite is always async, defer these to save a bit of time
+// doWrite is almost always async, defer these to save a bit of time
 // as the hot path ends with doWrite
 state.pendingcb++;
 state.lastBufferedRequest=null;
-state.corkedRequestsFree=holder.next;
-holder.next=null;
+if(holder.next){
+state.corkedRequestsFree=holder.next;
+holder.next=null;
+}else{
+state.corkedRequestsFree=newCorkedRequest(state);
+}
 }else{
 // Slow case, write chunks one-by-one
 while(entry){

test/parallel/test-net-sync-cork.js

@@ -0,0 +1,41 @@
+'use strict';
+
+constcommon=require('../common');
+constassert=require('assert');
+constnet=require('net');
+
+constserver=net.createServer(handle);
+
+constN=100;
+constbuf=Buffer('aa');
+
+server.listen(common.PORT,function(){
+constconn=net.connect(common.PORT);
+
+conn.on('connect',()=>{
+letres=true;
+leti=0;
+for(;i<N&&res;i++){
+conn.cork();
+conn.write(buf);
+res=conn.write(buf);
+conn.uncork();
+}
+assert.equal(i,N);
+conn.end();
+});
+});
+
+process.on('exit',function(){
+assert.equal(server.connections,0);
+});
+
+functionhandle(socket){
+socket.resume();
+
+socket.on('error',function(err){
+socket.destroy();
+}).on('close',function(){
+server.close();
+});
+}