stream: add fast-path for readable streams

Author: anonrig

benchmark/streams/readable-encoding.js

@@ -0,0 +1,35 @@
+'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,
+  });
+
+  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,10 @@ const {
   Promise,
   SafeSet,
   SymbolAsyncIterator,
-  Symbol
+  Symbol,
+  TypedArrayPrototypeGetBuffer,
+  TypedArrayPrototypeGetByteOffset,
+  TypedArrayPrototypeGetByteLength,
 } = primordials;
 
 module.exports = Readable;
@@ -42,6 +45,8 @@ const EE = require('events');
 const { Stream, prependListener } = require('internal/streams/legacy');
 const { Buffer } = require('buffer');
 
+const { TextEncoder } = require('internal/encoding');
+
 const {
   addAbortSignal,
 } = require('internal/streams/add-abort-signal');
@@ -73,6 +78,9 @@ const kPaused = Symbol('kPaused');
 
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
+const { normalizeEncoding } = require('internal/util');
+const { FastBuffer } = require('internal/buffer');
+const encoder = new TextEncoder();
 
 ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
 ObjectSetPrototypeOf(Readable, Stream);
@@ -251,9 +259,21 @@ 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);
+          const enc = normalizeEncoding(encoding);
+
+          if (enc === 'utf8') {
+            const buf = encoder.encode(chunk);
+            chunk = new FastBuffer(
+              TypedArrayPrototypeGetBuffer(buf),
+              TypedArrayPrototypeGetByteOffset(buf),
+              TypedArrayPrototypeGetByteLength(buf),
+            );
+          } else {
+            chunk = Buffer.from(chunk, enc);
+          }
           encoding = '';
         }
       }