net,stream: skip chunk check on incoming data

benchmark/streams/readable-skip-chunk-check.js

@@ -0,0 +1,28 @@
+'use strict';
+
+const common = require('../common');
+const { Readable } = require('stream');
+
+const bench = common.createBenchmark(main, {
+  skipChunkCheck: [1, 0],
+  n: [1e6]
+});
+
+function main({ n, skipChunkCheck }) {
+  const buf = Buffer.alloc(64);
+  const readable = new Readable({
+    skipChunkCheck,
+    read() {}
+  });
+
+  readable.on('end', function() {
+    bench.end(n);
+  });
+  readable.on('resume', function() {
+    bench.start();
+    for (var i = 0; i < n; i++)
+      this.push(buf);
+    this.push(null);
+  });
+  readable.resume();
+}

doc/api/stream.md

@@ -1662,16 +1662,24 @@ Custom Readable streams *must* call the `new stream.Readable([options])`
 constructor and implement the `readable._read()` method.
 
 #### new stream.Readable([options])
+<!-- YAML
+changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/19559
+    description: The `skipChunkCheck` option is supported now.
+-->
 
 * `options` {Object}
   * `highWaterMark` {number} The maximum [number of bytes][hwm-gotcha] to store
     in the internal buffer before ceasing to read from the underlying resource.
-    Defaults to `16384` (16kb), or `16` for `objectMode` streams
+    Defaults to `16384` (16kb), or `16` for `objectMode` streams.
   * `encoding` {string} If specified, then buffers will be decoded to
-    strings using the specified encoding. Defaults to `null`
+    strings using the specified encoding. Defaults to `null`.
   * `objectMode` {boolean} Whether this stream should behave
     as a stream of objects. Meaning that [`stream.read(n)`][stream-read] returns
-    a single value instead of a Buffer of size n. Defaults to `false`
+    a single value instead of a Buffer of size n. Defaults to `false`.
+  * `skipChunkCheck` {boolean} Make [`stream.push()`][stream-push] skip type
+    checking. Defaults to `false`.
   * `read` {Function} Implementation for the [`stream._read()`][stream-_read]
     method.
   * `destroy` {Function} Implementation for the

lib/_stream_readable.js

@@ -82,6 +82,9 @@ function ReadableState(options, stream, isDuplex) {
   if (isDuplex)
     this.objectMode = this.objectMode || !!options.readableObjectMode;
 
+  // Skip type checking when adding something to the readable buffer.
+  this.skipChunkCheck = this.objectMode || !!options.skipChunkCheck;
+
   // the point at which it stops calling _read() to fill the buffer
   // Note: 0 is a valid value, means "don't call _read preemptively ever"
   this.highWaterMark = getHighWaterMark(this, options, 'readableHighWaterMark',
@@ -199,18 +202,14 @@ Readable.prototype._destroy = function(err, cb) {
 // write() some more.
 Readable.prototype.push = function(chunk, encoding) {
   var state = this._readableState;
-  var skipChunkCheck;
-
-  if (!state.objectMode) {
-    if (typeof chunk === 'string') {
-      encoding = encoding || state.defaultEncoding;
-      if (encoding !== state.encoding) {
-        chunk = Buffer.from(chunk, encoding);
-        encoding = '';
-      }
-      skipChunkCheck = true;
+  var skipChunkCheck = state.skipChunkCheck;
+
+  if (!state.objectMode && typeof chunk === 'string') {
+    encoding = encoding || state.defaultEncoding;
+    if (encoding !== state.encoding) {
+      chunk = Buffer.from(chunk, encoding);
+      encoding = '';
     }
-  } else {
     skipChunkCheck = true;
   }
 
@@ -219,7 +218,8 @@ Readable.prototype.push = function(chunk, encoding) {
 
 // Unshift should *always* be something directly out of read()
 Readable.prototype.unshift = function(chunk) {
-  return readableAddChunk(this, chunk, null, true, false);
+  return readableAddChunk(this, chunk, null, true,
+                          this._readableState.skipChunkCheck);
 };
 
 function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) {

lib/net.js

@@ -239,6 +239,10 @@ function Socket(options) {
   // For backwards compat do not emit close on destroy.
   options.emitClose = false;
 
+  // Data chunks read from the handle are always buffers so there is no need to
+  // validate them.
+  options.skipChunkCheck = true;
+
   // `DuplexBase` is just a slimmed down constructor for `Duplex` which allow
   // us to not inherit the "no-half-open enforcer" as there is already one in
   // place. Instances of `Socket` are still instances of `Duplex`, that is,

test/parallel/test-net-socket-skip-chunk-check.js

@@ -0,0 +1,11 @@
+'use strict';
+require('../common');
+
+// This test ensures that `net.Socket` uses the `skipChunkCheck` option
+// when calling the `Readable` constructor.
+
+const { Socket } = require('net');
+const { strictEqual } = require('assert');
+
+const socket = new Socket();
+strictEqual(socket._readableState.skipChunkCheck, true);

test/parallel/test-stream-readable-ctor-options.js

@@ -0,0 +1,32 @@
+'use strict';
+const common = require('../common');
+
+const { Readable } = require('stream');
+const { strictEqual } = require('assert');
+
+{
+  const _read = common.mustCall(function _read(n) {
+    this.push(null);
+  });
+
+  const r = new Readable({ read: _read });
+  r.resume();
+}
+
+{
+  const { _readableState: state } = new Readable();
+  strictEqual(state.objectMode, false);
+  strictEqual(state.skipChunkCheck, false);
+}
+
+{
+  const { _readableState: state } = new Readable({ objectMode: true });
+  strictEqual(state.objectMode, true);
+  strictEqual(state.skipChunkCheck, true);
+}
+
+{
+  const { _readableState: state } = new Readable({ skipChunkCheck: true });
+  strictEqual(state.objectMode, false);
+  strictEqual(state.skipChunkCheck, true);
+}