stream: skip chunk validation for Buffers

benchmark/streams/writable-manywrites.js

@@ -0,0 +1,23 @@
+'use strict';
+
+const common = require('../common');
+const Writable = require('stream').Writable;
+
+const bench = common.createBenchmark(main, {
+  n: [2e6]
+});
+
+function main(conf) {
+  const n = +conf.n;
+  const b = Buffer.allocUnsafe(1024);
+  const s = new Writable();
+  s._write = function(chunk, encoding, cb) {
+    cb();
+  };
+
+  bench.start();
+  for (var k = 0; k < n; ++k) {
+    s.write(b);
+  }
+  bench.end(n);
+}

lib/_stream_writable.js

@@ -194,23 +194,18 @@ function writeAfterEnd(stream, cb) {
   process.nextTick(cb, er);
 }
 
-// If we get something that is not a buffer, string, null, or undefined,
-// and we're not in objectMode, then that's an error.
-// Otherwise stream chunks are all considered to be of length=1, and the
-// watermarks determine how many objects to keep in the buffer, rather than
-// how many bytes or characters.
+// Checks that a user-supplied chunk is valid, especially for the particular
+// mode the stream is in. Currently this means that `null` is never accepted
+// and undefined/non-string values are only allowed in object mode.
 function validChunk(stream, state, chunk, cb) {
   var valid = true;
   var er = false;
-  // Always throw error if a null is written
-  // if we are not in object mode then throw
-  // if it is not a buffer, string, or undefined.
+
   if (chunk === null) {
     er = new TypeError('May not write null values to stream');
-  } else if (!(chunk instanceof Buffer) &&
-      typeof chunk !== 'string' &&
-      chunk !== undefined &&
-      !state.objectMode) {
+  } else if (typeof chunk !== 'string' &&
+             chunk !== undefined &&
+             !state.objectMode) {
     er = new TypeError('Invalid non-string/buffer chunk');
   }
   if (er) {
@@ -224,13 +219,14 @@ function validChunk(stream, state, chunk, cb) {
 Writable.prototype.write = function(chunk, encoding, cb) {
   var state = this._writableState;
   var ret = false;
+  var isBuf = (chunk instanceof Buffer);
 
   if (typeof encoding === 'function') {
     cb = encoding;
     encoding = null;
   }
 
-  if (chunk instanceof Buffer)
+  if (isBuf)
     encoding = 'buffer';
   else if (!encoding)
     encoding = state.defaultEncoding;
@@ -240,9 +236,9 @@ Writable.prototype.write = function(chunk, encoding, cb) {
 
   if (state.ended)
     writeAfterEnd(this, cb);
-  else if (validChunk(this, state, chunk, cb)) {
+  else if (isBuf || validChunk(this, state, chunk, cb)) {
     state.pendingcb++;
-    ret = writeOrBuffer(this, state, chunk, encoding, cb);
+    ret = writeOrBuffer(this, state, isBuf, chunk, encoding, cb);
   }
 
   return ret;
@@ -291,11 +287,12 @@ function decodeChunk(state, chunk, encoding) {
 // if we're already writing something, then just put this
 // in the queue, and wait our turn.  Otherwise, call _write
 // If we return false, then we need a drain event, so set that flag.
-function writeOrBuffer(stream, state, chunk, encoding, cb) {
-  chunk = decodeChunk(state, chunk, encoding);
-
-  if (chunk instanceof Buffer)
-    encoding = 'buffer';
+function writeOrBuffer(stream, state, isBuf, chunk, encoding, cb) {
+  if (!isBuf) {
+    chunk = decodeChunk(state, chunk, encoding);
+    if (chunk instanceof Buffer)
+      encoding = 'buffer';
+  }
   var len = state.objectMode ? 1 : chunk.length;
 
   state.length += len;