stream: readable read one buffer at a time

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

Old behavior can be achieved by using `readable.read(readable.readableLength)` instead of `readable.read()`.

PR: #60441

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/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,17 @@ const readable = new Readable({
   highWaterMark: 16 * 1024,
   read: common.mustCall(function() {
     this.push(buf);
-  }, 31)
+  }, 13)
 });
 
 let i = 0;
 
 readable.on('readable', common.mustCall(function() {
-  if (i++ === 10) {
+  if (i++ > 10) {
     // We will just terminate now.
     process.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);
-  }
-}, 11));
+  assert.strictEqual(readable.read().length, 8192);
+}, 13));

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';
+const common = 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);
+  }
+}

test/parallel/test-stream-typedarray.js

@@ -83,9 +83,19 @@ const views = common.getArrayBufferViews(buffer);
   readable.push(views[2]);
   readable.unshift(views[0]);
 
-  const buf = readable.read();
+  let buf;
+
+  buf = readable.read();
+  assert(buf instanceof Buffer);
+  assert.deepStrictEqual([...buf], [...views[0]]);
+
+  buf = readable.read();
+  assert(buf instanceof Buffer);
+  assert.deepStrictEqual([...buf], [...views[1]]);
+
+  buf = readable.read();
   assert(buf instanceof Buffer);
-  assert.deepStrictEqual([...buf], [...views[0], ...views[1], ...views[2]]);
+  assert.deepStrictEqual([...buf], [...views[2]]);
 }
 
 {