Unreviewed, roll out http://trac.webkit.org/changeset/187972.
[WebKit-https.git] / 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         MutexLocker 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         MutexLocker 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         MutexLocker 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         MutexLocker 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             MutexLocker 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     MutexLocker 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         MutexLocker 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