Replace MessagePort/BroadcastChannel registries with MessagePortPipe primitive

src/bun.js/bindings/BunWorkerGlobalScope.cpp

@@ -1,18 +1,12 @@
 #include "config.h"
 
 #include "BunWorkerGlobalScope.h"
-#include "MessagePortChannelProviderImpl.h"
 #include <wtf/TZoneMallocInlines.h>
 
 namespace WebCore {
 
 WTF_MAKE_TZONE_ALLOCATED_IMPL(WorkerGlobalScope);
 
-MessagePortChannelProvider& WorkerGlobalScope::messagePortChannelProvider()
-{
-    return *reinterpret_cast<MessagePortChannelProvider*>(&MessagePortChannelProviderImpl::singleton());
-}
-
 void WorkerGlobalScope::onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind)
 {
     if (eventType == eventNames().messageEvent) {

src/bun.js/bindings/BunWorkerGlobalScope.h

@@ -12,9 +12,6 @@
 
 namespace WebCore {
 
-class MessagePortChannelProvider;
-class MessagePortChannelProviderImpl;
-
 class WorkerGlobalScope : public RefCounted<WorkerGlobalScope>, public EventTargetWithInlineData {
     WTF_MAKE_TZONE_ALLOCATED(WorkerGlobalScope);
 
@@ -46,8 +43,6 @@ class WorkerGlobalScope : public RefCounted<WorkerGlobalScope>, public EventTarg
     void derefEventTarget() final { deref(); }
     void eventListenersDidChange() final {}
 
-    MessagePortChannelProvider& messagePortChannelProvider();
-
     ScriptExecutionContext* m_context;
 };
 }

src/bun.js/bindings/ScriptExecutionContext.cpp

@@ -1,14 +1,12 @@
 #include "root.h"
 #include "headers.h"
 #include "ScriptExecutionContext.h"
-#include "MessagePort.h"
+#include "ContextDestructionObserver.h"
 
 #include "libusockets.h"
 #include "_libusockets.h"
 #include "BunClientData.h"
 #include "EventLoopTask.h"
-#include "BunBroadcastChannelRegistry.h"
-#include <wtf/LazyRef.h>
 extern "C" void Bun__startLoop(us_loop_t* loop);
 
 namespace WebCore {
@@ -39,9 +37,6 @@ ScriptExecutionContext::ScriptExecutionContext(JSC::VM* vm, JSC::JSGlobalObject*
     : m_vm(vm)
     , m_globalObject(globalObject)
     , m_identifier(initialIdentifier())
-    , m_broadcastChannelRegistry([](auto& owner, auto& lazyRef) {
-        lazyRef.set(BunBroadcastChannelRegistry::create());
-    })
 {
     relaxAdoptionRequirement();
     addToContextsMap();
@@ -51,9 +46,6 @@ ScriptExecutionContext::ScriptExecutionContext(JSC::VM* vm, JSC::JSGlobalObject*
     : m_vm(vm)
     , m_globalObject(globalObject)
     , m_identifier(identifier == std::numeric_limits<int32_t>::max() ? ++lastUniqueIdentifier : identifier)
-    , m_broadcastChannelRegistry([](auto& owner, auto& lazyRef) {
-        lazyRef.set(BunBroadcastChannelRegistry::create());
-    })
 {
     relaxAdoptionRequirement();
     addToContextsMap();
@@ -137,10 +129,6 @@ ScriptExecutionContext::~ScriptExecutionContext()
     m_inScriptExecutionContextDestructor = true;
 #endif // ASSERT_ENABLED
 
-    auto postMessageCompletionHandlers = WTF::move(m_processMessageWithMessagePortsSoonHandlers);
-    for (auto& completionHandler : postMessageCompletionHandlers)
-        completionHandler();
-
     while (auto* destructionObserver = m_destructionObservers.takeAny())
         destructionObserver->contextDestroyed();
 
@@ -228,71 +216,14 @@ ScriptExecutionContext* ScriptExecutionContext::getMainThreadScriptExecutionCont
     return allScriptExecutionContextsMap().get(1);
 }
 
-void ScriptExecutionContext::processMessageWithMessagePortsSoon(CompletionHandler<void()>&& completionHandler)
-{
-    ASSERT(isContextThread());
-    m_processMessageWithMessagePortsSoonHandlers.append(WTF::move(completionHandler));
-
-    if (m_willProcessMessageWithMessagePortsSoon) {
-        return;
-    }
-
-    m_willProcessMessageWithMessagePortsSoon = true;
-
-    postTask([](ScriptExecutionContext& context) {
-        context.dispatchMessagePortEvents();
-    });
-}
-
-void ScriptExecutionContext::dispatchMessagePortEvents()
-{
-    ASSERT(isContextThread());
-    checkConsistency();
-
-    ASSERT(m_willProcessMessageWithMessagePortsSoon);
-    m_willProcessMessageWithMessagePortsSoon = false;
-
-    auto completionHandlers = std::exchange(m_processMessageWithMessagePortsSoonHandlers, Vector<CompletionHandler<void()>> {});
-
-    // Make a frozen copy of the ports so we can iterate while new ones might be added or destroyed.
-    for (auto* messagePort : copyToVector(m_messagePorts)) {
-        // The port may be destroyed, and another one created at the same address,
-        // but this is harmless. The worst that can happen as a result is that
-        // dispatchMessages() will be called needlessly.
-        if (m_messagePorts.contains(messagePort) && messagePort->started())
-            messagePort->dispatchMessages();
-    }
-
-    for (auto& completionHandler : completionHandlers)
-        completionHandler();
-}
-
 void ScriptExecutionContext::checkConsistency() const
 {
 #if ASSERT_ENABLED
-    for (auto* messagePort : m_messagePorts)
-        ASSERT(messagePort->scriptExecutionContext() == this);
-
     for (auto* destructionObserver : m_destructionObservers)
         ASSERT(destructionObserver->scriptExecutionContext() == this);
-
 #endif // ASSERT_ENABLED
 }
 
-void ScriptExecutionContext::createdMessagePort(MessagePort& messagePort)
-{
-    ASSERT(isContextThread());
-
-    m_messagePorts.add(&messagePort);
-}
-
-void ScriptExecutionContext::destroyedMessagePort(MessagePort& messagePort)
-{
-    ASSERT(isContextThread());
-
-    m_messagePorts.remove(&messagePort);
-}
-
 us_socket_context_t* ScriptExecutionContext::webSocketContextNoSSL()
 {
     if (!m_client_websockets_ctx) {

src/bun.js/bindings/ScriptExecutionContext.h

@@ -12,7 +12,6 @@
 #include "CachedScript.h"
 #include "wtf/ThreadSafeWeakPtr.h"
 #include <wtf/URL.h>
-#include <wtf/LazyRef.h>
 
 namespace uWS {
 template<bool isServer, bool isClient, typename UserData>
@@ -26,8 +25,6 @@ struct us_loop_t;
 namespace WebCore {
 
 class WebSocket;
-class BunBroadcastChannelRegistry;
-class MessagePort;
 
 class ScriptExecutionContext;
 class EventLoopTask;
@@ -114,11 +111,6 @@ class ScriptExecutionContext : public CanMakeWeakPtr<ScriptExecutionContext>, pu
     void didCreateDestructionObserver(ContextDestructionObserver&);
     void willDestroyDestructionObserver(ContextDestructionObserver&);
 
-    void processMessageWithMessagePortsSoon(CompletionHandler<void()>&&);
-    void createdMessagePort(MessagePort&);
-    void destroyedMessagePort(MessagePort&);
-
-    void dispatchMessagePortEvents();
     void checkConsistency() const;
 
     void regenerateIdentifier();
@@ -149,8 +141,6 @@ class ScriptExecutionContext : public CanMakeWeakPtr<ScriptExecutionContext>, pu
         m_vm = &globalObject->vm();
     }
 
-    BunBroadcastChannelRegistry& broadcastChannelRegistry() { return m_broadcastChannelRegistry.get(*this); }
-
     static ScriptExecutionContext* getMainThreadScriptExecutionContext();
 
 private:
@@ -159,12 +149,7 @@ class ScriptExecutionContext : public CanMakeWeakPtr<ScriptExecutionContext>, pu
     WTF::URL m_url = WTF::URL();
     ScriptExecutionContextIdentifier m_identifier;
 
-    UncheckedKeyHashSet<MessagePort*> m_messagePorts;
     UncheckedKeyHashSet<ContextDestructionObserver*> m_destructionObservers;
-    Vector<CompletionHandler<void()>> m_processMessageWithMessagePortsSoonHandlers;
-    LazyRef<ScriptExecutionContext, BunBroadcastChannelRegistry> m_broadcastChannelRegistry;
-
-    bool m_willProcessMessageWithMessagePortsSoon { false };
 
     us_socket_context_t* webSocketContextSSL();
     us_socket_context_t* webSocketContextNoSSL();

src/bun.js/bindings/ZigGlobalObject.cpp

@@ -66,6 +66,7 @@
 #include "GeneratedBunObject.h"
 #include "BunPlugin.h"
 #include "BunProcess.h"
+#include "ProcessIdentifier.h"
 #include "BunWorkerGlobalScope.h"
 #include "CallSite.h"
 #include "CallSitePrototype.h"