LibWeb+WebWorker: Convert Workers to use MessagePorts for postMessage

This aligns Workers and Window and MessagePorts to all use the same
mechanism for transferring serialized messages across realms.

It also allows transferring more message ports into a worker.

Re-enable the Worker-echo test, as none of the MessagePort tests have
themselves been flaky, and those are now using the same underlying
implementation.

Author: ADKaster

Base/res/html/misc/worker.js

@@ -1,10 +1,10 @@
 onmessage = evt => {
     console.log("In Worker - Got message:", JSON.stringify(evt.data));
 
-    postMessage(evt.data, null);
+    postMessage(evt.data);
 };
 
 console.log("In Worker - Loaded", this);
 console.log("Keys: ", JSON.stringify(Object.keys(this)));
 
-postMessage("loaded", null);
+postMessage("loaded");

Tests/LibWeb/TestConfig.ini

@@ -1,3 +1 @@
 [Skipped]
-Text/input/Worker/Worker-echo.html
-

Tests/LibWeb/Text/expected/Worker/Worker-echo.txt

@@ -1,3 +1,4 @@
 Got message from worker: "loaded"
-Got message from worker: {"msg":"marco"}
+Got message from port: "Worker got message port!"
+Got message from port: "Extra Port got message: \"Hello from port2\""
 DONE

Tests/LibWeb/Text/input/Worker/Worker-echo.html

