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