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