2011-02-11 Anders Carlsson <andersca@apple.com>
[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 "BinarySemaphore.h"
30 #include "CoreIPCMessageKinds.h"
31 #include "RunLoop.h"
32 #include "WorkItem.h"
33 #include <wtf/CurrentTime.h>
34
35 using namespace std;
36
37 namespace CoreIPC {
38
39 class Connection::SyncMessageState : public RefCounted<Connection::SyncMessageState> {
40 public:
41     static PassRefPtr<SyncMessageState> getOrCreate(RunLoop*);
42     ~SyncMessageState();
43
44     void beginWaitForSyncReply();
45     void endWaitForSyncReply();
46
47     void wakeUpClientRunLoop()
48     {
49         m_waitForSyncReplySemaphore.signal();
50     }
51
52     bool wait(double absoluteTime)
53     {
54         return m_waitForSyncReplySemaphore.wait(absoluteTime);
55     }
56
57     // Returns true if this message will be handled on a client thread that is currently
58     // waiting for a reply to a synchronous message.
59     bool processIncomingMessage(Connection*, IncomingMessage&);
60
61     void dispatchMessages();
62
63 private:
64     explicit SyncMessageState(RunLoop*);
65
66     typedef HashMap<RunLoop*, SyncMessageState*> SyncMessageStateMap;
67     static SyncMessageStateMap& syncMessageStateMap()
68     {
69         DEFINE_STATIC_LOCAL(SyncMessageStateMap, syncMessageStateMap, ());
70         return syncMessageStateMap;
71     }
72
73     static Mutex& syncMessageStateMapMutex()
74     {
75         DEFINE_STATIC_LOCAL(Mutex, syncMessageStateMapMutex, ());
76         return syncMessageStateMapMutex;
77     }
78
79     RunLoop* m_runLoop;
80     BinarySemaphore m_waitForSyncReplySemaphore;
81
82     // Protects m_waitForSyncReplyCount and m_messagesToDispatchWhileWaitingForSyncReply.
83     Mutex m_mutex;
84
85     unsigned m_waitForSyncReplyCount;
86
87     struct ConnectionAndIncomingMessage {
88         Connection* connection;
89         IncomingMessage incomingMessage;
90     };
91     Vector<ConnectionAndIncomingMessage> m_messagesToDispatchWhileWaitingForSyncReply;
92 };
93
94 PassRefPtr<Connection::SyncMessageState> Connection::SyncMessageState::getOrCreate(RunLoop* runLoop)
95 {
96     MutexLocker locker(syncMessageStateMapMutex());
97     pair<SyncMessageStateMap::iterator, bool> result = syncMessageStateMap().add(runLoop, 0);
98
99     if (!result.second) {
100         ASSERT(result.first->second);
101         return result.first->second;
102     }
103
104     RefPtr<SyncMessageState> syncMessageState = adoptRef(new SyncMessageState(runLoop));
105     result.first->second = syncMessageState.get();
106
107     return syncMessageState.release();
108 }
109
110 Connection::SyncMessageState::SyncMessageState(RunLoop* runLoop)
111     : m_runLoop(runLoop)
112     , m_waitForSyncReplyCount(0)
113 {
114 }
115
116 Connection::SyncMessageState::~SyncMessageState()
117 {
118     MutexLocker locker(syncMessageStateMapMutex());
119     
120     ASSERT(syncMessageStateMap().contains(m_runLoop));
121     syncMessageStateMap().remove(m_runLoop);
122 }
123
124 void Connection::SyncMessageState::beginWaitForSyncReply()
125 {
126     ASSERT(RunLoop::current() == m_runLoop);
127
128     MutexLocker locker(m_mutex);
129     m_waitForSyncReplyCount++;
130 }
131
132 void Connection::SyncMessageState::endWaitForSyncReply()
133 {
134     ASSERT(RunLoop::current() == m_runLoop);
135
136     MutexLocker locker(m_mutex);
137     ASSERT(m_waitForSyncReplyCount);
138     --m_waitForSyncReplyCount;
139
140     if (m_waitForSyncReplyCount)
141         return;
142
143     // Dispatch any remaining incoming sync messages.
144     for (size_t i = 0; i < m_messagesToDispatchWhileWaitingForSyncReply.size(); ++i) {
145         ConnectionAndIncomingMessage& connectionAndIncomingMessage = m_messagesToDispatchWhileWaitingForSyncReply[i];
146         connectionAndIncomingMessage.connection->enqueueIncomingMessage(connectionAndIncomingMessage.incomingMessage);
147     }
148
149     m_messagesToDispatchWhileWaitingForSyncReply.clear();
150 }
151
152 bool Connection::SyncMessageState::processIncomingMessage(Connection* connection, IncomingMessage& incomingMessage)
153 {
154     MessageID messageID = incomingMessage.messageID();
155     if (!messageID.isSync() && !messageID.shouldDispatchMessageWhenWaitingForSyncReply())
156         return false;
157
158     MutexLocker locker(m_mutex);
159     if (!m_waitForSyncReplyCount)
160         return false;
161
162     ConnectionAndIncomingMessage connectionAndIncomingMessage;
163     connectionAndIncomingMessage.connection = connection;
164     connectionAndIncomingMessage.incomingMessage = incomingMessage;
165
166     m_messagesToDispatchWhileWaitingForSyncReply.append(connectionAndIncomingMessage);
167     wakeUpClientRunLoop();
168
169     return true;
170 }
171
172 void Connection::SyncMessageState::dispatchMessages()
173 {
174     ASSERT(m_runLoop == RunLoop::current());
175
176     Vector<ConnectionAndIncomingMessage> messagesToDispatchWhileWaitingForSyncReply;
177
178     {
179         MutexLocker locker(m_mutex);
180         m_messagesToDispatchWhileWaitingForSyncReply.swap(messagesToDispatchWhileWaitingForSyncReply);
181     }
182
183     for (size_t i = 0; i < messagesToDispatchWhileWaitingForSyncReply.size(); ++i) {
184         ConnectionAndIncomingMessage& connectionAndIncomingMessage = messagesToDispatchWhileWaitingForSyncReply[i];
185         connectionAndIncomingMessage.connection->dispatchMessage(connectionAndIncomingMessage.incomingMessage);
186     }
187 }
188
189 PassRefPtr<Connection> Connection::createServerConnection(Identifier identifier, Client* client, RunLoop* clientRunLoop)
190 {
191     return adoptRef(new Connection(identifier, true, client, clientRunLoop));
192 }
193
194 PassRefPtr<Connection> Connection::createClientConnection(Identifier identifier, Client* client, RunLoop* clientRunLoop)
195 {
196     return adoptRef(new Connection(identifier, false, client, clientRunLoop));
197 }
198
199 Connection::Connection(Identifier identifier, bool isServer, Client* client, RunLoop* clientRunLoop)
200     : m_client(client)
201     , m_isServer(isServer)
202     , m_syncRequestID(0)
203     , m_didCloseOnConnectionWorkQueueCallback(0)
204     , m_isConnected(false)
205     , m_connectionQueue("com.apple.CoreIPC.ReceiveQueue")
206     , m_clientRunLoop(clientRunLoop)
207     , m_inDispatchMessageCount(0)
208     , m_didReceiveInvalidMessage(false)
209     , m_syncMessageState(SyncMessageState::getOrCreate(clientRunLoop))
210     , m_shouldWaitForSyncReplies(true)
211 {
212     ASSERT(m_client);
213
214     platformInitialize(identifier);
215 }
216
217 Connection::~Connection()
218 {
219     ASSERT(!isValid());
220
221     m_connectionQueue.invalidate();
222 }
223
224 void Connection::setDidCloseOnConnectionWorkQueueCallback(DidCloseOnConnectionWorkQueueCallback callback)
225 {
226     ASSERT(!m_isConnected);
227
228     m_didCloseOnConnectionWorkQueueCallback = callback;    
229 }
230
231 void Connection::invalidate()
232 {
233     if (!isValid()) {
234         // Someone already called invalidate().
235         return;
236     }
237     
238     // Reset the client.
239     m_client = 0;
240
241     m_connectionQueue.scheduleWork(WorkItem::create(this, &Connection::platformInvalidate));
242 }
243
244 void Connection::markCurrentlyDispatchedMessageAsInvalid()
245 {
246     // This should only be called while processing a message.
247     ASSERT(m_inDispatchMessageCount > 0);
248
249     m_didReceiveInvalidMessage = true;
250 }
251
252 PassOwnPtr<ArgumentEncoder> Connection::createSyncMessageArgumentEncoder(uint64_t destinationID, uint64_t& syncRequestID)
253 {
254     OwnPtr<ArgumentEncoder> argumentEncoder = ArgumentEncoder::create(destinationID);
255
256     // Encode the sync request ID.
257     syncRequestID = ++m_syncRequestID;
258     argumentEncoder->encode(syncRequestID);
259
260     return argumentEncoder.release();
261 }
262
263 bool Connection::sendMessage(MessageID messageID, PassOwnPtr<ArgumentEncoder> arguments, unsigned messageSendFlags)
264 {
265     if (!isValid())
266         return false;
267
268     if (messageSendFlags & DispatchMessageEvenWhenWaitingForSyncReply)
269         messageID = messageID.messageIDWithAddedFlags(MessageID::DispatchMessageWhenWaitingForSyncReply);
270
271     MutexLocker locker(m_outgoingMessagesLock);
272     m_outgoingMessages.append(OutgoingMessage(messageID, arguments));
273     
274     // FIXME: We should add a boolean flag so we don't call this when work has already been scheduled.
275     m_connectionQueue.scheduleWork(WorkItem::create(this, &Connection::sendOutgoingMessages));
276     return true;
277 }
278
279 bool Connection::sendSyncReply(PassOwnPtr<ArgumentEncoder> arguments)
280 {
281     return sendMessage(MessageID(CoreIPCMessage::SyncMessageReply), arguments);
282 }
283
284 PassOwnPtr<ArgumentDecoder> Connection::waitForMessage(MessageID messageID, uint64_t destinationID, double timeout)
285 {
286     // First, check if this message is already in the incoming messages queue.
287     {
288         MutexLocker locker(m_incomingMessagesLock);
289
290         for (size_t i = 0; i < m_incomingMessages.size(); ++i) {
291             const IncomingMessage& message = m_incomingMessages[i];
292
293             if (message.messageID() == messageID && message.arguments()->destinationID() == destinationID) {
294                 OwnPtr<ArgumentDecoder> arguments(message.arguments());
295                 
296                 // Erase the incoming message.
297                 m_incomingMessages.remove(i);
298                 return arguments.release();
299             }
300         }
301     }
302     
303     double absoluteTime = currentTime() + timeout;
304     
305     std::pair<unsigned, uint64_t> messageAndDestination(std::make_pair(messageID.toInt(), destinationID));
306     
307     {
308         MutexLocker locker(m_waitForMessageMutex);
309
310         // We don't support having multiple clients wait for the same message.
311         ASSERT(!m_waitForMessageMap.contains(messageAndDestination));
312     
313         // Insert our pending wait.
314         m_waitForMessageMap.set(messageAndDestination, 0);
315     }
316     
317     // Now wait for it to be set.
318     while (true) {
319         MutexLocker locker(m_waitForMessageMutex);
320
321         HashMap<std::pair<unsigned, uint64_t>, ArgumentDecoder*>::iterator it = m_waitForMessageMap.find(messageAndDestination);
322         if (it->second) {
323             OwnPtr<ArgumentDecoder> arguments(it->second);
324             m_waitForMessageMap.remove(it);
325             
326             return arguments.release();
327         }
328         
329         // Now we wait.
330         if (!m_waitForMessageCondition.timedWait(m_waitForMessageMutex, absoluteTime)) {
331             // We timed out, now remove the pending wait.
332             m_waitForMessageMap.remove(messageAndDestination);
333
334             break;
335         }
336     }
337     
338     return PassOwnPtr<ArgumentDecoder>();
339 }
340
341 PassOwnPtr<ArgumentDecoder> Connection::sendSyncMessage(MessageID messageID, uint64_t syncRequestID, PassOwnPtr<ArgumentEncoder> encoder, double timeout)
342 {
343     // We only allow sending sync messages from the client run loop.
344     ASSERT(RunLoop::current() == m_clientRunLoop);
345
346     if (!isValid()) {
347         m_client->didFailToSendSyncMessage(this);
348         return 0;
349     }
350     
351     // Push the pending sync reply information on our stack.
352     {
353         MutexLocker locker(m_syncReplyStateMutex);
354         if (!m_shouldWaitForSyncReplies) {
355             m_client->didFailToSendSyncMessage(this);
356             return 0;
357         }
358
359         m_pendingSyncReplies.append(PendingSyncReply(syncRequestID));
360     }
361     
362     // First send the message.
363     sendMessage(messageID, encoder);
364
365     m_syncMessageState->beginWaitForSyncReply();
366
367     // Then wait for a reply. Waiting for a reply could involve dispatching incoming sync messages, so
368     // keep an extra reference to the connection here in case it's invalidated.
369     RefPtr<Connection> protect(this);
370     OwnPtr<ArgumentDecoder> reply = waitForSyncReply(syncRequestID, timeout);
371
372     // Finally, pop the pending sync reply information.
373     {
374         MutexLocker locker(m_syncReplyStateMutex);
375         ASSERT(m_pendingSyncReplies.last().syncRequestID == syncRequestID);
376         m_pendingSyncReplies.removeLast();
377     }
378
379     m_syncMessageState->endWaitForSyncReply();
380
381     if (!reply)
382         m_client->didFailToSendSyncMessage(this);
383
384     return reply.release();
385 }
386
387 PassOwnPtr<ArgumentDecoder> Connection::waitForSyncReply(uint64_t syncRequestID, double timeout)
388 {
389     double absoluteTime = currentTime() + timeout;
390
391     bool timedOut = false;
392     while (!timedOut) {
393         // First, check if we have any messages that we need to process.
394         m_syncMessageState->dispatchMessages();
395         
396         {
397             MutexLocker locker(m_syncReplyStateMutex);
398
399             // Second, check if there is a sync reply at the top of the stack.
400             ASSERT(!m_pendingSyncReplies.isEmpty());
401             
402             PendingSyncReply& pendingSyncReply = m_pendingSyncReplies.last();
403             ASSERT(pendingSyncReply.syncRequestID == syncRequestID);
404             
405             // We found the sync reply, or the connection was closed.
406             if (pendingSyncReply.didReceiveReply || !m_shouldWaitForSyncReplies)
407                 return pendingSyncReply.releaseReplyDecoder();
408         }
409
410         // We didn't find a sync reply yet, keep waiting.
411         timedOut = !m_syncMessageState->wait(absoluteTime);
412     }
413
414     // We timed out.
415     return 0;
416 }
417
418 void Connection::processIncomingMessage(MessageID messageID, PassOwnPtr<ArgumentDecoder> arguments)
419 {
420     // Check if this is a sync reply.
421     if (messageID == MessageID(CoreIPCMessage::SyncMessageReply)) {
422         MutexLocker locker(m_syncReplyStateMutex);
423         ASSERT(!m_pendingSyncReplies.isEmpty());
424
425         PendingSyncReply& pendingSyncReply = m_pendingSyncReplies.last();
426         ASSERT(pendingSyncReply.syncRequestID == arguments->destinationID());
427
428         pendingSyncReply.replyDecoder = arguments.leakPtr();
429         pendingSyncReply.didReceiveReply = true;
430         m_syncMessageState->wakeUpClientRunLoop();
431         return;
432     }
433
434     IncomingMessage incomingMessage(messageID, arguments);
435
436     // Check if this is a sync message or if it's a message that should be dispatched even when waiting for
437     // a sync reply. If it is, and we're waiting for a sync reply this message needs to be dispatched.
438     // If we don't we'll end up with a deadlock where both sync message senders are stuck waiting for a reply.
439     if (m_syncMessageState->processIncomingMessage(this, incomingMessage))
440         return;
441
442     // Check if we're waiting for this message.
443     {
444         MutexLocker locker(m_waitForMessageMutex);
445         
446         HashMap<std::pair<unsigned, uint64_t>, ArgumentDecoder*>::iterator it = m_waitForMessageMap.find(std::make_pair(messageID.toInt(), incomingMessage.destinationID()));
447         if (it != m_waitForMessageMap.end()) {
448             it->second = arguments.leakPtr();
449         
450             m_waitForMessageCondition.signal();
451             return;
452         }
453     }
454
455     enqueueIncomingMessage(incomingMessage);
456 }
457
458 void Connection::connectionDidClose()
459 {
460     // The connection is now invalid.
461     platformInvalidate();
462
463     {
464         MutexLocker locker(m_syncReplyStateMutex);
465
466         ASSERT(m_shouldWaitForSyncReplies);
467         m_shouldWaitForSyncReplies = false;
468
469         if (!m_pendingSyncReplies.isEmpty())
470             m_syncMessageState->wakeUpClientRunLoop();
471     }
472
473     if (m_didCloseOnConnectionWorkQueueCallback)
474         m_didCloseOnConnectionWorkQueueCallback(m_connectionQueue, this);
475
476     m_clientRunLoop->scheduleWork(WorkItem::create(this, &Connection::dispatchConnectionDidClose));
477 }
478
479 void Connection::dispatchConnectionDidClose()
480 {
481     // If the connection has been explicitly invalidated before dispatchConnectionDidClose was called,
482     // then the client will be null here.
483     if (!m_client)
484         return;
485
486
487     // Because we define a connection as being "valid" based on wheter it has a null client, we null out
488     // the client before calling didClose here. Otherwise, sendSync will try to send a message to the connection and
489     // will then wait indefinitely for a reply.
490     Client* client = m_client;
491     m_client = 0;
492     
493     client->didClose(this);
494 }
495
496 bool Connection::canSendOutgoingMessages() const
497 {
498     return m_isConnected && platformCanSendOutgoingMessages();
499 }
500
501 void Connection::sendOutgoingMessages()
502 {
503     if (!canSendOutgoingMessages())
504         return;
505
506     while (true) {
507         OutgoingMessage message;
508         {
509             MutexLocker locker(m_outgoingMessagesLock);
510             if (m_outgoingMessages.isEmpty())
511                 break;
512             message = m_outgoingMessages.takeFirst();
513         }
514
515         if (!sendOutgoingMessage(message.messageID(), adoptPtr(message.arguments())))
516             break;
517     }
518 }
519
520 void Connection::dispatchSyncMessage(MessageID messageID, ArgumentDecoder* arguments)
521 {
522     ASSERT(messageID.isSync());
523
524     // Decode the sync request ID.
525     uint64_t syncRequestID = 0;
526
527     if (!arguments->decodeUInt64(syncRequestID) || !syncRequestID) {
528         // We received an invalid sync message.
529         arguments->markInvalid();
530         return;
531     }
532
533     // Create our reply encoder.
534     ArgumentEncoder* replyEncoder = ArgumentEncoder::create(syncRequestID).leakPtr();
535     
536     // Hand off both the decoder and encoder to the client..
537     SyncReplyMode syncReplyMode = m_client->didReceiveSyncMessage(this, messageID, arguments, replyEncoder);
538
539     // FIXME: If the message was invalid, we should send back a SyncMessageError.
540     ASSERT(!arguments->isInvalid());
541
542     if (syncReplyMode == ManualReply) {
543         // The client will take ownership of the reply encoder and send it at some point in the future.
544         // We won't do anything here.
545         return;
546     }
547
548     // Send the reply.
549     sendSyncReply(replyEncoder);
550 }
551
552 void Connection::enqueueIncomingMessage(IncomingMessage& incomingMessage)
553 {
554     MutexLocker locker(m_incomingMessagesLock);
555     m_incomingMessages.append(incomingMessage);
556
557     m_clientRunLoop->scheduleWork(WorkItem::create(this, &Connection::dispatchMessages));
558 }
559
560 void Connection::dispatchMessage(IncomingMessage& message)
561 {
562     OwnPtr<ArgumentDecoder> arguments = message.releaseArguments();
563
564     // If there's no client, return. We do this after calling releaseArguments so that
565     // the ArgumentDecoder message will be freed.
566     if (!m_client)
567         return;
568
569     m_inDispatchMessageCount++;
570
571     bool oldDidReceiveInvalidMessage = m_didReceiveInvalidMessage;
572     m_didReceiveInvalidMessage = false;
573
574     if (message.messageID().isSync())
575         dispatchSyncMessage(message.messageID(), arguments.get());
576     else
577         m_client->didReceiveMessage(this, message.messageID(), arguments.get());
578
579     m_didReceiveInvalidMessage |= arguments->isInvalid();
580     m_inDispatchMessageCount--;
581
582     if (m_didReceiveInvalidMessage && m_client)
583         m_client->didReceiveInvalidMessage(this, message.messageID());
584
585     m_didReceiveInvalidMessage = oldDidReceiveInvalidMessage;
586 }
587
588 void Connection::dispatchMessages()
589 {
590     Vector<IncomingMessage> incomingMessages;
591     
592     {
593         MutexLocker locker(m_incomingMessagesLock);
594         m_incomingMessages.swap(incomingMessages);
595     }
596
597     for (size_t i = 0; i < incomingMessages.size(); ++i)
598         dispatchMessage(incomingMessages[i]);
599 }
600
601 } // namespace CoreIPC