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