Skip to content

Commit

Permalink
stream: improve tee perf by reduce ReflectConstruct usages
Browse files Browse the repository at this point in the history
also added more webstream creation benchmarks

PR-URL: nodejs#49546
Reviewed-By: Yagiz Nizipli <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
  • Loading branch information
rluvaton committed Sep 12, 2023
1 parent 022f1b7 commit 3f7b05e
Show file tree
Hide file tree
Showing 2 changed files with 96 additions and 35 deletions.
68 changes: 60 additions & 8 deletions benchmark/webstreams/creation.js
Original file line number Diff line number Diff line change
Expand Up @@ -2,47 +2,99 @@
const common = require('../common.js');
const {
ReadableStream,
ReadableStreamDefaultReader,
ReadableStreamBYOBReader,
TransformStream,
WritableStream,
} = require('node:stream/web');
const assert = require('assert');

const bench = common.createBenchmark(main, {
n: [50e3],
kind: ['ReadableStream', 'TransformStream', 'WritableStream'],
kind: [
'ReadableStream',
'TransformStream',
'WritableStream',

'ReadableStreamDefaultReader',
'ReadableStreamBYOBReader',

'ReadableStream.tee',
],
});

let rs, ws, ts;
let readableStream;
let transformStream;
let writableStream;
let readableStreamDefaultReader;
let readableStreamBYOBReader;
let teeResult;

function main({ n, kind }) {
switch (kind) {
case 'ReadableStream':
bench.start();
for (let i = 0; i < n; ++i)
rs = new ReadableStream();
readableStream = new ReadableStream();
bench.end(n);

// Avoid V8 deadcode (elimination)
assert.ok(rs);
assert.ok(readableStream);
break;
case 'WritableStream':
bench.start();
for (let i = 0; i < n; ++i)
ws = new WritableStream();
writableStream = new WritableStream();
bench.end(n);

// Avoid V8 deadcode (elimination)
assert.ok(ws);
assert.ok(writableStream);
break;
case 'TransformStream':
bench.start();
for (let i = 0; i < n; ++i)
ts = new TransformStream();
transformStream = new TransformStream();
bench.end(n);

// Avoid V8 deadcode (elimination)
assert.ok(transformStream);
break;
case 'ReadableStreamDefaultReader': {
const readers = Array.from({ length: n }, () => new ReadableStream());

bench.start();
for (let i = 0; i < n; ++i)
readableStreamDefaultReader = new ReadableStreamDefaultReader(readers[i]);
bench.end(n);

// Avoid V8 deadcode (elimination)
assert.ok(readableStreamDefaultReader);
break;
}
case 'ReadableStreamBYOBReader': {
const readers = Array.from({ length: n }, () => new ReadableStream({ type: 'bytes' }));

bench.start();
for (let i = 0; i < n; ++i)
readableStreamBYOBReader = new ReadableStreamBYOBReader(readers[i]);
bench.end(n);

// Avoid V8 deadcode (elimination)
assert.ok(readableStreamBYOBReader);
break;
}
case 'ReadableStream.tee': {
const streams = Array.from({ length: n }, () => new ReadableStream());

bench.start();
for (let i = 0; i < n; ++i)
teeResult = streams[i].tee();
bench.end(n);

// Avoid V8 deadcode (elimination)
assert.ok(ts);
assert.ok(teeResult);
break;
}
default:
throw new Error('Invalid kind');
}
Expand Down
63 changes: 36 additions & 27 deletions lib/internal/webstreams/readablestream.js
Original file line number Diff line number Diff line change
Expand Up @@ -1200,34 +1200,43 @@ ObjectDefineProperties(ReadableByteStreamController.prototype, {
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableByteStreamController.name),
});

function TeeReadableStream(start, pull, cancel) {
this[kType] = 'ReadableStream';
this[kState] = {
disturbed: false,
state: 'readable',
storedError: undefined,
stream: undefined,
transfer: {
writable: undefined,
port: undefined,
promise: undefined,
},
};
this[kIsClosedPromise] = createDeferredPromise();
setupReadableStreamDefaultControllerFromSource(
this,
ObjectCreate(null, {
start: { __proto__: null, value: start },
pull: { __proto__: null, value: pull },
cancel: { __proto__: null, value: cancel },
}),
1,
() => 1);


return makeTransferable(this);
}

ObjectSetPrototypeOf(TeeReadableStream.prototype, ReadableStream.prototype);
ObjectSetPrototypeOf(TeeReadableStream, ReadableStream);

function createTeeReadableStream(start, pull, cancel) {
return ReflectConstruct(
function() {
this[kType] = 'ReadableStream';
this[kState] = {
disturbed: false,
state: 'readable',
storedError: undefined,
stream: undefined,
transfer: {
writable: undefined,
port: undefined,
promise: undefined,
},
};
this[kIsClosedPromise] = createDeferredPromise();
setupReadableStreamDefaultControllerFromSource(
this,
ObjectCreate(null, {
start: { __proto__: null, value: start },
pull: { __proto__: null, value: pull },
cancel: { __proto__: null, value: cancel },
}),
1,
() => 1);
return makeTransferable(this);
}, [], ReadableStream,
);
const tee = new TeeReadableStream(start, pull, cancel);

// For spec compliance the Tee must be a ReadableStream
tee.constructor = ReadableStream;
return tee;
}

const isReadableStream =
Expand Down

0 comments on commit 3f7b05e

Please sign in to comment.