From fb7467881c9e91c2f3395f3350a9ad5bb9e28f8e Mon Sep 17 00:00:00 2001 From: Nick Nassiri Date: Sat, 4 Jul 2026 17:58:17 -0700 Subject: [PATCH] feat(worker_threads): compiled main-thread receiveMessageOnPort drain + MessageChannel messageerror (#1077) Closes the two compiled-mode gaps documented during epic #996 for the main-thread MessageChannel/$MessagePort path (the worker-driven paths were already dual-mode). 1. Synchronous receiveMessageOnPort (#1000 follow-up) $Runtime.WorkerThreadsReceiveMessageOnPort was an unconditional undefined stub because $MessagePort is emitted (EmitMessageChannelTypes) AFTER EmitRuntimeClass, so the helper could not reference the port queue at emit time. Split it: define the method during EmitWorkerThreadsModuleHelpers (so callers bind the token) and emit its BODY afterward via a new EmitWorkerThreadsReceiveMessageOnPortBody, wired right after EmitMessageChannelTypes and before EmitRuntimeClassFinalize. The body drains one message from the port's own _pending queue and returns { message }, else undefined - matching SharpTSMessagePort.ReceiveMessageSync. 2. messageerror on postMessage of an uncloneable value (#1001 follow-up) The emitted structured clone (cloneCore) returns uncloneable values by reference and never throws, so a compiled $MessagePort had no clone-failure point. $MessagePort.PostMessage now detects the uncloneable categories up front (typeof "function"/"symbol") and enqueues a shared static _cloneError sentinel; Drain compares by reference and emits 'messageerror' (no args) instead of 'message'. This mirrors the interpreter's receiver-side model (ClonedMessage IsError: true) without touching the shared cloneCore, so structuredClone/$BroadcastChannel are unaffected. Both paths stay fully standalone (pure emitted IL, no SharpTS.dll dependency; --standalone --verify clean). Deeply nested uncloneables remain aliased, consistent with the emitted clone's existing shallow fallback. Tests: ReceiveMessageOnPort_QueuedMessage_ReturnsMessageThenUndefined (new, dual-mode) and MessageChannelPort_PostUncloneable_FiresMessageError (was interpreter-only Fact -> dual-mode Theory). xUnit 15428/0, Test262 22/0/4skip, TSConf 31/0. --- Compilation/RuntimeEmitter.MessageChannel.cs | 73 ++++++++++++++- Compilation/RuntimeEmitter.Worker.cs | 90 ++++++++++++++++--- Compilation/RuntimeEmitter.cs | 5 ++ .../SharedTests/WorkerThreadsTests.cs | 57 +++++++++--- 4 files changed, 202 insertions(+), 23 deletions(-) diff --git a/Compilation/RuntimeEmitter.MessageChannel.cs b/Compilation/RuntimeEmitter.MessageChannel.cs index 4556ba51..9475ccf0 100644 --- a/Compilation/RuntimeEmitter.MessageChannel.cs +++ b/Compilation/RuntimeEmitter.MessageChannel.cs @@ -32,6 +32,11 @@ public partial class RuntimeEmitter private FieldBuilder _messagePortClosedField = null!; private FieldBuilder _messagePortRefedField = null!; private FieldBuilder _messagePortOnEnqueueField = null!; + // Shared sentinel enqueued (in place of a clone) when postMessage receives an + // uncloneable value. Drain turns it into a 'messageerror' event, mirroring the + // interpreter's ClonedMessage(IsError: true) path (#1077). One instance per + // emitted assembly, created by $MessagePort's static ctor. + private FieldBuilder _messagePortCloneErrorField = null!; private ConstructorBuilder _messagePortCtor = null!; private MethodBuilder _messagePortDrain = null!; private MethodBuilder _messagePortStart = null!; @@ -66,6 +71,18 @@ private void EmitMessagePortClass(ModuleBuilder moduleBuilder, EmittedRuntime ru // to drain _pending event-driven instead of the worker polling (#465). _messagePortOnEnqueueField = typeBuilder.DefineField("_onEnqueue", typeof(Action), FieldAttributes.Assembly); + // Static clone-failure sentinel (#1077). A single shared instance is enqueued in + // place of a message whose value cannot be structured-cloned; Drain compares by + // reference and emits 'messageerror' instead of 'message'. + _messagePortCloneErrorField = typeBuilder.DefineField( + "_cloneError", + _types.Object, + FieldAttributes.Assembly | FieldAttributes.Static | FieldAttributes.InitOnly); + var cctorIl = typeBuilder.DefineTypeInitializer().GetILGenerator(); + cctorIl.Emit(OpCodes.Newobj, _types.GetDefaultConstructor(_types.Object)); + cctorIl.Emit(OpCodes.Stsfld, _messagePortCloneErrorField); + cctorIl.Emit(OpCodes.Ret); + EmitMessagePortConstructorIl(typeBuilder, runtime); EmitMessagePortDrain(typeBuilder, runtime); EmitMessagePortRef(typeBuilder, runtime); @@ -143,6 +160,22 @@ private void EmitMessagePortDrain(TypeBuilder typeBuilder, EmittedRuntime runtim il.Emit(OpCodes.Callvirt, _types.ConcurrentQueueOfObject.GetMethod("TryDequeue", [_types.Object.MakeByRefType()])!); il.Emit(OpCodes.Brfalse, exitLabel); + // A clone-failure sentinel (from postMessage of an uncloneable value) is + // delivered as 'messageerror' (no args), not 'message' — Node's receiver-side + // model (#1077). ReferenceEquals(msg, _cloneError). + var notCloneErrorLabel = il.DefineLabel(); + il.Emit(OpCodes.Ldloc, msgLocal); + il.Emit(OpCodes.Ldsfld, _messagePortCloneErrorField); + il.Emit(OpCodes.Bne_Un, notCloneErrorLabel); + il.Emit(OpCodes.Ldarg_0); + il.Emit(OpCodes.Ldstr, "messageerror"); + il.Emit(OpCodes.Ldc_I4_0); + il.Emit(OpCodes.Newarr, _types.Object); + il.Emit(OpCodes.Callvirt, runtime.TSEventEmitterEmit); + il.Emit(OpCodes.Pop); + il.Emit(OpCodes.Br, loopTop); + il.MarkLabel(notCloneErrorLabel); + // this.Emit("message", new object[1] { msg }) il.Emit(OpCodes.Ldc_I4_1); il.Emit(OpCodes.Newarr, _types.Object); @@ -263,6 +296,7 @@ private void EmitMessagePortPostMessage(TypeBuilder typeBuilder, EmittedRuntime var exitLabel = il.DefineLabel(); var partnerLocal = il.DeclareLocal(typeBuilder); var clonedLocal = il.DeclareLocal(_types.Object); + var typeofLocal = il.DeclareLocal(_types.String); // if (_closed) return il.Emit(OpCodes.Ldarg_0); @@ -280,11 +314,48 @@ private void EmitMessagePortPostMessage(TypeBuilder typeBuilder, EmittedRuntime il.Emit(OpCodes.Ldfld, _messagePortClosedField); il.Emit(OpCodes.Brtrue, exitLabel); - // cloned = $Runtime.StructuredClone(msg, null) + // Node model: an uncloneable value is NOT thrown back on the sender — the receiver + // fires 'messageerror' (#1077). The emitted $Runtime.StructuredClone returns + // uncloneable values by reference (it has no throw path), so detect the uncloneable + // categories up front — functions/classes/callable wrappers (typeof "function") and + // symbols (typeof "symbol") — and enqueue the shared _cloneError sentinel instead of + // a clone. Drain converts it to 'messageerror'. This mirrors the try/catch around + // StructuredClone.Clone in SharpTSMessagePort.PostMessage. (Deeply nested uncloneables + // remain aliased, matching the emitted clone's existing shallow fallback.) + var markerLabel = il.DefineLabel(); + var haveValueLabel = il.DefineLabel(); + var strEquals = _types.GetMethod(_types.String, "op_Equality", _types.String, _types.String); + + // t = TypeOf(msg) + il.Emit(OpCodes.Ldarg_1); + il.Emit(OpCodes.Call, runtime.TypeOf); + il.Emit(OpCodes.Stloc, typeofLocal); + + // if (t == "function") goto markerLabel + il.Emit(OpCodes.Ldloc, typeofLocal); + il.Emit(OpCodes.Ldstr, "function"); + il.Emit(OpCodes.Call, strEquals); + il.Emit(OpCodes.Brtrue, markerLabel); + + // if (t == "symbol") goto markerLabel + il.Emit(OpCodes.Ldloc, typeofLocal); + il.Emit(OpCodes.Ldstr, "symbol"); + il.Emit(OpCodes.Call, strEquals); + il.Emit(OpCodes.Brtrue, markerLabel); + + // cloned = $Runtime.StructuredClone(msg, null); goto haveValue il.Emit(OpCodes.Ldarg_1); il.Emit(OpCodes.Ldnull); il.Emit(OpCodes.Call, runtime.StructuredCloneClone); il.Emit(OpCodes.Stloc, clonedLocal); + il.Emit(OpCodes.Br, haveValueLabel); + + // cloned = _cloneError sentinel + il.MarkLabel(markerLabel); + il.Emit(OpCodes.Ldsfld, _messagePortCloneErrorField); + il.Emit(OpCodes.Stloc, clonedLocal); + + il.MarkLabel(haveValueLabel); // partner._pending.Enqueue(cloned) il.Emit(OpCodes.Ldloc, partnerLocal); diff --git a/Compilation/RuntimeEmitter.Worker.cs b/Compilation/RuntimeEmitter.Worker.cs index d7018aeb..7887cc09 100644 --- a/Compilation/RuntimeEmitter.Worker.cs +++ b/Compilation/RuntimeEmitter.Worker.cs @@ -1704,6 +1704,10 @@ private void EmitStructuredCloneHelper(TypeBuilder runtimeType, EmittedRuntime r runtime.StructuredCloneClone = method; } + // WorkerThreadsReceiveMessageOnPort: defined during EmitWorkerThreadsModuleHelpers, body + // emitted later by EmitWorkerThreadsReceiveMessageOnPortBody once $MessagePort exists (#1077). + private MethodBuilder _receiveMessageOnPortMethod = null!; + /// /// Emits worker_threads module helper methods. /// Uses direct calls. @@ -1736,23 +1740,19 @@ private void EmitWorkerThreadsModuleHelpers(TypeBuilder runtimeType, EmittedRunt il2.Emit(OpCodes.Ret); runtime.WorkerThreadsThreadId = threadIdMethod; - // receiveMessageOnPort — main-thread compiled stub. The $MessagePort type is emitted - // AFTER this method (EmitMessageChannelTypes runs after EmitRuntimeClass), so this - // helper cannot read the port's queue. It returns `undefined` (Node's empty-port - // result) rather than null (#1000). Synchronously draining a $MessagePort on the main - // thread in compiled mode remains unimplemented (a worker drives receiveMessageOnPort - // through the interpreter, which is fully functional). - var receiveMethod = runtimeType.DefineMethod( + // receiveMessageOnPort — synchronous main-thread drain (#1077). The method is DEFINED + // here (so callers can bind runtime.WorkerThreadsReceiveMessageOnPort) but its body is + // emitted later by EmitWorkerThreadsReceiveMessageOnPortBody, after EmitMessageChannelTypes + // has created the $MessagePort type — this helper reads that type's _pending queue and + // _closed/_cloneError fields, which don't exist at this point in emission. The $Runtime + // type isn't finalized until EmitRuntimeClassFinalize, so filling the body afterward is safe. + _receiveMessageOnPortMethod = runtimeType.DefineMethod( "WorkerThreadsReceiveMessageOnPort", MethodAttributes.Public | MethodAttributes.Static, _types.Object, [_types.Object] ); - - var il3 = receiveMethod.GetILGenerator(); - il3.Emit(OpCodes.Ldsfld, runtime.UndefinedInstance); - il3.Emit(OpCodes.Ret); - runtime.WorkerThreadsReceiveMessageOnPort = receiveMethod; + runtime.WorkerThreadsReceiveMessageOnPort = _receiveMessageOnPortMethod; // getEnvironmentData / setEnvironmentData — route to the C# per-process // WorkerEnvironmentData store via reflection (worker programs co-locate SharpTS.dll; @@ -1761,6 +1761,72 @@ private void EmitWorkerThreadsModuleHelpers(TypeBuilder runtimeType, EmittedRunt EmitWorkerThreadsEnvironmentData(runtimeType, runtime); } + /// + /// Emits the body of WorkerThreadsReceiveMessageOnPort (#1077). Must be called AFTER + /// (so the $MessagePort type and its + /// _pending/_closed/_cloneError fields exist) and BEFORE + /// (which creates the $Runtime type). + /// + /// Synchronously drains one message from the port's own queue, matching + /// SharpTSMessagePort.ReceiveMessageSync: returns { message } when a value is + /// queued, or undefined when the argument is not a port, the port is closed, or the + /// queue is empty. A clone-failure sentinel dequeues as { message: undefined }. + /// + private void EmitWorkerThreadsReceiveMessageOnPortBody(EmittedRuntime runtime) + { + var il = _receiveMessageOnPortMethod.GetILGenerator(); + var portLocal = il.DeclareLocal(_messagePortType); + var msgLocal = il.DeclareLocal(_types.Object); + var valueLocal = il.DeclareLocal(_types.Object); + var dictLocal = il.DeclareLocal(_types.DictionaryStringObject); + var undefinedLabel = il.DefineLabel(); + var afterMarkerLabel = il.DefineLabel(); + + // port = arg0 as $MessagePort; if (port == null) return undefined + il.Emit(OpCodes.Ldarg_0); + il.Emit(OpCodes.Isinst, _messagePortType); + il.Emit(OpCodes.Stloc, portLocal); + il.Emit(OpCodes.Ldloc, portLocal); + il.Emit(OpCodes.Brfalse, undefinedLabel); + + // if (port._closed) return undefined + il.Emit(OpCodes.Ldloc, portLocal); + il.Emit(OpCodes.Ldfld, _messagePortClosedField); + il.Emit(OpCodes.Brtrue, undefinedLabel); + + // if (!port._pending.TryDequeue(out msg)) return undefined + il.Emit(OpCodes.Ldloc, portLocal); + il.Emit(OpCodes.Ldfld, _messagePortPendingField); + il.Emit(OpCodes.Ldloca, msgLocal); + il.Emit(OpCodes.Callvirt, _types.ConcurrentQueueOfObject.GetMethod("TryDequeue", [_types.Object.MakeByRefType()])!); + il.Emit(OpCodes.Brfalse, undefinedLabel); + + // value = msg; if (msg == _cloneError) value = undefined (a queued clone-failure + // sentinel has no cloneable payload — surface it as { message: undefined }). + il.Emit(OpCodes.Ldloc, msgLocal); + il.Emit(OpCodes.Stloc, valueLocal); + il.Emit(OpCodes.Ldloc, msgLocal); + il.Emit(OpCodes.Ldsfld, _messagePortCloneErrorField); + il.Emit(OpCodes.Bne_Un, afterMarkerLabel); + il.Emit(OpCodes.Ldsfld, runtime.UndefinedInstance); + il.Emit(OpCodes.Stloc, valueLocal); + il.MarkLabel(afterMarkerLabel); + + // return new Dictionary() { ["message"] = value } + il.Emit(OpCodes.Newobj, _types.DictionaryStringObjectCtor); + il.Emit(OpCodes.Stloc, dictLocal); + il.Emit(OpCodes.Ldloc, dictLocal); + il.Emit(OpCodes.Ldstr, "message"); + il.Emit(OpCodes.Ldloc, valueLocal); + il.Emit(OpCodes.Callvirt, _types.DictionaryStringObjectSetItem); + il.Emit(OpCodes.Ldloc, dictLocal); + il.Emit(OpCodes.Ret); + + il.MarkLabel(undefinedLabel); + il.Emit(OpCodes.Ldsfld, runtime.UndefinedInstance); + il.Emit(OpCodes.Ret); + } + /// /// Emits the getEnvironmentData/setEnvironmentData runtime helpers that reflectively call /// SharpTS.Runtime.Types.WorkerEnvironmentData. diff --git a/Compilation/RuntimeEmitter.cs b/Compilation/RuntimeEmitter.cs index cb118aa4..fdd0785a 100644 --- a/Compilation/RuntimeEmitter.cs +++ b/Compilation/RuntimeEmitter.cs @@ -438,6 +438,11 @@ public EmittedRuntime EmitAll(ModuleBuilder moduleBuilder, RuntimeFeatureSet fea // NOTE: Must stay in sync with SharpTS.Runtime.Types.SharpTSMessagePort EmitMessageChannelTypes(moduleBuilder, runtime); + // receiveMessageOnPort's body reads $MessagePort's _pending/_closed/_cloneError, so it + // must be filled now that EmitMessageChannelTypes has created the type (#1077). Still + // before EmitRuntimeClassFinalize, which closes the $Runtime type this method lives on. + EmitWorkerThreadsReceiveMessageOnPortBody(runtime); + // Web Streams — gated on UsesWebStreams. The only external references are // user-code `new ReadableStream(...)`/`new WritableStream(...)`/`new TransformStream(...)` // in ExpressionEmitterBase.Constructors.cs, which only fire when the diff --git a/SharpTS.Tests/SharedTests/WorkerThreadsTests.cs b/SharpTS.Tests/SharedTests/WorkerThreadsTests.cs index 34405aab..a8b8b990 100644 --- a/SharpTS.Tests/SharedTests/WorkerThreadsTests.cs +++ b/SharpTS.Tests/SharedTests/WorkerThreadsTests.cs @@ -769,6 +769,37 @@ public void ReceiveMessageOnPort_EmptyPort_IsUndefined(ExecutionMode mode) Assert.Contains("empty:true", output); } + /// + /// #1077: receiveMessageOnPort synchronously drains a queued message from a + /// main-thread MessageChannel port, returning { message }; a second call on + /// the now-empty port returns undefined. Dual-mode — the compiled helper reads the + /// emitted $MessagePort queue directly (it was previously a stub that always returned + /// undefined). + /// + [Theory] + [MemberData(nameof(ExecutionModes.All), MemberType = typeof(ExecutionModes))] + public void ReceiveMessageOnPort_QueuedMessage_ReturnsMessageThenUndefined(ExecutionMode mode) + { + var files = new Dictionary + { + ["main.ts"] = """ + import { MessageChannel, receiveMessageOnPort } from "worker_threads"; + const { port1, port2 } = new MessageChannel(); + // port2 has no 'message' listener, so it stays unstarted and the posted + // value waits in its queue for a synchronous receiveMessageOnPort drain. + port1.postMessage("hello"); + const r: any = receiveMessageOnPort(port2); + console.log("first:" + (r === undefined ? "undef" : r.message)); + const r2: any = receiveMessageOnPort(port2); + console.log("second-empty:" + (r2 === undefined)); + """ + }; + + var output = TestHarness.RunModules(files, "main.ts", mode); + Assert.Contains("first:hello", output); + Assert.Contains("second-empty:true", output); + } + #endregion #region introspection (#1004) @@ -1122,27 +1153,33 @@ public void Worker_ParentPostsUncloneable_WorkerParentPortFiresMessageError(Exec } /// - /// #1001: a MessageChannel port whose peer posts an uncloneable value fires - /// 'messageerror' on the receiver. Interpreter only — the compiled emitted - /// structured clone returns uncloneable values by reference (it does not throw), so a - /// compiled $MessagePort has no clone-failure point. The Worker paths above cover - /// the dual-mode behavior. + /// #1001/#1077: a MessageChannel port whose peer posts an uncloneable value (a + /// function) fires 'messageerror' — not 'message' — on the receiver. Now + /// dual-mode: the compiled $MessagePort.postMessage detects the uncloneable value + /// (typeof "function"/"symbol") up front and enqueues a clone-failure sentinel that + /// Drain converts to 'messageerror', matching the interpreter's + /// receiver-side model (previously compiled returned the function by reference and fired + /// 'message'). /// - [Fact] - public void MessageChannelPort_PostUncloneable_FiresMessageError_Interpreted() + [Theory] + [MemberData(nameof(ExecutionModes.All), MemberType = typeof(ExecutionModes))] + public void MessageChannelPort_PostUncloneable_FiresMessageError(ExecutionMode mode) { var files = new Dictionary { ["main.ts"] = """ import { MessageChannel } from "worker_threads"; const { port1, port2 } = new MessageChannel(); - port2.on("messageerror", () => { console.log("port-err"); }); - port2.on("message", () => { console.log("port-msg"); }); + // Close both ports after delivery so the compiled $MessagePort (which Refs the + // event loop on start) lets the process quiesce — same convention as the other + // compiled MessageChannel tests. + port2.on("messageerror", () => { console.log("port-err"); port1.close(); port2.close(); }); + port2.on("message", () => { console.log("port-msg"); port1.close(); port2.close(); }); port1.postMessage(() => {}); """ }; - var output = TestHarness.RunModules(files, "main.ts", ExecutionMode.Interpreted); + var output = TestHarness.RunModules(files, "main.ts", mode); Assert.Contains("port-err", output); Assert.DoesNotContain("port-msg", output); }