stream: improve Readable.read() performance

read() performance is improved most by switching from an array to
a linked list for storing buffered data. However, other changes that
also contribute include: making some hot functions inlinable, faster
read() argument checking, and misc code rearrangement to avoid
unnecessary code execution.

PR-URL: #7077
Reviewed-By: Calvin Metcalf <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>

Author: mscdex

benchmark/streams/readable-bigread.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const common = require('../common');
+const v8 = require('v8');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [100e1]
+});
+
+function main(conf) {
+  const n = +conf.n;
+  const b = new Buffer(32);
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  // Force optimization before starting the benchmark
+  s.push(b);
+  v8.setFlagsFromString('--allow_natives_syntax');
+  eval('%OptimizeFunctionOnNextCall(s.read)');
+  s.push(b);
+  while (s.read(128));
+
+  bench.start();
+  for (var k = 0; k < n; ++k) {
+    for (var i = 0; i < 1e4; ++i)
+      s.push(b);
+    while (s.read(128));
+  }
+  bench.end(n);
+}

benchmark/streams/readable-bigunevenread.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const common = require('../common');
+const v8 = require('v8');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [100e1]
+});
+
+function main(conf) {
+  const n = +conf.n;
+  const b = new Buffer(32);
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  // Force optimization before starting the benchmark
+  s.push(b);
+  v8.setFlagsFromString('--allow_natives_syntax');
+  eval('%OptimizeFunctionOnNextCall(s.read)');
+  s.push(b);
+  while (s.read(106));
+
+  bench.start();
+  for (var k = 0; k < n; ++k) {
+    for (var i = 0; i < 1e4; ++i)
+      s.push(b);
+    while (s.read(106));
+  }
+  bench.end(n);
+}

benchmark/streams/readable-boundaryread.js

@@ -0,0 +1,33 @@
+'use strict';
+
+const common = require('../common');
+const v8 = require('v8');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [200e1]
+});
+
+function main(conf) {
+  const n = +conf.n;
+  const b = new Buffer(32);
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  // Force optimization before starting the benchmark
+  s.push(b);
+  v8.setFlagsFromString('--allow_natives_syntax');
+  eval('%OptimizeFunctionOnNextCall(s.push)');
+  eval('%OptimizeFunctionOnNextCall(s.read)');
+  s.push(b);
+  while (s.read(32));
+
+  bench.start();
+  for (var k = 0; k < n; ++k) {
+    for (var i = 0; i < 1e4; ++i)
+      s.push(b);
+    while (s.read(32));
+  }
+  bench.end(n);
+}

benchmark/streams/readable-readall.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const common = require('../common');
+const v8 = require('v8');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [50e2]
+});
+
+function main(conf) {
+  const n = +conf.n;
+  const b = new Buffer(32);
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  // Force optimization before starting the benchmark
+  s.push(b);
+  v8.setFlagsFromString('--allow_natives_syntax');
+  eval('%OptimizeFunctionOnNextCall(s.read)');
+  s.push(b);
+  while (s.read());
+
+  bench.start();
+  for (var k = 0; k < n; ++k) {
+    for (var i = 0; i < 1e4; ++i)
+      s.push(b);
+    while (s.read());
+  }
+  bench.end(n);
+}

benchmark/streams/readable-unevenread.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const common = require('../common');
+const v8 = require('v8');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [100e1]
+});
+
+function main(conf) {
+  const n = +conf.n;
+  const b = new Buffer(32);
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  // Force optimization before starting the benchmark
+  s.push(b);
+  v8.setFlagsFromString('--allow_natives_syntax');
+  eval('%OptimizeFunctionOnNextCall(s.read)');
+  s.push(b);
+  while (s.read(12));
+
+  bench.start();
+  for (var k = 0; k < n; ++k) {
+    for (var i = 0; i < 1e4; ++i)
+      s.push(b);
+    while (s.read(12));
+  }
+  bench.end(n);
+}

lib/_stream_readable.js