@@ -2,17 +2,31 @@
 <script>
     asyncTest((done) => {
         let work = new Worker("worker.js");
+        let channel = new MessageChannel();
+
+        function finishTest() {
+            println("DONE");
+            work.onmessage = null;
+            work.terminate();
+            channel.port2.onmessage = null;
+            done();
+        }
         let count = 0;
         work.onmessage = (evt) => {
             println("Got message from worker: " + JSON.stringify(evt.data));
             count++;
-            work.postMessage({"msg": "marco"});
-            if (count === 2) {
-                println("DONE");
-                work.onmessage = null;
-                work.terminate();
-                done();
+            if (count === 3) {
+                finishTest();
             }
         };
+        channel.port2.onmessage = (evt) => {
+            println("Got message from port: " + JSON.stringify(evt.data));
+            channel.port2.postMessage("Hello from port2");
+            count++;
+            if (count === 3) {
+                finishTest();
+            }
+        };
+        work.postMessage({ port: channel.port1 }, { transfer : [channel.port1]});
     });
 </script>

Tests/LibWeb/Text/input/Worker/worker.js

@@ -1,4 +1,14 @@
+let extraPort = null;
+
 onmessage = evt => {
-    postMessage(evt.data, null);
+    if (evt.ports.length > 0) {
+        extraPort = evt.ports[0];
+        extraPort.onmessage = evt => {
+            extraPort.postMessage("Extra Port got message: " + JSON.stringify(evt.data));
+        };
+        extraPort.postMessage("Worker got message port!");
+    } else {
+        postMessage(evt.data);
+    }
 };
-postMessage("loaded", null);
+postMessage("loaded");

Userland/Libraries/LibWeb/Forward.h

@@ -446,6 +446,7 @@ class Timer;
 class TimeRanges;
 class ToggleEvent;
 class TrackEvent;
+struct TransferDataHolder;
 class TraversableNavigable;
 class VideoTrack;
 class VideoTrackList;

Userland/Libraries/LibWeb/HTML/MessagePort.cpp

@@ -19,6 +19,7 @@
 #include <LibWeb/HTML/MessageEvent.h>
 #include <LibWeb/HTML/MessagePort.h>
 #include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
+#include <LibWeb/HTML/WorkerGlobalScope.h>
 
 namespace Web::HTML {
 
@@ -53,6 +54,11 @@ void MessagePort::visit_edges(Cell::Visitor& visitor)
     visitor.visit(m_remote_port);
 }
 
+void MessagePort::set_worker_event_target(JS::NonnullGCPtr<DOM::EventTarget> target)
+{
+    m_worker_event_target = target;
+}
+
 // https://html.spec.whatwg.org/multipage/web-messaging.html#message-ports:transfer-steps
 WebIDL::ExceptionOr<void> MessagePort::transfer_steps(HTML::TransferDataHolder& data_holder)
 {
@@ -107,6 +113,10 @@ WebIDL::ExceptionOr<void> MessagePort::transfer_receiving_steps(HTML::TransferDa
         VERIFY(fd_tag == IPC_FILE_TAG);
         fd = data_holder.fds.take_first();
         m_fd_passing_socket = MUST(Core::LocalSocket::adopt_fd(fd.take_fd(), Core::LocalSocket::PreventSIGPIPE::Yes));
+
+        m_socket->on_ready_to_read = [strong_this = JS::make_handle(this)]() {
+            strong_this->read_from_socket();
+        };
     } else if (fd_tag != 0) {
         dbgln("Unexpected byte {:x} in MessagePort transfer data", fd_tag);
         VERIFY_NOT_REACHED();
@@ -348,6 +358,16 @@ void MessagePort::post_message_task_steps(SerializedTransferRecord& serialize_wi
     // NOTE: This can be different from targetPort, if targetPort itself was transferred and thus all its tasks moved along with it.
     auto* final_target_port = this;
 
+    // IMPLEMENTATION DEFINED:
+    // https://html.spec.whatwg.org/multipage/workers.html#dedicated-workers-and-the-worker-interface
+    //      Worker objects act as if they had an implicit MessagePort associated with them.
+    //      All messages received by that port must immediately be retargeted at the Worker object.
+    // We therefore set a special event target for those implicit ports on the Worker and the WorkerGlobalScope objects
+    EventTarget* message_event_target = final_target_port;
+    if (m_worker_event_target != nullptr) {
+        message_event_target = m_worker_event_target;
+    }
+
     // 2. Let targetRealm be finalTargetPort's relevant realm.
     auto& target_realm = relevant_realm(*final_target_port);
     auto& target_vm = target_realm.vm();
@@ -359,7 +379,7 @@ void MessagePort::post_message_task_steps(SerializedTransferRecord& serialize_wi
         // If this throws an exception, catch it, fire an event named messageerror at finalTargetPort, using MessageEvent, and then return.
         auto exception = deserialize_record_or_error.release_error();
         MessageEventInit event_init {};
-        final_target_port->dispatch_event(MessageEvent::create(target_realm, HTML::EventNames::messageerror, event_init));
+        message_event_target->dispatch_event(MessageEvent::create(target_realm, HTML::EventNames::messageerror, event_init));
         return;
     }
     auto deserialize_record = deserialize_record_or_error.release_value();
@@ -380,7 +400,7 @@ void MessagePort::post_message_task_steps(SerializedTransferRecord& serialize_wi
     MessageEventInit event_init {};
     event_init.data = message_clone;
     event_init.ports = move(new_ports);
-    final_target_port->dispatch_event(MessageEvent::create(target_realm, HTML::EventNames::message, event_init));
+    message_event_target->dispatch_event(MessageEvent::create(target_realm, HTML::EventNames::message, event_init));
 }
 
 // https://html.spec.whatwg.org/multipage/web-messaging.html#dom-messageport-start

Userland/Libraries/LibWeb/HTML/MessagePort.h

@@ -61,6 +61,8 @@ class MessagePort final : public DOM::EventTarget
     virtual WebIDL::ExceptionOr<void> transfer_receiving_steps(HTML::TransferDataHolder&) override;
     virtual HTML::TransferType primary_interface() const override { return HTML::TransferType::MessagePort; }
 
+    void set_worker_event_target(JS::NonnullGCPtr<DOM::EventTarget>);
+
 private:
     explicit MessagePort(JS::Realm&);
 
@@ -91,6 +93,8 @@ class MessagePort final : public DOM::EventTarget
         Error,
     } m_socket_state { SocketState::Header };
     size_t m_socket_incoming_message_size { 0 };
+
+    JS::GCPtr<DOM::EventTarget> m_worker_event_target;
 };
 
 }

Userland/Libraries/LibWeb/HTML/Worker.cpp

@@ -11,15 +11,14 @@
 #include <LibWeb/HTML/Scripting/Environments.h>
 #include <LibWeb/HTML/Scripting/TemporaryExecutionContext.h>
 #include <LibWeb/HTML/Worker.h>
-#include <LibWeb/HTML/WorkerDebugConsoleClient.h>
 #include <LibWeb/WebIDL/ExceptionOr.h>
 
 namespace Web::HTML {
 
 JS_DEFINE_ALLOCATOR(Worker);
 
 // https://html.spec.whatwg.org/multipage/workers.html#dedicated-workers-and-the-worker-interface
-Worker::Worker(String const& script_url, WorkerOptions const options, DOM::Document& document)
+Worker::Worker(String const& script_url, WorkerOptions const& options, DOM::Document& document)
     : DOM::EventTarget(document.realm())
     , m_script_url(script_url)
     , m_options(options)
@@ -42,7 +41,7 @@ void Worker::visit_edges(Cell::Visitor& visitor)
 }
 
 // https://html.spec.whatwg.org/multipage/workers.html#dom-worker
-WebIDL::ExceptionOr<JS::NonnullGCPtr<Worker>> Worker::create(String const& script_url, WorkerOptions const options, DOM::Document& document)
+WebIDL::ExceptionOr<JS::NonnullGCPtr<Worker>> Worker::create(String const& script_url, WorkerOptions const& options, DOM::Document& document)
 {
     dbgln_if(WEB_WORKER_DEBUG, "WebWorker: Creating worker with script_url = {}", script_url);
 
@@ -79,6 +78,7 @@ WebIDL::ExceptionOr<JS::NonnullGCPtr<Worker>> Worker::create(String const& scrip
 
     // 8. Associate the outside port with worker
     worker->m_outside_port = outside_port;
+    worker->m_outside_port->set_worker_event_target(worker);
 
     // 9. Run this step in parallel:
     //    1. Run a worker given worker, worker URL, outside settings, outside port, and options.
@@ -89,7 +89,7 @@ WebIDL::ExceptionOr<JS::NonnullGCPtr<Worker>> Worker::create(String const& scrip
 }
 
 // https://html.spec.whatwg.org/multipage/workers.html#run-a-worker
-void Worker::run_a_worker(AK::URL& url, EnvironmentSettingsObject& outside_settings, MessagePort&, WorkerOptions const& options)
+void Worker::run_a_worker(AK::URL& url, EnvironmentSettingsObject& outside_settings, JS::GCPtr<MessagePort> port, WorkerOptions const& options)
 {
     // 1. Let is shared be true if worker is a SharedWorker object, and false otherwise.
     // FIXME: SharedWorker support
@@ -110,55 +110,7 @@ void Worker::run_a_worker(AK::URL& url, EnvironmentSettingsObject& outside_setti
     // and is shared. Run the rest of these steps in that agent.
 
     // Note: This spawns a new process to act as the 'agent' for the worker.
-    m_agent = heap().allocate_without_realm<WorkerAgent>(url, options);
-
-    auto& socket = m_agent->socket();
-    // FIXME: Hide this logic in MessagePort
-    socket.set_notifications_enabled(true);
-    socket.on_ready_to_read = [this] {
-        auto& socket = this->m_agent->socket();
-        auto& vm = this->vm();
-        auto& realm = this->realm();
-
-        auto num_bytes_ready = MUST(socket.pending_bytes());
-        switch (m_outside_port_state) {
-        case PortState::Header: {
-            if (num_bytes_ready < 8)
-                break;
-            auto const magic = MUST(socket.read_value<u32>());
-            if (magic != 0xDEADBEEF) {
-                m_outside_port_state = PortState::Error;
-                break;
-            }
-            m_outside_port_incoming_message_size = MUST(socket.read_value<u32>());
-            num_bytes_ready -= 8;
-            m_outside_port_state = PortState::Data;
-        }
-            [[fallthrough]];
-        case PortState::Data: {
-            if (num_bytes_ready < m_outside_port_incoming_message_size)
-                break;
-            SerializationRecord rec; // FIXME: Keep in class scope
-            rec.resize(m_outside_port_incoming_message_size / sizeof(u32));
-
-            MUST(socket.read_until_filled(to_bytes(rec.span())));
-
-            TemporaryExecutionContext cxt(relevant_settings_object(*this));
-            VERIFY(&realm == vm.current_realm());
-            MessageEventInit event_init {};
-            event_init.data = MUST(structured_deserialize(vm, rec, realm, {}));
-            // FIXME: Fill in the rest of the info from MessagePort
-
-            this->dispatch_event(MessageEvent::create(realm, EventNames::message, event_init));
-
-            m_outside_port_state = PortState::Header;
-            break;
-        }
-        case PortState::Error:
-            VERIFY_NOT_REACHED();
-            break;
-        }
-    };
+    m_agent = heap().allocate<WorkerAgent>(outside_settings.realm(), url, options, port);
 }
 
 // https://html.spec.whatwg.org/multipage/workers.html#dom-worker-terminate
@@ -170,29 +122,15 @@ WebIDL::ExceptionOr<void> Worker::terminate()
 }
 
 // https://html.spec.whatwg.org/multipage/workers.html#dom-worker-postmessage
-WebIDL::ExceptionOr<void> Worker::post_message(JS::Value message, JS::Value)
+WebIDL::ExceptionOr<void> Worker::post_message(JS::Value message, StructuredSerializeOptions const& options)
 {
     dbgln_if(WEB_WORKER_DEBUG, "WebWorker: Post Message: {}", message.to_string_without_side_effects());
 
-    // FIXME: 1. Let targetPort be the port with which this is entangled, if any; otherwise let it be null.
-    // FIXME: 2. Let options be «[ "transfer" → transfer ]».
-    // FIXME: 3. Run the message port post message steps providing this, targetPort, message and options.
+    // The postMessage(message, transfer) and postMessage(message, options) methods on Worker objects act as if,
+    // when invoked, they immediately invoked the respective postMessage(message, transfer) and
+    // postMessage(message, options) on the port, with the same arguments, and returned the same return value.
 
-    auto& realm = this->realm();
-    auto& vm = this->vm();
-
-    // FIXME: Use the with-transfer variant, which should(?) prepend the magic + size at the front
-    auto data = TRY(structured_serialize(vm, message));
-
-    Array<u32, 2> header = { 0xDEADBEEF, static_cast<u32>(data.size() * sizeof(u32)) };
-
-    if (auto const err = m_agent->socket().write_until_depleted(to_readonly_bytes(header.span())); err.is_error())
-        return WebIDL::DataCloneError::create(realm, TRY_OR_THROW_OOM(vm, String::formatted("{}", err.error())));
-
-    if (auto const err = m_agent->socket().write_until_depleted(to_readonly_bytes(data.span())); err.is_error())
-        return WebIDL::DataCloneError::create(realm, TRY_OR_THROW_OOM(vm, String::formatted("{}", err.error())));
-
-    return {};
+    return m_outside_port->post_message(message, options);
 }
 
 #undef __ENUMERATE

Userland/Libraries/LibWeb/HTML/Worker.h

@@ -32,16 +32,16 @@ class Worker : public DOM::EventTarget {
     JS_DECLARE_ALLOCATOR(Worker);
 
 public:
-    static WebIDL::ExceptionOr<JS::NonnullGCPtr<Worker>> create(String const& script_url, WorkerOptions const options, DOM::Document& document);
-    static WebIDL::ExceptionOr<JS::NonnullGCPtr<Worker>> construct_impl(JS::Realm& realm, String const& script_url, WorkerOptions const options)
+    static WebIDL::ExceptionOr<JS::NonnullGCPtr<Worker>> create(String const& script_url, WorkerOptions const& options, DOM::Document& document);
+    static WebIDL::ExceptionOr<JS::NonnullGCPtr<Worker>> construct_impl(JS::Realm& realm, String const& script_url, WorkerOptions const& options)
     {
         auto& window = verify_cast<HTML::Window>(realm.global_object());
         return Worker::create(script_url, options, window.associated_document());
     }
 
     WebIDL::ExceptionOr<void> terminate();
 
-    WebIDL::ExceptionOr<void> post_message(JS::Value message, JS::Value transfer);
+    WebIDL::ExceptionOr<void> post_message(JS::Value message, StructuredSerializeOptions const&);
 
     virtual ~Worker() = default;
 
@@ -55,7 +55,7 @@ class Worker : public DOM::EventTarget {
 #undef __ENUMERATE
 
 protected:
-    Worker(String const&, const WorkerOptions, DOM::Document&);
+    Worker(String const&, WorkerOptions const&, DOM::Document&);
 
 private:
     virtual void initialize(JS::Realm&) override;
@@ -66,17 +66,10 @@ class Worker : public DOM::EventTarget {
 
     JS::GCPtr<DOM::Document> m_document;
     JS::GCPtr<MessagePort> m_outside_port;
-    // FIXME: Move tihs state into the message port (and actually use it :) )
-    enum class PortState : u8 {
-        Header,
-        Data,
-        Error,
-    } m_outside_port_state { PortState::Header };
-    size_t m_outside_port_incoming_message_size { 0 };
 
     JS::GCPtr<WorkerAgent> m_agent;
 
-    void run_a_worker(AK::URL& url, EnvironmentSettingsObject& outside_settings, MessagePort& outside_port, WorkerOptions const& options);
+    void run_a_worker(AK::URL& url, EnvironmentSettingsObject& outside_settings, JS::GCPtr<MessagePort> outside_port, WorkerOptions const& options);
 };
 
 }