Source/WebKit2/Platform/IPC/Connection.cpp

   1 /*
   2  * Copyright (C) 2010 Apple Inc. All rights reserved.
   3  *
   4  * Redistribution and use in source and binary forms, with or without
   5  * modification, are permitted provided that the following conditions
   6  * are met:
   7  * 1. Redistributions of source code must retain the above copyright
   8  *    notice, this list of conditions and the following disclaimer.
   9  * 2. Redistributions in binary form must reproduce the above copyright
  10  *    notice, this list of conditions and the following disclaimer in the
  11  *    documentation and/or other materials provided with the distribution.
  12  *
  13  * THIS SOFTWARE IS PROVIDED BY APPLE INC. AND ITS CONTRIBUTORS ``AS IS''
  14  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
  15  * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  16  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL APPLE INC. OR ITS CONTRIBUTORS
  17  * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  18  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  19  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  20  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  21  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  22  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  23  * THE POSSIBILITY OF SUCH DAMAGE.
  24  */
  25
  26 #include "config.h"
  27 #include "Connection.h"
  28
  29 #include <memory>
  30 #include <wtf/CurrentTime.h>
  31 #include <wtf/HashSet.h>
  32 #include <wtf/NeverDestroyed.h>
  33 #include <wtf/RunLoop.h>
  34 #include <wtf/text/WTFString.h>
  35 #include <wtf/threads/BinarySemaphore.h>
  36
  37 namespace IPC {
  38
  39 struct WaitForMessageState {
  40     WaitForMessageState(StringReference messageReceiverName, StringReference messageName, uint64_t destinationID, unsigned waitForMessageFlags)
  41         : messageReceiverName(messageReceiverName)
  42         , messageName(messageName)
  43         , destinationID(destinationID)
  44         , waitForMessageFlags(waitForMessageFlags)
  45     {
  46     }
  47
  48     StringReference messageReceiverName;
  49     StringReference messageName;
  50     uint64_t destinationID;
  51
  52     unsigned waitForMessageFlags;
  53     bool messageWaitingInterrupted = false;
  54
  55     std::unique_ptr<MessageDecoder> decoder;
  56 };
  57
  58 class Connection::SyncMessageState : public ThreadSafeRefCounted<Connection::SyncMessageState> {
  59 public:
  60     static Ref<SyncMessageState> getOrCreate(RunLoop&);
  61     ~SyncMessageState();
  62
  63     void wakeUpClientRunLoop()
  64     {
  65         m_waitForSyncReplySemaphore.signal();
  66     }
  67
  68     bool wait(double absoluteTime)
  69     {
  70         return m_waitForSyncReplySemaphore.wait(absoluteTime);
  71     }
  72
  73     // Returns true if this message will be handled on a client thread that is currently
  74     // waiting for a reply to a synchronous message.
  75     bool processIncomingMessage(Connection&, std::unique_ptr<MessageDecoder>&);
  76
  77     // Dispatch pending sync messages. if allowedConnection is not null, will only dispatch messages
  78     // from that connection and put the other messages back in the queue.
  79     void dispatchMessages(Connection* allowedConnection);
  80
  81 private:
  82     explicit SyncMessageState(RunLoop&);
  83
  84     typedef HashMap<RunLoop*, SyncMessageState*> SyncMessageStateMap;
  85     static SyncMessageStateMap& syncMessageStateMap()
  86     {
  87         static NeverDestroyed<SyncMessageStateMap> syncMessageStateMap;
  88         return syncMessageStateMap;
  89     }
  90
  91     static std::mutex& syncMessageStateMapMutex()
  92     {
  93         static LazyNeverDestroyed<std::mutex> syncMessageStateMapMutex;
  94         static std::once_flag onceFlag;
  95         std::call_once(onceFlag, [] {
  96             syncMessageStateMapMutex.construct();
  97         });
  98
  99         return syncMessageStateMapMutex;
 100     }
 101
 102     void dispatchMessageAndResetDidScheduleDispatchMessagesForConnection(Connection&);
 103
 104     RunLoop& m_runLoop;
 105     BinarySemaphore m_waitForSyncReplySemaphore;
 106
 107     // Protects m_didScheduleDispatchMessagesWorkSet and m_messagesToDispatchWhileWaitingForSyncReply.
 108     std::mutex m_mutex;
 109
 110     // The set of connections for which we've scheduled a call to dispatchMessageAndResetDidScheduleDispatchMessagesForConnection.
 111     HashSet<RefPtr<Connection>> m_didScheduleDispatchMessagesWorkSet;
 112
 113     struct ConnectionAndIncomingMessage {
 114         Ref<Connection> connection;
 115         std::unique_ptr<MessageDecoder> message;
 116     };
 117     Vector<ConnectionAndIncomingMessage> m_messagesToDispatchWhileWaitingForSyncReply;
 118 };
 119
 120 class Connection::SecondaryThreadPendingSyncReply {
 121 public:
 122     // The reply decoder, will be null if there was an error processing the sync message on the other side.
 123     std::unique_ptr<MessageDecoder> replyDecoder;
 124
 125     BinarySemaphore semaphore;
 126 };
 127
 128
 129 Ref<Connection::SyncMessageState> Connection::SyncMessageState::getOrCreate(RunLoop& runLoop)
 130 {
 131     std::lock_guard<std::mutex> lock(syncMessageStateMapMutex());
 132
 133     auto& slot = syncMessageStateMap().add(&runLoop, nullptr).iterator->value;
 134     if (slot)
 135         return *slot;
 136
 137     Ref<SyncMessageState> syncMessageState = adoptRef(*new SyncMessageState(runLoop));
 138     slot = syncMessageState.ptr();
 139
 140     return syncMessageState;
 141 }
 142
 143 Connection::SyncMessageState::SyncMessageState(RunLoop& runLoop)
 144     : m_runLoop(runLoop)
 145 {
 146 }
 147
 148 Connection::SyncMessageState::~SyncMessageState()
 149 {
 150     std::lock_guard<std::mutex> lock(syncMessageStateMapMutex());
 151
 152     ASSERT(syncMessageStateMap().contains(&m_runLoop));
 153     syncMessageStateMap().remove(&m_runLoop);
 154
 155     ASSERT(m_messagesToDispatchWhileWaitingForSyncReply.isEmpty());
 156 }
 157
 158 bool Connection::SyncMessageState::processIncomingMessage(Connection& connection, std::unique_ptr<MessageDecoder>& message)
 159 {
 160     if (!message->shouldDispatchMessageWhenWaitingForSyncReply())
 161         return false;
 162
 163     ConnectionAndIncomingMessage connectionAndIncomingMessage { connection, WTF::move(message) };
 164
 165     {
 166         std::lock_guard<std::mutex> lock(m_mutex);
 167
 168         if (m_didScheduleDispatchMessagesWorkSet.add(&connection).isNewEntry) {
 169             RefPtr<Connection> protectedConnection(&connection);
 170             m_runLoop.dispatch([this, protectedConnection] {
 171                 dispatchMessageAndResetDidScheduleDispatchMessagesForConnection(*protectedConnection);
 172             });
 173         }
 174
 175         m_messagesToDispatchWhileWaitingForSyncReply.append(WTF::move(connectionAndIncomingMessage));
 176     }
 177
 178     wakeUpClientRunLoop();
 179
 180     return true;
 181 }
 182
 183 void Connection::SyncMessageState::dispatchMessages(Connection* allowedConnection)
 184 {
 185     ASSERT(&m_runLoop == &RunLoop::current());
 186
 187     Vector<ConnectionAndIncomingMessage> messagesToDispatchWhileWaitingForSyncReply;
 188
 189     {
 190         std::lock_guard<std::mutex> lock(m_mutex);
 191         m_messagesToDispatchWhileWaitingForSyncReply.swap(messagesToDispatchWhileWaitingForSyncReply);
 192     }
 193
 194     Vector<ConnectionAndIncomingMessage> messagesToPutBack;
 195
 196     for (size_t i = 0; i < messagesToDispatchWhileWaitingForSyncReply.size(); ++i) {
 197         ConnectionAndIncomingMessage& connectionAndIncomingMessage = messagesToDispatchWhileWaitingForSyncReply[i];
 198
 199         if (allowedConnection && allowedConnection != connectionAndIncomingMessage.connection.ptr()) {
 200             // This incoming message belongs to another connection and we don't want to dispatch it now
 201             // so mark it to be put back in the message queue.
 202             messagesToPutBack.append(WTF::move(connectionAndIncomingMessage));
 203             continue;
 204         }
 205
 206         connectionAndIncomingMessage.connection->dispatchMessage(WTF::move(connectionAndIncomingMessage.message));
 207     }
 208
 209     if (!messagesToPutBack.isEmpty()) {
 210         std::lock_guard<std::mutex> lock(m_mutex);
 211
 212         for (auto& message : messagesToPutBack)
 213             m_messagesToDispatchWhileWaitingForSyncReply.append(WTF::move(message));
 214     }
 215 }
 216
 217 void Connection::SyncMessageState::dispatchMessageAndResetDidScheduleDispatchMessagesForConnection(Connection& connection)
 218 {
 219     {
 220         std::lock_guard<std::mutex> lock(m_mutex);
 221         ASSERT(m_didScheduleDispatchMessagesWorkSet.contains(&connection));
 222         m_didScheduleDispatchMessagesWorkSet.remove(&connection);
 223     }
 224
 225     dispatchMessages(&connection);
 226 }
 227
 228 Ref<Connection> Connection::createServerConnection(Identifier identifier, Client& client, RunLoop& clientRunLoop)
 229 {
 230     return adoptRef(*new Connection(identifier, true, client, clientRunLoop));
 231 }
 232
 233 Ref<Connection> Connection::createClientConnection(Identifier identifier, Client& client, RunLoop& clientRunLoop)
 234 {
 235     return adoptRef(*new Connection(identifier, false, client, clientRunLoop));
 236 }
 237
 238 Connection::Connection(Identifier identifier, bool isServer, Client& client, RunLoop& clientRunLoop)
 239     : m_client(&client)
 240     , m_isServer(isServer)
 241     , m_syncRequestID(0)
 242     , m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage(false)
 243     , m_shouldExitOnSyncMessageSendFailure(false)
 244     , m_didCloseOnConnectionWorkQueueCallback(0)
 245     , m_isConnected(false)
 246     , m_connectionQueue(WorkQueue::create("com.apple.IPC.ReceiveQueue"))
 247     , m_clientRunLoop(clientRunLoop)
 248     , m_inSendSyncCount(0)
 249     , m_inDispatchMessageCount(0)
 250     , m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount(0)
 251     , m_didReceiveInvalidMessage(false)
 252     , m_waitingForMessage(nullptr)
 253     , m_syncMessageState(SyncMessageState::getOrCreate(clientRunLoop))
 254     , m_shouldWaitForSyncReplies(true)
 255 {
 256     ASSERT(m_client);
 257
 258     platformInitialize(identifier);
 259 }
 260
 261 Connection::~Connection()
 262 {
 263     ASSERT(!isValid());
 264 }
 265
 266 void Connection::setOnlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage(bool flag)
 267 {
 268     ASSERT(!m_isConnected);
 269
 270     m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage = flag;
 271 }
 272
 273 void Connection::setShouldExitOnSyncMessageSendFailure(bool shouldExitOnSyncMessageSendFailure)
 274 {
 275     ASSERT(!m_isConnected);
 276
 277     m_shouldExitOnSyncMessageSendFailure = shouldExitOnSyncMessageSendFailure;
 278 }
 279
 280 void Connection::addWorkQueueMessageReceiver(StringReference messageReceiverName, WorkQueue* workQueue, WorkQueueMessageReceiver* workQueueMessageReceiver)
 281 {
 282     ASSERT(&RunLoop::current() == &m_clientRunLoop);
 283
 284     RefPtr<Connection> connection(this);
 285     m_connectionQueue->dispatch([connection, messageReceiverName, workQueue, workQueueMessageReceiver] {
 286         ASSERT(!connection->m_workQueueMessageReceivers.contains(messageReceiverName));
 287
 288         connection->m_workQueueMessageReceivers.add(messageReceiverName, std::make_pair(workQueue, workQueueMessageReceiver));
 289     });
 290 }
 291
 292 void Connection::removeWorkQueueMessageReceiver(StringReference messageReceiverName)
 293 {
 294     ASSERT(&RunLoop::current() == &m_clientRunLoop);
 295
 296     RefPtr<Connection> connection(this);
 297     m_connectionQueue->dispatch([connection, messageReceiverName] {
 298         ASSERT(connection->m_workQueueMessageReceivers.contains(messageReceiverName));
 299         connection->m_workQueueMessageReceivers.remove(messageReceiverName);
 300     });
 301 }
 302
 303 void Connection::dispatchWorkQueueMessageReceiverMessage(WorkQueueMessageReceiver& workQueueMessageReceiver, MessageDecoder& decoder)
 304 {
 305     if (!decoder.isSyncMessage()) {
 306         workQueueMessageReceiver.didReceiveMessage(*this, decoder);
 307         return;
 308     }
 309
 310     uint64_t syncRequestID = 0;
 311     if (!decoder.decode(syncRequestID) || !syncRequestID) {
 312         // We received an invalid sync message.
 313         // FIXME: Handle this.
 314         decoder.markInvalid();
 315         return;
 316     }
 317
 318 #if HAVE(DTRACE)
 319     auto replyEncoder = std::make_unique<MessageEncoder>("IPC", "SyncMessageReply", syncRequestID, decoder.UUID());
 320 #else
 321     auto replyEncoder = std::make_unique<MessageEncoder>("IPC", "SyncMessageReply", syncRequestID);
 322 #endif
 323
 324     // Hand off both the decoder and encoder to the work queue message receiver.
 325     workQueueMessageReceiver.didReceiveSyncMessage(*this, decoder, replyEncoder);
 326
 327     // FIXME: If the message was invalid, we should send back a SyncMessageError.
 328     ASSERT(!decoder.isInvalid());
 329
 330     if (replyEncoder)
 331         sendSyncReply(WTF::move(replyEncoder));
 332 }
 333
 334 void Connection::setDidCloseOnConnectionWorkQueueCallback(DidCloseOnConnectionWorkQueueCallback callback)
 335 {
 336     ASSERT(!m_isConnected);
 337
 338     m_didCloseOnConnectionWorkQueueCallback = callback;
 339 }
 340
 341 void Connection::invalidate()
 342 {
 343     if (!isValid()) {
 344         // Someone already called invalidate().
 345         return;
 346     }
 347
 348     m_client = nullptr;
 349
 350     RefPtr<Connection> protectedThis(this);
 351     m_connectionQueue->dispatch([protectedThis] {
 352         protectedThis->platformInvalidate();
 353     });
 354 }
 355
 356 void Connection::markCurrentlyDispatchedMessageAsInvalid()
 357 {
 358     // This should only be called while processing a message.
 359     ASSERT(m_inDispatchMessageCount > 0);
 360
 361     m_didReceiveInvalidMessage = true;
 362 }
 363
 364 std::unique_ptr<MessageEncoder> Connection::createSyncMessageEncoder(StringReference messageReceiverName, StringReference messageName, uint64_t destinationID, uint64_t& syncRequestID)
 365 {
 366     auto encoder = std::make_unique<MessageEncoder>(messageReceiverName, messageName, destinationID);
 367     encoder->setIsSyncMessage(true);
 368
 369     // Encode the sync request ID.
 370     syncRequestID = ++m_syncRequestID;
 371     *encoder << syncRequestID;
 372
 373     return encoder;
 374 }
 375
 376 bool Connection::sendMessage(std::unique_ptr<MessageEncoder> encoder, unsigned messageSendFlags, bool alreadyRecordedMessage)
 377 {
 378     if (!isValid())
 379         return false;
 380
 381     if (messageSendFlags & DispatchMessageEvenWhenWaitingForSyncReply
 382         && (!m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage
 383             || m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount))
 384         encoder->setShouldDispatchMessageWhenWaitingForSyncReply(true);
 385
 386 #if HAVE(DTRACE)
 387     std::unique_ptr<MessageRecorder::MessageProcessingToken> token;
 388     if (!alreadyRecordedMessage)
 389         token = MessageRecorder::recordOutgoingMessage(*this, *encoder);
 390 #else
 391     UNUSED_PARAM(alreadyRecordedMessage);
 392 #endif
 393
 394     {
 395         std::lock_guard<std::mutex> lock(m_outgoingMessagesMutex);
 396         m_outgoingMessages.append(WTF::move(encoder));
 397     }
 398
 399     // FIXME: We should add a boolean flag so we don't call this when work has already been scheduled.
 400     RefPtr<Connection> protectedThis(this);
 401     m_connectionQueue->dispatch([protectedThis] {
 402         protectedThis->sendOutgoingMessages();
 403     });
 404     return true;
 405 }
 406
 407 bool Connection::sendSyncReply(std::unique_ptr<MessageEncoder> encoder)
 408 {
 409     return sendMessage(WTF::move(encoder));
 410 }
 411
 412 std::unique_ptr<MessageDecoder> Connection::waitForMessage(StringReference messageReceiverName, StringReference messageName, uint64_t destinationID, std::chrono::milliseconds timeout, unsigned waitForMessageFlags)
 413 {
 414     ASSERT(&m_clientRunLoop == &RunLoop::current());
 415
 416     // First, check if this message is already in the incoming messages queue.
 417     {
 418         std::lock_guard<std::mutex> lock(m_incomingMessagesMutex);
 419
 420         for (auto it = m_incomingMessages.begin(), end = m_incomingMessages.end(); it != end; ++it) {
 421             std::unique_ptr<MessageDecoder>& message = *it;
 422
 423             if (message->messageReceiverName() == messageReceiverName && message->messageName() == messageName && message->destinationID() == destinationID) {
 424                 std::unique_ptr<MessageDecoder> returnedMessage = WTF::move(message);
 425
 426                 m_incomingMessages.remove(it);
 427                 return returnedMessage;
 428             }
 429         }
 430     }
 431
 432     WaitForMessageState waitingForMessage(messageReceiverName, messageName, destinationID, waitForMessageFlags);
 433
 434     {
 435         std::lock_guard<std::mutex> lock(m_waitForMessageMutex);
 436
 437         // We don't support having multiple clients waiting for messages.
 438         ASSERT(!m_waitingForMessage);
 439
 440         m_waitingForMessage = &waitingForMessage;
 441     }
 442
 443     // Now wait for it to be set.
 444     while (true) {
 445         std::unique_lock<std::mutex> lock(m_waitForMessageMutex);
 446
 447         if (m_waitingForMessage->decoder) {
 448             auto decoder = WTF::move(m_waitingForMessage->decoder);
 449             m_waitingForMessage = nullptr;
 450             return decoder;
 451         }
 452
 453         // Now we wait.
 454         std::cv_status status = m_waitForMessageCondition.wait_for(lock, timeout);
 455         // We timed out, lost our connection, or a sync message came in with InterruptWaitingIfSyncMessageArrives, so stop waiting.
 456         if (status == std::cv_status::timeout || m_waitingForMessage->messageWaitingInterrupted)
 457             break;
 458     }
 459
 460     m_waitingForMessage = nullptr;
 461
 462     return nullptr;
 463 }
 464
 465 std::unique_ptr<MessageDecoder> Connection::sendSyncMessage(uint64_t syncRequestID, std::unique_ptr<MessageEncoder> encoder, std::chrono::milliseconds timeout, unsigned syncSendFlags)
 466 {
 467     if (&RunLoop::current() != &m_clientRunLoop) {
 468         // No flags are supported for synchronous messages sent from secondary threads.
 469         ASSERT(!syncSendFlags);
 470         return sendSyncMessageFromSecondaryThread(syncRequestID, WTF::move(encoder), timeout);
 471     }
 472
 473     if (!isValid()) {
 474         didFailToSendSyncMessage();
 475         return nullptr;
 476     }
 477
 478     // Push the pending sync reply information on our stack.
 479     {
 480         MutexLocker locker(m_syncReplyStateMutex);
 481         if (!m_shouldWaitForSyncReplies) {
 482             didFailToSendSyncMessage();
 483             return nullptr;
 484         }
 485
 486         m_pendingSyncReplies.append(PendingSyncReply(syncRequestID));
 487     }
 488
 489     ++m_inSendSyncCount;
 490
 491 #if HAVE(DTRACE)
 492     auto token = MessageRecorder::recordOutgoingMessage(*this, *encoder);
 493 #endif
 494
 495     // First send the message.
 496     sendMessage(WTF::move(encoder), DispatchMessageEvenWhenWaitingForSyncReply, true);
 497
 498     // Then wait for a reply. Waiting for a reply could involve dispatching incoming sync messages, so
 499     // keep an extra reference to the connection here in case it's invalidated.
 500     Ref<Connection> protect(*this);
 501     std::unique_ptr<MessageDecoder> reply = waitForSyncReply(syncRequestID, timeout, syncSendFlags);
 502
 503     --m_inSendSyncCount;
 504
 505     // Finally, pop the pending sync reply information.
 506     {
 507         MutexLocker locker(m_syncReplyStateMutex);
 508         ASSERT(m_pendingSyncReplies.last().syncRequestID == syncRequestID);
 509         m_pendingSyncReplies.removeLast();
 510     }
 511
 512     if (!reply)
 513         didFailToSendSyncMessage();
 514
 515     return reply;
 516 }
 517
 518 std::unique_ptr<MessageDecoder> Connection::sendSyncMessageFromSecondaryThread(uint64_t syncRequestID, std::unique_ptr<MessageEncoder> encoder, std::chrono::milliseconds timeout)
 519 {
 520     ASSERT(&RunLoop::current() != &m_clientRunLoop);
 521
 522     if (!isValid())
 523         return nullptr;
 524
 525     SecondaryThreadPendingSyncReply pendingReply;
 526
 527     // Push the pending sync reply information on our stack.
 528     {
 529         MutexLocker locker(m_syncReplyStateMutex);
 530         if (!m_shouldWaitForSyncReplies)
 531             return nullptr;
 532
 533         ASSERT(!m_secondaryThreadPendingSyncReplyMap.contains(syncRequestID));
 534         m_secondaryThreadPendingSyncReplyMap.add(syncRequestID, &pendingReply);
 535     }
 536
 537 #if HAVE(DTRACE)
 538     auto token = MessageRecorder::recordOutgoingMessage(*this, *encoder);
 539 #endif
 540
 541     sendMessage(WTF::move(encoder), 0, true);
 542
 543     pendingReply.semaphore.wait(currentTime() + (timeout.count() / 1000.0));
 544
 545     // Finally, pop the pending sync reply information.
 546     {
 547         MutexLocker locker(m_syncReplyStateMutex);
 548         ASSERT(m_secondaryThreadPendingSyncReplyMap.contains(syncRequestID));
 549         m_secondaryThreadPendingSyncReplyMap.remove(syncRequestID);
 550     }
 551
 552     return WTF::move(pendingReply.replyDecoder);
 553 }
 554
 555 std::unique_ptr<MessageDecoder> Connection::waitForSyncReply(uint64_t syncRequestID, std::chrono::milliseconds timeout, unsigned syncSendFlags)
 556 {
 557     double absoluteTime = currentTime() + (timeout.count() / 1000.0);
 558
 559     willSendSyncMessage(syncSendFlags);
 560
 561     bool timedOut = false;
 562     while (!timedOut) {
 563         // First, check if we have any messages that we need to process.
 564         m_syncMessageState->dispatchMessages(nullptr);
 565
 566         {
 567             MutexLocker locker(m_syncReplyStateMutex);
 568
 569             // Second, check if there is a sync reply at the top of the stack.
 570             ASSERT(!m_pendingSyncReplies.isEmpty());
 571
 572             PendingSyncReply& pendingSyncReply = m_pendingSyncReplies.last();
 573             ASSERT_UNUSED(syncRequestID, pendingSyncReply.syncRequestID == syncRequestID);
 574
 575             // We found the sync reply, or the connection was closed.
 576             if (pendingSyncReply.didReceiveReply || !m_shouldWaitForSyncReplies) {
 577                 didReceiveSyncReply(syncSendFlags);
 578                 return WTF::move(pendingSyncReply.replyDecoder);
 579             }
 580         }
 581
 582         // Processing a sync message could cause the connection to be invalidated.
 583         // (If the handler ends up calling Connection::invalidate).
 584         // If that happens, we need to stop waiting, or we'll hang since we won't get
 585         // any more incoming messages.
 586         if (!isValid()) {
 587             didReceiveSyncReply(syncSendFlags);
 588             return nullptr;
 589         }
 590
 591         // We didn't find a sync reply yet, keep waiting.
 592         // This allows the WebProcess to still serve clients while waiting for the message to return.
 593         // Notably, it can continue to process accessibility requests, which are on the main thread.
 594         if (syncSendFlags & SpinRunLoopWhileWaitingForReply) {
 595 #if PLATFORM(COCOA)
 596             // FIXME: Although we run forever, any events incoming will cause us to drop out and exit out. This however doesn't
 597             // account for a timeout value passed in. Timeout is always NoTimeout in these cases, but that could change.
 598             RunLoop::current().runForDuration(1e10);
 599             timedOut = currentTime() >= absoluteTime;
 600 #endif
 601         } else
 602             timedOut = !m_syncMessageState->wait(absoluteTime);
 603
 604     }
 605
 606     didReceiveSyncReply(syncSendFlags);
 607
 608     return nullptr;
 609 }
 610
 611 void Connection::processIncomingSyncReply(std::unique_ptr<MessageDecoder> decoder)
 612 {
 613     MutexLocker locker(m_syncReplyStateMutex);
 614
 615     // Go through the stack of sync requests that have pending replies and see which one
 616     // this reply is for.
 617     for (size_t i = m_pendingSyncReplies.size(); i > 0; --i) {
 618         PendingSyncReply& pendingSyncReply = m_pendingSyncReplies[i - 1];
 619
 620         if (pendingSyncReply.syncRequestID != decoder->destinationID())
 621             continue;
 622
 623         ASSERT(!pendingSyncReply.replyDecoder);
 624
 625         pendingSyncReply.replyDecoder = WTF::move(decoder);
 626         pendingSyncReply.didReceiveReply = true;
 627
 628         // We got a reply to the last send message, wake up the client run loop so it can be processed.
 629         if (i == m_pendingSyncReplies.size())
 630             m_syncMessageState->wakeUpClientRunLoop();
 631
 632         return;
 633     }
 634
 635     // If it's not a reply to any primary thread message, check if it is a reply to a secondary thread one.
 636     SecondaryThreadPendingSyncReplyMap::iterator secondaryThreadReplyMapItem = m_secondaryThreadPendingSyncReplyMap.find(decoder->destinationID());
 637     if (secondaryThreadReplyMapItem != m_secondaryThreadPendingSyncReplyMap.end()) {
 638         SecondaryThreadPendingSyncReply* reply = secondaryThreadReplyMapItem->value;
 639         ASSERT(!reply->replyDecoder);
 640         reply->replyDecoder = WTF::move(decoder);
 641         reply->semaphore.signal();
 642     }
 643
 644     // If we get here, it means we got a reply for a message that wasn't in the sync request stack or map.
 645     // This can happen if the send timed out, so it's fine to ignore.
 646 }
 647
 648 void Connection::processIncomingMessage(std::unique_ptr<MessageDecoder> message)
 649 {
 650     ASSERT(!message->messageReceiverName().isEmpty());
 651     ASSERT(!message->messageName().isEmpty());
 652
 653     if (message->messageReceiverName() == "IPC" && message->messageName() == "SyncMessageReply") {
 654         processIncomingSyncReply(WTF::move(message));
 655         return;
 656     }
 657
 658     if (!m_workQueueMessageReceivers.isValidKey(message->messageReceiverName())) {
 659         RefPtr<Connection> protectedThis(this);
 660         StringReference messageReceiverName = message->messageReceiverName();
 661         StringCapture capturedMessageReceiverName(messageReceiverName.isEmpty() ? "<unknown message receiver>" : String(messageReceiverName.data(), messageReceiverName.size()));
 662         StringReference messageName = message->messageName();
 663         StringCapture capturedMessageName(messageName.isEmpty() ? "<unknown message>" : String(messageName.data(), messageName.size()));
 664
 665         m_clientRunLoop.dispatch([protectedThis, capturedMessageReceiverName, capturedMessageName] {
 666             protectedThis->dispatchDidReceiveInvalidMessage(capturedMessageReceiverName.string().utf8(), capturedMessageName.string().utf8());
 667         });
 668         return;
 669     }
 670
 671     auto it = m_workQueueMessageReceivers.find(message->messageReceiverName());
 672     if (it != m_workQueueMessageReceivers.end()) {
 673         RefPtr<Connection> protectedThis(this);
 674         RefPtr<WorkQueueMessageReceiver>& workQueueMessageReceiver = it->value.second;
 675         MessageDecoder* decoderPtr = message.release();
 676         it->value.first->dispatch([protectedThis, workQueueMessageReceiver, decoderPtr] {
 677             std::unique_ptr<MessageDecoder> decoder(decoderPtr);
 678             protectedThis->dispatchWorkQueueMessageReceiverMessage(*workQueueMessageReceiver, *decoder);
 679         });
 680         return;
 681     }
 682
 683     // Check if this is a sync message or if it's a message that should be dispatched even when waiting for
 684     // a sync reply. If it is, and we're waiting for a sync reply this message needs to be dispatched.
 685     // If we don't we'll end up with a deadlock where both sync message senders are stuck waiting for a reply.
 686     if (m_syncMessageState->processIncomingMessage(*this, message))
 687         return;
 688
 689     // Check if we're waiting for this message.
 690     {
 691         std::lock_guard<std::mutex> lock(m_waitForMessageMutex);
 692
 693         if (m_waitingForMessage && m_waitingForMessage->messageReceiverName == message->messageReceiverName() && m_waitingForMessage->messageName == message->messageName() && m_waitingForMessage->destinationID == message->destinationID()) {
 694             m_waitingForMessage->decoder = WTF::move(message);
 695             ASSERT(m_waitingForMessage->decoder);
 696             m_waitForMessageCondition.notify_one();
 697             return;
 698         }
 699
 700         if (m_waitingForMessage && (m_waitingForMessage->waitForMessageFlags & InterruptWaitingIfSyncMessageArrives) && message->isSyncMessage()) {
 701             m_waitingForMessage->messageWaitingInterrupted = true;
 702             m_waitForMessageCondition.notify_one();
 703         }
 704     }
 705
 706     enqueueIncomingMessage(WTF::move(message));
 707 }
 708
 709 void Connection::postConnectionDidCloseOnConnectionWorkQueue()
 710 {
 711     RefPtr<Connection> connection(this);
 712     m_connectionQueue->dispatch([connection] {
 713         connection->connectionDidClose();
 714     });
 715 }
 716
 717 void Connection::connectionDidClose()
 718 {
 719     // The connection is now invalid.
 720     platformInvalidate();
 721
 722     {
 723         MutexLocker locker(m_syncReplyStateMutex);
 724
 725         ASSERT(m_shouldWaitForSyncReplies);
 726         m_shouldWaitForSyncReplies = false;
 727
 728         if (!m_pendingSyncReplies.isEmpty())
 729             m_syncMessageState->wakeUpClientRunLoop();
 730
 731         for (SecondaryThreadPendingSyncReplyMap::iterator iter = m_secondaryThreadPendingSyncReplyMap.begin(); iter != m_secondaryThreadPendingSyncReplyMap.end(); ++iter)
 732             iter->value->semaphore.signal();
 733     }
 734
 735     {
 736         std::lock_guard<std::mutex> lock(m_waitForMessageMutex);
 737         if (m_waitingForMessage)
 738             m_waitingForMessage->messageWaitingInterrupted = true;
 739     }
 740     m_waitForMessageCondition.notify_all();
 741
 742     if (m_didCloseOnConnectionWorkQueueCallback)
 743         m_didCloseOnConnectionWorkQueueCallback(this);
 744
 745     RefPtr<Connection> connection(this);
 746     m_clientRunLoop.dispatch([connection] {
 747         // If the connection has been explicitly invalidated before dispatchConnectionDidClose was called,
 748         // then the client will be null here.
 749         if (!connection->m_client)
 750             return;
 751
 752         // Because we define a connection as being "valid" based on wheter it has a null client, we null out
 753         // the client before calling didClose here. Otherwise, sendSync will try to send a message to the connection and
 754         // will then wait indefinitely for a reply.
 755         Client* client = connection->m_client;
 756         connection->m_client = nullptr;
 757
 758         client->didClose(*connection);
 759     });
 760 }
 761
 762 bool Connection::canSendOutgoingMessages() const
 763 {
 764     return m_isConnected && platformCanSendOutgoingMessages();
 765 }
 766
 767 void Connection::sendOutgoingMessages()
 768 {
 769     if (!canSendOutgoingMessages())
 770         return;
 771
 772     while (true) {
 773         std::unique_ptr<MessageEncoder> message;
 774
 775         {
 776             std::lock_guard<std::mutex> lock(m_outgoingMessagesMutex);
 777             if (m_outgoingMessages.isEmpty())
 778                 break;
 779             message = m_outgoingMessages.takeFirst();
 780         }
 781
 782         if (!sendOutgoingMessage(WTF::move(message)))
 783             break;
 784     }
 785 }
 786
 787 void Connection::dispatchSyncMessage(MessageDecoder& decoder)
 788 {
 789     ASSERT(decoder.isSyncMessage());
 790
 791     uint64_t syncRequestID = 0;
 792     if (!decoder.decode(syncRequestID) || !syncRequestID) {
 793         // We received an invalid sync message.
 794         decoder.markInvalid();
 795         return;
 796     }
 797
 798 #if HAVE(DTRACE)
 799     auto replyEncoder = std::make_unique<MessageEncoder>("IPC", "SyncMessageReply", syncRequestID, decoder.UUID());
 800 #else
 801     auto replyEncoder = std::make_unique<MessageEncoder>("IPC", "SyncMessageReply", syncRequestID);
 802 #endif
 803
 804     // Hand off both the decoder and encoder to the client.
 805     m_client->didReceiveSyncMessage(*this, decoder, replyEncoder);
 806
 807     // FIXME: If the message was invalid, we should send back a SyncMessageError.
 808     ASSERT(!decoder.isInvalid());
 809
 810     if (replyEncoder)
 811         sendSyncReply(WTF::move(replyEncoder));
 812 }
 813
 814 void Connection::dispatchDidReceiveInvalidMessage(const CString& messageReceiverNameString, const CString& messageNameString)
 815 {
 816     ASSERT(&RunLoop::current() == &m_clientRunLoop);
 817
 818     if (!m_client)
 819         return;
 820
 821     m_client->didReceiveInvalidMessage(*this, StringReference(messageReceiverNameString.data(), messageReceiverNameString.length()), StringReference(messageNameString.data(), messageNameString.length()));
 822 }
 823
 824 void Connection::didFailToSendSyncMessage()
 825 {
 826     if (!m_shouldExitOnSyncMessageSendFailure)
 827         return;
 828
 829     exit(0);
 830 }
 831
 832 void Connection::enqueueIncomingMessage(std::unique_ptr<MessageDecoder> incomingMessage)
 833 {
 834     {
 835         std::lock_guard<std::mutex> lock(m_incomingMessagesMutex);
 836         m_incomingMessages.append(WTF::move(incomingMessage));
 837     }
 838
 839     RefPtr<Connection> protectedThis(this);
 840     m_clientRunLoop.dispatch([protectedThis] {
 841         protectedThis->dispatchOneMessage();
 842     });
 843 }
 844
 845 void Connection::dispatchMessage(MessageDecoder& decoder)
 846 {
 847     m_client->didReceiveMessage(*this, decoder);
 848 }
 849
 850 void Connection::dispatchMessage(std::unique_ptr<MessageDecoder> message)
 851 {
 852 #if HAVE(DTRACE)
 853     MessageRecorder::recordIncomingMessage(*this, *message);
 854 #endif
 855
 856     if (!m_client)
 857         return;
 858
 859     m_inDispatchMessageCount++;
 860
 861     if (message->shouldDispatchMessageWhenWaitingForSyncReply())
 862         m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount++;
 863
 864     bool oldDidReceiveInvalidMessage = m_didReceiveInvalidMessage;
 865     m_didReceiveInvalidMessage = false;
 866
 867     if (message->isSyncMessage())
 868         dispatchSyncMessage(*message);
 869     else
 870         dispatchMessage(*message);
 871
 872     m_didReceiveInvalidMessage |= message->isInvalid();
 873     m_inDispatchMessageCount--;
 874
 875     // FIXME: For Delayed synchronous messages, we should not decrement the counter until we send a response.
 876     // Otherwise, we would deadlock if processing the message results in a sync message back after we exit this function.
 877     if (message->shouldDispatchMessageWhenWaitingForSyncReply())
 878         m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount--;
 879
 880     if (m_didReceiveInvalidMessage && m_client)
 881         m_client->didReceiveInvalidMessage(*this, message->messageReceiverName(), message->messageName());
 882
 883     m_didReceiveInvalidMessage = oldDidReceiveInvalidMessage;
 884 }
 885
 886 void Connection::dispatchOneMessage()
 887 {
 888     std::unique_ptr<MessageDecoder> message;
 889
 890     {
 891         std::lock_guard<std::mutex> lock(m_incomingMessagesMutex);
 892         if (m_incomingMessages.isEmpty())
 893             return;
 894
 895         message = m_incomingMessages.takeFirst();
 896     }
 897
 898     dispatchMessage(WTF::move(message));
 899 }
 900
 901 void Connection::wakeUpRunLoop()
 902 {
 903     m_clientRunLoop.wakeUp();
 904 }
 905
 906 } // namespace IPC