stream: readable read one buffer at a time

ronag · web-flow · commit fadb214d9574 · 2026-01-21T08:29:40.000Z
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>
diff --git a/lib/internal/streams/readable.js b/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)
diff --git a/test/common/heap.js b/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 =
diff --git a/test/parallel/test-crypto-cipheriv-decipheriv.js b/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}`);
diff --git a/test/parallel/test-runner-run.mjs b/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/);
 });
 });
 
diff --git a/test/parallel/test-stream-compose.js b/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());
 }
 
diff --git a/test/parallel/test-stream-push-strings.js b/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);
diff --git a/test/parallel/test-stream-readable-emittedReadable.js b/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);
diff --git a/test/parallel/test-stream-readable-infinite-read.js b/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));
diff --git a/test/parallel/test-stream-readable-needReadable.js b/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.
diff --git a/test/parallel/test-stream-readable-readable-one.js b/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);
+ }
+}
diff --git a/test/parallel/test-stream-typedarray.js b/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]]);
 }
 
 {
diff --git a/test/parallel/test-stream-uint8array.js b/test/parallel/test-stream-uint8array.js
@@ -80,9 +80,15 @@ const GHI = new Uint8Array([0x47, 0x48, 0x49]);
 readable.push(DEF);
 readable.unshift(ABC);
 
- const buf = readable.read();
+ let buf;
+
+ buf = readable.read();
+ assert(buf instanceof Buffer);
+ assert.deepStrictEqual([...buf], [...ABC]);
+
+ buf = readable.read();
 assert(buf instanceof Buffer);
- assert.deepStrictEqual([...buf], [...ABC, ...DEF]);
+ assert.deepStrictEqual([...buf], [...DEF]);
 }
 
 {
diff --git a/test/parallel/test-stream2-transform.js b/test/parallel/test-stream2-transform.js
@@ -282,7 +282,9 @@ const { PassThrough, Transform } = require('stream');
 pt.write(Buffer.from('ef'), common.mustCall(function() {
 pt.end();
 }));
- assert.strictEqual(pt.read().toString(), 'abcdef');
+ assert.strictEqual(pt.read().toString(), 'abc');
+ assert.strictEqual(pt.read().toString(), 'd');
+ assert.strictEqual(pt.read().toString(), 'ef');
 assert.strictEqual(pt.read(), null);
 });
 });
diff --git a/test/parallel/test-webstreams-pipeline.js b/test/parallel/test-webstreams-pipeline.js
@@ -126,7 +126,7 @@ const http = require('http');
 });
 
 pipeline(r, ws, common.mustSucceed(() => {
- assert.deepStrictEqual(values, ['helloworld']);
+ assert.deepStrictEqual(values, ['hello', 'world']);
 }));
 
 r.push('hello');
@@ -181,7 +181,7 @@ const http = require('http');
 });
 
 pipeline(rs, t, ws, common.mustSucceed(() => {
- assert.deepStrictEqual(values, ['HELLOWORLD']);
+ assert.deepStrictEqual(values, ['HELLO', 'WORLD']);
 }));
 
 c.enqueue('hello');
diff --git a/test/parallel/test-worker-heap-snapshot.js b/test/parallel/test-worker-heap-snapshot.js
@@ -10,7 +10,8 @@ const { once } = require('events');
 const worker = new Worker('setInterval(() => {}, 1000);', { eval: true });
 await once(worker, 'online');
 const stream = await worker.getHeapSnapshot();
- assert.ok(JSON.parse(stream.read()));
+ stream.read(0); // Trigger the stream to start flowing
+ assert.ok(JSON.parse(stream.read(stream.readableLength)));
 
 await worker.terminate();
 })().then(common.mustCall());
diff --git a/test/parallel/test-worker-stdio-from-preload-module.js b/test/parallel/test-worker-stdio-from-preload-module.js
@@ -15,6 +15,6 @@ for (let i = 0; i < 10; i++) {
 stdout: true
 });
 w.on('exit', common.mustCall(() => {
- assert.strictEqual(w.stdout.read().toString(), 'A\nB\n');
+ assert.strictEqual(w.stdout.read(w.stdout.readableLength).toString(), 'A\nB\n');
 }));
 }
diff --git a/test/parallel/test-zlib-flush-write-sync-interleaved.js b/test/parallel/test-zlib-flush-write-sync-interleaved.js
@@ -19,7 +19,7 @@ for (const chunk of ['abc', 'def', 'ghi']) {
 compress.write(chunk, common.mustCall(() => events.push({ written: chunk })));
 compress.flush(Z_PARTIAL_FLUSH, common.mustCall(() => {
 events.push('flushed');
- const chunk = compress.read();
+ const chunk = compress.read(compress.readableLength);
 if (chunk !== null)
 compressedChunks.push(chunk);
 }));
@@ -36,7 +36,7 @@ function writeToDecompress() {
 const chunk = compressedChunks.shift();
 if (chunk === undefined) return decompress.end();
 decompress.write(chunk, common.mustCall(() => {
- events.push({ read: decompress.read() });
+ events.push({ read: decompress.read(decompress.readableLength) });
 writeToDecompress();
 }));
 }

-Original file line number
+Diff line change
 if((state[kState]&kObjectMode)!==0)
 return1;
 if(NumberIsNaN(n)){
 +// Fast path for buffers.
 +if((state[kState]&kDecoder)===0&&state.length)
 +returnstate.buffer[state.bufferIndex].length;
++
 // Only flow one buffer at a time.
 if((state[kState]&kFlowing)!==0&&state.length)
 returnstate.buffer[state.bufferIndex].length;
++
 returnstate.length;
+}
 if(n<=state.length)
-Original file line number
+Diff line change
 functioncreateJSHeapSnapshot(stream=getHeapSnapshot()){
 stream.pause();
 -constdump=JSON.parse(stream.read());
 +stream.read(0);
 +constdump=JSON.parse(stream.read(stream.readableLength));
 constmeta=dump.snapshot.meta;
 constnodes=
-Original file line number
+Diff line change
 // quite small, so there's no harm.
 constcStream=crypto.createCipheriv('des-ede3-cbc',key,iv);
 cStream.end(plaintext);
 -ciph=cStream.read();
 +ciph=cStream.read(cStream.readableLength);
 constdStream=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}`);
