diff --git a/packages/compiler/ambient/scriptc-node-fallback.d.ts b/packages/compiler/ambient/scriptc-node-fallback.d.ts index 3885e0fef..4e39d9e42 100644 --- a/packages/compiler/ambient/scriptc-node-fallback.d.ts +++ b/packages/compiler/ambient/scriptc-node-fallback.d.ts @@ -6736,12 +6736,12 @@ declare module "stream" { readonly closed: boolean; readonly destroyed: boolean; readonly errored: Error | null; - on(event: "drain" | "finish" | "close", listener: () => void): void; - on(event: "error", listener: (err: Error) => void): void; + on(event: "drain" | "finish" | "close", listener: () => void): this; + on(event: "error", listener: (err: Error) => void): this; on(event: "pipe" | "unpipe", listener: (src: Readable) => void): this; on(event: string, listener: (...args: any[]) => void): this; - once(event: "drain" | "finish" | "close", listener: () => void): void; - once(event: "error", listener: (err: Error) => void): void; + once(event: "drain" | "finish" | "close", listener: () => void): this; + once(event: "error", listener: (err: Error) => void): this; once(event: "pipe" | "unpipe", listener: (src: Readable) => void): this; once(event: string, listener: (...args: any[]) => void): this; addListener(event: "drain" | "finish" | "close", listener: () => void): this; diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index 4602a2bdd..d38379bf0 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -1106,6 +1106,7 @@ export const LIB_FN_SYMS: Record = { "readable.opFind": "scr_stream_op_find", "readable.opReduce": "scr_stream_op_reduce", "readable.pipe": "scr_stream_pipe", + "readable.pipeGeneral": "scr_stream_pipe_general", "writable.cork": "scr_stream_cork", "writable.uncork": "scr_stream_uncork", "stream.destroyErr": "scr_stream_destroy", @@ -1588,6 +1589,7 @@ export const USES_TIMERS_LIB_FNS = new Set([ "readable.opReduce", "readable.resume", "readable.pipe", + "readable.pipeGeneral", "readable.unpipe", "writable.write", "writable.writeStr", diff --git a/packages/compiler/src/frontend/lowering/lower-stream.ts b/packages/compiler/src/frontend/lowering/lower-stream.ts index 51c86fe39..88ba1f5b8 100644 --- a/packages/compiler/src/frontend/lowering/lower-stream.ts +++ b/packages/compiler/src/frontend/lowering/lower-stream.ts @@ -3371,6 +3371,16 @@ function pipeIntoStdio( }; } +/** The pipe() libCall for a destination of type `dst`. Under the general + * stream representation a Writable slot can hold the converted + * process.stdout / process.stderr, which a pipe never ends (as in Node), so + * the pipe takes readable.pipeGeneral, which checks for them. */ +function generalPipe(lowerer: Lowerer, dst: IrType): "readable.pipe" | "readable.pipeGeneral" { + return lowerer.generalNodeStreams && dst.kind === "object" && dst.className === "%Writable" + ? "readable.pipeGeneral" + : "readable.pipe"; +} + /** `readable.pipe(untyped)`: a checked-dynamic destination unwraps a boxed * native stream at run time (stream.unboxDyn; other values throw the * not-supported Error), and pipe() answers the destination value itself. */ @@ -3394,7 +3404,7 @@ function pipeIntoDyn( kind: "exprStmt", expr: { kind: "libCall", - fn: "readable.pipe", + fn: lowerer.generalNodeStreams ? "readable.pipeGeneral" : "readable.pipe", args: [ { kind: "varRef", localId: source.id, type: receiverValue.type, loc }, { @@ -3691,7 +3701,7 @@ export function lowerStreamMethodCall( const receiver = lowerer.lowerExpr(access.expression); return { kind: "libCall", - fn: "readable.pipe", + fn: generalPipe(lowerer, dst.type), args: [receiver, dst, end], type: dst.type, loc, diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts index 1c390bbe4..11a292a2e 100644 --- a/packages/compiler/src/ir/builtin-effects.ts +++ b/packages/compiler/src/ir/builtin-effects.ts @@ -326,6 +326,7 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "stream.onDyn", "readable.resume", "readable.pipe", + "readable.pipeGeneral", "readable.unpipe", "writable.write", "writable.writeStr", diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 597076cb1..82ecf461d 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -3612,7 +3612,9 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ * returns the below-hwm answer. readable.read: (recv, size — -1 for * absent) → Buffer|null union. pause/resume return recv +1 (`this`); * isPaused answers the flag. readable.pipe: (recv, dst, end) → dst +1, - * fires 'pipe' on dst; readable.unpipe (recv[, dst]) → recv +1. + * fires 'pipe' on dst; readable.pipeGeneral: the same, with `end` off + * when dst is the converted process.stdout / process.stderr; + * readable.unpipe (recv[, dst]) → recv +1. * writable.write/writeStr: (recv, chunk[, cb]) → below-hwm bool (cb * MOVES when present, called after the user write completes). * writable.end: (recv, flags[, chunk][, cb]) → recv +1. cork/uncork are @@ -3810,6 +3812,7 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ "readable.resume": { argTypes: [null], result: VOID }, "readable.isPaused": { argTypes: [null], result: BOOL }, "readable.pipe": { argTypes: [null, null, BOOL], result: VOID }, + "readable.pipeGeneral": { argTypes: [null, null, BOOL], result: VOID }, "readable.unpipe": { argTypes: [null], result: VOID }, "readable.flowing": { argTypes: [null], result: VOID }, "writable.write": { argTypes: [null, BYTES_U8], result: BOOL }, diff --git a/packages/compiler/src/ir/validate.ts b/packages/compiler/src/ir/validate.ts index 63078ae4a..742787d98 100644 --- a/packages/compiler/src/ir/validate.ts +++ b/packages/compiler/src/ir/validate.ts @@ -5729,9 +5729,9 @@ function validateFunction( } return true; } - if (e.fn === "readable.pipe") { + if (e.fn === "readable.pipe" || e.fn === "readable.pipeGeneral") { if (!isStreamObject(e.args[1]?.type) || !typeEquals(e.type, e.args[1]!.type)) { - err(`libCall readable.pipe must return its destination's type`, e.loc); + err(`libCall ${e.fn} must return its destination's type`, e.loc); } return true; } diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json new file mode 100644 index 000000000..baf428dee --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/6468-node-stdio-writable-slot.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/6468-node-stdio-writable-slot.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_stream.c b/packages/runtime/src/scr_stream.c index 1c5e339a0..7c7b2279c 100644 --- a/packages/runtime/src/scr_stream.c +++ b/packages/runtime/src/scr_stream.c @@ -6665,18 +6665,42 @@ ScrStream *scr_stream_child_stdio(void *child, const char *key, size_t key_len) * stream.stdioWritable libCall): a Writable whose _write hands each chunk * to the descriptor through process.stdout.write's synchronous path and * completes at once, so ordering against console output and direct writes - * is unchanged. The closure keeps the fd as given (1 or 2). */ + * is unchanged. The closure keeps the fd as given (1 or 2). Each thread + * keeps one per descriptor, so every conversion of process.stdout answers + * the same stream, as Node's process.stdout is one object: listeners, + * writable state and identity are shared. The cache is released at exit. */ +static SCR_TL ScrStream *scr_stream_stdio_writables[2]; static void scr_stream_stdio_write(ScrClosure *cb, ScrStream *s, ScrBytes *chunk) { int fd = (int)scr_box_get_f64(cb->caps[0]); (void)scr_stdio_write(fd == 2 ? 2 : 1, chunk->data, chunk->len * scr_bytes_elem_size(chunk->elem)); scr_stream_write_done(s, NULL); } +static void scr_stream_stdio_writables_free(void) { + for (int i = 0; i < 2; i++) { + ScrStream *s = scr_stream_stdio_writables[i]; + scr_stream_stdio_writables[i] = NULL; + if (s) scr_stream_release(s); + } +} ScrStream *scr_stream_stdio_writable(double fd) { - ScrClosure *write = scr_closure_new((void *)scr_stream_stdio_write, 1); - write->caps[0] = scr_box_new(SCR_BOX_F64); - scr_box_set_f64(write->caps[0], fd); - return scr_stream_new_writable(-1, true, true, write, scr_stream_stdio_write, - NULL, NULL, NULL, NULL); + int i = fd == 2 ? 1 : 0; + if (scr_stream_stdio_writables[i] == NULL) { + if (scr_stream_stdio_writables[1 - i] == NULL) scr_atexit(scr_stream_stdio_writables_free); + ScrClosure *write = scr_closure_new((void *)scr_stream_stdio_write, 1); + write->caps[0] = scr_box_new(SCR_BOX_F64); + scr_box_set_f64(write->caps[0], i ? 2 : 1); + scr_stream_stdio_writables[i] = scr_stream_new_writable(-1, true, true, write, + scr_stream_stdio_write, NULL, NULL, NULL, NULL); + } + return scr_stream_retain(scr_stream_stdio_writables[i]); +} + +/* readable.pipe() into a Writable slot under the general representation + * (the readable.pipeGeneral libCall): Node never ends process.stdout or + * process.stderr from a pipe, so their converted Writable keeps `end` off. */ +ScrStream *scr_stream_pipe_general(ScrStream *src, ScrStream *dst, bool end) { + if (dst == scr_stream_stdio_writables[0] || dst == scr_stream_stdio_writables[1]) end = false; + return scr_stream_pipe(src, dst, end); } /* A pipeline() stage that arrives as a checked-dynamic value (the diff --git a/packages/runtime/src/scr_stream.h b/packages/runtime/src/scr_stream.h index 63ed0c28a..69da3dc0e 100644 --- a/packages/runtime/src/scr_stream.h +++ b/packages/runtime/src/scr_stream.h @@ -130,6 +130,9 @@ ScrStream *scr_stream_new_writable_dyn(struct ScrDyn *opts); * Writable (fd 1 or 2). */ ScrStream *scr_stream_child_stdio(void *child, const char *key, size_t key_len); ScrStream *scr_stream_stdio_writable(double fd); +/* readable.pipe() under the general representation: a converted + * process.stdout / process.stderr destination is never ended (dst +1). */ +ScrStream *scr_stream_pipe_general(ScrStream *src, ScrStream *dst, bool end); /* A boxed native stream out of a checked-dynamic pipeline stage (+1), or * NULL with `message` thrown as the not-supported Error. */ ScrStream *scr_stream_unbox_dyn(const struct ScrDyn *value, const ScrStr *message); diff --git a/tests/corpus/6468-node-stdio-writable-slot.ts b/tests/corpus/6468-node-stdio-writable-slot.ts new file mode 100644 index 000000000..a326344d1 --- /dev/null +++ b/tests/corpus/6468-node-stdio-writable-slot.ts @@ -0,0 +1,49 @@ +// process.stdout and process.stderr in NodeJS.WritableStream slots of a +// program that also passes other streams there: every slot holds the same +// stream object, a pipe through such a slot leaves the process stream +// open, and Writable's listener methods return the stream for chaining. +import { Readable, Writable } from "node:stream"; + +class Sink extends Writable { + _write(_chunk: Buffer, _encoding: string, callback: (error?: Error | null) => void): void { + callback(); + } +} + +function same(a: NodeJS.WritableStream, b: NodeJS.WritableStream): boolean { + return a === b; +} + +function pipeInto(dst: NodeJS.WritableStream, parts: string[]): Promise { + return new Promise((resolve) => { + const src = Readable.from(parts); + src.on("end", () => resolve()); + src.pipe(dst); + }); +} + +function write(dst: NodeJS.WritableStream, text: string): boolean { + return dst.write(text); +} + +async function main(): Promise { + console.log("stdout same", same(process.stdout, process.stdout)); + console.log("stderr same", same(process.stderr, process.stderr)); + console.log("stdout and stderr", same(process.stdout, process.stderr)); + console.log("sink and stdout", same(new Sink(), process.stdout)); + + await pipeInto(process.stdout, ["piped ", "through ", "a slot\n"]); + // The pipe did not end process.stdout: writes through a slot still work. + console.log("write after pipe", write(process.stdout, "still open\n")); + await pipeInto(process.stderr, ["piped to stderr\n"]); + write(process.stderr, "stderr still open\n"); + + // Writable's listener methods return the stream. + const sink = new Sink(); + await new Promise((resolve) => { + sink.on("finish", () => resolve()).end("done"); + }); + console.log("sink finished", sink.writableFinished); +} + +void main(); diff --git a/tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json b/tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json new file mode 100644 index 000000000..3fafdfc4b --- /dev/null +++ b/tests/harness/corpus-costs/6468-node-stdio-writable-slot.ts.json @@ -0,0 +1 @@ +{ "plain": 3000, "san": 3500 }