Skip to content

stream: trim per-stream costs in webstreams #66052

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
mcollina wants to merge 1 commit into
base: main
Choose a base branch
from
+156 −150
Draft

stream: trim per-stream costs in webstreams #66052

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
163 changes: 69 additions & 94 deletions lib/internal/webstreams/readablestream.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,6 +253,13 @@ class ReadableStream {
*/
constructor(source = kEmptyObject, strategy = kEmptyObject) {
markTransferMode(this, false, true);
// Internal construction (tee, transform streams, adapters, transfer):
// the caller sets up the controller, so every ReadableStream shares
// one hidden class and no per-instance prototype swap is needed.
if (source === kSkipThrow) {
this[kState] = createReadableStreamState();
return;
}
validateObject(source, 'source', kValidateObjectAllowObjects);
validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull);
this[kState] = createReadableStreamState();
Expand Down Expand Up @@ -718,22 +725,8 @@ ObjectDefineProperties(ReadableStream, {
from: kEnumerableProperty,
});

function InternalTransferredReadableStream() {
ObjectSetPrototypeOf(this, ReadableStream.prototype);
markTransferMode(this, false, true);
this[kType] = 'ReadableStream';
this[kState] = createReadableStreamState();
}

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

function TransferredReadableStream() {
const stream = new InternalTransferredReadableStream();

stream.constructor = ReadableStream;

return stream;
return new ReadableStream(kSkipThrow);
}

TransferredReadableStream.prototype[kDeserialize] = () => {};
Expand Down Expand Up @@ -1350,57 +1343,29 @@ ObjectDefineProperties(ReadableByteStreamController.prototype, {
[SymbolToStringTag]: getNonWritablePropertyDescriptor(ReadableByteStreamController.name),
});

function InternalReadableStream(start, pull, cancel, highWaterMark, size) {
ObjectSetPrototypeOf(this, ReadableStream.prototype);
markTransferMode(this, false, true);
this[kType] = 'ReadableStream';
this[kState] = createReadableStreamState();
const controller = new ReadableStreamDefaultController(kSkipThrow);
function createReadableStream(start, pull, cancel, highWaterMark = 1, size = defaultSizeAlgorithm) {
const stream = new ReadableStream(kSkipThrow);
setupReadableStreamDefaultController(
this,
controller,
stream,
new ReadableStreamDefaultController(kSkipThrow),
start,
pull,
cancel,
highWaterMark,
size);
}

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

function createReadableStream(start, pull, cancel, highWaterMark = 1, size = defaultSizeAlgorithm) {
const stream = new InternalReadableStream(start, pull, cancel, highWaterMark, size);

// For spec compliance the InternalReadableStream must be a ReadableStream
stream.constructor = ReadableStream;
return stream;
}

function InternalReadableByteStream(start, pull, cancel) {
ObjectSetPrototypeOf(this, ReadableStream.prototype);
markTransferMode(this, false, true);
this[kType] = 'ReadableStream';
this[kState] = createReadableStreamState();
const controller = new ReadableByteStreamController(kSkipThrow);
function createReadableByteStream(start, pull, cancel) {
const stream = new ReadableStream(kSkipThrow);
setupReadableByteStreamController(
this,
controller,
stream,
new ReadableByteStreamController(kSkipThrow),
start,
pull,
cancel,
0,
undefined);
}

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

function createReadableByteStream(start, pull, cancel) {
const stream = new InternalReadableByteStream(start, pull, cancel);

// For spec compliance the InternalReadableByteStream must be a ReadableStream
stream.constructor = ReadableStream;
return stream;
}

Expand Down Expand Up @@ -1630,6 +1595,13 @@ function readableStreamPipeTo(
// tells us that the promise must be rejected even
// when error is undefine.
function finalize(rejected, error) {
// The pipe is the only observer of the reader's and writer's promise
// records (including the ready hook installed by parkOnReady), and
// it is done with them: dropping them lets release skip the
// pending-promise probes and the rejections nothing would handle.
writer[kState].ready = undefined;
writer[kState].close = undefined;
reader[kState].close = undefined;
writableStreamDefaultWriterRelease(writer);
readableStreamReaderGenericRelease(reader);
if (signal !== undefined)
Expand Down Expand Up @@ -1727,12 +1699,6 @@ function readableStreamPipeTo(
PromisePrototypeThen(promise, undefined, action);
}

function watchClosed(stream, promise, action) {
if (stream[kState].state === 'closed')
action();
else
PromisePrototypeThen(promise, action, () => {});
}

// The pump loop is callback-driven to avoid per-iteration promise
// allocations. At most one read is in flight at a time, so one read
Expand Down Expand Up @@ -1863,15 +1829,34 @@ function readableStreamPipeTo(

pump();

watchErrored(source, readerClosedPromise(reader).promise, (error) => {
function onSourceErrored(error) {
if (!preventAbort) {
return shutdownWithAnAction(
() => writableStreamAbort(dest, error),
true,
error);
}
shutdown(true, error);
});
}

function onSourceClosed() {
if (!preventClose) {
return shutdownWithAnAction(
() => writableStreamDefaultWriterCloseWithErrorPropagation(writer));
}
shutdown();
}

// The spec installs the source-errored watcher before the dest-errored
// one and the source-closed watcher last; a source that is already
// errored is handled before the dest watcher is installed, and an
// already-closed source after it, as before.
if (source[kState].state === 'errored') {
onSourceErrored(source[kState].storedError);
} else if (source[kState].state !== 'closed') {
PromisePrototypeThen(
readerClosedPromise(reader).promise, onSourceClosed, onSourceErrored);
}

watchErrored(dest, writerClosedPromise(writer).promise, (error) => {
if (!preventCancel) {
Expand All @@ -1883,13 +1868,8 @@ function readableStreamPipeTo(
shutdown(true, error);
});

watchClosed(source, readerClosedPromise(reader).promise, () => {
if (!preventClose) {
return shutdownWithAnAction(
() => writableStreamDefaultWriterCloseWithErrorPropagation(writer));
}
shutdown();
});
if (source[kState].state === 'closed')
onSourceClosed();

if (writableStreamCloseQueuedOrInFlight(dest) ||
dest[kState].state === 'closed') {
Expand Down Expand Up @@ -2899,29 +2879,27 @@ function setupReadableStreamDefaultController(

const startResult = startAlgorithm();

const started = () => {
controller[kState].started = true;
assert(!controller[kState].pulling);
assert(!controller[kState].pullAgain);
readableStreamDefaultControllerCallPullIfNeeded(controller);
};

if (startResult === null ||
(typeof startResult !== 'object' && typeof startResult !== 'function')) {
// Non-thenable start result: fulfillment is guaranteed and no .then
// lookup on the result is observable, so run the post-start step
// directly at the exact microtask position the promise reaction
// would have had, skipping two promise allocations.
queueMicrotask(() => {
controller[kState].started = true;
assert(!controller[kState].pulling);
assert(!controller[kState].pullAgain);
readableStreamDefaultControllerCallPullIfNeeded(controller);
});
// lookup on the result is observable, so the post-start step runs at
// the exact microtask position the promise reaction would have had.
queueMicrotask(started);
return;
}

// The wrapper promise matches the reference implementation's
// promiseResolvedWith(), whose extra microtask hops WPT relies on.
PromisePrototypeThen(
new Promise((r) => r(startResult)),
() => {
controller[kState].started = true;
assert(!controller[kState].pulling);
assert(!controller[kState].pullAgain);
readableStreamDefaultControllerCallPullIfNeeded(controller);
},
started,
(error) => readableStreamDefaultControllerError(controller, error));
}

Expand Down Expand Up @@ -3783,26 +3761,23 @@ function setupReadableByteStreamController(

const startResult = startAlgorithm();

const started = () => {
controller[kState].started = true;
assert(!controller[kState].pulling);
assert(!controller[kState].pullAgain);
readableByteStreamControllerCallPullIfNeeded(controller);
};

// See setupReadableStreamDefaultController.
if (startResult === null ||
(typeof startResult !== 'object' && typeof startResult !== 'function')) {
// See setupReadableStreamDefaultController.
queueMicrotask(() => {
controller[kState].started = true;
assert(!controller[kState].pulling);
assert(!controller[kState].pullAgain);
readableByteStreamControllerCallPullIfNeeded(controller);
});
queueMicrotask(started);
return;
}

PromisePrototypeThen(
new Promise((r) => r(startResult)),
() => {
controller[kState].started = true;
assert(!controller[kState].pulling);
assert(!controller[kState].pullAgain);
readableByteStreamControllerCallPullIfNeeded(controller);
},
started,
(error) => readableByteStreamControllerError(controller, error));
}

Expand Down
16 changes: 9 additions & 7 deletions lib/internal/webstreams/util.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,12 +179,12 @@ class Queue {
// Single-slot entries (readable byte controller chunk records).

push(entry) {
if (this.length === this.list.length)
this.grow();
const tail = this.tail;
this.list[tail] = entry;
this.tail = (tail + 1) & this.capacityMask;
this.length++;
if (this.tail === this.head)
this.grow();
}

shift() {
Expand All @@ -207,14 +207,14 @@ class Queue {
// never need to wrap.

pushPair(value, size) {
if (this.length * 2 === this.list.length)
this.grow();
const tail = this.tail;
const list = this.list;
list[tail] = value;
list[tail + 1] = size;
this.tail = (tail + 2) & this.capacityMask;
this.length++;
if (this.tail === this.head)
this.grow();
}

// Returns the dequeued value; the size of the same entry is left in
Expand All @@ -237,9 +237,11 @@ class Queue {
return this.list[this.head];
}

// The ring is completely full (the post-push tail caught up with the
// head): double the capacity, re-linearizing from the head so index
// arithmetic stays trivial.
// The ring is completely full (the tail has caught up with the head, so
// the next push would overwrite the oldest entry): double the capacity,
// re-linearizing from the head so index arithmetic stays trivial.
// Growing before the push rather than after it lets the initial 8-slot
// ring hold four (value, size) pairs without reallocating.
grow() {
const list = this.list;
const capacity = list.length;
Expand Down
Loading