#include "config.h"
#include "Connection.h"
#include <memory>
#include <wtf/CurrentTime.h>
#include <wtf/HashSet.h>
#include <wtf/NeverDestroyed.h>
#include <wtf/RunLoop.h>
#include <wtf/text/WTFString.h>
#include <wtf/threads/BinarySemaphore.h>
namespace IPC {
struct WaitForMessageState {
WaitForMessageState(StringReference messageReceiverName, StringReference messageName, uint64_t destinationID, unsigned waitForMessageFlags)
: messageReceiverName(messageReceiverName)
, messageName(messageName)
, destinationID(destinationID)
, waitForMessageFlags(waitForMessageFlags)
{
}
StringReference messageReceiverName;
StringReference messageName;
uint64_t destinationID;
unsigned waitForMessageFlags;
bool messageWaitingInterrupted = false;
std::unique_ptr<MessageDecoder> decoder;
};
class Connection::SyncMessageState {
public:
static SyncMessageState& singleton();
SyncMessageState();
~SyncMessageState() = delete;
void wakeUpClientRunLoop()
{
m_waitForSyncReplySemaphore.signal();
}
bool wait(double absoluteTime)
{
return m_waitForSyncReplySemaphore.wait(absoluteTime);
}
bool processIncomingMessage(Connection&, std::unique_ptr<MessageDecoder>&);
void dispatchMessages(Connection* allowedConnection);
private:
void dispatchMessageAndResetDidScheduleDispatchMessagesForConnection(Connection&);
BinarySemaphore m_waitForSyncReplySemaphore;
Lock m_mutex;
HashSet<RefPtr<Connection>> m_didScheduleDispatchMessagesWorkSet;
struct ConnectionAndIncomingMessage {
Ref<Connection> connection;
std::unique_ptr<MessageDecoder> message;
};
Vector<ConnectionAndIncomingMessage> m_messagesToDispatchWhileWaitingForSyncReply;
};
class Connection::SecondaryThreadPendingSyncReply {
public:
std::unique_ptr<MessageDecoder> replyDecoder;
BinarySemaphore semaphore;
};
Connection::SyncMessageState& Connection::SyncMessageState::singleton()
{
static std::once_flag onceFlag;
static LazyNeverDestroyed<SyncMessageState> syncMessageState;
std::call_once(onceFlag, [] {
syncMessageState.construct();
});
return syncMessageState;
}
Connection::SyncMessageState::SyncMessageState()
{
}
bool Connection::SyncMessageState::processIncomingMessage(Connection& connection, std::unique_ptr<MessageDecoder>& message)
{
if (!message->shouldDispatchMessageWhenWaitingForSyncReply())
return false;
ConnectionAndIncomingMessage connectionAndIncomingMessage { connection, WTFMove(message) };
{
std::lock_guard<Lock> lock(m_mutex);
if (m_didScheduleDispatchMessagesWorkSet.add(&connection).isNewEntry) {
RunLoop::main().dispatch([this, protectedConnection = Ref<Connection>(connection)]() mutable {
dispatchMessageAndResetDidScheduleDispatchMessagesForConnection(protectedConnection);
});
}
m_messagesToDispatchWhileWaitingForSyncReply.append(WTFMove(connectionAndIncomingMessage));
}
wakeUpClientRunLoop();
return true;
}
void Connection::SyncMessageState::dispatchMessages(Connection* allowedConnection)
{
ASSERT(RunLoop::isMain());
Vector<ConnectionAndIncomingMessage> messagesToDispatchWhileWaitingForSyncReply;
{
std::lock_guard<Lock> lock(m_mutex);
m_messagesToDispatchWhileWaitingForSyncReply.swap(messagesToDispatchWhileWaitingForSyncReply);
}
Vector<ConnectionAndIncomingMessage> messagesToPutBack;
for (size_t i = 0; i < messagesToDispatchWhileWaitingForSyncReply.size(); ++i) {
ConnectionAndIncomingMessage& connectionAndIncomingMessage = messagesToDispatchWhileWaitingForSyncReply[i];
if (allowedConnection && allowedConnection != connectionAndIncomingMessage.connection.ptr()) {
messagesToPutBack.append(WTFMove(connectionAndIncomingMessage));
continue;
}
connectionAndIncomingMessage.connection->dispatchMessage(WTFMove(connectionAndIncomingMessage.message));
}
if (!messagesToPutBack.isEmpty()) {
std::lock_guard<Lock> lock(m_mutex);
for (auto& message : messagesToPutBack)
m_messagesToDispatchWhileWaitingForSyncReply.append(WTFMove(message));
}
}
void Connection::SyncMessageState::dispatchMessageAndResetDidScheduleDispatchMessagesForConnection(Connection& connection)
{
{
std::lock_guard<Lock> lock(m_mutex);
ASSERT(m_didScheduleDispatchMessagesWorkSet.contains(&connection));
m_didScheduleDispatchMessagesWorkSet.remove(&connection);
}
dispatchMessages(&connection);
}
Ref<Connection> Connection::createServerConnection(Identifier identifier, Client& client)
{
return adoptRef(*new Connection(identifier, true, client));
}
Ref<Connection> Connection::createClientConnection(Identifier identifier, Client& client)
{
return adoptRef(*new Connection(identifier, false, client));
}
Connection::Connection(Identifier identifier, bool isServer, Client& client)
: m_client(&client)
, m_isServer(isServer)
, m_syncRequestID(0)
, m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage(false)
, m_shouldExitOnSyncMessageSendFailure(false)
, m_didCloseOnConnectionWorkQueueCallback(0)
, m_isConnected(false)
, m_connectionQueue(WorkQueue::create("com.apple.IPC.ReceiveQueue"))
, m_inSendSyncCount(0)
, m_inDispatchMessageCount(0)
, m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount(0)
, m_didReceiveInvalidMessage(false)
, m_waitingForMessage(nullptr)
, m_shouldWaitForSyncReplies(true)
{
ASSERT(RunLoop::isMain());
platformInitialize(identifier);
#if HAVE(QOS_CLASSES)
ASSERT(pthread_main_np());
m_mainThread = pthread_self();
#endif
}
Connection::~Connection()
{
ASSERT(!isValid());
}
void Connection::setOnlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage(bool flag)
{
ASSERT(!m_isConnected);
m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage = flag;
}
void Connection::setShouldExitOnSyncMessageSendFailure(bool shouldExitOnSyncMessageSendFailure)
{
ASSERT(!m_isConnected);
m_shouldExitOnSyncMessageSendFailure = shouldExitOnSyncMessageSendFailure;
}
void Connection::addWorkQueueMessageReceiver(StringReference messageReceiverName, WorkQueue* workQueue, WorkQueueMessageReceiver* workQueueMessageReceiver)
{
ASSERT(RunLoop::isMain());
m_connectionQueue->dispatch([protectedThis = makeRef(*this), messageReceiverName = WTFMove(messageReceiverName), workQueue, workQueueMessageReceiver]() mutable {
ASSERT(!protectedThis->m_workQueueMessageReceivers.contains(messageReceiverName));
protectedThis->m_workQueueMessageReceivers.add(messageReceiverName, std::make_pair(workQueue, workQueueMessageReceiver));
});
}
void Connection::removeWorkQueueMessageReceiver(StringReference messageReceiverName)
{
ASSERT(RunLoop::isMain());
m_connectionQueue->dispatch([protectedThis = makeRef(*this), messageReceiverName = WTFMove(messageReceiverName)]() mutable {
ASSERT(protectedThis->m_workQueueMessageReceivers.contains(messageReceiverName));
protectedThis->m_workQueueMessageReceivers.remove(messageReceiverName);
});
}
void Connection::dispatchWorkQueueMessageReceiverMessage(WorkQueueMessageReceiver& workQueueMessageReceiver, MessageDecoder& decoder)
{
if (!decoder.isSyncMessage()) {
workQueueMessageReceiver.didReceiveMessage(*this, decoder);
return;
}
uint64_t syncRequestID = 0;
if (!decoder.decode(syncRequestID) || !syncRequestID) {
decoder.markInvalid();
return;
}
#if HAVE(DTRACE)
auto replyEncoder = std::make_unique<MessageEncoder>("IPC", "SyncMessageReply", syncRequestID, decoder.UUID());
#else
auto replyEncoder = std::make_unique<MessageEncoder>("IPC", "SyncMessageReply", syncRequestID);
#endif
workQueueMessageReceiver.didReceiveSyncMessage(*this, decoder, replyEncoder);
ASSERT(!decoder.isInvalid());
if (replyEncoder)
sendSyncReply(WTFMove(replyEncoder));
}
void Connection::setDidCloseOnConnectionWorkQueueCallback(DidCloseOnConnectionWorkQueueCallback callback)
{
ASSERT(!m_isConnected);
m_didCloseOnConnectionWorkQueueCallback = callback;
}
void Connection::invalidate()
{
if (!isValid()) {
return;
}
m_client = nullptr;
m_connectionQueue->dispatch([protectedThis = makeRef(*this)]() mutable {
protectedThis->platformInvalidate();
});
}
void Connection::markCurrentlyDispatchedMessageAsInvalid()
{
ASSERT(m_inDispatchMessageCount > 0);
m_didReceiveInvalidMessage = true;
}
std::unique_ptr<MessageEncoder> Connection::createSyncMessageEncoder(StringReference messageReceiverName, StringReference messageName, uint64_t destinationID, uint64_t& syncRequestID)
{
auto encoder = std::make_unique<MessageEncoder>(messageReceiverName, messageName, destinationID);
encoder->setIsSyncMessage(true);
syncRequestID = ++m_syncRequestID;
*encoder << syncRequestID;
return encoder;
}
bool Connection::sendMessage(std::unique_ptr<MessageEncoder> encoder, unsigned messageSendFlags, bool alreadyRecordedMessage)
{
if (!isValid())
return false;
if (m_inDispatchMessageMarkedToUseFullySynchronousModeForTesting && !encoder->isSyncMessage() && !(encoder->messageReceiverName() == "IPC")) {
uint64_t syncRequestID;
auto wrappedMessage = createSyncMessageEncoder("IPC", "WrappedAsyncMessageForTesting", encoder->destinationID(), syncRequestID);
wrappedMessage->setFullySynchronousModeForTesting();
wrappedMessage->wrapForTesting(WTFMove(encoder));
return static_cast<bool>(sendSyncMessage(syncRequestID, WTFMove(wrappedMessage), std::chrono::milliseconds::max()));
}
if (messageSendFlags & DispatchMessageEvenWhenWaitingForSyncReply
&& (!m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage
|| m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount))
encoder->setShouldDispatchMessageWhenWaitingForSyncReply(true);
#if HAVE(DTRACE)
std::unique_ptr<MessageRecorder::MessageProcessingToken> token;
if (!alreadyRecordedMessage)
token = MessageRecorder::recordOutgoingMessage(*this, *encoder);
#else
UNUSED_PARAM(alreadyRecordedMessage);
#endif
{
std::lock_guard<Lock> lock(m_outgoingMessagesMutex);
m_outgoingMessages.append(WTFMove(encoder));
}
m_connectionQueue->dispatch([protectedThis = makeRef(*this)]() mutable {
protectedThis->sendOutgoingMessages();
});
return true;
}
bool Connection::sendSyncReply(std::unique_ptr<MessageEncoder> encoder)
{
return sendMessage(WTFMove(encoder));
}
std::chrono::milliseconds Connection::timeoutRespectingIgnoreTimeoutsForTesting(std::chrono::milliseconds timeout) const
{
return m_ignoreTimeoutsForTesting ? std::chrono::milliseconds::max() : timeout;
}
std::unique_ptr<MessageDecoder> Connection::waitForMessage(StringReference messageReceiverName, StringReference messageName, uint64_t destinationID, std::chrono::milliseconds timeout, unsigned waitForMessageFlags)
{
ASSERT(RunLoop::isMain());
timeout = timeoutRespectingIgnoreTimeoutsForTesting(timeout);
bool hasIncomingSynchronousMessage = false;
{
std::lock_guard<Lock> lock(m_incomingMessagesMutex);
for (auto it = m_incomingMessages.begin(), end = m_incomingMessages.end(); it != end; ++it) {
std::unique_ptr<MessageDecoder>& message = *it;
if (message->messageReceiverName() == messageReceiverName && message->messageName() == messageName && message->destinationID() == destinationID) {
std::unique_ptr<MessageDecoder> returnedMessage = WTFMove(message);
m_incomingMessages.remove(it);
return returnedMessage;
}
if (message->isSyncMessage())
hasIncomingSynchronousMessage = true;
}
}
if (hasIncomingSynchronousMessage && waitForMessageFlags & InterruptWaitingIfSyncMessageArrives) {
m_waitingForMessage = nullptr;
return nullptr;
}
WaitForMessageState waitingForMessage(messageReceiverName, messageName, destinationID, waitForMessageFlags);
{
std::lock_guard<Lock> lock(m_waitForMessageMutex);
ASSERT(!m_waitingForMessage);
m_waitingForMessage = &waitingForMessage;
}
auto now = Condition::Clock::now();
auto remainingClockTime = std::chrono::duration_cast<std::chrono::milliseconds>(Condition::Clock::time_point::max() - now);
auto absoluteTimeout = now + std::min(remainingClockTime, timeout);
while (true) {
std::unique_lock<Lock> lock(m_waitForMessageMutex);
if (m_waitingForMessage->decoder) {
auto decoder = WTFMove(m_waitingForMessage->decoder);
m_waitingForMessage = nullptr;
return decoder;
}
bool didTimeout = !m_waitForMessageCondition.waitUntil(lock, absoluteTimeout);
if (didTimeout || m_waitingForMessage->messageWaitingInterrupted) {
m_waitingForMessage = nullptr;
break;
}
}
return nullptr;
}
std::unique_ptr<MessageDecoder> Connection::sendSyncMessage(uint64_t syncRequestID, std::unique_ptr<MessageEncoder> encoder, std::chrono::milliseconds timeout, unsigned syncSendFlags)
{
if (!RunLoop::isMain()) {
ASSERT(!syncSendFlags);
return sendSyncMessageFromSecondaryThread(syncRequestID, WTFMove(encoder), timeout);
}
if (!isValid()) {
didFailToSendSyncMessage();
return nullptr;
}
{
LockHolder locker(m_syncReplyStateMutex);
if (!m_shouldWaitForSyncReplies) {
didFailToSendSyncMessage();
return nullptr;
}
m_pendingSyncReplies.append(PendingSyncReply(syncRequestID));
}
++m_inSendSyncCount;
#if HAVE(DTRACE)
auto token = MessageRecorder::recordOutgoingMessage(*this, *encoder);
#endif
sendMessage(WTFMove(encoder), DispatchMessageEvenWhenWaitingForSyncReply, true);
Ref<Connection> protect(*this);
std::unique_ptr<MessageDecoder> reply = waitForSyncReply(syncRequestID, timeout, syncSendFlags);
--m_inSendSyncCount;
{
LockHolder locker(m_syncReplyStateMutex);
ASSERT(m_pendingSyncReplies.last().syncRequestID == syncRequestID);
m_pendingSyncReplies.removeLast();
}
if (!reply)
didFailToSendSyncMessage();
return reply;
}
std::unique_ptr<MessageDecoder> Connection::sendSyncMessageFromSecondaryThread(uint64_t syncRequestID, std::unique_ptr<MessageEncoder> encoder, std::chrono::milliseconds timeout)
{
ASSERT(!RunLoop::isMain());
if (!isValid())
return nullptr;
SecondaryThreadPendingSyncReply pendingReply;
{
LockHolder locker(m_syncReplyStateMutex);
if (!m_shouldWaitForSyncReplies)
return nullptr;
ASSERT(!m_secondaryThreadPendingSyncReplyMap.contains(syncRequestID));
m_secondaryThreadPendingSyncReplyMap.add(syncRequestID, &pendingReply);
}
#if HAVE(DTRACE)
auto token = MessageRecorder::recordOutgoingMessage(*this, *encoder);
#endif
sendMessage(WTFMove(encoder), 0, true);
timeout = timeoutRespectingIgnoreTimeoutsForTesting(timeout);
pendingReply.semaphore.wait(currentTime() + (timeout.count() / 1000.0));
{
LockHolder locker(m_syncReplyStateMutex);
ASSERT(m_secondaryThreadPendingSyncReplyMap.contains(syncRequestID));
m_secondaryThreadPendingSyncReplyMap.remove(syncRequestID);
}
return WTFMove(pendingReply.replyDecoder);
}
std::unique_ptr<MessageDecoder> Connection::waitForSyncReply(uint64_t syncRequestID, std::chrono::milliseconds timeout, unsigned syncSendFlags)
{
timeout = timeoutRespectingIgnoreTimeoutsForTesting(timeout);
double absoluteTime = currentTime() + (timeout.count() / 1000.0);
willSendSyncMessage(syncSendFlags);
bool timedOut = false;
while (!timedOut) {
SyncMessageState::singleton().dispatchMessages(nullptr);
{
LockHolder locker(m_syncReplyStateMutex);
ASSERT(!m_pendingSyncReplies.isEmpty());
PendingSyncReply& pendingSyncReply = m_pendingSyncReplies.last();
ASSERT_UNUSED(syncRequestID, pendingSyncReply.syncRequestID == syncRequestID);
if (pendingSyncReply.didReceiveReply || !m_shouldWaitForSyncReplies) {
didReceiveSyncReply(syncSendFlags);
return WTFMove(pendingSyncReply.replyDecoder);
}
}
if (!isValid()) {
RELEASE_LOG_ERROR("Connection::waitForSyncReply: Connection no longer valid, id = %" PRIu64, syncRequestID);
didReceiveSyncReply(syncSendFlags);
return nullptr;
}
timedOut = !SyncMessageState::singleton().wait(absoluteTime);
}
RELEASE_LOG_ERROR("Connection::waitForSyncReply: Timed-out while waiting for reply, id = %" PRIu64, syncRequestID);
didReceiveSyncReply(syncSendFlags);
return nullptr;
}
void Connection::processIncomingSyncReply(std::unique_ptr<MessageDecoder> decoder)
{
LockHolder locker(m_syncReplyStateMutex);
for (size_t i = m_pendingSyncReplies.size(); i > 0; --i) {
PendingSyncReply& pendingSyncReply = m_pendingSyncReplies[i - 1];
if (pendingSyncReply.syncRequestID != decoder->destinationID())
continue;
ASSERT(!pendingSyncReply.replyDecoder);
pendingSyncReply.replyDecoder = WTFMove(decoder);
pendingSyncReply.didReceiveReply = true;
if (i == m_pendingSyncReplies.size())
SyncMessageState::singleton().wakeUpClientRunLoop();
return;
}
SecondaryThreadPendingSyncReplyMap::iterator secondaryThreadReplyMapItem = m_secondaryThreadPendingSyncReplyMap.find(decoder->destinationID());
if (secondaryThreadReplyMapItem != m_secondaryThreadPendingSyncReplyMap.end()) {
SecondaryThreadPendingSyncReply* reply = secondaryThreadReplyMapItem->value;
ASSERT(!reply->replyDecoder);
reply->replyDecoder = WTFMove(decoder);
reply->semaphore.signal();
}
}
void Connection::processIncomingMessage(std::unique_ptr<MessageDecoder> message)
{
ASSERT(!message->messageReceiverName().isEmpty());
ASSERT(!message->messageName().isEmpty());
if (message->messageReceiverName() == "IPC" && message->messageName() == "SyncMessageReply") {
processIncomingSyncReply(WTFMove(message));
return;
}
if (!m_workQueueMessageReceivers.isValidKey(message->messageReceiverName())) {
RefPtr<Connection> protectedThis(this);
StringReference messageReceiverNameReference = message->messageReceiverName();
String messageReceiverName(messageReceiverNameReference.isEmpty() ? "<unknown message receiver>" : String(messageReceiverNameReference.data(), messageReceiverNameReference.size()));
StringReference messageNameReference = message->messageName();
String messageName(messageNameReference.isEmpty() ? "<unknown message>" : String(messageNameReference.data(), messageNameReference.size()));
RunLoop::main().dispatch([protectedThis = makeRef(*this), messageReceiverName = WTFMove(messageReceiverName), messageName = WTFMove(messageName)]() mutable {
protectedThis->dispatchDidReceiveInvalidMessage(messageReceiverName.utf8(), messageName.utf8());
});
return;
}
auto it = m_workQueueMessageReceivers.find(message->messageReceiverName());
if (it != m_workQueueMessageReceivers.end()) {
it->value.first->dispatch([protectedThis = makeRef(*this), workQueueMessageReceiver = it->value.second, decoder = WTFMove(message)]() mutable {
protectedThis->dispatchWorkQueueMessageReceiverMessage(*workQueueMessageReceiver, *decoder);
});
return;
}
#if HAVE(QOS_CLASSES)
if (message->isSyncMessage() && m_shouldBoostMainThreadOnSyncMessage) {
pthread_override_t override = pthread_override_qos_class_start_np(m_mainThread, QOS_CLASS_USER_INTERACTIVE, 0);
message->setQOSClassOverride(override);
}
#endif
if (message->isSyncMessage()) {
std::lock_guard<Lock> lock(m_incomingSyncMessageCallbackMutex);
for (auto& callback : m_incomingSyncMessageCallbacks.values())
m_incomingSyncMessageCallbackQueue->dispatch(WTFMove(callback));
m_incomingSyncMessageCallbacks.clear();
}
if (SyncMessageState::singleton().processIncomingMessage(*this, message))
return;
{
std::lock_guard<Lock> lock(m_waitForMessageMutex);
if (m_waitingForMessage && !m_waitingForMessage->decoder) {
if (m_waitingForMessage->messageReceiverName == message->messageReceiverName() && m_waitingForMessage->messageName == message->messageName() && m_waitingForMessage->destinationID == message->destinationID()) {
m_waitingForMessage->decoder = WTFMove(message);
ASSERT(m_waitingForMessage->decoder);
m_waitForMessageCondition.notifyOne();
return;
}
if ((m_waitingForMessage->waitForMessageFlags & InterruptWaitingIfSyncMessageArrives) && message->isSyncMessage()) {
m_waitingForMessage->messageWaitingInterrupted = true;
m_waitForMessageCondition.notifyOne();
}
}
}
enqueueIncomingMessage(WTFMove(message));
}
uint64_t Connection::installIncomingSyncMessageCallback(std::function<void ()> callback)
{
std::lock_guard<Lock> lock(m_incomingSyncMessageCallbackMutex);
m_nextIncomingSyncMessageCallbackID++;
if (!m_incomingSyncMessageCallbackQueue)
m_incomingSyncMessageCallbackQueue = WorkQueue::create("com.apple.WebKit.IPC.IncomingSyncMessageCallbackQueue");
m_incomingSyncMessageCallbacks.add(m_nextIncomingSyncMessageCallbackID, callback);
return m_nextIncomingSyncMessageCallbackID;
}
void Connection::uninstallIncomingSyncMessageCallback(uint64_t callbackID)
{
std::lock_guard<Lock> lock(m_incomingSyncMessageCallbackMutex);
m_incomingSyncMessageCallbacks.remove(callbackID);
}
bool Connection::hasIncomingSyncMessage()
{
std::lock_guard<Lock> lock(m_incomingMessagesMutex);
for (auto& message : m_incomingMessages) {
if (message->isSyncMessage())
return true;
}
return false;
}
void Connection::postConnectionDidCloseOnConnectionWorkQueue()
{
m_connectionQueue->dispatch([protectedThis = makeRef(*this)]() mutable {
protectedThis->connectionDidClose();
});
}
void Connection::connectionDidClose()
{
platformInvalidate();
{
LockHolder locker(m_syncReplyStateMutex);
ASSERT(m_shouldWaitForSyncReplies);
m_shouldWaitForSyncReplies = false;
if (!m_pendingSyncReplies.isEmpty())
SyncMessageState::singleton().wakeUpClientRunLoop();
for (SecondaryThreadPendingSyncReplyMap::iterator iter = m_secondaryThreadPendingSyncReplyMap.begin(); iter != m_secondaryThreadPendingSyncReplyMap.end(); ++iter)
iter->value->semaphore.signal();
}
{
std::lock_guard<Lock> lock(m_waitForMessageMutex);
if (m_waitingForMessage)
m_waitingForMessage->messageWaitingInterrupted = true;
}
m_waitForMessageCondition.notifyAll();
if (m_didCloseOnConnectionWorkQueueCallback)
m_didCloseOnConnectionWorkQueueCallback(this);
RunLoop::main().dispatch([protectedThis = makeRef(*this)]() mutable {
if (!protectedThis->m_client)
return;
Client* client = protectedThis->m_client;
protectedThis->m_client = nullptr;
client->didClose(protectedThis.get());
});
}
bool Connection::canSendOutgoingMessages() const
{
return m_isConnected && platformCanSendOutgoingMessages();
}
void Connection::sendOutgoingMessages()
{
if (!canSendOutgoingMessages())
return;
while (true) {
std::unique_ptr<MessageEncoder> message;
{
std::lock_guard<Lock> lock(m_outgoingMessagesMutex);
if (m_outgoingMessages.isEmpty())
break;
message = m_outgoingMessages.takeFirst();
}
if (!sendOutgoingMessage(WTFMove(message)))
break;
}
}
void Connection::dispatchSyncMessage(MessageDecoder& decoder)
{
ASSERT(decoder.isSyncMessage());
uint64_t syncRequestID = 0;
if (!decoder.decode(syncRequestID) || !syncRequestID) {
decoder.markInvalid();
return;
}
#if HAVE(DTRACE)
auto replyEncoder = std::make_unique<MessageEncoder>("IPC", "SyncMessageReply", syncRequestID, decoder.UUID());
#else
auto replyEncoder = std::make_unique<MessageEncoder>("IPC", "SyncMessageReply", syncRequestID);
#endif
if (decoder.messageReceiverName() == "IPC" && decoder.messageName() == "WrappedAsyncMessageForTesting") {
if (!m_fullySynchronousModeIsAllowedForTesting) {
decoder.markInvalid();
return;
}
std::unique_ptr<MessageDecoder> unwrappedDecoder = MessageDecoder::unwrapForTesting(decoder);
RELEASE_ASSERT(unwrappedDecoder);
processIncomingMessage(WTFMove(unwrappedDecoder));
SyncMessageState::singleton().dispatchMessages(nullptr);
} else {
m_client->didReceiveSyncMessage(*this, decoder, replyEncoder);
}
ASSERT(!decoder.isInvalid());
if (replyEncoder)
sendSyncReply(WTFMove(replyEncoder));
}
void Connection::dispatchDidReceiveInvalidMessage(const CString& messageReceiverNameString, const CString& messageNameString)
{
ASSERT(RunLoop::isMain());
if (!m_client)
return;
m_client->didReceiveInvalidMessage(*this, StringReference(messageReceiverNameString.data(), messageReceiverNameString.length()), StringReference(messageNameString.data(), messageNameString.length()));
}
void Connection::didFailToSendSyncMessage()
{
if (!m_shouldExitOnSyncMessageSendFailure)
return;
exit(0);
}
void Connection::enqueueIncomingMessage(std::unique_ptr<MessageDecoder> incomingMessage)
{
{
std::lock_guard<Lock> lock(m_incomingMessagesMutex);
m_incomingMessages.append(WTFMove(incomingMessage));
}
RunLoop::main().dispatch([protectedThis = makeRef(*this)]() mutable {
protectedThis->dispatchOneMessage();
});
}
void Connection::dispatchMessage(MessageDecoder& decoder)
{
m_client->didReceiveMessage(*this, decoder);
}
void Connection::dispatchMessage(std::unique_ptr<MessageDecoder> message)
{
#if HAVE(DTRACE)
MessageRecorder::recordIncomingMessage(*this, *message);
#endif
if (!m_client)
return;
if (message->shouldUseFullySynchronousModeForTesting()) {
if (!m_fullySynchronousModeIsAllowedForTesting) {
m_client->didReceiveInvalidMessage(*this, message->messageReceiverName(), message->messageName());
return;
}
m_inDispatchMessageMarkedToUseFullySynchronousModeForTesting++;
}
m_inDispatchMessageCount++;
if (message->shouldDispatchMessageWhenWaitingForSyncReply())
m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount++;
bool oldDidReceiveInvalidMessage = m_didReceiveInvalidMessage;
m_didReceiveInvalidMessage = false;
if (message->isSyncMessage())
dispatchSyncMessage(*message);
else
dispatchMessage(*message);
m_didReceiveInvalidMessage |= message->isInvalid();
m_inDispatchMessageCount--;
if (message->shouldDispatchMessageWhenWaitingForSyncReply())
m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount--;
if (message->shouldUseFullySynchronousModeForTesting())
m_inDispatchMessageMarkedToUseFullySynchronousModeForTesting--;
if (m_didReceiveInvalidMessage && m_client)
m_client->didReceiveInvalidMessage(*this, message->messageReceiverName(), message->messageName());
m_didReceiveInvalidMessage = oldDidReceiveInvalidMessage;
}
void Connection::dispatchOneMessage()
{
std::unique_ptr<MessageDecoder> message;
{
std::lock_guard<Lock> lock(m_incomingMessagesMutex);
if (m_incomingMessages.isEmpty())
return;
message = m_incomingMessages.takeFirst();
}
dispatchMessage(WTFMove(message));
}
void Connection::wakeUpRunLoop()
{
RunLoop::main().wakeUp();
}
}