stream: readable read one buffer at a time

Instead of wasting cycles concatenating buffers, just return each
one by one.

Similar (but not exact) old behavior can be achieved by using
`readable.read(readable.readableLength)` instead of
`readable.read()`. In some edge cases it might be necessary
to do a `readable.read(0)` first.

PR: #60441
PR-URL: #60441
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>

Author: ronag

lib/internal/streams/readable.js

@@ -636,9 +636,14 @@ function howMuchToRead(n, state) {
   if ((state[kState] & kObjectMode) !== 0)
     return 1;
   if (NumberIsNaN(n)) {
+    // Fast path for buffers.
+    if ((state[kState] & kDecoder) === 0 && state.length)
+      return state.buffer[state.bufferIndex].length;
+
     // Only flow one buffer at a time.
     if ((state[kState] & kFlowing) !== 0 && state.length)
       return state.buffer[state.bufferIndex].length;
+
     return state.length;
   }
   if (n <= state.length)

test/common/heap.js

@@ -21,7 +21,8 @@ const { getHeapSnapshot } = require('v8');
 
 function createJSHeapSnapshot(stream = getHeapSnapshot()) {
   stream.pause();
-  const dump = JSON.parse(stream.read());
+  stream.read(0);
+  const dump = JSON.parse(stream.read(stream.readableLength));
   const meta = dump.snapshot.meta;
 
   const nodes =

test/parallel/test-crypto-cipheriv-decipheriv.js

@@ -31,11 +31,11 @@ function testCipher1(key, iv) {
   // quite small, so there's no harm.
   const cStream = crypto.createCipheriv('des-ede3-cbc', key, iv);
   cStream.end(plaintext);
-  ciph = cStream.read();
+  ciph = cStream.read(cStream.readableLength);
 
   const dStream = crypto.createDecipheriv('des-ede3-cbc', key, iv);
   dStream.end(ciph);
-  txt = dStream.read().toString('utf8');
+  txt = dStream.read(dStream.readableLength).toString('utf8');
 
   assert.strictEqual(txt, plaintext,
                      `streaming cipher with key ${key} and iv ${iv}`);

test/parallel/test-runner-run.mjs

@@ -3,6 +3,7 @@ import * as fixtures from '../common/fixtures.mjs';
 import { join } from 'node:path';
 import { describe, it, run } from 'node:test';
 import { dot, spec, tap } from 'node:test/reporters';
+import consumers from 'node:stream/consumers';
 import assert from 'node:assert';
 import util from 'node:util';
 
@@ -111,34 +112,31 @@ describe('require(\'node:test\').run', { concurrency: true }, () => {
   describe('should be piped with spec reporter', () => {
     it('new spec', async () => {
       const specReporter = new spec();
-      const result = await run({
+      const result = await consumers.text(run({
         files: [join(testFixtures, 'default-behavior/test/random.cjs')]
-      }).compose(specReporter).toArray();
-      const stringResults = result.map((bfr) => bfr.toString());
-      assert.match(stringResults[0], /this should pass/);
-      assert.match(stringResults[1], /tests 1/);
-      assert.match(stringResults[1], /pass 1/);
+      }).compose(specReporter));
+      assert.match(result, /this should pass/);
+      assert.match(result, /tests 1/);
+      assert.match(result, /pass 1/);
     });
 
     it('spec()', async () => {
       const specReporter = spec();
-      const result = await run({
+      const result = await consumers.text(run({
         files: [join(testFixtures, 'default-behavior/test/random.cjs')]
-      }).compose(specReporter).toArray();
-      const stringResults = result.map((bfr) => bfr.toString());
-      assert.match(stringResults[0], /this should pass/);
-      assert.match(stringResults[1], /tests 1/);
-      assert.match(stringResults[1], /pass 1/);
+      }).compose(specReporter));
+      assert.match(result, /this should pass/);
+      assert.match(result, /tests 1/);
+      assert.match(result, /pass 1/);
     });
 
     it('spec', async () => {
-      const result = await run({
+      const result = await consumers.text(run({
         files: [join(testFixtures, 'default-behavior/test/random.cjs')]
-      }).compose(spec).toArray();
-      const stringResults = result.map((bfr) => bfr.toString());
-      assert.match(stringResults[0], /this should pass/);
-      assert.match(stringResults[1], /tests 1/);
-      assert.match(stringResults[1], /pass 1/);
+      }).compose(spec));
+      assert.match(result, /this should pass/);
+      assert.match(result, /tests 1/);
+      assert.match(result, /pass 1/);
     });
   });
 

test/parallel/test-stream-compose.js

@@ -490,7 +490,7 @@ const assert = require('assert');
 
     newStream.end();
 
-    assert.deepStrictEqual(await newStream.toArray(), [Buffer.from('Steve RogersOn your left')]);
+    assert.deepStrictEqual(await newStream.toArray(), [Buffer.from('Steve Rogers'), Buffer.from('On your left')]);
   })().then(common.mustCall());
 }
 

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

@@ -59,7 +59,7 @@ ms.on('readable', function() {
     results.push(String(chunk));
 });
 
-const expect = [ 'first chunksecond to last chunk', 'last chunk' ];
+const expect = [ 'first chunk', 'second to last chunk', 'last chunk' ];
 process.on('exit', function() {
   assert.strictEqual(ms._chunks, -1);
   assert.deepStrictEqual(results, expect);

test/parallel/test-stream-readable-emittedReadable.js

@@ -10,7 +10,7 @@ const readable = new Readable({
 // Initialized to false.
 assert.strictEqual(readable._readableState.emittedReadable, false);
 
-const expected = [Buffer.from('foobar'), Buffer.from('quo'), null];
+const expected = [Buffer.from('foo'), Buffer.from('bar'), Buffer.from('quo'), null];
 readable.on('readable', common.mustCall(() => {
   // emittedReadable should be true when the readable event is emitted
   assert.strictEqual(readable._readableState.emittedReadable, true);

test/parallel/test-stream-readable-infinite-read.js

@@ -10,24 +10,16 @@ const readable = new Readable({
   highWaterMark: 16 * 1024,
   read: common.mustCall(function() {
     this.push(buf);
-  }, 31)
+  }, 12)
 });
 
 let i = 0;
 
 readable.on('readable', common.mustCall(function() {
   if (i++ === 10) {
-    // We will just terminate now.
-    process.removeAllListeners('readable');
+    readable.removeAllListeners('readable');
     return;
   }
 
-  const data = readable.read();
-  // TODO(mcollina): there is something odd in the highWaterMark logic
-  // investigate.
-  if (i === 1) {
-    assert.strictEqual(data.length, 8192 * 2);
-  } else {
-    assert.strictEqual(data.length, 8192 * 3);
-  }
+  assert.strictEqual(readable.read().length, 8192);
 }, 11));

test/parallel/test-stream-readable-needReadable.js

@@ -32,7 +32,7 @@ const asyncReadable = new Readable({
 });
 
 asyncReadable.on('readable', common.mustCall(() => {
-  if (asyncReadable.read() !== null) {
+  if (asyncReadable.read(asyncReadable.readableLength) !== null) {
     // After each read(), the buffer is empty.
     // If the stream doesn't end now,
     // then we need to notify the reader on future changes.

test/parallel/test-stream-readable-readable-one.js

@@ -0,0 +1,20 @@
+'use strict';
+require('../common');
+const assert = require('assert');
+
+const { Readable } = require('stream');
+
+// Read one buffer at a time and don't waste cycles allocating
+// and copying into a new larger buffer.
+{
+  const r = new Readable({
+    read() {}
+  });
+  const buffers = [Buffer.allocUnsafe(5), Buffer.allocUnsafe(10)];
+  for (const buf of buffers) {
+    r.push(buf);
+  }
+  for (const buf of buffers) {
+    assert.strictEqual(r.read(), buf);
+  }
+}