streams: 5% throughput gain when sending small chunks

benchmark/net/net-c2s-cork.js

@@ -0,0 +1,96 @@
+// test the speed of .pipe() with sockets
+
+var common = require('../common.js');
+var PORT = common.PORT;
+
+var bench = common.createBenchmark(main, {
+  len: [4, 8, 16, 32, 64, 128, 512, 1024],
+  type: ['buf'],
+  dur: [5],
+});
+
+var dur;
+var len;
+var type;
+var chunk;
+var encoding;
+
+function main(conf) {
+  dur = +conf.dur;
+  len = +conf.len;
+  type = conf.type;
+
+  switch (type) {
+    case 'buf':
+      chunk = new Buffer(len);
+      chunk.fill('x');
+      break;
+    case 'utf':
+      encoding = 'utf8';
+      chunk = new Array(len / 2 + 1).join('ü');
+      break;
+    case 'asc':
+      encoding = 'ascii';
+      chunk = new Array(len + 1).join('x');
+      break;
+    default:
+      throw new Error('invalid type: ' + type);
+      break;
+  }
+
+  server();
+}
+
+var net = require('net');
+
+function Writer() {
+  this.received = 0;
+  this.writable = true;
+}
+
+Writer.prototype.write = function(chunk, encoding, cb) {
+  this.received += chunk.length;
+
+  if (typeof encoding === 'function')
+    encoding();
+  else if (typeof cb === 'function')
+    cb();
+
+  return true;
+};
+
+// doesn't matter, never emits anything.
+Writer.prototype.on = function() {};
+Writer.prototype.once = function() {};
+Writer.prototype.emit = function() {};
+
+function server() {
+  var writer = new Writer();
+
+  // the actual benchmark.
+  var server = net.createServer(function(socket) {
+    socket.pipe(writer);
+  });
+
+  server.listen(PORT, function() {
+    var socket = net.connect(PORT);
+    socket.on('connect', function() {
+      bench.start();
+
+      socket.on('drain', send)
+      send()
+
+      setTimeout(function() {
+        var bytes = writer.received;
+        var gbits = (bytes * 8) / (1024 * 1024 * 1024);
+        bench.end(gbits);
+      }, dur * 1000);
+
+      function send() {
+        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 = new CorkedRequest(this);
+  this.corkedRequestsFree.next = new CorkedRequest(this);
 }
 
 WritableState.prototype.getBuffer = function writableStateGetBuffer() {
@@ -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
 function clearBuffer(stream, state) {
   state.bufferProcessing = true;
   var entry = state.bufferedRequest;
 
   if (stream._writev && entry && entry.next) {
     // Fast case, write everything using _writev()
-    var buffer = [];
-    var cbs = [];
+    var l = state.bufferedRequestCount;
+    var buffer = new Array(l);
+    var holder = state.corkedRequestsFree;
+    holder.entry = entry;
+
+    var count = 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 (var i = 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
+function CorkedRequest(state) {
+  this.next = null;
+  this.entry = null;
+
+  this.finish = (err) => {
+    var entry = this.entry;
+    this.entry = null;
+    while (entry) {
+      var cb = entry.callback;
+      state.pendingcb--;
+      cb(err);
+      entry = entry.next;
+    }
+    if (state.corkedRequestsFree) {
+      state.corkedRequestsFree.next = this;
+    } else {
+      state.corkedRequestsFree = this;
+    }
+  };
+}