node:worker_threads: implement postMessageToThread [1bx5ty]

src/bun.js/bindings/ErrorCode.cpp

@@ -2390,6 +2390,14 @@ JSC_DEFINE_HOST_FUNCTION(Bun::jsFunctionMakeErrorWithCode, (JSC::JSGlobalObject
         return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_IPC_CHANNEL_CLOSED, "Channel closed."_s));
     case ErrorCode::ERR_SOCKET_BAD_TYPE:
         return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_SOCKET_BAD_TYPE, "Bad socket type specified. Valid types are: udp4, udp6"_s));
+    case ErrorCode::ERR_WORKER_MESSAGING_ERRORED:
+        return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_WORKER_MESSAGING_ERRORED, "The destination thread threw an error while processing the message"_s));
+    case ErrorCode::ERR_WORKER_MESSAGING_FAILED:
+        return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_WORKER_MESSAGING_FAILED, "Cannot find the destination thread or listener"_s));
+    case ErrorCode::ERR_WORKER_MESSAGING_SAME_THREAD:
+        return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_WORKER_MESSAGING_SAME_THREAD, "Cannot sent a message to the same thread"_s));
+    case ErrorCode::ERR_WORKER_MESSAGING_TIMEOUT:
+        return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_WORKER_MESSAGING_TIMEOUT, "Sending a message to another thread timed out"_s));
     case ErrorCode::ERR_ZLIB_INITIALIZATION_FAILED:
         return JSC::JSValue::encode(createError(globalObject, ErrorCode::ERR_ZLIB_INITIALIZATION_FAILED, "Initialization failed"_s));
     case ErrorCode::ERR_IPC_ONE_PIPE:

src/bun.js/bindings/ErrorCode.ts

@@ -271,6 +271,10 @@ const errors: ErrorCodeMapping = [
   ["ERR_WASI_NOT_STARTED", Error],
   ["ERR_WEBASSEMBLY_RESPONSE", TypeError],
   ["ERR_WORKER_INIT_FAILED", Error],
+  ["ERR_WORKER_MESSAGING_ERRORED", Error],
+  ["ERR_WORKER_MESSAGING_FAILED", Error],
+  ["ERR_WORKER_MESSAGING_SAME_THREAD", Error],
+  ["ERR_WORKER_MESSAGING_TIMEOUT", Error],
   ["ERR_WORKER_NOT_RUNNING", Error],
   ["ERR_ZLIB_INITIALIZATION_FAILED", Error],
   ["MODULE_NOT_FOUND", Error],

src/bun.js/bindings/ScriptExecutionContext.cpp

@@ -134,13 +134,21 @@ ScriptExecutionContext::~ScriptExecutionContext()
         Locker locker { allScriptExecutionContextsMapLock };
         ASSERT_WITH_MESSAGE(!allScriptExecutionContextsMap().contains(m_identifier), "A ScriptExecutionContext subclass instance implementing postTask should have already removed itself from the map");
     }
-    m_inScriptExecutionContextDestructor = true;
 #endif // ASSERT_ENABLED
 
+    // Draining these handlers may drop the last reference to a MessagePort whose JS wrapper is
+    // already gone (e.g. when a worker is terminated while a port still has a pending
+    // processMessageWithMessagePortsSoon callback). Do this before setting
+    // m_inScriptExecutionContextDestructor so the observer can unregister itself without
+    // tripping the assertion below; takeAny() makes the subsequent loop safe regardless.
     auto postMessageCompletionHandlers = WTF::move(m_processMessageWithMessagePortsSoonHandlers);
     for (auto& completionHandler : postMessageCompletionHandlers)
         completionHandler();
 
+#if ASSERT_ENABLED
+    m_inScriptExecutionContextDestructor = true;
+#endif // ASSERT_ENABLED
+
     while (auto* destructionObserver = m_destructionObservers.takeAny())
         destructionObserver->contextDestroyed();
 

