diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx index 060a0bf95..cfd52e054 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -22,8 +22,9 @@ `Readable.from` accepts `objectMode` and `highWaterMark` options. The `Readable` operators (`map`, `filter`, `forEach`, `toArray`, `some`, `find`, `every`, `flatMap`, `drop`, `take`, and `reduce`) accept `concurrency`, `highWaterMark`, and `signal` options; their callbacks take only the chunk, `map` returns strings or Buffers, and `flatMap` returns strings or arrays of them. `Duplex.from` accepts an async generator function, which reads what is written to the Duplex, a `Duplex`, `Transform` or `PassThrough`, which it returns, and the values `Readable.from` accepts, which give a Duplex that is not writable. - A lone `Readable` or `Writable`, a promise, a `{ readable, writable }` pair, a web stream, and a body function that declares the options argument are unsupported in `Duplex.from`. - `stream.compose` accepts streams and async generator functions in any position, and an array, a string, or a generator as the first stage; one stage is `Duplex.from` of it. + It also accepts a lone `Readable` or `Writable` and a `{ readable, writable }` pair, which the Duplex reads and writes through, a promise, whose value it reads, and an async function, which reads what is written and must resolve to `null` or `undefined`. + A web stream and a body function that declares the options argument are unsupported in `Duplex.from`. + `stream.compose` accepts streams and async generator functions in any position, an async function as the last stage, and an array, a string, or a generator as the first stage; one stage is `Duplex.from` of it. Web streams as compose stages are unsupported, and the `ERR_INVALID_ARG_VALUE` message for a stream that is no longer readable or writable names the stream's class instead of printing the stream. `readable.compose(stage, { signal })` is `stream.compose(readable, stage)` with the signal added; the options are an object literal whose `signal` is a variable or a property read. `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, with or without function stages, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags. diff --git a/packages/cli/test/native-link-info.test.ts b/packages/cli/test/native-link-info.test.ts index 484b03e8b..1f2fb241a 100644 --- a/packages/cli/test/native-link-info.test.ts +++ b/packages/cli/test/native-link-info.test.ts @@ -72,7 +72,7 @@ describe.runIf(supported)("macOS arm64 native link info", () => { minimum_os: "14.0", }, program: { object, entry_symbol: "main" }, - runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" }, + runtime_abi: { version: 30, marker: "scr_runtime_abi_v30" }, runtime_pack: { kind: "precompiled", package: "@scriptc/runtime-darwin-arm64", @@ -165,6 +165,6 @@ describe.runIf(supported)("macOS arm64 native link info", () => { () => null, (failure: { stderr?: string }) => failure, ); - expect(error?.stderr).toContain("scr_runtime_abi_v29"); + expect(error?.stderr).toContain("scr_runtime_abi_v30"); }); }); diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index 94318c3aa..a790ad197 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -1129,6 +1129,11 @@ export const LIB_FN_SYMS: Record = { "stream.decodeStrings": "scr_stream_decode_strings", "stream.duplexFromGen": "scr_stream_duplex_from_gen", "stream.duplexFromDyn": "scr_stream_duplex_from_dyn", + "stream.duplexifyReadable": "scr_stream_duplexify_readable", + "stream.duplexifyWritable": "scr_stream_duplexify_writable", + "stream.duplexifyPair": "scr_stream_duplexify_pair", + "stream.duplexFromPromise": "scr_stream_duplex_from_promise", + "stream.duplexFromAsyncFn": "scr_stream_duplex_from_async_fn", "writable.writeEncoding": "scr_stream_write_encoding", "writable.setDefaultEncoding": "scr_stream_set_default_encoding", "stream.chunkDone": "scr_stream_chunk_done", @@ -1737,6 +1742,11 @@ export const USES_TIMERS_LIB_FNS = new Set([ "stream.setConstruct", "stream.duplexFromGen", "stream.duplexFromDyn", + "stream.duplexifyReadable", + "stream.duplexifyWritable", + "stream.duplexifyPair", + "stream.duplexFromPromise", + "stream.duplexFromAsyncFn", "stream.setWritev", "stream.setWritevDefault", "process.activeResources", diff --git a/packages/compiler/src/backend/runtime-abi.ts b/packages/compiler/src/backend/runtime-abi.ts index 948fa6359..fcc8d160e 100644 --- a/packages/compiler/src/backend/runtime-abi.ts +++ b/packages/compiler/src/backend/runtime-abi.ts @@ -2,8 +2,8 @@ * of the link contract: a program object keeps this symbol undefined and the * matching runtime defines it, so incompatible manual links fail before the * program can start. */ -export const RUNTIME_ABI_VERSION = 29 as const; -export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v29" as const; +export const RUNTIME_ABI_VERSION = 30 as const; +export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v30" as const; /** Object consumption is useful today, but the complete scr_* surface may * still change before 1.0. The versioned marker prevents accidental mixing; diff --git a/packages/compiler/src/backend/runtime-pack-targets.test.ts b/packages/compiler/src/backend/runtime-pack-targets.test.ts index a97769328..58e2b4855 100644 --- a/packages/compiler/src/backend/runtime-pack-targets.test.ts +++ b/packages/compiler/src/backend/runtime-pack-targets.test.ts @@ -52,7 +52,7 @@ async function fixture(target: NativeTargetSpec, workers = false) { object_format: target.objectFormat, minimum_os: target.minimumOs, }, - runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" }, + runtime_abi: { version: 30, marker: "scr_runtime_abi_v30" }, compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple }, macros: { executable: [], excluded: ["SCR_LIB"], sanitizer: "external-toolchain-required" }, flavors: { diff --git a/packages/compiler/src/backend/runtime-pack.test.ts b/packages/compiler/src/backend/runtime-pack.test.ts index 85038d8c5..b898a3e62 100644 --- a/packages/compiler/src/backend/runtime-pack.test.ts +++ b/packages/compiler/src/backend/runtime-pack.test.ts @@ -127,7 +127,7 @@ async function fixture(target: NativeTargetSpec = MACOS_ARM64_TARGET) { object_format: target.objectFormat, minimum_os: target.minimumOs, }, - runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" }, + runtime_abi: { version: 30, marker: "scr_runtime_abi_v30" }, compiler: { command: "clang", identity: "fixture clang", diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index 6aa8aef96..7c0c2b46c 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -3772,11 +3772,11 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE if (fn !== null) { const isAsync = (fn.modifiers?.nodes ?? []).some((m) => m.kind === ts.SyntaxKind.AsyncKeyword); const isGen = !ts.isArrowFunction(fn) && fn.asteriskToken !== undefined; - if (!isAsync || !isGen) { + if (!isAsync) { lowerer.noLowering( - "Duplex.from over a function that is not an async generator function", + "Duplex.from over a function that is not async", node, - "write the body as async function* (source) { for await (const chunk of source) ... }", + "write the body as async function* (source) { ... } or async function (source) { ... }", ); } if (fn.parameters.nodes.length > 1) { @@ -3794,6 +3794,34 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE nextT: UNDEFINED_T, }; const lowered = lowerStageFunction(lowerer, fn, sourceT); + if (!isGen) { + // an async function: a writable Duplexify whose final waits for + // the function's promise + const asyncSource = duplexSourceHelper(lowerer, sourceT, loc); + const out = lowered.out; + if (out.kind !== "promise" || !asyncSource) + return lowerer.noLowering("Duplex.from over a function with this return type", node); + if ( + out.inner.kind !== "void" && + !canConvertToDyn( + out.inner, + (id) => lowerer.shapes.get(id), + (id) => lowerer.unions.get(id), + ) + ) { + lowerer.noLowering( + `Duplex.from over a function resolving to '${lowerer.fmt(out.inner)}'`, + node, + ); + } + return { + kind: "libCall", + fn: "stream.duplexFromAsyncFn", + args: [lowered.closure, asyncSource, pipelineBoxHelper(lowerer, out, loc)], + type: duplex, + loc, + }; + } const helpers = lowered.out.kind === "generator" ? fromGenHelpers(lowerer, lowered.out, loc) : null; const source = duplexSourceHelper(lowerer, sourceT, loc); @@ -3818,6 +3846,45 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE } const bodySides = bodyT?.kind === "object" ? streamSidesOf(lowerer, lowerer.classes.get(bodyT.className)) : null; + let literal: ts.Expression = node; + while (ts.isParenthesizedExpression(literal)) literal = literal.expression; + if (bodySides === null && ts.isObjectLiteralExpression(literal)) + return duplexFromPair(lowerer, literal, loc); + if (bodyT?.kind === "promise") { + // a Duplexify that reads the promise's value + const valueT = bodyT.inner; + if ( + valueT.kind !== "void" && + !canConvertToDyn( + valueT, + (id) => lowerer.shapes.get(id), + (id) => lowerer.unions.get(id), + ) + ) { + lowerer.noLowering(`Duplex.from over a promise of '${lowerer.fmt(valueT)}'`, node); + } + const promise = lowerer.lowerExpr(node); + if (promise.type.kind !== "promise") { + // a checked-dynamic promise (an untyped function's result) + return { + kind: "libCall", + fn: "stream.duplexFromDyn", + args: [lowerer.coerceToExpected(promise, DYN)], + type: { kind: "object", className: "%Duplex" }, + loc, + }; + } + return { + kind: "libCall", + fn: "stream.duplexFromPromise", + args: [ + promise, + pipelineBoxHelper(lowerer, promise.type as IrType & { kind: "promise" }, loc), + ], + type: { kind: "object", className: "%Duplex" }, + loc, + }; + } if (bodySides === null) return lowerFromSource(lowerer, node, node, undefined, true); let stream: IrExpr; try { @@ -3830,16 +3897,88 @@ function duplexFromBody(lowerer: Lowerer, node: ts.Expression, loc: SrcLoc): IrE stream.type.kind === "object" ? streamSidesOf(lowerer, lowerer.classes.get(stream.type.className)) : null; + // _duplexify({ readable }) / _duplexify({ writable }) + if (sides === "r" || sides === "w") { + return { + kind: "libCall", + fn: sides === "r" ? "stream.duplexifyReadable" : "stream.duplexifyWritable", + args: [stream], + type: duplex, + loc, + }; + } if (sides !== "rw") { lowerer.noLowering( - `Duplex.from over a ${sides === "r" ? "Readable" : sides === "w" ? "Writable" : "value of this type"}`, + "Duplex.from over a value of this type", node, - "an async generator function, an array, a string, a generator, an async generator, or a Duplex, Transform or PassThrough is the lowered body", + "an async generator function, an iterable, a stream, or a { readable, writable } pair is the lowered body", ); } return stream; } +/** Duplex.from({ readable, writable }): each member that is a stream with + * that side is used as it is, any other member is Duplex.from of it, and + * the pair reads and writes through _duplexify (duplexify.js). The + * members evaluate in source order. */ +function duplexFromPair( + lowerer: Lowerer, + literal: ts.ObjectLiteralExpression, + loc: SrcLoc, +): IrExpr { + const duplex: IrType = { kind: "object", className: "%Duplex" }; + const sides: { readable?: IrExpr; writable?: IrExpr } = {}; + for (const p of literal.properties.nodes) { + const key = + (ts.isPropertyAssignment(p) || ts.isShorthandPropertyAssignment(p)) && ts.isIdentifier(p.name) + ? p.name.text + : null; + if (key !== "readable" && key !== "writable") { + lowerer.noLowering( + "Duplex.from over an object with members other than readable and writable", + p, + "pass { readable, writable } with streams or Duplex.from bodies", + ); + } + const value = ts.isShorthandPropertyAssignment(p) + ? (p.name as ts.Identifier) + : (p as ts.PropertyAssignment).initializer; + let valueT: IrType | null; + try { + valueT = lowerer.mapTypeOf(lowerer.typeOf(value)); + } catch { + valueT = null; + } + const valueSides = + valueT?.kind === "object" + ? streamSidesOf(lowerer, lowerer.classes.get(valueT.className)) + : null; + const wanted = key === "readable" ? "r" : "w"; + const direct = valueSides === "rw" || valueSides === wanted; + sides[key!] = direct + ? lowerStreamArg(lowerer, value, "compose") + : duplexFromBody(lowerer, value, loc); + } + const { readable, writable } = sides; + if (readable && writable) + return { + kind: "libCall", + fn: "stream.duplexifyPair", + args: [readable, writable], + type: duplex, + loc, + }; + if (readable) + return { kind: "libCall", fn: "stream.duplexifyReadable", args: [readable], type: duplex, loc }; + if (writable) + return { kind: "libCall", fn: "stream.duplexifyWritable", args: [writable], type: duplex, loc }; + return lowerer.noLowering( + "Duplex.from over an object with neither readable nor writable", + literal, + "pass { readable, writable } with streams or Duplex.from bodies", + ); +} + /** The checker type of a stage's string or Buffer chunks, or null for * other chunk types. */ function stageChunkCheckerType(lowerer: Lowerer, chunk: IrType, at: ts.Node): ts.Type | null { diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts index bab07c458..734f9ee80 100644 --- a/packages/compiler/src/ir/builtin-effects.ts +++ b/packages/compiler/src/ir/builtin-effects.ts @@ -384,6 +384,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "writable.setDefaultEncoding", "stream.duplexFromGen", "stream.duplexFromDyn", + "stream.duplexFromPromise", + "stream.duplexFromAsyncFn", "stream.destroy", "stream.destroyErr", "stream.iteratorClose", diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 1708ad01a..ad8b907da 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -4069,6 +4069,16 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ "stream.decodeStrings": { argTypes: [null, BOOL], result: VOID }, "stream.duplexFromGen": { argTypes: [null, null, null, null, null, null], result: VOID }, "stream.duplexFromDyn": { argTypes: [DYN], result: VOID }, + /** Duplex.from over a lone Readable or Writable, or a { readable, + * writable } pair of streams (duplexify.js's _duplexify). */ + "stream.duplexifyReadable": { argTypes: [null], result: VOID }, + "stream.duplexifyWritable": { argTypes: [null], result: VOID }, + "stream.duplexifyPair": { argTypes: [null, null], result: VOID }, + /** Duplex.from(promise): the promise and its box helper. */ + "stream.duplexFromPromise": { argTypes: [null, null], result: VOID }, + /** Duplex.from(asyncFunction): the stage, its source helper and the + * promise's box helper. */ + "stream.duplexFromAsyncFn": { argTypes: [null, null, null], result: VOID }, "writable.writeEncoding": { argTypes: [null, STRING], result: VOID }, "writable.setDefaultEncoding": { argTypes: [null, DYN], result: VOID }, /** Whether a readable.nextChunkDyn step answered the end (the done diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts index da9cab1c3..84b767fab 100644 --- a/packages/compiler/src/ir/runtime-features.ts +++ b/packages/compiler/src/ir/runtime-features.ts @@ -221,6 +221,11 @@ const STREAM_LIB_FNS: ReadonlySet = new Set([ "stream.decodeStrings", "stream.duplexFromGen", "stream.duplexFromDyn", + "stream.duplexFromPromise", + "stream.duplexFromAsyncFn", + "stream.duplexifyReadable", + "stream.duplexifyWritable", + "stream.duplexifyPair", "console.bind", "crypto.hashStream", "crypto.hashStreamPiped", @@ -277,6 +282,8 @@ const READABLE_DYN_ASYNC_LIB_FNS: ReadonlySet = new Set([ "readable.fromIter", "stream.duplexFromGen", "stream.duplexFromDyn", + "stream.duplexFromPromise", + "stream.duplexFromAsyncFn", ]); /** libCall name prefixes of the Readable entry points that await through the dyn-async unit. */ diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts index a76784083..d08d0084f 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/packages/compiler/src/ir/validate.ts @@ -5499,6 +5499,11 @@ function validateFunction( e.fn === "stream.decodeStrings" || e.fn === "stream.duplexFromGen" || e.fn === "stream.duplexFromDyn" || + e.fn === "stream.duplexifyReadable" || + e.fn === "stream.duplexifyWritable" || + e.fn === "stream.duplexifyPair" || + e.fn === "stream.duplexFromPromise" || + e.fn === "stream.duplexFromAsyncFn" || e.fn === "stream.setConstruct" || e.fn === "stream.setWritev" || e.fn === "stream.setWritevDefault" || @@ -5554,7 +5559,30 @@ function validateFunction( } return true; } - if (e.fn === "stream.duplexFromGen" || e.fn === "stream.duplexFromDyn") { + if (e.fn === "stream.duplexFromPromise") { + // Duplex.from(promise): (promise, box) → a Duplex. + if (e.args[0]?.type.kind !== "promise" || e.args[1]?.type.kind !== "func") + err(`libCall ${e.fn} takes a promise and its box helper`, e.loc); + if (!isStreamObject(e.type)) err(`libCall ${e.fn} must return a stream class`, e.loc); + return true; + } + if ( + e.fn === "stream.duplexifyReadable" || + e.fn === "stream.duplexifyWritable" || + e.fn === "stream.duplexifyPair" + ) { + // Duplex.from over streams: (r), (w) or (r, w) → a Duplex. + const want = e.fn === "stream.duplexifyPair" ? 2 : 1; + if (e.args.length !== want || !e.args.every((a) => isStreamObject(a.type))) + err(`libCall ${e.fn} takes ${want} stream argument(s)`, e.loc); + if (!isStreamObject(e.type)) err(`libCall ${e.fn} must return a stream class`, e.loc); + return true; + } + if ( + e.fn === "stream.duplexFromGen" || + e.fn === "stream.duplexFromDyn" || + e.fn === "stream.duplexFromAsyncFn" + ) { // Duplex.from: (stage, source, helpers...) or (body) → a Duplex. if (!isStreamObject(e.type)) { err(`libCall ${e.fn} must return a stream class, got ${e.type.kind}`, e.loc); diff --git a/packages/compiler/test/native-codegen-integration.test.ts b/packages/compiler/test/native-codegen-integration.test.ts index bb6ccf874..169e89c2e 100644 --- a/packages/compiler/test/native-codegen-integration.test.ts +++ b/packages/compiler/test/native-codegen-integration.test.ts @@ -378,7 +378,7 @@ done: expect(await symbols(helperObject, ["-u"])).toEqual(await symbols(clangObject, ["-u"])); expect(await symbols(helperObject, ["-gU"])).toEqual(await symbols(clangObject, ["-gU"])); expect(await symbols(helperObject, ["-gU"])).toEqual(["0000000000000000 T _main"]); - expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v29"); + expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v30"); const clangExe = join(dir, "clang-program"); const linkDriver = join(dir, "clang-driver.c"); @@ -712,7 +712,7 @@ done: .split("\n"); expect(undefinedSymbols).toContain("_sf_scale"); expect(undefinedSymbols).toContain("_sf_callback_mix"); - expect(undefinedSymbols).toContain("_scr_runtime_abi_v29"); + expect(undefinedSymbols).toContain("_scr_runtime_abi_v30"); }); test("helper and clang object paths retain the same outbound FFI C ABI", async () => { diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7424-duplex-from-streams.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7424-duplex-from-streams.mjs.json new file mode 100644 index 000000000..a3801c775 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7424-duplex-from-streams.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7424-duplex-from-streams.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7425-duplex-from-streams.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7425-duplex-from-streams.ts.json new file mode 100644 index 000000000..069ae4ef4 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7425-duplex-from-streams.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7425-duplex-from-streams.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7426-duplex-from-promise.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7426-duplex-from-promise.mjs.json new file mode 100644 index 000000000..e44b3f001 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7426-duplex-from-promise.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7426-duplex-from-promise.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7427-duplex-from-async-function.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7427-duplex-from-async-function.mjs.json new file mode 100644 index 000000000..22d092c60 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7427-duplex-from-async-function.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7427-duplex-from-async-function.mjs" + ], + "diags": [] +} diff --git a/packages/runtime-darwin-arm64/scripts/build.mjs b/packages/runtime-darwin-arm64/scripts/build.mjs index 71126f806..aa251dc34 100644 --- a/packages/runtime-darwin-arm64/scripts/build.mjs +++ b/packages/runtime-darwin-arm64/scripts/build.mjs @@ -340,7 +340,7 @@ async function build() { package: packageManifest.name, version: packageManifest.version, target: RUNTIME_PACK_MATRIX.target, - runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" }, + runtime_abi: { version: 30, marker: "scr_runtime_abi_v30" }, compiler: { command: compiler, identity: compilerVersion, diff --git a/packages/runtime-pack-common/scripts/build.mjs b/packages/runtime-pack-common/scripts/build.mjs index b0ec0005b..acf463824 100644 --- a/packages/runtime-pack-common/scripts/build.mjs +++ b/packages/runtime-pack-common/scripts/build.mjs @@ -437,7 +437,7 @@ async function build() { package: packageManifest.name, version: packageManifest.version, target: matrix.target, - runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" }, + runtime_abi: { version: 30, marker: "scr_runtime_abi_v30" }, compiler: { command: compiler, identity: compilerVersion, diff --git a/packages/runtime/src/scr_console.c b/packages/runtime/src/scr_console.c index 317ee124b..5b7f8da0f 100644 --- a/packages/runtime/src/scr_console.c +++ b/packages/runtime/src/scr_console.c @@ -148,7 +148,7 @@ static void scr_enable_utf8_console_output(void) { } #endif -void scr_runtime_abi_v29(void) {} +void scr_runtime_abi_v30(void) {} void scr_init(void) { #ifdef SCR_WORKERS diff --git a/packages/runtime/src/scr_core.h b/packages/runtime/src/scr_core.h index 6f3aae93a..57928cfb5 100644 --- a/packages/runtime/src/scr_core.h +++ b/packages/runtime/src/scr_core.h @@ -81,7 +81,7 @@ void scr_init(void); /* Program objects emitted by the bundled LLVM helper reference this symbol. * Its versioned spelling makes a mismatched manual runtime link fail before * the program can start. */ -void scr_runtime_abi_v29(void); +void scr_runtime_abi_v30(void); void scr_runtime_workers_v8(void); /* ── the trap funnel (scr_console.c; scr_library.c under -DSCR_LIB) ────── diff --git a/packages/runtime/src/scr_stream.c b/packages/runtime/src/scr_stream.c index 4a53da4c0..992203eb4 100644 --- a/packages/runtime/src/scr_stream.c +++ b/packages/runtime/src/scr_stream.c @@ -1769,18 +1769,26 @@ ScrStream *scr_stream_from_dyn(ScrDyn *src, ScrDyn *opts) { } /* Duplex.from(body) over a checked-dynamic body (duplexify.js): a native - * Duplex answers itself, an iterable reads as Duplex.from does, and the - * other bodies Node accepts (a lone Readable or Writable, a function, a - * promise, a { readable, writable } pair) are not supported yet. */ + * Duplex answers itself, a lone native Readable or Writable is wrapped, a + * promise reads as its value, an iterable reads as Duplex.from does, and + * the other bodies Node accepts (a function, a thenable, a { readable, + * writable } pair) are not supported yet. */ static ScrStream *scr_stream_boxed(const ScrDyn *value); +static ScrStream *scr_dupx_from_boxed(ScrDyn *boxed); ScrStream *scr_stream_duplex_from_dyn(ScrDyn *body) { ScrStream *native = scr_stream_of_handle(body); /* +1 */ if (!native) native = scr_stream_boxed(body); /* +1 */ - if (native && native->st->has_r && native->st->has_w) return native; - bool unsupported = native != NULL; - if (native) scr_stream_release(native); - if (!unsupported && (body->kind == SCR_DYN_FUNC || body->kind == SCR_DYN_PROMISE)) unsupported = true; + if (native) { + if (native->st->has_r && native->st->has_w) return native; + /* _duplexify({ readable }) / _duplexify({ writable }) */ + ScrStream *d = native->st->has_r ? scr_stream_duplexify(native, NULL) + : scr_stream_duplexify(NULL, native); + scr_stream_release(native); + return d; + } + if (body->kind == SCR_DYN_PROMISE) return scr_dupx_from_boxed(scr_dyn_retain(body)); + bool unsupported = body->kind == SCR_DYN_FUNC; if (!unsupported && body->kind == SCR_DYN_OBJ) { static const char *const keys[] = {"then", "readable", "writable"}; for (size_t i = 0; i < 3 && !unsupported; i++) { @@ -4709,6 +4717,344 @@ ScrStream *scr_stream_compose_stages(double n, ScrStream **streams, double count return scr_stream_compose_impl((size_t)n, streams, &fs, abort); } +/* ── Duplex.from over a stream or a { readable, writable } pair ───────── + * duplexify.js's _duplexify({ readable, writable }): a Duplexify that + * writes to w (write() answering false waits for 'drain'; final ends w and + * waits for its 'finish') and reads r through 'readable' and read(), with + * each side's end of stream (eos) finishing the Duplex. The readable and + * writable object modes follow r's and w's. */ + +enum { SCR_DUPX_UNSET, SCR_DUPX_NULL, SCR_DUPX_WAIT }; + +typedef struct ScrDuplexify { + size_t rc; + ScrStream *r, *w; /* +1 or NULL */ + ScrStream *d; /* +1 until teardown */ + ScrClosure *drain_l, *finish_l, *readable_l, *end_l; /* +1, the listeners, or NULL */ + bool readable, writable; + bool ondrain, onfinish, onreadable; + int onclose; /* _duplexify's onclose: undefined, null, or d's destroy callback */ + struct ScrDuplexify *live_prev, *live_next; +} ScrDuplexify; + +static SCR_TL ScrDuplexify *scr_dupx_live = NULL; + +static void *scr_dupx_retain_v(void *p) { + ((ScrDuplexify *)p)->rc++; + return p; +} + +static void scr_dupx_release_v(void *p) { + ScrDuplexify *x = p; + if (--x->rc > 0) return; + if (x->r) scr_stream_release(x->r); + if (x->w) scr_stream_release(x->w); + free(x); +} + +static ScrClosure *scr_dupx_closure(void *fn, ScrDuplexify *x) { + ScrClosure *cl = scr_closure_new(fn, 1); + cl->caps[0] = scr_box_new_obj(&scr_dupx_retain_v, &scr_dupx_release_v, NULL); + scr_box_set_ref(cl->caps[0], scr_dupx_retain_v(x)); + return cl; +} + +static void scr_dupx_on(ScrStream *s, const char *name, ScrClosure **slot, void *fn, ScrDuplexify *x) { + ScrStr *n = scr_str_new(name, strlen(name)); + *slot = scr_dupx_closure(fn, x); + scr_emitter_release( + scr_emitter_on((ScrEmitter *)s, n, scr_closure_retain(*slot), scr_ee_inv_fixed0, false, false)); + scr_str_release(n); +} + +/* Drop the Duplex and the listeners on r and w. */ +static void scr_dupx_teardown(ScrDuplexify *x) { + if (!x->d) return; + if (x->w) { + scr_compose_off(x->w, "drain", &x->drain_l); + scr_compose_off(x->w, "finish", &x->finish_l); + } + if (x->r) { + scr_compose_off(x->r, "readable", &x->readable_l); + scr_compose_off(x->r, "end", &x->end_l); + } + if (x->live_prev) x->live_prev->live_next = x->live_next; + else scr_dupx_live = x->live_next; + if (x->live_next) x->live_next->live_prev = x->live_prev; + x->live_prev = x->live_next = NULL; + ScrStream *d = x->d; + x->d = NULL; + scr_stream_release(d); + scr_dupx_release_v(x); /* the live list's reference */ +} + +static void scr_dupx_atexit(void) { + while (scr_dupx_live) scr_dupx_teardown(scr_dupx_live); +} + +/* d's destroy callback: callback(err), then the context lets go. */ +static void scr_dupx_destroyed(ScrDuplexify *x, ScrError *err) { + ScrStream *d = scr_stream_retain(x->d); + scr_dupx_teardown(x); + scr_stream_destroy_done(d, err ? scr_error_retain(err) : NULL); + scr_stream_release(d); +} + +/* onfinished(err): the pending destroy callback, else d.destroy(err) */ +static void scr_dupx_onfinished(ScrDuplexify *x, ScrError *err) { + int onclose = x->onclose; + x->onclose = SCR_DUPX_NULL; + if (!x->d) return; + if (onclose == SCR_DUPX_WAIT) { + scr_dupx_destroyed(x, err); + } else if (err) { + ScrStream *d = scr_stream_retain(x->d); + scr_stream_release(scr_stream_destroy(d, err)); + scr_stream_release(d); + } +} + +/* eos(w): writable = false; an error destroys r */ +static void scr_dupx_w_eos(ScrClosure *cb, ScrStream *s, ScrError *err) { + (void)s; + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + x->writable = false; + if (err && x->r) scr_stream_release(scr_stream_destroy(x->r, err)); + scr_dupx_onfinished(x, err); + scr_dupx_release_v(x); +} + +/* eos(r): readable = false; an error destroys w */ +static void scr_dupx_r_eos(ScrClosure *cb, ScrStream *s, ScrError *err) { + (void)s; + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + x->readable = false; + if (err && x->w) scr_stream_release(scr_stream_destroy(x->w, err)); + scr_dupx_onfinished(x, err); + scr_dupx_release_v(x); +} + +/* d._write: w.write(chunk) ? callback() : (ondrain = callback) */ +static void scr_dupx_write_inv(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) { + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + bool ok = scr_stream_writes_values(s) ? scr_stream_write_dyn(x->w, (ScrDyn *)chunk, NULL) + : scr_stream_write(x->w, chunk, NULL); + if (ok) { + if (!scr_exc_pending()) scr_stream_write_done(s, NULL); + } else { + x->ondrain = true; + } + scr_dupx_release_v(x); +} + +/* d._final: w.end(); onfinish = callback */ +static void scr_dupx_final_inv(ScrClosure *cb, ScrStream *s) { + (void)s; + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + ScrStream *r = scr_stream_end(x->w, NULL, NULL, NULL); + if (r) scr_stream_release(r); + x->onfinish = true; + scr_dupx_release_v(x); +} + +static void scr_dupx_drain_l(ScrClosure *cb) { + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + if (x->ondrain && x->d) { + x->ondrain = false; + scr_stream_write_done(x->d, NULL); + } + scr_dupx_release_v(x); +} + +static void scr_dupx_finish_l(ScrClosure *cb) { + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + if (x->onfinish && x->d) { + x->onfinish = false; + scr_stream_final_done(x->d, NULL); + } + scr_dupx_release_v(x); +} + +/* d._read: push r.read() until it answers null (then wait for 'readable') + * or d is full */ +static void scr_dupx_pull(ScrDuplexify *x) { + ScrStream *d = x->d; + while (d && x->r) { + ScrEntryKind kind = SCR_ENTRY_BYTES; + void *chunk = scr_stream_read_n_kind(x->r, -1, &kind); + if (scr_exc_pending()) return; + if (!chunk) { + x->onreadable = true; + return; + } + bool more; + if (d->st->r.object_entries) { + more = scr_stream_add_entry(d, chunk, kind) > 0; + } else if (kind == SCR_ENTRY_STR) { + more = scr_stream_push_str(d, chunk); + scr_str_release(chunk); + } else if (kind == SCR_ENTRY_DYN) { + more = scr_stream_push_dyn(d, chunk); + scr_dyn_release(chunk); + } else { + more = scr_stream_push(d, chunk); + scr_bytes_release(chunk); + } + if (!more || scr_exc_pending()) return; + } +} + +static void scr_dupx_read_inv(ScrClosure *cb, ScrStream *s, double size) { + (void)s; + (void)size; + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + x->onreadable = false; + scr_dupx_pull(x); + scr_dupx_release_v(x); +} + +static void scr_dupx_readable_l(ScrClosure *cb) { + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + if (x->onreadable && x->d) { + x->onreadable = false; + scr_dupx_pull(x); + } + scr_dupx_release_v(x); +} + +static void scr_dupx_end_l(ScrClosure *cb) { + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + if (x->d) scr_stream_push_null(x->d); + scr_dupx_release_v(x); +} + +/* d._destroy */ +static void scr_dupx_destroy_inv(ScrClosure *cb, ScrStream *s, ScrError *err) { + (void)s; + ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); + ScrError *e = err ? scr_error_retain(err) : NULL; + if (!e && x->onclose != SCR_DUPX_NULL) e = scr_compose_abort_error(); + x->onreadable = x->ondrain = x->onfinish = false; + if (x->onclose == SCR_DUPX_NULL) { + if (x->d) scr_dupx_destroyed(x, e); + } else { + x->onclose = SCR_DUPX_WAIT; + if (x->w && !x->w->st->destroyed) scr_stream_release(scr_stream_destroy(x->w, e)); + if (x->r && !x->r->st->destroyed) scr_stream_release(scr_stream_destroy(x->r, e)); + } + if (e) scr_error_release(e); + scr_dupx_release_v(x); +} + +/* _duplexify({ readable: r, writable: w }): r and w borrowed, either NULL. + * +1. */ +ScrStream *scr_stream_duplexify(ScrStream *r, ScrStream *w) { + ScrDuplexify *x = calloc(1, sizeof *x); + if (!x) scr_stream_oom(); + x->rc = 1; + x->r = r ? scr_stream_retain(r) : NULL; + x->w = w ? scr_stream_retain(w) : NULL; + x->onclose = SCR_DUPX_UNSET; + x->readable = r && scr_stream_prop(r, "readable") != 0; + x->writable = w && scr_stream_prop(w, "writable") != 0; + /* new Duplexify({ readableObjectMode, writableObjectMode, readable, writable }) */ + bool r_object = r && r->st->has_r && r->st->r.object_entries; + bool w_object = w && w->st->has_w && w->st->w.object_mode; + ScrStream *d = scr_stream_new_duplex( + r_object ? 16 : -1, w_object ? 16 : -1, true, true, true, x->readable, x->writable, + x->readable ? scr_dupx_closure((void *)&scr_dupx_read_inv, x) : NULL, + x->readable ? &scr_dupx_read_inv : NULL, + x->writable ? scr_dupx_closure((void *)&scr_dupx_write_inv, x) : NULL, + x->writable ? &scr_dupx_write_inv : NULL, + x->writable ? scr_dupx_closure((void *)&scr_dupx_final_inv, x) : NULL, + x->writable ? &scr_dupx_final_inv : NULL, + scr_dupx_closure((void *)&scr_dupx_destroy_inv, x), &scr_dupx_destroy_inv); + if (r_object) d->st->r.object_entries = true; + if (w_object) scr_stream_set_object_mode(d, false, true); + d->cls = "Duplexify"; + x->d = scr_stream_retain(d); + if (x->writable) { + scr_closure_release( + scr_stream_finished(w, scr_dupx_closure((void *)&scr_dupx_w_eos, x), &scr_dupx_w_eos)); + scr_dupx_on(w, "drain", &x->drain_l, (void *)&scr_dupx_drain_l, x); + scr_dupx_on(w, "finish", &x->finish_l, (void *)&scr_dupx_finish_l, x); + } + if (x->readable) { + scr_closure_release( + scr_stream_finished(r, scr_dupx_closure((void *)&scr_dupx_r_eos, x), &scr_dupx_r_eos)); + scr_dupx_on(r, "readable", &x->readable_l, (void *)&scr_dupx_readable_l, x); + scr_dupx_on(r, "end", &x->end_l, (void *)&scr_dupx_end_l, x); + } + static SCR_TL bool registered = false; + if (!registered) { + registered = true; + scr_atexit(scr_dupx_atexit); + } + x->live_next = scr_dupx_live; + if (scr_dupx_live) scr_dupx_live->live_prev = x; + scr_dupx_live = x; /* the reference x was created with */ + return d; +} + +/* Duplex.from(promise) (duplexify.js): a Duplexify({ objectMode: true, + * writable: false, read() {} }) that pushes the promise's value unless it + * is nullish, then ends; a rejection destroys it with the reason. */ +typedef struct { + ScrStream *d; /* +1 */ + ScrDyn *boxed; /* +1, the promise boxed */ +} ScrDupxPromise; + +static void scr_dupx_noop_read(ScrClosure *cb, ScrStream *s, double size) { + (void)cb; + (void)s; + (void)size; +} + +static void scr_dupx_promise_entry(ScrFiber *fiber, void *arg) { + (void)fiber; + ScrDupxPromise p = *(ScrDupxPromise *)arg; + free(arg); + /* the boxed promise's payload is a dyn value (NULL: undefined) */ + ScrDyn *v = scr_await_ref(p.boxed->v.promise); + if (!scr_exc_pending()) { + if (v && v->kind != SCR_DYN_NULL && v->kind != SCR_DYN_UNDEF) scr_stream_push_object(p.d, v); + if (!scr_exc_pending()) scr_stream_push_null(p.d); + if (scr_exc_pending()) scr_exc_clear(); + if (v) scr_dyn_release(v); + } else { + ScrError *err = scr_pump_caught(); + scr_stream_release(scr_stream_destroy(p.d, err)); + if (scr_exc_pending()) scr_exc_clear(); + scr_error_release(err); + } + scr_stream_release(p.d); + scr_dyn_release(p.boxed); +} + +/* The readable Duplexify over a boxed promise (moves boxed). +1. */ +static ScrStream *scr_dupx_from_boxed(ScrDyn *boxed) { + ScrClosure *read = scr_closure_new((void *)&scr_dupx_noop_read, 0); + ScrStream *d = scr_stream_new_duplex(16, 16, true, true, true, true, false, read, + &scr_dupx_noop_read, NULL, NULL, NULL, NULL, NULL, NULL); + scr_stream_set_object_mode(d, true, true); + d->cls = "Duplexify"; + ScrDupxPromise *ctx = malloc(sizeof *ctx); + if (!ctx) scr_stream_oom(); + ctx->d = scr_stream_retain(d); + ctx->boxed = boxed; + scr_promise_release(scr_async_spawn_after(ctx->boxed->v.promise, &scr_dupx_promise_entry, ctx)); + return d; +} + +/* p and box borrowed (box: the compiler's promise → dyn promise helper). +1. */ +ScrStream *scr_stream_duplex_from_promise(ScrPromise *p, ScrClosure *box) { + return scr_dupx_from_boxed(((ScrPipeBoxFn)box->fn)(box, scr_promise_retain(p))); +} + +ScrStream *scr_stream_duplexify_readable(ScrStream *r) { return scr_stream_duplexify(r, NULL); } +ScrStream *scr_stream_duplexify_writable(ScrStream *w) { return scr_stream_duplexify(NULL, w); } +ScrStream *scr_stream_duplexify_pair(ScrStream *r, ScrStream *w) { return scr_stream_duplexify(r, w); } + /* ── node:stream/promises ───────────────────────────────────────────── * The promise forms ride the callback machinery above with a settling * watcher: caps[0] boxes the pending promise; the terminal status diff --git a/packages/runtime/src/scr_stream.h b/packages/runtime/src/scr_stream.h index 623f60465..d276700c8 100644 --- a/packages/runtime/src/scr_stream.h +++ b/packages/runtime/src/scr_stream.h @@ -545,6 +545,20 @@ void scr_stream_set_writev_default(ScrStream *s, ScrClosure *cb, ScrStreamWritev ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, ScrClosure *next, ScrClosure *read, ScrClosure *ret, ScrClosure *thr); ScrStream *scr_stream_duplex_from_dyn(struct ScrDyn *body); +/* Duplex.from over an async function (the compiled stage takes the source + * generator and answers its promise; source and the promise's box helper + * borrowed): a writable Duplexify whose final waits for the promise. +1. */ +ScrStream *scr_stream_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source, ScrClosure *box); +/* Duplex.from over a Readable, a Writable, or a { readable, writable } + * pair (duplexify.js's _duplexify): a Duplexify that reads r and writes w + * (either NULL in scr_stream_duplexify; streams borrowed). +1. */ +ScrStream *scr_stream_duplexify(ScrStream *r, ScrStream *w); +ScrStream *scr_stream_duplexify_readable(ScrStream *r); +ScrStream *scr_stream_duplexify_writable(ScrStream *w); +ScrStream *scr_stream_duplexify_pair(ScrStream *r, ScrStream *w); +/* Duplex.from(promise): a readable Duplexify of the promise's value. p and + * box (the compiler's promise → dyn promise helper) borrowed. +1. */ +ScrStream *scr_stream_duplex_from_promise(struct ScrPromise *p, ScrClosure *box); /* decodeStrings (false keeps string writes as strings), the writable * defaultEncoding option (a known encoding, borrowed), and * setDefaultEncoding (validated; answers the stream +1). The chaining diff --git a/packages/runtime/src/scr_stream_ops.c b/packages/runtime/src/scr_stream_ops.c index 1db90bc5a..75ce416ac 100644 --- a/packages/runtime/src/scr_stream_ops.c +++ b/packages/runtime/src/scr_stream_ops.c @@ -2042,7 +2042,7 @@ ScrStream *scr_stream_from_async_iter(ScrDyn *iter, double flags, double hwm) { * (the chunk, or the done marker), and ack() is fromAsyncGen's * `process.nextTick(cb)` followed by a fresh deferred. */ -enum { CHAN_NONE, CHAN_WRITE, CHAN_FINAL, CHAN_ABORT }; +enum { CHAN_NONE, CHAN_WRITE, CHAN_FINAL, CHAN_ABORT, CHAN_FINAL_ASYNC }; typedef struct OpChan OpChan; struct OpChan { @@ -2052,6 +2052,12 @@ struct OpChan { ScrPromise *resolve; /* +1: the deferred a write or the final settles, NULL once settled */ int settled; /* CHAN_*: what settled the deferred, for ack() */ bool aborted; /* destroy ran (fromAsyncGen's ac.abort()) */ + /* An async function body: its promise's then() (duplexify.js) settles + * `promise` there; the final waits for it. */ + bool finalized; /* final() has started */ + bool body_settled; /* the then() callbacks ran */ + bool final_waiting; /* the final's callback awaits the settlement */ + ScrError *body_err; /* +1: what awaiting the then() result throws, or NULL */ OpChan *live_prev, *live_next; }; @@ -2059,6 +2065,7 @@ static SCR_TL OpChan *op_chan_live = NULL; static void op_chan_drop(OpObj *o) { OpChan *c = (OpChan *)o; + if (c->body_err) scr_error_release(c->body_err); if (c->d) scr_stream_release(c->d); if (c->promise) scr_promise_release(c->promise); if (c->resolve) scr_promise_release(c->resolve); @@ -2136,10 +2143,49 @@ static ScrError *op_chan_abort_error(void) { /* After the source's await: process.nextTick(cb), then (unless done) the * aborted check and a fresh deferred. */ +static void op_chan_final_async_wait(OpChan *c); + +/* The final's completion after `await promise`: process.nextTick(cb, err) */ +static void op_chan_final_async_done(ScrClosure *job) { + OpChan *c = op_chan_of(job); + if (c->d) { + ScrError *err = c->body_err ? scr_error_retain(c->body_err) : NULL; + ScrStream *d = scr_stream_retain(c->d); + scr_stream_final_done(d, err); + scr_stream_release(d); + } + op_unref(c); +} + +/* `await promise` resumed: the completion follows on the next tick */ +static void op_chan_final_async_resume(ScrClosure *job) { + OpChan *c = op_chan_of(job); + scr_next_tick(op_chan_closure((void *)&op_chan_final_async_done, c)); + op_unref(c); +} + +/* The final's async callback (run by the source's nextTick): it awaits + * the then() result, which may still be pending. */ +static void op_chan_final_async_tick(ScrClosure *job) { + OpChan *c = op_chan_of(job); + op_chan_final_async_wait(c); + op_unref(c); +} + +static void op_chan_final_async_wait(OpChan *c) { + if (c->body_settled) scr_queue_microtask(op_chan_closure((void *)&op_chan_final_async_resume, c)); + else c->final_waiting = true; +} + static void op_chan_ack_inv(ScrClosure *cb) { OpChan *c = op_chan_of(cb); int settled = c->settled; c->settled = CHAN_NONE; + if (settled == CHAN_FINAL_ASYNC) { + scr_next_tick(op_chan_closure((void *)&op_chan_final_async_tick, c)); + op_unref(c); + return; + } if (c->d && (settled == CHAN_WRITE || settled == CHAN_FINAL)) { ScrClosure *tick = scr_closure_new((void *)&op_chan_done_tick, 2); tick->caps[0] = scr_box_new_obj(&scr_stream_retain_v, &scr_stream_release_v, &scr_stream_trace); @@ -2261,6 +2307,126 @@ ScrStream *scr_stream_duplex_from_gen(ScrClosure *stage, ScrClosure *source, Scr return d; } +/* ── Duplex.from(asyncFunction) ── duplexify.js: the function's promise + * must settle to a nullish value (else ERR_INVALID_RETURN_VALUE when the + * final awaits it), a return before the input was consumed destroys the + * Duplex, a rejection destroys it with the reason, and the Duplexify is + * writable only, its final waiting for the promise. */ + +typedef ScrPromise *(*OpAsyncStageFn)(ScrClosure *, ScrGen *); +typedef ScrDyn *(*OpBoxFn)(ScrClosure *, ScrPromise *); + +typedef struct { + OpChan *c; /* +1 */ + ScrDyn *boxed; /* +1, the function's promise boxed */ +} OpAsyncBody; + +static void op_async_body_entry(ScrFiber *fiber, void *arg) { + (void)fiber; + OpAsyncBody b = *(OpAsyncBody *)arg; + free(arg); + OpChan *c = b.c; + ScrDyn *v = scr_await_ref(b.boxed->v.promise); /* NULL: undefined */ + ScrStream *d = c->d ? scr_stream_retain(c->d) : NULL; + if (!scr_exc_pending()) { + if (v && v->kind != SCR_DYN_NULL && v->kind != SCR_DYN_UNDEF) { + /* the then() callback throws: awaiting its result rejects */ + char detail[160]; + const char *got = scr_dyn_specific_type(v, detail, sizeof detail); + char msg[320]; + int n = snprintf(msg, sizeof msg, + "Expected nully to be returned from the \"body\" function but got %s.", got); + c->body_err = op_coded_error(SCR_ERR_TYPE, "ERR_INVALID_RETURN_VALUE", msg); + (void)n; + } else if (!c->finalized && d) { + /* returned without consuming the input: destroyer(d) */ + ScrError *e = op_chan_abort_error(); + scr_stream_release(scr_stream_destroy(d, e)); + scr_error_release(e); + } + if (v) scr_dyn_release(v); + } else { + ScrCaught *caught = scr_exc_take(); + ScrError *err = op_caught_error(caught); + scr_caught_release(caught); + if (d) scr_stream_release(scr_stream_destroy(d, err)); + scr_error_release(err); + } + if (scr_exc_pending()) scr_exc_clear(); + if (d) scr_stream_release(d); + c->body_settled = true; + if (c->final_waiting) { + c->final_waiting = false; + op_chan_final_async_wait(c); + } + op_unref(c); + scr_dyn_release(b.boxed); +} + +/* Duplexify's final(cb) for an async function body: finalized, then + * fromAsyncGen's final with the awaiting callback */ +static void op_chan_final_async_inv(ScrClosure *cb, ScrStream *s) { + (void)s; + OpChan *c = op_chan_of(cb); + c->finalized = true; + op_chan_settle(c, scr_stream_from_done(), CHAN_FINAL_ASYNC); + op_unref(c); +} + +/* fromAsyncGen's destroy alone (no from() iterator to close) */ +static void op_chan_destroy_inv(ScrClosure *cb, ScrStream *s, ScrError *err) { + OpChan *c = op_chan_of(cb); + c->aborted = true; + if (c->resolve) op_chan_settle(c, scr_stream_from_done(), CHAN_ABORT); + op_chan_retire(c); + op_unref(c); + scr_stream_destroy_done(s, err ? scr_error_retain(err) : NULL); +} + +/* Duplex.from(fn) for an async function: the compiled stage (borrowed) + * takes the source generator and answers its promise; source and box + * (promise → dyn promise) borrowed. +1, or NULL with the function's throw + * pending. */ +ScrStream *scr_stream_duplex_from_async_fn(ScrClosure *stage, ScrClosure *source, ScrClosure *box) { + op_sentinels_init(); + static SCR_TL bool registered = false; + if (!registered) { + registered = true; + scr_atexit(op_chan_atexit); + } + OpChan *c = op_alloc(sizeof *c); + c->base.rc = 1; + c->base.drop = &op_chan_drop; + op_chan_defer(c); + ScrGen *src = ((OpSourceFn)source->fn)(source, op_chan_closure((void *)&op_chan_next_inv, c), + op_chan_closure((void *)&op_chan_ack_inv, c)); + ScrPromise *p = scr_exc_pending() ? NULL : ((OpAsyncStageFn)stage->fn)(stage, src); + if (scr_exc_pending()) { + if (p) scr_promise_release(p); + op_unref(c); + return NULL; + } + /* Duplexify({ objectMode: true, readable: false, write, final, destroy }) */ + ScrStream *d = scr_stream_new_duplex(16, 16, true, true, true, false, true, NULL, NULL, + op_chan_closure((void *)&op_chan_write_inv, c), &op_chan_write_inv, + op_chan_closure((void *)&op_chan_final_async_inv, c), + &op_chan_final_async_inv, + op_chan_closure((void *)&op_chan_destroy_inv, c), &op_chan_destroy_inv); + scr_stream_set_object_mode(d, true, true); + d->cls = "Duplexify"; + c->d = scr_stream_retain(d); + c->live_next = op_chan_live; + if (op_chan_live) op_chan_live->live_prev = c; + op_chan_live = c; /* the reference c was created with */ + /* value.then(onFulfilled, onRejected) */ + OpAsyncBody *b = malloc(sizeof *b); + if (!b) op_oom(); + b->c = op_ref(c); + b->boxed = ((OpBoxFn)box->fn)(box, p); /* p moves */ + scr_promise_release(scr_async_spawn_after(b->boxed->v.promise, &op_async_body_entry, b)); + return d; +} + /* ── the promise operators ─────────────────────────────────────────── */ enum { PK_DYN, PK_F64, PK_BOOL, PK_STR, PK_VOID }; diff --git a/tests/corpus/7424-duplex-from-streams.mjs b/tests/corpus/7424-duplex-from-streams.mjs new file mode 100644 index 000000000..5a9bdc01a --- /dev/null +++ b/tests/corpus/7424-duplex-from-streams.mjs @@ -0,0 +1,39 @@ +// Duplex.from over a lone Readable, a lone Writable and a { readable, +// writable } pair, as duplexify.js's _duplexify wraps them: the Duplex +// reads through the Readable and writes through the Writable, takes each +// side's object mode from the stream behind it, and closes once both ends +// finish; destroying it without an error destroys the streams behind it. +import { Duplex, PassThrough, Readable, Writable } from "node:stream"; +const log = []; +const r = Readable.from(["a", "b"]); +const dr = Duplex.from(r); +log.push(`dr ${dr.readable} ${dr.writable} ${dr.readableObjectMode} ${dr.writableObjectMode}`); +dr.on("end", () => log.push("dr end")); +dr.on("close", () => log.push("dr close")); +for await (const c of dr) log.push(`dr ${c}`); +const got = []; +const w = new Writable({ write(c, e, cb) { got.push(String(c)); log.push(`w write ${c}`); cb(); } }); +const dw = Duplex.from(w); +log.push(`dw ${dw.readable} ${dw.writable} ${dw.writableObjectMode}`); +dw.on("finish", () => log.push("dw finish")); +dw.on("close", () => log.push("dw close")); +dw.write("x", () => log.push("dw cb x")); +dw.end("y", () => log.push("dw end cb")); +await new Promise((res) => dw.on("close", res)); +const inner = new PassThrough({ objectMode: true }); +const out = new PassThrough({ objectMode: true }); +const pair = Duplex.from({ readable: out, writable: inner }); +log.push(`pair ${pair.readable} ${pair.writable} ${pair.readableObjectMode} ${pair.writableObjectMode}`); +inner.on("data", (v) => { log.push(`inner ${JSON.stringify(v)}`); out.write({ echo: v }); }); +inner.on("finish", () => { log.push("inner finish"); out.end(); }); +pair.on("data", (v) => log.push(`pair data ${JSON.stringify(v)}`)); +pair.write({ n: 1 }); +pair.end({ n: 2 }); +await new Promise((res) => pair.on("close", res)); +log.push("pair closed"); +const killed = Duplex.from(new PassThrough()); +killed.on("error", (e) => log.push(`killed error ${e.name}`)); +killed.on("close", () => log.push(`killed close ${killed.destroyed}`)); +killed.destroy(); +await new Promise((res) => setTimeout(res, 5)); +console.log(log.join("\n")); diff --git a/tests/corpus/7425-duplex-from-streams.ts b/tests/corpus/7425-duplex-from-streams.ts new file mode 100644 index 000000000..a9d85b797 --- /dev/null +++ b/tests/corpus/7425-duplex-from-streams.ts @@ -0,0 +1,72 @@ +// Duplex.from over streams in typed code: a Readable that fails destroys +// the Duplex with its error, a slow Writable's callbacks and the end +// callback follow its pace, a pair whose writable member is an async +// generator function's Duplex, compose() with an async function as its +// last stage, and a Duplex over a promise. +import { Duplex, PassThrough, Readable, Writable, compose } from "node:stream"; +const log: string[] = []; +async function main(): Promise { + const failing = new Readable({ read(): void { this.destroy(new Error("source failed")); } }); + const df = Duplex.from(failing); + df.on("error", (e: Error) => log.push(`df error ${e.message}`)); + df.on("close", () => log.push(`df close ${df.destroyed}`)); + df.resume(); + await new Promise((res) => df.on("close", res)); + const slow = new Writable({ + highWaterMark: 2, + write(c: Buffer, _e: string, cb: (err?: Error | null) => void): void { + log.push(`slow write ${c.toString()}`); + setTimeout(cb, 2); + }, + }); + const ds = Duplex.from(slow); + log.push(`write1 ${ds.write("aa", () => log.push("cb aa"))}`); + log.push(`write2 ${ds.write("bb", () => log.push("cb bb"))}`); + ds.on("drain", () => log.push("ds drain")); + ds.end(() => log.push("ds end cb")); + await new Promise((res) => ds.on("finish", res)); + log.push("ds finished"); + const upper = new PassThrough(); + const dp = Duplex.from({ + readable: Readable.from(["r1", "r2"]), + writable: async function* (source: AsyncIterable): AsyncGenerator { + for await (const c of source) log.push(`gen got ${String(c)}`); + yield "unused"; + }, + }); + dp.on("data", (c: unknown) => log.push(`dp data ${String(c)}`)); + dp.write("w1"); + dp.end(); + await new Promise((res) => dp.on("end", res)); + upper.end(); + await more(); + for (const line of log) console.log(line); +} +async function more(): Promise { + const sink = compose(new PassThrough({ objectMode: true }), async function (source: AsyncIterable): Promise { + for await (const c of source) log.push(`sink ${String(c)}`); + }); + log.push(`sink r=${sink.readable} w=${sink.writable}`); + sink.write("p"); + sink.end("q"); + await new Promise((res) => sink.on("finish", res)); + log.push("sink finished"); + const fromReadable: Duplex = Duplex.from(Readable.from([1, 2])); + const values: unknown[] = []; + for await (const v of fromReadable) values.push(v); + log.push(`readable ${JSON.stringify(values)}`); + const written: string[] = []; + const target = new Writable({ + write(c: Buffer, _e: string, cb: (err?: Error | null) => void): void { + written.push(c.toString()); + cb(); + }, + }); + const fromWritable: Duplex = Duplex.from(target); + fromWritable.end("only"); + await new Promise((res) => fromWritable.on("finish", res)); + log.push(`writable ${written.join(",")}`); + const value: Duplex = Duplex.from(Promise.resolve("resolved")); + for await (const v of value) log.push(`promise ${String(v)}`); +} +void main(); diff --git a/tests/corpus/7426-duplex-from-promise.mjs b/tests/corpus/7426-duplex-from-promise.mjs new file mode 100644 index 000000000..46d19586b --- /dev/null +++ b/tests/corpus/7426-duplex-from-promise.mjs @@ -0,0 +1,27 @@ +// Duplex.from over a promise: a Duplex that is not writable and reads the +// promise's value once it settles (nothing for null), then ends; a +// rejection destroys it with the reason. The events interleave with +// microtasks and nextTick callbacks as in Node. +import { Duplex } from "node:stream"; +const log = []; +const tick = (n) => { if (n < 5) { log.push(`tick ${n}`); process.nextTick(() => tick(n + 1)); } }; +const later = (v, ms) => new Promise((res) => setTimeout(() => res(v), ms)); +const a = Duplex.from(Promise.resolve({ answer: 42 })); +log.push(`a ${a.readable} ${a.writable} ${a.readableObjectMode}`); +a.on("data", (v) => log.push(`a data ${JSON.stringify(v)}`)); +a.on("end", () => log.push("a end")); +a.on("close", () => log.push("a close")); +Promise.resolve().then(() => log.push("microtask 1")).then(() => log.push("microtask 2")); +tick(0); +await new Promise((res) => a.on("close", res)); +const b = Duplex.from(later(null, 1)); +b.on("data", () => log.push("b data")); +b.on("end", () => log.push("b end")); +await new Promise((res) => b.on("close", res)); +const c = Duplex.from(Promise.reject(new Error("nope"))); +c.on("error", (e) => log.push(`c error ${e.message}`)); +c.on("close", () => log.push(`c close ${c.destroyed}`)); +await new Promise((res) => c.on("close", res)); +const d = Duplex.from(later("text", 1)); +for await (const v of d) log.push(`d ${v}`); +console.log(log.join("\n")); diff --git a/tests/corpus/7427-duplex-from-async-function.mjs b/tests/corpus/7427-duplex-from-async-function.mjs new file mode 100644 index 000000000..66ce35f9a --- /dev/null +++ b/tests/corpus/7427-duplex-from-async-function.mjs @@ -0,0 +1,40 @@ +// Duplex.from over an async function: a Duplex that is only writable, whose +// writes reach the function's source and whose end waits for the +// function's promise. A value other than null or undefined is +// ERR_INVALID_RETURN_VALUE, a return before the input is consumed aborts +// the Duplex, and a throw destroys it. +import { Duplex } from "node:stream"; +const log = []; +const sink = Duplex.from(async function (source) { + for await (const c of source) log.push(`sink got ${c}`); + log.push("sink done"); +}); +log.push(`sink ${sink.readable} ${sink.writable} ${sink.writableObjectMode}`); +sink.on("finish", () => log.push("sink finish")); +sink.on("close", () => log.push("sink close")); +sink.write("a", () => log.push("cb a")); +sink.end("b", () => log.push("end cb")); +await new Promise((res) => sink.on("close", res)); +const bad = Duplex.from(async function (source) { + for await (const c of source) log.push(`bad got ${c}`); + return 5; +}); +bad.on("error", (e) => log.push(`bad error ${e.code} ${e.message}`)); +bad.on("close", () => log.push("bad close")); +bad.end("x"); +await new Promise((res) => bad.on("close", res)); +const early = Duplex.from(async function (source) { + for await (const c of source) { log.push(`early got ${c}`); break; } +}); +early.on("error", (e) => log.push(`early error ${e.name}`)); +early.on("close", () => log.push(`early close ${early.destroyed}`)); +early.write("1"); +early.write("2"); +await new Promise((res) => early.on("close", res)); +const failing = Duplex.from(async function (source) { + for await (const c of source) throw new Error("fail " + c); +}); +failing.on("error", (e) => log.push(`failing error ${e.message}`)); +failing.write("z"); +await new Promise((res) => failing.on("close", res)); +console.log(log.join("\n")); diff --git a/tests/harness/corpus-costs/7424-duplex-from-streams.mjs.json b/tests/harness/corpus-costs/7424-duplex-from-streams.mjs.json new file mode 100644 index 000000000..14a2052b9 --- /dev/null +++ b/tests/harness/corpus-costs/7424-duplex-from-streams.mjs.json @@ -0,0 +1 @@ +{ "plain": 2607, "san": 3440 } diff --git a/tests/harness/corpus-costs/7425-duplex-from-streams.ts.json b/tests/harness/corpus-costs/7425-duplex-from-streams.ts.json new file mode 100644 index 000000000..9dde8d70e --- /dev/null +++ b/tests/harness/corpus-costs/7425-duplex-from-streams.ts.json @@ -0,0 +1 @@ +{ "plain": 2009, "san": 2743 } diff --git a/tests/harness/corpus-costs/7426-duplex-from-promise.mjs.json b/tests/harness/corpus-costs/7426-duplex-from-promise.mjs.json new file mode 100644 index 000000000..9195268ab --- /dev/null +++ b/tests/harness/corpus-costs/7426-duplex-from-promise.mjs.json @@ -0,0 +1 @@ +{ "plain": 1877, "san": 2357 } diff --git a/tests/harness/corpus-costs/7427-duplex-from-async-function.mjs.json b/tests/harness/corpus-costs/7427-duplex-from-async-function.mjs.json new file mode 100644 index 000000000..c572d9930 --- /dev/null +++ b/tests/harness/corpus-costs/7427-duplex-from-async-function.mjs.json @@ -0,0 +1 @@ +{ "plain": 1811, "san": 2417 } diff --git a/tests/harness/self-hosting-native-toolchain.test.ts b/tests/harness/self-hosting-native-toolchain.test.ts index f143419dc..81ff5b293 100644 --- a/tests/harness/self-hosting-native-toolchain.test.ts +++ b/tests/harness/self-hosting-native-toolchain.test.ts @@ -108,7 +108,7 @@ for (const backend of ["llvm"] as const) { object_format: target.objectFormat, minimum_os: target.minimumOs, }, - runtime_abi: { version: 29, marker: "scr_runtime_abi_v29" }, + runtime_abi: { version: 30, marker: "scr_runtime_abi_v30" }, compiler: { command: "fixture", identity: "fixture", target: target.llvmTriple }, macros: { executable: [], excluded: [], sanitizer: "external-toolchain-required" }, flavors: {