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