Optimize postMessage between two MessageChannel ports living in the s…

…ame process

https://bugs.webkit.org/show_bug.cgi?id=246943
rdar://problem/101497189

Reviewed by Alex Christensen.

Some objects may be heavy to transfer from one process to another (RTCDataChannel, encoded chunks, video frames...).
It is best if we can use the same code path as when doing Worker::postMessage.
This is feasible as long as we keep the order of the messages, since each MessagePort has its own event queue.

We introduce a map of port id to messages in WebMessagePortChannelProvider.
We fill the map whenever creating or whenever entangling a port in the process.
When postMessaging to that port, we check whether it is in process and if so add messages to the local map.
When taking messages, we still go to network process but complement any message we may find by adding the local messages.

* Source/WebKit/WebProcess/WebCoreSupport/WebMessagePortChannelProvider.cpp:
(WebKit::WebMessagePortChannelProvider::createNewMessagePortChannel):
(WebKit::WebMessagePortChannelProvider::entangleLocalPortInThisProcessToRemote):
(WebKit::WebMessagePortChannelProvider::messagePortDisentangled):
(WebKit::WebMessagePortChannelProvider::messagePortClosed):
(WebKit::WebMessagePortChannelProvider::takeAllMessagesForPort):
(WebKit::WebMessagePortChannelProvider::postMessageToRemote):
(WebKit::WebMessagePortChannelProvider::checkRemotePortForActivity):
* Source/WebKit/WebProcess/WebCoreSupport/WebMessagePortChannelProvider.h:

Canonical link: https://commits.webkit.org/255948@main

Source/WebKit/WebProcess/WebCoreSupport/WebMessagePortChannelProvider.cpp

