From 21b88b8d669cd8501d692a929ede1fde4b3cfeda Mon Sep 17 00:00:00 2001 From: Chris Tate Date: Fri, 16 Oct 2026 04:27:08 -0500 Subject: [PATCH] Add MessageChannel, receiveMessageOnPort and MessagePort transfer new MessageChannel() from node:worker_threads makes two entangled MessagePorts. They share the handle model of parentPort: postMessage with the existing message clone, 'message' and 'close' events, on/once, start, close, ref, unref and hasRef. A port's 'close' reaches the other port after its queued messages. receiveMessageOnPort(port) answers the next queued message as { message }, or undefined. A channel is a thread-neutral record (scr_mailbox.c): each end has its queue and the inbox of the thread that owns its port. A port listed in a transfer list, given as postMessage's array or { transfer }, or as the Worker constructor's transferList, travels with the message, and the decoding thread adopts it. The sending port becomes unusable. A port in a message without that listing throws DataCloneError, and so does a port listing itself. Port events follow Node's loop phases. Each port's queued messages arrive as one batch, ports in creation order, before setImmediate callbacks, with nextTicks and microtasks after each message (a new loop hook keeps the pass going until the batch is delivered). A port's own 'close' runs in the following close phase, the latest close() first. The other port's 'close' arrives on a later poll, in the order the notices were sent, and only while the loop keeps running. --- .../limitations/native-worker-threads.mdx | 4 +- .../node-v24-compatibility-meta.json | 2 +- .../src/generated/node-v24-compatibility.json | 16 +- .../generated/node-v24-backlog.json | 36 +- .../generated/node-v24-internal.json | 25 +- internal/compatibility/static-support.json | 1 + .../ambient/scriptc-node-fallback.d.ts | 7 + .../compiler/src/backend/llvm/lib-shared.ts | 4 + .../src/frontend/lowering/builtin-calls.ts | 4 + .../src/frontend/lowering/builtins/workers.ts | 68 ++++ .../src/frontend/lowering/lower-classes.ts | 4 +- packages/compiler/src/frontend/type-mapper.ts | 38 +- packages/compiler/src/ir/builtin-effects.ts | 2 + .../compiler/src/ir/builtin-signatures.ts | 4 + packages/compiler/src/library/ir-policy.ts | 2 + .../5793-worker-message-channel.ts.json | 6 + ...94-worker-message-port-close-order.ts.json | 6 + .../5795-worker-message-port-batches.ts.json | 6 + .../5796-worker-message-channel.cjs.json | 6 + .../5797-worker-message-port-transfer.ts.json | 6 + packages/runtime/src/scr_async.c | 6 + packages/runtime/src/scr_bytes.h | 4 + packages/runtime/src/scr_mailbox.c | 189 ++++++++++ packages/runtime/src/scr_mailbox.h | 21 ++ packages/runtime/src/scr_message.c | 67 +++- packages/runtime/src/scr_message.h | 18 + packages/runtime/src/scr_worker.h | 3 + packages/runtime/src/scr_worker_events.c | 331 +++++++++++++++++- tests/corpus/5793-worker-message-channel.ts | 51 +++ .../5794-worker-message-port-close-order.ts | 14 + .../5795-worker-message-port-batches.ts | 16 + tests/corpus/5796-worker-message-channel.cjs | 45 +++ .../5797-worker-message-port-transfer.ts | 50 +++ 33 files changed, 1016 insertions(+), 46 deletions(-) create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5793-worker-message-channel.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5794-worker-message-port-close-order.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5795-worker-message-port-batches.ts.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5796-worker-message-channel.cjs.json create mode 100644 packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5797-worker-message-port-transfer.ts.json create mode 100644 tests/corpus/5793-worker-message-channel.ts create mode 100644 tests/corpus/5794-worker-message-port-close-order.ts create mode 100644 tests/corpus/5795-worker-message-port-batches.ts create mode 100644 tests/corpus/5796-worker-message-channel.cjs create mode 100644 tests/corpus/5797-worker-message-port-transfer.ts diff --git a/docs/content/limitations/native-worker-threads.mdx b/docs/content/limitations/native-worker-threads.mdx index 061d844f2e..0527c24b02 100644 --- a/docs/content/limitations/native-worker-threads.mdx +++ b/docs/content/limitations/native-worker-threads.mdx @@ -1,6 +1,6 @@ -`node:worker_threads` runs workers as native threads. Worker entry points must resolve at build time: use a source-relative `new URL("./worker.ts", import.meta.url)`, the current module URL, an absolute path, or CommonJS `__filename`. Each worker initializes its own module state, event loop, environment snapshot, and reference-counted heap. Nested workers inherit their parent's environment snapshot. Supported constructor options are `workerData` and `argv`; runtime paths, `eval`, loaders, custom stdio, shared environments, resource limits, and transfer lists are unsupported. Worker argument strings containing null bytes are also unsupported. +`node:worker_threads` runs workers as native threads. Worker entry points must resolve at build time: use a source-relative `new URL("./worker.ts", import.meta.url)`, the current module URL, an absolute path, or CommonJS `__filename`. Each worker initializes its own module state, event loop, environment snapshot, and reference-counted heap. Nested workers inherit their parent's environment snapshot. Supported constructor options are `workerData`, `argv`, and a `transferList` of MessagePorts; runtime paths, `eval`, loaders, custom stdio, shared environments, and resource limits are unsupported. Worker argument strings containing null bytes are also unsupported. -Workers support `online`, `message`, `error`, and `exit` events, `postMessage`, `terminate`, `ref`, and `unref`. The parent port supports message listeners, `postMessage`, `close`, `start`, `ref`, `unref`, and `hasRef`. Messages clone primitives, plain object graphs, arrays, errors, ArrayBuffers, typed arrays, and DataViews, preserving cycles and aliases. Program class instances arrive as plain objects with their own data properties, without their prototype, methods, accessors, or private fields, as in Node. Functions, proxies, other native handles, and nonempty transfer lists are refused. SharedArrayBuffers share their backing storage while retaining separate wrappers in each thread. Graphs made immutable with `@scriptc/threads` are passed by reference instead of cloned; see [Sharing Data Between Workers](/docs/threads). +Workers support `online`, `message`, `error`, and `exit` events, `postMessage`, `terminate`, `ref`, and `unref`. `new MessageChannel()` from `node:worker_threads` makes two entangled MessagePorts, in one thread or across threads once a port is transferred, with `receiveMessageOnPort`. Port messages arrive in Node's order: each port's queued messages as one batch before `setImmediate` callbacks, a port's own `'close'` after the poll phase, and the other port's `'close'` after its queued messages while the loop keeps running. A port's `onmessage` property, `addEventListener`, `receiveMessageOnPort(parentPort)`, and `BroadcastChannel` are unsupported. The parent port supports message listeners, `postMessage`, `close`, `start`, `ref`, `unref`, and `hasRef`. Messages clone primitives, plain object graphs, arrays, errors, ArrayBuffers, typed arrays, and DataViews, preserving cycles and aliases. Program class instances arrive as plain objects with their own data properties, without their prototype, methods, accessors, or private fields, as in Node. A MessagePort travels with a message when the message's transfer list (an array or `{ transfer }`) names it, and the sending port is unusable afterwards; a port in a message without that listing throws `DataCloneError`. Other transferable values, functions, proxies, and other native handles are refused. SharedArrayBuffers share their backing storage while retaining separate wrappers in each thread. Graphs made immutable with `@scriptc/threads` are passed by reference instead of cloned; see [Sharing Data Between Workers](/docs/threads). SharedArrayBuffer storage is fixed-length. Atomics supports integer `load`, `store`, `exchange`, `compareExchange`, `add`, `sub`, `and`, `or`, and `xor` over 8-, 16-, and 32-bit integer arrays. `wait` and `notify` use Int32Array views; BigInt atomics, `waitAsync`, and growable buffers are unsupported. Atomics operations run as sequentially consistent atomic instructions and take no lock. Other reads and writes of shared storage access each element atomically; a copy, fill, or conversion that runs while another thread writes the same storage can observe some of those writes, as in Node. `terminate` unwinds at runtime checkpoints, including loops and parked atomic waits, without running JavaScript catch or finally blocks. Blocking native calls finish before the next checkpoint. diff --git a/docs/src/generated/node-v24-compatibility-meta.json b/docs/src/generated/node-v24-compatibility-meta.json index 7f49aebbb4..984fe2b0bc 100644 --- a/docs/src/generated/node-v24-compatibility-meta.json +++ b/docs/src/generated/node-v24-compatibility-meta.json @@ -2,6 +2,6 @@ "schemaVersion": 3, "nodeVersion": "24.22.0", "nodeCommit": "f14cd3aa7bf9ca8df2cc5b8bb14beb0f335b1d93", - "artifactVersion": "aaa7e74233dc5f826dcd", + "artifactVersion": "a2be648ccaabb8375f04", "rowCount": 3880 } diff --git a/docs/src/generated/node-v24-compatibility.json b/docs/src/generated/node-v24-compatibility.json index 6ab3bd3019..95784d77a3 100644 --- a/docs/src/generated/node-v24-compatibility.json +++ b/docs/src/generated/node-v24-compatibility.json @@ -884,9 +884,9 @@ "apiEntries": 66, "static": { "supported": 0, - "partial": 24, + "partial": 26, "refused": 0, - "not-implemented": 42, + "not-implemented": 40, "by-design": 0, "unreviewed": 0, "not-applicable": 0 @@ -83081,9 +83081,9 @@ "inherited": true }, "static": { - "status": "not-implemented", - "detail": "No static compiler lowering is registered for this API yet.", - "verification": "registry-gap" + "status": "partial", + "detail": "Implemented for the documented scriptc module-loader subset.", + "verification": "test-backed" } }, { @@ -84024,9 +84024,9 @@ "inherited": true }, "static": { - "status": "not-implemented", - "detail": "No static compiler lowering is registered for this API yet.", - "verification": "registry-gap" + "status": "partial", + "detail": "Implemented for the documented scriptc module-loader subset.", + "verification": "test-backed" } }, { diff --git a/internal/compatibility/generated/node-v24-backlog.json b/internal/compatibility/generated/node-v24-backlog.json index fdc2805d01..4e3d55d2ce 100644 --- a/internal/compatibility/generated/node-v24-backlog.json +++ b/internal/compatibility/generated/node-v24-backlog.json @@ -7,8 +7,8 @@ "tierItemCount": 3472, "tiers": { "static": { - "audit-partial": 1035, - "verify-gap": 2310, + "audit-partial": 1037, + "verify-gap": 2308, "replace-refusal": 40, "classify": 2, "implement": 85 @@ -83869,12 +83869,16 @@ "priority": "high", "tiers": { "static": { - "status": "not-implemented", - "action": "verify-gap", - "verification": "registry-gap", - "confidence": "low", - "source": "compiler-unmatched:worker_threads", - "tests": [] + "status": "partial", + "action": "audit-partial", + "verification": "test-backed", + "confidence": "high", + "source": "compiler-feature:worker_threads.MessageChannel", + "tests": [ + "tests/corpus/5793-worker-message-channel.ts", + "tests/corpus/5796-worker-message-channel.cjs", + "tests/corpus/5797-worker-message-port-transfer.ts" + ] } } }, @@ -84958,12 +84962,16 @@ "priority": "high", "tiers": { "static": { - "status": "not-implemented", - "action": "verify-gap", - "verification": "registry-gap", - "confidence": "low", - "source": "compiler-unmatched:worker_threads", - "tests": [] + "status": "partial", + "action": "audit-partial", + "verification": "test-backed", + "confidence": "high", + "source": "compiler-feature:worker_threads.worker_threads.receiveMessageOnPort", + "tests": [ + "tests/corpus/5793-worker-message-channel.ts", + "tests/corpus/5796-worker-message-channel.cjs", + "tests/corpus/5797-worker-message-port-transfer.ts" + ] } } }, diff --git a/internal/compatibility/generated/node-v24-internal.json b/internal/compatibility/generated/node-v24-internal.json index a56cf56717..23d55260bd 100644 --- a/internal/compatibility/generated/node-v24-internal.json +++ b/internal/compatibility/generated/node-v24-internal.json @@ -941,9 +941,9 @@ "apiEntries": 66, "static": { "supported": 0, - "partial": 24, + "partial": 26, "refused": 0, - "not-implemented": 42, + "not-implemented": 40, "by-design": 0, "unreviewed": 0, "not-applicable": 0 @@ -112380,6 +112380,9 @@ "status": "partial", "evidence": "derived:descendants", "tests": [ + "tests/corpus/5793-worker-message-channel.ts", + "tests/corpus/5796-worker-message-channel.cjs", + "tests/corpus/5797-worker-message-port-transfer.ts", "tests/corpus/native-worker-clone-errors.ts", "tests/corpus/native-worker-commonjs.cjs", "tests/corpus/native-worker-delayed-messages.ts", @@ -112648,8 +112651,13 @@ "inherited": true }, "static": { - "status": "not-implemented", - "evidence": "compiler-unmatched:worker_threads" + "status": "partial", + "evidence": "compiler-feature:worker_threads.MessageChannel", + "tests": [ + "tests/corpus/5793-worker-message-channel.ts", + "tests/corpus/5796-worker-message-channel.cjs", + "tests/corpus/5797-worker-message-port-transfer.ts" + ] }, "anchorSource": "exact" }, @@ -113720,8 +113728,13 @@ "inherited": true }, "static": { - "status": "not-implemented", - "evidence": "compiler-unmatched:worker_threads" + "status": "partial", + "evidence": "compiler-feature:worker_threads.worker_threads.receiveMessageOnPort", + "tests": [ + "tests/corpus/5793-worker-message-channel.ts", + "tests/corpus/5796-worker-message-channel.cjs", + "tests/corpus/5797-worker-message-port-transfer.ts" + ] }, "anchorSource": "exact" }, diff --git a/internal/compatibility/static-support.json b/internal/compatibility/static-support.json index a0e0e3a7c5..6af5b6b9c0 100644 --- a/internal/compatibility/static-support.json +++ b/internal/compatibility/static-support.json @@ -199,6 +199,7 @@ {"chapter": "worker_threads", "symbols": ["port.close", "port.start", "port.ref", "port.unref", "port.hasRef"], "status": "partial", "evidence": ["tests/corpus/native-worker-port-lifetime.ts"]}, {"chapter": "worker_threads", "parent": "Class: `Worker`", "signatures": ["Event: `'online'`", "Event: `'message'`", "Event: `'error'`", "Event: `'exit'`"], "status": "partial", "evidence": ["tests/corpus/native-worker-references.ts", "tests/corpus/native-worker-error.ts", "tests/corpus/native-worker-exit-listeners.ts"]}, {"chapter": "worker_threads", "parent": "Class: `MessagePort`", "signatures": ["Event: `'message'`", "Event: `'close'`"], "status": "partial", "evidence": ["tests/corpus/native-worker-port-lifetime.ts", "tests/corpus/native-worker-messages.ts"]}, + {"chapter": "worker_threads", "symbols": ["MessageChannel", "worker_threads.receiveMessageOnPort"], "status": "partial", "evidence": ["tests/corpus/5793-worker-message-channel.ts", "tests/corpus/5796-worker-message-channel.cjs", "tests/corpus/5797-worker-message-port-transfer.ts"]}, {"chapter": "util", "symbols": ["util.types.isAnyArrayBuffer", "util.types.isArrayBuffer", "util.types.isArrayBufferView", "util.types.isDataView", "util.types.isTypedArray", "util.types.isUint8Array", "util.types.isUint8ClampedArray", "util.types.isUint16Array", "util.types.isUint32Array", "util.types.isInt8Array", "util.types.isInt16Array", "util.types.isInt32Array", "util.types.isFloat32Array", "util.types.isFloat64Array", "util.types.isMap", "util.types.isSet", "util.types.isWeakMap", "util.types.isWeakSet", "util.types.isRegExp", "util.types.isNativeError", "util.types.isPromise", "util.types.isProxy", "util.types.isAsyncFunction", "util.types.isGeneratorFunction", "util.types.isGeneratorObject"], "status": "partial", "evidence": ["tests/corpus/node24-util-types.ts", "tests/corpus/node24-util-types.js", "tests/corpus/node24-util-types.cjs", "tests/fixtures/node-types/util-types.ts"]}, { "chapter": "util", "symbols": ["util.styleText"], "status": "partial", "evidence": ["tests/corpus/node24-style-text.ts", "tests/corpus/node24-style-text-validation.js", "tests/corpus/node24-style-text-environment.js", "tests/corpus/node24-style-text.cjs", "tests/fixtures/node-types/style-text.ts"] }, { "chapter": "util", "symbols": ["util.isDeepStrictEqual"], "status": "partial", "evidence": ["tests/corpus/node24-deep-equality.ts", "tests/corpus/node24-deep-equality.cjs", "tests/corpus/node24-deep-equality-properties.js", "tests/fixtures/node-types/deep-equality.ts", "packages/compiler/test/deep-equality.test.ts"] }, diff --git a/packages/compiler/ambient/scriptc-node-fallback.d.ts b/packages/compiler/ambient/scriptc-node-fallback.d.ts index afa9b0d31e..7395bb6ca3 100644 --- a/packages/compiler/ambient/scriptc-node-fallback.d.ts +++ b/packages/compiler/ambient/scriptc-node-fallback.d.ts @@ -5459,6 +5459,13 @@ declare module "worker_threads" { removeAllListeners(event?: string): this; listenerCount(event: string, listener?: (...args: any[]) => void): number; } + /* Two entangled ports; a port moves to another thread through a + * postMessage or workerData transferList. */ + export class MessageChannel { + readonly port1: MessagePort; + readonly port2: MessagePort; + } + export function receiveMessageOnPort(port: MessagePort): { message: any } | undefined; export const parentPort: MessagePort | null; export const workerData: any; } diff --git a/packages/compiler/src/backend/llvm/lib-shared.ts b/packages/compiler/src/backend/llvm/lib-shared.ts index 192c20aa8a..9bea776264 100644 --- a/packages/compiler/src/backend/llvm/lib-shared.ts +++ b/packages/compiler/src/backend/llvm/lib-shared.ts @@ -913,6 +913,8 @@ export const LIB_FN_SYMS: Record = { "worker.root": "scr_worker_root", "worker.data": "scr_worker_data", "worker.parentPort": "scr_worker_parent_port", + "worker.channel": "scr_worker_message_channel", + "worker.receiveMessageOnPort": "scr_worker_receive_message_on_port", "process.stdinDestroy": "scr_process_stdin_destroy", "process.stdinSetRawMode": "scr_process_stdin_set_raw_mode", // node:events EventEmitter (scr_events_emitter.c): receivers borrowed, @@ -1327,6 +1329,8 @@ export const USES_TIMERS_LIB_FNS = new Set([ "util.debuglog", "worker.new", "worker.parentPort", + "worker.channel", + "worker.receiveMessageOnPort", "process.stdio", "fetch.start", "fetch.input", diff --git a/packages/compiler/src/frontend/lowering/builtin-calls.ts b/packages/compiler/src/frontend/lowering/builtin-calls.ts index d9aa998e1c..1bc55b8843 100644 --- a/packages/compiler/src/frontend/lowering/builtin-calls.ts +++ b/packages/compiler/src/frontend/lowering/builtin-calls.ts @@ -1,3 +1,4 @@ +import { lowerReceiveMessageOnPort } from "./builtins/workers.js"; import * as ts from "../ts7/adapter.js"; import { dynUndefinedExpr } from "../../ir/build.js"; import { DYN, type IrExpr, type SrcLoc } from "../../ir/ir.js"; @@ -86,6 +87,9 @@ export function lowerBuiltinCall( if (timersInterval) return timersInterval; const nodeModuleServed = lowerNodeModuleCall(lowerer, expr, bi, loc); if (nodeModuleServed) return nodeModuleServed; + if (bi.module === "worker_threads" && bi.member === "receiveMessageOnPort") { + return lowerReceiveMessageOnPort(lowerer, expr); + } const builtinFn = builtinModuleFnOf(lowerer, bi.module, bi.member); if (!builtinFn) { lowerer.noLowering( diff --git a/packages/compiler/src/frontend/lowering/builtins/workers.ts b/packages/compiler/src/frontend/lowering/builtins/workers.ts index e2f22bfccc..3144f208f1 100644 --- a/packages/compiler/src/frontend/lowering/builtins/workers.ts +++ b/packages/compiler/src/frontend/lowering/builtins/workers.ts @@ -98,3 +98,71 @@ export function lowerWorkerNew(lowerer: Lowerer, expression: ts.NewExpression): loc, }; } + +/** True when `node`'s type is worker_threads' `name` class (also reached + * through its global). */ +function isWorkerThreadsClass(lowerer: Lowerer, node: ts.Expression, name: string): boolean { + const t = lowerer.checker.getTypeAtLocation(node); + const sym = t.getSymbol(); + if (sym?.name !== name) return false; + return lowerer.checker.declarationsOf(sym).some((d) => { + if (!ts.isClassDeclaration(d) && !ts.isInterfaceDeclaration(d)) return false; + if (!lowerer.isStdlibFile(d.getSourceFile())) return false; + let up: ts.Node | undefined = d.parent; + while (up) { + if (ts.isModuleDeclaration(up) && ts.isStringLiteral(up.name)) + return up.name.text === "worker_threads" || up.name.text === "node:worker_threads"; + up = up.parent; + } + return false; + }); +} + +/** new MessageChannel(): { port1, port2 }, two entangled MessagePorts. */ +export function lowerMessageChannelNew( + lowerer: Lowerer, + expression: ts.NewExpression, +): IrExpr | null { + if (!isWorkerThreadsClass(lowerer, expression, "MessageChannel")) return null; + if ((expression.arguments ?? []).length > 0) { + lowerer.noLowering( + "MessageChannel constructor arguments", + expression, + "use new MessageChannel()", + ); + } + if (lowerer.targetPlatform === "wasi") { + lowerer.noLowering( + "MessageChannel in this target", + expression, + "use a native static executable", + ); + } + lowerer.usesWorkers = true; + return { kind: "libCall", fn: "worker.channel", args: [], type: DYN, loc: locOf(expression) }; +} + +/** receiveMessageOnPort(port): { message } or undefined. */ +export function lowerReceiveMessageOnPort(lowerer: Lowerer, call: ts.CallExpression): IrExpr { + const args = call.arguments; + if (args.length > 1 || args.some(ts.isSpreadElement)) { + lowerer.noLowering(`receiveMessageOnPort with ${args.length} arguments`, call); + } + lowerer.usesWorkers = true; + const loc = locOf(call); + const port: IrExpr = args[0] + ? lowerer.lowerExprExpecting(args[0], DYN) + : { + kind: "dynFrom", + value: { kind: "unitLit", unit: "undefined", type: UNDEFINED_T, loc }, + type: DYN, + loc, + }; + return { + kind: "libCall", + fn: "worker.receiveMessageOnPort", + args: [port], + type: DYN, + loc, + }; +} diff --git a/packages/compiler/src/frontend/lowering/lower-classes.ts b/packages/compiler/src/frontend/lowering/lower-classes.ts index 6a81f9a72e..d518c68225 100644 --- a/packages/compiler/src/frontend/lowering/lower-classes.ts +++ b/packages/compiler/src/frontend/lowering/lower-classes.ts @@ -76,7 +76,7 @@ import { bufEncoding } from "./containers/bytes.js"; import { isSafeToRepeat } from "./expressions/evaluation-safety.js"; import { lowerSearchParamsNew } from "./builtins/url.js"; import { lowerTextCodecNew } from "./builtins/text-codecs.js"; -import { lowerWorkerNew } from "./builtins/workers.js"; +import { lowerMessageChannelNew, lowerWorkerNew } from "./builtins/workers.js"; import { unsupportedPackageValueDiag, unsupportedDiag, @@ -8138,6 +8138,8 @@ export function lowerNew(lowerer: Lowerer, expr: ts.NewExpression): IrExpr { const loc = locOf(expr); const worker = lowerWorkerNew(lowerer, expr); if (worker) return worker; + const channel = lowerMessageChannelNew(lowerer, expr); + if (channel) return channel; let intlConstructor = expr.expression; if (ts.isIdentifier(intlConstructor)) { const symbol = lowerer.resolveValueSymbol(intlConstructor); diff --git a/packages/compiler/src/frontend/type-mapper.ts b/packages/compiler/src/frontend/type-mapper.ts index 767e0abff4..0e757efca1 100644 --- a/packages/compiler/src/frontend/type-mapper.ts +++ b/packages/compiler/src/frontend/type-mapper.ts @@ -1220,7 +1220,9 @@ function isWorkerHandleSymbol( ctx: Pick, ): boolean { return ( - (symbol?.name === "Worker" || symbol?.name === "MessagePort") && + (symbol?.name === "Worker" || + symbol?.name === "MessagePort" || + symbol?.name === "MessageChannel") && ctx.checker .declarationsOf(symbol) .some( @@ -1232,6 +1234,37 @@ function isWorkerHandleSymbol( ); } +/** The `{ message: any }` object type (or its union with undefined) that + * worker_threads declares as receiveMessageOnPort's result. */ +function isReceiveMessageResult( + type: ts.Type, + ctx: Pick, +): boolean { + const members = type.isUnionType() ? ts.constituentTypes(type) : [type]; + let found = false; + for (const member of members) { + if (member.flags & ts.TypeFlags.Undefined) continue; + const decl = member.getSymbol() + ? ctx.checker.declarationsOf(member.getSymbol()!)[0] + : undefined; + let fn: ts.Node | undefined = decl?.parent; + while (fn && ts.isUnionTypeNode(fn)) fn = fn.parent; + if ( + !decl || + !ts.isTypeLiteralNode(decl) || + !fn || + !ts.isFunctionDeclaration(fn) || + fn.name?.text !== "receiveMessageOnPort" || + !ctx.isStdlibFile(decl.getSourceFile()) || + !isDeclaredInAmbientModule(fn, "worker_threads") + ) { + return false; + } + found = true; + } + return found; +} + /** True for a type built only from worker handles, null/undefined, and * arrays of those: values that keep the live DYN representation. */ export function isWorkerHandleType( @@ -1316,6 +1349,9 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null { const moduleId = ctx.moduleNamespaceId?.(type) ?? null; if (moduleId !== null) return { kind: "moduleNs", moduleId }; } + // worker_threads.receiveMessageOnPort's `{ message: any } | undefined`: + // the runtime's { message } object, a checked-dynamic value. + if (isReceiveMessageResult(type, ctx)) return DYN; // @types/node spells child_process.ExecFileException as an intersection // type alias over two Error-derived metadata interfaces. The value is // still one ordinary Error; recognize the owned alias before generic diff --git a/packages/compiler/src/ir/builtin-effects.ts b/packages/compiler/src/ir/builtin-effects.ts index 9cb5db6839..891d825b44 100644 --- a/packages/compiler/src/ir/builtin-effects.ts +++ b/packages/compiler/src/ir/builtin-effects.ts @@ -18,6 +18,8 @@ export const MAY_THROW_LIB_FNS: ReadonlySet = new Set([ "worker.new", "worker.data", "worker.parentPort", + "worker.channel", + "worker.receiveMessageOnPort", "process.exit", "text.encodeInto", "buffer.isAscii", diff --git a/packages/compiler/src/ir/builtin-signatures.ts b/packages/compiler/src/ir/builtin-signatures.ts index 80ccf9a3e1..611af659de 100644 --- a/packages/compiler/src/ir/builtin-signatures.ts +++ b/packages/compiler/src/ir/builtin-signatures.ts @@ -1096,6 +1096,10 @@ export const LIB_FN_SIGS = defineLibFnSignatures({ "worker.root": { argTypes: [], result: F64 }, "worker.data": { argTypes: [], result: DYN }, "worker.parentPort": { argTypes: [], result: DYN }, + /** new MessageChannel() ({ port1, port2 }) and receiveMessageOnPort(port) + * (scr_worker_events.c; the latter may throw). */ + "worker.channel": { argTypes: [], result: DYN }, + "worker.receiveMessageOnPort": { argTypes: [DYN], result: DYN }, /** child_process.spawnSync (scr_child.c): posix_spawn + waitpid + piped * utf8 capture — cmd borrowed, args one borrowed string[] (the frontend * completes an omitted list to an empty literal), result an owned (+1) diff --git a/packages/compiler/src/library/ir-policy.ts b/packages/compiler/src/library/ir-policy.ts index 9ca7bdca4e..c13c7694b7 100644 --- a/packages/compiler/src/library/ir-policy.ts +++ b/packages/compiler/src/library/ir-policy.ts @@ -61,6 +61,8 @@ const LIB_MODE_REFUSED_PREFIXES: readonly [string, string][] = [ ["urj.", "unhandled-rejection tracking"], ["dc.", "the diagnostics_channel surface"], ["worker.new", "native worker threads"], + ["worker.channel", "native worker threads"], + ["worker.receiveMessageOnPort", "native worker threads"], ]; /** Value/type kinds whose mere presence means an excluded unit's code (or diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5793-worker-message-channel.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5793-worker-message-channel.ts.json new file mode 100644 index 0000000000..c00734b297 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5793-worker-message-channel.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5793-worker-message-channel.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5794-worker-message-port-close-order.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5794-worker-message-port-close-order.ts.json new file mode 100644 index 0000000000..7922436a14 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5794-worker-message-port-close-order.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5794-worker-message-port-close-order.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5795-worker-message-port-batches.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5795-worker-message-port-batches.ts.json new file mode 100644 index 0000000000..1281e40ebf --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5795-worker-message-port-batches.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5795-worker-message-port-batches.ts" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5796-worker-message-channel.cjs.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5796-worker-message-channel.cjs.json new file mode 100644 index 0000000000..a039905b46 --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5796-worker-message-channel.cjs.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5796-worker-message-channel.cjs" + ], + "diags": [] +} diff --git a/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5797-worker-message-port-transfer.ts.json b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5797-worker-message-port-transfer.ts.json new file mode 100644 index 0000000000..600a41db5c --- /dev/null +++ b/packages/compiler/test/ts7/baselines/order-parity/tests/corpus/5797-worker-message-port-transfer.ts.json @@ -0,0 +1,6 @@ +{ + "order": [ + "/tests/corpus/5797-worker-message-port-transfer.ts" + ], + "diags": [] +} diff --git a/packages/runtime/src/scr_async.c b/packages/runtime/src/scr_async.c index 0665828f4e..7630b2f724 100644 --- a/packages/runtime/src/scr_async.c +++ b/packages/runtime/src/scr_async.c @@ -3719,6 +3719,7 @@ void scr_loop_set_stream(bool (*pending)(void), void (*dispatch)(void)) { #ifdef SCR_WORKERS static SCR_TL bool (*scr_workers_pending_fn)(void); static SCR_TL void (*scr_workers_dispatch_fn)(void); +static SCR_TL bool (*scr_workers_more_fn)(void); static SCR_TL int (*scr_workers_pollfd_fn)(void); static SCR_TL bool (*scr_workers_ready_fn)(void); static SCR_TL void (*scr_workers_wait_fn)(double); @@ -3731,6 +3732,8 @@ void scr_loop_set_workers(bool (*pending)(void), bool (*ready)(void), void (*dis scr_workers_ready_fn = ready; scr_workers_wait_fn = wait; } + +void scr_loop_set_workers_more(bool (*more)(void)) { scr_workers_more_fn = more; } #endif static bool scr_workers_pending(void) { @@ -3889,6 +3892,9 @@ static bool scr_loop_run_pass(ScrPromise *top_level, bool first_checkpoint) { scr_workers_dispatch(); if (scr_exc_pending()) return false; if (scr_ready_len > 0 || scr_nt_head != NULL) continue; +#ifdef SCR_WORKERS + if (scr_workers_more_fn != NULL && scr_workers_more_fn()) continue; +#endif /* Event dispatch (scr_events.c, when linked): watched signals * delivered since the last turn fire their listeners now (macrotasks, * like timers), then stdin — while a consumer exists, probe fd 0 and diff --git a/packages/runtime/src/scr_bytes.h b/packages/runtime/src/scr_bytes.h index 5050e3a1b3..6a43a68d47 100644 --- a/packages/runtime/src/scr_bytes.h +++ b/packages/runtime/src/scr_bytes.h @@ -90,6 +90,10 @@ void scr_bytes_require_unshared(const ScrBytes *); ScrDyn *scr_shared_array_buffer_new(ScrDyn *); ScrDyn *scr_worker_data(void); ScrDyn *scr_worker_parent_port(void); +/* new MessageChannel() ({ port1, port2 }) and receiveMessageOnPort(port) + * (scr_worker_events.c). */ +ScrDyn *scr_worker_message_channel(void); +ScrDyn *scr_worker_receive_message_on_port(ScrDyn *port); ScrDyn *scr_worker_new(double root, ScrStr *filename, ScrDyn *options); bool scr_shared_array_buffer_is(const ScrDyn *); bool scr_buffer_storage_is(const ScrDyn *); diff --git a/packages/runtime/src/scr_mailbox.c b/packages/runtime/src/scr_mailbox.c index 0bc9abf04a..89cb6ce830 100644 --- a/packages/runtime/src/scr_mailbox.c +++ b/packages/runtime/src/scr_mailbox.c @@ -231,3 +231,192 @@ int scr_mailbox_pollfd(const ScrMailbox *mailbox) { return mailbox->read_fd; #endif } + +/* ── MessageChannel ──────────────────────────────────────────────────── */ + +typedef struct { + ScrMailEvent *head; /* incoming events for this end, in order */ + ScrMailEvent **tail; + size_t count; + ScrMailbox *owner; /* the owning thread's inbox (+1), or NULL in transit */ + bool closed; /* this end called close() */ + bool peer_closed; /* the other end's close reached this end */ + uint64_t peer_closed_seq; /* when: notices are delivered in this order */ +} ScrChannelEnd; + +static atomic_uint_fast64_t scr_channel_notice_seq; + +struct ScrChannel { + atomic_size_t references; + ScrChannelEnd end[2]; +#ifdef _WIN32 + SRWLOCK lock; +#else + pthread_mutex_t lock; +#endif +}; + +static void scr_channel_lock(ScrChannel *channel) { +#ifdef _WIN32 + AcquireSRWLockExclusive(&channel->lock); +#else + (void)pthread_mutex_lock(&channel->lock); +#endif +} + +static void scr_channel_unlock(ScrChannel *channel) { +#ifdef _WIN32 + ReleaseSRWLockExclusive(&channel->lock); +#else + (void)pthread_mutex_unlock(&channel->lock); +#endif +} + +ScrChannel *scr_channel_new(void) { + ScrChannel *channel = calloc(1, sizeof(*channel)); + if (!channel) scr_trap("scriptc: out of memory\n"); + atomic_init(&channel->references, 1); + for (int side = 0; side < 2; side++) channel->end[side].tail = &channel->end[side].head; +#ifdef _WIN32 + InitializeSRWLock(&channel->lock); +#else + if (pthread_mutex_init(&channel->lock, NULL) != 0) scr_trap("scriptc: out of memory\n"); +#endif + return channel; +} + +ScrChannel *scr_channel_retain(ScrChannel *channel) { + atomic_fetch_add_explicit(&channel->references, 1, memory_order_relaxed); + return channel; +} + +void scr_channel_release(ScrChannel *channel) { + if (!channel || atomic_fetch_sub_explicit(&channel->references, 1, memory_order_acq_rel) != 1) + return; + for (int side = 0; side < 2; side++) { + ScrChannelEnd *end = &channel->end[side]; + while (end->head) { + ScrMailEvent *event = end->head; + end->head = event->next; + scr_mail_event_free(event); + } + scr_mailbox_release(end->owner); + } +#ifndef _WIN32 + pthread_mutex_destroy(&channel->lock); +#endif + free(channel); +} + +/* Wakes the owner's loop with a wake_kind event. */ +static void scr_channel_wake(ScrMailbox *owner, unsigned wake_kind) { + if (!owner) return; + ScrMailEvent *wake = calloc(1, sizeof(*wake)); + if (!wake) scr_trap("scriptc: out of memory\n"); + wake->kind = wake_kind; + scr_mailbox_post(owner, wake); + scr_mailbox_release(owner); +} + +void scr_channel_set_owner(ScrChannel *channel, int side, ScrMailbox *owner, unsigned wake_kind) { + scr_channel_lock(channel); + ScrChannelEnd *end = &channel->end[side]; + ScrMailbox *old = end->owner; + end->owner = owner ? scr_mailbox_retain(owner) : NULL; + ScrMailbox *wake = end->owner && (end->head || end->peer_closed) ? scr_mailbox_retain(end->owner) : NULL; + scr_channel_unlock(channel); + scr_mailbox_release(old); + scr_channel_wake(wake, wake_kind); +} + +/* postMessage from `side`: the message (consumed) queues on the other end + * unless either end has closed. */ +bool scr_channel_post(ScrChannel *channel, int side, ScrMessage *message, unsigned message_kind, + unsigned wake_kind) { + ScrMailEvent *event = calloc(1, sizeof(*event)); + if (!event) scr_trap("scriptc: out of memory\n"); + event->kind = message_kind; + event->message = message; + scr_channel_lock(channel); + ScrChannelEnd *peer = &channel->end[1 - side]; + if (channel->end[side].closed || peer->closed) { + scr_channel_unlock(channel); + scr_mail_event_free(event); + return false; + } + *peer->tail = event; + peer->tail = &event->next; + peer->count++; + ScrMailbox *wake = peer->owner ? scr_mailbox_retain(peer->owner) : NULL; + scr_channel_unlock(channel); + scr_channel_wake(wake, wake_kind); + return true; +} + +/* close() on `side`: nothing more is posted in either direction. */ +void scr_channel_close(ScrChannel *channel, int side) { + scr_channel_lock(channel); + channel->end[side].closed = true; + scr_channel_unlock(channel); +} + +/* The closed port's 'close' ran: the other end learns it, after its queued + * messages. */ +void scr_channel_notify_close(ScrChannel *channel, int side, unsigned wake_kind) { + scr_channel_lock(channel); + ScrChannelEnd *peer = &channel->end[1 - side]; + ScrMailbox *wake = NULL; + if (!peer->peer_closed && !peer->closed) { + peer->peer_closed = true; + peer->peer_closed_seq = atomic_fetch_add(&scr_channel_notice_seq, 1) + 1; + wake = peer->owner ? scr_mailbox_retain(peer->owner) : NULL; + } + scr_channel_unlock(channel); + scr_channel_wake(wake, wake_kind); +} + +/* The next queued message for `side`, or NULL. */ +ScrMailEvent *scr_channel_take(ScrChannel *channel, int side) { + scr_channel_lock(channel); + ScrChannelEnd *end = &channel->end[side]; + ScrMailEvent *event = end->head; + if (event) { + end->head = event->next; + if (!end->head) end->tail = &end->head; + end->count--; + event->next = NULL; + } + scr_channel_unlock(channel); + return event; +} + +size_t scr_channel_count(ScrChannel *channel, int side) { + scr_channel_lock(channel); + size_t count = channel->end[side].count; + scr_channel_unlock(channel); + return count; +} + +/* The other end's close notice order (0: none). */ +uint64_t scr_channel_peer_closed(ScrChannel *channel, int side) { + scr_channel_lock(channel); + uint64_t seq = channel->end[side].peer_closed ? channel->end[side].peer_closed_seq : 0; + scr_channel_unlock(channel); + return seq; +} + +/* Drops the queued messages for `side` (its port closed). */ +void scr_channel_drop(ScrChannel *channel, int side) { + scr_channel_lock(channel); + ScrChannelEnd *end = &channel->end[side]; + ScrMailEvent *head = end->head; + end->head = NULL; + end->tail = &end->head; + end->count = 0; + scr_channel_unlock(channel); + while (head) { + ScrMailEvent *event = head; + head = event->next; + scr_mail_event_free(event); + } +} diff --git a/packages/runtime/src/scr_mailbox.h b/packages/runtime/src/scr_mailbox.h index 792660bdb1..75d99bbc86 100644 --- a/packages/runtime/src/scr_mailbox.h +++ b/packages/runtime/src/scr_mailbox.h @@ -25,4 +25,25 @@ bool scr_mailbox_pending(ScrMailbox *mailbox); void scr_mailbox_wait(ScrMailbox *mailbox, double milliseconds); int scr_mailbox_pollfd(const ScrMailbox *mailbox); +/* A MessageChannel: two ends, each with its queue of incoming messages and + * the inbox of the thread that owns its port (NULL while the port is in + * transit). Posting and closing wake the receiving owner with a wake_kind + * event. Ends are thread-neutral, so a port can move between threads. */ +typedef struct ScrChannel ScrChannel; +ScrChannel *scr_channel_new(void); +ScrChannel *scr_channel_retain(ScrChannel *channel); +void scr_channel_release(ScrChannel *channel); +void scr_channel_set_owner(ScrChannel *channel, int side, ScrMailbox *owner, unsigned wake_kind); +/* Consumes message; false when either end has closed (the message is dropped). */ +bool scr_channel_post(ScrChannel *channel, int side, ScrMessage *message, unsigned message_kind, + unsigned wake_kind); +/* close() stops posting both ways; notify_close, once the closed port's + * 'close' ran, tells the other end (after its queued messages). */ +void scr_channel_close(ScrChannel *channel, int side); +void scr_channel_notify_close(ScrChannel *channel, int side, unsigned wake_kind); +ScrMailEvent *scr_channel_take(ScrChannel *channel, int side); +size_t scr_channel_count(ScrChannel *channel, int side); +uint64_t scr_channel_peer_closed(ScrChannel *channel, int side); /* the notice's order, 0: none */ +void scr_channel_drop(ScrChannel *channel, int side); + #endif diff --git a/packages/runtime/src/scr_message.c b/packages/runtime/src/scr_message.c index 5b27bfc4ed..84f3e5098d 100644 --- a/packages/runtime/src/scr_message.c +++ b/packages/runtime/src/scr_message.c @@ -21,8 +21,12 @@ typedef enum { /* A typed value published with @scriptc/threads (immortal), delivered as * a fresh typed-ref capsule over the same pointer. */ SCR_MESSAGE_TYPED_REF, + /* A transferred MessagePort: its channel end (+1), the side in number. */ + SCR_MESSAGE_PORT, } ScrMessageKind; +const ScrMessagePortOps *scr_message_port_ops = NULL; + typedef struct { char *key; size_t length; @@ -43,6 +47,7 @@ typedef struct { ScrBytesElem element; bool data_view; ScrDyn *published; /* SCR_MESSAGE_PUBLISHED: immortal root, owns nothing */ + ScrChannel *channel; /* SCR_MESSAGE_PORT */ /* SCR_MESSAGE_TYPED_REF: the capsule's process-wide adapters and type key * (vtables, adapters and type keys are shared by every thread). */ struct { @@ -85,8 +90,18 @@ typedef struct { size_t *stack; size_t depth; size_t stack_capacity; + const ScrDyn *transfer; /* the transfer list (a dyn array), or NULL */ } ScrMessageEncoder; +static bool scr_message_listed(const ScrMessageEncoder *encoder, const ScrDyn *value) { + if (!encoder->transfer) return false; + for (size_t i = 0; i < encoder->transfer->v.arr.len; i++) { + const ScrDyn *item = encoder->transfer->v.arr.items[i]; + if (item && item->kind == SCR_DYN_HANDLE && item->v.handle.ptr == value->v.handle.ptr) return true; + } + return false; +} + static void *scr_message_alloc(size_t count, size_t size) { if (count > SIZE_MAX / size) scr_trap("scriptc: message too large\n"); void *data = calloc(count ? count : 1, size); @@ -303,6 +318,20 @@ static size_t scr_message_value(ScrMessageEncoder *encoder, const ScrDyn *input) break; case SCR_DYN_BYTES: case SCR_DYN_HANDLE: { + if (value->kind == SCR_DYN_HANDLE && scr_message_port_ops && scr_message_port_ops->is_port(value)) { + int side = 0; + ScrChannel *channel = scr_message_listed(encoder, value) + ? scr_message_port_ops->channel_of(value, &side) : NULL; + if (!channel) { + scr_throw_domex("DataCloneError", + "Object that needs transfer was found in message but not listed in transferList"); + return SIZE_MAX; + } + node->kind = SCR_MESSAGE_PORT; + node->channel = channel; + node->number = side; + break; + } bool buffer = scr_buffer_storage_is(value); if (value->kind == SCR_DYN_HANDLE && !buffer) goto unsupported; const ScrBytes *bytes = buffer ? value->v.handle.ptr : value->v.bytes; @@ -491,11 +520,39 @@ static ScrMessage *scr_message_try_publish(const ScrDyn *value) { return message; } -ScrMessage *scr_message_encode(const ScrDyn *value) { - ScrMessage *published = scr_message_try_publish(value); +static ScrMessage *scr_message_encode_with(const ScrDyn *value, const ScrDyn *transfer); + +ScrMessage *scr_message_encode(const ScrDyn *value) { return scr_message_encode_with(value, NULL); } + +ScrMessage *scr_message_encode_transfer(const ScrDyn *value, const ScrDyn *transfer) { + if (!transfer || transfer->kind != SCR_DYN_ARR || transfer->v.arr.len == 0) + return scr_message_encode_with(value, NULL); + for (size_t i = 0; i < transfer->v.arr.len; i++) { + const ScrDyn *item = transfer->v.arr.items[i]; + int side = 0; + ScrChannel *channel = item && item->kind == SCR_DYN_HANDLE && scr_message_port_ops + ? scr_message_port_ops->channel_of(item, &side) : NULL; + if (!channel) { + static const char text[] = "Transfer lists with values other than MessagePorts are not supported yet"; + ScrStr *m = scr_str_new(text, sizeof text - 1); + scr_throw_lowering_fence(m); + scr_str_release(m); + return NULL; + } + scr_message_port_ops->release(channel); + } + ScrMessage *message = scr_message_encode_with(value, transfer); + if (!message) return NULL; + /* the listed ports now belong to the message */ + for (size_t i = 0; i < transfer->v.arr.len; i++) scr_message_port_ops->detach(transfer->v.arr.items[i]); + return message; +} + +static ScrMessage *scr_message_encode_with(const ScrDyn *value, const ScrDyn *transfer) { + ScrMessage *published = transfer ? NULL : scr_message_try_publish(value); if (published) return published; ScrMessage *message = scr_message_alloc(1, sizeof(*message)); - ScrMessageEncoder encoder = {.message = message}; + ScrMessageEncoder encoder = {.message = message, .transfer = transfer}; if (scr_message_value(&encoder, value) == SIZE_MAX) goto done; while (encoder.depth && !scr_exc_pending()) { size_t id = encoder.stack[encoder.depth - 1]; @@ -598,6 +655,9 @@ ScrDyn *scr_message_decode(const ScrMessage *message) { node->typed.ptr, node->typed.retain, node->typed.release, node->typed.type_key, node->typed.type_key_len, node->typed.materialize, node->typed.commit); break; + case SCR_MESSAGE_PORT: + values[i] = scr_message_port_ops->adopt(node->channel, (int)node->number); + break; } if (!values[i] || scr_exc_pending()) goto done; } @@ -643,6 +703,7 @@ void scr_message_free(ScrMessage *message) { free(node->properties); free(node->data); scr_shared_release(node->shared); + if (node->kind == SCR_MESSAGE_PORT) scr_message_port_ops->release(node->channel); } free(message->nodes); free(message); diff --git a/packages/runtime/src/scr_message.h b/packages/runtime/src/scr_message.h index 29ed5f7a78..ba0a7391cf 100644 --- a/packages/runtime/src/scr_message.h +++ b/packages/runtime/src/scr_message.h @@ -8,6 +8,24 @@ * values; release the message separately after delivery or cancellation. */ typedef struct ScrMessage ScrMessage; ScrMessage *scr_message_encode(const ScrDyn *value); +/* postMessage(value, transferList): the listed MessagePorts move with the + * message (a port in the value that is not listed throws DataCloneError); + * transfer is NULL or a dyn array. */ +ScrMessage *scr_message_encode_transfer(const ScrDyn *value, const ScrDyn *transfer); + +/* MessagePort transfer, installed by the worker events unit: channel_of + * answers a channel port's end (+1 channel; NULL for other values, side + * set), is_port any MessagePort, detach leaves the sender's port unusable, + * adopt makes a port in the decoding thread (the channel borrowed). */ +typedef struct ScrChannel ScrChannel; +typedef struct { + ScrChannel *(*channel_of)(const ScrDyn *value, int *side); + bool (*is_port)(const ScrDyn *value); + void (*detach)(const ScrDyn *value); + ScrDyn *(*adopt)(ScrChannel *channel, int side); + void (*release)(ScrChannel *channel); +} ScrMessagePortOps; +extern const ScrMessagePortOps *scr_message_port_ops; ScrDyn *scr_message_decode(const ScrMessage *message); void scr_message_free(ScrMessage *message); diff --git a/packages/runtime/src/scr_worker.h b/packages/runtime/src/scr_worker.h index cc00a8f021..008159fa2b 100644 --- a/packages/runtime/src/scr_worker.h +++ b/packages/runtime/src/scr_worker.h @@ -38,5 +38,8 @@ void scr_worker_port_close(void); ScrMailbox *scr_worker_inbox(void); void scr_loop_set_workers(bool (*pending)(void), bool (*ready)(void), void (*dispatch)(void), int (*pollfd)(void), void (*wait)(double)); +/* After a dispatch: true keeps the loop pass delivering (a MessagePort poll + * phase in progress), with microtasks drained between events. */ +void scr_loop_set_workers_more(bool (*more)(void)); #endif diff --git a/packages/runtime/src/scr_worker_events.c b/packages/runtime/src/scr_worker_events.c index ab4b4878a2..51fbce59fc 100644 --- a/packages/runtime/src/scr_worker_events.c +++ b/packages/runtime/src/scr_worker_events.c @@ -17,12 +17,30 @@ typedef struct ScrWorkerView { bool closing; ScrMailEvent *queued_head; ScrMailEvent *queued_tail; + /* A MessageChannel port: its channel (+1) and end; NULL for parentPort. */ + ScrChannel *channel; + int side; + bool detached; /* transferred to another thread */ + uint64_t close_seq; /* when close() ran: the latest close emits first */ + uint64_t close_gen; /* the poll generation close() ran in */ + size_t phase_left; /* messages this poll phase still delivers */ + bool phase_close; /* the other end's close, delivered this phase */ struct ScrWorkerView *next; } ScrWorkerView; static SCR_TL ScrMailbox *scr_worker_events; static SCR_TL ScrWorkerView *scr_worker_views; static SCR_TL ScrWorkerView *scr_worker_port; +/* this thread's MessageChannel ports (registry references) */ +static SCR_TL ScrWorkerView *scr_channel_ports; +static SCR_TL uint64_t scr_channel_close_seq; +static SCR_TL bool scr_channel_phase; /* a poll phase is being delivered */ +/* completed poll phases: a port's 'close' runs in the close phase after + * the poll phase its close() came before or during */ +static SCR_TL uint64_t scr_channel_poll_gen; +/* the loop pass continues a phase (scr_channel_more answered true) */ +static SCR_TL bool scr_channel_continuing; +static SCR_TL bool scr_channel_closing_phase; static const char *const scr_worker_method_names[] = { "ref", "unref", "hasRef", "terminate", "postMessage", "on", "once", "addListener", "prependListener", "prependOnceListener", "off", "removeListener", "removeAllListeners", @@ -38,6 +56,7 @@ static void scr_worker_view_trace(void *ptr, ScrTraceVisit visit, void *ctx) { static void scr_worker_view_gcfree(void *ptr) { ScrWorkerView *view = ptr; scr_worker_thread_release(view->thread); + scr_channel_release(view->channel); while (view->queued_head) { ScrMailEvent *event = view->queued_head; view->queued_head = event->next; @@ -106,6 +125,44 @@ static bool scr_worker_empty_transfer(const ScrDyn *value) { return value->kind == SCR_DYN_UNDEF || (value->kind == SCR_DYN_ARR && value->v.arr.len == 0); } +/* postMessage's transfer argument: an array, or { transfer } (+1, or NULL + * when empty) */ +static ScrDyn *scr_worker_transfer_list(ScrDyn *arg) { + ScrDyn *value = arg; + ScrDyn *own = NULL; + if (value->kind == SCR_DYN_TYPED_REF) { + own = scr_dyn_typed_ref_materialize(value); + if (!own || scr_exc_pending()) { + scr_dyn_release(own); + return NULL; + } + value = own; + } + if (value->kind == SCR_DYN_OBJ) { + ScrDyn *list = scr_dyn_obj_get(value, "transfer", 8); + ScrDyn *result = list ? scr_worker_transfer_list(list) : NULL; + scr_dyn_release(own); + return result; + } + if (value->kind == SCR_DYN_ARR && value->v.arr.len > 0) { + ScrDyn *result = scr_dyn_retain(value); + scr_dyn_release(own); + return result; + } + bool empty = scr_worker_empty_transfer(value); + scr_dyn_release(own); + if (!empty) scr_worker_refusal("This postMessage transfer argument is not supported yet"); + return NULL; +} + +static bool scr_worker_transfer_has(const ScrDyn *transfer, const ScrWorkerView *view) { + for (size_t i = 0; i < transfer->v.arr.len; i++) { + const ScrDyn *item = transfer->v.arr.items[i]; + if (item && item->kind == SCR_DYN_HANDLE && item->v.handle.ptr == view) return true; + } + return false; +} + static ScrDyn *scr_worker_view_invoke(void *ptr, ScrDyn *self, const char *method, ScrDyn *const *args, size_t argc, const char *what) { (void)what; @@ -115,7 +172,7 @@ static ScrDyn *scr_worker_view_invoke(void *ptr, ScrDyn *self, const char *metho return scr_dyn_retain(scr_dyn_undefined()); } if (view->port && !strcmp(method, "hasRef")) - return scr_dyn_new_bool(view->referenced && !view->closed); + return scr_dyn_new_bool(view->referenced && (!view->closed || (view->channel && view->closing))); if (!view->port && !strcmp(method, "terminate")) { if (!view->termination) { view->termination = scr_promise_new(); @@ -129,16 +186,30 @@ static ScrDyn *scr_worker_view_invoke(void *ptr, ScrDyn *self, const char *metho /* Worker drops messages after exit before validating or cloning. A * closed MessagePort still validates its payload, as Node does. */ if (!view->port && view->closed) return scr_dyn_retain(scr_dyn_undefined()); + if (view->detached) return scr_dyn_retain(scr_dyn_undefined()); if (!argc) { - static const char message[] = "The \"value\" argument must be specified"; - scr_throw_error_msg_code(SCR_ERR_TYPE, message, sizeof message - 1, "ERR_MISSING_ARGS"); + if (view->channel) { + static const char message[] = "Not enough arguments to MessagePort.postMessage"; + scr_throw_error_msg_code(SCR_ERR_TYPE, message, sizeof message - 1, "ERR_MISSING_ARGS"); + } else { + static const char message[] = "The \"value\" argument must be specified"; + scr_throw_error_msg_code(SCR_ERR_TYPE, message, sizeof message - 1, "ERR_MISSING_ARGS"); + } return NULL; } - if (argc > 1 && !scr_worker_empty_transfer(args[1])) - return scr_exc_pending() ? NULL : scr_worker_refusal("Worker postMessage transfer lists are not supported yet"); - ScrMessage *message = scr_message_encode(args[0]); + ScrDyn *transfer = argc > 1 ? scr_worker_transfer_list(args[1]) : NULL; + if (scr_exc_pending()) return NULL; + if (transfer && scr_worker_transfer_has(transfer, view)) { + scr_dyn_release(transfer); + scr_throw_domex("DataCloneError", "Transfer list contains source port"); + return NULL; + } + ScrMessage *message = scr_message_encode_transfer(args[0], transfer); + scr_dyn_release(transfer); if (!message) return NULL; if (view->closed) scr_message_free(message); + else if (view->channel) + scr_channel_post(view->channel, view->side, message, SCR_WORKER_MESSAGE, SCR_WORKER_WAKE); else if (view->port) scr_worker_post_parent(message); else scr_worker_thread_post(view->thread, message); return scr_dyn_retain(scr_dyn_undefined()); @@ -185,9 +256,17 @@ static ScrDyn *scr_worker_view_invoke(void *ptr, ScrDyn *self, const char *metho scr_emitter_release(scr_emitter_on_flex(view->events, name, args[0], true, false)); scr_str_release(name); } - if (!view->closed) view->closing = true; - view->closed = true; - scr_worker_port_close(); + if (!view->closed && !view->detached) { + view->closing = true; + view->closed = true; + if (view->channel) { + view->close_seq = ++scr_channel_close_seq; + view->close_gen = scr_channel_poll_gen; + scr_channel_close(view->channel, view->side); + } else { + scr_worker_port_close(); + } + } return scr_dyn_retain(scr_dyn_undefined()); } if (view->port && !strcmp(method, "start")) { @@ -235,6 +314,9 @@ static ScrDyn *scr_worker_view_get(void *ptr, const char *key, size_t length) { static bool scr_workers_pending(void) { for (ScrWorkerView *view = scr_worker_views; view; view = view->next) if (view->referenced) return true; + for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) + if (view->closing || (view->started && view->referenced && !view->closed && !view->detached)) + return true; return scr_worker_port && (scr_worker_port->closing || (scr_worker_port->referenced && !scr_worker_port->closed)); } @@ -243,7 +325,19 @@ static int scr_worker_events_pollfd(void) { return scr_worker_events ? scr_mailbox_pollfd(scr_worker_events) : -1; } +/* a channel port with something to deliver: its 'close', the other end's + * close, or messages once started */ +static bool scr_channel_port_ready(ScrWorkerView *view) { + if (view->closing || view->phase_left || view->phase_close) return true; + if (view->closed || view->detached) return false; + if (view->started && scr_channel_count(view->channel, view->side) > 0) return true; + return scr_channel_peer_closed(view->channel, view->side) && + scr_channel_count(view->channel, view->side) == 0; +} + static bool scr_worker_events_ready(void) { + for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) + if (scr_channel_port_ready(view)) return true; return (scr_worker_port && (scr_worker_port->closing || (scr_worker_port->started && scr_worker_port->queued_head))) || (scr_worker_events && scr_mailbox_pending(scr_worker_events)); @@ -253,7 +347,123 @@ static void scr_worker_events_wait(double milliseconds) { if (!scr_worker_events_ready()) scr_mailbox_wait(scr_worker_events, milliseconds); } +static void scr_worker_port_emit(ScrWorkerView *view, const char *event, ScrDyn *argument) { + ScrStr *name = scr_str_new(event, strlen(event)); + scr_dyn_this_push(view, SCR_DYNH_MESSAGE_PORT); + scr_emitter_emit_flex(view->events, name, argument ? &argument : NULL, argument ? 1 : 0); + scr_dyn_this_pop(); + scr_str_release(name); +} + +/* The channel ports' turn, one event per call, in Node's phase order: + * - the close phase: ports this thread closed, the latest close() first + * (libuv's close callbacks), each telling its other end; + * - the poll phase: a snapshot of each port's queued messages (the batch + * Node delivers), ports in creation order, then the other ends' close + * notices in the order they were sent. scr_channel_more keeps the loop + * pass going until the phase is delivered, with microtasks between. + * False when nothing was ready. */ +static bool scr_channel_close_due(const ScrWorkerView *view) { + return view->closing && view->close_gen < scr_channel_poll_gen; +} + +static bool scr_channel_dispatch(void) { + bool continuing = scr_channel_continuing; + scr_channel_continuing = false; + /* a loop pass runs one phase: the close phase, or the poll phase */ + if (scr_channel_closing_phase && !continuing) scr_channel_closing_phase = false; + if (!scr_channel_phase && (!continuing || scr_channel_closing_phase)) { + ScrWorkerView *latest = NULL; + for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) + if (scr_channel_close_due(view) && (!latest || view->close_seq > latest->close_seq)) + latest = view; + if (latest) { + scr_worker_view_retain(latest); + latest->closing = false; + scr_channel_closing_phase = true; + scr_channel_drop(latest->channel, latest->side); + scr_channel_notify_close(latest->channel, latest->side, SCR_WORKER_WAKE); + scr_worker_port_emit(latest, "close", NULL); + scr_worker_view_release(latest); + return true; + } + if (scr_channel_closing_phase) return false; + bool any = false; + for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) { + if (view->closed || view->detached) continue; + size_t count = scr_channel_count(view->channel, view->side); + view->phase_left = view->started ? count : 0; + view->phase_close = count == 0 && scr_channel_peer_closed(view->channel, view->side) != 0; + any = any || view->phase_left || view->phase_close; + } + if (!any) { + scr_channel_poll_gen++; + return false; + } + scr_channel_phase = true; + } + for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) { + if (!view->phase_left) continue; + view->phase_left--; + ScrMailEvent *event = scr_channel_take(view->channel, view->side); + if (!event) { + view->phase_left = 0; + continue; + } + scr_worker_view_retain(view); + ScrDyn *argument = event->message ? scr_message_decode(event->message) : NULL; + if (argument) scr_worker_port_emit(view, "message", argument); + scr_dyn_release(argument); + scr_mail_event_free(event); + scr_worker_view_release(view); + return true; + } + ScrWorkerView *first = NULL; + uint64_t first_seq = 0; + for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) { + if (!view->phase_close) continue; + uint64_t seq = scr_channel_peer_closed(view->channel, view->side); + if (!first || seq < first_seq) { + first = view; + first_seq = seq; + } + } + if (!first) { + scr_channel_phase = false; + scr_channel_poll_gen++; + return false; + } + first->phase_close = false; + if (first->closed) return true; + scr_worker_view_retain(first); + first->closed = true; + scr_channel_drop(first->channel, first->side); + scr_worker_port_emit(first, "close", NULL); + scr_worker_view_release(first); + return true; +} + +/* more of the current poll phase to deliver in this loop pass */ +static bool scr_channel_more(void) { + bool more = false; + if (scr_channel_closing_phase) { + /* the close phase runs its callbacks back to back */ + for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) + if (scr_channel_close_due(view)) more = true; + } else if (scr_channel_phase) { + for (ScrWorkerView *view = scr_channel_ports; view; view = view->next) + if (view->phase_left || view->phase_close) more = true; + if (!more) { + scr_channel_phase = false; + scr_channel_poll_gen++; + } + } + scr_channel_continuing = more; + return more; +} + static void scr_worker_dispatch(void) { + if (scr_channel_dispatch()) return; ScrWorkerView *port = scr_worker_port; if (port && port->closing) { port->closing = false; @@ -319,6 +529,7 @@ static void scr_worker_dispatch(void) { static void scr_worker_events_cleanup(void) { scr_loop_set_workers(NULL, NULL, NULL, NULL, NULL); + scr_loop_set_workers_more(NULL); scr_mailbox_close(scr_worker_events); for (ScrWorkerView *view = scr_worker_views; view; view = view->next) scr_worker_thread_terminate(view->thread); @@ -331,6 +542,13 @@ static void scr_worker_events_cleanup(void) { } scr_worker_view_release(scr_worker_port); scr_worker_port = NULL; + while (scr_channel_ports) { + ScrWorkerView *view = scr_channel_ports; + scr_channel_ports = view->next; + view->next = NULL; + if (view->channel) scr_channel_set_owner(view->channel, view->side, NULL, SCR_WORKER_WAKE); + scr_worker_view_release(view); + } for (size_t i = 0; i < sizeof scr_worker_methods / sizeof scr_worker_methods[0]; i++) { scr_dyn_release(scr_worker_methods[i]); scr_worker_methods[i] = NULL; @@ -339,6 +557,8 @@ static void scr_worker_events_cleanup(void) { scr_worker_events = NULL; } +static void scr_port_ops_install(void); + static bool scr_worker_events_install(void) { if (scr_worker_events) return true; ScrMailbox *inbox = scr_worker_inbox(); @@ -355,12 +575,98 @@ static bool scr_worker_events_install(void) { }; scr_dyn_handle_install(SCR_DYNH_WORKER, &worker_ops); scr_dyn_handle_install(SCR_DYNH_MESSAGE_PORT, &port_ops); + scr_port_ops_install(); if (scr_context_atexit(scr_worker_events_cleanup) != 0) scr_trap("scriptc: out of memory\n"); scr_loop_set_workers(scr_workers_pending, scr_worker_events_ready, scr_worker_dispatch, scr_worker_events_pollfd, scr_worker_events_wait); + scr_loop_set_workers_more(scr_channel_more); return true; } +/* ── MessageChannel ports ─────────────────────────────────────────────── */ + +static ScrWorkerView *scr_channel_port_new(ScrChannel *channel /*borrowed*/, int side) { + ScrWorkerView *view = scr_worker_view_new(true); + view->channel = scr_channel_retain(channel); + view->side = side; + /* the registry reference, in creation order (Node delivers ports in the + * order their handles were made) */ + ScrWorkerView **tail = &scr_channel_ports; + while (*tail) tail = &(*tail)->next; + *tail = view; + scr_channel_set_owner(channel, side, scr_worker_events, SCR_WORKER_WAKE); + return view; +} + +static ScrWorkerView *scr_port_view_of(const ScrDyn *value) { + if (!value || value->kind != SCR_DYN_HANDLE || value->v.handle.tag != SCR_DYNH_MESSAGE_PORT) + return NULL; + return value->v.handle.ptr; +} + +static ScrChannel *scr_port_channel_of(const ScrDyn *value, int *side) { + ScrWorkerView *view = scr_port_view_of(value); + if (!view || !view->channel || view->detached) return NULL; + *side = view->side; + return scr_channel_retain(view->channel); +} + +static bool scr_port_is(const ScrDyn *value) { return scr_port_view_of(value) != NULL; } + +/* the sender's port after a transfer: inert, and no longer this thread's */ +static void scr_port_detach(const ScrDyn *value) { + ScrWorkerView *view = scr_port_view_of(value); + if (!view || !view->channel || view->detached) return; + view->detached = true; + view->closing = false; + scr_channel_set_owner(view->channel, view->side, NULL, SCR_WORKER_WAKE); +} + +static ScrDyn *scr_port_adopt(ScrChannel *channel, int side) { + if (!scr_worker_events_install()) return NULL; + ScrWorkerView *view = scr_channel_port_new(channel, side); + return scr_dyn_new_handle(view, SCR_DYNH_MESSAGE_PORT); +} + +static const ScrMessagePortOps scr_port_ops = { + &scr_port_channel_of, &scr_port_is, &scr_port_detach, &scr_port_adopt, &scr_channel_release, +}; + +static void scr_port_ops_install(void) { scr_message_port_ops = &scr_port_ops; } + +/* new MessageChannel(): { port1, port2 } */ +ScrDyn *scr_worker_message_channel(void) { + if (!scr_worker_events_install()) return NULL; + ScrChannel *channel = scr_channel_new(); + ScrWorkerView *a = scr_channel_port_new(channel, 0); + ScrWorkerView *b = scr_channel_port_new(channel, 1); + scr_channel_release(channel); + ScrDyn *object = scr_dyn_new_obj(); + scr_dyn_obj_set(object, "port1", 5, scr_dyn_new_handle(a, SCR_DYNH_MESSAGE_PORT)); + scr_dyn_obj_set(object, "port2", 5, scr_dyn_new_handle(b, SCR_DYNH_MESSAGE_PORT)); + return object; +} + +/* receiveMessageOnPort(port): { message } of the next queued message, or + * undefined */ +ScrDyn *scr_worker_receive_message_on_port(ScrDyn *port) { + ScrWorkerView *view = scr_port_view_of(port); + if (!view) { + static const char message[] = "The \"port\" argument must be a MessagePort instance"; + scr_throw_error_msg_code(SCR_ERR_TYPE, message, sizeof message - 1, "ERR_INVALID_ARG_TYPE"); + return NULL; + } + if (!view->channel) return scr_worker_refusal("receiveMessageOnPort on parentPort is not supported yet"); + ScrMailEvent *event = view->closed || view->detached ? NULL : scr_channel_take(view->channel, view->side); + if (!event) return scr_dyn_retain(scr_dyn_undefined()); + ScrDyn *value = event->message ? scr_message_decode(event->message) : NULL; + scr_mail_event_free(event); + if (!value) return NULL; + ScrDyn *object = scr_dyn_new_obj(); + scr_dyn_obj_set(object, "message", 7, value); + return object; +} + ScrDyn *scr_worker_parent_port(void) { if (scr_context_is_main()) return scr_dyn_new_null(); if (!scr_worker_events_install()) return NULL; @@ -407,6 +713,7 @@ static ScrDyn *scr_worker_new_options(double root, ScrStr *filename, ScrDyn *opt if (!scr_worker_events_install()) return NULL; ScrDyn *data = scr_dyn_undefined(); ScrDyn *argv = scr_dyn_undefined(); + ScrDyn *transfer_list = scr_dyn_undefined(); if (options->kind != SCR_DYN_UNDEF) { if (options->kind != SCR_DYN_OBJ) { scr_dyn_arg_type_fail("options", "object", options); @@ -417,12 +724,16 @@ static ScrDyn *scr_worker_new_options(double root, ScrStr *filename, ScrDyn *opt if (entry->accessor) return scr_worker_refusal("Worker option accessors are not supported yet"); if (entry->key_len == 10 && !memcmp(entry->key, "workerData", 10)) data = entry->value; else if (entry->key_len == 4 && !memcmp(entry->key, "argv", 4)) argv = entry->value; + else if (entry->key_len == 12 && !memcmp(entry->key, "transferList", 12)) transfer_list = entry->value; else return scr_worker_refusal("This Worker constructor option is not supported yet"); } } ScrArr *arguments = scr_worker_arguments(filename, argv); if (!arguments) return NULL; - ScrMessage *wire = scr_message_encode(data); + ScrDyn *transfer = scr_worker_transfer_list(transfer_list); + if (scr_exc_pending()) { scr_arr_release(arguments); return NULL; } + ScrMessage *wire = scr_message_encode_transfer(data, transfer); + scr_dyn_release(transfer); if (!wire) { scr_arr_release(arguments); return NULL; } ScrWorkerThread *thread = scr_worker_thread_start(sc_worker_entry, (uint32_t)root, wire, scr_worker_events, arguments); scr_arr_release(arguments); diff --git a/tests/corpus/5793-worker-message-channel.ts b/tests/corpus/5793-worker-message-channel.ts new file mode 100644 index 0000000000..5bdfb053a3 --- /dev/null +++ b/tests/corpus/5793-worker-message-channel.ts @@ -0,0 +1,51 @@ +import { MessageChannel, receiveMessageOnPort, type MessagePort } from "node:worker_threads"; + +// MessageChannel within one thread: hasRef, the validation errors, +// receiveMessageOnPort, close() reaching the other port after its queued +// messages, and the batch that close() inside a listener still delivers. +const a = new MessageChannel(); +console.log("hasRef new", a.port1.hasRef(), a.port2.hasRef()); +a.port1.on("message", () => {}); +console.log("hasRef listener", a.port1.hasRef()); +a.port1.unref(); +console.log("hasRef unref", a.port1.hasRef()); +a.port1.ref(); +console.log("hasRef ref", a.port1.hasRef()); +a.port1.close(); +console.log("hasRef closed", a.port1.hasRef()); +const c = new MessageChannel(); +const checks: Array<() => void> = [ + () => c.port1.postMessage(c.port2), + () => c.port1.postMessage({ p: c.port2 }), + () => c.port1.postMessage(1, [c.port1]), + () => receiveMessageOnPort({} as MessagePort), +]; +for (const check of checks) { + try { + check(); + console.log("ok"); + } catch (err) { + const e = err as Error; + console.log(e.name, e.message); + } +} +c.port1.close(); +const d = new MessageChannel(); +d.port1.postMessage("before close"); +d.port1.close(); +d.port1.postMessage("after close"); +console.log("d2 receive", JSON.stringify(receiveMessageOnPort(d.port2)), JSON.stringify(receiveMessageOnPort(d.port2))); +d.port2.on("close", () => console.log("d2 close")); +d.port1.on("close", () => console.log("d1 close")); +const e = new MessageChannel(); +e.port2.on("message", (m: unknown) => { + console.log("e message", m); + if (m === 2) e.port2.close(); +}); +e.port1.on("close", () => console.log("e1 close")); +e.port2.on("close", () => console.log("e2 close")); +e.port1.postMessage(1); +e.port1.postMessage(2); +e.port1.postMessage(3); +setImmediate(() => console.log("immediate")); +process.on("exit", () => console.log("exit")); diff --git a/tests/corpus/5794-worker-message-port-close-order.ts b/tests/corpus/5794-worker-message-port-close-order.ts new file mode 100644 index 0000000000..7aeaf3e149 --- /dev/null +++ b/tests/corpus/5794-worker-message-port-close-order.ts @@ -0,0 +1,14 @@ +// MessagePort close order: a port's own 'close' runs after the loop's +// poll phase, and the other end's close notice arrives in the next one, +// after its queued messages, while something keeps the loop alive. +import { MessageChannel } from "node:worker_threads"; +const d = new MessageChannel(); +d.port2.on("close", () => console.log("d2 close (peer)")); +d.port1.on("close", () => console.log("d1 close (own)")); +d.port1.close(); +const f = new MessageChannel(); +f.port1.on("message", (m: unknown) => console.log("f got", m)); +f.port2.postMessage("x"); +f.port2.close(); +f.port1.on("close", () => console.log("f1 close (peer, after messages)")); +setTimeout(() => console.log("timer"), 30); diff --git a/tests/corpus/5795-worker-message-port-batches.ts b/tests/corpus/5795-worker-message-port-batches.ts new file mode 100644 index 0000000000..5bde69a738 --- /dev/null +++ b/tests/corpus/5795-worker-message-port-batches.ts @@ -0,0 +1,16 @@ +// Queued MessagePort messages arrive as one batch before setImmediate, +// with nextTicks and microtasks run after each message. +import { MessageChannel } from "node:worker_threads"; +const { port1, port2 } = new MessageChannel(); +port1.on("message", (m: unknown) => { + console.log("message", m); + Promise.resolve().then(() => console.log("microtask after", m)); + process.nextTick(() => console.log("tick after", m)); +}); +port2.postMessage(1); +port2.postMessage(2); +port2.postMessage(3); +setImmediate(() => { + console.log("immediate"); + port1.close(); +}); diff --git a/tests/corpus/5796-worker-message-channel.cjs b/tests/corpus/5796-worker-message-channel.cjs new file mode 100644 index 0000000000..d7231a5f73 --- /dev/null +++ b/tests/corpus/5796-worker-message-channel.cjs @@ -0,0 +1,45 @@ +'use strict'; +// MessageChannel in JavaScript: receiveMessageOnPort before a listener +// starts the port, message delivery before setImmediate, the error codes, +// the { transfer } options form, and close order (this thread's closes, +// then the other ends'). +const { MessageChannel, receiveMessageOnPort } = require('node:worker_threads'); + +const { port1, port2 } = new MessageChannel(); +port2.postMessage({ a: 1 }); +port2.postMessage('two'); +console.log('receive', JSON.stringify(receiveMessageOnPort(port1))); +port1.on('message', (m) => console.log('message', JSON.stringify(m))); +port2.postMessage([3]); +setImmediate(() => console.log('immediate')); + +const c = new MessageChannel(); +for (const f of [ + () => c.port1.postMessage({ p: c.port2 }), + () => c.port1.postMessage(1, { transfer: [c.port1] }), + () => c.port1.postMessage(), + () => receiveMessageOnPort({}), + () => receiveMessageOnPort(), +]) { + try { + f(); + console.log('ok'); + } catch (e) { + console.log(e.name, e.code, e.message); + } +} +c.port1.close(); + +const d = new MessageChannel(); +d.port2.on('close', () => console.log('d2 close (peer)')); +d.port1.on('close', () => console.log('d1 close (own)')); +d.port1.close(); +const f = new MessageChannel(); +f.port1.on('message', (m) => console.log('f got', m)); +f.port2.postMessage('x'); +f.port2.close(); +f.port1.on('close', () => console.log('f1 close (peer, after messages)')); +setTimeout(() => { + console.log('timer', JSON.stringify(receiveMessageOnPort(port1))); + port1.close(); +}, 30); diff --git a/tests/corpus/5797-worker-message-port-transfer.ts b/tests/corpus/5797-worker-message-port-transfer.ts new file mode 100644 index 0000000000..fa898b6bb5 --- /dev/null +++ b/tests/corpus/5797-worker-message-port-transfer.ts @@ -0,0 +1,50 @@ +import { MessageChannel, MessagePort, Worker, isMainThread, parentPort, workerData } from "node:worker_threads"; + +// MessagePorts moved to a worker through workerData's and postMessage's +// transferList: the worker answers on them and sees the other end close, +// a port in a message without the transferList throws DataCloneError, +// and a transferred port is inert. +if (isMainThread) { + const { port1, port2 } = new MessageChannel(); + const worker = new Worker(new URL(import.meta.url), { + workerData: { port: port2 }, + transferList: [port2], + }); + const second = new MessageChannel(); + port1.on("message", (m: unknown) => { + console.log("main got", JSON.stringify(m)); + if (m === "done") port1.close(); + }); + port1.on("close", () => console.log("port1 close")); + port1.postMessage("hello"); + worker.on("message", (m: unknown) => { + console.log("from worker", JSON.stringify(m)); + try { + worker.postMessage({ reply: second.port1 }); + } catch (err) { + const e = err as Error; + console.log(e.name, e.message); + } + worker.postMessage({ reply: second.port2 }, [second.port2]); + second.port2.postMessage("ignored after transfer"); + }); + second.port1.on("message", (m: unknown) => { + console.log("second got", JSON.stringify(m)); + second.port1.close(); + }); + worker.on("exit", (code: number) => console.log("exit", code)); +} else { + const port = (workerData as { port: MessagePort }).port; + port.on("message", (m: unknown) => { + port.postMessage(["echo", m]); + port.postMessage("done"); + }); + port.on("close", () => { + parentPort!.postMessage("worker port closed"); + parentPort!.once("message", (msg: { reply: MessagePort }) => { + // the reply port stays open until the main thread closes its end + msg.reply.on("message", () => {}); + msg.reply.postMessage({ from: "worker" }); + }); + }); +}