#include "config.h"
#include "Connection.h"
#include "BinarySemaphore.h"
#include "CoreIPCMessageKinds.h"
#include <WebCore/RunLoop.h>
#include <wtf/CurrentTime.h>
using namespace std;
using namespace WebCore;
namespace CoreIPC {
class Connection::SyncMessageState : public ThreadSafeRefCounted<Connection::SyncMessageState> {
public:
static PassRefPtr<SyncMessageState> getOrCreate(RunLoop*);
~SyncMessageState();
void wakeUpClientRunLoop()
{
m_waitForSyncReplySemaphore.signal();
}
bool wait(double absoluteTime)
{
return m_waitForSyncReplySemaphore.wait(absoluteTime);
}
#if PLATFORM(WIN)
bool waitWhileDispatchingSentWin32Messages(double absoluteTime, const Vector<HWND>& windowsToReceiveMessages)
{
return Connection::dispatchSentMessagesUntil(windowsToReceiveMessages, m_waitForSyncReplySemaphore, absoluteTime);
}
#endif
bool processIncomingMessage(Connection*, IncomingMessage&);
void dispatchMessages();
private:
explicit SyncMessageState(RunLoop*);
typedef HashMap<RunLoop*, SyncMessageState*> SyncMessageStateMap;
static SyncMessageStateMap& syncMessageStateMap()
{
DEFINE_STATIC_LOCAL(SyncMessageStateMap, syncMessageStateMap, ());
return syncMessageStateMap;
}
static Mutex& syncMessageStateMapMutex()
{
DEFINE_STATIC_LOCAL(Mutex, syncMessageStateMapMutex, ());
return syncMessageStateMapMutex;
}
void dispatchMessageAndResetDidScheduleDispatchMessagesWork();
RunLoop* m_runLoop;
BinarySemaphore m_waitForSyncReplySemaphore;
Mutex m_mutex;
bool m_didScheduleDispatchMessagesWork;
struct ConnectionAndIncomingMessage {
RefPtr<Connection> connection;
IncomingMessage incomingMessage;
};
Vector<ConnectionAndIncomingMessage> m_messagesToDispatchWhileWaitingForSyncReply;
};
PassRefPtr<Connection::SyncMessageState> Connection::SyncMessageState::getOrCreate(RunLoop* runLoop)
{
MutexLocker locker(syncMessageStateMapMutex());
SyncMessageStateMap::AddResult result = syncMessageStateMap().add(runLoop, 0);
if (!result.isNewEntry) {
ASSERT(result.iterator->second);
return result.iterator->second;
}
RefPtr<SyncMessageState> syncMessageState = adoptRef(new SyncMessageState(runLoop));
result.iterator->second = syncMessageState.get();
return syncMessageState.release();
}
Connection::SyncMessageState::SyncMessageState(RunLoop* runLoop)
: m_runLoop(runLoop)
, m_didScheduleDispatchMessagesWork(false)
{
}
Connection::SyncMessageState::~SyncMessageState()
{
MutexLocker locker(syncMessageStateMapMutex());
ASSERT(syncMessageStateMap().contains(m_runLoop));
syncMessageStateMap().remove(m_runLoop);
}
bool Connection::SyncMessageState::processIncomingMessage(Connection* connection, IncomingMessage& incomingMessage)
{
MessageID messageID = incomingMessage.messageID();
if (!messageID.shouldDispatchMessageWhenWaitingForSyncReply())
return false;
ConnectionAndIncomingMessage connectionAndIncomingMessage;
connectionAndIncomingMessage.connection = connection;
connectionAndIncomingMessage.incomingMessage = incomingMessage;
{
MutexLocker locker(m_mutex);
if (!m_didScheduleDispatchMessagesWork) {
m_runLoop->dispatch(bind(&SyncMessageState::dispatchMessageAndResetDidScheduleDispatchMessagesWork, this));
m_didScheduleDispatchMessagesWork = true;
}
m_messagesToDispatchWhileWaitingForSyncReply.append(connectionAndIncomingMessage);
}
wakeUpClientRunLoop();
return true;
}
void Connection::SyncMessageState::dispatchMessages()
{
ASSERT(m_runLoop == RunLoop::current());
Vector<ConnectionAndIncomingMessage> messagesToDispatchWhileWaitingForSyncReply;
{
MutexLocker locker(m_mutex);
m_messagesToDispatchWhileWaitingForSyncReply.swap(messagesToDispatchWhileWaitingForSyncReply);
}
for (size_t i = 0; i < messagesToDispatchWhileWaitingForSyncReply.size(); ++i) {
ConnectionAndIncomingMessage& connectionAndIncomingMessage = messagesToDispatchWhileWaitingForSyncReply[i];
connectionAndIncomingMessage.connection->dispatchMessage(connectionAndIncomingMessage.incomingMessage);
}
}
void Connection::SyncMessageState::dispatchMessageAndResetDidScheduleDispatchMessagesWork()
{
{
MutexLocker locker(m_mutex);
ASSERT(m_didScheduleDispatchMessagesWork);
m_didScheduleDispatchMessagesWork = false;
}
dispatchMessages();
}
PassRefPtr<Connection> Connection::createServerConnection(Identifier identifier, Client* client, RunLoop* clientRunLoop)
{
return adoptRef(new Connection(identifier, true, client, clientRunLoop));
}
PassRefPtr<Connection> Connection::createClientConnection(Identifier identifier, Client* client, RunLoop* clientRunLoop)
{
return adoptRef(new Connection(identifier, false, client, clientRunLoop));
}
Connection::Connection(Identifier identifier, bool isServer, Client* client, RunLoop* clientRunLoop)
: m_client(client)
, m_isServer(isServer)
, m_syncRequestID(0)
, m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage(false)
, m_shouldExitOnSyncMessageSendFailure(false)
, m_didCloseOnConnectionWorkQueueCallback(0)
, m_isConnected(false)
, m_connectionQueue("com.apple.CoreIPC.ReceiveQueue")
, m_clientRunLoop(clientRunLoop)
, m_inDispatchMessageCount(0)
, m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount(0)
, m_didReceiveInvalidMessage(false)
, m_defaultSyncMessageTimeout(NoTimeout)
, m_syncMessageState(SyncMessageState::getOrCreate(clientRunLoop))
, m_shouldWaitForSyncReplies(true)
{
ASSERT(m_client);
platformInitialize(identifier);
}
Connection::~Connection()
{
ASSERT(!isValid());
m_connectionQueue.invalidate();
}
void Connection::setOnlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage(bool flag)
{
ASSERT(!m_isConnected);
m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage = flag;
}
void Connection::setShouldExitOnSyncMessageSendFailure(bool shouldExitOnSyncMessageSendFailure)
{
ASSERT(!m_isConnected);
m_shouldExitOnSyncMessageSendFailure = shouldExitOnSyncMessageSendFailure;
}
void Connection::addQueueClient(QueueClient* queueClient)
{
m_connectionQueue.dispatch(bind(&Connection::addQueueClientOnWorkQueue, this, queueClient));
}
void Connection::removeQueueClient(QueueClient* queueClient)
{
m_connectionQueue.dispatch(bind(&Connection::removeQueueClientOnWorkQueue, this, queueClient));
}
void Connection::addQueueClientOnWorkQueue(QueueClient* queueClient)
{
ASSERT(!m_connectionQueueClients.contains(queueClient));
m_connectionQueueClients.append(queueClient);
}
void Connection::removeQueueClientOnWorkQueue(QueueClient* queueClient)
{
size_t index = m_connectionQueueClients.find(queueClient);
ASSERT(index != notFound);
m_connectionQueueClients.remove(index);
}
void Connection::setDidCloseOnConnectionWorkQueueCallback(DidCloseOnConnectionWorkQueueCallback callback)
{
ASSERT(!m_isConnected);
m_didCloseOnConnectionWorkQueueCallback = callback;
}
void Connection::invalidate()
{
if (!isValid()) {
return;
}
m_client = 0;
m_connectionQueue.dispatch(bind(&Connection::platformInvalidate, this));
}
void Connection::markCurrentlyDispatchedMessageAsInvalid()
{
ASSERT(m_inDispatchMessageCount > 0);
m_didReceiveInvalidMessage = true;
}
void Connection::setDefaultSyncMessageTimeout(double defaultSyncMessageTimeout)
{
ASSERT(defaultSyncMessageTimeout != DefaultTimeout);
m_defaultSyncMessageTimeout = defaultSyncMessageTimeout;
}
PassOwnPtr<ArgumentEncoder> Connection::createSyncMessageArgumentEncoder(uint64_t destinationID, uint64_t& syncRequestID)
{
OwnPtr<ArgumentEncoder> argumentEncoder = ArgumentEncoder::create(destinationID);
syncRequestID = ++m_syncRequestID;
argumentEncoder->encode(syncRequestID);
return argumentEncoder.release();
}
bool Connection::sendMessage(MessageID messageID, PassOwnPtr<ArgumentEncoder> arguments, unsigned messageSendFlags)
{
if (!isValid())
return false;
if (messageSendFlags & DispatchMessageEvenWhenWaitingForSyncReply
&& (!m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage
|| m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount))
messageID = messageID.messageIDWithAddedFlags(MessageID::DispatchMessageWhenWaitingForSyncReply);
MutexLocker locker(m_outgoingMessagesLock);
m_outgoingMessages.append(OutgoingMessage(messageID, arguments));
m_connectionQueue.dispatch(bind(&Connection::sendOutgoingMessages, this));
return true;
}
bool Connection::sendSyncReply(PassOwnPtr<ArgumentEncoder> arguments)
{
return sendMessage(MessageID(CoreIPCMessage::SyncMessageReply), arguments);
}
PassOwnPtr<ArgumentDecoder> Connection::waitForMessage(MessageID messageID, uint64_t destinationID, double timeout)
{
{
MutexLocker locker(m_incomingMessagesLock);
for (Deque<IncomingMessage>::iterator it = m_incomingMessages.begin(), end = m_incomingMessages.end(); it != end; ++it) {
IncomingMessage& message = *it;
if (message.messageID() == messageID && message.arguments()->destinationID() == destinationID) {
OwnPtr<ArgumentDecoder> arguments = message.releaseArguments();
m_incomingMessages.remove(it);
return arguments.release();
}
}
}
double absoluteTime = currentTime() + timeout;
std::pair<unsigned, uint64_t> messageAndDestination(std::make_pair(messageID.toInt(), destinationID));
{
MutexLocker locker(m_waitForMessageMutex);
ASSERT(!m_waitForMessageMap.contains(messageAndDestination));
m_waitForMessageMap.set(messageAndDestination, 0);
}
while (true) {
MutexLocker locker(m_waitForMessageMutex);
HashMap<std::pair<unsigned, uint64_t>, ArgumentDecoder*>::iterator it = m_waitForMessageMap.find(messageAndDestination);
if (it->second) {
OwnPtr<ArgumentDecoder> arguments = adoptPtr(it->second);
m_waitForMessageMap.remove(it);
return arguments.release();
}
if (!m_waitForMessageCondition.timedWait(m_waitForMessageMutex, absoluteTime)) {
m_waitForMessageMap.remove(messageAndDestination);
break;
}
}
return nullptr;
}
PassOwnPtr<ArgumentDecoder> Connection::sendSyncMessage(MessageID messageID, uint64_t syncRequestID, PassOwnPtr<ArgumentEncoder> encoder, double timeout, unsigned syncSendFlags)
{
ASSERT(RunLoop::current() == m_clientRunLoop);
if (!isValid()) {
didFailToSendSyncMessage();
return nullptr;
}
{
MutexLocker locker(m_syncReplyStateMutex);
if (!m_shouldWaitForSyncReplies) {
didFailToSendSyncMessage();
return nullptr;
}
m_pendingSyncReplies.append(PendingSyncReply(syncRequestID));
}
sendMessage(messageID.messageIDWithAddedFlags(MessageID::SyncMessage), encoder, DispatchMessageEvenWhenWaitingForSyncReply);
RefPtr<Connection> protect(this);
OwnPtr<ArgumentDecoder> reply = waitForSyncReply(syncRequestID, timeout, syncSendFlags);
{
MutexLocker locker(m_syncReplyStateMutex);
ASSERT(m_pendingSyncReplies.last().syncRequestID == syncRequestID);
m_pendingSyncReplies.removeLast();
}
if (!reply)
didFailToSendSyncMessage();
return reply.release();
}
PassOwnPtr<ArgumentDecoder> Connection::waitForSyncReply(uint64_t syncRequestID, double timeout, unsigned syncSendFlags)
{
if (timeout == DefaultTimeout)
timeout = m_defaultSyncMessageTimeout;
if (timeout == NoTimeout)
timeout = 1e10;
double absoluteTime = currentTime() + timeout;
bool timedOut = false;
while (!timedOut) {
m_syncMessageState->dispatchMessages();
{
MutexLocker locker(m_syncReplyStateMutex);
ASSERT(!m_pendingSyncReplies.isEmpty());
PendingSyncReply& pendingSyncReply = m_pendingSyncReplies.last();
ASSERT(pendingSyncReply.syncRequestID == syncRequestID);
if (pendingSyncReply.didReceiveReply || !m_shouldWaitForSyncReplies)
return pendingSyncReply.releaseReplyDecoder();
}
if (!isValid())
return nullptr;
#if PLATFORM(WIN)
timedOut = !m_syncMessageState->waitWhileDispatchingSentWin32Messages(absoluteTime, m_client->windowsToReceiveSentMessagesWhileWaitingForSyncReply());
#else
if (syncSendFlags & SpinRunLoopWhileWaitingForReply) {
#if PLATFORM(MAC)
RunLoop::current()->runForDuration(1e10);
timedOut = currentTime() >= absoluteTime;
#endif
} else
timedOut = !m_syncMessageState->wait(absoluteTime);
#endif
}
if (m_client)
m_client->syncMessageSendTimedOut(this);
return nullptr;
}
void Connection::processIncomingSyncReply(PassOwnPtr<ArgumentDecoder> arguments)
{
MutexLocker locker(m_syncReplyStateMutex);
for (size_t i = m_pendingSyncReplies.size(); i > 0; --i) {
PendingSyncReply& pendingSyncReply = m_pendingSyncReplies[i - 1];
if (pendingSyncReply.syncRequestID != arguments->destinationID())
continue;
ASSERT(!pendingSyncReply.replyDecoder);
pendingSyncReply.replyDecoder = arguments.leakPtr();
pendingSyncReply.didReceiveReply = true;
if (i == m_pendingSyncReplies.size())
m_syncMessageState->wakeUpClientRunLoop();
return;
}
}
void Connection::processIncomingMessage(MessageID messageID, PassOwnPtr<ArgumentDecoder> arguments)
{
if (messageID == MessageID(CoreIPCMessage::SyncMessageReply)) {
processIncomingSyncReply(arguments);
return;
}
IncomingMessage incomingMessage(messageID, arguments);
if (m_syncMessageState->processIncomingMessage(this, incomingMessage))
return;
{
MutexLocker locker(m_waitForMessageMutex);
HashMap<std::pair<unsigned, uint64_t>, ArgumentDecoder*>::iterator it = m_waitForMessageMap.find(std::make_pair(messageID.toInt(), incomingMessage.destinationID()));
if (it != m_waitForMessageMap.end()) {
it->second = incomingMessage.releaseArguments().leakPtr();
ASSERT(it->second);
m_waitForMessageCondition.signal();
return;
}
}
for (size_t i = 0; i < m_connectionQueueClients.size(); ++i) {
bool didHandleMessage = false;
m_connectionQueueClients[i]->didReceiveMessageOnConnectionWorkQueue(this, incomingMessage.messageID(), incomingMessage.arguments(), didHandleMessage);
if (didHandleMessage) {
incomingMessage.releaseArguments();
return;
}
}
enqueueIncomingMessage(incomingMessage);
}
void Connection::postConnectionDidCloseOnConnectionWorkQueue()
{
m_connectionQueue.dispatch(bind(&Connection::connectionDidClose, this));
}
void Connection::connectionDidClose()
{
platformInvalidate();
{
MutexLocker locker(m_syncReplyStateMutex);
ASSERT(m_shouldWaitForSyncReplies);
m_shouldWaitForSyncReplies = false;
if (!m_pendingSyncReplies.isEmpty())
m_syncMessageState->wakeUpClientRunLoop();
}
if (m_didCloseOnConnectionWorkQueueCallback)
m_didCloseOnConnectionWorkQueueCallback(m_connectionQueue, this);
m_clientRunLoop->dispatch(bind(&Connection::dispatchConnectionDidClose, this));
}
void Connection::dispatchConnectionDidClose()
{
if (!m_client)
return;
Client* client = m_client;
m_client = 0;
client->didClose(this);
}
bool Connection::canSendOutgoingMessages() const
{
return m_isConnected && platformCanSendOutgoingMessages();
}
void Connection::sendOutgoingMessages()
{
if (!canSendOutgoingMessages())
return;
while (true) {
OutgoingMessage message;
{
MutexLocker locker(m_outgoingMessagesLock);
if (m_outgoingMessages.isEmpty())
break;
message = m_outgoingMessages.takeFirst();
}
if (!sendOutgoingMessage(message.messageID(), adoptPtr(message.arguments())))
break;
}
}
void Connection::dispatchSyncMessage(MessageID messageID, ArgumentDecoder* arguments)
{
ASSERT(messageID.isSync());
uint64_t syncRequestID = 0;
if (!arguments->decodeUInt64(syncRequestID) || !syncRequestID) {
arguments->markInvalid();
return;
}
OwnPtr<ArgumentEncoder> replyEncoder = ArgumentEncoder::create(syncRequestID);
m_client->didReceiveSyncMessage(this, messageID, arguments, replyEncoder);
ASSERT(!arguments->isInvalid());
if (replyEncoder)
sendSyncReply(replyEncoder.release());
}
void Connection::didFailToSendSyncMessage()
{
if (!m_shouldExitOnSyncMessageSendFailure)
return;
exit(0);
}
void Connection::enqueueIncomingMessage(IncomingMessage& incomingMessage)
{
MutexLocker locker(m_incomingMessagesLock);
m_incomingMessages.append(incomingMessage);
m_clientRunLoop->dispatch(bind(&Connection::dispatchOneMessage, this));
}
void Connection::dispatchMessage(IncomingMessage& message)
{
OwnPtr<ArgumentDecoder> arguments = message.releaseArguments();
if (!m_client)
return;
m_inDispatchMessageCount++;
if (message.messageID().shouldDispatchMessageWhenWaitingForSyncReply())
m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount++;
bool oldDidReceiveInvalidMessage = m_didReceiveInvalidMessage;
m_didReceiveInvalidMessage = false;
if (message.messageID().isSync())
dispatchSyncMessage(message.messageID(), arguments.get());
else
m_client->didReceiveMessage(this, message.messageID(), arguments.get());
m_didReceiveInvalidMessage |= arguments->isInvalid();
m_inDispatchMessageCount--;
if (message.messageID().shouldDispatchMessageWhenWaitingForSyncReply())
m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount--;
if (m_didReceiveInvalidMessage && m_client)
m_client->didReceiveInvalidMessage(this, message.messageID());
m_didReceiveInvalidMessage = oldDidReceiveInvalidMessage;
}
void Connection::dispatchOneMessage()
{
IncomingMessage incomingMessage;
{
MutexLocker locker(m_incomingMessagesLock);
if (m_incomingMessages.isEmpty())
return;
incomingMessage = m_incomingMessages.takeFirst();
}
dispatchMessage(incomingMessage);
}
void Connection::wakeUpRunLoop()
{
m_clientRunLoop->wakeUp();
}
}