@@ -58,27 +58,46 @@ static inline IPC::Connection& networkProcessConnection()
 
 void WebMessagePortChannelProvider::createNewMessagePortChannel(const MessagePortIdentifier& port1, const MessagePortIdentifier& port2)
 {
+    ASSERT(!m_inProcessPortMessages.contains(port1));
+    ASSERT(!m_inProcessPortMessages.contains(port2));
+    m_inProcessPortMessages.add(port1, Vector<MessageWithMessagePorts> { });
+    m_inProcessPortMessages.add(port2, Vector<MessageWithMessagePorts> { });
+
     networkProcessConnection().send(Messages::NetworkConnectionToWebProcess::CreateNewMessagePortChannel { port1, port2 }, 0);
 }
 
 void WebMessagePortChannelProvider::entangleLocalPortInThisProcessToRemote(const MessagePortIdentifier& local, const MessagePortIdentifier& remote)
 {
+    m_inProcessPortMessages.add(local, Vector<MessageWithMessagePorts> { });
+    ASSERT(!m_inProcessPortMessages.get(local).size());
+
     networkProcessConnection().send(Messages::NetworkConnectionToWebProcess::EntangleLocalPortInThisProcessToRemote { local, remote }, 0);
 }
 
 void WebMessagePortChannelProvider::messagePortDisentangled(const MessagePortIdentifier& port)
 {
+    auto inProcessPortMessages = m_inProcessPortMessages.take(port);
     networkProcessConnection().send(Messages::NetworkConnectionToWebProcess::MessagePortDisentangled { port }, 0);
+    for (auto& message : inProcessPortMessages)
+        postMessageToRemote(WTFMove(message), port);
 }
 
 void WebMessagePortChannelProvider::messagePortClosed(const MessagePortIdentifier& port)
 {
+    m_inProcessPortMessages.remove(port);
     networkProcessConnection().send(Messages::NetworkConnectionToWebProcess::MessagePortClosed { port }, 0);
 }
 
 void WebMessagePortChannelProvider::takeAllMessagesForPort(const MessagePortIdentifier& port, CompletionHandler<void(Vector<MessageWithMessagePorts>&&, CompletionHandler<void()>&&)>&& completionHandler)
 {
-    networkProcessConnection().sendWithAsyncReply(Messages::NetworkConnectionToWebProcess::TakeAllMessagesForPort { port }, [completionHandler = WTFMove(completionHandler)](auto&& messages, uint64_t messageBatchIdentifier) mutable {
+    networkProcessConnection().sendWithAsyncReply(Messages::NetworkConnectionToWebProcess::TakeAllMessagesForPort { port }, [completionHandler = WTFMove(completionHandler), port](auto&& messages, uint64_t messageBatchIdentifier) mutable {
+
+        auto& inProcessPortMessages = WebMessagePortChannelProvider::singleton().m_inProcessPortMessages;
+        auto iterator = inProcessPortMessages.find(port);
+        if (iterator != inProcessPortMessages.end()) {
+            auto pendingMessages = std::exchange(iterator->value, { });
+            messages.appendVector(WTFMove(pendingMessages));
+        }
         completionHandler(WTFMove(messages), [messageBatchIdentifier] {
             networkProcessConnection().send(Messages::NetworkConnectionToWebProcess::DidDeliverMessagePortMessages { messageBatchIdentifier }, 0);
         });
@@ -87,12 +106,24 @@ void WebMessagePortChannelProvider::takeAllMessagesForPort(const MessagePortIden
 
 void WebMessagePortChannelProvider::postMessageToRemote(MessageWithMessagePorts&& message, const MessagePortIdentifier& remoteTarget)
 {
+    auto iterator = m_inProcessPortMessages.find(remoteTarget);
+    if (iterator != m_inProcessPortMessages.end()) {
+        iterator->value.append(WTFMove(message));
+        WebProcess::singleton().messagesAvailableForPort(remoteTarget);
+        return;
+    }
+
     networkProcessConnection().send(Messages::NetworkConnectionToWebProcess::PostMessageToRemote { message, remoteTarget }, 0);
 }
 
 void WebMessagePortChannelProvider::checkRemotePortForActivity(const MessagePortIdentifier& remoteTarget, CompletionHandler<void(HasActivity)>&& completionHandler)
 {
-    networkProcessConnection().sendWithAsyncReply(Messages::NetworkConnectionToWebProcess::CheckRemotePortForActivity { remoteTarget }, [completionHandler = WTFMove(completionHandler)](bool hasActivity) mutable {
+    networkProcessConnection().sendWithAsyncReply(Messages::NetworkConnectionToWebProcess::CheckRemotePortForActivity { remoteTarget }, [completionHandler = WTFMove(completionHandler), remoteTarget](bool hasActivity) mutable {
+        if (!hasActivity) {
+            auto& inProcessPorts = WebMessagePortChannelProvider::singleton().m_inProcessPortMessages;
+            auto iterator = inProcessPorts.find(remoteTarget);
+            hasActivity = iterator != inProcessPorts.end() && iterator->value.size();
+        }
         completionHandler(hasActivity ? HasActivity::Yes : HasActivity::No);
     }, 0);
 }

Source/WebKit/WebProcess/WebCoreSupport/WebMessagePortChannelProvider.h

@@ -45,6 +45,8 @@ class WebMessagePortChannelProvider final : public WebCore::MessagePortChannelPr
     void takeAllMessagesForPort(const WebCore::MessagePortIdentifier&, CompletionHandler<void(Vector<WebCore::MessageWithMessagePorts>&&, CompletionHandler<void()>&&)>&&) final;
     void postMessageToRemote(WebCore::MessageWithMessagePorts&&, const WebCore::MessagePortIdentifier& remoteTarget) final;
     void checkRemotePortForActivity(const WebCore::MessagePortIdentifier& remoteTarget, CompletionHandler<void(HasActivity)>&& callback) final;
+
+    HashMap<WebCore::MessagePortIdentifier, Vector<WebCore::MessageWithMessagePorts>> m_inProcessPortMessages;
 };
 
 } // namespace WebKit