Skip to content
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.

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

streams: implement TransformStream cleanup using "transformer.cancel" #50126

Merged
merged 7 commits into from
Dec 15, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
5 changes: 3 additions & 2 deletions lib/internal/webstreams/readablestream.js
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ const {
ObjectCreate,
ObjectDefineProperties,
ObjectSetPrototypeOf,
Promise,
PromisePrototypeThen,
PromiseResolve,
PromiseReject,
Expand Down Expand Up @@ -2440,7 +2441,7 @@ function setupReadableStreamDefaultController(
const startResult = startAlgorithm();

PromisePrototypeThen(
PromiseResolve(startResult),
new Promise((r) => r(startResult)),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why this change?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

to align with the WHATWG spec, we need to do it in more places too Ref: #50126 (comment)

() => {
controller[kState].started = true;
assert(!controller[kState].pulling);
Expand Down Expand Up @@ -3239,7 +3240,7 @@ function setupReadableByteStreamController(
const startResult = startAlgorithm();

PromisePrototypeThen(
PromiseResolve(startResult),
new Promise((r) => r(startResult)),
() => {
controller[kState].started = true;
assert(!controller[kState].pulling);
Expand Down
116 changes: 105 additions & 11 deletions lib/internal/webstreams/transformstream.js
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ const {
ObjectDefineProperties,
ObjectSetPrototypeOf,
PromisePrototypeThen,
debadree25 marked this conversation as resolved.
Show resolved Hide resolved
PromiseResolve,
SymbolToStringTag,
Symbol,
} = primordials;
Expand Down Expand Up @@ -47,6 +46,7 @@ const {
nonOpFlush,
kType,
kState,
nonOpCancel,
} = require('internal/webstreams/util');

const {
Expand Down Expand Up @@ -384,8 +384,7 @@ function initializeTransformStream(
return transformStreamDefaultSourcePullAlgorithm(stream);
},
cancel(reason) {
transformStreamErrorWritableAndUnblockWrite(stream, reason);
return PromiseResolve();
return transformStreamDefaultSourceCancelAlgorithm(stream, reason);
},
}, {
highWaterMark: readableHighWaterMark,
Expand Down Expand Up @@ -427,6 +426,10 @@ function transformStreamErrorWritableAndUnblockWrite(stream, error) {
writableStreamDefaultControllerErrorIfNeeded(
writable[kState].controller,
error);
transformStreamUnblockWrite(stream);
}

function transformStreamUnblockWrite(stream) {
if (stream[kState].backpressure)
transformStreamSetBackpressure(stream, false);
}
Expand All @@ -443,13 +446,15 @@ function setupTransformStreamDefaultController(
stream,
controller,
transformAlgorithm,
flushAlgorithm) {
flushAlgorithm,
cancelAlgorithm) {
assert(isTransformStream(stream));
assert(stream[kState].controller === undefined);
controller[kState] = {
stream,
transformAlgorithm,
flushAlgorithm,
cancelAlgorithm,
};
stream[kState].controller = controller;
}
Expand All @@ -460,21 +465,26 @@ function setupTransformStreamDefaultControllerFromTransformer(
const controller = new TransformStreamDefaultController(kSkipThrow);
const transform = transformer?.transform || defaultTransformAlgorithm;
const flush = transformer?.flush || nonOpFlush;
const cancel = transformer?.cancel || nonOpCancel;
const transformAlgorithm =
FunctionPrototypeBind(transform, transformer);
const flushAlgorithm =
FunctionPrototypeBind(flush, transformer);
const cancelAlgorithm =
FunctionPrototypeBind(cancel, transformer);

setupTransformStreamDefaultController(
stream,
controller,
transformAlgorithm,
flushAlgorithm);
flushAlgorithm,
cancelAlgorithm);
}

function transformStreamDefaultControllerClearAlgorithms(controller) {
controller[kState].transformAlgorithm = undefined;
controller[kState].flushAlgorithm = undefined;
controller[kState].cancelAlgorithm = undefined;
}

function transformStreamDefaultControllerEnqueue(controller, chunk) {
Expand Down Expand Up @@ -563,7 +573,40 @@ function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
}

async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
transformStreamError(stream, reason);
const {
controller,
readable,
} = stream[kState];

if (controller[kState].finishPromise !== undefined) {
return controller[kState].finishPromise;
}

const { promise, resolve, reject } = createDeferredPromise();
controller[kState].finishPromise = promise;
const cancelPromise = ensureIsPromise(
controller[kState].cancelAlgorithm,
controller,
reason);
transformStreamDefaultControllerClearAlgorithms(controller);

PromisePrototypeThen(
cancelPromise,
() => {
if (readable[kState].state === 'errored')
reject(readable[kState].storedError);
else {
readableStreamDefaultControllerError(readable[kState].controller, reason);
resolve();
}
},
(error) => {
readableStreamDefaultControllerError(readable[kState].controller, error);
reject(error);
},
);

return controller[kState].finishPromise;
}

function transformStreamDefaultSinkCloseAlgorithm(stream) {
Expand All @@ -572,23 +615,32 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
controller,
} = stream[kState];

if (controller[kState].finishPromise !== undefined) {
return controller[kState].finishPromise;
}
const { promise, resolve, reject } = createDeferredPromise();
controller[kState].finishPromise = promise;
const flushPromise =
ensureIsPromise(
controller[kState].flushAlgorithm,
controller,
controller);
transformStreamDefaultControllerClearAlgorithms(controller);
return PromisePrototypeThen(
PromisePrototypeThen(
flushPromise,
() => {
if (readable[kState].state === 'errored')
throw readable[kState].storedError;
readableStreamDefaultControllerClose(readable[kState].controller);
reject(readable[kState].storedError);
else {
readableStreamDefaultControllerClose(readable[kState].controller);
resolve();
}
},
(error) => {
transformStreamError(stream, error);
throw readable[kState].storedError;
readableStreamDefaultControllerError(readable[kState].controller, error);
reject(error);
});
return controller[kState].finishPromise;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this mean we are adding one more promise all the time?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

indeed yes i guess, seems like spec demands!

}

function transformStreamDefaultSourcePullAlgorithm(stream) {
Expand All @@ -598,6 +650,48 @@ function transformStreamDefaultSourcePullAlgorithm(stream) {
return stream[kState].backpressureChange.promise;
}

function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {
const {
controller,
writable,
} = stream[kState];

if (controller[kState].finishPromise !== undefined) {
return controller[kState].finishPromise;
}

const { promise, resolve, reject } = createDeferredPromise();
controller[kState].finishPromise = promise;
const cancelPromise = ensureIsPromise(
controller[kState].cancelAlgorithm,
controller,
reason);
transformStreamDefaultControllerClearAlgorithms(controller);

PromisePrototypeThen(cancelPromise,
() => {
if (writable[kState].state === 'errored')
reject(writable[kState].storedError);
else {
writableStreamDefaultControllerErrorIfNeeded(
writable[kState].controller,
reason);
transformStreamUnblockWrite(stream);
resolve();
}
},
(error) => {
writableStreamDefaultControllerErrorIfNeeded(
writable[kState].controller,
error);
transformStreamUnblockWrite(stream);
reject(error);
},
);

return controller[kState].finishPromise;
}

module.exports = {
TransformStream,
TransformStreamDefaultController,
Expand Down
3 changes: 2 additions & 1 deletion lib/internal/webstreams/writablestream.js
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ const {
FunctionPrototypeCall,
ObjectDefineProperties,
ObjectSetPrototypeOf,
Promise,
PromisePrototypeThen,
PromiseResolve,
PromiseReject,
Expand Down Expand Up @@ -1295,7 +1296,7 @@ function setupWritableStreamDefaultController(
const startResult = startAlgorithm();

PromisePrototypeThen(
PromiseResolve(startResult),
new Promise((r) => r(startResult)),
() => {
assert(stream[kState].state === 'writable' ||
stream[kState].state === 'erroring');
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ Last update:
- performance-timeline: https://github.com/web-platform-tests/wpt/tree/17ebc3aea0/performance-timeline
- resource-timing: https://github.com/web-platform-tests/wpt/tree/22d38586d0/resource-timing
- resources: https://github.com/web-platform-tests/wpt/tree/1e140d63ec/resources
- streams: https://github.com/web-platform-tests/wpt/tree/517e945bbf/streams
- streams: https://github.com/web-platform-tests/wpt/tree/a8872d92b1/streams
- url: https://github.com/web-platform-tests/wpt/tree/c2d7e70b52/url
- user-timing: https://github.com/web-platform-tests/wpt/tree/5ae85bf826/user-timing
- wasm/jsapi: https://github.com/web-platform-tests/wpt/tree/cde25e7e3c/wasm/jsapi
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/piping/abort.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/recording-streams.js
// META: script=../resources/test-utils.js
'use strict';
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/recording-streams.js
'use strict';

Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/test-utils.js
// META: script=../resources/recording-streams.js
'use strict';
Expand Down
12 changes: 12 additions & 0 deletions test/fixtures/wpt/streams/piping/crashtests/cross-piping.html
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
<!DOCTYPE html>
<script type="module">
let a = new ReadableStream();
let b = self.open()
let f = new b.WritableStream();
a.pipeThrough(
{ "readable": a, "writable": f },
{ "signal": AbortSignal.abort() }
)
await new Promise(setTimeout);
structuredClone(undefined, { "transfer": [f] })
</script>
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/test-utils.js
// META: script=../resources/recording-streams.js
'use strict';
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/test-utils.js
// META: script=../resources/recording-streams.js
'use strict';
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/piping/flow-control.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/test-utils.js
// META: script=../resources/rs-utils.js
// META: script=../resources/recording-streams.js
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/piping/general-addition.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
'use strict';

promise_test(async t => {
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/piping/general.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/test-utils.js
// META: script=../resources/recording-streams.js
'use strict';
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/test-utils.js
// META: script=../resources/recording-streams.js
'use strict';
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/piping/pipe-through.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/rs-utils.js
// META: script=../resources/test-utils.js
// META: script=../resources/recording-streams.js
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/piping/then-interception.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/test-utils.js
// META: script=../resources/recording-streams.js
'use strict';
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/piping/throwing-options.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
'use strict';

class ThrowingOptions {
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/piping/transform-streams.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
'use strict';

promise_test(() => {
Expand Down
2 changes: 1 addition & 1 deletion test/fixtures/wpt/streams/queuing-strategies.any.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
'use strict';

const highWaterMarkConversions = new Map([
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
'use strict';

promise_test(() => {
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/rs-utils.js
'use strict';

Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm

promise_test(async t => {
const error = new Error('cannot proceed');
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
// META: script=../resources/rs-utils.js
// META: script=../resources/test-utils.js
'use strict';
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm
'use strict';

promise_test(async t => {
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// META: global=window,worker
// META: global=window,worker,shadowrealm

'use strict';

Expand Down
Loading