stream: add fast-path for readable streams

benchmark/streams/readable-encoding.js

@@ -0,0 +1,37 @@
+'use strict';
+
+const common = require('../common');
+const Readable = require('stream').Readable;
+
+const BASE = 'hello world\n\n';
+
+const bench = common.createBenchmark(main, {
+  encoding: ['utf-8', 'latin1'],
+  len: [256, 512, 1024 * 16],
+  op: ['unshift', 'push'],
+  n: [1e3]
+});
+
+function main({ n, encoding, len, op }) {
+  const b = BASE.repeat(len);
+  const s = new Readable({
+    objectMode: false,
+  });
+  function noop() {}
+  s._read = noop;
+
+  bench.start();
+  switch (op) {
+    case 'unshift': {
+      for (let i = 0; i < n; i++)
+        s.unshift(b, encoding);
+      break;
+    }
+    case 'push': {
+      for (let i = 0; i < n; i++)
+        s.push(b, encoding);
+      break;
+    }
+  }
+  bench.end(n);
+}

lib/internal/streams/readable.js

@@ -32,7 +32,7 @@ const {
   Promise,
   SafeSet,
   SymbolAsyncIterator,
-  Symbol
+  Symbol,
 } = primordials;
 
 module.exports = Readable;
@@ -73,6 +73,7 @@ const kPaused = Symbol('kPaused');
 
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
+const { encodeWithFastPath } = require('internal/streams/utils');
 
 ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
 ObjectSetPrototypeOf(Readable, Stream);
@@ -251,9 +252,10 @@ function readableAddChunk(stream, chunk, encoding, addToFront) {
         if (addToFront && state.encoding) {
           // When unshifting, if state.encoding is set, we have to save
           // the string in the BufferList with the state encoding.
+
           chunk = Buffer.from(chunk, encoding).toString(state.encoding);
         } else {
-          chunk = Buffer.from(chunk, encoding);
+          chunk = encodeWithFastPath(chunk, encoding);
           encoding = '';
         }
       }

lib/internal/streams/utils.js

@@ -4,8 +4,23 @@ const {
   Symbol,
   SymbolAsyncIterator,
   SymbolIterator,
+  TypedArrayPrototypeGetBuffer,
+  TypedArrayPrototypeGetByteOffset,
+  TypedArrayPrototypeGetByteLength,
 } = primordials;
 
+const { normalizeEncoding } = require('internal/util');
+const { FastBuffer } = require('internal/buffer');
+const { TextEncoder } = require('internal/encoding');
+const { isTypedArray } = require('internal/util/types');
+
+let buffer;
+function lazyLoadBuffer() {
+  buffer ??= require('buffer').Buffer;
+  return buffer;
+}
+const encoder = new TextEncoder();
+
 const kDestroyed = Symbol('kDestroyed');
 const kIsErrored = Symbol('kIsErrored');
 const kIsReadable = Symbol('kIsReadable');
@@ -261,7 +276,24 @@ function isErrored(stream) {
   ));
 }
 
+function encodeWithFastPath(input, encoding) {
+  const enc = normalizeEncoding(encoding);
+  const byteLength = enc === 'utf8' && isTypedArray(input) && TypedArrayPrototypeGetByteLength(input);
+
+  if (byteLength > 512) {
+    const buf = encoder.encode(input);
+    return new FastBuffer(
+      TypedArrayPrototypeGetBuffer(buf),
+      TypedArrayPrototypeGetByteOffset(buf),
+      byteLength,
+    );
+  }
+
+  return lazyLoadBuffer().from(input, enc);
+}
+
 module.exports = {
+  encodeWithFastPath,
   kDestroyed,
   isDisturbed,
   kIsDisturbed,