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:
nodejs/node@89aeab9
(nodejs/node#4354).

Fixes: nodejs/node#6154
PR-URL: nodejs/node#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 = new CorkedRequest(this);
-  this.corkedRequestsFree.next = new CorkedRequest(this);
 }
 
 WritableState.prototype.getBuffer = function writableStateGetBuffer() {
@@ -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 = new CorkedRequest(state);
+    }
   } else {
     // Slow case, write chunks one-by-one
     while (entry) {

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

@@ -0,0 +1,41 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const net = require('net');
+
+const server = net.createServer(handle);
+
+const N = 100;
+const buf = Buffer('aa');
+
+server.listen(common.PORT, function() {
+  const conn = net.connect(common.PORT);
+
+  conn.on('connect', () => {
+    let res = true;
+    let i = 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);
+});
+
+function handle(socket) {
+  socket.resume();
+
+  socket.on('error', function(err) {
+    socket.destroy();
+  }).on('close', function() {
+    server.close();
+  });
+}