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