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