Changeset 139514 in webkit

Timestamp:

Jan 11, 2013, 4:01:56 PM (12 years ago)

Author:

ap@apple.com

Message:

[WK2] Make it possible to send sync messages from secondary threads
​https://bugs.webkit.org/show_bug.cgi?id=106708

Reviewed by Anders Carlsson.

It is hugely beneficial to implement sync messages at Connection level, because
ad hoc code that blocks a thread and wakes it up when a reply arrives on main
thread can't be made equally performant. A CoreOPC MessageDecoder can be moved across
threads, which can't be done with a decoded argument passed by reference to client code.

Sync messages from secondary threads are tracked in much simpler data structure
than client thread ones, because we don't need to be concerned with incoming messages.

• Platform/CoreIPC/Connection.cpp: (Connection::SecondaryThreadPendingSyncReply): (CoreIPC::Connection::SecondaryThreadPendingSyncReply::SecondaryThreadPendingSyncReply): (CoreIPC::Connection::createSyncMessageEncoder): (CoreIPC::Connection::sendSyncMessage): (CoreIPC::Connection::sendSyncMessageFromSecondaryThread): (CoreIPC::Connection::processIncomingSyncReply): (CoreIPC::Connection::connectionDidClose):

• Platform/CoreIPC/Connection.h: Also corrected a misleading comment.

Location:

trunk/Source/WebKit2

Files:

• ChangeLog (modified) (1 diff)
• Platform/CoreIPC/Connection.cpp (modified) (6 diffs)
• Platform/CoreIPC/Connection.h (modified) (4 diffs)

trunk/Source/WebKit2/ChangeLog

@@ +1 @@
+2013-01-11  Alexey Proskuryakov  <ap@apple.com>
+
+        [WK2] Make it possible to send sync messages from secondary threads
+        https://bugs.webkit.org/show_bug.cgi?id=106708
+
+        Reviewed by Anders Carlsson.
+
+        It is hugely beneficial to implement sync messages at Connection level, because
+        ad hoc code that blocks a thread and wakes it up when a reply arrives on main
+        thread can't be made equally performant. A CoreOPC MessageDecoder can be moved across
+        threads, which can't be done with a decoded argument passed by reference to client code.
+
+        Sync messages from secondary threads are tracked in much simpler data structure
+        than client thread ones, because we don't need to be concerned with incoming messages.
+
+        * Platform/CoreIPC/Connection.cpp:
+        (Connection::SecondaryThreadPendingSyncReply):
+        (CoreIPC::Connection::SecondaryThreadPendingSyncReply::SecondaryThreadPendingSyncReply):
+        (CoreIPC::Connection::createSyncMessageEncoder):
+        (CoreIPC::Connection::sendSyncMessage):
+        (CoreIPC::Connection::sendSyncMessageFromSecondaryThread):
+        (CoreIPC::Connection::processIncomingSyncReply):
+        (CoreIPC::Connection::connectionDidClose):
+
+        * Platform/CoreIPC/Connection.h: Also corrected a misleading comment.
+
 2013-01-11  Dan Bernstein  <mitz@apple.com>
 

trunk/Source/WebKit2/Platform/CoreIPC/Connection.cpp

@@ -94 +94 @@
 };
 
