#include "config.h"
#include "Connection.h"
#include "Logging.h"
#include "MessageFlags.h"
#include <memory>
#include <wtf/HashSet.h>
#include <wtf/Lock.h>
#include <wtf/NeverDestroyed.h>
#include <wtf/RunLoop.h>
#include <wtf/text/WTFString.h>
#include <wtf/threads/BinarySemaphore.h>
#if PLATFORM(COCOA)
#include "MachMessage.h"
#endif
#if USE(UNIX_DOMAIN_SOCKETS)
#include "UnixMessage.h"
#endif
namespace IPC {
#if PLATFORM(COCOA)
const size_t maxPendingIncomingMessagesKillingThreshold { 50000 };
#endif
std::atomic<unsigned> UnboundedSynchronousIPCScope::unboundedSynchronousIPCCount = 0;
struct Connection::WaitForMessageState {
WaitForMessageState(MessageName messageName, uint64_t destinationID, OptionSet<WaitForOption> waitForOptions)
: messageName(messageName)
, destinationID(destinationID)
, waitForOptions(waitForOptions)
{
}
MessageName messageName;
uint64_t destinationID;
OptionSet<WaitForOption> waitForOptions;
bool messageWaitingInterrupted = false;
std::unique_ptr<Decoder> decoder;
};
class Connection::SyncMessageState {
public:
static SyncMessageState& singleton();
~SyncMessageState() = delete;
void wakeUpClientRunLoop()
{
m_waitForSyncReplySemaphore.signal();
}
bool wait(TimeWithDynamicClockType absoluteTime)
{
return m_waitForSyncReplySemaphore.waitUntil(absoluteTime);
}
bool processIncomingMessage(Connection&, std::unique_ptr<Decoder>&);
void dispatchMessages();
private:
friend class LazyNeverDestroyed<Connection::SyncMessageState>;
SyncMessageState() = default;
void dispatchMessagesAndResetDidScheduleDispatchMessagesForConnection(Connection&);
BinarySemaphore m_waitForSyncReplySemaphore;
Lock m_mutex;
HashSet<RefPtr<Connection>> m_didScheduleDispatchMessagesWorkSet;
struct ConnectionAndIncomingMessage {
Ref<Connection> connection;
std::unique_ptr<Decoder> message;
void dispatch()
{
connection->dispatchMessage(WTFMove(message));
}
};
Vector<ConnectionAndIncomingMessage> m_messagesToDispatchWhileWaitingForSyncReply;
};
Connection::SyncMessageState& Connection::SyncMessageState::singleton()
{
static std::once_flag onceFlag;
static LazyNeverDestroyed<SyncMessageState> syncMessageState;
std::call_once(onceFlag, [] {
syncMessageState.construct();
});
return syncMessageState;
}
bool Connection::SyncMessageState::processIncomingMessage(Connection& connection, std::unique_ptr<Decoder>& message)
{
switch (message->shouldDispatchMessageWhenWaitingForSyncReply()) {
case ShouldDispatchWhenWaitingForSyncReply::No:
return false;
case ShouldDispatchWhenWaitingForSyncReply::YesDuringUnboundedIPC:
if (!UnboundedSynchronousIPCScope::hasOngoingUnboundedSyncIPC())
return false;
break;
case ShouldDispatchWhenWaitingForSyncReply::Yes:
break;
}
bool shouldDispatch;
{
auto locker = holdLock(m_mutex);
shouldDispatch = m_didScheduleDispatchMessagesWorkSet.add(&connection).isNewEntry;
auto incomingMessagesLocker = holdLock(connection.m_incomingMessagesMutex);
if (message->shouldMaintainOrderingWithAsyncMessages()) {
while (!connection.m_incomingMessages.isEmpty())
m_messagesToDispatchWhileWaitingForSyncReply.append(ConnectionAndIncomingMessage { connection, connection.m_incomingMessages.takeFirst() });
}
m_messagesToDispatchWhileWaitingForSyncReply.append(ConnectionAndIncomingMessage { connection, WTFMove(message) });
}
if (shouldDispatch) {
RunLoop::main().dispatch([this, protectedConnection = makeRef(connection)]() mutable {
dispatchMessagesAndResetDidScheduleDispatchMessagesForConnection(protectedConnection);
});
}
wakeUpClientRunLoop();
return true;
}
void Connection::SyncMessageState::dispatchMessages()
{
ASSERT(RunLoop::isMain());
Vector<ConnectionAndIncomingMessage> messagesToDispatchWhileWaitingForSyncReply;
{
auto locker = holdLock(m_mutex);
m_messagesToDispatchWhileWaitingForSyncReply.swap(messagesToDispatchWhileWaitingForSyncReply);
}
for (auto& connectionAndIncomingMessage : messagesToDispatchWhileWaitingForSyncReply)
connectionAndIncomingMessage.dispatch();
}
void Connection::SyncMessageState::dispatchMessagesAndResetDidScheduleDispatchMessagesForConnection(Connection& connection)
{
ASSERT(RunLoop::isMain());
Vector<ConnectionAndIncomingMessage> messagesToDispatchWhileWaitingForSyncReply;
{
auto locker = holdLock(m_mutex);
ASSERT(m_didScheduleDispatchMessagesWorkSet.contains(&connection));
m_didScheduleDispatchMessagesWorkSet.remove(&connection);
m_messagesToDispatchWhileWaitingForSyncReply.swap(messagesToDispatchWhileWaitingForSyncReply);
}
Vector<ConnectionAndIncomingMessage> messagesToPutBack;
for (auto& connectionAndIncomingMessage : messagesToDispatchWhileWaitingForSyncReply) {
if (&connection == connectionAndIncomingMessage.connection.ptr())
connectionAndIncomingMessage.dispatch();
else
messagesToPutBack.append(WTFMove(connectionAndIncomingMessage));
}
if (!messagesToPutBack.isEmpty()) {
auto locker = holdLock(m_mutex);
messagesToPutBack.appendVector(WTFMove(m_messagesToDispatchWhileWaitingForSyncReply));
m_messagesToDispatchWhileWaitingForSyncReply = WTFMove(messagesToPutBack);
}
}
struct Connection::PendingSyncReply {
uint64_t syncRequestID { 0 };
std::unique_ptr<Decoder> replyDecoder;
bool didReceiveReply { false };
PendingSyncReply() = default;
explicit PendingSyncReply(uint64_t syncRequestID)
: syncRequestID(syncRequestID)
{
}
};
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));
}
static HashMap<IPC::Connection::UniqueID, Connection*>& allConnections()
{
static NeverDestroyed<HashMap<IPC::Connection::UniqueID, Connection*>> map;
return map;
}
static Lock& asyncReplyHandlerMapLock()
{
static Lock lock;
return lock;
}
static HashMap<uintptr_t, HashMap<uint64_t, CompletionHandler<void(Decoder*)>>>& asyncReplyHandlerMap(const LockHolder&)
{
ASSERT(asyncReplyHandlerMapLock().isHeld());
static NeverDestroyed<HashMap<uintptr_t, HashMap<uint64_t, CompletionHandler<void(Decoder*)>>>> map;
return map.get();
}
static void clearAsyncReplyHandlers(const Connection&);
Connection::Connection(Identifier identifier, bool isServer, Client& client)
: m_client(client)
, m_uniqueID(UniqueID::generate())
, 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_shouldWaitForSyncReplies(true)
, m_shouldWaitForMessages(true)
{
ASSERT(RunLoop::isMain());
allConnections().add(m_uniqueID, this);
platformInitialize(identifier);
#if HAVE(QOS_CLASSES)
ASSERT(pthread_main_np());
m_mainThread = pthread_self();
#endif
}
Connection::~Connection()
{
ASSERT(RunLoop::isMain());
ASSERT(!isValid());
allConnections().remove(m_uniqueID);
clearAsyncReplyHandlers(*this);
}
Connection* Connection::connection(UniqueID uniqueID)
{
ASSERT(RunLoop::isMain());
return allConnections().get(uniqueID);
}
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(ReceiverName messageReceiverName, WorkQueue& workQueue, WorkQueueMessageReceiver* workQueueMessageReceiver, uint64_t destinationID)
{
ASSERT(RunLoop::isMain());
auto locker = holdLock(m_workQueueMessageReceiversMutex);
auto key = std::make_pair(static_cast<uint8_t>(messageReceiverName), destinationID);
ASSERT(!m_workQueueMessageReceivers.contains(key));
m_workQueueMessageReceivers.add(key, std::make_pair(&workQueue, workQueueMessageReceiver));
}
void Connection::removeWorkQueueMessageReceiver(ReceiverName messageReceiverName, uint64_t destinationID)
{
ASSERT(RunLoop::isMain());
auto locker = holdLock(m_workQueueMessageReceiversMutex);
auto key = std::make_pair(static_cast<uint8_t>(messageReceiverName), destinationID);
ASSERT(m_workQueueMessageReceivers.contains(key));
m_workQueueMessageReceivers.remove(key);
}
void Connection::dispatchWorkQueueMessageReceiverMessage(WorkQueueMessageReceiver& workQueueMessageReceiver, Decoder& decoder)
{
if (!decoder.isSyncMessage()) {
workQueueMessageReceiver.didReceiveMessage(*this, decoder);
return;
}
uint64_t syncRequestID = 0;
if (!decoder.decode(syncRequestID) || !syncRequestID) {
decoder.markInvalid();
return;
}
auto replyEncoder = makeUnique<Encoder>(MessageName::SyncMessageReply, syncRequestID);
workQueueMessageReceiver.didReceiveSyncMessage(*this, decoder, replyEncoder);
ASSERT(decoder.isValid());
if (replyEncoder)
sendSyncReply(WTFMove(replyEncoder));
}
void Connection::addThreadMessageReceiver(ReceiverName messageReceiverName, ThreadMessageReceiver* threadMessageReceiver, uint64_t destinationID)
{
ASSERT(RunLoop::isMain());
auto locker = holdLock(m_threadMessageReceiversLock);
auto key = std::make_pair(static_cast<uint8_t>(messageReceiverName), destinationID);
ASSERT(!m_threadMessageReceivers.contains(key));
m_threadMessageReceivers.add(key, threadMessageReceiver);
}
void Connection::removeThreadMessageReceiver(ReceiverName messageReceiverName, uint64_t destinationID)
{
ASSERT(RunLoop::isMain());
auto locker = holdLock(m_threadMessageReceiversLock);
auto key = std::make_pair(static_cast<uint8_t>(messageReceiverName), destinationID);
ASSERT(m_threadMessageReceivers.contains(key));
m_threadMessageReceivers.remove(key);
}
void Connection::dispatchThreadMessageReceiverMessage(ThreadMessageReceiver& threadMessageReceiver, Decoder& decoder)
{
if (!decoder.isSyncMessage()) {
threadMessageReceiver.didReceiveMessage(*this, decoder);
return;
}
uint64_t syncRequestID = 0;
if (!decoder.decode(syncRequestID) || !syncRequestID) {
decoder.markInvalid();
return;
}
auto replyEncoder = makeUnique<Encoder>(MessageName::SyncMessageReply, syncRequestID);
threadMessageReceiver.didReceiveSyncMessage(*this, decoder, replyEncoder);
ASSERT(decoder.isValid());
if (replyEncoder)
sendSyncReply(WTFMove(replyEncoder));
}
void Connection::setDidCloseOnConnectionWorkQueueCallback(DidCloseOnConnectionWorkQueueCallback callback)
{
ASSERT(!m_isConnected);
m_didCloseOnConnectionWorkQueueCallback = callback;
}
void Connection::invalidate()
{
ASSERT(RunLoop::isMain());
if (!isValid()) {
return;
}
m_isValid = false;
m_connectionQueue->dispatch([protectedThis = makeRef(*this)]() mutable {
protectedThis->platformInvalidate();
});
}
void Connection::markCurrentlyDispatchedMessageAsInvalid()
{
ASSERT(m_inDispatchMessageCount > 0);
m_didReceiveInvalidMessage = true;
}
std::unique_ptr<Encoder> Connection::createSyncMessageEncoder(MessageName messageName, uint64_t destinationID, uint64_t& syncRequestID)
{
auto encoder = makeUnique<Encoder>(messageName, destinationID);
syncRequestID = ++m_syncRequestID;
*encoder << syncRequestID;
return encoder;
}
bool Connection::sendMessage(std::unique_ptr<Encoder> encoder, OptionSet<SendOption> sendOptions)
{
if (!isValid())
return false;
#if ENABLE(IPC_TESTING_API)
if (isMainThread()) {
bool hasDeadObservers = false;
for (auto& observerWeakPtr : m_messageObservers) {
if (auto* observer = observerWeakPtr.get())
observer->willSendMessage(*encoder, sendOptions);
else
hasDeadObservers = true;
}
if (hasDeadObservers)
m_messageObservers.removeAllMatching([](auto& observer) { return !observer; });
}
#endif
if (isMainThread() && m_inDispatchMessageMarkedToUseFullySynchronousModeForTesting && !encoder->isSyncMessage() && !(encoder->messageReceiverName() == ReceiverName::IPC) && !sendOptions.contains(SendOption::IgnoreFullySynchronousMode)) {
uint64_t syncRequestID;
auto wrappedMessage = createSyncMessageEncoder(MessageName::WrappedAsyncMessageForTesting, encoder->destinationID(), syncRequestID);
wrappedMessage->setFullySynchronousModeForTesting();
wrappedMessage->wrapForTesting(WTFMove(encoder));
return static_cast<bool>(sendSyncMessage(syncRequestID, WTFMove(wrappedMessage), Seconds::infinity(), { }));
}
if (sendOptions.contains(SendOption::DispatchMessageEvenWhenWaitingForSyncReply)
&& (!m_onlySendMessagesAsDispatchWhenWaitingForSyncReplyWhenProcessingSuchAMessage
|| m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount))
encoder->setShouldDispatchMessageWhenWaitingForSyncReply(ShouldDispatchWhenWaitingForSyncReply::Yes);
else if (sendOptions.contains(SendOption::DispatchMessageEvenWhenWaitingForUnboundedSyncReply))
encoder->setShouldDispatchMessageWhenWaitingForSyncReply(ShouldDispatchWhenWaitingForSyncReply::YesDuringUnboundedIPC);
#if ENABLE(IPC_TESTING_API)
#endif
{
auto locker = holdLock(m_outgoingMessagesMutex);
m_outgoingMessages.append(WTFMove(encoder));
}
m_connectionQueue->dispatch([protectedThis = makeRef(*this)]() mutable {
protectedThis->sendOutgoingMessages();
});
return true;
}
bool Connection::sendSyncReply(std::unique_ptr<Encoder> encoder)
{
return sendMessage(WTFMove(encoder), { });
}
Seconds Connection::timeoutRespectingIgnoreTimeoutsForTesting(Seconds timeout) const
{
return m_ignoreTimeoutsForTesting ? Seconds::infinity() : timeout;
}
std::unique_ptr<Decoder> Connection::waitForMessage(MessageName messageName, uint64_t destinationID, Seconds timeout, OptionSet<WaitForOption> waitForOptions)
{
ASSERT(RunLoop::isMain());
auto protectedThis = makeRef(*this);
timeout = timeoutRespectingIgnoreTimeoutsForTesting(timeout);
WaitForMessageState waitingForMessage(messageName, destinationID, waitForOptions);
{
auto locker = holdLock(m_waitForMessageMutex);
ASSERT(!m_waitingForMessage);
if (m_waitingForMessage)
return nullptr;
if (!m_shouldWaitForMessages)
return nullptr;
bool hasIncomingSynchronousMessage = false;
{
auto locker = holdLock(m_incomingMessagesMutex);
for (auto it = m_incomingMessages.begin(), end = m_incomingMessages.end(); it != end; ++it) {
std::unique_ptr<Decoder>& message = *it;
if (message->messageName() == messageName && message->destinationID() == destinationID) {
std::unique_ptr<Decoder> returnedMessage = WTFMove(message);
m_incomingMessages.remove(it);
return returnedMessage;
}
if (message->isSyncMessage())
hasIncomingSynchronousMessage = true;
}
}
if (hasIncomingSynchronousMessage && waitForOptions.contains(WaitForOption::InterruptWaitingIfSyncMessageArrives)) {
#if ASSERT_ENABLED
ASSERT(!m_waitingForMessage);
#endif
return nullptr;
}
m_waitingForMessage = &waitingForMessage;
}
MonotonicTime absoluteTimeout = MonotonicTime::now() + timeout;
while (true) {
SyncMessageState::singleton().dispatchMessages();
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<Decoder> Connection::sendSyncMessage(uint64_t syncRequestID, std::unique_ptr<Encoder> encoder, Seconds timeout, OptionSet<SendSyncOption> sendSyncOptions)
{
ASSERT(RunLoop::isMain());
if (!isValid()) {
didFailToSendSyncMessage();
return nullptr;
}
{
LockHolder locker(m_syncReplyStateMutex);
if (!m_shouldWaitForSyncReplies) {
didFailToSendSyncMessage();
return nullptr;
}
m_pendingSyncReplies.append(PendingSyncReply(syncRequestID));
}
++m_inSendSyncCount;
OptionSet<SendOption> sendOptions = IPC::SendOption::DispatchMessageEvenWhenWaitingForSyncReply;
if (sendSyncOptions.contains(SendSyncOption::ForceDispatchWhenDestinationIsWaitingForUnboundedSyncReply))
sendOptions = sendOptions | IPC::SendOption::DispatchMessageEvenWhenWaitingForUnboundedSyncReply;
if (sendSyncOptions.contains(IPC::SendSyncOption::MaintainOrderingWithAsyncMessages))
encoder->setShouldMaintainOrderingWithAsyncMessages();
auto messageName = encoder->messageName();
sendMessage(WTFMove(encoder), sendOptions);
Ref<Connection> protect(*this);
std::unique_ptr<Decoder> reply = waitForSyncReply(syncRequestID, messageName, timeout, sendSyncOptions);
--m_inSendSyncCount;
{
LockHolder locker(m_syncReplyStateMutex);
ASSERT(m_pendingSyncReplies.last().syncRequestID == syncRequestID);
m_pendingSyncReplies.removeLast();
}
if (!reply)
didFailToSendSyncMessage();
return reply;
}
std::unique_ptr<Decoder> Connection::waitForSyncReply(uint64_t syncRequestID, MessageName messageName, Seconds timeout, OptionSet<SendSyncOption> sendSyncOptions)
{
timeout = timeoutRespectingIgnoreTimeoutsForTesting(timeout);
MonotonicTime absoluteTime = MonotonicTime::now() + timeout;
willSendSyncMessage(sendSyncOptions);
bool timedOut = false;
while (!timedOut) {
SyncMessageState::singleton().dispatchMessages();
{
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(sendSyncOptions);
return WTFMove(pendingSyncReply.replyDecoder);
}
}
if (!isValid()) {
RELEASE_LOG_ERROR(IPC, "Connection::waitForSyncReply: Connection no longer valid, id = %" PRIu64, syncRequestID);
didReceiveSyncReply(sendSyncOptions);
return nullptr;
}
timedOut = !SyncMessageState::singleton().wait(absoluteTime);
}
#if OS(DARWIN)
RELEASE_LOG_ERROR(IPC, "Connection::waitForSyncReply: Timed-out while waiting for reply for %{public}s from process %d, id = %" PRIu64, description(messageName), remoteProcessID(), syncRequestID);
#else
RELEASE_LOG_ERROR(IPC, "Connection::waitForSyncReply: Timed-out while waiting for reply for %s, id = %" PRIu64, description(messageName), syncRequestID);
#endif
didReceiveSyncReply(sendSyncOptions);
return nullptr;
}
void Connection::processIncomingSyncReply(std::unique_ptr<Decoder> 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;
}
}
}
void Connection::processIncomingMessage(std::unique_ptr<Decoder> message)
{
ASSERT(message->messageReceiverName() != ReceiverName::Invalid);
if (message->messageName() == MessageName::SyncMessageReply) {
processIncomingSyncReply(WTFMove(message));
return;
}
auto threadedReceiverKey = std::make_pair(static_cast<uint8_t>(message->messageReceiverName()), message->destinationID());
if (!WorkQueueMessageReceiverMap::isValidKey(threadedReceiverKey) || !ThreadMessageReceiverMap::isValidKey(threadedReceiverKey)) {
RunLoop::main().dispatch([protectedThis = makeRef(*this), messageName = message->messageName()]() mutable {
protectedThis->dispatchDidReceiveInvalidMessage(messageName);
});
return;
}
if (dispatchMessageToWorkQueueReceiver(message))
return;
if (dispatchMessageToThreadReceiver(message))
return;
if (message->isSyncMessage()) {
auto locker = holdLock(m_incomingSyncMessageCallbackMutex);
for (auto& callback : m_incomingSyncMessageCallbacks.values())
m_incomingSyncMessageCallbackQueue->dispatch(WTFMove(callback));
m_incomingSyncMessageCallbacks.clear();
}
{
auto locker = holdLock(m_waitForMessageMutex);
if (m_waitingForMessage && !m_waitingForMessage->decoder) {
if (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->waitForOptions.contains(WaitForOption::DispatchIncomingSyncMessagesWhileWaiting) && message->isSyncMessage() && SyncMessageState::singleton().processIncomingMessage(*this, message)) {
m_waitForMessageCondition.notifyOne();
return;
}
if (m_waitingForMessage->waitForOptions.contains(WaitForOption::InterruptWaitingIfSyncMessageArrives) && message->isSyncMessage()) {
m_waitingForMessage->messageWaitingInterrupted = true;
m_waitForMessageCondition.notifyOne();
enqueueIncomingMessage(WTFMove(message));
return;
}
}
if (SyncMessageState::singleton().processIncomingMessage(*this, message))
return;
enqueueIncomingMessage(WTFMove(message));
}
}
uint64_t Connection::installIncomingSyncMessageCallback(WTF::Function<void ()>&& callback)
{
auto locker = holdLock(m_incomingSyncMessageCallbackMutex);
m_nextIncomingSyncMessageCallbackID++;
if (!m_incomingSyncMessageCallbackQueue)
m_incomingSyncMessageCallbackQueue = WorkQueue::create("com.apple.WebKit.IPC.IncomingSyncMessageCallbackQueue");
m_incomingSyncMessageCallbacks.add(m_nextIncomingSyncMessageCallbackID, WTFMove(callback));
return m_nextIncomingSyncMessageCallbackID;
}
void Connection::uninstallIncomingSyncMessageCallback(uint64_t callbackID)
{
auto locker = holdLock(m_incomingSyncMessageCallbackMutex);
m_incomingSyncMessageCallbacks.remove(callbackID);
}
bool Connection::hasIncomingSyncMessage()
{
auto locker = holdLock(m_incomingMessagesMutex);
for (auto& message : m_incomingMessages) {
if (message->isSyncMessage())
return true;
}
return false;
}
void Connection::enableIncomingMessagesThrottling()
{
if (isIncomingMessagesThrottlingEnabled())
return;
m_incomingMessagesThrottler = makeUnique<MessagesThrottler>(*this, &Connection::dispatchIncomingMessages);
}
#if ENABLE(IPC_TESTING_API)
void Connection::addMessageObserver(const MessageObserver& observer)
{
m_messageObservers.append(makeWeakPtr(observer));
}
#endif
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();
}
{
auto locker = holdLock(m_waitForMessageMutex);
ASSERT(m_shouldWaitForMessages);
m_shouldWaitForMessages = false;
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->isValid())
return;
protectedThis->m_isValid = false;
protectedThis->m_client.didClose(protectedThis.get());
clearAsyncReplyHandlers(protectedThis.get());
});
}
bool Connection::canSendOutgoingMessages() const
{
return m_isConnected && platformCanSendOutgoingMessages();
}
void Connection::sendOutgoingMessages()
{
if (!canSendOutgoingMessages())
return;
while (true) {
std::unique_ptr<Encoder> message;
{
auto locker = holdLock(m_outgoingMessagesMutex);
if (m_outgoingMessages.isEmpty())
break;
message = m_outgoingMessages.takeFirst();
}
if (!sendOutgoingMessage(WTFMove(message)))
break;
}
}
void Connection::dispatchSyncMessage(Decoder& decoder)
{
ASSERT(decoder.isSyncMessage());
uint64_t syncRequestID = 0;
if (!decoder.decode(syncRequestID) || !syncRequestID) {
decoder.markInvalid();
return;
}
auto replyEncoder = makeUnique<Encoder>(MessageName::SyncMessageReply, syncRequestID);
if (decoder.messageName() == MessageName::WrappedAsyncMessageForTesting) {
if (!m_fullySynchronousModeIsAllowedForTesting) {
decoder.markInvalid();
return;
}
std::unique_ptr<Decoder> unwrappedDecoder = Decoder::unwrapForTesting(decoder);
RELEASE_ASSERT(unwrappedDecoder);
processIncomingMessage(WTFMove(unwrappedDecoder));
SyncMessageState::singleton().dispatchMessages();
} else {
m_client.didReceiveSyncMessage(*this, decoder, replyEncoder);
}
#if ENABLE(IPC_TESTING_API)
ASSERT(decoder.isValid() || m_ignoreInvalidMessageForTesting);
#else
ASSERT(decoder.isValid());
#endif
if (replyEncoder)
sendSyncReply(WTFMove(replyEncoder));
}
void Connection::dispatchDidReceiveInvalidMessage(MessageName messageName)
{
ASSERT(RunLoop::isMain());
if (!isValid())
return;
m_client.didReceiveInvalidMessage(*this, messageName);
}
void Connection::didFailToSendSyncMessage()
{
if (!m_shouldExitOnSyncMessageSendFailure)
return;
exit(0);
}
void Connection::enqueueIncomingMessage(std::unique_ptr<Decoder> incomingMessage)
{
{
auto locker = holdLock(m_incomingMessagesMutex);
#if PLATFORM(COCOA)
if (m_wasKilled)
return;
if (isIncomingMessagesThrottlingEnabled() && m_incomingMessages.size() >= maxPendingIncomingMessagesKillingThreshold) {
if (kill()) {
RELEASE_LOG_FAULT(IPC, "%p - Connection::enqueueIncomingMessage: Over %zu incoming messages have been queued without the main thread processing them, killing the connection as the remote process seems to be misbehaving", this, maxPendingIncomingMessagesKillingThreshold);
m_incomingMessages.clear();
}
return;
}
#endif
m_incomingMessages.append(WTFMove(incomingMessage));
if (isIncomingMessagesThrottlingEnabled() && m_incomingMessages.size() != 1)
return;
}
RunLoop::main().dispatch([protectedThis = makeRef(*this)]() mutable {
if (protectedThis->isIncomingMessagesThrottlingEnabled())
protectedThis->dispatchIncomingMessages();
else
protectedThis->dispatchOneIncomingMessage();
});
}
void Connection::dispatchMessage(Decoder& decoder)
{
RELEASE_ASSERT(isValid());
if (decoder.messageReceiverName() == ReceiverName::AsyncReply) {
Optional<uint64_t> listenerID;
decoder >> listenerID;
if (!listenerID) {
ASSERT_NOT_REACHED();
return;
}
auto handler = takeAsyncReplyHandler(*this, *listenerID);
if (!handler) {
ASSERT_NOT_REACHED();
return;
}
handler(&decoder);
return;
}
#if ENABLE(IPC_TESTING_API)
if (isMainThread()) {
bool hasDeadObservers = false;
for (auto& observerWeakPtr : m_messageObservers) {
if (auto* observer = observerWeakPtr.get())
observer->didReceiveMessage(decoder);
else
hasDeadObservers = true;
}
if (hasDeadObservers)
m_messageObservers.removeAllMatching([](auto& observer) { return !observer; });
}
#endif
m_client.didReceiveMessage(*this, decoder);
}
auto Connection::threadMessageReceiver(std::unique_ptr<Decoder>& message) -> RefPtr<ThreadMessageReceiver>
{
auto locker = holdLock(m_threadMessageReceiversLock);
auto key = std::make_pair(static_cast<uint8_t>(message->messageReceiverName()), 0);
auto it = m_threadMessageReceivers.find(key);
if (it != m_threadMessageReceivers.end())
return it->value;
if (auto destinationID = message->destinationID()) {
key.second = destinationID;
return m_threadMessageReceivers.get(key);
}
return nullptr;
}
auto Connection::workQueueMessageReceiver(std::unique_ptr<Decoder>& message) -> std::pair<RefPtr<WorkQueue>, RefPtr<WorkQueueMessageReceiver>>
{
auto locker = holdLock(m_workQueueMessageReceiversMutex);
auto key = std::make_pair(static_cast<uint8_t>(message->messageReceiverName()), 0);
auto it = m_workQueueMessageReceivers.find(key);
if (it != m_workQueueMessageReceivers.end())
return it->value;
if (auto destinationID = message->destinationID()) {
key.second = destinationID;
return m_workQueueMessageReceivers.get(key);
}
return { };
}
bool Connection::dispatchMessageToWorkQueueReceiver(std::unique_ptr<Decoder>& message)
{
auto receiver = workQueueMessageReceiver(message);
if (!receiver.first)
return false;
receiver.first->dispatch([protectedThis = makeRef(*this), workQueueMessageReceiver = receiver.second, decoder = WTFMove(message)]() mutable {
protectedThis->dispatchWorkQueueMessageReceiverMessage(*workQueueMessageReceiver, *decoder);
});
return true;
}
bool Connection::dispatchMessageToThreadReceiver(std::unique_ptr<Decoder>& message)
{
if (auto receiver = threadMessageReceiver(message)) {
receiver->dispatchToThread([protectedThis = makeRef(*this), receiver, decoder = WTFMove(message)]() mutable {
protectedThis->dispatchThreadMessageReceiverMessage(*receiver, *decoder);
});
return true;
}
return false;
}
void Connection::dispatchMessage(std::unique_ptr<Decoder> message)
{
ASSERT(RunLoop::isMain());
if (!isValid())
return;
if (dispatchMessageToWorkQueueReceiver(message))
return;
if (message->shouldUseFullySynchronousModeForTesting()) {
if (!m_fullySynchronousModeIsAllowedForTesting) {
m_client.didReceiveInvalidMessage(*this, message->messageName());
return;
}
m_inDispatchMessageMarkedToUseFullySynchronousModeForTesting++;
}
m_inDispatchMessageCount++;
bool isDispatchingMessageWhileWaitingForSyncReply = (message->shouldDispatchMessageWhenWaitingForSyncReply() == ShouldDispatchWhenWaitingForSyncReply::Yes)
|| (message->shouldDispatchMessageWhenWaitingForSyncReply() == ShouldDispatchWhenWaitingForSyncReply::YesDuringUnboundedIPC && UnboundedSynchronousIPCScope::hasOngoingUnboundedSyncIPC());
if (isDispatchingMessageWhileWaitingForSyncReply)
m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount++;
bool oldDidReceiveInvalidMessage = m_didReceiveInvalidMessage;
m_didReceiveInvalidMessage = false;
if (message->isSyncMessage())
dispatchSyncMessage(*message);
else
dispatchMessage(*message);
m_didReceiveInvalidMessage |= !message->isValid();
m_inDispatchMessageCount--;
if (isDispatchingMessageWhileWaitingForSyncReply)
m_inDispatchMessageMarkedDispatchWhenWaitingForSyncReplyCount--;
if (message->shouldUseFullySynchronousModeForTesting())
m_inDispatchMessageMarkedToUseFullySynchronousModeForTesting--;
if (m_didReceiveInvalidMessage && isValid())
m_client.didReceiveInvalidMessage(*this, message->messageName());
m_didReceiveInvalidMessage = oldDidReceiveInvalidMessage;
}
Connection::MessagesThrottler::MessagesThrottler(Connection& connection, DispatchMessagesFunction dispatchMessages)
: m_dispatchMessagesTimer(RunLoop::main(), &connection, dispatchMessages)
, m_connection(connection)
, m_dispatchMessages(dispatchMessages)
{
ASSERT(RunLoop::isMain());
}
void Connection::MessagesThrottler::scheduleMessagesDispatch()
{
ASSERT(RunLoop::isMain());
if (m_throttlingLevel) {
m_dispatchMessagesTimer.startOneShot(0_s);
return;
}
RunLoop::main().dispatch([this, protectedConnection = makeRefPtr(&m_connection)]() mutable {
(protectedConnection.get()->*m_dispatchMessages)();
});
}
size_t Connection::MessagesThrottler::numberOfMessagesToProcess(size_t totalMessages)
{
ASSERT(RunLoop::isMain());
static const size_t maxIncomingMessagesDispatchingBatchSize { 600 };
static const unsigned maxThrottlingLevel = 9;
size_t batchSize = maxIncomingMessagesDispatchingBatchSize / (m_throttlingLevel + 1);
if (totalMessages > maxIncomingMessagesDispatchingBatchSize)
m_throttlingLevel = std::min(m_throttlingLevel + 1, maxThrottlingLevel);
else if (m_throttlingLevel)
--m_throttlingLevel;
return std::min(totalMessages, batchSize);
}
void Connection::dispatchOneIncomingMessage()
{
std::unique_ptr<Decoder> message;
{
auto locker = holdLock(m_incomingMessagesMutex);
if (m_incomingMessages.isEmpty())
return;
message = m_incomingMessages.takeFirst();
}
dispatchMessage(WTFMove(message));
}
void Connection::dispatchIncomingMessages()
{
ASSERT(RunLoop::isMain());
std::unique_ptr<Decoder> message;
size_t messagesToProcess = 0;
{
auto locker = holdLock(m_incomingMessagesMutex);
if (m_incomingMessages.isEmpty())
return;
message = m_incomingMessages.takeFirst();
messagesToProcess = m_incomingMessagesThrottler->numberOfMessagesToProcess(m_incomingMessages.size());
if (messagesToProcess < m_incomingMessages.size()) {
RELEASE_LOG_ERROR(IPC, "%p - Connection::dispatchIncomingMessages: IPC throttling was triggered (has %zu pending incoming messages, will only process %zu before yielding)", this, m_incomingMessages.size(), messagesToProcess);
#if PLATFORM(COCOA)
RELEASE_LOG_ERROR(IPC, "%p - Connection::dispatchIncomingMessages: first IPC message in queue is %{public}s", this, description(message->messageName()));
#endif
}
if (!m_incomingMessages.isEmpty())
m_incomingMessagesThrottler->scheduleMessagesDispatch();
}
dispatchMessage(WTFMove(message));
for (size_t i = 1; i < messagesToProcess; ++i) {
{
auto locker = holdLock(m_incomingMessagesMutex);
if (m_incomingMessages.isEmpty())
return;
message = m_incomingMessages.takeFirst();
}
dispatchMessage(WTFMove(message));
}
}
uint64_t nextAsyncReplyHandlerID()
{
static std::atomic<uint64_t> identifier { 0 };
return ++identifier;
}
void addAsyncReplyHandler(Connection& connection, uint64_t identifier, CompletionHandler<void(Decoder*)>&& completionHandler)
{
LockHolder locker(asyncReplyHandlerMapLock());
auto result = asyncReplyHandlerMap(locker).ensure(reinterpret_cast<uintptr_t>(&connection), [] {
return HashMap<uint64_t, CompletionHandler<void(Decoder*)>>();
}).iterator->value.add(identifier, WTFMove(completionHandler));
ASSERT_UNUSED(result, result.isNewEntry);
}
void clearAsyncReplyHandlers(const Connection& connection)
{
HashMap<uint64_t, CompletionHandler<void(Decoder*)>> map;
{
LockHolder locker(asyncReplyHandlerMapLock());
map = asyncReplyHandlerMap(locker).take(reinterpret_cast<uintptr_t>(&connection));
}
for (auto& handler : map.values()) {
if (handler)
handler(nullptr);
}
}
CompletionHandler<void(Decoder*)> takeAsyncReplyHandler(Connection& connection, uint64_t identifier)
{
LockHolder locker(asyncReplyHandlerMapLock());
auto& map = asyncReplyHandlerMap(locker);
auto iterator = map.find(reinterpret_cast<uintptr_t>(&connection));
if (iterator != map.end()) {
if (!iterator->value.isValidKey(identifier)) {
ASSERT_NOT_REACHED();
connection.markCurrentlyDispatchedMessageAsInvalid();
return nullptr;
}
ASSERT(iterator->value.contains(identifier));
return iterator->value.take(identifier);
}
ASSERT_NOT_REACHED();
return nullptr;
}
void Connection::wakeUpRunLoop()
{
RunLoop::main().wakeUp();
}
}