src/bun.js/bindings/webcore/EventEmitter.cpp

@@ -116,8 +116,7 @@ bool EventEmitter::emitForBindings(const Identifier& eventType, const MarkedArgu
     if (!scriptExecutionContext())
         return false;
 
-    emit(eventType, arguments);
-    return true;
+    return emit(eventType, arguments);
 }
 
 bool EventEmitter::emit(const Identifier& eventType, const MarkedArgumentBuffer& arguments)

src/bun.js/bindings/webcore/JSMessagePort.cpp

@@ -231,7 +231,7 @@
     vm.writeBarrier(&thisObject, value);
     ensureStillAliveHere(value);
 
-    thisObject.wrapped().jsRef(&lexicalGlobalObject);
+    // Unlike onmessage, Node.js does not ref the port when only onmessageerror is set.
 
     return true;
 }

src/bun.js/bindings/webcore/JSWorker.cpp

@@ -145,8 +145,10 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSWorkerDOMConstructor::
 
     WorkerOptions options {};
     JSValue nodeWorkerObject {};
-    if (callFrame->argumentCount() == 3) {
+    JSValue mainThreadPort {};
+    if (callFrame->argumentCount() >= 3) {
         nodeWorkerObject = callFrame->argument(2);
+        mainThreadPort = callFrame->argument(3);
         options.kind = WorkerOptions::Kind::Node;
     }
     JSValue workerData = jsUndefined();
@@ -307,13 +309,22 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSWorkerDOMConstructor::
         }
     }
 
+    // The internal port linking this worker to the main thread for worker_threads.postMessageToThread.
+    // Only present when constructed via node:worker_threads.
+    if (mainThreadPort && mainThreadPort.isObject()) {
+        transferList.append({ vm, mainThreadPort.getObject() });
+    } else {
+        mainThreadPort = jsUndefined();
+    }
+
     Vector<RefPtr<MessagePort>> ports;
-    auto* valueToTransfer = constructEmptyArray(globalObject, nullptr, 2);
+    auto* valueToTransfer = constructEmptyArray(globalObject, nullptr, 3);
     RETURN_IF_EXCEPTION(throwScope, {});
     valueToTransfer->putDirectIndex(globalObject, 0, workerData);
     auto* environmentData = globalObject->nodeWorkerEnvironmentData();
     // If node:worker_threads has not been imported, environment data will not be set up yet.
     valueToTransfer->putDirectIndex(globalObject, 1, environmentData ? environmentData : jsUndefined());
+    valueToTransfer->putDirectIndex(globalObject, 2, mainThreadPort);
 
     ExceptionOr<Ref<SerializedScriptValue>> serialized = SerializedScriptValue::create(*lexicalGlobalObject, valueToTransfer, WTF::move(transferList), ports, SerializationForStorage::No, SerializationContext::WorkerPostMessage);
     if (serialized.hasException()) {

src/bun.js/bindings/webcore/MessagePort.cpp

@@ -45,8 +45,6 @@
 #include <wtf/Lock.h>
 #include <wtf/Scope.h>
 
-extern "C" void Bun__eventLoop__incrementRefConcurrently(void* bunVM, int delta);
-
 namespace WebCore {
 
 WTF_MAKE_TZONE_ALLOCATED_IMPL(MessagePort);
@@ -241,6 +239,7 @@ void MessagePort::close()
     MessagePortChannelProvider::singleton().messagePortClosed(m_identifier);
 
     removeAllEventListeners();
+    updateEventLoopRef();
 }
 
 void MessagePort::dispatchMessages()
@@ -383,36 +382,40 @@ void MessagePort::contextDestroyed()
     // ActiveDOMObject::contextDestroyed();
 }
 