+class Connection::SecondaryThreadPendingSyncReply {
+WTF_MAKE_NONCOPYABLE(SecondaryThreadPendingSyncReply);
+public:
+    SecondaryThreadPendingSyncReply() : replyDecoder(0) { }
+
+    // The reply decoder, will be null if there was an error processing the sync message on the other side.
+    MessageDecoder* replyDecoder;
+
+    BinarySemaphore semaphore;
+};
+
+
 PassRefPtr<Connection::SyncMessageState> Connection::SyncMessageState::getOrCreate(RunLoop* runLoop)
 {
@@ -298 +310 @@
 
     // Encode the sync request ID.
-    syncRequestID = ++m_syncRequestID;
+    COMPILE_ASSERT(sizeof(m_syncRequestID) == sizeof(int64_t), CanUseAtomicIncrement);
+    syncRequestID = atomicIncrement(reinterpret_cast<int64_t volatile*>(&m_syncRequestID));
     encoder->encode(syncRequestID);
 
@@ -387 +400 @@
 PassOwnPtr<MessageDecoder> Connection::sendSyncMessage(MessageID messageID, uint64_t syncRequestID, PassOwnPtr<MessageEncoder> encoder, double timeout, unsigned syncSendFlags)
 {
-    // We only allow sending sync messages from the client run loop.
-    ASSERT(RunLoop::current() == m_clientRunLoop);
+    if (RunLoop::current() != m_clientRunLoop) {
+        // No flags are supported for synchronous messages sent from secondary threads.
+        ASSERT(!syncSendFlags);
+        return sendSyncMessageFromSecondaryThread(messageID, syncRequestID, encoder, timeout);
+    }
 
     if (!isValid()) {
@@ -421 +437 @@
     }
 
-    if (!reply)
+    // FIXME: Should we call didFailToSendSyncMessage()? It may be unexpected to get in on a background thread.
+
+    return reply.release();
+}
+
+PassOwnPtr<MessageDecoder> Connection::sendSyncMessageFromSecondaryThread(MessageID messageID, uint64_t syncRequestID, PassOwnPtr<MessageEncoder> encoder, double timeout)
+{
+    ASSERT(RunLoop::current() != m_clientRunLoop);
+
+    if (!isValid()) {
         didFailToSendSyncMessage();
-
-    return reply.release();
+        return nullptr;
+    }
+
+    SecondaryThreadPendingSyncReply pendingReply;
+
+    // Push the pending sync reply information on our stack.
+    {
+        MutexLocker locker(m_syncReplyStateMutex);
+        if (!m_shouldWaitForSyncReplies) {
+            didFailToSendSyncMessage();
+            return nullptr;
+        }
+
+        ASSERT(!m_secondaryThreadPendingSyncReplyMap.contains(syncRequestID));
+        m_secondaryThreadPendingSyncReplyMap.add(syncRequestID, &pendingReply);
+    }
+
+    sendMessage(messageID.messageIDWithAddedFlags(MessageID::SyncMessage), encoder, 0);
+
+    // Use a really long timeout.
+    if (timeout == NoTimeout)
+        timeout = 1e10;
+
+    pendingReply.semaphore.wait(timeout);
+
+    // Finally, pop the pending sync reply information.
+    {
+        MutexLocker locker(m_syncReplyStateMutex);
+        ASSERT(m_secondaryThreadPendingSyncReplyMap.contains(syncRequestID));
+        m_secondaryThreadPendingSyncReplyMap.remove(syncRequestID);
+    }
+
+    if (!pendingReply.replyDecoder)
+        didFailToSendSyncMessage();
+
+    return adoptPtr(pendingReply.replyDecoder);
 }
 
@@ -503 +562 @@
     }
 
-    // If we get here, it means we got a reply for a message that wasn't in the sync request stack.
+    // If it's not a reply to any primary thread message, check if it is a reply to a secondary thread one.
+    SecondaryThreadPendingSyncReplyMap::iterator secondaryThreadReplyMapItem = m_secondaryThreadPendingSyncReplyMap.find(decoder->destinationID());
+    if (secondaryThreadReplyMapItem != m_secondaryThreadPendingSyncReplyMap.end()) {
+        SecondaryThreadPendingSyncReply* reply = secondaryThreadReplyMapItem->value;
+        ASSERT(!reply->replyDecoder);
+        reply->replyDecoder = decoder.leakPtr();
+        reply->semaphore.signal();
+    }
+
+    // If we get here, it means we got a reply for a message that wasn't in the sync request stack or map.
     // This can happen if the send timed out, so it's fine to ignore.
 }
@@ -571 +639 @@
         if (!m_pendingSyncReplies.isEmpty())
             m_syncMessageState->wakeUpClientRunLoop();
+
+        for (SecondaryThreadPendingSyncReplyMap::iterator iter = m_secondaryThreadPendingSyncReplyMap.begin(); iter != m_secondaryThreadPendingSyncReplyMap.end(); ++iter)
+            iter->value->semaphore.signal();
     }
 

trunk/Source/WebKit2/Platform/CoreIPC/Connection.h

@@ -183 +183 @@
     bool sendMessage(MessageID, PassOwnPtr<MessageEncoder>, unsigned messageSendFlags = 0);
     PassOwnPtr<MessageDecoder> sendSyncMessage(MessageID, uint64_t syncRequestID, PassOwnPtr<MessageEncoder>, double timeout, unsigned syncSendFlags = 0);
+    PassOwnPtr<MessageDecoder> sendSyncMessageFromSecondaryThread(MessageID, uint64_t syncRequestID, PassOwnPtr<MessageEncoder>, double timeout);
     bool sendSyncReply(PassOwnPtr<MessageEncoder>);
 
@@ -306 +307 @@
         MessageDecoder* replyDecoder;
 
-        // Will be set to true once a reply has been received or an error occurred.
+        // Will be set to true once a reply has been received.
         bool didReceiveReply;
     
@@ -331 +332 @@
         }
     };
-    
+
     class SyncMessageState;
     friend class SyncMessageState;
@@ -339 +340 @@
     bool m_shouldWaitForSyncReplies;
     Vector<PendingSyncReply> m_pendingSyncReplies;
+
+    class SecondaryThreadPendingSyncReply;
+    typedef HashMap<uint64_t, SecondaryThreadPendingSyncReply*> SecondaryThreadPendingSyncReplyMap;
+    SecondaryThreadPendingSyncReplyMap m_secondaryThreadPendingSyncReplyMap;
 
 #if OS(DARWIN)