From 2353a045404cd8dd18acc6caa97aa19f8f76a90e Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Thu, 29 Oct 2026 23:46:36 -0500 Subject: [PATCH] Keep typed records unboxed in object-mode streams --- .../src/backend/llvm/expr-stream-bridges.ts | 48 +++ .../src/backend/llvm/expr-stream-callbacks.ts | 102 ++++++- .../compiler/src/backend/llvm/lib-stream.ts | 34 +++ .../compiler/src/backend/llvm/record-views.ts | 11 + .../src/frontend/lowering/lower-stream.ts | 88 +++++- .../stream-typed-chunk-boxed-kinds.ts.json | 6 + .../stream-typed-chunk-data-and-pipe.ts.json | 6 + .../stream-typed-chunk-for-await.ts.json | 6 + .../stream-typed-chunk-identity.ts.json | 6 + .../stream-typed-chunk-mismatch.ts.json | 6 + .../stream-typed-chunk-mutation.ts.json | 6 + .../stream-typed-chunk-read-paused.ts.json | 6 + .../corpus/stream-typed-chunk-weakmap.ts.json | 6 + .../corpus/stream-typed-chunk-writev.ts.json | 6 + packages/runtime/src/scr_stream.c | 273 ++++++++++++++---- packages/runtime/src/scr_stream.h | 31 ++ .../corpus/stream-typed-chunk-boxed-kinds.ts | 89 ++++++ .../stream-typed-chunk-data-and-pipe.ts | 55 ++++ tests/corpus/stream-typed-chunk-for-await.ts | 44 +++ tests/corpus/stream-typed-chunk-identity.ts | 54 ++++ tests/corpus/stream-typed-chunk-mismatch.ts | 41 +++ tests/corpus/stream-typed-chunk-mutation.ts | 38 +++ .../corpus/stream-typed-chunk-read-paused.ts | 26 ++ tests/corpus/stream-typed-chunk-weakmap.ts | 53 ++++ tests/corpus/stream-typed-chunk-writev.ts | 39 +++ .../stream-typed-chunk-boxed-kinds.ts.json | 1 + .../stream-typed-chunk-data-and-pipe.ts.json | 1 + .../stream-typed-chunk-for-await.ts.json | 1 + .../stream-typed-chunk-identity.ts.json | 1 + .../stream-typed-chunk-mismatch.ts.json | 1 + .../stream-typed-chunk-mutation.ts.json | 1 + .../stream-typed-chunk-read-paused.ts.json | 1 + .../stream-typed-chunk-weakmap.ts.json | 1 + .../stream-typed-chunk-writev.ts.json | 1 + 34 files changed, 1010 insertions(+), 79 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-boxed-kinds.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-data-and-pipe.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-for-await.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-identity.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mismatch.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mutation.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-read-paused.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-weakmap.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-writev.ts.json create mode 100644 tests/corpus/stream-typed-chunk-boxed-kinds.ts create mode 100644 tests/corpus/stream-typed-chunk-data-and-pipe.ts create mode 100644 tests/corpus/stream-typed-chunk-for-await.ts create mode 100644 tests/corpus/stream-typed-chunk-identity.ts create mode 100644 tests/corpus/stream-typed-chunk-mismatch.ts create mode 100644 tests/corpus/stream-typed-chunk-mutation.ts create mode 100644 tests/corpus/stream-typed-chunk-read-paused.ts create mode 100644 tests/corpus/stream-typed-chunk-weakmap.ts create mode 100644 tests/corpus/stream-typed-chunk-writev.ts create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-boxed-kinds.ts.json create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-data-and-pipe.ts.json create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-for-await.ts.json create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-identity.ts.json create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-mismatch.ts.json create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-mutation.ts.json create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-read-paused.ts.json create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-weakmap.ts.json create mode 100644 tests/harness/corpus-costs/stream-typed-chunk-writev.ts.json diff --git a/packages/compiler/src/backend/llvm/expr-stream-bridges.ts b/packages/compiler/src/backend/llvm/expr-stream-bridges.ts index 56fb5458c..2042b9078 100644 --- a/packages/compiler/src/backend/llvm/expr-stream-bridges.ts +++ b/packages/compiler/src/backend/llvm/expr-stream-bridges.ts @@ -40,6 +40,7 @@ import { llFieldType, loadStructField, releaseSym, + traceAdapter, traceArg, vAdapters, type ShapeHost, @@ -754,6 +755,50 @@ function symbolKeyLines( }; } +/** The ScrStreamTypedDesc constant (scr_stream.h) of a record type an + * object-mode stream holds as the record itself, or null for a type whose + * chunks stay boxed: records without declared fields, with an index + * signature or accessor slots, tuples, the text codec records (they box as + * handles), and records the collector traces (a stream does not trace the + * records it holds). The constant's `box` converts the record as a + * live-reference conversion to dyn does (streamTypedRefBoxValue). */ +export function streamTypedDesc(host: LlvmEmitterContext, t: IrType): string | null { + if (t.kind !== "record") return null; + const shape = host.recordsById.get(t.shapeId); + if ( + !shape || + shape.fields.length === 0 || + shape.indexValue !== undefined || + shape.tuple || + shapeHasAccessorSlots(shape) || + shape.fields.some((field) => field.name === "handleEvent" && field.type.kind === "func") || + textCodecRecordClass(t, (id) => host.recordsById.get(id)) !== null || + traceAdapter(host.shapeHost, t) !== null + ) + return null; + const key = typeKey(t); + const cacheKey = `std:${key}`; + const existing = host.resolveThunks.get(cacheKey); + if (existing) return `@${existing}`; + const sym = host.contentSymbol("sc_std_", key); + host.resolveThunks.set(cacheKey, sym); + host.journal?.allocated(sym, "streamTypedDesc", [t]); + const B = new BlockBuilder(); + const boxed = streamTypedRefBoxValue(host, B, t, "%p"); + B.terminate(`ret ptr ${boxed}`); + const rc = vAdapters(host.shapeHost, t); + host.resolveThunkDefs.push( + `define internal ptr @${sym}_box(ptr %p) ${FN_ATTRS} { ; box a typed stream chunk ${key}`, + B.render(), + `}`, + ``, + `@${sym} = internal constant { ptr, ${host.sizeType}, ptr, ptr, ptr } { ptr ${host.cstr(key)}, ${host.sizeType} ${Buffer.byteLength(key, "utf8")}, ptr ${rc.retain}, ptr ${rc.release}, ptr @${sym}_box }`, + ``, + ); + host.journal?.built(); + return `@${sym}`; +} + export function streamTypedRefBoxValue( host: LlvmEmitterContext, B: BlockBuilder, @@ -2018,6 +2063,9 @@ export function streamFromArrayAdapter( } registerReplay("dynPromiseAdapter", (host, args) => dynPromiseAdapter(host, args[0] as IrType)); +registerReplay("streamTypedDesc", (host, args) => + (streamTypedDesc(host, args[0] as IrType) ?? "").slice(1), +); registerReplay("liveDynUnionRefAdapter", (host, args) => liveDynUnionRefAdapter(host, args[0] as IrType & { kind: "union" }), ); diff --git a/packages/compiler/src/backend/llvm/expr-stream-callbacks.ts b/packages/compiler/src/backend/llvm/expr-stream-callbacks.ts index 78dbc3639..c680822ea 100644 --- a/packages/compiler/src/backend/llvm/expr-stream-callbacks.ts +++ b/packages/compiler/src/backend/llvm/expr-stream-callbacks.ts @@ -1,10 +1,12 @@ /* Focused LLVM expression emission extracted from emitter.ts. */ import { InternalCompilerError } from "../../errors.js"; import { type IrType, isRefCounted, STRING, typeKey } from "../../ir/ir.js"; +import { streamTypedRefEligible } from "../../ir/analysis.js"; import { FN_ATTRS, releaseSym, retainSym, traceArg, vAdapters } from "./shapes.js"; import type { LlvmEmitterContext } from "./expr-context.js"; import { unionPeekLines, unionTagLines } from "./union-repr.js"; import { registerReplay } from "./emit-journal.js"; +import { streamTypedDesc } from "./expr-stream-bridges.js"; export function streamDataAdapter( host: LlvmEmitterContext, @@ -259,6 +261,31 @@ export function streamDoneFnFor( strTag = def!.arms.findIndex((a) => a.kind === "string"); } } + // data of one record type or nothing (a typed completion, + // pinTypedDataCompletion): a record with a typed-chunk descriptor + // completes as a typed chunk, and another record boxes by reference + // (the conversion a dyn `data` parameter receives). + let typedTag = -1; + let typedDesc: string | null = null; + let liveData = false; + if (valueData && dataT?.kind === "union") { + const arms = host.unionsById.get(dataT.unionId)?.arms ?? []; + const record = arms.findIndex((a) => a.kind === "record"); + if (arms.length === 2 && record >= 0 && arms.some((a) => a.kind === "undefinedT")) { + typedDesc = streamTypedDesc(host, arms[record]!); + if (typedDesc !== null) typedTag = record; + else { + // the records a conversion to dyn keeps by reference (Lowerer's + // liveReference) + const armT = arms[record]!; + const shape = armT.kind === "record" ? host.recordsById.get(armT.shapeId) : undefined; + liveData = + streamTypedRefEligible(armT) && + shape !== undefined && + (shape.fields.length > 0 || shape.indexValue !== undefined); + } + } + } const entry = kind === "c" ? "scr_stream_construct_done" @@ -269,15 +296,19 @@ export function streamDoneFnFor( : kind === "d" ? "scr_stream_destroy_done" : kind === "t" - ? valueData - ? "scr_stream_transform_done_value" - : "scr_stream_transform_done" - : valueData - ? "scr_stream_flush_done_value" - : "scr_stream_flush_done"; + ? typedDesc !== null + ? "scr_stream_transform_done_typed" + : valueData + ? "scr_stream_transform_done_value" + : "scr_stream_transform_done" + : typedDesc !== null + ? "scr_stream_flush_done_typed" + : valueData + ? "scr_stream_flush_done_value" + : "scr_stream_flush_done"; const twoSlot = (kind === "t" || kind === "l") && !valueData; host.declare( - `declare void @${entry}(ptr, ptr${twoSlot ? ", ptr, ptr" : valueData ? ", ptr" : ""})`, + `declare void @${entry}(ptr, ptr${typedDesc !== null ? ", ptr, ptr" : twoSlot ? ", ptr, ptr" : valueData ? ", ptr" : ""})`, ); host.declare(`declare ptr @scr_error_retain_v(ptr)`); host.declare(`declare void @scr_union_release(ptr)`); @@ -291,6 +322,7 @@ export function streamDoneFnFor( `entry:`, ` %eslot = alloca ptr`, ...(twoSlot ? [` %bslot = alloca ptr`, ` %sslot = alloca ptr`] : []), + ...(typedDesc !== null ? [` %tslot = alloca ptr`] : []), ` %capsp = getelementptr inbounds %ScrClosure, ptr %self, i64 1`, ` %bx = load ptr, ptr %capsp`, // the stream, borrowed: the box holds it, and the closure (alive for @@ -299,10 +331,12 @@ export function streamDoneFnFor( ` %s = load ptr, ptr %sp`, ` store ptr null, ptr %eslot`, ...(twoSlot ? [` store ptr null, ptr %bslot`, ` store ptr null, ptr %sslot`] : []), + ...(typedDesc !== null ? [` store ptr null, ptr %tslot`] : []), ]; - // A tag-guarded retained peek out of a union argument into a slot. + // A tag-guarded peek out of a union argument into a slot, retained + // unless retain is null. let arm = 0; - const unwrap = (u: string, tag: number, retain: string, slot: string): void => { + const unwrap = (u: string, tag: number, retain: string | null, slot: string): void => { const a = arm++; const union = u === "%e" ? errT : dataT; const nullable = union?.kind === "union" ? host.nullableUnions.get(union.unionId) : null; @@ -315,8 +349,9 @@ export function streamDoneFnFor( ` br i1 %hit${a}, label %yes${a}, label %done${a}`, `yes${a}:`, ...unionPeekLines(nullable, u, `%pv${a}`), - ` %rt${a} = call ptr ${retain}(ptr %pv${a})`, - ` store ptr %rt${a}, ptr ${slot}`, + ...(retain !== null + ? [` %rt${a} = call ptr ${retain}(ptr %pv${a})`, ` store ptr %rt${a}, ptr ${slot}`] + : [` store ptr %pv${a}, ptr ${slot}`]), ` br label %done${a}`, `done${a}:`, ); @@ -330,8 +365,21 @@ export function streamDoneFnFor( host.declare(`declare ptr @scr_str_retain_v(ptr)`); unwrap("%d", strTag, "@scr_str_retain_v", "%sslot"); } + if (typedDesc !== null) unwrap("%d", typedTag, null, "%tslot"); d.push(` %ev = load ptr, ptr %eslot`); - if (valueData) { + if (typedDesc !== null) { + d.push( + ` %tv = load ptr, ptr %tslot`, + ` call void @${entry}(ptr %s, ptr %ev, ptr %tv, ptr ${typedDesc}) ; moves err; borrows s and the record`, + ); + } else if (liveData) { + host.declare(`declare void @scr_dyn_release(ptr)`); + d.push( + ` %dv = call ptr @${host.liveDynUnionRefAdapter(dataT as IrType & { kind: "union" })}(ptr %d)`, + ` call void @${entry}(ptr %s, ptr %ev, ptr %dv) ; moves err; borrows s and the value`, + ` call void @scr_dyn_release(ptr %dv)`, + ); + } else if (valueData) { const conv = host.dyn.toDynHelper(dataT!); host.declare(`declare void @scr_dyn_release(ptr)`); d.push( @@ -840,18 +888,42 @@ export function streamCbThunkFor( host.declare(`declare void @scr_dyn_release(ptr)`); host.declare(`declare zeroext i1 @scr_exc_pending()`); if (hasThis) host.declare(`declare void @scr_stream_release_v(ptr)`); + // A typed chunk of the declared record type (scr_stream_push_typed) + // is the record itself; any other chunk converts. + const desc = streamTypedDesc(host, p); + if (desc !== null) { + host.declare(`declare ptr @scr_stream_chunk_typed(ptr, ptr, ptr)`); + d.push( + ` br label %ctchk${i}`, + `ctchk${i}:`, + ` %ctt${i} = call ptr @scr_stream_chunk_typed(ptr %s, ptr %chunk, ptr ${desc})`, + ` %cthit${i} = icmp ne ptr %ctt${i}, null`, + ` br i1 %cthit${i}, label %ctok${i}, label %ctconv${i}`, + `ctconv${i}:`, + ); + } d.push( ` %cv${i} = call ptr @scr_stream_chunk_value(ptr %s, ptr %chunk)`, ` %ct${i} = call ${ty === "i1" ? "zeroext i1" : ty} @${helper}(ptr %cv${i}, ptr null)`, ` call void @scr_dyn_release(ptr %cv${i})`, ` %ctbad${i} = call zeroext i1 @scr_exc_pending()`, - ` br i1 %ctbad${i}, label %ctfail${i}, label %ctok${i}`, + ` br i1 %ctbad${i}, label %ctfail${i}, label %${desc !== null ? "ctconvok" : "ctok"}${i}`, `ctfail${i}:`, ...(hasThis ? [` call void @scr_stream_release_v(ptr %sr)`] : []), ` ret void`, - `ctok${i}:`, ); - passed.push(`${ty} %ct${i}`); + if (desc !== null) { + d.push( + `ctconvok${i}:`, + ` br label %ctok${i}`, + `ctok${i}:`, + ` %ctv${i} = phi ptr [ %ctt${i}, %ctchk${i} ], [ %ct${i}, %ctconvok${i} ]`, + ); + passed.push(`ptr %ctv${i}`); + } else { + d.push(`ctok${i}:`); + passed.push(`${ty} %ct${i}`); + } } else if (isEncPos && p.kind === "string") { // Node's encoding for decoded (Buffer) chunks is 'buffer'; an // object-mode side passes the write's encoding argument ("" for diff --git a/packages/compiler/src/backend/llvm/lib-stream.ts b/packages/compiler/src/backend/llvm/lib-stream.ts index 89a97d002..11566dc5b 100644 --- a/packages/compiler/src/backend/llvm/lib-stream.ts +++ b/packages/compiler/src/backend/llvm/lib-stream.ts @@ -3,6 +3,8 @@ import { InternalCompilerError } from "../../errors.js"; import type { LlvmEmitterContext, LibCallExpr, LlValue } from "./expr-context.js"; import { f64Lit } from "./common.js"; import { pipelinePumpedRuns } from "../../ir/builtin-signatures.js"; +import { emitBorrowedInput } from "./borrowed-inputs.js"; +import { streamTypedDesc } from "./expr-stream-bridges.js"; const STREAM_CANONICAL_CBS: Record = { readable: ["r", "d"], @@ -14,6 +16,10 @@ const STREAM_CANONICAL_CBS: Record { + if (!ok) return; + if ( + ts.isIdentifier(node) && + node !== declaration && + lowerer.checker.getSymbolAtLocation(node) === symbol + ) { + const call = node.parent; + if (!call || !ts.isCallExpression(call) || call.expression !== node) { + ok = false; + return; + } + const data = call.arguments.nodes[1]; + if (data !== undefined) { + let t: IrType | null; + try { + t = lowerer.mapTypeOf(lowerer.typeOf(data)); + } catch { + t = null; + } + if (t?.kind === "undefinedT") return; + const shape = t?.kind === "record" ? lowerer.shapes.get(t.shapeId) : undefined; + if ( + t?.kind !== "record" || + shape === undefined || + shape.fields.length === 0 || + shape.indexValue !== undefined || + (found.record !== null && found.record.shapeId !== t.shapeId) + ) { + ok = false; + return; + } + found.record = t; + } + } + ts.forEachChild(node, visit); + }; + visit(body); + return ok ? found.record : null; +} + /** Whether every reference to the completion callback `symbol` in `body` * calls it with a `data` argument that is a Buffer, a string, null or * undefined (or with no data). */ @@ -989,7 +1073,9 @@ function parseStreamOptions( continue; } const full = tuples[name]!.map((t) => (t === "fn" ? null : t)); - const pinned = out.readableObject ? null : pinAnyDataCompletion(lowerer, name, fnNode); + const pinned = out.readableObject + ? pinTypedDataCompletion(lowerer, name, fnNode) + : pinAnyDataCompletion(lowerer, name, fnNode); let anyChunk: ts.Symbol | undefined; try { // The completion-callback tail: typed by the checker contextually; diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-boxed-kinds.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-boxed-kinds.ts.json new file mode 100644 index 000000000..0a6f23ff3 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-boxed-kinds.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-boxed-kinds.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-data-and-pipe.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-data-and-pipe.ts.json new file mode 100644 index 000000000..7d3e9902c --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-data-and-pipe.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-data-and-pipe.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-for-await.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-for-await.ts.json new file mode 100644 index 000000000..218dcc215 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-for-await.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-for-await.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-identity.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-identity.ts.json new file mode 100644 index 000000000..be6571be0 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-identity.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-identity.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mismatch.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mismatch.ts.json new file mode 100644 index 000000000..88f11ff72 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mismatch.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-mismatch.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mutation.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mutation.ts.json new file mode 100644 index 000000000..02b9028e9 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-mutation.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-mutation.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-read-paused.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-read-paused.ts.json new file mode 100644 index 000000000..29e67ad0e --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-read-paused.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-read-paused.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-weakmap.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-weakmap.ts.json new file mode 100644 index 000000000..ef4b5ce10 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-weakmap.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-weakmap.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-writev.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-writev.ts.json new file mode 100644 index 000000000..ac4661baf --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/stream-typed-chunk-writev.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/stream-typed-chunk-writev.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_stream.c b/packages/runtime/src/scr_stream.c index a029ece82..57b93d666 100644 --- a/packages/runtime/src/scr_stream.c +++ b/packages/runtime/src/scr_stream.c @@ -80,11 +80,13 @@ static ScrBytes *scr_stream_buffer(ScrBytes *b) { /* ── state ────────────────────────────────────────────────────────────── */ typedef struct ScrWEntry { - ScrBytes *chunk; /* owned; a ScrDyn* value on an object-mode writable side */ + ScrBytes *chunk; /* owned; a ScrDyn* value on an object-mode writable side, + * or a typed record when desc is set */ ScrClosure *cb; /* owned or NULL (the user's write(chunk, cb) callback) */ ScrDyn *enc; /* owned or NULL = undefined: the encoding argument an * object-mode write passed (Node hands it to _write * unchanged; null when the callback took its place) */ + const ScrStreamTypedDesc *desc; /* the typed chunk's type, or NULL */ } ScrWEntry; struct ScrStreamState { @@ -150,6 +152,11 @@ struct ScrStreamState { * Buffer, or both kinds: every entry is an owned ScrDyn* * (scr_stream_add_value). Takes precedence over `encoded`. */ bool dyn_entries; + /* The type of the typed records an object-mode buffer holds + * (scr_stream_push_typed), or NULL. Set by the first entry and kept: + * an entry taken out of the buffer is released through it even after + * the buffer turned to dyn entries, which take precedence. */ + const ScrStreamTypedDesc *typed; /* The objectMode or readableObjectMode option made the buffer hold * object entries (a stream without _read is not a seeded * Readable.from source). */ @@ -220,6 +227,9 @@ struct ScrStreamState { bool final_pending; /* _final called, its callback not yet run */ ScrClosure *inflight_cb; /* owned or NULL */ ScrDyn *cur_enc; /* borrowed: the running write's encoding argument */ + /* the running write's typed chunk type, or NULL (a dyn value or a + * Buffer) */ + const ScrStreamTypedDesc *cur_desc; ScrClosure *write_cb; ScrStreamChunkInv write_inv; ScrClosure *final_cb; @@ -341,12 +351,21 @@ static bool scr_stream_is(const ScrEmitter *em) { /* Readable-buffer entry accessors: bytes chunks, strings once an * encoding is set (lengths then count JS string units, Node's encoded - * accounting), or checked-dynamic values in an object-mode buffer that - * holds other values. */ -typedef enum { SCR_ENTRY_BYTES, SCR_ENTRY_STR, SCR_ENTRY_DYN } ScrEntryKind; + * accounting), checked-dynamic values in an object-mode buffer that holds + * other values, or typed records of the buffer's one type (r.typed). */ +typedef enum { SCR_ENTRY_BYTES, SCR_ENTRY_STR, SCR_ENTRY_DYN, SCR_ENTRY_TYPED } ScrEntryKind; static ScrEntryKind scr_stream_rkind(const ScrStreamState *st) { - return st->r.dyn_entries ? SCR_ENTRY_DYN : st->r.encoded ? SCR_ENTRY_STR : SCR_ENTRY_BYTES; + return st->r.dyn_entries ? SCR_ENTRY_DYN + : st->r.typed ? SCR_ENTRY_TYPED + : st->r.encoded ? SCR_ENTRY_STR + : SCR_ENTRY_BYTES; +} + +/* Two descriptors of one type: the same constant, or the same type key + * (separately emitted constants). */ +static bool scr_stream_desc_same(const ScrStreamTypedDesc *a, const ScrStreamTypedDesc *b) { + return a == b || (a->key_len == b->key_len && memcmp(a->key, b->key, a->key_len) == 0); } static size_t scr_stream_entry_len(const ScrStreamState *st, void *e) { @@ -357,21 +376,25 @@ static size_t scr_stream_entry_len(const ScrStreamState *st, void *e) { /* Releases an entry by the kind it had when it left the buffer: a 'data' * listener may call setEncoding() or push another kind of value while the * entry is out. */ -static void scr_stream_entry_release_kind(void *e, ScrEntryKind kind) { +static void scr_stream_entry_release_kind(const ScrStreamState *st, void *e, ScrEntryKind kind) { if (kind == SCR_ENTRY_DYN) scr_dyn_release((ScrDyn *)e); + else if (kind == SCR_ENTRY_TYPED) st->r.typed->release(e); else if (kind == SCR_ENTRY_STR) scr_str_release((ScrStr *)e); else scr_bytes_release((ScrBytes *)e); } static void scr_stream_entry_release(const ScrStreamState *st, void *e) { - scr_stream_entry_release_kind(e, scr_stream_rkind(st)); + scr_stream_entry_release_kind(st, e, scr_stream_rkind(st)); } /* An entry of the given kind as a dyn value (+1; moves entry). */ -static ScrDyn *scr_stream_entry_box(void *entry, ScrEntryKind kind) { +static ScrDyn *scr_stream_entry_box(const ScrStreamState *st, void *entry, ScrEntryKind kind) { if (kind == SCR_ENTRY_DYN) return entry; ScrDyn *d; - if (kind == SCR_ENTRY_STR) { + if (kind == SCR_ENTRY_TYPED) { + d = st->r.typed->box(entry); + st->r.typed->release(entry); + } else if (kind == SCR_ENTRY_STR) { d = scr_dyn_new_str((ScrStr *)entry); scr_str_release((ScrStr *)entry); } else { @@ -394,13 +417,19 @@ static size_t scr_stream_wchunk_len(const ScrStreamState *st, ScrBytes *chunk) { return 1; } -static ScrBytes *scr_stream_wchunk_retain(const ScrStreamState *st, ScrBytes *chunk) { +/* desc: the chunk's type when it is a typed record (an object-mode side's + * typed write), else NULL. */ +static ScrBytes *scr_stream_wchunk_retain(const ScrStreamState *st, ScrBytes *chunk, + const ScrStreamTypedDesc *desc) { + if (desc) return desc->retain(chunk); if (st->w.values) return (ScrBytes *)scr_dyn_retain((ScrDyn *)chunk); return scr_bytes_retain(chunk); } -static void scr_stream_wchunk_release(const ScrStreamState *st, ScrBytes *chunk) { - if (st->w.values) scr_dyn_release((ScrDyn *)chunk); +static void scr_stream_wchunk_release(const ScrStreamState *st, ScrBytes *chunk, + const ScrStreamTypedDesc *desc) { + if (desc) desc->release(chunk); + else if (st->w.values) scr_dyn_release((ScrDyn *)chunk); else scr_bytes_release(chunk); } @@ -442,8 +471,10 @@ static void scr_stream_state_drop(ScrStreamState *st, bool gc) { for (size_t i = 0; i < st->drain_srcs.n; i++) scr_stream_release(st->drain_srcs.src[i]); } for (size_t i = st->w.head; i < st->w.n; i++) { - bool traced = st->w.values && scr_stream_dyn_entry_traced((ScrDyn *)st->w.q[i].chunk); - if (!gc || !traced) scr_stream_wchunk_release(st, st->w.q[i].chunk); + /* typed chunks are of types the collector does not trace */ + const ScrStreamTypedDesc *desc = st->w.q[i].desc; + bool traced = !desc && st->w.values && scr_stream_dyn_entry_traced((ScrDyn *)st->w.q[i].chunk); + if (!gc || !traced) scr_stream_wchunk_release(st, st->w.q[i].chunk, desc); if (!gc && st->w.q[i].cb) scr_closure_release(st->w.q[i].cb); ScrDyn *enc = st->w.q[i].enc; if (enc && (!gc || !scr_stream_dyn_entry_traced(enc))) scr_dyn_release(enc); @@ -490,7 +521,7 @@ void scr_stream_st_trace(ScrStreamState *st, ScrTraceVisit visit, void *ctx) { for (size_t i = 0; i < st->fin.n; i++) visit(st->fin.cb[i], ctx); for (size_t i = st->w.head; i < st->w.n; i++) { if (st->w.q[i].cb) visit(st->w.q[i].cb, ctx); - if (st->w.values && scr_stream_dyn_entry_traced((ScrDyn *)st->w.q[i].chunk)) + if (!st->w.q[i].desc && st->w.values && scr_stream_dyn_entry_traced((ScrDyn *)st->w.q[i].chunk)) visit(st->w.q[i].chunk, ctx); if (st->w.q[i].enc && scr_stream_dyn_entry_traced(st->w.q[i].enc)) visit(st->w.q[i].enc, ctx); } @@ -896,18 +927,20 @@ static void *scr_stream_rbuf_take(ScrStreamState *st, size_t n) { * slot, so at most one of the first two is non-NULL; the compiler's * stream-data listener thunks dispatch on them (typed listeners unwrap * their declared side; dyn listeners box by tag or take the value). */ -static void scr_stream_emit_data(ScrStream *s, void *chunk, ScrEntryKind kind) { +static void scr_stream_emit_data_entry(ScrStream *s, void *chunk, ScrEntryKind kind) { ScrStreamState *st = s->st; ScrBytes *b = NULL; ScrStr *str = NULL; ScrDyn *value = NULL; + /* a typed record, delivered to pipes as one (no 'data' listeners) */ + const ScrStreamTypedDesc *typed = kind == SCR_ENTRY_TYPED ? st->r.typed : NULL; if (kind == SCR_ENTRY_DYN) { value = chunk; if (value->kind == SCR_DYN_BYTES) b = value->v.bytes; else if (value->kind == SCR_DYN_STR) str = value->v.str; } else if (kind == SCR_ENTRY_STR) { str = chunk; - } else { + } else if (kind == SCR_ENTRY_BYTES) { b = chunk; } /* emit('data') with no listeners does nothing; a piped stream delivers @@ -932,9 +965,10 @@ static void scr_stream_emit_data(ScrStream *s, void *chunk, ScrEntryKind kind) { if (scr_exc_pending()) return; continue; /* the pipes array shifted down over i */ } - bool ok = value ? scr_stream_write_dyn(dst, value, NULL) - : b ? scr_stream_write(dst, b, NULL) - : scr_stream_write_str(dst, str, NULL); + bool ok = typed ? scr_stream_write_typed(dst, chunk, typed, NULL) + : value ? scr_stream_write_dyn(dst, value, NULL) + : b ? scr_stream_write(dst, b, NULL) + : scr_stream_write_str(dst, str, NULL); if (scr_exc_pending()) return; i++; if (!ok && !dst->st->destroyed) { @@ -958,6 +992,20 @@ static void scr_stream_emit_data(ScrStream *s, void *chunk, ScrEntryKind kind) { } } +/* A typed record reaches 'data' listeners as its capsule, and the pipes + * then receive that capsule too; without listeners the pipes receive the + * record. */ +static void scr_stream_emit_data(ScrStream *s, void *chunk, ScrEntryKind kind) { + ScrStreamState *st = s->st; + if (kind != SCR_ENTRY_TYPED || !st->r.data_listeners) { + scr_stream_emit_data_entry(s, chunk, kind); + return; + } + ScrDyn *v = st->r.typed->box(chunk); + scr_stream_emit_data_entry(s, v, SCR_ENTRY_DYN); + scr_dyn_release(v); +} + static void scr_stream_emit_readable_nt(ScrStream *s) { ScrStreamState *st = s->st; st->r.need_readable = false; /* Node's emitReadable clears the ask */ @@ -1076,7 +1124,7 @@ static bool scr_stream_add_chunk(ScrStream *s, ScrBytes *chunk, bool front) { if (st->r.emitted_readable) st->r.emitted_readable = false; ScrEntryKind kind = scr_stream_rkind(st); scr_stream_emit_data(s, entry, kind); - scr_stream_entry_release_kind(entry, kind); + scr_stream_entry_release_kind(st, entry, kind); if (scr_exc_pending()) return false; } else { scr_stream_rbuf_push(st, entry, scr_stream_rkind(st), front); @@ -1188,7 +1236,7 @@ static void *scr_stream_read_n_kind(ScrStream *s, double size, ScrEntryKind *kin if (ret != NULL && !st->error_emitted && !st->close_emitted) { scr_stream_emit_data(s, ret, *kind); if (scr_exc_pending()) { - scr_stream_entry_release_kind(ret, *kind); + scr_stream_entry_release_kind(st, ret, *kind); return NULL; } } @@ -1201,7 +1249,7 @@ static void scr_stream_flow(ScrStream *s) { ScrEntryKind kind; void *b = scr_stream_read_n_kind(s, -1, &kind); if (b == NULL) break; - scr_stream_entry_release_kind(b, kind); + scr_stream_entry_release_kind(st, b, kind); } } @@ -1376,6 +1424,10 @@ static void scr_stream_next_fulfill(ScrStreamState *st, ScrPromise *w /*moves*/, void *chunk /*moves or NULL=EOF*/, ScrEntryKind kind) { void *payload; bool dyn = st->r.next_dyn; + if (chunk != NULL && kind == SCR_ENTRY_TYPED) { + chunk = scr_stream_entry_box(st, chunk, kind); + kind = SCR_ENTRY_DYN; + } if (!dyn && chunk != NULL && kind == SCR_ENTRY_DYN) { ScrDyn *d = chunk; if (d->kind == SCR_DYN_BYTES) { @@ -1400,11 +1452,11 @@ static void scr_stream_next_fulfill(ScrStreamState *st, ScrPromise *w /*moves*/, ScrDyn *result = scr_dyn_new_obj(); scr_dyn_obj_set(result, "value", 5, chunk == NULL ? scr_dyn_retain(scr_dyn_undefined()) - : scr_stream_entry_box(chunk, kind)); + : scr_stream_entry_box(st, chunk, kind)); scr_dyn_obj_set(result, "done", 4, scr_dyn_new_bool(chunk == NULL)); payload = result; } else if (dyn) { - payload = chunk == NULL ? scr_stream_from_done() : scr_stream_entry_box(chunk, kind); + payload = chunk == NULL ? scr_stream_from_done() : scr_stream_entry_box(st, chunk, kind); } else { payload = chunk ? (ScrBytes *)chunk : scr_bytes_new(SCR_BYTES_U8, 0); } @@ -1438,7 +1490,7 @@ static void scr_stream_iter_step(ScrStream *s) { void *chunk = scr_stream_read_n_kind(s, -1, &kind); st->r.next_reading = false; if (scr_exc_pending()) { - if (chunk) scr_stream_entry_release_kind(chunk, kind); + if (chunk) scr_stream_entry_release_kind(st, chunk, kind); ScrPromise *w = st->r.next_waiter; st->r.next_waiter = NULL; if (w) { @@ -1984,7 +2036,7 @@ ScrStream *scr_stream_new_object_readable(ScrClosure *read, ScrStreamReadInv rea static void scr_stream_rbuf_to_dyn(ScrStreamState *st) { ScrEntryKind kind = scr_stream_rkind(st); for (size_t i = st->r.first; i < st->r.first + st->r.n; i++) - st->r.buf[i] = scr_stream_entry_box(st->r.buf[i], kind); + st->r.buf[i] = scr_stream_entry_box(st, st->r.buf[i], kind); st->r.dyn_entries = true; st->r.object_kind_set = true; } @@ -2005,7 +2057,7 @@ static void *scr_stream_entry_convert_kind(ScrStreamState *st, void *entry, ScrE return dec; } if (want != SCR_ENTRY_DYN) scr_stream_rbuf_to_dyn(st); - ScrDyn *d = scr_stream_entry_box(entry, *kind); + ScrDyn *d = scr_stream_entry_box(st, entry, *kind); *kind = SCR_ENTRY_DYN; return d; } @@ -2027,14 +2079,14 @@ static int scr_stream_add_entry(ScrStream *s, void *entry /*moves*/, ScrEntryKin } entry = scr_stream_entry_convert(st, entry, &kind); if (st->r.ended) { - scr_stream_entry_release_kind(entry, kind); + scr_stream_entry_release_kind(st, entry, kind); ScrError *e = scr_stream_mkerr("ERR_STREAM_PUSH_AFTER_EOF", "stream.push() after EOF"); scr_stream_error_or_destroy(s, e); scr_error_release(e); return 0; } if (st->destroyed || st->errored) { - scr_stream_entry_release_kind(entry, kind); + scr_stream_entry_release_kind(st, entry, kind); return 0; } st->r.reading = false; @@ -2042,7 +2094,7 @@ static int scr_stream_add_entry(ScrStream *s, void *entry /*moves*/, ScrEntryKin (st->r.data_listeners || st->pipes.n > 0)) { if (st->r.emitted_readable) st->r.emitted_readable = false; scr_stream_emit_data(s, entry, kind); - scr_stream_entry_release_kind(entry, kind); + scr_stream_entry_release_kind(st, entry, kind); if (scr_exc_pending()) return 0; } else { scr_stream_rbuf_push(st, entry, kind, false); @@ -2053,6 +2105,24 @@ static int scr_stream_add_entry(ScrStream *s, void *entry /*moves*/, ScrEntryKin return !st->r.ended && (st->r.length < st->r.hwm || st->r.length == 0); } +/* push(rec) of a typed record: an object-mode buffer that holds records + * of its type, or nothing yet and no encoding, keeps the record itself; + * any other stream takes its capsule, as push(value) takes it (a typed + * buffer turns to dyn entries before an encoding is set). Borrows rec. */ +bool scr_stream_push_typed(ScrStream *s, void *rec, const ScrStreamTypedDesc *desc) { + ScrStreamState *st = s->st; + if (st->has_r && st->r.object_entries && !st->r.dyn_entries && + (st->r.typed ? scr_stream_desc_same(st->r.typed, desc) + : !st->r.object_kind_set && st->r.length == 0 && !st->r.encoded)) { + if (!st->r.typed) st->r.typed = desc; + return scr_stream_add_entry(s, desc->retain(rec), SCR_ENTRY_TYPED) > 0; + } + ScrDyn *v = desc->box(rec); + bool ret = scr_stream_push_dyn(s, v); + scr_dyn_release(v); + return ret; +} + /* push(value) on an object-mode stream: a string or a Buffer keeps its * kind, any other value is a dyn entry. Borrows v. */ int scr_stream_push_object(ScrStream *s, const ScrDyn *v) { @@ -2110,7 +2180,7 @@ static void scr_stream_on_listener(ScrEmitter *em, ScrStr *name) { /* ── writable internals ───────────────────────────────────────────────── */ static void scr_stream_wq_push(ScrStreamState *st, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/, - ScrDyn *enc /*moves*/) { + ScrDyn *enc /*moves*/, const ScrStreamTypedDesc *desc) { if (st->w.n == st->w.cap) { st->w.cap = st->w.cap ? st->w.cap * 2 : 4; st->w.q = realloc(st->w.q, st->w.cap * sizeof *st->w.q); @@ -2119,32 +2189,41 @@ static void scr_stream_wq_push(ScrStreamState *st, ScrBytes *chunk /*moves*/, Sc st->w.q[st->w.n].chunk = chunk; st->w.q[st->w.n].cb = cb; st->w.q[st->w.n].enc = enc; + st->w.q[st->w.n].desc = desc; st->w.n++; } static void scr_stream_transform_callback(ScrStream *s); -static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/); -static ScrDyn *scr_stream_wentry_value(const ScrStreamState *st, ScrBytes *chunk); +static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/, + const ScrStreamTypedDesc *desc); +static ScrDyn *scr_stream_wentry_value(const ScrStreamState *st, ScrBytes *chunk, + const ScrStreamTypedDesc *desc); static ScrDyn *scr_stream_wentry_encoding(const ScrStreamState *st, ScrDyn *enc); /* One write reaching _write/_transform; the callback reads the write's - * encoding argument through w.cur_enc while it runs. */ + * encoding argument through w.cur_enc, and the type of a typed chunk + * through w.cur_desc, while it runs. */ static void scr_stream_do_write(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/, - ScrDyn *enc /*moves*/) { + ScrDyn *enc /*moves*/, const ScrStreamTypedDesc *desc) { ScrStreamState *st = s->st; + const ScrStreamTypedDesc *outer_desc = st->w.cur_desc; + st->w.cur_desc = desc; if (enc == NULL) { - scr_stream_do_write_inner(s, chunk, cb); + scr_stream_do_write_inner(s, chunk, cb, desc); + st->w.cur_desc = outer_desc; return; } ScrDyn *outer = st->w.cur_enc; st->w.cur_enc = enc; - scr_stream_do_write_inner(s, chunk, cb); + scr_stream_do_write_inner(s, chunk, cb, desc); st->w.cur_enc = outer; + st->w.cur_desc = outer_desc; scr_dyn_release(enc); } -static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/) { +static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, ScrClosure *cb /*moves*/, + const ScrStreamTypedDesc *desc) { ScrStreamState *st = s->st; st->w.writing = true; st->w.wsync = true; @@ -2153,16 +2232,17 @@ static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, S if (st->is_transform || st->passthrough) st->xf_length = st->r.length; if (st->passthrough && st->transform_cb == NULL) { /* identity: push through, complete synchronously */ - if (st->w.values) scr_stream_push_dyn(s, (ScrDyn *)chunk); + if (desc) scr_stream_push_typed(s, chunk, desc); + else if (st->w.values) scr_stream_push_dyn(s, (ScrDyn *)chunk); else scr_stream_add_chunk(s, chunk, false); - scr_stream_wchunk_release(st, chunk); + scr_stream_wchunk_release(st, chunk, desc); if (!scr_exc_pending()) scr_stream_transform_callback(s); st->w.wsync = false; return; } if (st->is_transform || st->passthrough) { if (st->transform_cb == NULL) { - scr_stream_wchunk_release(st, chunk); + scr_stream_wchunk_release(st, chunk, desc); ScrError *e = scr_stream_mkerr("ERR_METHOD_NOT_IMPLEMENTED", "The _transform() method is not implemented"); st->w.wsync = false; scr_stream_error_or_destroy(s, e); @@ -2170,7 +2250,7 @@ static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, S return; } st->transform_inv(st->transform_cb, s, chunk); - scr_stream_wchunk_release(st, chunk); + scr_stream_wchunk_release(st, chunk, desc); st->w.wsync = false; return; } @@ -2178,17 +2258,17 @@ static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, S /* Writable.prototype._write: this._writev([{ chunk, encoding }], cb) */ ScrDyn *chunks = scr_dyn_new_arr(); ScrDyn *entry = scr_dyn_new_obj(); - scr_dyn_obj_set(entry, "chunk", 5, scr_stream_wentry_value(st, chunk)); + scr_dyn_obj_set(entry, "chunk", 5, scr_stream_wentry_value(st, chunk, desc)); scr_dyn_obj_set(entry, "encoding", 8, scr_stream_wentry_encoding(st, st->w.cur_enc)); scr_dyn_arr_push(chunks, entry); - scr_stream_wchunk_release(st, chunk); + scr_stream_wchunk_release(st, chunk, desc); st->w.writev_inv(st->w.writev_cb, s, chunks); scr_dyn_release(chunks); st->w.wsync = false; return; } if (st->w.write_cb == NULL) { - scr_stream_wchunk_release(st, chunk); + scr_stream_wchunk_release(st, chunk, desc); ScrError *e = scr_stream_mkerr("ERR_METHOD_NOT_IMPLEMENTED", "The _write() method is not implemented"); st->w.wsync = false; scr_stream_error_or_destroy(s, e); @@ -2196,7 +2276,7 @@ static void scr_stream_do_write_inner(ScrStream *s, ScrBytes *chunk /*moves*/, S return; } st->w.write_inv(st->w.write_cb, s, chunk); - scr_stream_wchunk_release(st, chunk); + scr_stream_wchunk_release(st, chunk, desc); st->w.wsync = false; } @@ -2263,7 +2343,7 @@ static void scr_stream_error_buffer(ScrStream *s) { while (st->w.head < st->w.n && !scr_exc_pending()) { ScrWEntry e = st->w.q[st->w.head++]; st->w.length -= scr_stream_wchunk_len(st, e.chunk); - scr_stream_wchunk_release(st, e.chunk); + scr_stream_wchunk_release(st, e.chunk, e.desc); if (e.enc) scr_dyn_release(e.enc); if (e.cb) { SCR_WCB_CALL_ERROF(e.cb, st); @@ -2360,11 +2440,13 @@ static SCR_COLD void scr_stream_writev_buffer(ScrStream *s) { } st->w.head = 0; st->w.n = 0; - scr_stream_do_write(s, all, cbs, NULL); + scr_stream_do_write(s, all, cbs, NULL, NULL); } /* A write's chunk and encoding as _writev receives them (+1 each). */ -static ScrDyn *scr_stream_wentry_value(const ScrStreamState *st, ScrBytes *chunk) { +static ScrDyn *scr_stream_wentry_value(const ScrStreamState *st, ScrBytes *chunk, + const ScrStreamTypedDesc *desc) { + if (desc) return desc->box(chunk); if (st->w.values) return scr_dyn_retain((ScrDyn *)chunk); return chunk->is_buffer ? scr_dyn_new_buffer(chunk) : scr_dyn_new_bytes(chunk); } @@ -2451,11 +2533,11 @@ static void scr_stream_writev_batch(ScrStream *s) { if (!(encoding->kind == SCR_DYN_STR && encoding->v.str->len == 6 && memcmp(encoding->v.str->data, "buffer", 6) == 0)) all_buffers = false; - scr_dyn_obj_set(entry, "chunk", 5, scr_stream_wentry_value(st, e.chunk)); + scr_dyn_obj_set(entry, "chunk", 5, scr_stream_wentry_value(st, e.chunk, e.desc)); scr_dyn_obj_set(entry, "encoding", 8, encoding); scr_dyn_obj_set(entry, "callback", 8, scr_stream_writev_entry_cb(e.cb)); scr_dyn_arr_push(chunks, entry); - scr_stream_wchunk_release(st, e.chunk); + scr_stream_wchunk_release(st, e.chunk, e.desc); if (e.enc) scr_dyn_release(e.enc); if (e.cb) { cbs->caps[k] = scr_box_new_obj(&scr_closure_retain_v, &scr_closure_release_v, @@ -2489,7 +2571,7 @@ static void scr_stream_clear_buffer_entries(ScrStream *s) { st->w.head = 0; st->w.n = 0; } - scr_stream_do_write(s, e.chunk, e.cb, e.enc); + scr_stream_do_write(s, e.chunk, e.cb, e.enc, e.desc); } st->w.buffer_processing = false; } @@ -4851,7 +4933,9 @@ static void scr_compose_destroyed(ScrCompose *c, ScrError *err) { /* d._write: head.write(chunk) ? callback() : (ondrain = callback) */ static void scr_compose_write_inv(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) { ScrCompose *c = scr_box_get_ref(cb->caps[0]); - bool ok = scr_stream_writes_values(s) ? scr_stream_write_dyn(c->head, (ScrDyn *)chunk, NULL) + const ScrStreamTypedDesc *desc = s->st->w.cur_desc; + bool ok = desc ? scr_stream_write_typed(c->head, chunk, desc, NULL) + : scr_stream_writes_values(s) ? scr_stream_write_dyn(c->head, (ScrDyn *)chunk, NULL) : scr_stream_write(c->head, chunk, NULL); if (ok) { if (!scr_exc_pending()) scr_stream_write_done(s, NULL); @@ -5214,7 +5298,9 @@ static void scr_dupx_r_eos(ScrClosure *cb, ScrStream *s, ScrError *err) { /* d._write: w.write(chunk) ? callback() : (ondrain = callback) */ static void scr_dupx_write_inv(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) { ScrDuplexify *x = scr_box_get_ref(cb->caps[0]); - bool ok = scr_stream_writes_values(s) ? scr_stream_write_dyn(x->w, (ScrDyn *)chunk, NULL) + const ScrStreamTypedDesc *desc = s->st->w.cur_desc; + bool ok = desc ? scr_stream_write_typed(x->w, chunk, desc, NULL) + : scr_stream_writes_values(s) ? scr_stream_write_dyn(x->w, (ScrDyn *)chunk, NULL) : scr_stream_write(x->w, chunk, NULL); if (ok) { if (!scr_exc_pending()) scr_stream_write_done(s, NULL); @@ -5265,7 +5351,7 @@ static void scr_dupx_pull(ScrDuplexify *x) { return; } bool more; - if (d->st->r.object_entries) { + if (d->st->r.object_entries && kind != SCR_ENTRY_TYPED) { more = scr_stream_add_entry(d, chunk, kind) > 0; } else if (kind == SCR_ENTRY_STR) { more = scr_stream_push_str(d, chunk); @@ -5273,6 +5359,9 @@ static void scr_dupx_pull(ScrDuplexify *x) { } else if (kind == SCR_ENTRY_DYN) { more = scr_stream_push_dyn(d, chunk); scr_dyn_release(chunk); + } else if (kind == SCR_ENTRY_TYPED) { + more = scr_stream_push_typed(d, chunk, x->r->st->r.typed); + x->r->st->r.typed->release(chunk); } else { more = scr_stream_push(d, chunk); scr_bytes_release(chunk); @@ -6755,7 +6844,7 @@ bool scr_stream_emit_value(ScrStream *s, ScrStr *name, ScrDyn *v) { ScrBytes *scr_stream_chunk_bytes(ScrStream *s, ScrBytes *chunk) { if (!scr_stream_writes_values(s)) return scr_bytes_retain(chunk); ScrDyn *v = (ScrDyn *)chunk; - if (v->kind == SCR_DYN_BYTES) return scr_bytes_raw_view(v->v.bytes); + if (!s->st->w.cur_desc && v->kind == SCR_DYN_BYTES) return scr_bytes_raw_view(v->v.bytes); static const char msg[] = "a stream callback declaring a Buffer chunk received an object-mode value that is not a Buffer"; scr_throw_error_msg(SCR_ERR_TYPE, msg, sizeof msg - 1); @@ -6763,10 +6852,17 @@ ScrBytes *scr_stream_chunk_bytes(ScrStream *s, ScrBytes *chunk) { } ScrDyn *scr_stream_chunk_value(ScrStream *s, ScrBytes *chunk) { + const ScrStreamTypedDesc *desc = s->st->w.cur_desc; + if (desc) return desc->box(chunk); if (scr_stream_writes_values(s)) return scr_dyn_retain((ScrDyn *)chunk); return scr_dyn_new_buffer(chunk); } +void *scr_stream_chunk_typed(ScrStream *s, ScrBytes *chunk, const ScrStreamTypedDesc *desc) { + const ScrStreamTypedDesc *cur = s->st->w.cur_desc; + return cur && scr_stream_desc_same(cur, desc) ? desc->retain(chunk) : NULL; +} + ScrStr *scr_stream_chunk_encoding(ScrStream *s) { if (!scr_stream_writes_values(s)) return scr_str_new("buffer", 6); ScrDyn *enc = s->st->w.cur_enc; @@ -7018,7 +7114,7 @@ ScrDyn *scr_stream_read_dyn(ScrStream *s, ScrDyn *size) { scr_bytes_release(chunk); return result; } - return scr_stream_entry_box(chunk, kind); + return scr_stream_entry_box(s->st, chunk, kind); } /* read() answers STRINGS once encoded: the static result type here is @@ -7041,6 +7137,10 @@ ScrBytes *scr_stream_read(ScrStream *s, double size) { } ScrEntryKind kind; void *chunk = scr_stream_read_n_kind(s, size, &kind); + if (chunk && kind == SCR_ENTRY_TYPED) { + chunk = scr_stream_entry_box(s->st, chunk, kind); + kind = SCR_ENTRY_DYN; + } if (chunk && kind == SCR_ENTRY_DYN) { /* an object-mode value: a Buffer is the result, anything else is not */ ScrDyn *d = chunk; @@ -7074,6 +7174,8 @@ ScrBytes *scr_stream_read(ScrStream *s, double size) { ScrStream *scr_stream_set_encoding(ScrStream *s, ScrStr *enc) { ScrStreamState *st = s->st; if (!st->has_r) return scr_stream_retain(s); + /* typed records are values like the dyn entries */ + if (st->r.typed && !st->r.dyn_entries) scr_stream_rbuf_to_dyn(st); if (st->r.dyn_entries) { /* an object-mode buffer of values: Buffers pushed later decode */ if (st->r.enc) scr_str_release(st->r.enc); @@ -7215,8 +7317,9 @@ ScrStream *scr_stream_unpipe(ScrStream *src, ScrStream *dst) { /* ── the writable surface ─────────────────────────────────────────────── */ -static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed*/, - ScrClosure *cb /*moves*/, ScrDyn *enc /*borrowed*/) { +static bool scr_stream_write_chunk_desc(ScrStream *s, ScrBytes *chunk /*borrowed*/, + ScrClosure *cb /*moves*/, ScrDyn *enc /*borrowed*/, + const ScrStreamTypedDesc *desc) { ScrStreamState *st = s->st; if (!st->has_w || st->destroyed || st->w.ending) { ScrError *e = st->has_w && !st->destroyed @@ -7236,9 +7339,9 @@ static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed* st->w.length += scr_stream_wchunk_len(st, chunk); if (enc) enc = scr_dyn_retain(enc); if (st->w.writing || st->w.corked > 0 || st->constructing) { - scr_stream_wq_push(st, scr_stream_wchunk_retain(st, chunk), cb, enc); + scr_stream_wq_push(st, scr_stream_wchunk_retain(st, chunk, desc), cb, enc, desc); } else { - scr_stream_do_write(s, scr_stream_wchunk_retain(st, chunk), cb, enc); + scr_stream_do_write(s, scr_stream_wchunk_retain(st, chunk, desc), cb, enc, desc); } /* Node computes the answer AFTER the (possibly synchronous) write: * a sync completion has already drained its bytes from length. Only @@ -7249,6 +7352,11 @@ static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed* return below && !st->destroyed && !st->errored; } +static bool scr_stream_write_chunk_enc(ScrStream *s, ScrBytes *chunk /*borrowed*/, + ScrClosure *cb /*moves*/, ScrDyn *enc /*borrowed*/) { + return scr_stream_write_chunk_desc(s, chunk, cb, enc, NULL); +} + static bool scr_stream_write_chunk(ScrStream *s, ScrBytes *chunk /*borrowed*/, ScrClosure *cb /*moves*/) { return scr_stream_write_chunk_enc(s, chunk, cb, NULL); } @@ -7264,6 +7372,19 @@ static bool scr_stream_write_value(ScrStream *s, ScrDyn *v, ScrClosure *cb) { return scr_stream_write_chunk_enc(s, (ScrBytes *)v, cb, scr_stream_cb_enc(cb)); } +/* write(rec) of a typed record: on an object-mode side the record itself + * is the chunk; another side takes its capsule, as write(value) takes it. + * Borrows rec. */ +bool scr_stream_write_typed(ScrStream *s, void *rec, const ScrStreamTypedDesc *desc, + ScrClosure *cb) { + if (s->st->has_w && s->st->w.object_mode) + return scr_stream_write_chunk_desc(s, rec, cb, scr_stream_cb_enc(cb), desc); + ScrDyn *v = desc->box(rec); + bool ret = scr_stream_write_dyn(s, v, cb); + scr_dyn_release(v); + return ret; +} + /* A Buffer write on a byte-mode side: with decodeStrings: false the * queue holds values, so the Buffer is boxed with its 'buffer' encoding. * Borrows chunk. */ @@ -7642,6 +7763,32 @@ void scr_stream_transform_done_value(ScrStream *s, ScrError *err /*moves*/, ScrD else scr_stream_transform_callback(s); } +void scr_stream_transform_done_typed(ScrStream *s, ScrError *err /*moves*/, void *rec, + const ScrStreamTypedDesc *desc) { + if (!err && rec) scr_stream_push_typed(s, rec, desc); + if (scr_exc_pending()) { + if (err) scr_error_release(err); + return; + } + if (err) scr_stream_write_done(s, err); + else scr_stream_transform_callback(s); +} + +void scr_stream_flush_done_typed(ScrStream *s, ScrError *err /*moves*/, void *rec, + const ScrStreamTypedDesc *desc) { + s->st->flush_pending = false; + if (err) { + scr_stream_error_or_destroy(s, err); + scr_error_release(err); + return; + } + if (rec) scr_stream_push_typed(s, rec, desc); + if (scr_exc_pending()) return; + scr_stream_push_null(s); + if (scr_exc_pending()) return; + scr_stream_finish_maybe(s); +} + void scr_stream_flush_done_value(ScrStream *s, ScrError *err /*moves*/, ScrDyn *v) { s->st->flush_pending = false; if (err) { @@ -8191,7 +8338,7 @@ static void scr_stream_run_tick(ScrStreamTick *t) { if (st->has_r) { ScrEntryKind kind; void *chunk = scr_stream_read_n_kind(s, 0, &kind); - if (chunk) scr_stream_entry_release_kind(chunk, kind); + if (chunk) scr_stream_entry_release_kind(st, chunk, kind); } break; } diff --git a/packages/runtime/src/scr_stream.h b/packages/runtime/src/scr_stream.h index f5a8207f2..6b8a4dfdc 100644 --- a/packages/runtime/src/scr_stream.h +++ b/packages/runtime/src/scr_stream.h @@ -574,6 +574,37 @@ bool scr_stream_chunk_done(const struct ScrDyn *v); void scr_stream_transform_done_value(ScrStream *s, ScrError *err, struct ScrDyn *v); void scr_stream_flush_done_value(ScrStream *s, ScrError *err, struct ScrDyn *v); +/* Typed object-mode chunks. A compiled push, completion or write of a + * record whose type has a descriptor (one internal constant per type, + * emitted by the compiler) keeps the record itself in the stream's + * buffers. A consumer that reads the chunk as a checked-dynamic value + * gets the record's capsule from `box`, the conversion compiled code + * uses for the record (same referent, so the same JS object). A side + * that is not in object mode, or a buffer that holds other values, takes + * the chunk boxed, as push(value) takes it. */ +typedef struct ScrStreamTypedDesc { + const char *key; /* the type key (the capsule's type key) */ + size_t key_len; + void *(*retain)(void *); + void (*release)(void *); + struct ScrDyn *(*box)(void *rec); /* +1 capsule; borrows rec */ +} ScrStreamTypedDesc; +/* push(rec): borrows rec. */ +bool scr_stream_push_typed(ScrStream *s, void *rec, const ScrStreamTypedDesc *desc); +/* write(rec[, cb]): borrows rec; cb moves. */ +bool scr_stream_write_typed(ScrStream *s, void *rec, const ScrStreamTypedDesc *desc, + ScrClosure *cb); +/* callback(err, rec) of a transform or flush: rec borrowed, NULL when the + * callback passed no data; err moves. */ +void scr_stream_transform_done_typed(ScrStream *s, ScrError *err, void *rec, + const ScrStreamTypedDesc *desc); +void scr_stream_flush_done_typed(ScrStream *s, ScrError *err, void *rec, + const ScrStreamTypedDesc *desc); +/* A compiled write or transform callback's chunk declared with desc's + * type: the record (+1) when the write in progress carries one of that + * type, else NULL (the caller converts scr_stream_chunk_value). */ +void *scr_stream_chunk_typed(ScrStream *s, ScrBytes *chunk, const ScrStreamTypedDesc *desc); + /* Runtime-backed streams (the fs streams in scr_fs_stream.c). * - construct: Node's _construct. The hook runs on a tick after * construction; until construct_done's tick, reads, buffered writes, diff --git a/tests/corpus/stream-typed-chunk-boxed-kinds.ts b/tests/corpus/stream-typed-chunk-boxed-kinds.ts new file mode 100644 index 000000000..39bd8789a --- /dev/null +++ b/tests/corpus/stream-typed-chunk-boxed-kinds.ts @@ -0,0 +1,89 @@ +// Arrays, union-typed values, strings and Buffers in object-mode streams, +// alone and mixed with records in one stream, reach their consumers as +// the values that were pushed. +import { Readable, Transform, Writable } from "node:stream"; + +type Pair = { a: number; b: number }; +type Shape = { kind: "circle"; r: number } | { kind: "square"; side: number }; + +const log: string[] = []; +const arrays: number[][] = [[1, 2], [3], []]; +const shapes: Shape[] = [ + { kind: "circle", r: 2 }, + { kind: "square", side: 3 }, +]; +const first: Pair = { a: 1, b: 2 }; +const second: Pair = { a: 3, b: 4 }; +const pairs: Pair[] = [first, second]; + +const arraySource = new Readable({ objectMode: true, read() {} }); +for (const arr of arrays) arraySource.push(arr); +arraySource.push(null); +const arraySink = new Writable({ + objectMode: true, + write(arr: number[], _enc, cb) { + log.push(`array [${arr.join(",")}] same=${arrays.includes(arr)}`); + cb(); + }, +}); + +const shapeSource = new Readable({ objectMode: true, read() {} }); +for (const shape of shapes) shapeSource.push(shape); +shapeSource.push(null); +const area = new Transform({ + objectMode: true, + transform(shape: Shape, _enc, cb) { + const size = shape.kind === "circle" ? shape.r : shape.side; + log.push(`shape ${shape.kind} ${size} same=${shapes.includes(shape)}`); + cb(null, shape); + }, +}); +const shapeSink = new Writable({ + objectMode: true, + write(shape: Shape, _enc, cb) { + log.push(`shape out ${shape.kind} same=${shapes.includes(shape)}`); + cb(); + }, +}); + +const mixed = new Readable({ objectMode: true, read() {} }); +mixed.push(first); +mixed.push("text"); +mixed.push(second); +mixed.push(Buffer.from("buf")); +mixed.push(null); +const mixedSink = new Writable({ + objectMode: true, + write(value: any, _enc, cb) { + if (typeof value === "string") log.push(`mixed string ${value}`); + else if (Buffer.isBuffer(value)) log.push(`mixed buffer ${value.toString()}`); + else log.push(`mixed pair ${value.a},${value.b} same=${pairs.includes(value)}`); + cb(); + }, +}); + +const later = new Readable({ objectMode: true, read() {} }); +later.push("first"); +later.push(first); +later.push(null); +const laterSink = new Writable({ + objectMode: true, + write(value: any, _enc, cb) { + log.push(`later ${typeof value} ${value === first}`); + cb(); + }, +}); + +let finished = 0; +const done = (): void => { + finished++; + if (finished === 4) for (const line of log) console.log(line); +}; +arraySink.on("finish", done); +shapeSink.on("finish", done); +mixedSink.on("finish", done); +laterSink.on("finish", done); +arraySource.pipe(arraySink); +shapeSource.pipe(area).pipe(shapeSink); +mixed.pipe(mixedSink); +later.pipe(laterSink); diff --git a/tests/corpus/stream-typed-chunk-data-and-pipe.ts b/tests/corpus/stream-typed-chunk-data-and-pipe.ts new file mode 100644 index 000000000..ba5db96e0 --- /dev/null +++ b/tests/corpus/stream-typed-chunk-data-and-pipe.ts @@ -0,0 +1,55 @@ +// A 'data' listener and a pipe on one object-mode stream receive the same +// object; a property the listener adds is visible to the untyped +// consumer downstream, and the typed consumer sees the same object. +import { Readable, Writable } from "node:stream"; + +type Point = { x: number; y: number }; + +const points: Point[] = []; +let i = 0; +const source = new Readable({ + objectMode: true, + read() { + if (i < 3) { + const p = { x: i, y: i * i }; + points.push(p); + i++; + this.push(p); + } else { + this.push(null); + } + }, +}); +const log: string[] = []; +let lastData: any = null; +source.on("data", (p: any) => { + lastData = p; + p.label = `p${p.x}`; + log.push(`data ${p.x},${p.y} same=${p === points[p.x]}`); +}); +const typed = new Writable({ + objectMode: true, + write(p: Point, _enc, cb) { + log.push(`typed ${p.x},${p.y} same=${p === points[p.x]} last=${p === lastData}`); + cb(); + }, +}); +const untyped = new Writable({ + objectMode: true, + write(p: any, _enc, cb) { + log.push(`untyped ${p.x} ${p.label} keys=${Object.keys(p).join("/")} same=${p === points[p.x]}`); + cb(); + }, +}); +let finished = 0; +const done = (): void => { + finished++; + if (finished === 2) { + for (const line of log) console.log(line); + console.log(JSON.stringify(points)); + } +}; +typed.on("finish", done); +untyped.on("finish", done); +source.pipe(typed); +source.pipe(untyped); diff --git a/tests/corpus/stream-typed-chunk-for-await.ts b/tests/corpus/stream-typed-chunk-for-await.ts new file mode 100644 index 000000000..f43bcd835 --- /dev/null +++ b/tests/corpus/stream-typed-chunk-for-await.ts @@ -0,0 +1,44 @@ +// for await over an object-mode Readable and over a Transform's output +// yields the objects that were pushed. +import { Readable, Transform } from "node:stream"; + +type Entry = { key: string; value: number }; + +const entries: Entry[] = []; +let n = 0; +const source = new Readable({ + objectMode: true, + read() { + if (n < 4) { + const entry = { key: `k${n}`, value: n * 10 }; + entries.push(entry); + n++; + this.push(entry); + } else { + this.push(null); + } + }, +}); +const outputs: Entry[] = []; +const double = new Transform({ + objectMode: true, + transform(entry: Entry, _enc, cb) { + const out = { key: entry.key.toUpperCase(), value: entry.value * 2 }; + outputs.push(out); + cb(null, out); + }, +}); + +async function main(): Promise { + const copy = new Readable({ objectMode: true, read() {} }); + for (const entry of [{ key: "a", value: 1 }, { key: "b", value: 2 }]) copy.push(entry); + copy.push(null); + for await (const entry of copy) console.log(`copy ${entry.key}=${entry.value}`); + let index = 0; + for await (const entry of source.pipe(double)) { + console.log(`${entry.key}=${entry.value} same=${entry === outputs[index]} source=${entries.includes(entry)}`); + index++; + } + console.log(`count ${index}`); +} +void main(); diff --git a/tests/corpus/stream-typed-chunk-identity.ts b/tests/corpus/stream-typed-chunk-identity.ts new file mode 100644 index 000000000..a22f7bfc4 --- /dev/null +++ b/tests/corpus/stream-typed-chunk-identity.ts @@ -0,0 +1,54 @@ +// Object-mode chunks keep their identity through a Readable, a Transform +// and a Writable: each stage receives the object the previous stage +// pushed or passed to its callback. +import { Readable, Transform, Writable } from "node:stream"; + +type Row = { id: number; qty: number }; + +const pushed: Row[] = []; +const made: Row[] = []; +const seen = new Map(); +let produced = 0; +const source = new Readable({ + objectMode: true, + highWaterMark: 2, + read() { + while (produced < 6) { + produced++; + const row = { id: produced, qty: produced * 3 }; + pushed.push(row); + seen.set(row, produced); + if (!this.push(row)) return; + } + this.push(null); + }, +}); +const lines: string[] = []; +const pass = new Transform({ + objectMode: true, + transform(row: Row, _enc, cb) { + lines.push(`transform ${row.id} same=${row === pushed[row.id - 1]} map=${seen.get(row)}`); + if (row.id % 2 === 0) { + cb(null, row); + return; + } + const next = { id: row.id, qty: row.qty + 1 }; + made.push(next); + cb(null, next); + }, +}); +const sink = new Writable({ + objectMode: true, + highWaterMark: 1, + write(row: Row, _enc, cb) { + const original = row === pushed[row.id - 1]; + const fresh = made.includes(row); + lines.push(`write ${row.id} qty=${row.qty} original=${original} fresh=${fresh}`); + cb(); + }, +}); +sink.on("finish", () => { + for (const line of lines) console.log(line); + console.log(`pushed ${pushed.length} made ${made.length}`); +}); +source.pipe(pass).pipe(sink); diff --git a/tests/corpus/stream-typed-chunk-mismatch.ts b/tests/corpus/stream-typed-chunk-mismatch.ts new file mode 100644 index 000000000..667c96064 --- /dev/null +++ b/tests/corpus/stream-typed-chunk-mismatch.ts @@ -0,0 +1,41 @@ +// A Transform that declares fewer fields than the chunks carry: the +// fields it does not declare stay on the chunks it passes on, so the +// Writable after it, which declares them, reads them, and an untyped +// 'data' listener on the Transform lists them. +import { Readable, Transform, Writable } from "node:stream"; + +type Full = { id: number; name: string; score: number }; +type Narrow = { id: number }; + +let k = 0; +const source = new Readable({ + objectMode: true, + read() { + if (k < 3) { + this.push({ id: k, name: `n${k}`, score: k * 1.5 }); + k++; + } else { + this.push(null); + } + }, +}); +const log: string[] = []; +const narrow = new Transform({ + objectMode: true, + transform(row: Narrow, _enc, cb) { + log.push(`narrow ${row.id} ${JSON.stringify(row)}`); + cb(null, row); + }, +}); +narrow.on("data", (row: any) => log.push(`data ${Object.keys(row).join("/")} ${row.name}`)); +const full = new Writable({ + objectMode: true, + write(row: Full, _enc, cb) { + log.push(`full ${row.id} ${row.name} ${row.score} ${JSON.stringify(row)}`); + cb(); + }, +}); +full.on("finish", () => { + for (const line of log) console.log(line); +}); +source.pipe(narrow).pipe(full); diff --git a/tests/corpus/stream-typed-chunk-mutation.ts b/tests/corpus/stream-typed-chunk-mutation.ts new file mode 100644 index 000000000..d0ca0b0a2 --- /dev/null +++ b/tests/corpus/stream-typed-chunk-mutation.ts @@ -0,0 +1,38 @@ +// Chunks are shared, not copied: a change to an object after push() is +// visible to the consumer that reads it later, and a change a Transform +// makes reaches the next stage. +import { PassThrough, Transform, Writable } from "node:stream"; + +type Cell = { row: number; value: number }; + +const cells: Cell[] = []; +const log: string[] = []; +const bump = new Transform({ + objectMode: true, + transform(cell: Cell, _enc, cb) { + log.push(`transform ${cell.row} value=${cell.value}`); + cell.value += 100; + cb(null, cell); + }, +}); +const sink = new Writable({ + objectMode: true, + write(cell: Cell, _enc, cb) { + log.push(`write ${cell.row} value=${cell.value} same=${cell === cells[cell.row]}`); + cb(); + }, +}); +const source = new PassThrough({ objectMode: true }); +source.pause(); +for (let r = 0; r < 3; r++) { + const cell = { row: r, value: r }; + cells.push(cell); + source.write(cell); + cell.value = r * 10; +} +sink.on("finish", () => { + for (const line of log) console.log(line); + console.log(cells.map((c) => c.value).join(",")); +}); +source.pipe(bump).pipe(sink); +source.end(); diff --git a/tests/corpus/stream-typed-chunk-read-paused.ts b/tests/corpus/stream-typed-chunk-read-paused.ts new file mode 100644 index 000000000..16813698d --- /dev/null +++ b/tests/corpus/stream-typed-chunk-read-paused.ts @@ -0,0 +1,26 @@ +// read() on a paused object-mode stream answers the pushed objects +// themselves, in order, one per call. +import { Readable } from "node:stream"; + +type Job = { id: number; done: boolean }; + +const jobs: Job[] = []; +const source = new Readable({ objectMode: true, read() {} }); +for (let i = 0; i < 3; i++) { + const job = { id: i, done: false }; + jobs.push(job); + source.push(job); +} +console.log(`length ${source.readableLength}`); +const first = source.read(); +console.log(`first ${first.id} same=${first === jobs[0]}`); +first.done = true; +console.log(`jobs[0].done ${jobs[0]!.done}`); +source.push(null); +source.on("readable", () => { + let job: any; + while ((job = source.read()) !== null) { + console.log(`read ${job.id} same=${job === jobs[job.id]} done=${job.done}`); + } +}); +source.on("end", () => console.log("end")); diff --git a/tests/corpus/stream-typed-chunk-weakmap.ts b/tests/corpus/stream-typed-chunk-weakmap.ts new file mode 100644 index 000000000..aa1219c7b --- /dev/null +++ b/tests/corpus/stream-typed-chunk-weakmap.ts @@ -0,0 +1,53 @@ +// A WeakMap and a WeakSet keyed by the objects a Readable pushes find +// them again in a typed Writable, in an untyped 'data' listener, and in +// an untyped Writable downstream. +import { PassThrough, Readable, Writable } from "node:stream"; + +type Item = { name: string; n: number }; + +const tags = new WeakMap(); +const marked = new WeakSet(); +const items: Item[] = []; +for (let i = 0; i < 4; i++) { + const item = { name: `item${i}`, n: i }; + items.push(item); + tags.set(item, `tag-${i}`); + if (i % 2 === 0) marked.add(item); +} +let next = 0; +const source = new Readable({ + objectMode: true, + read() { + const item = items[next++]; + if (item === undefined) this.push(null); + else this.push(item); + }, +}); +const out: string[] = []; +const typed = new Writable({ + objectMode: true, + write(item: Item, _enc, cb) { + out.push(`typed ${item.name} ${tags.get(item)} ${marked.has(item)}`); + cb(); + }, +}); +const middle = new PassThrough({ objectMode: true }); +middle.on("data", (item: any) => { + out.push(`data ${item.name} ${tags.get(item)} ${marked.has(item)}`); +}); +const untyped = new Writable({ + objectMode: true, + write(item: any, _enc, cb) { + out.push(`untyped ${item.name} ${tags.get(item)} ${items.indexOf(item)}`); + cb(); + }, +}); +let finished = 0; +const done = (): void => { + finished++; + if (finished === 2) for (const line of out) console.log(line); +}; +typed.on("finish", done); +untyped.on("finish", done); +source.pipe(typed); +source.pipe(middle).pipe(untyped); diff --git a/tests/corpus/stream-typed-chunk-writev.ts b/tests/corpus/stream-typed-chunk-writev.ts new file mode 100644 index 000000000..8ddca02be --- /dev/null +++ b/tests/corpus/stream-typed-chunk-writev.ts @@ -0,0 +1,39 @@ +// writev on an object-mode Writable receives the written objects in its +// batch, with their encodings, and write() receives them one at a time. +import { Writable } from "node:stream"; + +type Msg = { seq: number; text: string }; + +const sent: Msg[] = []; +const lines: string[] = []; +const batched = new Writable({ + objectMode: true, + write(msg: Msg, _enc, cb) { + lines.push(`write ${msg.seq} same=${msg === sent[msg.seq]}`); + cb(); + }, + writev(chunks: Array<{ chunk: Msg; encoding: any }>, cb) { + const parts = chunks.map( + (c) => `${c.chunk.seq}:${c.chunk.text}:${c.chunk === sent[c.chunk.seq]}:${String(c.encoding)}`, + ); + lines.push(`writev ${parts.join(" ")}`); + cb(); + }, +}); +const make = (seq: number): Msg => { + const msg = { seq, text: `m${seq}` }; + sent.push(msg); + return msg; +}; +batched.write(make(0)); +batched.cork(); +batched.write(make(1)); +batched.write(make(2)); +batched.write(make(3)); +process.nextTick(() => { + batched.uncork(); + batched.end(make(4)); +}); +batched.on("finish", () => { + for (const line of lines) console.log(line); +}); diff --git a/tests/harness/corpus-costs/stream-typed-chunk-boxed-kinds.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-boxed-kinds.ts.json new file mode 100644 index 000000000..81838ef3a --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-boxed-kinds.ts.json @@ -0,0 +1 @@ +{ "plain": 2797, "san": 4727 } diff --git a/tests/harness/corpus-costs/stream-typed-chunk-data-and-pipe.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-data-and-pipe.ts.json new file mode 100644 index 000000000..32e63fa22 --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-data-and-pipe.ts.json @@ -0,0 +1 @@ +{ "plain": 2910, "san": 4166 } diff --git a/tests/harness/corpus-costs/stream-typed-chunk-for-await.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-for-await.ts.json new file mode 100644 index 000000000..3078acbc2 --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-for-await.ts.json @@ -0,0 +1 @@ +{ "plain": 2833, "san": 5275 } diff --git a/tests/harness/corpus-costs/stream-typed-chunk-identity.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-identity.ts.json new file mode 100644 index 000000000..9c68a0fba --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-identity.ts.json @@ -0,0 +1 @@ +{ "plain": 2525, "san": 4060 } diff --git a/tests/harness/corpus-costs/stream-typed-chunk-mismatch.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-mismatch.ts.json new file mode 100644 index 000000000..23ca83a0b --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-mismatch.ts.json @@ -0,0 +1 @@ +{ "plain": 2953, "san": 4748 } diff --git a/tests/harness/corpus-costs/stream-typed-chunk-mutation.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-mutation.ts.json new file mode 100644 index 000000000..f194675f7 --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-mutation.ts.json @@ -0,0 +1 @@ +{ "plain": 2493, "san": 4446 } diff --git a/tests/harness/corpus-costs/stream-typed-chunk-read-paused.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-read-paused.ts.json new file mode 100644 index 000000000..1f697e7e0 --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-read-paused.ts.json @@ -0,0 +1 @@ +{ "plain": 2615, "san": 4274 } diff --git a/tests/harness/corpus-costs/stream-typed-chunk-weakmap.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-weakmap.ts.json new file mode 100644 index 000000000..e5eda35cb --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-weakmap.ts.json @@ -0,0 +1 @@ +{ "plain": 2399, "san": 4694 } diff --git a/tests/harness/corpus-costs/stream-typed-chunk-writev.ts.json b/tests/harness/corpus-costs/stream-typed-chunk-writev.ts.json new file mode 100644 index 000000000..d1ac79034 --- /dev/null +++ b/tests/harness/corpus-costs/stream-typed-chunk-writev.ts.json @@ -0,0 +1 @@ +{ "plain": 2486, "san": 4615 }