+void MessagePort::updateEventLoopRef()
+{
+    bool shouldRef = m_hasRef && (m_messageEventCount > 0 || m_wantsExplicitRef) && !m_isDetached;
+    if (shouldRef == m_isRefingEventLoop)
+        return;
+    auto* context = scriptExecutionContext();
+    if (!context) {
+        m_isRefingEventLoop = false;
+        return;
+    }
+    m_isRefingEventLoop = shouldRef;
+    if (shouldRef)
+        context->refEventLoop();
+    else
+        context->unrefEventLoop();
+}
+
 void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind)
 {
     if (eventType == eventNames().messageEvent) {
         auto& port = static_cast<MessagePort&>(self);
         switch (kind) {
         case Add:
-            if (port.m_messageEventCount == 0) {
-                auto* context = port.scriptExecutionContext();
-                if (context)
-                    context->refEventLoop();
-            }
             port.m_messageEventCount++;
             break;
         case Remove:
-            port.m_messageEventCount--;
-            if (port.m_messageEventCount == 0) {
-                auto* context = port.scriptExecutionContext();
-                if (context)
-                    context->unrefEventLoop();
-            }
+            if (port.m_messageEventCount > 0)
+                port.m_messageEventCount--;
             break;
         case Clear:
-            if (port.m_messageEventCount > 0) {
-                auto* context = port.scriptExecutionContext();
-                if (context)
-                    context->unrefEventLoop();
-            }
             port.m_messageEventCount = 0;
             break;
         }
+        port.updateEventLoopRef();
     }
 };
 
@@ -453,22 +456,18 @@ WebCoreOpaqueRoot root(MessagePort* port)
     return WebCoreOpaqueRoot { port };
 }
 
-void MessagePort::jsRef(JSGlobalObject* lexicalGlobalObject)
+void MessagePort::jsRef(JSGlobalObject*)
 {
-    if (!m_hasRef) {
-        m_hasRef = true;
-        ref();
-        Bun__eventLoop__incrementRefConcurrently(WebCore::clientData(lexicalGlobalObject->vm())->bunVM, 1);
-    }
+    m_hasRef = true;
+    m_wantsExplicitRef = true;
+    updateEventLoopRef();
 }
 
-void MessagePort::jsUnref(JSGlobalObject* lexicalGlobalObject)
+void MessagePort::jsUnref(JSGlobalObject*)
 {
-    if (m_hasRef) {
-        m_hasRef = false;
-        deref();
-        Bun__eventLoop__incrementRefConcurrently(WebCore::clientData(lexicalGlobalObject->vm())->bunVM, -1);
-    }
+    m_hasRef = false;
+    m_wantsExplicitRef = false;
+    updateEventLoopRef();
 }
 
 } // namespace WebCore

src/bun.js/bindings/webcore/MessagePort.h

@@ -113,7 +113,7 @@ class MessagePort final : /* public ActiveDOMObject, */ public ContextDestructio
 
     void jsRef(JSGlobalObject*);
     void jsUnref(JSGlobalObject*);
-    bool jsHasRef() { return m_hasRef; }
+    bool jsHasRef() { return m_isRefingEventLoop; }
 
 private:
     explicit MessagePort(ScriptExecutionContext&, const MessagePortIdentifier& local, const MessagePortIdentifier& remote);
@@ -139,9 +139,18 @@ class MessagePort final : /* public ActiveDOMObject, */ public ContextDestructio
 
     mutable std::atomic<unsigned> m_refCount { 1 };
 
-    bool m_hasRef { false };
+    // Whether this port should keep the event loop alive when it is active.
+    // Toggled by ref()/unref() from JS; unref() wins over everything else.
+    bool m_hasRef { true };
+    // Whether jsRef() has been called (from .ref() or the onmessage setter). This is one of the
+    // ways a port becomes "active" for event-loop-ref purposes; the other is m_messageEventCount,
+    // which is only tracked for ports whose onDidChangeListener is wired (transferred ports).
+    bool m_wantsExplicitRef { false };
+    // Whether this port is currently holding a ref on the event loop.
+    bool m_isRefingEventLoop { false };
 
     uint32_t m_messageEventCount { 0 };
