diff --git a/docs/content/limitations/builtin-apis/streams-and-readline.mdx b/docs/content/limitations/builtin-apis/streams-and-readline.mdx index b03151da9..b0768b01d 100644 --- a/docs/content/limitations/builtin-apis/streams-and-readline.mdx +++ b/docs/content/limitations/builtin-apis/streams-and-readline.mdx @@ -1 +1 @@ -- **Streams and readline:** Stream chunks are Buffers or strings. `Readable.from` accepts arrays, generators and async generators whose values are strings or Buffers; a stream that mixes strings and Buffers is unsupported. Object-mode streams of other values, `Readable.from` over other iterables, and generator stages in `pipeline` are unsupported. `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. `readable.compose` is unsupported. `stream.compose` accepts two or more streams, or one `Duplex`; functions, iterables, and web streams as 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. `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags. Stream constructors accept a `signal` option in an options object literal, and `stream.addAbortSignal` accepts any stream. `node:readline` and `node:readline/promises` interfaces read `process.stdin` or a readable stream without terminal mode: `question`, `prompt`, `write`, `pause`, `resume`, `'line'`/`'close'`/`'pause'`/`'resume'`/`'error'` listeners, and `for await` compile. Terminal mode, `completer`, history options, and chained listener registration are unsupported. +- **Streams and readline:** Stream chunks are Buffers or strings. `Readable.from` accepts arrays, generators and async generators whose values are strings or Buffers; a stream that mixes strings and Buffers is unsupported. Object-mode streams of other values and `Readable.from` over other iterables are unsupported. `pipeline` accepts one run of async generator function stages between streams (`async function* (source)`, iterating `source` with `for await`); functions as the source or destination, stages that take the `{ signal }` argument, and function stages together with pipeline options are unsupported. `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. `readable.compose` is unsupported. `stream.compose` accepts two or more streams, or one `Duplex`; functions, iterables, and web streams as 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. `pipeline` from `node:stream/promises` accepts a trailing `{ signal, end }` object, and `finished` accepts `readable`, `writable`, `error`, `cleanup`, and `signal` options, with boolean literals for the flags. Stream constructors accept a `signal` option in an options object literal, and `stream.addAbortSignal` accepts any stream. `node:readline` and `node:readline/promises` interfaces read `process.stdin` or a readable stream without terminal mode: `question`, `prompt`, `write`, `pause`, `resume`, `'line'`/`'close'`/`'pause'`/`'resume'`/`'error'` listeners, and `for await` compile. Terminal mode, `completer`, history options, and chained listener registration are unsupported. diff --git a/internal/compatibility/generated/node-v24-backlog.json b/internal/compatibility/generated/node-v24-backlog.json index 934ecadc1..0270f047e 100644 --- a/internal/compatibility/generated/node-v24-backlog.json +++ b/internal/compatibility/generated/node-v24-backlog.json @@ -61644,7 +61644,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -61675,7 +61677,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -61706,7 +61710,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -61737,7 +61743,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -61768,7 +61776,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -61799,7 +61809,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62216,7 +62228,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62272,7 +62286,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62303,7 +62319,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62334,7 +62352,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62365,7 +62385,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62396,7 +62418,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62813,7 +62837,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62894,7 +62920,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62952,7 +62980,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -62983,7 +63013,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63014,7 +63046,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63070,7 +63104,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63606,7 +63642,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63637,7 +63675,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63668,7 +63708,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63699,7 +63741,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63730,7 +63774,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63761,7 +63807,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63792,7 +63840,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63823,7 +63873,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -63854,7 +63906,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -65146,7 +65200,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -65302,7 +65358,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } @@ -65333,7 +65391,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] } } diff --git a/internal/compatibility/generated/node-v24-internal.json b/internal/compatibility/generated/node-v24-internal.json index 0bc785724..8737e922e 100644 --- a/internal/compatibility/generated/node-v24-internal.json +++ b/internal/compatibility/generated/node-v24-internal.json @@ -85812,7 +85812,9 @@ "tests/corpus/5342-stream-compose.ts", "tests/corpus/5343-stream-compose-edges.ts", "tests/corpus/5420-readable-from-generators.ts", - "tests/corpus/5421-readable-from-generators.cjs" + "tests/corpus/5421-readable-from-generators.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "chapter" @@ -85934,7 +85936,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -85964,7 +85968,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -85994,7 +86000,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -86047,7 +86055,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -86077,7 +86087,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -86107,7 +86119,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -86496,7 +86510,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -86549,7 +86565,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -86579,7 +86597,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -86632,7 +86652,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "ancestor" @@ -86662,7 +86684,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "ancestor" @@ -86692,7 +86716,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "ancestor" @@ -87081,7 +87107,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "ancestor" @@ -87180,7 +87208,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -87236,7 +87266,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -87266,7 +87298,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -87296,7 +87330,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -87349,7 +87385,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -87938,7 +87976,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -87968,7 +88008,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -87998,7 +88040,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -88028,7 +88072,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -88058,7 +88104,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -88088,7 +88136,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -88118,7 +88168,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -88148,7 +88200,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -88178,7 +88232,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -89593,7 +89649,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "exact" @@ -89853,7 +89911,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "ancestor" @@ -89883,7 +89943,9 @@ "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", - "tests/corpus/5303-stream-finished-callback-opts.cjs" + "tests/corpus/5303-stream-finished-callback-opts.cjs", + "tests/corpus/5425-pipeline-generator-stage.ts", + "tests/corpus/5428-pipeline-generator-stages.cjs" ] }, "anchorSource": "ancestor" diff --git a/internal/compatibility/static-support.json b/internal/compatibility/static-support.json index 3c3c53dab..0b06b0a2a 100644 --- a/internal/compatibility/static-support.json +++ b/internal/compatibility/static-support.json @@ -350,7 +350,7 @@ { "symbols": ["AsyncLocalStorage", "asyncLocalStorage.run", "asyncLocalStorage.exit", "asyncLocalStorage.getStore", "asyncLocalStorage.enterWith", "asyncLocalStorage.disable"], "status": "partial", "evidence": ["tests/corpus/2213-async-local-storage.cjs"] }, { "symbols": ["StringDecoder", "stringDecoder.write", "stringDecoder.end"], "status": "partial", "evidence": ["tests/corpus/1662-string-decoder-encodings.ts", "tests/corpus/node24-string-decoder.ts"] }, { "symbols": ["setImmediate", "setInterval", "setTimeout", "clearImmediate", "clearInterval", "clearTimeout", "timersPromises.setTimeout", "timersPromises.setImmediate", "timersPromises.setInterval", "immediate.hasRef", "immediate.ref", "immediate.unref", "timeout.hasRef", "timeout.ref", "timeout.unref", "timeout.refresh"], "status": "partial", "evidence": ["tests/corpus/1804-timers-module.ts", "tests/corpus/1805-timer-callback-args.ts", "tests/corpus/2093-timers-promises.ts", "tests/corpus/2852-timers-promises-interval.ts"] }, - { "symbols": ["stream.finished", "stream.pipeline", "stream.getDefaultHighWaterMark", "readable.push", "readable.unshift", "readable.read", "readable.pause", "readable.resume", "readable.isPaused", "readable.setEncoding", "readable.pipe", "readable.unpipe", "readable.destroy", "writable.write", "writable.end", "writable.cork", "writable.uncork", "writable.setDefaultEncoding"], "status": "partial", "evidence": ["tests/corpus/1685-stream-readable-basics.ts", "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", "tests/corpus/5303-stream-finished-callback-opts.cjs"] }, + { "symbols": ["stream.finished", "stream.pipeline", "stream.getDefaultHighWaterMark", "readable.push", "readable.unshift", "readable.read", "readable.pause", "readable.resume", "readable.isPaused", "readable.setEncoding", "readable.pipe", "readable.unpipe", "readable.destroy", "writable.write", "writable.end", "writable.cork", "writable.uncork", "writable.setDefaultEncoding"], "status": "partial", "evidence": ["tests/corpus/1685-stream-readable-basics.ts", "tests/corpus/1688-stream-writable-basics.ts", "tests/corpus/1814-stream-pipeline.cjs", "tests/corpus/5302-stream-promises-options.ts", "tests/corpus/5303-stream-finished-callback-opts.cjs", "tests/corpus/5425-pipeline-generator-stage.ts", "tests/corpus/5428-pipeline-generator-stages.cjs"] }, { "symbols": ["net.createServer", "net.connect", "net.createConnection", "net.getDefaultAutoSelectFamilyAttemptTimeout", "net.setDefaultAutoSelectFamilyAttemptTimeout", "server.address", "server.close", "server.listen", "socket.destroy", "socket.end", "socket.pause", "socket.resume", "socket.setEncoding", "socket.setTimeout", "socket.write"], "status": "partial", "evidence": ["tests/corpus/2500-net-autosel-timeout.ts", "tests/corpus/2596-net-arg-ladders.cjs"] }, { "symbols": ["dgram.createSocket", "socket.address", "socket.bind", "socket.close", "socket.connect", "socket.ref", "socket.remoteAddress", "socket.send", "socket.unref"], "status": "partial", "evidence": ["tests/corpus/2597-dgram-send-ladders.cjs"] }, { "symbols": ["readline.createInterface", "rl.close"], "status": "partial", "evidence": ["tests/corpus/1475-readline-closed-stdin.ts"] }, diff --git a/packages/compiler/src/backend/llvm/lib-dispatch.ts b/packages/compiler/src/backend/llvm/lib-dispatch.ts index f7e67383a..86cfb79a4 100644 --- a/packages/compiler/src/backend/llvm/lib-dispatch.ts +++ b/packages/compiler/src/backend/llvm/lib-dispatch.ts @@ -440,6 +440,7 @@ export function emitLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlValue { if (e.fn === "process.nextTick") return host.emitAsyncContextLibCall(e); if ( e.fn === "sp.pipeline" || + e.fn === "sp.pipelineStages" || e.fn.startsWith("sp.pipelineOpts") || e.fn.startsWith("sp.finishedOpts") ) diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index f0f1d211a..0ac7bd185 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -1380,6 +1380,9 @@ export const USES_TIMERS_LIB_FNS = new Set([ "stream.compose", "sp.finished", "sp.pipeline", + "sp.pipelineStages", + "stream.pipelineStages", + "stream.pipelineStagesDyn", "sp.pipelineOpts", "sp.pipelineOptsSignal", "sp.finishedOpts", diff --git a/packages/compiler/src/backend/llvm/lib-stream.ts b/packages/compiler/src/backend/llvm/lib-stream.ts index ce8ac6d6a..98e9d1b9e 100644 --- a/packages/compiler/src/backend/llvm/lib-stream.ts +++ b/packages/compiler/src/backend/llvm/lib-stream.ts @@ -633,6 +633,65 @@ export function emitStreamLibCall(host: LlvmEmitterContext, e: LibCallExpr): LlV host.emitPendingCheck(); return out; } + if ( + e.fn === "sp.pipelineStages" || + e.fn === "stream.pipelineStages" || + e.fn === "stream.pipelineStagesDyn" + ) { + // (n, at, k, s1..sn, stage1..stagek, next, read, return[, cb]): + // the streams and the stage closures ride stack arrays. + const [nArg, atArg, kArg] = e.args; + if (nArg?.kind !== "numLit" || atArg?.kind !== "numLit" || kArg?.kind !== "numLit") + throw new InternalCompilerError(`llvm emitter bug: ${e.fn} counts not literals`); + const n = nArg.value; + const k = kArg.value; + const args = e.args.map((a) => host.emitExpr(a)); + const packed = (from: number, count: number): string => { + const arr = B.slot(); + B.entryAllocas.push(`${arr} = alloca [${count} x ptr]`); + for (let i = 0; i < count; i++) { + const p = B.tmp(); + B.line(`${p} = getelementptr inbounds [${count} x ptr], ptr ${arr}, i64 0, i64 ${i}`); + B.line(`store ptr ${args[from + i]!.name}, ptr ${p}`); + } + return arr; + }; + const streams = packed(3, n); + const stages = packed(3 + n, k); + const [next, read, ret] = [3 + n + k, 4 + n + k, 5 + n + k].map((i) => args[i]!.name); + const head = `double ${f64Lit(n)}, ptr ${streams}, double ${f64Lit(atArg.value)}, double ${f64Lit(k)}, ptr ${stages}, ptr ${next}, ptr ${read}, ptr ${ret}`; + const t = B.tmp(); + if (e.fn === "sp.pipelineStages") { + host.declare( + `declare ptr @scr_sp_pipeline_stages(double, ptr, double, double, ptr, ptr, ptr, ptr)`, + ); + B.line(`${t} = call ptr @scr_sp_pipeline_stages(${head})`); + } else if (e.fn === "stream.pipelineStagesDyn") { + host.declare( + `declare ptr @scr_stream_pipeline_stages_dyn(double, ptr, double, double, ptr, ptr, ptr, ptr, ptr)`, + ); + B.line( + `${t} = call ptr @scr_stream_pipeline_stages_dyn(${head}, ptr ${args[args.length - 1]!.name})`, + ); + } else { + const cbT = e.args[e.args.length - 1]!.type; + if (cbT.kind !== "func") + throw new InternalCompilerError( + "llvm emitter bug: stream.pipelineStages callback not a func", + ); + host.moveTemp(args[args.length - 1]!); + const thunk = host.streamCbThunkFor("e", cbT); + host.declare( + `declare ptr @scr_stream_pipeline_stages(double, ptr, double, double, ptr, ptr, ptr, ptr, ptr, ptr)`, + ); + B.line( + `${t} = call ptr @scr_stream_pipeline_stages(${head}, ptr ${args[args.length - 1]!.name}, ptr @${thunk})`, + ); + } + const out = host.own({ name: t, type: e.type }); + host.emitPendingCheck(); + return out; + } if (e.fn === "stream.pipeline" || e.fn === "stream.pipelineDyn") { // pipeline(count, s1..sn, cb): the destination answers +1. The // stream list rides a stack array (the C compound literal). diff --git a/packages/compiler/src/frontend/lowering/call-signatures.ts b/packages/compiler/src/frontend/lowering/call-signatures.ts index f24675061..0499223f1 100644 --- a/packages/compiler/src/frontend/lowering/call-signatures.ts +++ b/packages/compiler/src/frontend/lowering/call-signatures.ts @@ -140,6 +140,8 @@ export function generatorMeta( * - `...xs: T[]`: the ABI type is the array; call sites pack the surplus. */ export function paramShape(lowerer: Lowerer, param: ts.ParameterDeclaration): ParamShape { + const stage = lowerer.stageSourceType(param.name); + if (stage) return { type: stage, mode: "required" }; if (isJsSourceFile(param.getSourceFile()) && ts.isSetAccessorDeclaration(param.parent)) return { type: DYN, mode: "required" }; if ( @@ -344,6 +346,8 @@ export function paramShapes( return params.filter((p) => !isThisParameter(p)).map((param) => lowerer.paramShape(param)); const symbols = signature.getParameters(); return params.map((declParam, i) => { + const stage = lowerer.stageSourceType(declParam.name); + if (stage) return { type: stage, mode: "required" as const }; const symbol = symbols[i]; const tsType = symbol ? lowerer.checker.getTypeOfSymbol(symbol) diff --git a/packages/compiler/src/frontend/lowering/lower-async-iteration.ts b/packages/compiler/src/frontend/lowering/lower-async-iteration.ts index daade569b..1bd074dd7 100644 --- a/packages/compiler/src/frontend/lowering/lower-async-iteration.ts +++ b/packages/compiler/src/frontend/lowering/lower-async-iteration.ts @@ -136,7 +136,8 @@ function literalBooleanOption( } function nodeReadablePlan(lowerer: Lowerer, expression: ts.Expression): NodeReadablePlan | null { - const direct = lowerer.mapTypeOf(lowerer.typeOf(expression)); + const direct = + lowerer.stageSourceType(expression) ?? lowerer.mapTypeOf(lowerer.typeOf(expression)); if (direct?.kind === "object") { const sides = streamSidesOf(lowerer, lowerer.classes.get(direct.className)); if (sides === "r" || sides === "rw") { diff --git a/packages/compiler/src/frontend/lowering/lower-stmts.ts b/packages/compiler/src/frontend/lowering/lower-stmts.ts index 5ae5bef1a..835bbc92d 100644 --- a/packages/compiler/src/frontend/lowering/lower-stmts.ts +++ b/packages/compiler/src/frontend/lowering/lower-stmts.ts @@ -7849,7 +7849,9 @@ export function lowerForOf(lowerer: Lowerer, stmt: ts.ForOfStatement): IrStmt { !!stmt.awaitModifier && lowerer.mapTypeOf(lowerer.typeOf(stmt.expression))?.kind === "array"; if (stmt.awaitModifier && !awaitArray) { { - const genT = lowerer.mapTypeOf(lowerer.typeOf(stmt.expression)); + const genT = + lowerer.stageSourceType(stmt.expression) ?? + lowerer.mapTypeOf(lowerer.typeOf(stmt.expression)); if (genT?.kind === "generator" && genT.async) { const iterable = lowerer.lowerExpr(stmt.expression); if (iterable.type.kind === "generator" && iterable.type.async) { diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index 2abd87aeb..546c69727 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -2026,6 +2026,23 @@ export function lowerStreamModuleCall( ? pipelineOptions(lowerer, lastArg) : null; const stageArgs = opts ? args.slice(0, -1) : args; + if (stageArgs.some(isPipelineFunctionStage)) { + if (opts) { + lowerer.noLowering( + "pipeline with function stages and options", + lastArg, + "pipeline(source, async function* (source) { ... }, destination) without options is supported", + ); + } + const staged = lowerPipelineStages(lowerer, stageArgs, loc); + return { + kind: "libCall", + fn: "sp.pipelineStages", + args: staged.args, + type: { kind: "promise", inner: VOID }, + loc, + }; + } let streams: IrExpr[]; try { streams = stageArgs.map((a) => lowerStreamArg(lowerer, a, "pipeline")); @@ -2200,6 +2217,22 @@ export function lowerStreamModuleCall( "the supported form is pipeline(source, ...streams, callback)", ); } + if (args.slice(0, -1).some(isPipelineFunctionStage)) { + const staged = lowerPipelineStages(lowerer, args.slice(0, -1), loc); + const { cb, dyn } = lowerEosCallback( + lowerer, + "pipeline", + args[args.length - 1]!, + staged.last, + ); + return { + kind: "libCall", + fn: dyn ? "stream.pipelineStagesDyn" : "stream.pipelineStages", + args: [...staged.args, cb], + type: staged.last, + loc, + }; + } let streams: IrExpr[]; try { streams = args.slice(0, -1).map((a) => lowerStreamArg(lowerer, a, "pipeline")); @@ -2223,6 +2256,177 @@ export function lowerStreamModuleCall( return null; } +/** A pipeline() argument written as a function: an async generator + * function stage. */ +function isPipelineFunctionStage(node: ts.Expression): boolean { + return ts.isFunctionExpression(node) || ts.isArrowFunction(node); +} + +/** pipeline(s1, ..., stage1, ..., stagek, ..., sn): one run of async + * generator function stages between node streams (pipeline.js). Each + * stage's `source` parameter is typed as what the pipeline passes it: + * makeAsyncIterable answers the previous stream itself (a Readable is + * async iterable) or the previous stage's generator. Answers the libcall + * head [n, at, k, streams..., stages..., next, read, return] and the last + * stream's type. */ +function lowerPipelineStages( + lowerer: Lowerer, + stageArgs: readonly ts.Expression[], + loc: SrcLoc, +): { args: IrExpr[]; last: IrType } { + const first = stageArgs.findIndex(isPipelineFunctionStage); + let end = first; + while (end < stageArgs.length && isPipelineFunctionStage(stageArgs[end]!)) end++; + if (stageArgs.slice(end).some(isPipelineFunctionStage)) { + lowerer.noLowering( + "pipeline with function stages separated by streams", + stageArgs[end]!, + "one run of function stages between two streams is supported", + ); + } + if (first === 0) { + lowerer.noLowering( + "pipeline with a function as its source", + stageArgs[0]!, + "the source must be a stream; Readable.from(generator()) turns a generator into one", + ); + } + if (end === stageArgs.length) { + lowerer.noLowering( + "pipeline with a function as its destination", + stageArgs[end - 1]!, + "the destination must be a stream", + ); + } + const before: IrExpr[] = []; + const after: IrExpr[] = []; + try { + for (const a of stageArgs.slice(0, first)) before.push(lowerStreamArg(lowerer, a, "pipeline")); + } catch (e) { + if (e instanceof StreamArgTypeThrow) + lowerer.noLowering("pipeline over this value", stageArgs[0]!); + throw e; + } + let srcT: IrType = before[before.length - 1]!.type; + const stages: IrExpr[] = []; + for (const node of stageArgs.slice(first, end)) { + const fn = node as ts.FunctionExpression | ts.ArrowFunction; + const isAsyncGen = + ts.isFunctionExpression(fn) && + fn.asteriskToken !== undefined && + (fn.modifiers ?? []).some((m) => m.kind === ts.SyntaxKind.AsyncKeyword); + if (!isAsyncGen) { + lowerer.noLowering( + "pipeline stages that are not async generator functions", + fn, + "write the stage as async function* (source) { for await (const chunk of source) ... }", + ); + } + if (fn.parameters.length > 1) { + lowerer.noLowering( + "pipeline stages that declare the options argument", + fn.parameters[1]!, + "the { signal } argument has no lowering — take only the source", + ); + } + const lowered = lowerStageFunction(lowerer, fn, srcT); + stages.push(lowered.closure); + srcT = lowered.out; + } + try { + for (const a of stageArgs.slice(end)) after.push(lowerStreamArg(lowerer, a, "pipeline")); + } catch (e) { + if (e instanceof StreamArgTypeThrow) + lowerer.noLowering("pipeline over this value", stageArgs[end]!); + throw e; + } + const lastGen = srcT as IrType & { kind: "generator" }; + if (!fromGenChunk(lowerer, lastGen.yieldT)) { + lowerer.noLowering( + `a pipeline stage yielding '${lowerer.fmt(lastGen.yieldT)}'`, + stageArgs[end - 1]!, + "stage values written to a stream are strings or Buffers", + ); + } + const helpers = fromGenHelpers(lowerer, lastGen, loc); + if (!helpers) + lowerer.noLowering("a pipeline stage with this generator type", stageArgs[end - 1]!); + const streams = [...before, ...after]; + return { + args: [ + numLit(streams.length, loc), + numLit(before.length, loc), + numLit(stages.length, loc), + ...streams, + ...stages, + helpers![0]!, + helpers![1]!, + helpers![2]!, + ], + last: streams[streams.length - 1]!.type, + }; +} + +/** An async generator stage, lowered with its `source` parameter typed + * as what the pipeline passes (a stream or the previous stage's + * generator). The parameter may only head `for await` loops. */ +function lowerStageFunction( + lowerer: Lowerer, + fn: ts.FunctionExpression, + srcT: IrType, +): { closure: IrExpr; out: IrType & { kind: "generator" } } { + const param = fn.parameters[0]; + const symbol = + param && ts.isIdentifier(param.name) + ? lowerer.checker.getSymbolAtLocation(param.name) + : undefined; + if (param && !symbol) { + lowerer.noLowering( + "pipeline stage parameters that are patterns", + param, + "name the source parameter", + ); + } + if (symbol && fn.body) { + const visit = (node: ts.Node): void => { + if ( + ts.isIdentifier(node) && + node !== param!.name && + lowerer.checker.getSymbolAtLocation(node) === symbol + ) { + const parent = node.parent; + if ( + !( + parent && + ts.isForOfStatement(parent) && + parent.awaitModifier && + parent.expression === node + ) + ) { + lowerer.noLowering( + "a pipeline stage's source used outside a for await loop head", + node, + "iterate the source with for await (const chunk of source)", + ); + } + } + ts.forEachChild(node, visit); + }; + visit(fn.body); + } + if (symbol) lowerer.stageSourceTypes.set(symbol, srcT); + try { + const closure = lowerer.lowerExpr(fn); + const out = closure.type.kind === "func" ? closure.type.ret : null; + if (out?.kind !== "generator" || !out.async) { + lowerer.noLowering("a pipeline stage that does not return an async generator", fn); + } + return { closure, out: out as IrType & { kind: "generator" } }; + } finally { + if (symbol) lowerer.stageSourceTypes.delete(symbol); + } +} + /** `stream.compose(...streams)` over stream instances: a Duplex writing * to the first stream and reading from the last (scr_stream_compose). The * classes fix which positions can be readable and writable; a stream in diff --git a/packages/compiler/src/frontend/lowering/lowerer.ts b/packages/compiler/src/frontend/lowering/lowerer.ts index 1c8d92a0a..7022aa9db 100644 --- a/packages/compiler/src/frontend/lowering/lowerer.ts +++ b/packages/compiler/src/frontend/lowering/lowerer.ts @@ -1801,6 +1801,11 @@ export class Lowerer { * prepass. This includes explicit annotations: unchecked TS annotations do * not prove that an array property read produced a value. */ readonly runtimeOptionalBindingTypes = new Map(); + /** A pipeline() function stage's `source` parameter: the IR type of what + * the pipeline passes it (the previous stage as an async iterable), set + * while the stage lowers. Parameter typing and `for await` heads read + * it before the checker's `AsyncIterable`. */ + readonly stageSourceTypes = new Map(); /** Set when a match result lowers to execGroups/matchGroups, the inputs * of regex span folding. */ regexGroupResults = false; @@ -9029,6 +9034,16 @@ export class Lowerer { return paramShape(this, param); } + /** The pipeline stage source type bound to `node` (a parameter name or + * a reference to it), or null. */ + stageSourceType(node: ts.Node): IrType | null { + if (this.stageSourceTypes.size === 0) return null; + while (ts.isParenthesizedExpression(node)) node = node.expression; + if (!ts.isIdentifier(node)) return null; + const symbol = this.checker.getSymbolAtLocation(node); + return (symbol && this.stageSourceTypes.get(symbol)) ?? null; + } + checkDefaultParamBodyType(param: ts.ParameterDeclaration, bodyType: IrType): void { return checkDefaultParamBodyType(this, param, bodyType); } diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts index f8226587c..fb3515fb4 100644 --- a/packages/compiler/src/ir/builtin-effects.ts +++ b/packages/compiler/src/ir/builtin-effects.ts @@ -245,6 +245,9 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "stream.compose", "sp.finished", "sp.pipeline", + "sp.pipelineStages", + "stream.pipelineStages", + "stream.pipelineStagesDyn", "sc.text", "sc.json", "sc.buffer", diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 0884138c8..341f641d8 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -3207,6 +3207,8 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ "stream.finishedDyn": { argTypes: [null, DYN], result: VOID }, "stream.pipeline": { argTypes: [F64], result: VOID }, "stream.pipelineDyn": { argTypes: [F64], result: VOID }, + "stream.pipelineStages": { argTypes: [F64, F64, F64], result: VOID }, + "stream.pipelineStagesDyn": { argTypes: [F64, F64, F64], result: VOID }, /** stream.compose(count, s1..sn): the Duplex over a pipeline of the * streams (+1). MAY THROW (ERR_INVALID_ARG_VALUE for a stream that is * no longer readable or writable). */ @@ -3237,6 +3239,11 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ "stream.finishedOpts": { argTypes: [null, F64, F64, BOOL, null], result: VOID }, "stream.finishedOptsSignal": { argTypes: [null, F64, F64, BOOL, DYN, null], result: VOID }, "sp.pipeline": { argTypes: [F64], result: { kind: "promise", inner: VOID } }, + /** pipeline with a run of async generator stages: [n, at, k, s1..sn, + * stage1..stagek, next, read, return] (+ the callback for the stream.* + * forms): the user's stage functions, then the frontend's helpers over + * the last stage's generator (scr_stream_pipeline_stages). */ + "sp.pipelineStages": { argTypes: [F64, F64, F64], result: { kind: "promise", inner: VOID } }, /** node:stream/consumers — the promise consumers over the readable * machinery: (s) → a pending promise settled at the terminal point * with the accumulated result (sc.text: the utf8 decode, sc.json: the diff --git a/packages/compiler/src/ir/runtime-features.ts b/packages/compiler/src/ir/runtime-features.ts index e7edc56bb..8cb35576b 100644 --- a/packages/compiler/src/ir/runtime-features.ts +++ b/packages/compiler/src/ir/runtime-features.ts @@ -221,6 +221,8 @@ function scanRuntimeFeatures(mod: IrModule, stopAt?: keyof RuntimeFeatures): Run fn === "stream.errored" || fn === "sp.finished" || fn === "sp.pipeline" || + fn === "sp.pipelineStages" || + fn.startsWith("stream.pipelineStages") || fn === "console.bind" || fn === "stream.addAbortSignal" || fn === "stream.addAbortSignalFn" || diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts index 9f1d11d30..e4d85c863 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/packages/compiler/src/ir/validate.ts @@ -3961,6 +3961,9 @@ function validateFunction( e.fn === "stream.pipeline" || e.fn === "stream.pipelineDyn" || e.fn === "stream.compose" || + e.fn === "stream.pipelineStages" || + e.fn === "stream.pipelineStagesDyn" || + e.fn === "sp.pipelineStages" || e.fn === "sp.pipeline" || e.fn === "sp.pipelineOpts" || e.fn === "sp.pipelineOptsSignal" || @@ -5222,6 +5225,8 @@ function validateFunction( e.fn === "stream.pipeline" || e.fn === "stream.pipelineDyn" || e.fn === "stream.compose" || + e.fn === "stream.pipelineStages" || + e.fn === "stream.pipelineStagesDyn" || e.fn === "stream.addAbortSignal" || e.fn === "stream.addAbortSignalFn" || e.fn === "stream.finishedOpts" || @@ -5328,6 +5333,28 @@ function validateFunction( if (!isStreamObject(e.type)) err(`libCall ${e.fn} must return a stream`, e.loc); return true; } + if (e.fn === "stream.pipelineStages" || e.fn === "stream.pipelineStagesDyn") { + // (n, at, k, s1..sn, stage1..stagek, next, read, return, cb) + const [n, at, k] = e.args.slice(0, 3).map((a) => (a?.kind === "numLit" ? a.value : -1)); + if (n! < 2 || at! < 1 || at! >= n! || k! < 1 || e.args.length !== 3 + n! + k! + 3 + 1) { + err(`libCall ${e.fn} count/arity mismatch`, e.loc); + return true; + } + for (let i = 3; i < 3 + n!; i++) { + if (!isStreamObject(e.args[i]?.type)) { + err(`libCall ${e.fn} stage ${i - 3} must be a stream-hierarchy object`, e.loc); + } + } + for (let i = 3 + n!; i < e.args.length - 1; i++) { + if (e.args[i]?.type.kind !== "func") + err(`libCall ${e.fn} helper ${i} must be a func`, e.loc); + } + const cbK = e.args[e.args.length - 1]?.type.kind; + if (e.fn === "stream.pipelineStages" ? cbK !== "func" : cbK !== "dyn") { + err(`libCall ${e.fn} callback has the wrong kind`, e.loc); + } + return true; + } if (e.fn === "stream.pipeline" || e.fn === "stream.pipelineDyn") { // (count, s1..sn, cb) → the destination's type. const count = e.args[0]; diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5425-pipeline-generator-stage.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5425-pipeline-generator-stage.ts.json new file mode 100644 index 000000000..11893caad --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5425-pipeline-generator-stage.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5425-pipeline-generator-stage.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5426-pipeline-generator-errors.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5426-pipeline-generator-errors.ts.json new file mode 100644 index 000000000..8388abdf9 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5426-pipeline-generator-errors.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5426-pipeline-generator-errors.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5427-pipeline-generator-backpressure.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5427-pipeline-generator-backpressure.ts.json new file mode 100644 index 000000000..aacd0d055 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5427-pipeline-generator-backpressure.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5427-pipeline-generator-backpressure.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5428-pipeline-generator-stages.cjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5428-pipeline-generator-stages.cjs.json new file mode 100644 index 000000000..6070cc629 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5428-pipeline-generator-stages.cjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5428-pipeline-generator-stages.cjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5429-pipeline-generator-buffers.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5429-pipeline-generator-buffers.ts.json new file mode 100644 index 000000000..a1149659a --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5429-pipeline-generator-buffers.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5429-pipeline-generator-buffers.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_stream.c b/packages/runtime/src/scr_stream.c index 065a94b06..16cc7048f 100644 --- a/packages/runtime/src/scr_stream.c +++ b/packages/runtime/src/scr_stream.c @@ -2992,9 +2992,310 @@ static void scr_pipe_throw(const char *code, const char *msg, int kind) { scr_throw_obj(e, &scr_error_retain_v, &scr_error_release_v, scr_error_trace_arg()); } +/* The generators' done marker (Readable.from's and the pipeline stages' + * helpers answer it): a process-lived object compared by identity. */ +static SCR_TL ScrDyn *scr_stream_done_marker = NULL; + +static void scr_stream_done_marker_free(void) { + scr_dyn_release(scr_stream_done_marker); + scr_stream_done_marker = NULL; +} + +ScrDyn *scr_stream_from_done(void) { + if (!scr_stream_done_marker) { + scr_stream_done_marker = scr_dyn_new_obj(); + scr_atexit(scr_stream_done_marker_free); + } + return scr_dyn_retain(scr_stream_done_marker); +} + +/* ── pipeline function stages (pipeline.js: makeAsyncIterable, pumpToNode) ── + * A run of async generator functions between two node streams. Each + * stage is called while the pipeline is set up, in pipeline order: + * makeAsyncIterable hands the first one the previous stream itself (a + * Readable is async iterable) and each later one the previous stage's + * generator. The last generator is pumped into the stream after the run + * with pumpToNode. The stages and helpers use the closure ABI (the + * closure first; the callee owns its parameters). */ + +typedef ScrGen *(*ScrPipeFirstFn)(ScrClosure *, ScrStream *); +typedef ScrGen *(*ScrPipeStageFn)(ScrClosure *, ScrGen *); +typedef ScrPromise *(*ScrPipeNextFn)(ScrClosure *, ScrGen *); +typedef ScrDyn *(*ScrPipeReadFn)(ScrClosure *, void *); + +typedef struct { + size_t at; /* streams[0..at-1] come before the stages */ + size_t count; /* the function stages */ + ScrClosure **stages; /* (stream or async generator) => async generator */ + ScrClosure *next, *read, *ret; /* the last generator's helpers */ +} ScrPipeStages; + +enum { SCR_PUMP_DRAIN0, SCR_PUMP_WRITE, SCR_PUMP_END }; + +typedef struct { + size_t rc; + ScrStreamPipeCtx *ctx; /* +1 */ + ScrGen *gen; /* +1 */ + ScrClosure *next, *read, *ret; /* +1 */ + ScrStream *w; /* +1 */ + bool end; + ScrError *error; /* +1: what resume() recorded, or NULL */ + ScrPromise *waiter; /* +1: the pending wait(), or NULL */ + int stage; /* what the pending wait() resumes */ + ScrClosure *drain_l; /* +1 */ + ScrClosure *eos_cleanup; /* +1 */ + ScrError *fail; /* +1: the loop's error while return() runs */ +} ScrPump; + +static void *scr_pump_retain_v(void *p) { + ((ScrPump *)p)->rc++; + return p; +} + +static void scr_pump_release_v(void *p) { + ScrPump *pm = p; + if (--pm->rc > 0) return; + scr_stream_pipectx_release_v(pm->ctx); + scr_gen_release(pm->gen); + scr_closure_release(pm->next); + scr_closure_release(pm->read); + scr_closure_release(pm->ret); + scr_stream_release(pm->w); + if (pm->error) scr_error_release(pm->error); + if (pm->waiter) scr_promise_release(pm->waiter); + if (pm->drain_l) scr_closure_release(pm->drain_l); + if (pm->eos_cleanup) scr_closure_release(pm->eos_cleanup); + if (pm->fail) scr_error_release(pm->fail); + free(pm); +} + +static ScrClosure *scr_pump_closure(void *fn, ScrPump *pm) { + ScrClosure *c = scr_closure_new(fn, 1); + c->caps[0] = scr_box_new_obj(&scr_pump_retain_v, &scr_pump_release_v, NULL); + scr_box_set_ref(c->caps[0], scr_pump_retain_v(pm)); + return c; +} + +/* A rejection as a pipeline error: Error values as themselves. +1. */ +static ScrError *scr_pump_caught(void) { + ScrCaught *c = scr_exc_take(); + ScrDyn *d = scr_caught_to_dyn(c); + scr_caught_release(c); + ScrError *e = d ? scr_errdyn_err_of(d) : NULL; + if (d) scr_dyn_release(d); + if (e) return e; + return scr_stream_mkerr("ERR_INTERNAL_ASSERTION", + "a pipeline stage rejection that is not an Error object is not supported yet"); +} + +static void scr_pump_loop(ScrPump *pm); +static void scr_pump_wait(ScrPump *pm, int stage); +static void scr_pump_fail_in_loop(ScrPump *pm, ScrError *err); + +/* finish(err) and the finally block: the eos cleanup, then the 'drain' + * listener. Moves err. */ +static void scr_pump_done(ScrPump *pm, ScrError *err) { + ScrStreamPipeCtx *ctx = pm->ctx; + scr_pipe_finish_impl(ctx, err, --ctx->finish_count == 0); + if (err) scr_error_release(err); + if (pm->eos_cleanup) { + ScrClosure *c = pm->eos_cleanup; + pm->eos_cleanup = NULL; + ((void (*)(ScrClosure *))c->fn)(c); + scr_closure_release(c); + } + if (pm->drain_l) { + ScrStr *n = scr_str_new("drain", 5); + scr_emitter_release(scr_emitter_off((ScrEmitter *)pm->w, n, pm->drain_l)); + scr_str_release(n); + scr_closure_release(pm->drain_l); + pm->drain_l = NULL; + } +} + +/* resume(err): the 'drain' listener and the eos callback. */ +static void scr_pump_resume(ScrPump *pm, ScrError *err) { + if (err && !pm->error) pm->error = scr_error_retain(err); + ScrPromise *w = pm->waiter; + if (!w) return; + pm->waiter = NULL; + if (pm->error) { + scr_throw_obj(scr_error_retain(pm->error), &scr_error_retain_v, &scr_error_release_v, + scr_error_trace_arg()); + scr_promise_reject_pending(w); + } else { + scr_promise_fulfill_void(w); + } + scr_promise_release(w); +} + +static void scr_pump_drain_l(ScrClosure *c) { + ScrPump *pm = scr_box_get_ref(c->caps[0]); + scr_pump_resume(pm, NULL); + scr_pump_release_v(pm); +} + +static void scr_pump_eos(ScrClosure *c, ScrStream *s, ScrError *err) { + (void)s; + ScrPump *pm = scr_box_get_ref(c->caps[0]); + scr_pump_resume(pm, err); + scr_pump_release_v(pm); +} + +/* `await p` for the pump: kind 0 a wait(), 1 the generator's next(), + * 2 its return(). */ +typedef struct { + ScrPump *pm; /* +1 */ + ScrPromise *p; /* +1 */ + int kind; + int stage; +} ScrPumpAwait; + +static void scr_pump_after_wait(ScrPump *pm, int stage, bool rejected); +static void scr_pump_after_next(ScrPump *pm, ScrDyn *v); +static void scr_pump_after_return(ScrPump *pm); + +static void scr_pump_await_entry(ScrFiber *fiber, void *arg) { + (void)fiber; + ScrPumpAwait a = *(ScrPumpAwait *)arg; + free(arg); + if (a.kind == 1) { + void *rec = scr_await_ref(a.p); + ScrDyn *v = scr_exc_pending() ? NULL : ((ScrPipeReadFn)a.pm->read->fn)(a.pm->read, rec); + scr_pump_after_next(a.pm, v); + } else { + scr_await_void(a.p); + if (a.kind == 2) { + if (scr_exc_pending()) scr_exc_clear(); /* the loop's error wins */ + scr_pump_after_return(a.pm); + } else { + scr_pump_after_wait(a.pm, a.stage, scr_exc_pending()); + } + } + if (scr_exc_pending()) scr_exc_clear(); + scr_promise_release(a.p); + scr_pump_release_v(a.pm); +} + +static void scr_pump_await(ScrPump *pm, ScrPromise *p /*borrowed*/, int kind, int stage) { + ScrPumpAwait *a = malloc(sizeof *a); + if (!a) scr_stream_oom(); + a->pm = scr_pump_retain_v(pm); + a->p = scr_promise_retain(p); + a->kind = kind; + a->stage = stage; + scr_promise_release(scr_async_spawn_after(p, &scr_pump_await_entry, a)); +} + +/* `await wait()`: rejects at once with a recorded error, else waits for + * resume(). */ +static void scr_pump_wait(ScrPump *pm, int stage) { + ScrPromise *p = scr_promise_new(); + if (pm->error) { + scr_throw_obj(scr_error_retain(pm->error), &scr_error_retain_v, &scr_error_release_v, + scr_error_trace_arg()); + scr_promise_reject_pending(p); + } else { + pm->waiter = scr_promise_retain(p); + } + scr_pump_await(pm, p, 0, stage); + scr_promise_release(p); +} + +/* `for await (const chunk of iterable)`: one next() per pass. */ +static void scr_pump_loop(ScrPump *pm) { + ScrPromise *p = ((ScrPipeNextFn)pm->next->fn)(pm->next, scr_gen_retain(pm->gen)); + scr_pump_await(pm, p, 1, 0); + scr_promise_release(p); +} + +static void scr_pump_after_next(ScrPump *pm, ScrDyn *v) { + if (!v) { + /* the generator threw: catch (err) */ + scr_pump_done(pm, scr_pump_caught()); + return; + } + if (v == scr_stream_done_marker) { + scr_dyn_release(v); + if (pm->end) { + scr_stream_release(scr_stream_end(pm->w, NULL, NULL, NULL)); + if (scr_exc_pending()) { + scr_pump_done(pm, scr_pump_caught()); + return; + } + scr_pump_wait(pm, SCR_PUMP_END); + return; + } + scr_pump_done(pm, NULL); + return; + } + bool ok = scr_stream_write_dyn(pm->w, v, NULL); + scr_dyn_release(v); + if (scr_exc_pending()) { + scr_pump_fail_in_loop(pm, scr_pump_caught()); + return; + } + if (!ok) { + scr_pump_wait(pm, SCR_PUMP_WRITE); + return; + } + scr_pump_loop(pm); +} + +/* An error inside the loop body: for-await's AsyncIteratorClose awaits + * the generator's return() before the catch block runs. Moves err. */ +static void scr_pump_fail_in_loop(ScrPump *pm, ScrError *err) { + pm->fail = err; + ScrPromise *p = ((ScrPipeNextFn)pm->ret->fn)(pm->ret, scr_gen_retain(pm->gen)); + scr_pump_await(pm, p, 2, 0); + scr_promise_release(p); +} + +static void scr_pump_after_return(ScrPump *pm) { + ScrError *err = pm->fail; + pm->fail = NULL; + scr_pump_done(pm, err); +} + +static void scr_pump_after_wait(ScrPump *pm, int stage, bool rejected) { + ScrError *err = rejected ? scr_pump_caught() : NULL; + if (stage == SCR_PUMP_WRITE) { + if (err) scr_pump_fail_in_loop(pm, err); + else scr_pump_loop(pm); + } else if (stage == SCR_PUMP_DRAIN0) { + if (err) scr_pump_done(pm, err); + else scr_pump_loop(pm); + } else { + scr_pump_done(pm, err); + } +} + +/* pumpToNode(gen, w, finish, { end }): starts at once, as the async + * function does. The helpers are borrowed. */ +static void scr_pump_start(ScrStreamPipeCtx *ctx, ScrGen *gen /*moves*/, const ScrPipeStages *fs, + ScrStream *w, bool end) { + ScrPump *pm = calloc(1, sizeof *pm); + if (!pm) scr_stream_oom(); + pm->rc = 1; + pm->ctx = scr_stream_pipectx_retain_v(ctx); + pm->gen = gen; + pm->next = scr_closure_retain(fs->next); + pm->read = scr_closure_retain(fs->read); + pm->ret = scr_closure_retain(fs->ret); + pm->w = scr_stream_retain(w); + pm->end = end; + pm->drain_l = scr_pump_closure((void *)&scr_pump_drain_l, pm); + scr_pipe_on(w, "drain", scr_closure_retain(pm->drain_l), scr_ee_inv_fixed0, false); + pm->eos_cleanup = scr_pipe_eos(w, scr_pump_closure((void *)&scr_pump_eos, pm), &scr_pump_eos, false, + w->st->has_w); + if (w->st->has_w && w->st->w.need_drain) scr_pump_wait(pm, SCR_PUMP_DRAIN0); + else scr_pump_loop(pm); + scr_pump_release_v(pm); +} + static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosure *cb, ScrStreamErrInv inv, bool end_opt, ScrDyn *signal, - ScrAbortAddFn add) { + ScrAbortAddFn add, const ScrPipeStages *fs) { ScrStreamPipeCtx *ctx = calloc(1, sizeof *ctx); if (!ctx) scr_stream_oom(); ctx->rc = 1; @@ -3026,14 +3327,35 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu ctx->dispose = add(signal, listener); scr_dyn_release(listener); } + /* pipelineImpl's loop over the stages: the n streams, with fs->count + * function stages before streams[fs->at] */ + size_t stages = n + (fs ? fs->count : 0); + size_t fn_first = fs ? fs->at : stages, fn_end = fs ? fs->at + fs->count : stages; ScrStream *ret = NULL; - for (size_t i = 0; i < n; i++) { - ScrStream *s = streams[i]; - bool reading = i + 1 < n; - bool writing = i > 0; + ScrGen *ret_gen = NULL; /* +1: ret when the previous stage was a function */ + size_t i = 0; /* the next stream */ + for (size_t j = 0; j < stages; j++) { + bool reading = j + 1 < stages; bool end = reading || end_opt; - bool last = i + 1 == n; - if (reading && (streams[i + 1]->st->close_emitted || streams[i + 1]->st->destroyed)) { + bool last = j + 1 == stages; + if (j >= fn_first && j < fn_end) { + /* ret = stage(makeAsyncIterable(ret)) */ + ScrClosure *stage = fs->stages[j - fn_first]; + ScrGen *src = ret_gen; + ret_gen = src ? ((ScrPipeStageFn)stage->fn)(stage, src) + : ((ScrPipeFirstFn)stage->fn)(stage, scr_stream_retain(ret)); + if (scr_exc_pending()) { + scr_gen_release(ret_gen); + scr_stream_pipectx_drop(ctx); + scr_stream_pipectx_release_v(ctx); + return NULL; + } + continue; + } + ScrStream *s = streams[i]; + bool writing = j > 0; + if (reading && !(j + 1 >= fn_first && j + 1 < fn_end) && + (streams[i + 1]->st->close_emitted || streams[i + 1]->st->destroyed)) { scr_pipe_throw("ERR_STREAM_UNABLE_TO_PIPE", "Cannot pipe to a closed or destroyed stream", SCR_ERR_ERROR); scr_stream_pipectx_drop(ctx); @@ -3054,7 +3376,13 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu scr_pipe_on(s, "error", scr_closure_retain(on_error), scr_ee_inv_fixed1, false); if (last_readable) ctx->last_error_l = on_error; else scr_closure_release(on_error); - if (i > 0) { + if (ret_gen) { + /* an async iterable before a stream: pumpToNode(ret, s, finish, { end }) */ + ctx->finish_count++; + ScrGen *g = ret_gen; + ret_gen = NULL; + scr_pump_start(ctx, g, fs, s, end); + } else if (j > 0) { if (!ret->st->has_r) { static const char msg[] = "The \"val\" argument must be an instance of Readable, Iterable, AsyncIterable, " @@ -3084,15 +3412,33 @@ static ScrStream *scr_stream_pipeline_ex(size_t n, ScrStream **streams, ScrClosu else scr_closure_release(cleanup); } ret = s; + i++; } if (aborted) scr_next_tick(scr_pipe_closure((void *)&scr_pipe_abort_tick, ctx, 0)); scr_stream_pipectx_release_v(ctx); /* the listeners and the live list hold it */ return scr_stream_retain(ret); } +/* pipeline(...) with a run of async generator stages before streams[at]: + * the helpers are borrowed (see ScrPipeStages). */ +ScrStream *scr_stream_pipeline_stages(double n, ScrStream **streams, double at, double count, + ScrClosure **stages, ScrClosure *next, ScrClosure *read, + ScrClosure *ret, ScrClosure *cb /*moves*/, ScrStreamErrInv inv) { + ScrPipeStages fs = {(size_t)at, (size_t)count, stages, next, read, ret}; + return scr_stream_pipeline_ex((size_t)n, streams, cb, inv, true, NULL, NULL, &fs); +} + +ScrStream *scr_stream_pipeline_stages_dyn(double n, ScrStream **streams, double at, double count, + ScrClosure **stages, ScrClosure *next, ScrClosure *read, + ScrClosure *ret, ScrDyn *cb) { + ScrClosure *c = scr_stream_dynopt_clo(cb, (void *)&scr_stream_finished_dyn_inv); + return scr_stream_pipeline_stages(n, streams, at, count, stages, next, read, ret, c, + &scr_stream_finished_dyn_inv); +} + ScrStream *scr_stream_pipeline(double n_d, ScrStream **streams /*borrowed*/, ScrClosure *cb /*moves*/, ScrStreamErrInv inv) { - return scr_stream_pipeline_ex((size_t)n_d, streams, cb, inv, true, NULL, NULL); + return scr_stream_pipeline_ex((size_t)n_d, streams, cb, inv, true, NULL, NULL, NULL); } /* The dyn-valued twins: the callback rides as a checked-dynamic value @@ -3360,7 +3706,7 @@ ScrStream *scr_stream_compose(double n_d, ScrStream **streams) { c->tail = scr_stream_retain(streams[n - 1]); c->onclose = SCR_COMPOSE_RUNNING; ScrStream *tail = scr_stream_pipeline_ex(n, streams, scr_compose_closure((void *)&scr_compose_finished, c), - &scr_compose_finished, true, NULL, NULL); + &scr_compose_finished, true, NULL, NULL, NULL); if (!tail) { scr_compose_release_v(c); return NULL; @@ -3444,6 +3790,17 @@ ScrPromise *scr_sp_pipeline(double n, ScrStream **streams) { return p; } +ScrPromise *scr_sp_pipeline_stages(double n, ScrStream **streams, double at, double count, + ScrClosure **stages, ScrClosure *next, ScrClosure *read, + ScrClosure *ret) { + ScrPromise *p = scr_promise_new(); + ScrStream *dst = scr_stream_pipeline_stages(n, streams, at, count, stages, next, read, ret, + scr_sp_watcher(p), &scr_sp_settle_inv); + if (dst) scr_stream_release(dst); + else scr_promise_reject_pending(p); + return p; +} + /* pipeline(...streams, { signal, end }): end false leaves the destination * open (the promise settles once the stages before it finish); signal is * NULL or borrowed — a value that is not an AbortSignal rejects with @@ -3465,7 +3822,7 @@ ScrPromise *scr_sp_pipeline_opts(double n, ScrStream **streams, bool end, ScrDyn scr_dyn_release(aborted); } ScrStream *dst = scr_stream_pipeline_ex((size_t)n, streams, scr_sp_watcher(p), - &scr_sp_settle_inv, end, signal, add); + &scr_sp_settle_inv, end, signal, add, NULL); if (dst) scr_stream_release(dst); else scr_promise_reject_pending(p); return p; diff --git a/packages/runtime/src/scr_stream.h b/packages/runtime/src/scr_stream.h index 0b5687255..0909e5f19 100644 --- a/packages/runtime/src/scr_stream.h +++ b/packages/runtime/src/scr_stream.h @@ -171,6 +171,22 @@ ScrStream *scr_stream_pipeline(double n, ScrStream **streams, ScrClosure *cb, Sc * mustCall wrappers; borrowed). */ ScrClosure *scr_stream_finished_dyn(ScrStream *s, struct ScrDyn *cb); ScrStream *scr_stream_pipeline_dyn(double n, ScrStream **streams, struct ScrDyn *cb); +/* pipeline(s1.., stage1..stagek, ..sn[, cb]) with a run of async generator + * function stages before streams[at] (pipeline.js): the first stage gets + * streams[at-1] itself, each later one the previous stage's generator, + * and the last generator is pumped into streams[at] (pumpToNode) through + * its helpers: next/ret answer the resume's promise, read maps its + * IteratorResult to the value or scr_stream_from_done. Streams and + * closures borrowed; cb moves. */ +ScrStream *scr_stream_pipeline_stages(double n, ScrStream **streams, double at, double count, + ScrClosure **stages, ScrClosure *next, ScrClosure *read, + ScrClosure *ret, ScrClosure *cb, ScrStreamErrInv inv); +ScrStream *scr_stream_pipeline_stages_dyn(double n, ScrStream **streams, double at, double count, + ScrClosure **stages, ScrClosure *next, ScrClosure *read, + ScrClosure *ret, struct ScrDyn *cb); +struct ScrPromise *scr_sp_pipeline_stages(double n, ScrStream **streams, double at, double count, + ScrClosure **stages, ScrClosure *next, ScrClosure *read, + ScrClosure *ret); /* stream.compose(s1..sn): a pipeline over the streams and a Duplex that * writes to the first and reads from the last (compose.js). One stream * answers itself (Duplex.from of a duplex). +1, or NULL with @@ -293,7 +309,7 @@ ScrStream *scr_stream_from_gen(struct ScrGen *g, ScrClosure *next, ScrClosure *r ScrStream *scr_stream_from_async_gen(struct ScrGen *g, ScrClosure *next, ScrClosure *read, ScrClosure *ret, ScrClosure *thr, double flags, double hwm); -ScrDyn *scr_stream_from_done(void); /* +1 */ +ScrDyn *scr_stream_from_done(void); /* +1, the process-lived marker (scr_stream.c) */ ScrStream *scr_stream_pause(ScrStream *s); /* recv +1 */ ScrStream *scr_stream_resume(ScrStream *s); /* recv +1 */ bool scr_stream_is_paused(ScrStream *s); diff --git a/packages/runtime/src/scr_stream_ops.c b/packages/runtime/src/scr_stream_ops.c index a8f1cacf9..5ad50248d 100644 --- a/packages/runtime/src/scr_stream_ops.c +++ b/packages/runtime/src/scr_stream_ops.c @@ -78,7 +78,7 @@ static void op_sentinels_init(void) { if (op_k_empty) return; op_k_empty = scr_dyn_new_obj(); op_k_eof = scr_dyn_new_obj(); - op_k_done = scr_dyn_new_obj(); + op_k_done = scr_stream_from_done(); /* shared with pumpToNode (scr_stream.c) */ scr_atexit(op_sentinels_free); } @@ -1726,10 +1726,6 @@ ScrStream *scr_stream_from_async_gen(ScrGen *g, ScrClosure *next, ScrClosure *re return op_from_compiled(FROM_ASYNC, g, next, read, ret, thr, flags, hwm); } -ScrDyn *scr_stream_from_done(void) { - op_sentinels_init(); - return scr_dyn_retain(op_k_done); -} /* ── the promise operators ─────────────────────────────────────────── */ diff --git a/tests/corpus/5425-pipeline-generator-stage.ts b/tests/corpus/5425-pipeline-generator-stage.ts new file mode 100644 index 000000000..dc7f1b8cb --- /dev/null +++ b/tests/corpus/5425-pipeline-generator-stage.ts @@ -0,0 +1,35 @@ +// pipeline() with an async generator stage between two streams: the stage +// iterates the source stream itself, and its values reach the destination +// on the same microtasks as in Node. +import { Readable, Writable, pipeline } from "node:stream"; + +const log: string[] = []; +let k = 0; +const chain = (): void => { + if (k++ < 40) { + log.push(`m${k}`); + Promise.resolve().then(chain); + } +}; +const src = Readable.from(["a", "b"]); +const sink = new Writable({ + write(c: Buffer, _e, cb) { + log.push(`write ${c.toString()}`); + cb(); + }, +}); +Promise.resolve().then(chain); +pipeline( + src, + async function* (source: AsyncIterable) { + for await (const c of source) { + log.push(`stage ${c}`); + yield c.toUpperCase(); + } + }, + sink, + (err?: Error | null) => { + log.push(`done ${err ? err.message : "ok"}`); + console.log(log.join(" ")); + }, +); diff --git a/tests/corpus/5426-pipeline-generator-errors.ts b/tests/corpus/5426-pipeline-generator-errors.ts new file mode 100644 index 000000000..948089799 --- /dev/null +++ b/tests/corpus/5426-pipeline-generator-errors.ts @@ -0,0 +1,118 @@ +// pipeline() generator stages: two chained stages, the promise form with a +// slow sink, a stage that throws, a destination that fails (the stage is +// returned and its finally runs), and a source that errors. +import { PassThrough, Readable, Writable, pipeline } from "node:stream"; +import { pipeline as pipelineP } from "node:stream/promises"; + +const log: string[] = []; +let hop = 0; +const chain = (): void => { + if (hop++ < 60) { + log.push(`h${hop}`); + Promise.resolve().then(chain); + } +}; +const collect = (label: string, hwm = 16384): Writable => + new Writable({ + highWaterMark: hwm, + write(c: Buffer, _e, cb) { + log.push(`${label} write ${c.toString()}`); + if (hwm === 1) setTimeout(cb, 1); + else cb(); + }, + }); + +Promise.resolve().then(chain); +// two chained stages +await new Promise((resolve) => + pipeline( + Readable.from(["a", "b", "c"]), + async function* (source: AsyncIterable) { + for await (const c of source) yield c + "1"; + }, + async function* (source: AsyncIterable) { + for await (const c of source) { + log.push(`second ${c}`); + yield c + "2"; + } + }, + collect("chained"), + (err?: Error | null) => { + log.push(`chained done ${err ? err.message : "ok"}`); + resolve(); + }, + ), +); +// the promise form, with backpressure into a slow sink +await pipelineP( + Readable.from(["x", "y", "z"]), + async function* (source: AsyncIterable) { + for await (const c of source) yield c.repeat(3); + }, + collect("slow", 1), +); +log.push("promise form done"); +// a stage that throws +await new Promise((resolve) => + pipeline( + Readable.from(["p", "q"]), + async function* (source: AsyncIterable) { + try { + for await (const c of source) { + if (c === "q") throw new Error("stage failed"); + yield c; + } + } finally { + log.push("thrower finally"); + } + }, + collect("thrower"), + (err?: Error | null) => { + log.push(`thrower done ${err ? err.message : "ok"}`); + resolve(); + }, + ), +); +// a destination that fails: the stage's generator is returned +const failing = new Writable({ + write(_c: Buffer, _e, cb) { + cb(new Error("sink failed")); + }, +}); +await new Promise((resolve) => + pipeline( + Readable.from(["u", "v", "w"]), + async function* (source: AsyncIterable) { + try { + for await (const c of source) yield c; + } finally { + log.push("returned stage finally"); + } + }, + failing, + (err?: Error | null) => { + log.push(`sink done ${err ? err.message : "ok"}`); + resolve(); + }, + ), +); +// a source that errors +const bad = new Readable({ + read() { + this.destroy(new Error("source failed")); + }, +}); +await new Promise((resolve) => + pipeline( + bad, + async function* (source: AsyncIterable) { + for await (const c of source) yield c; + }, + new PassThrough(), + (err?: Error | null) => { + log.push(`source done ${err ? err.message : "ok"}`); + resolve(); + }, + ), +); +console.log(log.join("\n")); diff --git a/tests/corpus/5427-pipeline-generator-backpressure.ts b/tests/corpus/5427-pipeline-generator-backpressure.ts new file mode 100644 index 000000000..a7f56d006 --- /dev/null +++ b/tests/corpus/5427-pipeline-generator-backpressure.ts @@ -0,0 +1,42 @@ +// pipeline() pumps a generator stage into a highWaterMark: 1 sink: each +// write waits for drain, timed against microtasks and ticks. +import { Readable, Writable } from "node:stream"; +import { pipeline } from "node:stream/promises"; + +const log: string[] = []; +let hop = 0; +const chain = (): void => { + if (hop++ < 40) { + log.push(`h${hop}`); + Promise.resolve().then(chain); + } +}; +let tick = 0; +const ticks = (): void => { + if (tick++ < 10) { + log.push(`t${tick}`); + process.nextTick(ticks); + } +}; +const sink = new Writable({ + highWaterMark: 1, + write(c: Buffer, _e, cb) { + log.push(`write ${c.toString()} ${sink.writableLength}`); + Promise.resolve().then(() => cb()); + }, +}); +sink.on("drain", () => log.push("drain")); +Promise.resolve().then(chain); +process.nextTick(ticks); +await pipeline( + Readable.from(["1", "2", "3", "4"]), + async function* (source: AsyncIterable) { + for await (const c of source) { + log.push(`stage ${c}`); + yield c + c; + } + }, + sink, +); +log.push("done"); +console.log(log.join(" ")); diff --git a/tests/corpus/5428-pipeline-generator-stages.cjs b/tests/corpus/5428-pipeline-generator-stages.cjs new file mode 100644 index 000000000..8c6acd92e --- /dev/null +++ b/tests/corpus/5428-pipeline-generator-stages.cjs @@ -0,0 +1,28 @@ +// @no-node-shims +// pipeline() generator stages from CommonJS: untyped stages over Buffer +// chunks, the promise and callback forms, and the returned destination. +'use strict'; +const { Readable, PassThrough, Writable, pipeline } = require('node:stream'); +const { pipeline: pipelineP } = require('node:stream/promises'); +async function main() { + const seen = []; + const sink = new Writable({ write(c, e, cb) { seen.push(String(c)); cb(); } }); + await pipelineP( + Readable.from([Buffer.from('ab'), Buffer.from('cd')]), + async function* (source) { + for await (const chunk of source) yield String(chunk).toUpperCase(); + }, + async function* (source) { + for await (const chunk of source) yield chunk + '|'; + }, + sink, + ); + console.log('seen', seen.join('')); + const out = new PassThrough(); + const ret = pipeline(Readable.from(['x']), async function* (s) { for await (const c of s) yield c + c; }, out, (err) => { + console.log('cb', err === undefined ? 'undefined' : err); + }); + console.log('returns destination', ret === out); + out.on('data', (c) => console.log('out', String(c))); +} +main(); diff --git a/tests/corpus/5429-pipeline-generator-buffers.ts b/tests/corpus/5429-pipeline-generator-buffers.ts new file mode 100644 index 000000000..ad1b4a435 --- /dev/null +++ b/tests/corpus/5429-pipeline-generator-buffers.ts @@ -0,0 +1,43 @@ +// pipeline() with a stage typed AsyncIterable between streams, more +// streams after it, and the callback form's return value. +import { PassThrough, Readable, Transform, Writable, pipeline } from "node:stream"; + +const log: string[] = []; +const upper = new Transform({ + transform(c: Buffer, _e, cb) { + cb(null, c.toString().toUpperCase()); + }, +}); +const sink = new Writable({ + write(c: Buffer, _e, cb) { + log.push(`sink ${c.toString()}`); + cb(); + }, +}); +const source = new Readable({ + read() { + this.push(Buffer.from("hello ")); + this.push(Buffer.from("world")); + this.push(null); + }, +}); +const ret = pipeline( + source, + new PassThrough(), + async function* (chunks: AsyncIterable) { + let total = 0; + for await (const chunk of chunks) { + total += chunk.length; + log.push(`stage ${chunk.length} ${chunk.toString("hex")}`); + yield chunk.subarray(0, 3); + } + log.push(`stage total ${total}`); + }, + upper, + sink, + (err?: Error | null) => { + log.push(`done ${err ? err.message : "ok"} ${sink.writableFinished}`); + console.log(log.join("\n")); + }, +); +console.log("returns the destination", ret === sink);