-Original file line number
+Diff line change
 import{join}from'node:path';
 import{describe,it,run}from'node:test';
 import{dot,spec,tap}from'node:test/reporters';
 +importconsumersfrom'node:stream/consumers';
 importassertfrom'node:assert';
 importutilfrom'node:util';
 describe('should be piped with spec reporter',()=>{
 it('new spec',async()=>{
 constspecReporter=newspec();
 -constresult=awaitrun({
 +constresult=awaitconsumers.text(run({
 files: [join(testFixtures,'default-behavior/test/random.cjs')]
 -}).compose(specReporter).toArray();
 -conststringResults=result.map((bfr)=>bfr.toString());
 -assert.match(stringResults[0],/thisshouldpass/);
 -assert.match(stringResults[1],/tests1/);
 -assert.match(stringResults[1],/pass1/);
 +}).compose(specReporter));
 +assert.match(result,/thisshouldpass/);
 +assert.match(result,/tests1/);
 +assert.match(result,/pass1/);
 });
 it('spec()',async()=>{
 constspecReporter=spec();
 -constresult=awaitrun({
 +constresult=awaitconsumers.text(run({
 files: [join(testFixtures,'default-behavior/test/random.cjs')]
 -}).compose(specReporter).toArray();
 -conststringResults=result.map((bfr)=>bfr.toString());
 -assert.match(stringResults[0],/thisshouldpass/);
 -assert.match(stringResults[1],/tests1/);
 -assert.match(stringResults[1],/pass1/);
 +}).compose(specReporter));
 +assert.match(result,/thisshouldpass/);
 +assert.match(result,/tests1/);
 +assert.match(result,/pass1/);
 });
 it('spec',async()=>{
 -constresult=awaitrun({
 +constresult=awaitconsumers.text(run({
 files: [join(testFixtures,'default-behavior/test/random.cjs')]
 -}).compose(spec).toArray();
 -conststringResults=result.map((bfr)=>bfr.toString());
 -assert.match(stringResults[0],/thisshouldpass/);
 -assert.match(stringResults[1],/tests1/);
 -assert.match(stringResults[1],/pass1/);
 +}).compose(spec));
 +assert.match(result,/thisshouldpass/);
 +assert.match(result,/tests1/);
 +assert.match(result,/pass1/);
 });
 });
Original file line number	Diff line number	Diff line change
`@@ -490,7 +490,7 @@ const assert = require('assert');`
`490`	`490`
`491`	`491`	`newStream.end();`
`492`	`492`
`493`		`-assert.deepStrictEqual(awaitnewStream.toArray(),[Buffer.from('Steve RogersOn your left')]);`
	`493`	`+assert.deepStrictEqual(awaitnewStream.toArray(),[Buffer.from('Steve Rogers'),Buffer.from('On your left')]);`
`494`	`494`	`})().then(common.mustCall());`
`495`	`495`	`}`
`496`	`496`
-Original file line number
+Diff line change
 results.push(String(chunk));
 });
 -constexpect=['first chunksecond to last chunk','last chunk'];
 +constexpect=['first chunk','second to last chunk','last chunk'];
 process.on('exit',function(){
 assert.strictEqual(ms._chunks,-1);
 assert.deepStrictEqual(results,expect);
-Original file line number
+Diff line change
 // Initialized to false.
 assert.strictEqual(readable._readableState.emittedReadable,false);
 -constexpected=[Buffer.from('foobar'),Buffer.from('quo'),null];
 +constexpected=[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);
-Original file line number
+Diff line change
 highWaterMark: 16*1024,
 read: common.mustCall(function(){
 this.push(buf);
 -},31)
 +},12)
 });
 leti=0;
 readable.on('readable',common.mustCall(function(){
 if(i++===10){
 -// We will just terminate now.
 -process.removeAllListeners('readable');
 +readable.removeAllListeners('readable');
 return;
+}
 -constdata=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));
-Original file line number
+Diff line change
@@ @@ -0,0 +1,20 @@ @@
 +'use strict';
 +require('../common');
 +constassert=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.
 +{
 +constr=newReadable({
 +read(){}
 +});
 +constbuffers=[Buffer.allocUnsafe(5),Buffer.allocUnsafe(10)];
 +for(constbufofbuffers){
 +r.push(buf);
 +}
 +for(constbufofbuffers){
 +assert.strictEqual(r.read(),buf);
 +}
 +}