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