diff --git a/benchmark/webstreams/pipe-through.js b/benchmark/webstreams/pipe-through.js new file mode 100644 index 000000000000..8af088f4eed1 --- /dev/null +++ b/benchmark/webstreams/pipe-through.js @@ -0,0 +1,38 @@ +'use strict'; +const common = require('../common.js'); +const { + ReadableStream, + TransformStream, +} = require('node:stream/web'); + +const bench = common.createBenchmark(main, { + n: [5e5], + kind: ['default', 'transform'], +}); + +async function main({ n, kind }) { + const b = Buffer.alloc(64); + let i = 0; + const rs = new ReadableStream({ + pull(controller) { + if (i++ < n) { + controller.enqueue(b); + } else { + controller.close(); + } + }, + }); + const ts = kind === 'default' ? + new TransformStream() : + new TransformStream({ + transform(chunk, controller) { controller.enqueue(chunk); }, + }); + + const reader = rs.pipeThrough(ts).getReader(); + bench.start(); + for (;;) { + const { done } = await reader.read(); + if (done) break; + } + bench.end(n); +} diff --git a/benchmark/webstreams/pipe-to.js b/benchmark/webstreams/pipe-to.js index 38324cd20822..e902f67a9887 100644 --- a/benchmark/webstreams/pipe-to.js +++ b/benchmark/webstreams/pipe-to.js @@ -7,8 +7,8 @@ const { const bench = common.createBenchmark(main, { n: [5e5], - highWaterMarkR: [512, 1024, 2048, 4096], - highWaterMarkW: [512, 1024, 2048, 4096], + highWaterMarkR: [1, 1024, 4096], + highWaterMarkW: [1, 1024, 4096], }); @@ -16,7 +16,6 @@ async function main({ n, highWaterMarkR, highWaterMarkW }) { const b = Buffer.alloc(1024); let i = 0; const rs = new ReadableStream({ - highWaterMark: highWaterMarkR, pull: function(controller) { if (i++ < n) { controller.enqueue(b); @@ -24,12 +23,11 @@ async function main({ n, highWaterMarkR, highWaterMarkW }) { controller.close(); } }, - }); + }, { highWaterMark: highWaterMarkR }); const ws = new WritableStream({ - highWaterMark: highWaterMarkW, write(chunk, controller) {}, close() { bench.end(n); }, - }); + }, { highWaterMark: highWaterMarkW }); bench.start(); rs.pipeTo(ws); diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index e1e80eb953c0..d102a2aaaf94 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -101,6 +101,7 @@ const { cloneAsUint8Array, copyArrayBuffer, createPromiseCallback1Param, + createRawCallback1Param, customInspect, defaultSizeAlgorithm, dequeueValue, @@ -110,17 +111,18 @@ const { getNonWritablePropertyDescriptor, isBrandCheck, kEmptyQueue, + kResolvedPromise, kState, kType, lazyTransfer, materializeQueue, + nonOpCallback, nonOpCancel, - nonOpPull, - nonOpStart, rejectedHandledRecord, resetQueue, resolvedRecord, setPromiseHandled, + thenAlgorithmResult, } = require('internal/webstreams/util'); const { @@ -137,7 +139,6 @@ const { writableStreamDefaultWriterRelease, writableStreamDefaultWriterWriteWithRequest, writerClosedPromise, - writerReadyPromise, } = require('internal/webstreams/writablestream'); const { Buffer } = require('buffer'); @@ -1455,7 +1456,7 @@ function readableStreamFromIterable(iterable) { if (iterator === null || (typeof iterator !== 'object' && typeof iterator !== 'function')) { throw new ERR_INVALID_STATE.TypeError('The iterator method must return an object'); } - const startAlgorithm = nonOpStart; + const startAlgorithm = nonOpCallback; async function pullAlgorithm() { const iterResult = await iterator.next(); @@ -1674,11 +1675,31 @@ function readableStreamPipeTo( // the chunk travels through `pendingChunk`. let pendingChunk; let readRequest; + let readyHook; // Ready promise rejection is handled by the destination-errored // watcher. function ignoreReadyRejection() {} + // Parks the pump on the destination's backpressure by installing a + // record that duck-types the writer's lazily-materialized + // [[readyPromise]] record: writableStreamUpdateBackpressure resolves it + // when backpressure clears (after publishing the new backpressure + // state), which re-enters the pump directly instead of rotating a + // fresh promise record plus reaction per flip. The pipe holds the only + // reference to the writer, so the record is never observable as a real + // ready promise; the erroring/release paths probe `promise` via + // isPromisePending() and call `reject`, so it carries a real + // forever-pending promise and a no-op reject. + function parkOnReady() { + readyHook ??= { + promise: PromiseWithResolvers().promise, + resolve: pump, + reject: ignoreReadyRejection, + }; + writer[kState].ready = readyHook; + } + function forwardChunk() { const chunk = pendingChunk; pendingChunk = undefined; @@ -1690,10 +1711,7 @@ function readableStreamPipeTo( if (shuttingDown) return; if (dest[kState].backpressure) { - PromisePrototypeThen( - writerReadyPromise(writer).promise, - pump, - ignoreReadyRejection); + parkOnReady(); return; } @@ -1738,9 +1756,18 @@ function readableStreamPipeTo( return; } - // Yield to microtask queue between batches to allow events/signals - // to fire - queueMicrotask(pump); + // Park on backpressure directly: the ready hook resumes the pump + // when a completed write clears it. + if (dest[kState].backpressure) { + parkOnReady(); + return; + } + + // Yield to the microtask queue between batches so completed-write + // reactions and events/signals fire; a shared resolved promise + // enqueues the continuation at the same position as queueMicrotask + // without the per-batch scheduling overhead. + PromisePrototypeThen(kResolvedPromise, pump); return; } @@ -1752,7 +1779,7 @@ function readableStreamPipeTo( // synchronous write during enqueue(). See WHATWG Streams spec // "ReadableStreamPipeTo" step 15's "chunk steps". pendingChunk = chunk; - queueMicrotask(forwardChunk); + PromisePrototypeThen(kResolvedPromise, forwardChunk); }, [kClose]() {}, [kError]() {}, @@ -1867,7 +1894,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) { // The microtask is required by the spec (ReadableStreamTee's // "chunk steps" queue one). pendingChunk = value; - queueMicrotask(forwardChunk); + PromisePrototypeThen(kResolvedPromise, forwardChunk); }, [kClose]() { // The `process.nextTick()` is not part of the spec. @@ -1911,9 +1938,9 @@ function readableStreamDefaultTee(stream, cloneForBranch2) { } branch1 = - createReadableStream(nonOpStart, pullAlgorithm, cancel1Algorithm); + createReadableStream(nonOpCallback, pullAlgorithm, cancel1Algorithm); branch2 = - createReadableStream(nonOpStart, pullAlgorithm, cancel2Algorithm); + createReadableStream(nonOpCallback, pullAlgorithm, cancel2Algorithm); PromisePrototypeThen( readerClosedPromise(reader).promise, @@ -2020,7 +2047,7 @@ function readableByteStreamTee(stream) { defaultReadRequest ??= { [kChunk](chunk) { pendingChunk = chunk; - queueMicrotask(forwardChunk); + PromisePrototypeThen(kResolvedPromise, forwardChunk); }, [kClose]() { reading = false; @@ -2199,9 +2226,9 @@ function readableByteStreamTee(stream) { } branch1 = - createReadableByteStream(nonOpStart, pull1Algorithm, cancel1Algorithm); + createReadableByteStream(nonOpCallback, pull1Algorithm, cancel1Algorithm); branch2 = - createReadableByteStream(nonOpStart, pull2Algorithm, cancel2Algorithm); + createReadableByteStream(nonOpCallback, pull2Algorithm, cancel2Algorithm); forwardReaderError(reader); @@ -2709,8 +2736,18 @@ function readableStreamDefaultControllerPull(controller) { controller[kState].pullRejected = (error) => readableStreamDefaultControllerError(controller, error); } - PromisePrototypeThen( - controller[kState].pullAlgorithm(controller), + // The pull algorithm may be a raw callback (a wrapped user source.pull + // returns its result uncoerced; a synchronous throw surfaces here) or an + // internal algorithm that always returns a promise; thenAlgorithmResult + // handles both. + let result; + try { + result = controller[kState].pullAlgorithm(controller); + } catch (error) { + result = PromiseReject(error); + } + thenAlgorithmResult( + result, controller[kState].pullFulfilled, controller[kState].pullRejected); } @@ -2834,10 +2871,10 @@ function setupReadableStreamDefaultControllerFromSource( const cancel = source?.cancel; const startAlgorithm = start ? FunctionPrototypeBind(start, source, controller) : - nonOpStart; + nonOpCallback; const pullAlgorithm = pull ? - createPromiseCallback1Param('source.pull', pull, source) : - nonOpPull; + createRawCallback1Param('source.pull', pull, source) : + nonOpCallback; const cancelAlgorithm = cancel ? createPromiseCallback1Param('source.cancel', cancel, source) : nonOpCancel; @@ -3529,8 +3566,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) { controller[kState].pullRejected = (error) => readableByteStreamControllerError(controller, error); } - PromisePrototypeThen( - controller[kState].pullAlgorithm(controller), + // See readableStreamDefaultControllerPull for the raw-callback contract. + let result; + try { + result = controller[kState].pullAlgorithm(controller); + } catch (error) { + result = PromiseReject(error); + } + thenAlgorithmResult( + result, controller[kState].pullFulfilled, controller[kState].pullRejected); } @@ -3708,10 +3752,10 @@ function setupReadableByteStreamControllerFromSource( const autoAllocateChunkSize = source?.autoAllocateChunkSize; const startAlgorithm = start ? FunctionPrototypeBind(start, source, controller) : - nonOpStart; + nonOpCallback; const pullAlgorithm = pull ? - createPromiseCallback1Param('source.pull', pull, source) : - nonOpPull; + createRawCallback1Param('source.pull', pull, source) : + nonOpCallback; const cancelAlgorithm = cancel ? createPromiseCallback1Param('source.cancel', cancel, source) : nonOpCancel; diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 535c783a3a31..30b7b1c8fac1 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -5,6 +5,8 @@ const { ObjectDefineProperties, ObjectSetPrototypeOf, PromisePrototypeThen, + PromiseReject, + PromiseResolve, PromiseWithResolvers, Symbol, SymbolToStringTag, @@ -44,12 +46,14 @@ const { const { createPromiseCallback1Param, - createPromiseCallback2Params, + createRawCallback2Params, customInspect, extractHighWaterMark, extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + kParkedAlgorithmResult, + kResolvedPromise, kState, kType, nonOpCancel, @@ -258,7 +262,10 @@ function InternalTransferredTransformStream() { readable: undefined, writable: undefined, backpressure: undefined, - backpressureChange: undefined, + pullPending: false, + pendingWrite: undefined, + pendingWriteChunk: undefined, + writeContinuation: undefined, controller: undefined, }; } @@ -348,7 +355,9 @@ const isTransformStream = const isTransformStreamDefaultController = isBrandCheck('TransformStreamDefaultController'); -async function defaultTransformAlgorithm(chunk, controller) { +// Raw callback (see createRawCallback*): invoked inside the try/catch of +// transformStreamDefaultControllerPerformTransform. +function defaultTransformAlgorithm(chunk, controller) { transformStreamDefaultControllerEnqueue(controller, chunk); } @@ -385,7 +394,12 @@ function initializeTransformStream( writable, controller: undefined, backpressure: undefined, - backpressureChange: undefined, + // Continuation slots replacing the spec's + // [[backpressureChangePromise]]; see transformStreamSetBackpressure. + pullPending: false, + pendingWrite: undefined, + pendingWriteChunk: undefined, + writeContinuation: undefined, }; transformStreamSetBackpressure(stream, true); @@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) { // The spec's [[backpressureChangePromise]] is only ever observed by the // source pull algorithm (settles when backpressure next becomes true) and // by a sink write arriving while backpressure is set (settles when -// backpressure next becomes false). Instead of allocating a fresh promise -// record on every flip, the record is materialized lazily on first -// observation and dropped once settled; flips nobody is waiting on -// allocate nothing. -function transformStreamBackpressureChangePromise(stream) { - const state = stream[kState]; - return (state.backpressureChange ??= PromiseWithResolvers()).promise; -} - +// backpressure next becomes false). Both observers are internal, so the +// promise record is replaced by continuation slots: a parked pull is +// completed by delivering the readable controller's pull-fulfilled step, +// and a parked write by the cached write continuation (see +// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the +// shared resolved promise at the exact microtask position the old +// record's reaction would have had. function transformStreamSetBackpressure(stream, backpressure) { const state = stream[kState]; assert(state.backpressure !== backpressure); - const backpressureChange = state.backpressureChange; - if (backpressureChange !== undefined) { - state.backpressureChange = undefined; - backpressureChange.resolve(); - } state.backpressure = backpressure; + if (backpressure) { + if (state.pullPending) { + state.pullPending = false; + // The pull-fulfilled step exists: a pull parked it (see + // transformStreamDefaultSourcePullAlgorithm), and the readable + // controller creates it before invoking the pull algorithm. + PromisePrototypeThen( + kResolvedPromise, + state.readable[kState].controller[kState].pullFulfilled); + } + } else if (state.pendingWrite !== undefined) { + PromisePrototypeThen(kResolvedPromise, state.writeContinuation); + } } function setupTransformStreamDefaultController( @@ -456,6 +476,7 @@ function setupTransformStreamDefaultController( transformAlgorithm, flushAlgorithm, cancelAlgorithm, + performTransformRejected: undefined, }; stream[kState].controller = controller; } @@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer( const flush = transformer?.flush; const cancel = transformer?.cancel; const transformAlgorithm = transform ? - createPromiseCallback2Params('transformer.transform', transform, transformer) : + createRawCallback2Params('transformer.transform', transform, transformer) : defaultTransformAlgorithm; const flushAlgorithm = flush ? createPromiseCallback1Param('transformer.flush', flush, transformer) : @@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) { transformStreamError(controller[kState].stream, error); } -async function transformStreamDefaultControllerPerformTransform(controller, chunk) { +// Mirrors the reference implementation's +// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`: +// the returned promise settles one microtask after the (coerced) result +// does, and a rejection errors the transform stream before propagating. +// The raw transform callback plus the shared resolved promise for +// non-thenable results replace the previous async wrapper's two implicit +// promises per chunk. +function transformStreamDefaultControllerPerformTransform(controller, chunk) { + const controllerState = controller[kState]; + const transformAlgorithm = controllerState.transformAlgorithm; + if (transformAlgorithm === undefined) { + // Algorithms were cleared by a concurrent cancel/abort/close. + return kResolvedPromise; + } + let result; try { - const transformAlgorithm = controller[kState].transformAlgorithm; - if (transformAlgorithm === undefined) { - // Algorithms were cleared by a concurrent cancel/abort/close. - return; - } - return await transformAlgorithm(chunk, controller); + result = transformAlgorithm(chunk, controller); } catch (error) { + result = PromiseReject(error); + } + if (result === null || + (typeof result !== 'object' && typeof result !== 'function')) { + result = kResolvedPromise; + } else { + result = PromiseResolve(result); + } + controllerState.performTransformRejected ??= (error) => { transformStreamError(controller[kState].stream, error); throw error; - } + }; + return PromisePrototypeThen( + result, + undefined, + controllerState.performTransformRejected); } function transformStreamDefaultControllerTerminate(controller) { @@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) { } function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) { + const state = stream[kState]; const { writable, controller, - } = stream[kState]; + } = state; assert(writable[kState].state === 'writable'); - if (stream[kState].backpressure) { - const backpressureChange = transformStreamBackpressureChangePromise(stream); - return PromisePrototypeThen( - backpressureChange, - () => { - const { - writable, - } = stream[kState]; - if (writable[kState].state === 'erroring') - throw writable[kState].storedError; - assert(writable[kState].state === 'writable'); - return transformStreamDefaultControllerPerformTransform( + if (state.backpressure) { + // Park the chunk and one promise record; the backpressure -> false + // flip delivers the cached continuation (see + // transformStreamSetBackpressure) at the same microtask position as + // the old [[backpressureChangePromise]] reaction. The continuation + // resolves the sink promise with the perform-transform promise, so + // adoption reproduces the old derived-chain settle depth exactly. + // The writable dispatches a single write at a time, so one pending + // slot suffices. + assert(state.pendingWrite === undefined); + const pendingWrite = PromiseWithResolvers(); + state.pendingWrite = pendingWrite; + state.pendingWriteChunk = chunk; + state.writeContinuation ??= () => { + const pending = state.pendingWrite; + const pendingChunk = state.pendingWriteChunk; + state.pendingWrite = undefined; + state.pendingWriteChunk = undefined; + const writableState = state.writable[kState]; + if (writableState.state === 'erroring') { + pending.reject(writableState.storedError); + return; + } + assert(writableState.state === 'writable'); + pending.resolve( + transformStreamDefaultControllerPerformTransform( controller, - chunk); - }); + pendingChunk)); + }; + return pendingWrite.promise; } return transformStreamDefaultControllerPerformTransform(controller, chunk); } @@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) { } function transformStreamDefaultSourcePullAlgorithm(stream) { - assert(stream[kState].backpressure); + const state = stream[kState]; + assert(state.backpressure); transformStreamSetBackpressure(stream, false); - return transformStreamBackpressureChangePromise(stream); + // Park the pull: the next backpressure -> true flip delivers the + // pull-fulfilled step (see transformStreamSetBackpressure). The old + // [[backpressureChangePromise]] this replaces was only ever resolved, + // so the parked pull needs no rejection delivery. + state.pullPending = true; + return kParkedAlgorithmResult; } function transformStreamDefaultSourceCancelAlgorithm(stream, reason) { diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index 8bc4c02be31e..9598796f35c8 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -338,6 +338,48 @@ function createPromiseCallbackNoParams(name, fn, thisArg) { return async () => FunctionPrototypeCall(fn, thisArg); } +// Raw variants that skip the async wrapper's implicit result promise. +// Consumers of a raw callback invoke it inside try/catch and route the +// result through thenAlgorithmResult() below. +function createRawCallback1Param(name, fn, thisArg) { + validateFunction(fn, name); + return (arg) => FunctionPrototypeCall(fn, thisArg, arg); +} + +function createRawCallback2Params(name, fn, thisArg) { + validateFunction(fn, name); + return (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2); +} + +// A single shared, forever-resolved promise used to enqueue a reaction at +// the next microtask checkpoint without allocating a fresh promise. +const kResolvedPromise = PromiseResolve(); + +// Returned by an internal algorithm to signal that it parked the +// operation and takes responsibility for delivering the fulfilled (or +// rejected) continuation itself later, instead of settling a promise +// (see the transform stream source pull algorithm). +const kParkedAlgorithmResult = { __proto__: null }; + +// Wires the (possibly non-thenable) result of an underlying algorithm +// callback to its fulfilled/rejected continuations. A non-thenable result +// means fulfillment is guaranteed and no then() lookup is observable, so +// the fulfillment step is enqueued directly at the exact microtask +// position the coerced promise's reaction would have had, skipping the +// per-chunk promise allocation. For thenable results PromiseResolve() +// matches the spec's "a promise resolved with" conversion (identity for +// native promises). +function thenAlgorithmResult(result, onFulfilled, onRejected) { + if (result === kParkedAlgorithmResult) + return; + if (result === null || + (typeof result !== 'object' && typeof result !== 'function')) { + PromisePrototypeThen(kResolvedPromise, onFulfilled); + } else { + PromisePrototypeThen(PromiseResolve(result), onFulfilled, onRejected); + } +} + function createPromiseCallback1Param(name, fn, thisArg) { validateFunction(fn, name); return async (arg) => FunctionPrototypeCall(fn, thisArg, arg); @@ -384,14 +426,14 @@ function setPromiseHandled(promise) { async function nonOpFlush() {} -function nonOpStart() {} - -async function nonOpPull() {} +// Shared non-op for the start/pull/write algorithm callbacks, which all +// follow the raw-callback contract (see createRawCallback*): the +// non-thenable return takes the allocation-free fast path in +// thenAlgorithmResult(). +function nonOpCallback() {} async function nonOpCancel() {} -async function nonOpWrite() {} - let transfer; function lazyTransfer() { if (transfer === undefined) @@ -411,6 +453,8 @@ module.exports = { createPromiseCallbackNoParams, createPromiseCallback1Param, createPromiseCallback2Params, + createRawCallback1Param, + createRawCallback2Params, customInspect, defaultSizeAlgorithm, dequeueValue, @@ -421,18 +465,20 @@ module.exports = { isBrandCheck, isPromisePending, kEmptyQueue, + kParkedAlgorithmResult, + kResolvedPromise, kState, kType, lazyTransfer, materializeQueue, + nonOpCallback, nonOpCancel, nonOpFlush, - nonOpPull, - nonOpStart, - nonOpWrite, + peekQueueValue, rejectedHandledRecord, resetQueue, resolvedRecord, setPromiseHandled, + thenAlgorithmResult, }; diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 10b7dbcf277c..1e9ca02cfe96 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -56,7 +56,7 @@ const { Queue, createPromiseCallbackNoParams, createPromiseCallback1Param, - createPromiseCallback2Params, + createRawCallback2Params, customInspect, defaultSizeAlgorithm, dequeueValue, @@ -70,14 +70,14 @@ const { kState, kType, lazyTransfer, + nonOpCallback, nonOpCancel, - nonOpStart, - nonOpWrite, peekQueueValue, rejectedHandledRecord, resetQueue, resolvedRecord, setPromiseHandled, + thenAlgorithmResult, } = require('internal/webstreams/util'); const { @@ -769,7 +769,12 @@ function writableStreamUpdateBackpressure(controller, streamState) { const backpressure = controllerState.highWaterMark - controllerState.queueTotalSize <= 0; const writer = streamState.writer; - if (writer !== undefined && streamState.backpressure !== backpressure) { + const changed = streamState.backpressure !== backpressure; + // The state field is published before the ready record is resolved so + // that a ready resolve hook (pipeTo's pump continuation) observes the + // new value. + streamState.backpressure = backpressure; + if (writer !== undefined && changed) { if (backpressure) { // The spec replaces [[readyPromise]] with a fresh pending promise; // dropping the cache lets the next observation derive it. @@ -778,7 +783,6 @@ function writableStreamUpdateBackpressure(controller, streamState) { writer[kState].ready?.resolve(); } } - streamState.backpressure = backpressure; } function writableStreamStartErroring(stream, reason) { @@ -1197,8 +1201,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { }; } - PromisePrototypeThen( - writeAlgorithm(chunk, controller), + // The write algorithm may be a raw callback (a wrapped user sink.write + // returns its result uncoerced; a synchronous throw surfaces here) or an + // internal algorithm that always returns a promise; thenAlgorithmResult + // handles both. + let result; + try { + result = writeAlgorithm(chunk, controller); + } catch (error) { + result = PromiseReject(error); + } + thenAlgorithmResult( + result, controller[kState].writeFulfilled, controller[kState].writeRejected); } @@ -1321,10 +1335,10 @@ function setupWritableStreamDefaultControllerFromSink( const abort = sink?.abort; const startAlgorithm = start ? FunctionPrototypeBind(start, sink, controller) : - nonOpStart; + nonOpCallback; const writeAlgorithm = write ? - createPromiseCallback2Params('sink.write', write, sink) : - nonOpWrite; + createRawCallback2Params('sink.write', write, sink) : + nonOpCallback; const closeAlgorithm = close ? createPromiseCallbackNoParams('sink.close', close, sink) : nonOpCancel;