+    void updateEventLoopRef();
     static void onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind);
 };
 

src/bun.js/bindings/webcore/Worker.cpp

@@ -584,6 +584,7 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject)
     auto scope = DECLARE_THROW_SCOPE(globalObject->vm());
     JSValue workerData = jsNull();
     JSValue threadId = jsNumber(0);
+    JSValue mainThreadPort = jsUndefined();
     JSMap* environmentData = nullptr;
 
     if (auto* worker = WebWorker__getParentWorker(globalObject->bunVM())) {
@@ -592,17 +593,20 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject)
         RefPtr<WebCore::SerializedScriptValue> serialized = WTF::move(options.workerDataAndEnvironmentData);
         JSValue deserialized = serialized->deserialize(*globalObject, globalObject, WTF::move(ports));
         RETURN_IF_EXCEPTION(scope, {});
-        // Should always be set to an Array of length 2 in the constructor in JSWorker.cpp
+        // Should always be set to an Array of length 3 in the constructor in JSWorker.cpp
         auto* pair = uncheckedDowncast<JSArray>(deserialized);
-        ASSERT(pair->length() == 2);
+        ASSERT(pair->length() == 3);
         ASSERT(pair->canGetIndexQuickly(0u));
         ASSERT(pair->canGetIndexQuickly(1u));
+        ASSERT(pair->canGetIndexQuickly(2u));
         workerData = pair->getIndexQuickly(0);
         RETURN_IF_EXCEPTION(scope, {});
         auto environmentDataValue = pair->getIndexQuickly(1);
         // it might not be a Map if the parent had not set up environmentData yet
         environmentData = environmentDataValue ? dynamicDowncast<JSMap>(environmentDataValue) : nullptr;
         RETURN_IF_EXCEPTION(scope, {});
+        mainThreadPort = pair->getIndexQuickly(2);
+        RETURN_IF_EXCEPTION(scope, {});
 
         // Main thread starts at 1
         threadId = jsNumber(worker->clientIdentifier() - 1);
@@ -614,12 +618,13 @@ JSValue createNodeWorkerThreadsBinding(Zig::GlobalObject* globalObject)
     ASSERT(environmentData);
     globalObject->setNodeWorkerEnvironmentData(environmentData);
 
-    JSObject* array = constructEmptyArray(globalObject, nullptr, 4);
+    JSObject* array = constructEmptyArray(globalObject, nullptr, 5);
     RETURN_IF_EXCEPTION(scope, {});
     array->putDirectIndex(globalObject, 0, workerData);
     array->putDirectIndex(globalObject, 1, threadId);
     array->putDirectIndex(globalObject, 2, JSFunction::create(vm, globalObject, 1, "receiveMessageOnPort"_s, jsReceiveMessageOnPort, ImplementationVisibility::Public, NoIntrinsic));
     array->putDirectIndex(globalObject, 3, environmentData);
+    array->putDirectIndex(globalObject, 4, mainThreadPort);
     return array;
 }
 

src/js/builtins.d.ts

@@ -710,6 +710,10 @@ declare function $ERR_IPC_DISCONNECTED(): Error;
 declare function $ERR_SERVER_NOT_RUNNING(): Error;
 declare function $ERR_IPC_CHANNEL_CLOSED(): Error;
 declare function $ERR_SOCKET_BAD_TYPE(): Error;
+declare function $ERR_WORKER_MESSAGING_ERRORED(): Error;
+declare function $ERR_WORKER_MESSAGING_FAILED(): Error;
+declare function $ERR_WORKER_MESSAGING_SAME_THREAD(): Error;
+declare function $ERR_WORKER_MESSAGING_TIMEOUT(): Error;
 declare function $ERR_ZLIB_INITIALIZATION_FAILED(): Error;
 declare function $ERR_IPC_ONE_PIPE(): Error;
 declare function $ERR_SOCKET_ALREADY_BOUND(): Error;