diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx index 561840268..25427a307 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -4,7 +4,13 @@ When an untyped value reaches `NodeJS.ReadableStream` or `NodeJS.WritableStream`, those two interfaces hold any value, so JavaScript stream objects (constructor-function streams, plain objects with `on` and `pipe`) pass through TypeScript functions typed with them and come back to JavaScript as the same objects. Member calls on such a value use the native stream when it is one and call the object's own members otherwise; `readable.pipe()` into a JavaScript destination follows Node's pipe protocol. The first `pipeline` stage may arrive as an untyped value that is a native stream, a `Cipheriv` or `Decipheriv` value, or an iterable: a string, an array, a generator object, or any value with a `Symbol.asyncIterator` or `Symbol.iterator` method, which the pipeline pumps into the next stage as Node does. - Later `pipeline` stages and `for await` sources that arrive as untyped values must be native streams, or `Cipheriv` and `Decipheriv` values, at run time; other values throw a not-supported error. + It may also be a source function, which the pipeline calls with `{ signal }` and whose iterable result it pumps, or a `{ readable, writable }` pair, whose readable side it pumps. + Later `pipeline` stages that arrive as untyped values may be native streams, `Cipheriv` and `Decipheriv` values, functions, and iterables. + A function is called with the previous stage and `{ signal }` as a function stage: before the last stage it must return an async iterable, and as the last stage it may return a promise, whose value resolves the pipeline, or an async iterable, which feeds the PassThrough that `pipeline()` returns. + An untyped source function, function stage, or destination that returns a stream is unsupported. + An iterable becomes `Duplex.from` of it, which the next stage reads from while the stages before it are not read, as in Node. + A pipeline branches at run time on whether each of its first three untyped stages after the first is a function; a function in a later untyped stage, an untyped web stream, and a `{ readable, writable }` pair after the first stage throw a not-supported error. + `for await` sources that arrive as untyped values must be native streams, or `Cipheriv` and `Decipheriv` values, at run time; other values throw a not-supported error. Native streams expose no `emit` to JavaScript, so a JavaScript stream whose own `pipe()` emits on its destination (the legacy `Stream` pipe) cannot pipe into one. `Readable.from` accepts strings, Buffers, arrays, Sets, Maps, Map and Set iterators, generators, async generators, and class iterables and async iterables, and in JavaScript any iterable or async iterable. In JavaScript, a sync iterator value with a `then` method that is not a promise, and an object whose `[Symbol.asyncIterator]` or `[Symbol.iterator]` method is a generator method, are unsupported. diff --git a/packages/cli/test/native-link-info.test.ts b/packages/cli/test/native-link-info.test.ts index f63cc790c..24db2c4c2 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: 45, marker: "scr_runtime_abi_v45" }, + runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" }, 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_v45"); + expect(error?.stderr).toContain("scr_runtime_abi_v46"); }); }); diff --git a/packages/compiler/src/backend/llvm/lib-dispatch.ts b/packages/compiler/src/backend/llvm/lib-dispatch.ts index 8cefda8f0..30eef4b99 100644 --- a/packages/compiler/src/backend/llvm/lib-dispatch.ts +++ b/packages/compiler/src/backend/llvm/lib-dispatch.ts @@ -671,11 +671,13 @@ export function emitGenericLibCall(host: LlvmEmitterContext, e: LibCallExpr): Ll // which also takes those streams. e.fn === "stream.unboxDyn" && host.usesDynBuiltinStream ? "scr_dyn_builtin_stream_unbox" - : // for await over a checked-dynamic value iterates a native - // stream behind it (scr_dyn_for_await_stream.c) - e.fn === "dyn.asyncIteratorOpen" && host.dynForAwaitStreams - ? "scr_stream_dyn_async_iterator_open" - : LIB_FN_SYMS[e.fn]); + : e.fn === "stream.unboxDynStage" && host.usesDynBuiltinStream + ? "scr_dyn_builtin_stream_unbox_stage" + : // for await over a checked-dynamic value iterates a native + // stream behind it (scr_dyn_for_await_stream.c) + e.fn === "dyn.asyncIteratorOpen" && host.dynForAwaitStreams + ? "scr_stream_dyn_async_iterator_open" + : LIB_FN_SYMS[e.fn]); if (sym === undefined) throw new LlvmUnsupportedError(`libCall:${e.fn}`, e.loc); // The runtime's Date functions take and answer time values: a Date // operand passes its time value, read at once from the borrowed object, diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index 4b3f34b80..016815819 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -818,6 +818,9 @@ export const LIB_FN_SYMS: Record = { "stream.stdioWritable": "scr_stream_stdio_writable", "stream.unboxDyn": "scr_stream_unbox_dyn", "stream.dynIterableSource": "scr_stream_dyn_iterable_source", + "stream.dynSourceValue": "scr_stream_dyn_source_value", + "stream.unboxDynStage": "scr_stream_unbox_dyn_stage", + "stream.dynTransformResult": "scr_stream_dyn_transform_result", "stream.handleStream": "scr_stream_handle", "stream.pipeDyn": "scr_stream_pipe_dyn", "stream.unpipeDyn": "scr_stream_unpipe_dyn", diff --git a/packages/compiler/src/backend/llvm/lib-stream.ts b/packages/compiler/src/backend/llvm/lib-stream.ts index fdbf13d4f..53f455964 100644 --- a/packages/compiler/src/backend/llvm/lib-stream.ts +++ b/packages/compiler/src/backend/llvm/lib-stream.ts @@ -743,7 +743,7 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV const helpers = packed(5 + n + k, helperCount); let at = 5 + n + k + helperCount; const take = (present: boolean): string => (present ? args[at++]!.name : "null"); - const box = take(((flags >> 1) & 3) === 2); + const box = take(((flags >> 1) & 3) >= 2); const signal = take((flags & 32) !== 0); const abort = take((flags & 32) !== 0); const head = diff --git a/packages/compiler/src/backend/runtime-abi.ts b/packages/compiler/src/backend/runtime-abi.ts index fe13879a2..892e9281e 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 = 45 as const; -export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v45" as const; +export const RUNTIME_ABI_VERSION = 46 as const; +export const RUNTIME_ABI_MARKER = "scr_runtime_abi_v46" 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 492b71217..9c5b61f58 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: 45, marker: "scr_runtime_abi_v45" }, + runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" }, 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 fdb0ba29e..812fd9531 100644 --- a/packages/compiler/src/backend/runtime-pack.test.ts +++ b/packages/compiler/src/backend/runtime-pack.test.ts @@ -129,7 +129,7 @@ async function fixture(target: NativeTargetSpec = MACOS_ARM64_TARGET) { object_format: target.objectFormat, minimum_os: target.minimumOs, }, - runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" }, + runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" }, 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 c8699d856..2ae952aaa 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -2837,20 +2837,24 @@ export function streamClassAliasDecl( const PIPELINE_STAGE_EXPECTED = "of type function or an instance of Blob, ReadableStream, WritableStream, Stream, Iterable, AsyncIterable, or Promise or { readable, writable } pair"; -/** A checked-dynamic pipeline stage (an untyped source or destination): - * a boxed native stream unwraps at run time; other values (functions, - * pairs) throw the not-supported Error there. */ +/** A checked-dynamic pipeline stage taken as a stream: a boxed native + * stream unwraps at run time, and a value that is not stream-like or a + * function (an iterable after the first stage) becomes Duplex.from of it, + * as pipelineImpl's last branch does. Stream-like objects, web streams and + * { readable, writable } pairs throw the not-supported Error there, and so + * do functions, which reach this lowering only past + * MAX_DYN_FUNCTION_STAGES. */ function unboxDynStage(lowerer: Lowerer, v: IrExpr, node: ts.Expression): IrExpr { noteUntypedStreamFlow(lowerer, null); return { kind: "libCall", - fn: "stream.unboxDyn", + fn: "stream.unboxDynStage", args: [ v, strLit( lowerer.dynNodeStreams - ? "pipeline stages other than native streams (JavaScript stream objects, iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet" - : "pipeline stages other than native streams (iterables, generators, plain functions, and { readable, writable } pairs) are not supported yet", + ? "pipeline stages other than native streams, iterables and functions (JavaScript stream objects, web streams, { readable, writable } pairs after the first stage, and functions after the first three untyped stages) are not supported yet" + : "pipeline stages other than native streams, iterables and functions (web streams, { readable, writable } pairs after the first stage, and functions after the first three untyped stages) are not supported yet", locOf(node), ), ], @@ -2930,6 +2934,102 @@ const BOXED_SOURCE_KINDS: ReadonlySet = new Set([ "string", ]); +/** The most checked-dynamic stages after the first that a pipeline + * branches on as functions; each one doubles the lowered pipeline calls. */ +const MAX_DYN_FUNCTION_STAGES = 3; + +/** A pipeline's arguments for one of the run-time branches + * lowerDynStagePipeline lowers. */ +interface DynStagePlan { + /** the first stage when it is checked-dynamic */ + source: DynPipelineSource | null; + /** argument values evaluated before the branches, in argument order */ + bound: ReadonlyMap; + /** a typed iterable first stage's source, lowered before the branches */ + preSource?: { closure: IrExpr; out: SyncGenType } | undefined; + /** the checked-dynamic stages this branch takes as functions */ + fns: ReadonlySet; +} + +/** pipeline() whose stages after the first include untyped values: + * pipelineImpl calls a stage that is a function with the previous stage + * and { signal }, and pipes into or takes any other value, so the call is + * lowered once per combination of those stages being functions or not, + * up to MAX_DYN_FUNCTION_STAGES of them, and a typeof test on each + * picks the branch at run time. The arguments other than function stages + * are evaluated first, in argument order. */ +function lowerDynStagePipeline( + lowerer: Lowerer, + stageArgs: readonly ts.Expression[], + src: DynPipelineSource | null, + loc: SrcLoc, + build: (plan: DynStagePlan) => IrExpr, +): IrExpr { + const candidates: number[] = []; + for (let j = 1; j < stageArgs.length; j++) { + const arg = stageArgs[j]!; + if (pipelineStageFunction(lowerer, arg) !== null || isValueStream(lowerer, arg)) continue; + if ((lowerer.typeOf(arg).flags & (ts.TypeFlags.Any | ts.TypeFlags.Unknown)) !== 0) { + candidates.push(j); + } + } + if (candidates.length === 0) return build({ source: src, bound: new Map(), fns: new Set() }); + const stmts: IrStmt[] = []; + const bind = (init: IrExpr): IrExpr => { + const local = lowerer.declareHiddenLocal("%pipelineArg", init.type); + stmts.push({ kind: "varDecl", localId: local.id, init, loc: init.loc }); + return varRef(local.id, init.type, init.loc); + }; + const source = src !== null && !src.iterable ? { expr: bind(src.expr), iterable: false } : src; + const bound = new Map(); + let preSource: DynStagePlan["preSource"]; + for (let j = src !== null ? 1 : 0; j < stageArgs.length; j++) { + const arg = stageArgs[j]!; + if (pipelineStageFunction(lowerer, arg) !== null) continue; + if (j === 0 && iterableSourceType(lowerer, arg) !== null) { + const lowered = lowerIterableSource(lowerer, arg, null, loc); + if (lowered === null) lowerer.noLowering("pipeline over this source", arg); + preSource = { closure: bind(lowered!.closure), out: lowered!.out }; + continue; + } + let value = lowerer.lowerExpr(arg); + // an untyped array, string, Set, Map or generator object that keeps + // its static type is taken through its checked-dynamic value + if ( + candidates.includes(j) && + BOXED_SOURCE_KINDS.has(value.type.kind) && + lowerer.dynConvertible(value.type) + ) { + value = { kind: "dynFrom", value, type: DYN, loc: value.loc }; + } + bound.set(j, bind(value)); + } + const dynStages = candidates + .filter((j) => bound.get(j)!.type.kind === "dyn") + .slice(0, MAX_DYN_FUNCTION_STAGES); + const branch = (i: number, fns: ReadonlySet): IrExpr => { + if (i === dynStages.length) return build({ source, bound, preSource, fns }); + const j = dynStages[i]!; + const asStream = branch(i + 1, fns); + const asFunction = branch(i + 1, new Set([...fns, j])); + // a function destination's promise resolves to its value + const type = + asFunction.type.kind === "promise" && asFunction.type.inner.kind === "dyn" + ? asFunction.type + : asStream.type; + return { + kind: "ternary", + cond: { kind: "dynTest", test: "function", value: bound.get(j)!, type: BOOL, loc }, + then: lowerer.coerceToExpected(asFunction, type), + else_: lowerer.coerceToExpected(asStream, type), + type, + loc, + }; + }; + const result = branch(0, new Set()); + return { kind: "seqExpr", stmts, result, type: result.type, loc }; +} + /** lowerPipelineStages' source arguments for a checked-dynamic first stage. */ function stagesSource( src: DynPipelineSource | null, @@ -2938,6 +3038,22 @@ function stagesSource( return src.iterable ? [undefined, src.expr] : [new Map([[0, src.expr]]), undefined]; } +/** A pipeline stage taken as a stream under a DynStagePlan: the + * checked-dynamic first stage's stream, a value bound before the + * branches, or the argument lowered here. */ +function planStreamArg( + lowerer: Lowerer, + plan: DynStagePlan, + node: ts.Expression, + i: number, +): IrExpr { + if (i === 0 && plan.source !== null) return plan.source.expr; + const bound = plan.bound.get(i); + return bound !== undefined + ? streamArgOf(lowerer, node, bound, "pipeline") + : lowerStreamArg(lowerer, node, "pipeline"); +} + /** A stream-rooted argument, lowered — or the pointed fence (pipeline's * iterable/generator/function stages have no lowering yet). `what` picks * the validation ladder: finished() rejects EVERY non-stream through @@ -3304,7 +3420,9 @@ export function lowerStreamModuleCall( : null; const stageArgs = opts ? args.slice(0, -1) : args; const build = (src: DynPipelineSource | null): IrExpr => - lowerPromisePipeline(lowerer, stageArgs, opts, src, loc); + lowerDynStagePipeline(lowerer, stageArgs, src, loc, (plan) => + lowerPromisePipeline(lowerer, stageArgs, opts, plan, loc), + ); return lowerDynSourcePipeline(lowerer, stageArgs[0]!, loc, build) ?? build(null); } return null; @@ -3312,17 +3430,23 @@ export function lowerStreamModuleCall( return lowerStreamModuleCallbackCall(lowerer, call, bi, loc); } -/** stream/promises pipeline(...stages[, options]), with the first stage - * pre-lowered when it is checked-dynamic (lowerDynSourcePipeline). */ +/** stream/promises pipeline(...stages[, options]), for one branch of + * lowerDynStagePipeline's plan. */ function lowerPromisePipeline( lowerer: Lowerer, stageArgs: readonly ts.Expression[], opts: ReturnType | null, - src: DynPipelineSource | null, + plan: DynStagePlan, loc: SrcLoc, ): IrExpr { - if (src?.iterable === true || hasPipelineFunctionStage(lowerer, stageArgs)) { - const staged = lowerPipelineStages(lowerer, stageArgs, loc, ...stagesSource(src)); + const src = plan.source; + if ( + src?.iterable === true || + plan.fns.size > 0 || + plan.preSource !== undefined || + hasPipelineFunctionStage(lowerer, stageArgs) + ) { + const staged = lowerPipelineStages(lowerer, stageArgs, loc, ...stagesSource(src), plan); const type = staged.destPromise ?? { kind: "promise", inner: VOID }; // A promise destination's value settles the result (same type). if (!opts) { @@ -3346,9 +3470,7 @@ function lowerPromisePipeline( } let streams: IrExpr[]; try { - streams = stageArgs.map((a, i) => - i === 0 && src !== null ? src.expr : lowerStreamArg(lowerer, a, "pipeline"), - ); + streams = stageArgs.map((a, i) => planStreamArg(lowerer, plan, a, i)); } catch (e) { if (e instanceof StreamArgTypeThrow) return e.expr; throw e; @@ -3530,7 +3652,9 @@ function lowerStreamModuleCallbackCall( ); } const build = (src: DynPipelineSource | null): IrExpr => - lowerCallbackPipeline(lowerer, call, args, src, loc); + lowerDynStagePipeline(lowerer, args.slice(0, -1), src, loc, (plan) => + lowerCallbackPipeline(lowerer, call, args, plan, loc), + ); return lowerDynSourcePipeline(lowerer, args[0]!, loc, build) ?? build(null); } @@ -3544,23 +3668,32 @@ function lowerStreamModuleCallbackCall( return null; } -/** pipeline(...stages, callback), with the first stage pre-lowered when it - * is checked-dynamic (lowerDynSourcePipeline). */ +/** pipeline(...stages, callback), for one branch of lowerDynStagePipeline's + * plan. */ function lowerCallbackPipeline( lowerer: Lowerer, call: ts.CallExpression, args: readonly ts.Expression[], - src: DynPipelineSource | null, + plan: DynStagePlan, loc: SrcLoc, ): IrExpr { - if (src?.iterable === true || hasPipelineFunctionStage(lowerer, args.slice(0, -1))) { - const staged = lowerPipelineStages(lowerer, args.slice(0, -1), loc, ...stagesSource(src)); - const cbNode = args[args.length - 1]!; - const cbParams = - ts.isArrowFunction(cbNode) || ts.isFunctionExpression(cbNode) - ? cbNode.parameters.nodes.length - : 0; - if (staged.dest === 2 || (staged.dest === 1 && cbParams > 1)) { + const src = plan.source; + const cbNode = args[args.length - 1]!; + const cbParams = + ts.isArrowFunction(cbNode) || ts.isFunctionExpression(cbNode) + ? cbNode.parameters.nodes.length + : 0; + // A branch over untyped stages whose destination is a stream: the + // callback still receives (err, value), value undefined, as in Node. + const dynCallback = plan.bound.size > 0 && cbParams > 1; + if ( + src?.iterable === true || + plan.fns.size > 0 || + plan.preSource !== undefined || + hasPipelineFunctionStage(lowerer, args.slice(0, -1)) + ) { + const staged = lowerPipelineStages(lowerer, args.slice(0, -1), loc, ...stagesSource(src), plan); + if (staged.dest >= 2 || (staged.dest === 1 && cbParams > 1) || dynCallback) { // callback(err, value): the destination's value (undefined for // an async generator) reaches the callback through the // checked-dynamic path. @@ -3595,15 +3728,15 @@ function lowerCallbackPipeline( } let streams: IrExpr[]; try { - streams = args - .slice(0, -1) - .map((a, i) => (i === 0 && src !== null ? src.expr : lowerStreamArg(lowerer, a, "pipeline"))); + streams = args.slice(0, -1).map((a, i) => planStreamArg(lowerer, plan, a, i)); } catch (e) { if (e instanceof StreamArgTypeThrow) return e.expr; throw e; } const last = streams[streams.length - 1]!; - const { cb, dyn } = lowerEosCallback(lowerer, "pipeline", args[args.length - 1]!, last.type); + const { cb, dyn } = dynCallback + ? { cb: lowerer.lowerExprExpecting(cbNode, DYN), dyn: true } + : lowerEosCallback(lowerer, "pipeline", cbNode, last.type); return { kind: "libCall", fn: dyn ? "stream.pipelineDyn" : "stream.pipeline", @@ -4213,6 +4346,397 @@ function lowerDynIterableSource( }; } +/** The async generator a checked-dynamic stage's iterable is pumped + * through. */ +function dynStageGenType(): SyncGenType { + return { kind: "generator", async: true, yieldT: DYN, retT: VOID, nextT: UNDEFINED_T }; +} + +/** A checked-dynamic pipeline stage after the first that is a function at + * run time, as the stage closure the runtime calls with the previous + * stage and { signal } (pipelineImpl's `typeof stream === 'function'` + * branch): it calls the function with the previous stream or generator + * as an untyped value. Before the last stage the result must be async + * iterable (ERR_INVALID_RETURN_VALUE otherwise) and is pumped on through + * lowerDynIterableSource's generator; the last stage answers the result, + * which the runtime classifies (SCR_PIPE_DEST_DYN). A previous stage + * whose type has no untyped form throws the not-supported Error. */ +function lowerDynFunctionStage( + lowerer: Lowerer, + fnValue: IrExpr, + srcT: IrType, + j: number, + isLast: boolean, + loc: SrcLoc, + handoff?: DynStageHandoff, +): { closure: IrExpr; out: IrType } { + const out: IrType = isLast ? DYN : dynStageGenType(); + const stageT = funcOf([srcT, DYN], out); + const take = handoff?.take === true; + const give = handoff?.give === true && !isLast; + const key = `dynFunctionStage:${isLast ? "dest" : "transform"}:${typeKey(srcT)}:${take ? 1 : 0}${give ? 1 : 0}`; + let maker = lowerer.valueHelpers.get(key); + if (!maker) { + maker = `%stream.dynFunctionStage.${lowerer.valueHelpers.size}`; + lowerer.valueHelpers.set(key, maker); + const stageName = `${maker}.stage`; + const frame = new LiftedFrame(loc); + const fnId = frame.local("fn", DYN, false); + const labelId = frame.local("label", STRING, false); + const cellId = frame.local("cell", DYN, false); + const srcId = frame.param("src", srcT); + const optsId = frame.param("opts", DYN); + const inputId = frame.local("input", DYN, false); + const resultId = frame.local("result", DYN, false); + const nextId = frame.local("next", DYN, false); + const cell = (): IrExpr => varRef(cellId, DYN, loc); + const body: IrStmt[] = []; + if (take) { + // the previous untyped stage's value, handed over as it is + // (makeAsyncIterable answers an iterable itself) + body.push( + { kind: "varDecl", localId: inputId, init: handoffTake(cell(), loc), loc }, + handoffPut(cell(), dynUndefinedExpr(loc), loc), + ); + } + if (!take && !lowerer.dynConvertible(srcT)) { + body.push({ + kind: "return", + value: nodeThrowExpr( + 0, + "SC2020", + `pipeline function stages after a stage of type '${lowerer.fmt(srcT)}' are not supported yet`, + out, + loc, + ), + loc, + }); + } else { + body.push({ + kind: "varDecl", + localId: resultId, + init: { + kind: "dynCall", + callee: varRef(fnId, DYN, loc), + receiver: dynUndefinedExpr(loc), + calleeName: "transform", + args: [ + take + ? varRef(inputId, DYN, loc) + : { kind: "dynFrom", value: varRef(srcId, srcT, loc), type: DYN, loc }, + varRef(optsId, DYN, loc), + ], + type: DYN, + loc, + }, + loc, + }); + if (isLast) { + body.push({ kind: "return", value: varRef(resultId, DYN, loc), loc }); + } else { + body.push({ + kind: "varDecl", + localId: nextId, + init: { + kind: "libCall", + fn: "stream.dynTransformResult", + args: [varRef(resultId, DYN, loc), varRef(labelId, STRING, loc)], + type: DYN, + loc, + }, + loc, + }); + // the next untyped function stage takes the result itself; the + // generator returned for the runtime is then never read + if (give) body.push(handoffPut(cell(), varRef(nextId, DYN, loc), loc)); + const iterable = lowerDynIterableSource(lowerer, varRef(nextId, DYN, loc), loc); + if (!iterable) + throw new InternalCompilerError("dyn iterable generator without a result record"); + body.push({ + kind: "return", + value: { kind: "callValue", callee: iterable.closure, args: [], type: out, loc }, + loc, + }); + } + } + const captured = take || give ? [fnId, labelId, cellId] : [fnId, labelId]; + for (const id of captured) frame.locals.find((l) => l.id === id)!.boxed = true; + lowerer.liftedFns.push({ + name: stageName, + params: frame.params, + returnType: out, + captures: [ + { localId: fnId, name: "fn", type: DYN }, + { localId: labelId, name: "label", type: STRING }, + ...(take || give ? [{ localId: cellId, name: "cell", type: DYN }] : []), + ], + locals: frame.locals, + body, + loc, + }); + const makerFn: IrLocal = { id: "fn.0", name: "fn", type: DYN, mutable: false, boxed: true }; + const makerLabel: IrLocal = { + id: "label.1", + name: "label", + type: STRING, + mutable: false, + boxed: true, + }; + const makerCell: IrLocal = { + id: "cell.2", + name: "cell", + type: DYN, + mutable: false, + boxed: true, + }; + const makerLocals = take || give ? [makerFn, makerLabel, makerCell] : [makerFn, makerLabel]; + lowerer.liftedFns.push({ + name: maker, + params: makerLocals.map((l) => ({ localId: l.id, name: l.name, type: l.type })), + returnType: stageT, + locals: makerLocals, + body: [ + { + kind: "return", + value: { + kind: "closure", + fnName: stageName, + captures: makerLocals.map((l) => l.id), + type: stageT, + loc, + }, + loc, + }, + ], + loc, + }); + } + return { + closure: { + kind: "call", + callee: maker, + args: [fnValue, strLit(`transform[${j - 1}]`, loc), ...(take || give ? [handoff!.cell] : [])], + type: stageT, + loc, + }, + out, + }; +} + +/** The cell that carries an untyped stage's value to the untyped function + * stage after it, created per pipeline call. A stage that gives stores its + * value; the next stage takes it instead of the generator the runtime + * passes, as Node's makeAsyncIterable answers an iterable itself. */ +interface DynStageHandoff { + cell: IrExpr; + take: boolean; + give: boolean; +} + +function handoffPut(cell: IrExpr, value: IrExpr, loc: SrcLoc): IrStmt { + return { + kind: "exprStmt", + expr: { + kind: "libCall", + fn: "dyn.keySet", + args: [cell, strLit("v", loc), value], + type: VOID, + loc, + }, + loc, + }; +} + +function handoffTake(cell: IrExpr, loc: SrcLoc): IrExpr { + return { kind: "dynKeyGet", value: cell, key: strLit("v", loc), type: DYN, loc }; +} + +/** `(value) => generator over value`: the async generator an untyped + * destination function's async iterable result is pumped through + * (SCR_PIPE_DEST_DYN's box), interned. */ +function dynStageIterHelper(lowerer: Lowerer, loc: SrcLoc): IrExpr { + const genT = dynStageGenType(); + const type = funcOf([DYN], genT); + const key = "dynStageIter"; + let name = lowerer.valueHelpers.get(key); + if (!name) { + name = `%stream.dynStageIter.${lowerer.valueHelpers.size}`; + lowerer.valueHelpers.set(key, name); + const v: IrLocal = { id: "value.0", name: "value", type: DYN, mutable: false }; + const iterable = lowerDynIterableSource(lowerer, varRef(v.id, DYN, loc), loc); + if (!iterable) + throw new InternalCompilerError("dyn iterable generator without a result record"); + lowerer.liftedFns.push({ + name, + params: [{ localId: v.id, name: v.name, type: DYN }], + returnType: genT, + locals: [v], + body: [ + { + kind: "return", + value: { kind: "callValue", callee: iterable.closure, args: [], type: genT, loc }, + loc, + }, + ], + loc, + }); + } + return { kind: "closure", fnName: name, captures: [], type, loc }; +} + +/** A checked-dynamic first pipeline stage pumped as an iterable + * (pipelineImpl's i === 0 branch), as the stage closure the runtime calls + * with { signal } while it sets the pipeline up: a function is called + * with { signal } and its result must be iterable, a { readable, writable } + * pair pumps its readable side, and any other value is pumped itself, + * through lowerDynIterableSource's generator. */ +function lowerDynSourceStage( + lowerer: Lowerer, + value: IrExpr, + loc: SrcLoc, + cell?: IrExpr, +): { closure: IrExpr; out: SyncGenType } | null { + const out: SyncGenType = { + kind: "generator", + async: true, + yieldT: DYN, + retT: VOID, + nextT: UNDEFINED_T, + }; + const stageT = funcOf([DYN], out); + const give = cell !== undefined; + const key = give ? "dynSourceStage:give" : "dynSourceStage"; + let maker = lowerer.valueHelpers.get(key); + if (!maker) { + maker = `%stream.dynSourceStage.${lowerer.valueHelpers.size}`; + const stageName = `${maker}.stage`; + const frame = new LiftedFrame(loc); + const srcId = frame.local("src", DYN, false); + const cellId = frame.local("cell", DYN, false); + const optsId = frame.param("opts", DYN); + const valueId = frame.local("value", DYN, false); + const genId = frame.local("gen", out, false); + const src = (): IrExpr => varRef(srcId, DYN, loc); + const sourceValue = (v: IrExpr, returned: boolean): IrExpr => ({ + kind: "libCall", + fn: "stream.dynSourceValue", + args: [v, boolLit(returned, loc)], + type: DYN, + loc, + }); + const iterable = lowerDynIterableSource(lowerer, varRef(valueId, DYN, loc), loc); + if (!iterable) return null; + lowerer.valueHelpers.set(key, maker); + const body: IrStmt[] = [ + { + kind: "varDecl", + localId: valueId, + init: { + kind: "ternary", + cond: { kind: "dynTest", test: "function", value: src(), type: BOOL, loc }, + then: sourceValue( + { + kind: "dynCall", + callee: src(), + receiver: dynUndefinedExpr(loc), + calleeName: "source", + args: [varRef(optsId, DYN, loc)], + type: DYN, + loc, + }, + true, + ), + else_: sourceValue(src(), false), + type: DYN, + loc, + }, + loc, + }, + { + kind: "varDecl", + localId: genId, + init: { kind: "callValue", callee: iterable.closure, args: [], type: out, loc }, + loc, + }, + ]; + if (give) { + // the untyped function stage after the source takes the iterable + // itself, as Node's makeAsyncIterable answers it; a stream (a pair's + // readable side) goes as the generator over it, as Node iterates a + // stream there + const v = varRef(valueId, DYN, loc); + body.push( + handoffPut( + varRef(cellId, DYN, loc), + { + kind: "ternary", + cond: { kind: "libCall", fn: "stream.dynIterableSource", args: [v], type: BOOL, loc }, + then: v, + else_: { kind: "dynFrom", value: varRef(genId, out, loc), type: DYN, loc }, + type: DYN, + loc, + }, + loc, + ), + ); + } + body.push({ kind: "return", value: varRef(genId, out, loc), loc }); + const captured = give ? [srcId, cellId] : [srcId]; + for (const id of captured) frame.locals.find((l) => l.id === id)!.boxed = true; + lowerer.liftedFns.push({ + name: stageName, + params: frame.params, + returnType: out, + captures: [ + { localId: srcId, name: "src", type: DYN }, + ...(give ? [{ localId: cellId, name: "cell", type: DYN }] : []), + ], + locals: frame.locals, + body, + loc, + }); + const makerSrc: IrLocal = { id: "src.0", name: "src", type: DYN, mutable: false, boxed: true }; + const makerCell: IrLocal = { + id: "cell.1", + name: "cell", + type: DYN, + mutable: false, + boxed: true, + }; + const makerLocals = give ? [makerSrc, makerCell] : [makerSrc]; + lowerer.liftedFns.push({ + name: maker, + params: makerLocals.map((l) => ({ localId: l.id, name: l.name, type: l.type })), + returnType: stageT, + locals: makerLocals, + body: [ + { + kind: "return", + value: { + kind: "closure", + fnName: stageName, + captures: makerLocals.map((l) => l.id), + type: stageT, + loc, + }, + loc, + }, + ], + loc, + }); + } + return { + closure: { + kind: "call", + callee: maker, + args: give ? [value, cell!] : [value], + type: stageT, + loc, + }, + out, + }; +} + function lowerIterableSource( lowerer: Lowerer, node: ts.Expression, @@ -4356,7 +4880,7 @@ function lowerIterableSource( interface PipelineStagesLowering { args: IrExpr[]; last: IrType; - dest: 0 | 1 | 2; + dest: 0 | 1 | 2 | 3; destPromise: (IrType & { kind: "promise" }) | null; wrap: (call: IrExpr) => IrExpr; } @@ -4377,6 +4901,7 @@ function lowerPipelineStages( loc: SrcLoc, preLowered?: ReadonlyMap, dynSource?: IrExpr, + plan?: DynStagePlan, ): PipelineStagesLowering { if (stageArgs.length > 52) { lowerer.noLowering( @@ -4390,7 +4915,7 @@ function lowerPipelineStages( const helperArgs: IrExpr[] = []; let positions = 0; let mask = 0; - let dest: 0 | 1 | 2 = 0; + let dest: 0 | 1 | 2 | 3 = 0; let destPromise: (IrType & { kind: "promise" }) | null = null; // What the next function stage receives: null before the first stage, // else the previous stream's type or the previous stage's generator. @@ -4399,17 +4924,69 @@ function lowerPipelineStages( // before all stages, so when a stage precedes a stream the arguments // are bound to locals first, keeping JavaScript's evaluation order. const argOrder: { list: IrExpr[]; index: number }[] = []; + // An untyped stage followed by an untyped function stage hands its value + // over through one cell per call (DynStageHandoff). + let handoffCell: IrLocal | null = null; + const cellRef = (): IrExpr => { + handoffCell ??= lowerer.declareHiddenLocal("%pipelineHandoff", DYN); + return varRef(handoffCell.id, DYN, loc); + }; + let takes = false; for (let j = 0; j < stageArgs.length; j++) { const arg = stageArgs[j]!; + const givesToNext = plan?.fns.has(j + 1) === true; if (j === 0 && dynSource !== undefined) { - const source = lowerDynIterableSource(lowerer, dynSource, loc); + const source = lowerDynSourceStage( + lowerer, + dynSource, + loc, + givesToNext ? cellRef() : undefined, + ); + takes = givesToNext; if (source === null) lowerer.noLowering("pipeline over this source", arg); + // a source function takes { signal } + mask |= 1 << stages.length; positions += 1; argOrder.push({ list: stages, index: stages.length }); stages.push(source!.closure); srcT = source!.out; continue; } + if (plan?.fns.has(j) === true) { + // a checked-dynamic stage that is a function at run time + const isLast = j === stageArgs.length - 1; + const take = takes; + const give = givesToNext && !isLast; + takes = give; + const lowered = lowerDynFunctionStage( + lowerer, + plan.bound.get(j)!, + srcT!, + j, + isLast, + loc, + take || give ? { cell: cellRef(), take, give } : undefined, + ); + mask |= 1 << stages.length; + positions += 2 ** j; + argOrder.push({ list: stages, index: stages.length }); + stages.push(lowered.closure); + if (isLast) { + // its result is classified at run time: a promise settles the + // pipeline's value, and an async iterable is pumped into the + // PassThrough + const genT = dynStageGenType(); + const helpers = fromGenHelpers(lowerer, genT, loc); + if (!helpers) lowerer.noLowering("a pipeline destination with this generator type", arg); + helperArgs.push(helpers![0]!, helpers![1]!, helpers![2]!); + dest = 3; + destPromise = { kind: "promise", inner: DYN }; + srcT = null; + } else { + srcT = lowered.out; + } + continue; + } // compose's head and tail functions arrive as Duplex.from streams const fn = preLowered?.has(j) ? null : pipelineStageFunction(lowerer, arg); const syncSourceFn = @@ -4423,7 +5000,9 @@ function lowerPipelineStages( !preLowered?.has(j) && (syncSourceFn || (fn === null && iterableSourceType(lowerer, arg) !== null)) ) { - const source = lowerIterableSource(lowerer, arg, syncSourceFn ? fn : null, loc); + const source = + (syncSourceFn ? undefined : plan?.preSource) ?? + lowerIterableSource(lowerer, arg, syncSourceFn ? fn : null, loc); if (source === null) lowerer.noLowering("pipeline over this source", arg); positions += 1; argOrder.push({ list: stages, index: stages.length }); @@ -4448,7 +5027,12 @@ function lowerPipelineStages( } let stream: IrExpr; try { - stream = preLowered?.get(j) ?? lowerStreamArg(lowerer, arg, "pipeline"); + const bound = plan?.bound.get(j); + stream = + preLowered?.get(j) ?? + (bound !== undefined + ? streamArgOf(lowerer, arg, bound, "pipeline") + : lowerStreamArg(lowerer, arg, "pipeline")); } catch (e) { if (e instanceof StreamArgTypeThrow) lowerer.noLowering("pipeline over this value", arg); throw e; @@ -4527,6 +5111,8 @@ function lowerPipelineStages( ); } helperArgs.push(pipelineBoxHelper(lowerer, destPromise!, loc)); + } else if (dest === 3) { + helperArgs.push(dynStageIterHelper(lowerer, loc)); } const argDecls: IrStmt[] = []; const firstStage = argOrder.findIndex((a) => a.list === stages); @@ -4538,6 +5124,15 @@ function lowerPipelineStages( a.list[a.index] = varRef(local.id, init.type, init.loc); } } + if (handoffCell !== null) { + const cell: IrLocal = handoffCell; + argDecls.unshift({ + kind: "varDecl", + localId: cell.id, + init: { kind: "dynObjLit", fields: [], type: DYN, loc }, + loc, + }); + } const bindArgs = (call: IrExpr): IrExpr => argDecls.length === 0 ? call diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts index b0cc96687..86f9ad233 100644 --- a/packages/compiler/src/ir/builtin-effects.ts +++ b/packages/compiler/src/ir/builtin-effects.ts @@ -92,6 +92,9 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "child.stdioStream", "stream.unboxDyn", "stream.dynIterableSource", + "stream.dynSourceValue", + "stream.unboxDynStage", + "stream.dynTransformResult", "stream.handleStream", "stream.pipeDyn", "stream.unpipeDyn", diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index e230932ba..b7ef7d660 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -2870,9 +2870,27 @@ function buildLibFnSignatures() { result: { kind: "object", className: "%Duplex" }, }, /** Whether pipeline() pumps a checked-dynamic first stage as an - * iterable (a string, or a value with an iterator method that is not a - * stream with pipe() and on()) instead of piping it as a stream. */ + * iterable (a string, a value with an iterator method that is not a + * stream with pipe() and on(), a { readable, writable } pair, or a + * source function's result) instead of piping it as a stream. */ "stream.dynIterableSource": { argTypes: [DYN], result: BOOL }, + /** The value pipeline() pumps for a checked-dynamic first stage: a + * source function's result (the flag set), which must be iterable, else + * a { readable, writable } pair's readable side or the value itself. */ + "stream.dynSourceValue": { argTypes: [DYN, BOOL], result: DYN }, + /** An untyped pipeline function stage's result before the last stage, + * which must be async iterable: ERR_INVALID_RETURN_VALUE naming the + * stage ("transform[i]") otherwise. Answers the value (+1). */ + "stream.dynTransformResult": { argTypes: [DYN, STRING], result: DYN }, + /** A checked-dynamic pipeline() stage taken as a stream: a native stream + * (+1), or Duplex.from of a value that is not stream-like or a function + * (an iterable), which the pipeline uses without piping into it. Other + * values throw: Duplex.from's ERR_INVALID_ARG_TYPE, or the given + * not-supported message. */ + "stream.unboxDynStage": { + argTypes: [DYN, STRING], + result: { kind: "object", className: "%Duplex" }, + }, /** The Transform a Cipheriv or Decipheriv reads and writes as a stream * (+1, created on first use and owned by the cipher). */ "stream.handleStream": { @@ -5237,10 +5255,12 @@ export type IrLibFn = keyof ReturnType; /** The function-stage runs of a pipeline of `total` stages that a stream * follows (positions bit j: stage j is a function), and an async * generator destination's run (flags bits 1-2 = 1, pumped into the - * PassThrough pipeline() answers): each pumps its last generator through - * one (next, read, return) triple. */ + * PassThrough pipeline() answers, or 3, an untyped destination whose + * result may be pumped there): each pumps its last generator through one + * (next, read, return) triple. */ export function pipelinePumpedRuns(total: number, positions: number, flags: number): number { - let runs = ((flags >> 1) & 3) === 1 ? 1 : 0; + const dest = (flags >> 1) & 3; + let runs = dest === 1 || dest === 3 ? 1 : 0; for (let j = 0; j + 1 < total; j++) { if (Math.floor(positions / 2 ** j) % 2 === 1 && Math.floor(positions / 2 ** (j + 1)) % 2 === 0) runs++; diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts index 494c4e3e8..630ebdc67 100644 --- a/packages/compiler/src/ir/runtime-features.ts +++ b/packages/compiler/src/ir/runtime-features.ts @@ -278,6 +278,9 @@ const STREAM_LIB_FNS: ReadonlySet = new Set([ "stream.stdioWritable", "stream.unboxDyn", "stream.dynIterableSource", + "stream.dynSourceValue", + "stream.dynTransformResult", + "stream.unboxDynStage", "stream.handleStream", "stream.unpipeDyn", "stream.workerStdio", @@ -314,6 +317,7 @@ const READABLE_DYN_ASYNC_LIB_FNS: ReadonlySet = new Set([ "stream.duplexFromAsyncFn", "stream.duplexFromGenSignal", "stream.duplexFromAsyncFnSignal", + "stream.unboxDynStage", ]); /** 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 45ebf5c79..84751cc06 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/packages/compiler/src/ir/validate.ts @@ -5761,7 +5761,7 @@ function validateFunction( } const lastIsFunction = Math.floor(positions! / 2 ** (total - 1)) % 2 === 1; const helperCount = 3 * pipelinePumpedRuns(total, positions!, flags!); - const groups = helperCount + (destKind === 2 ? 1 : 0) + (flags! & 32 ? 2 : 0); + const groups = helperCount + (destKind >= 2 ? 1 : 0) + (flags! & 32 ? 2 : 0); const cbArgs = e.fn === "sp.pipelineStagesOpts" ? 1 + (flags! & 64 ? 1 : 0) @@ -5776,7 +5776,7 @@ function validateFunction( total > 52 || positions! >= 2 ** total || functions !== k || - destKind === 3 || + (destKind === 3 && e.fn === "stream.composeStages") || (destKind !== 0) !== lastIsFunction || total < 2 || e.args.length !== 5 + n! + k! + groups + cbArgs @@ -5795,7 +5795,7 @@ function validateFunction( err(`libCall ${e.fn} stage function ${i} must be a func`, e.loc); } at2 += k!; - const helpers = helperCount + (destKind === 2 ? 1 : 0); + const helpers = helperCount + (destKind >= 2 ? 1 : 0); for (let i = at2; i < at2 + helpers; i++) { if (e.args[i]?.type.kind !== "func") err(`libCall ${e.fn} helper ${i} must be a func`, e.loc); diff --git a/packages/compiler/test/native-codegen-integration.test.ts b/packages/compiler/test/native-codegen-integration.test.ts index a57d96f01..bdad62203 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_v45"); + expect(await symbols(helperObject, ["-u"])).toContain("_scr_runtime_abi_v46"); 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_v45"); + expect(undefinedSymbols).toContain("_scr_runtime_abi_v46"); }); 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/7508-pipeline-untyped-source-functions.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7508-pipeline-untyped-source-functions.mjs.json new file mode 100644 index 000000000..4027063d4 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7508-pipeline-untyped-source-functions.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7508-pipeline-untyped-source-functions.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7509-pipeline-untyped-function-stages.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7509-pipeline-untyped-function-stages.mjs.json new file mode 100644 index 000000000..028a5c1c6 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7509-pipeline-untyped-function-stages.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7509-pipeline-untyped-function-stages.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7510-pipeline-untyped-later-iterables.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7510-pipeline-untyped-later-iterables.mjs.json new file mode 100644 index 000000000..be18c801a --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7510-pipeline-untyped-later-iterables.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7510-pipeline-untyped-later-iterables.mjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7511-pipeline-untyped-stages.cjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7511-pipeline-untyped-stages.cjs.json new file mode 100644 index 000000000..2353cc44c --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7511-pipeline-untyped-stages.cjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7511-pipeline-untyped-stages.cjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7512-pipeline-any-stages.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7512-pipeline-any-stages.ts.json new file mode 100644 index 000000000..acea0e98d --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7512-pipeline-any-stages.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7512-pipeline-any-stages.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7513-pipeline-untyped-stage-errors.mjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7513-pipeline-untyped-stage-errors.mjs.json new file mode 100644 index 000000000..39a5dba57 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/7513-pipeline-untyped-stage-errors.mjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/7513-pipeline-untyped-stage-errors.mjs" + ], + "diags": [] +} diff --git a/packages/runtime-darwin-arm64/scripts/build.mjs b/packages/runtime-darwin-arm64/scripts/build.mjs index 3305d69de..f8f5d0bf1 100644 --- a/packages/runtime-darwin-arm64/scripts/build.mjs +++ b/packages/runtime-darwin-arm64/scripts/build.mjs @@ -366,7 +366,7 @@ async function build() { package: packageManifest.name, version: packageManifest.version, target: RUNTIME_PACK_MATRIX.target, - runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" }, + runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" }, compiler: { command: compiler, identity: compilerVersion, diff --git a/packages/runtime-pack-common/scripts/build.mjs b/packages/runtime-pack-common/scripts/build.mjs index 8c0863b04..ed32f43fb 100644 --- a/packages/runtime-pack-common/scripts/build.mjs +++ b/packages/runtime-pack-common/scripts/build.mjs @@ -469,7 +469,7 @@ async function build() { package: packageManifest.name, version: packageManifest.version, target: matrix.target, - runtime_abi: { version: 45, marker: "scr_runtime_abi_v45" }, + runtime_abi: { version: 46, marker: "scr_runtime_abi_v46" }, compiler: { command: compiler, identity: compilerVersion, diff --git a/packages/runtime/src/scr_console.c b/packages/runtime/src/scr_console.c index 67d4aca15..18c715cb2 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_v45(void) {} +void scr_runtime_abi_v46(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 206463e18..92392897a 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_v45(void); +void scr_runtime_abi_v46(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_dyn.h b/packages/runtime/src/scr_dyn.h index 2d916a825..63dc0d209 100644 --- a/packages/runtime/src/scr_dyn.h +++ b/packages/runtime/src/scr_dyn.h @@ -1548,6 +1548,9 @@ ScrDyn *scr_dyn_builtin_stream_object(struct ScrStream *s); ScrDyn *scr_dyn_builtin_stream_view(struct ScrStream *s); struct ScrStream *scr_dyn_builtin_stream_check(const ScrDyn *value, double kind); struct ScrStream *scr_dyn_builtin_stream_unbox(const ScrDyn *value, const ScrStr *message); +/* stream.unboxDynStage in those programs (scr_stream_unbox_dyn_stage's + * contract, also taking constructor-function streams). */ +struct ScrStream *scr_dyn_builtin_stream_unbox_stage(const ScrDyn *value, const ScrStr *message); /* stream.dynNative in those programs: scr_stream_dyn_native's contract, * also taking constructor-function streams that keep the runtime's write * and end (+1 or NULL). */ diff --git a/packages/runtime/src/scr_dyn_builtin_stream.c b/packages/runtime/src/scr_dyn_builtin_stream.c index 2875cfa06..52c1e7269 100644 --- a/packages/runtime/src/scr_dyn_builtin_stream.c +++ b/packages/runtime/src/scr_dyn_builtin_stream.c @@ -1752,3 +1752,11 @@ ScrStream *scr_dyn_builtin_stream_unbox(const ScrDyn *value, const ScrStr *messa } return scr_stream_unbox_dyn(value, message); } + +/* stream.unboxDynStage in programs that link this unit: a + * constructor-function stream as scr_dyn_builtin_stream_unbox takes it, + * any other value as scr_stream_unbox_dyn_stage does. */ +ScrStream *scr_dyn_builtin_stream_unbox_stage(const ScrDyn *value, const ScrStr *message) { + if (bcs_of(value) && value->kind == SCR_DYN_OBJ) return scr_dyn_builtin_stream_unbox(value, message); + return scr_stream_unbox_dyn_stage(value, message); +} diff --git a/packages/runtime/src/scr_stream.c b/packages/runtime/src/scr_stream.c index 201d6e34b..a7593b551 100644 --- a/packages/runtime/src/scr_stream.c +++ b/packages/runtime/src/scr_stream.c @@ -93,6 +93,10 @@ struct ScrStreamState { bool is_transform; /* the writable side routes through transform/flush */ bool allow_half_open; bool auto_destroy, emit_close; + /* Duplex.from of a pipeline() argument that is not a stream: the + * pipeline takes it as that stage's stream without wiring the stage + * (scr_stream_unbox_dyn_stage) */ + bool pipeline_body; /* shared lifecycle */ bool destroyed, destroy_calling, error_emitted, close_emitted, error_scheduled, close_scheduled; @@ -3850,7 +3854,9 @@ typedef ScrPromise *(*ScrPipeNextFn)(ScrClosure *, ScrGen *); typedef ScrDyn *(*ScrPipeReadFn)(ScrClosure *, void *); typedef ScrDyn *(*ScrPipeBoxFn)(ScrClosure *, ScrPromise *); -enum { SCR_PIPE_DEST_STREAM, SCR_PIPE_DEST_GEN, SCR_PIPE_DEST_PROMISE }; +/* SCR_PIPE_DEST_DYN: an untyped destination function's result, a + * checked-dynamic value the pipeline classifies (scr_pipe_dest_dyn_kind) */ +enum { SCR_PIPE_DEST_STREAM, SCR_PIPE_DEST_GEN, SCR_PIPE_DEST_PROMISE, SCR_PIPE_DEST_DYN }; /* flags bit 7: the callback is typed code's (stream.pipelineStagesDyn) */ enum { SCR_PIPE_TYPED_CB = 128 }; @@ -3868,7 +3874,9 @@ typedef struct { * async generator destination's PassThrough included), in pipeline * order, or NULL */ ScrClosure **helpers; - ScrClosure *box; /* SCR_PIPE_DEST_PROMISE: boxes its promise as a dyn promise */ + /* SCR_PIPE_DEST_PROMISE: boxes its promise as a dyn promise; + * SCR_PIPE_DEST_DYN: answers the async generator over a dyn iterable */ + ScrClosure *box; } ScrPipeStages; static bool scr_pipe_is_stage(const ScrPipeStages *fs, size_t j) { @@ -4234,6 +4242,71 @@ static void scr_pipe_dest_promise(ScrStreamPipeCtx *ctx, ScrPromise *p /*moves*/ scr_promise_release(scr_async_spawn_after(d->boxed->v.promise, &scr_pipe_dest_entry, d)); } +/* An untyped destination's promise: scr_pipe_dest_promise over a promise + * that already holds dyn values. */ +static void scr_pipe_dest_dyn_promise(ScrStreamPipeCtx *ctx, ScrDyn *boxed /*moves*/, size_t pi, + bool end) { + ScrPipeDest *d = malloc(sizeof *d); + if (!d) scr_stream_oom(); + d->ctx = scr_stream_pipectx_retain_v(ctx); + d->typed = scr_promise_retain(boxed->v.promise); + d->boxed = boxed; + d->pi = pi; + d->end = end; + scr_promise_release(scr_async_spawn_after(d->boxed->v.promise, &scr_pipe_dest_entry, d)); +} + +typedef ScrGen *(*ScrPipeIterFn)(ScrClosure *, ScrDyn *); +static bool scr_stream_dyn_has_method(const ScrDyn *value, const char *key); +static bool scr_stream_dyn_transform_like(const ScrDyn *value); +static void scr_stream_throw_web_stage(void); +static bool scr_stream_dyn_stage_returned_stream(const ScrDyn *value); +static void scr_stream_throw_invalid_return(const char *expected, const char *name, + const ScrDyn *got); + +/* What pipelineImpl does with an untyped destination function's result: + * a promise settles the pipeline's value (the `then` branch), and an + * async iterable, or a { readable, writable } pair's readable side, is + * pumped into the PassThrough. *out answers the promise or that + * iterable's async generator (+1). Other values throw + * ERR_INVALID_RETURN_VALUE, and thenables that are not promises and web + * streams throw the not-supported Error; -1 then. */ +enum { SCR_PIPE_DYN_PROMISE, SCR_PIPE_DYN_ITER }; +static int scr_pipe_dest_dyn_kind(const ScrPipeStages *fs, ScrDyn *r, void **out) { + if (r->kind == SCR_DYN_PROMISE) { + *out = scr_dyn_retain(r); + return SCR_PIPE_DYN_PROMISE; + } + if (r->kind == SCR_DYN_OBJ && scr_stream_dyn_has_method(r, "then")) { + static const char msg[] = + "pipeline destinations that return a thenable other than a promise are not supported yet"; + scr_throw_error_msg_code(SCR_ERR_ERROR, msg, sizeof msg - 1, "SC2020"); + return -1; + } + if (scr_exc_pending() || scr_stream_dyn_stage_returned_stream(r)) return -1; + ScrDyn *read = NULL; + int kind = scr_dyn_iterable_kind(r); + if (kind < 0) return -1; + if (r->kind == SCR_DYN_HANDLE && + (r->v.handle.tag == SCR_DYNH_WEB_STREAM || r->v.handle.tag == SCR_DYNH_WEB_TRANSFORM)) { + scr_stream_throw_web_stage(); + return -1; + } + if (kind == 2) { + read = scr_dyn_retain(r); + } else if (scr_stream_dyn_transform_like(r)) { + read = scr_dyn_obj_read(r, "readable", 8); + } + if (!read) { + if (!scr_exc_pending()) scr_stream_throw_invalid_return("AsyncIterable or Promise", "destination", r); + return -1; + } + ScrGen *g = ((ScrPipeIterFn)fs->box->fn)(fs->box, read); + if (!g) return -1; + *out = g; + return SCR_PIPE_DYN_ITER; +} + static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosure *cb, ScrStreamErrInv inv, bool end_opt, ScrDyn *signal, ScrAbortAddFn add, const ScrPipeStages *fs, @@ -4294,8 +4367,18 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu : (void *)scr_stream_retain(ret); ret_gen = NULL; void *out = scr_pipe_call_stage(fs, k++, src); + int dyn_kind = -1; + if (!scr_exc_pending() && dest && fs->dest == SCR_PIPE_DEST_DYN) { + ScrDyn *r = out ? out : scr_dyn_retain(scr_dyn_undefined()); + out = NULL; + dyn_kind = scr_pipe_dest_dyn_kind(fs, r, &out); + scr_dyn_release(r); + } if (scr_exc_pending()) { - if (out && dest) scr_promise_release(out); + if (out && dest && fs->dest == SCR_PIPE_DEST_DYN) { + if (dyn_kind == SCR_PIPE_DYN_ITER) scr_gen_release(out); + else scr_dyn_release(out); + } else if (out && dest) scr_promise_release(out); else if (out) scr_gen_release(out); scr_stream_pipectx_drop(ctx); scr_stream_pipectx_release_v(ctx); @@ -4315,9 +4398,12 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu size_t pi = ctx->n++; ctx->streams[pi] = pt; ctx->finish_count++; - if (fs->dest == SCR_PIPE_DEST_GEN) { + if (fs->dest == SCR_PIPE_DEST_GEN || + (fs->dest == SCR_PIPE_DEST_DYN && dyn_kind == SCR_PIPE_DYN_ITER)) { /* pumpToNode(ret, pt, finish, { end }) */ scr_pump_start(ctx, out, fs->helpers + 3 * h++, pt, end); + } else if (fs->dest == SCR_PIPE_DEST_DYN) { + scr_pipe_dest_dyn_promise(ctx, out, pi, end); } else { scr_pipe_dest_promise(ctx, out, fs->box, pi, end); } @@ -4332,6 +4418,20 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu continue; } ScrStream *s = streams[i]; + if (s->st->pipeline_body) { + /* ret = Duplex.from(stream) for an argument that is not a stream: + * no destroyer or 'error' listener, and the previous stage is + * neither piped nor pumped into it */ + s->st->pipeline_body = false; + if (ret_gen) { + scr_gen_release(ret_gen); + ret_gen = NULL; + h++; + } + ret = s; + i++; + continue; + } bool writing = j > 0; if (reading && !scr_pipe_is_stage(fs, j + 1) && (streams[i + 1]->st->close_emitted || streams[i + 1]->st->destroyed)) { @@ -8593,10 +8693,90 @@ ScrStream *scr_stream_unbox_dyn(const ScrDyn *value, const ScrStr *message) { return NULL; } -/* pipeline.js's first stage: a value with pipe() and on() methods is a - * readable node stream, which the stage ladder tests before isIterable; - * a string, or a value with a Symbol.asyncIterator or Symbol.iterator - * method, is an iterable the pipeline pumps into the next stage. */ +/* Whether a checked-dynamic object's member is of type "object". */ +static bool scr_stream_dyn_member_object(const ScrDyn *value, const char *key) { + ScrDyn *m = scr_dyn_obj_read(value, key, strlen(key)); + if (!m) return false; + ScrStr *t = scr_dyn_typeof(m); + bool object = t->len == 6 && memcmp(t->data, "object", 6) == 0; + scr_str_release(t); + scr_dyn_release(m); + return object; +} + +/* pipeline.js's isTransformStream for a value that is not a stream: an + * object whose readable and writable members are both of type "object". */ +static bool scr_stream_dyn_transform_like(const ScrDyn *value) { + if (value->kind == SCR_DYN_HANDLE) return value->v.handle.tag == SCR_DYNH_WEB_TRANSFORM; + return value->kind == SCR_DYN_OBJ && scr_stream_dyn_member_object(value, "readable") && + !scr_exc_pending() && scr_stream_dyn_member_object(value, "writable"); +} + +/* Whether a checked-dynamic object has a function member `key`. */ +static bool scr_stream_dyn_has_method(const ScrDyn *value, const char *key) { + ScrDyn *m = scr_dyn_obj_read(value, key, strlen(key)); + if (!m) return false; + bool fn = m->kind == SCR_DYN_FUNC; + scr_dyn_release(m); + return fn; +} + +/* pipeline.js's isNodeStream and isWebStream for a checked-dynamic value + * that is not a native stream: a stream-like object or a web stream, which + * the pipeline would pipe or pump into and has no lowering for. */ +static bool scr_stream_dyn_stream_like(const ScrDyn *value) { + if (value->kind == SCR_DYN_HANDLE) { + int tag = value->v.handle.tag; + return tag == SCR_DYNH_WEB_STREAM || tag == SCR_DYNH_WEB_TRANSFORM || tag == SCR_DYNH_WEB_WRITABLE; + } + if (value->kind != SCR_DYN_OBJ) return false; + static const char *const state_keys[] = {"_readableState", "_writableState"}; + for (size_t i = 0; i < 2; i++) { + ScrDyn *v = scr_dyn_obj_read(value, state_keys[i], strlen(state_keys[i])); + bool set = v && scr_dyn_truthy(v); + if (v) scr_dyn_release(v); + if (set) return true; + } + bool on = scr_stream_dyn_has_method(value, "on"); + if (on && (scr_stream_dyn_has_method(value, "write") || scr_stream_dyn_has_method(value, "pipe"))) + return true; + if (scr_stream_dyn_has_method(value, "pipeThrough") && + scr_stream_dyn_has_method(value, "getReader") && scr_stream_dyn_has_method(value, "cancel")) + return true; + if (scr_stream_dyn_has_method(value, "getWriter") && scr_stream_dyn_has_method(value, "abort")) + return true; + return scr_stream_dyn_transform_like(value); +} + +/* A checked-dynamic pipeline() stage that the pipeline takes as a stream + * (the stream.unboxDynStage libCall): a boxed native stream or a stream + * handle answers its stream (+1). Any other value that is not stream-like + * or a function takes pipelineImpl's last branch, Duplex.from(stream): + * an iterable answers a readable Duplex the pipeline uses without piping + * the previous stage into it, and values Duplex.from rejects throw its + * ERR_INVALID_ARG_TYPE. Stream-like objects, web streams and functions + * throw `message` as the not-supported Error. */ +ScrStream *scr_stream_unbox_dyn_stage(const ScrDyn *value, const ScrStr *message) { + ScrStream *stream = scr_stream_of_handle(value); + if (!stream) stream = scr_stream_boxed(value); + if (stream) return stream; + if (value->kind == SCR_DYN_FUNC || scr_stream_dyn_stream_like(value)) { + if (!scr_exc_pending()) + scr_throw_error_msg_code(SCR_ERR_ERROR, message->data, message->len, "SC2020"); + return NULL; + } + if (scr_exc_pending()) return NULL; + ScrStream *d = scr_stream_duplex_from_dyn((ScrDyn *)value); + if (d) d->st->pipeline_body = true; + return d; +} + +/* pipeline.js's first stage: a function is a source function, which the + * pipeline calls; a value with pipe() and on() methods is a readable node + * stream, which the stage ladder tests before isIterable; a string, a + * value with a Symbol.asyncIterator or Symbol.iterator method, or a + * { readable, writable } pair is an iterable the pipeline pumps into the + * next stage. */ bool scr_stream_dyn_iterable_source(const ScrDyn *value) { ScrStream *stream = scr_stream_of_handle(value); if (!stream) stream = scr_stream_boxed(value); @@ -8604,7 +8784,8 @@ bool scr_stream_dyn_iterable_source(const ScrDyn *value) { scr_stream_release(stream); return false; } - if (value->kind == SCR_DYN_STR) return true; + if (value->kind == SCR_DYN_STR || value->kind == SCR_DYN_FUNC) return true; + if (value->kind == SCR_DYN_HANDLE && value->v.handle.tag == SCR_DYNH_WEB_TRANSFORM) return true; if (value->kind == SCR_DYN_OBJ) { ScrDyn *pipe = scr_dyn_obj_read(value, "pipe", 4); if (!pipe) return false; @@ -8617,6 +8798,79 @@ bool scr_stream_dyn_iterable_source(const ScrDyn *value) { scr_dyn_release(on); if (streamish) return false; } + if (scr_stream_dyn_transform_like(value)) return true; + if (scr_exc_pending()) return false; } return scr_dyn_iterable_kind(value) > 0; } + +/* Untyped web streams as pipeline stages have no lowering. */ +static void scr_stream_throw_web_stage(void) { + static const char msg[] = "pipeline stages that are untyped web streams are not supported yet"; + scr_throw_error_msg_code(SCR_ERR_ERROR, msg, sizeof msg - 1, "SC2020"); +} + +/* A native stream returned by an untyped pipeline function, which Node + * pipes or passes on as the stream: throws the not-supported Error. */ +static bool scr_stream_dyn_stage_returned_stream(const ScrDyn *value) { + ScrStream *stream = scr_stream_of_handle(value); + if (!stream) stream = scr_stream_boxed(value); + if (!stream) return false; + scr_stream_release(stream); + static const char msg[] = "pipeline function stages that return a stream are not supported yet"; + scr_throw_error_msg_code(SCR_ERR_ERROR, msg, sizeof msg - 1, "SC2020"); + return true; +} + +/* Node's ERR_INVALID_RETURN_VALUE TypeError: "Expected to be + * returned from the "" function but got ." */ +static void scr_stream_throw_invalid_return(const char *expected, const char *name, + const ScrDyn *got) { + char detail[160]; + const char *type = scr_dyn_specific_type(got, detail, sizeof detail); + char msg[384]; + int n = snprintf(msg, sizeof msg, "Expected %s to be returned from the \"%s\" function but got %s.", + expected, name, type); + if (n < 0) n = 0; + if ((size_t)n >= sizeof msg) n = (int)sizeof msg - 1; + scr_throw_error_msg_code(SCR_ERR_TYPE, msg, (size_t)n, "ERR_INVALID_RETURN_VALUE"); +} + +/* An untyped pipeline function stage's result before the last stage + * (pipelineImpl's `isIterable(ret, true)` check): an async iterable + * answers itself (+1); other values throw ERR_INVALID_RETURN_VALUE naming + * the stage, as `label` ("transform[i]") does. */ +ScrDyn *scr_stream_dyn_transform_result(const ScrDyn *value, const ScrStr *label) { + if (scr_stream_dyn_stage_returned_stream(value)) return NULL; + int kind = scr_dyn_iterable_kind(value); + if (kind < 0) return NULL; + if (kind != 2) { + scr_stream_throw_invalid_return("AsyncIterable", label->data, value); + return NULL; + } + return scr_dyn_retain((ScrDyn *)value); +} + +/* The value a checked-dynamic first stage pumps (pipelineImpl's i === 0 + * branch): for `returned`, a source function's result, which must be + * iterable; otherwise a { readable, writable } pair's readable side, or + * the value itself. +1, or NULL with an exception pending. */ +ScrDyn *scr_stream_dyn_source_value(const ScrDyn *value, bool returned) { + if (!returned) { + if (value->kind == SCR_DYN_HANDLE && value->v.handle.tag == SCR_DYNH_WEB_TRANSFORM) { + scr_stream_throw_web_stage(); + return NULL; + } + if (scr_stream_dyn_transform_like(value)) return scr_dyn_obj_read(value, "readable", 8); + if (scr_exc_pending()) return NULL; + return scr_dyn_retain((ScrDyn *)value); + } + if (scr_stream_dyn_stage_returned_stream(value)) return NULL; + int kind = scr_dyn_iterable_kind(value); + if (kind < 0) return NULL; + if (kind == 0) { + scr_stream_throw_invalid_return("Iterable, AsyncIterable or Stream", "source", value); + return NULL; + } + return scr_dyn_retain((ScrDyn *)value); +} diff --git a/packages/runtime/src/scr_stream.h b/packages/runtime/src/scr_stream.h index 7acad0eeb..c10978298 100644 --- a/packages/runtime/src/scr_stream.h +++ b/packages/runtime/src/scr_stream.h @@ -149,6 +149,18 @@ ScrStream *scr_stream_unbox_dyn(const struct ScrDyn *value, const ScrStr *messag * to pump rather than as a stream (false with an exception pending when * reading the value's members throws). */ bool scr_stream_dyn_iterable_source(const struct ScrDyn *value); +/* A checked-dynamic pipeline() stage the pipeline takes as a stream: a + * native stream (+1), or Duplex.from of a value that is not stream-like + * or a function, which the pipeline uses without piping into it; NULL + * with an exception otherwise (`message` for the unsupported shapes). */ +ScrStream *scr_stream_unbox_dyn_stage(const struct ScrDyn *value, const ScrStr *message); +/* The iterable pipeline() pumps for a checked-dynamic first stage: a + * source function's result when `returned`, else a { readable, writable } + * pair's readable side or the value (+1, or NULL with an exception). */ +struct ScrDyn *scr_stream_dyn_source_value(const struct ScrDyn *value, bool returned); +/* An untyped function stage's result before the last pipeline stage: the + * async iterable itself (+1), or NULL with ERR_INVALID_RETURN_VALUE. */ +struct ScrDyn *scr_stream_dyn_transform_result(const struct ScrDyn *value, const ScrStr *label); /* Handles that are Transforms (ScrDynHandleOps.transform: Cipheriv and * Decipheriv). scr_stream_of_handle: the handle's Transform (+1), built on * first use, or NULL when the value is not such a handle. diff --git a/tests/corpus/7508-pipeline-untyped-source-functions.mjs b/tests/corpus/7508-pipeline-untyped-source-functions.mjs new file mode 100644 index 000000000..987221ffd --- /dev/null +++ b/tests/corpus/7508-pipeline-untyped-source-functions.mjs @@ -0,0 +1,85 @@ +// pipeline() over untyped first stages that are functions or +// { readable, writable } pairs: a source function is called with +// { signal } while the pipeline is set up and its iterable result is +// pumped; a result that is not iterable throws ERR_INVALID_RETURN_VALUE +// from the callback form and rejects the promise form; a pair pumps its +// readable side. +import { pipeline, Writable, Readable, PassThrough } from "node:stream"; +import { pipeline as pipelineP } from "node:stream/promises"; + +function pick(v) { + return v; +} +function sink(label) { + return new Writable({ + objectMode: true, + write(chunk, _enc, cb) { + console.log(label, "write", JSON.stringify(String(chunk))); + cb(); + }, + final(cb) { + console.log(label, "final"); + cb(); + }, + }); +} + +const arraySource = pick(function (opts) { + console.log("array source called", typeof opts, typeof opts.signal, opts.signal.aborted); + return ["a1", "a2"]; +}); +const before = pipelineP(arraySource, sink("array")); +console.log("after pipeline() returned"); +await before; +console.log("array resolved"); + +await pipelineP(pick(() => "xy"), sink("string")); + +await pipelineP( + pick(async function* ({ signal }) { + console.log("generator source", typeof signal.aborted); + yield "g1"; + await new Promise((r) => setTimeout(r, 1)); + yield "g2"; + }), + pick(async function* (source) { + for await (const chunk of source) yield `[${chunk}]`; + }), + sink("generator"), +); + +await new Promise((resolve) => { + pipeline(pick(() => new Set(["s1", "s2"])), sink("set"), (err) => { + console.log("set done", err); + resolve(undefined); + }); +}); + +// a result that is not iterable +try { + pipeline(pick(() => 5), sink("five"), () => console.log("five callback")); +} catch (err) { + console.log("five threw", err.name, err.code, err.message); +} +try { + await pipelineP(pick(() => ({ value: 1 })), sink("object")); +} catch (err) { + console.log("object rejected", err.name, err.code, err.message); +} + +// a source function that throws +try { + await pipelineP( + pick(() => { + throw new Error("source failed"); + }), + sink("throws"), + ); +} catch (err) { + console.log("throws rejected", err.message); +} + +// a { readable, writable } pair pumps its readable side +const side = new PassThrough(); +await pipelineP(pick({ readable: Readable.from(["p1", "p2"]), writable: side }), sink("pair")); +console.log("done"); diff --git a/tests/corpus/7509-pipeline-untyped-function-stages.mjs b/tests/corpus/7509-pipeline-untyped-function-stages.mjs new file mode 100644 index 000000000..18749e55a --- /dev/null +++ b/tests/corpus/7509-pipeline-untyped-function-stages.mjs @@ -0,0 +1,112 @@ +// pipeline() with untyped stages after the first that are functions: a +// transform is called with the previous stream or generator and +// { signal } while the pipeline is set up, and must return an async +// iterable; an async function destination's value resolves the promise +// form and reaches the callback; an async generator destination feeds +// the PassThrough that pipeline() returns. Results of the wrong kind throw +// ERR_INVALID_RETURN_VALUE naming the stage. A stage after an untyped +// source function or transform receives that stage's own iterator, as +// Node's makeAsyncIterable answers an iterable itself. +import { pipeline, Writable, Readable, PassThrough } from "node:stream"; +import { pipeline as pipelineP } from "node:stream/promises"; + +function pick(v) { + return v; +} +function sink(label) { + return new Writable({ + objectMode: true, + write(chunk, _enc, cb) { + console.log(label, "write", JSON.stringify(String(chunk))); + cb(); + }, + }); +} +const upper = pick(async function* (source, { signal }) { + console.log("upper called", source instanceof Readable, typeof signal.aborted); + for await (const chunk of source) yield String(chunk).toUpperCase(); +}); +const join = pick(async (source) => { + let text = ""; + for await (const chunk of source) text += chunk; + return text; +}); + +await pipelineP(Readable.from(["a", "b"]), upper, sink("transform")); +console.log("joined", await pipelineP(Readable.from(["x", "y"]), upper, join)); +console.log("two transforms", await pipelineP(pick(["m", "n"]), upper, upper, join)); + +await new Promise((resolve) => { + pipeline(Readable.from(["c1", "c2"]), upper, join, (err, value) => { + console.log("callback", err, value); + resolve(undefined); + }); +}); + +// an async generator destination +const wrap = pick(async function* (source) { + for await (const chunk of source) yield `<${chunk}>`; +}); +const out = pipeline(Readable.from(["p1", "p2"]), wrap, (err, value) => { + console.log("generator destination done", err, value); +}); +console.log("returns a PassThrough", out instanceof PassThrough); +for await (const chunk of out) console.log("passthrough", chunk); + +// a destination whose promise rejects +try { + await pipelineP( + Readable.from(["r"]), + pick(async (source) => { + for await (const chunk of source) console.log("rejecting after", chunk); + throw new Error("destination failed"); + }), + ); +} catch (err) { + console.log("rejected", err.message); +} + +// results of the wrong kind +try { + await pipelineP(Readable.from(["i"]), pick(() => 5), sink("number")); +} catch (err) { + console.log("transform result", err.name, err.code, err.message); +} +try { + await pipelineP(Readable.from(["i"]), upper, pick(() => ["sync"]), sink("array")); +} catch (err) { + console.log("second transform result", err.code, err.message); +} +try { + pipeline(Readable.from(["i"]), pick(() => 7), () => console.log("never")); +} catch (err) { + console.log("destination result", err.name, err.code, err.message); +} +// the iterators handed from stage to stage +let transformIter = null; +const handed = await pipelineP( + pick(function* () { + yield "s1"; + yield "s2"; + }), + pick((source) => { + // a sync generator, not an async iterator over it + console.log( + "transform gets the source's iterator", + typeof source[Symbol.asyncIterator], + typeof source[Symbol.iterator], + ); + transformIter = (async function* () { + for await (const chunk of source) yield `${chunk}!`; + })(); + return transformIter; + }), + pick(async (source) => { + console.log("destination gets the transform's iterator", source === transformIter); + let text = ""; + for await (const chunk of source) text += chunk; + return text; + }), +); +console.log("handed", handed); +console.log("done"); diff --git a/tests/corpus/7510-pipeline-untyped-later-iterables.mjs b/tests/corpus/7510-pipeline-untyped-later-iterables.mjs new file mode 100644 index 000000000..7fe6e27b3 --- /dev/null +++ b/tests/corpus/7510-pipeline-untyped-later-iterables.mjs @@ -0,0 +1,64 @@ +// pipeline() with an untyped stage after the first that is not a stream or +// a function: Node takes Duplex.from of it, so an iterable becomes the +// readable stage the next stream is piped from, the stages before it are +// not read and are destroyed when the pipeline finishes, and a value +// Duplex.from rejects throws ERR_INVALID_ARG_TYPE. +import { pipeline, Writable, Readable, Duplex } from "node:stream"; +import { pipeline as pipelineP } from "node:stream/promises"; + +function pick(v) { + return v; +} +function sink(label) { + return new Writable({ + objectMode: true, + write(chunk, _enc, cb) { + console.log(label, "write", JSON.stringify(String(chunk))); + cb(); + }, + }); +} +function watched(label, values) { + const source = Readable.from(values); + source.on("error", (err) => console.log(label, "error", err.code)); + source.on("close", () => console.log(label, "close", source.destroyed)); + return source; +} + +await pipelineP(watched("first", ["unread"]), pick(["z1", "z2"]), sink("array")); +console.log("array resolved"); +await new Promise((r) => setTimeout(r, 5)); + +await new Promise((resolve) => { + pipeline(watched("second", ["unread"]), pick("text"), sink("string"), (err) => { + console.log("string done", err); + resolve(undefined); + }); +}); +await new Promise((r) => setTimeout(r, 5)); + +// a generator stage before the iterable is never read +await pipelineP( + Readable.from(["s"]), + pick(async function* (source) { + console.log("generator stage called"); + for await (const chunk of source) yield chunk; + }), + pick(new Set(["set1", "set2"])), + sink("set"), +); +console.log("set resolved"); + +// the iterable as the last stage is what pipeline() returns +const last = pipeline(Readable.from(["u"]), pick(["tail"]), () => console.log("never")); +console.log(last instanceof Duplex, last.readable, last.writable); +for await (const chunk of last) console.log("last", chunk); + +for (const value of [42, { plain: true }, null]) { + try { + await pipelineP(Readable.from(["i"]), pick(value), sink("rejected")); + } catch (err) { + console.log(err.name, err.code, err.message); + } +} +console.log("done"); diff --git a/tests/corpus/7511-pipeline-untyped-stages.cjs b/tests/corpus/7511-pipeline-untyped-stages.cjs new file mode 100644 index 000000000..124372e12 --- /dev/null +++ b/tests/corpus/7511-pipeline-untyped-stages.cjs @@ -0,0 +1,50 @@ +// A CommonJS helper that passes its untyped parameters to pipeline(): the +// same call runs over streams, function stages and iterables, whichever +// each argument is at run time. +const { pipeline, Transform, Writable, Readable } = require("node:stream"); + +function run(label, source, middle, destination) { + return new Promise((resolve) => { + pipeline(source, middle, destination, (err, value) => { + console.log(label, "done", err ? err.code || err.message : err, value); + resolve(undefined); + }); + }); +} +function sink(label) { + return new Writable({ + objectMode: true, + write(chunk, _enc, cb) { + console.log(label, "write", String(chunk)); + cb(); + }, + }); +} +const double = new Transform({ + objectMode: true, + transform(chunk, _enc, cb) { + cb(null, String(chunk) + String(chunk)); + }, +}); +async function* exclaim(source) { + for await (const chunk of source) yield `${chunk}!`; +} +async function count(source) { + let n = 0; + for await (const _ of source) n++; + return n; +} + +(async () => { + await run("streams", Readable.from(["a", "b"]), double, sink("streams")); + await run("function middle", Readable.from(["c"]), exclaim, sink("function middle")); + await run("function ends", () => ["d", "e", "f"], exclaim, count); + await run("iterable middle", Readable.from(["skipped"]), ["g"], sink("iterable middle")); + await run("generator into stream", ["h"], exclaim, new Transform({ + objectMode: true, + transform(chunk, _enc, cb) { + console.log("transform saw", chunk); + cb(null, chunk); + }, + }).on("data", (d) => console.log("data", d))); +})(); diff --git a/tests/corpus/7512-pipeline-any-stages.ts b/tests/corpus/7512-pipeline-any-stages.ts new file mode 100644 index 000000000..a2039a533 --- /dev/null +++ b/tests/corpus/7512-pipeline-any-stages.ts @@ -0,0 +1,45 @@ +// pipeline() in TypeScript with stages after the first typed any or +// unknown: functions run as function stages, streams are piped, and an +// iterable is taken as Duplex.from of it. +import { pipeline, Readable, Writable, PassThrough } from "node:stream"; +import { pipeline as pipelineP } from "node:stream/promises"; + +function sink(label: string): Writable { + return new Writable({ + objectMode: true, + write(chunk: unknown, _enc, cb) { + console.log(label, JSON.stringify(chunk)); + cb(); + }, + }); +} + +async function main(): Promise { + const shout: any = async function* (source: any) { + for await (const chunk of source) yield String(chunk).toUpperCase(); + }; + await pipelineP(Readable.from(["a", "b"]), shout, sink("function")); + + const total: unknown = async (source: any) => { + let sum = 0; + for await (const chunk of source) sum += Number(chunk); + return sum; + }; + const sum = await pipelineP(Readable.from(["1", "2", "3"]), total as any); + console.log("sum", sum); + + const relay: any = new PassThrough({ objectMode: true }); + await pipelineP(Readable.from([{ k: 1 }]), relay, sink("stream")); + + const values: any = ["v1", "v2"]; + await pipelineP(Readable.from(["unread"]), values, sink("iterable")); + + await new Promise((resolve) => { + pipeline(Readable.from(["x"]), shout, sink("callback"), (err: NodeJS.ErrnoException | null) => { + console.log("callback done", err == null); + resolve(); + }); + }); +} + +main().then(() => console.log("done")); diff --git a/tests/corpus/7513-pipeline-untyped-stage-errors.mjs b/tests/corpus/7513-pipeline-untyped-stage-errors.mjs new file mode 100644 index 000000000..ade4370b4 --- /dev/null +++ b/tests/corpus/7513-pipeline-untyped-stage-errors.mjs @@ -0,0 +1,86 @@ +// Untyped pipeline function stages next to typed stages: an untyped stage +// after a typed generator stage or a Transform, the { signal } an untyped +// stage receives aborting when the pipeline fails, and an untyped stage +// that throws while the pipeline is set up. +import { pipeline, Writable, Readable, Transform } from "node:stream"; +import { pipeline as pipelineP } from "node:stream/promises"; + +function pick(v) { + return v; +} +function sink(label) { + return new Writable({ + objectMode: true, + write(chunk, _enc, cb) { + console.log(label, "write", JSON.stringify(String(chunk))); + cb(); + }, + }); +} + +await pipelineP( + Readable.from(["a", "b"]), + async function* (source) { + for await (const chunk of source) yield `${chunk}1`; + }, + pick(async function* (source) { + for await (const chunk of source) yield `${chunk}2`; + }), + sink("after generator"), +); + +await pipelineP( + Readable.from(["q"]), + new Transform({ + transform(chunk, _enc, cb) { + cb(null, chunk); + }, + }), + pick(async (source) => { + for await (const chunk of source) console.log("destination saw", String(chunk)); + }), +); + +let seen; +try { + await pipelineP( + Readable.from(["x", "y"]), + pick(async function* (source, { signal }) { + seen = signal; + signal.addEventListener("abort", () => console.log("signal abort event")); + for await (const chunk of source) yield chunk; + }), + new Writable({ + write(_chunk, _enc, cb) { + cb(new Error("write failed")); + }, + }), + ); +} catch (err) { + console.log("rejected", err.message, seen.aborted); +} + +try { + pipeline( + Readable.from(["t"]), + pick(() => { + throw new Error("stage threw"); + }), + sink("never"), + () => console.log("never called"), + ); +} catch (err) { + console.log("callback form threw", err.message); +} +try { + await pipelineP( + Readable.from(["t"]), + pick(() => { + throw new Error("stage threw again"); + }), + sink("never"), + ); +} catch (err) { + console.log("promise form rejected", err.message); +} +console.log("done"); diff --git a/tests/harness/corpus-costs/7508-pipeline-untyped-source-functions.mjs.json b/tests/harness/corpus-costs/7508-pipeline-untyped-source-functions.mjs.json new file mode 100644 index 000000000..805df1394 --- /dev/null +++ b/tests/harness/corpus-costs/7508-pipeline-untyped-source-functions.mjs.json @@ -0,0 +1 @@ +{ "plain": 4900, "san": 6400 } diff --git a/tests/harness/corpus-costs/7509-pipeline-untyped-function-stages.mjs.json b/tests/harness/corpus-costs/7509-pipeline-untyped-function-stages.mjs.json new file mode 100644 index 000000000..93a0d0f3b --- /dev/null +++ b/tests/harness/corpus-costs/7509-pipeline-untyped-function-stages.mjs.json @@ -0,0 +1 @@ +{ "plain": 6100, "san": 8500 } diff --git a/tests/harness/corpus-costs/7510-pipeline-untyped-later-iterables.mjs.json b/tests/harness/corpus-costs/7510-pipeline-untyped-later-iterables.mjs.json new file mode 100644 index 000000000..e1920ef92 --- /dev/null +++ b/tests/harness/corpus-costs/7510-pipeline-untyped-later-iterables.mjs.json @@ -0,0 +1 @@ +{ "plain": 5100, "san": 8600 } diff --git a/tests/harness/corpus-costs/7511-pipeline-untyped-stages.cjs.json b/tests/harness/corpus-costs/7511-pipeline-untyped-stages.cjs.json new file mode 100644 index 000000000..eafd03265 --- /dev/null +++ b/tests/harness/corpus-costs/7511-pipeline-untyped-stages.cjs.json @@ -0,0 +1 @@ +{ "plain": 5300, "san": 6900 } diff --git a/tests/harness/corpus-costs/7512-pipeline-any-stages.ts.json b/tests/harness/corpus-costs/7512-pipeline-any-stages.ts.json new file mode 100644 index 000000000..ac0e8497d --- /dev/null +++ b/tests/harness/corpus-costs/7512-pipeline-any-stages.ts.json @@ -0,0 +1 @@ +{ "plain": 4500, "san": 6100 } diff --git a/tests/harness/corpus-costs/7513-pipeline-untyped-stage-errors.mjs.json b/tests/harness/corpus-costs/7513-pipeline-untyped-stage-errors.mjs.json new file mode 100644 index 000000000..f363efcf0 --- /dev/null +++ b/tests/harness/corpus-costs/7513-pipeline-untyped-stage-errors.mjs.json @@ -0,0 +1 @@ +{ "plain": 4700, "san": 6300 }