Skip to content

feat(worker_threads): compiled main-thread receiveMessageOnPort drain + MessageChannel messageerror (#1077) #1253

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nickna merged 1 commit into from
Jul 5, 2026
Merged

feat(worker_threads): compiled main-thread receiveMessageOnPort drain + MessageChannel messageerror (#1077) #1253

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 72 additions & 1 deletion Compilation/RuntimeEmitter.MessageChannel.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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!;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand Down
90 changes: 78 additions & 12 deletions Compilation/RuntimeEmitter.Worker.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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!;

/// <summary>
/// Emits worker_threads module helper methods.
/// Uses direct calls.
Expand Down Expand Up @@ -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;
Expand All @@ -1761,6 +1761,72 @@ private void EmitWorkerThreadsModuleHelpers(TypeBuilder runtimeType, EmittedRunt
EmitWorkerThreadsEnvironmentData(runtimeType, runtime);
}

/// <summary>
/// Emits the body of <c>WorkerThreadsReceiveMessageOnPort</c> (#1077). Must be called AFTER
/// <see cref="EmitMessageChannelTypes"/> (so the <c>$MessagePort</c> type and its
/// <c>_pending</c>/<c>_closed</c>/<c>_cloneError</c> fields exist) and BEFORE
/// <see cref="EmitRuntimeClassFinalize"/> (which creates the <c>$Runtime</c> type).
///
/// Synchronously drains one message from the port's own queue, matching
/// <c>SharpTSMessagePort.ReceiveMessageSync</c>: returns <c>{ message }</c> when a value is
/// queued, or <c>undefined</c> when the argument is not a port, the port is closed, or the
/// queue is empty. A clone-failure sentinel dequeues as <c>{ message: undefined }</c>.
/// </summary>
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<string, object>() { ["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);
}

/// <summary>
/// Emits the getEnvironmentData/setEnvironmentData runtime helpers that reflectively call
/// <c>SharpTS.Runtime.Types.WorkerEnvironmentData</c>.
Expand Down
5 changes: 5 additions & 0 deletions Compilation/RuntimeEmitter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
57 changes: 47 additions & 10 deletions SharpTS.Tests/SharedTests/WorkerThreadsTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -769,6 +769,37 @@ public void ReceiveMessageOnPort_EmptyPort_IsUndefined(ExecutionMode mode)
Assert.Contains("empty:true", output);
}

/// <summary>
/// #1077: <c>receiveMessageOnPort</c> synchronously drains a queued message from a
/// main-thread <c>MessageChannel</c> port, returning <c>{ message }</c>; a second call on
/// the now-empty port returns <c>undefined</c>. Dual-mode — the compiled helper reads the
/// emitted <c>$MessagePort</c> queue directly (it was previously a stub that always returned
/// <c>undefined</c>).
/// </summary>
[Theory]
[MemberData(nameof(ExecutionModes.All), MemberType = typeof(ExecutionModes))]
public void ReceiveMessageOnPort_QueuedMessage_ReturnsMessageThenUndefined(ExecutionMode mode)
{
var files = new Dictionary<string, string>
{
["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)
Expand Down Expand Up @@ -1122,27 +1153,33 @@ public void Worker_ParentPostsUncloneable_WorkerParentPortFiresMessageError(Exec
}

/// <summary>
/// #1001: a <c>MessageChannel</c> port whose peer posts an uncloneable value fires
/// <c>'messageerror'</c> on the receiver. Interpreter only — the compiled emitted
/// structured clone returns uncloneable values by reference (it does not throw), so a
/// compiled <c>$MessagePort</c> has no clone-failure point. The Worker paths above cover
/// the dual-mode behavior.
/// #1001/#1077: a <c>MessageChannel</c> port whose peer posts an uncloneable value (a
/// function) fires <c>'messageerror'</c> — not <c>'message'</c> — on the receiver. Now
/// dual-mode: the compiled <c>$MessagePort.postMessage</c> detects the uncloneable value
/// (<c>typeof</c> "function"/"symbol") up front and enqueues a clone-failure sentinel that
/// <c>Drain</c> converts to <c>'messageerror'</c>, matching the interpreter's
/// receiver-side model (previously compiled returned the function by reference and fired
/// <c>'message'</c>).
/// </summary>
[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<string, string>
{
["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);
}
Expand Down
Loading