mirror of
https://github.com/nodejs/node.git
synced 2026-10-28 10:15:07 +08:00
stream: keep consumer state in fast mode
Create null-prototype share and broadcast consumer state with fast properties instead of V8 dictionary properties. Assisted-by: Pi Signed-off-by: Matteo Collina <hello@matteocollina.com> PR-URL: https://github.com/nodejs/node/pull/66266 Reviewed-By: Mattias Buelens <mattias@buelens.com> Reviewed-By: James M Snell <jasnell@gmail.com>
This commit is contained in:
1 parent
147ade5ff9
commit
7ff62671a5
3 files changed
+45
-9
No files matched your search
@@ -0,0 +1,37 @@
|
||||
'use strict';
|
||||
|
||||
const common = require('../common.js');
|
||||
|
||||
const bench = common.createBenchmark(main, {
|
||||
consumers: [2, 8, 32],
|
||||
batches: [1e4],
|
||||
n: [5],
|
||||
}, {
|
||||
flags: ['--experimental-stream-iter'],
|
||||
});
|
||||
|
||||
function main({ consumers, batches, n }) {
|
||||
const { shareSync } = require('stream/iter');
|
||||
const chunk = Buffer.alloc(1024);
|
||||
let bytes = 0;
|
||||
|
||||
function* source() {
|
||||
for (let i = 0; i < batches; i++) yield [chunk];
|
||||
}
|
||||
|
||||
bench.start();
|
||||
for (let run = 0; run < n; run++) {
|
||||
const shared = shareSync(source(), { budget: 65536 });
|
||||
const readers = Array.from({ length: consumers }, () =>
|
||||
shared.pull()[Symbol.iterator]());
|
||||
for (let i = 0; i < batches; i++) {
|
||||
for (let j = 0; j < consumers; j++) {
|
||||
bytes += readers[j].next().value[0].byteLength;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (bytes !== batches * consumers * n * chunk.byteLength) {
|
||||
throw new Error('Incorrect byte count');
|
||||
}
|
||||
bench.end(batches * consumers * n);
|
||||
}
|
||||
@@ -11,6 +11,7 @@ const {
|
||||
ArrayPrototypePush,
|
||||
ArrayPrototypeShift,
|
||||
FunctionPrototypeCall,
|
||||
ObjectSetPrototypeOf,
|
||||
PromisePrototypeThen,
|
||||
PromiseReject,
|
||||
PromiseResolve,
|
||||
@@ -185,8 +186,7 @@ class BroadcastImpl {
|
||||
}
|
||||
|
||||
#createRawConsumer() {
|
||||
const state = {
|
||||
__proto__: null,
|
||||
const state = ObjectSetPrototypeOf({
|
||||
// Start at the oldest buffered entry so late-joining consumers
|
||||
// can read data already in the buffer.
|
||||
cursor: this.#bufferStart,
|
||||
@@ -195,7 +195,7 @@ class BroadcastImpl {
|
||||
pending: [],
|
||||
detached: false,
|
||||
error: kNoBroadcastError,
|
||||
};
|
||||
}, null);
|
||||
|
||||
this.#consumers.add(state);
|
||||
if (this.#consumers.size === 1) {
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
const {
|
||||
ArrayPrototypePush,
|
||||
FunctionPrototypeCall,
|
||||
ObjectSetPrototypeOf,
|
||||
PromisePrototypeThen,
|
||||
PromiseResolve,
|
||||
PromiseWithResolvers,
|
||||
@@ -143,15 +144,14 @@ class ShareImpl {
|
||||
}
|
||||
|
||||
#createRawConsumer() {
|
||||
const state = {
|
||||
__proto__: null,
|
||||
const state = ObjectSetPrototypeOf({
|
||||
cursor: this.#bufferStart,
|
||||
resolve: null,
|
||||
reject: null,
|
||||
detached: false,
|
||||
error: kNoShareError,
|
||||
pendingNext: PromiseResolve(),
|
||||
};
|
||||
}, null);
|
||||
|
||||
this.#consumers.add(state);
|
||||
if (this.#consumers.size === 1) {
|
||||
@@ -549,12 +549,11 @@ class SyncShareImpl {
|
||||
}
|
||||
|
||||
#createRawConsumer() {
|
||||
const state = {
|
||||
__proto__: null,
|
||||
const state = ObjectSetPrototypeOf({
|
||||
cursor: this.#bufferStart,
|
||||
detached: false,
|
||||
error: kNoShareError,
|
||||
};
|
||||
}, null);
|
||||
|
||||
this.#consumers.add(state);
|
||||
if (this.#consumers.size === 1) {
|
||||
|
||||
Reference in new issue
Block a user