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