@@ -0,0 +1,72 @@
+'use strict';
+
+const Buffer = require('buffer').Buffer;
+
+module.exports = BufferList;
+
+function BufferList() {
+  this.head = null;
+  this.tail = null;
+  this.length = 0;
+}
+
+BufferList.prototype.push = function(v) {
+  const entry = { data: v, next: null };
+  if (this.length > 0)
+    this.tail.next = entry;
+  else
+    this.head = entry;
+  this.tail = entry;
+  ++this.length;
+};
+
+BufferList.prototype.unshift = function(v) {
+  const entry = { data: v, next: this.head };
+  if (this.length === 0)
+    this.tail = entry;
+  this.head = entry;
+  ++this.length;
+};
+
+BufferList.prototype.shift = function() {
+  if (this.length === 0)
+    return;
+  const ret = this.head.data;
+  if (this.length === 1)
+    this.head = this.tail = null;
+  else
+    this.head = this.head.next;
+  --this.length;
+  return ret;
+};
+
+BufferList.prototype.clear = function() {
+  this.head = this.tail = null;
+  this.length = 0;
+};
+
+BufferList.prototype.join = function(s) {
+  if (this.length === 0)
+    return '';
+  var p = this.head;
+  var ret = '' + p.data;
+  while (p = p.next)
+    ret += s + p.data;
+  return ret;
+};
+
+BufferList.prototype.concat = function(n) {
+  if (this.length === 0)
+    return Buffer.alloc(0);
+  if (this.length === 1)
+    return this.head.data;
+  const ret = Buffer.allocUnsafe(n >>> 0);
+  var p = this.head;
+  var i = 0;
+  while (p) {
+    p.data.copy(ret, i);
+    i += p.data.length;
+    p = p.next;
+  }
+  return ret;
+};

lib/internal/streams/BufferList.js

@@ -88,6 +88,7 @@
       'lib/internal/v8_prof_polyfill.js',
       'lib/internal/v8_prof_processor.js',
       'lib/internal/streams/lazy_transform.js',
+      'lib/internal/streams/BufferList.js',
       'deps/v8/tools/splaytree.js',
       'deps/v8/tools/codemap.js',
       'deps/v8/tools/consarray.js',

node.gyp

@@ -26,7 +26,6 @@ s.read(0);
 // ACTUALLY [1, 3, 5, 6, 4, 2]
 
 process.on('exit', function() {
-  assert.deepStrictEqual(s._readableState.buffer,
-                         ['1', '2', '3', '4', '5', '6']);
+  assert.deepStrictEqual(s._readableState.buffer.join(','), '1,2,3,4,5,6');
   console.log('ok');
 });

test/parallel/test-stream-push-order.js

@@ -1,7 +1,9 @@
+// Flags: --expose_internals
 'use strict';
 require('../common');
 var assert = require('assert');
 var fromList = require('_stream_readable')._fromList;
+var BufferList = require('internal/streams/BufferList');
 
 // tiny node-tap lookalike.
 var tests = [];
@@ -30,6 +32,13 @@ function run() {
   });
 }
 
+function bufferListFromArray(arr) {
+  const bl = new BufferList();
+  for (var i = 0; i < arr.length; ++i)
+    bl.push(arr[i]);
+  return bl;
+}
+
 // ensure all tests have run
 process.on('exit', function() {
   assert.equal(count, 0);
@@ -43,6 +52,7 @@ test('buffers', function(t) {
                Buffer.from('bark'),
                Buffer.from('bazy'),
                Buffer.from('kuel') ];
+  list = bufferListFromArray(list);
 
   // read more than the first element.
   var ret = fromList(6, { buffer: list, length: 16 });
@@ -61,7 +71,7 @@ test('buffers', function(t) {
   t.equal(ret.toString(), 'zykuel');
 
   // all consumed.
-  t.same(list, []);
+  t.same(list, new BufferList());
 
   t.end();
 });
@@ -71,6 +81,7 @@ test('strings', function(t) {
                'bark',
                'bazy',
                'kuel' ];
+  list = bufferListFromArray(list);
 
   // read more than the first element.
   var ret = fromList(6, { buffer: list, length: 16, decoder: true });
@@ -89,7 +100,7 @@ test('strings', function(t) {
   t.equal(ret, 'zykuel');
 
   // all consumed.
-  t.same(list, []);
+  t.same(list, new BufferList());
 
   t.end();
 });