#include "config.h"
#include "MessagePort.h"
#include "Document.h"
#include "EventNames.h"
#include "Logging.h"
#include "MessageEvent.h"
#include "MessagePortChannelProvider.h"
#include "MessageWithMessagePorts.h"
#include "WorkerGlobalScope.h"
#include "WorkerThread.h"
#include <wtf/CompletionHandler.h>
#include <wtf/IsoMallocInlines.h>
#include <wtf/Scope.h>
namespace WebCore {
WTF_MAKE_ISO_ALLOCATED_IMPL(MessagePort);
static Lock allMessagePortsLock;
static HashMap<MessagePortIdentifier, MessagePort*>& allMessagePorts()
{
static NeverDestroyed<HashMap<MessagePortIdentifier, MessagePort*>> map;
return map;
}
void MessagePort::ref() const
{
++m_refCount;
}
void MessagePort::deref() const
{
if (!--m_refCount) {
Locker<Lock> locker(allMessagePortsLock);
if (m_refCount)
return;
auto iterator = allMessagePorts().find(m_identifier);
if (iterator != allMessagePorts().end() && iterator->value == this)
allMessagePorts().remove(iterator);
delete this;
}
}
bool MessagePort::isExistingMessagePortLocallyReachable(const MessagePortIdentifier& identifier)
{
Locker<Lock> locker(allMessagePortsLock);
auto* port = allMessagePorts().get(identifier);
return port && port->isLocallyReachable();
}
void MessagePort::notifyMessageAvailable(const MessagePortIdentifier& identifier)
{
Locker<Lock> locker(allMessagePortsLock);
if (auto* port = allMessagePorts().get(identifier))
port->messageAvailable();
}
Ref<MessagePort> MessagePort::create(ScriptExecutionContext& scriptExecutionContext, const MessagePortIdentifier& local, const MessagePortIdentifier& remote)
{
return adoptRef(*new MessagePort(scriptExecutionContext, local, remote));
}
MessagePort::MessagePort(ScriptExecutionContext& scriptExecutionContext, const MessagePortIdentifier& local, const MessagePortIdentifier& remote)
: ActiveDOMObject(&scriptExecutionContext)
, m_identifier(local)
, m_remoteIdentifier(remote)
{
LOG(MessagePorts, "Created MessagePort %s (%p) in process %" PRIu64, m_identifier.logString().utf8().data(), this, Process::identifier().toUInt64());
Locker<Lock> locker(allMessagePortsLock);
allMessagePorts().set(m_identifier, this);
m_scriptExecutionContext->createdMessagePort(*this);
suspendIfNeeded();
}
MessagePort::~MessagePort()
{
LOG(MessagePorts, "Destroyed MessagePort %s (%p) in process %" PRIu64, m_identifier.logString().utf8().data(), this, Process::identifier().toUInt64());
ASSERT(allMessagePortsLock.isLocked());
if (m_entangled)
close();
if (m_scriptExecutionContext)
m_scriptExecutionContext->destroyedMessagePort(*this);
}
void MessagePort::entangle()
{
MessagePortChannelProvider::fromContext(*m_scriptExecutionContext).entangleLocalPortInThisProcessToRemote(m_identifier, m_remoteIdentifier);
}
ExceptionOr<void> MessagePort::postMessage(JSC::JSGlobalObject& state, JSC::JSValue messageValue, PostMessageOptions&& options)
{
LOG(MessagePorts, "Attempting to post message to port %s (to be received by port %s)", m_identifier.logString().utf8().data(), m_remoteIdentifier.logString().utf8().data());
registerLocalActivity();
Vector<RefPtr<MessagePort>> ports;
auto messageData = SerializedScriptValue::create(state, messageValue, WTFMove(options.transfer), ports);
if (messageData.hasException())
return messageData.releaseException();
if (!isEntangled())
return { };
ASSERT(m_scriptExecutionContext);
TransferredMessagePortArray transferredPorts;
if (!ports.isEmpty()) {
for (auto& port : ports) {
if (port->identifier() == m_identifier || port->identifier() == m_remoteIdentifier)
return Exception { DataCloneError };
}
auto disentangleResult = MessagePort::disentanglePorts(WTFMove(ports));
if (disentangleResult.hasException())
return disentangleResult.releaseException();
transferredPorts = disentangleResult.releaseReturnValue();
}
MessageWithMessagePorts message { messageData.releaseReturnValue(), WTFMove(transferredPorts) };
LOG(MessagePorts, "Actually posting message to port %s (to be received by port %s)", m_identifier.logString().utf8().data(), m_remoteIdentifier.logString().utf8().data());
MessagePortChannelProvider::fromContext(*m_scriptExecutionContext).postMessageToRemote(WTFMove(message), m_remoteIdentifier);
return { };
}
void MessagePort::disentangle()
{
ASSERT(m_entangled);
m_entangled = false;
registerLocalActivity();
MessagePortChannelProvider::fromContext(*m_scriptExecutionContext).messagePortDisentangled(m_identifier);
m_scriptExecutionContext->destroyedMessagePort(*this);
m_scriptExecutionContext->willDestroyActiveDOMObject(*this);
m_scriptExecutionContext->willDestroyDestructionObserver(*this);
m_scriptExecutionContext = nullptr;
}
void MessagePort::registerLocalActivity()
{
m_hasHadLocalActivitySinceLastCheck = true;
m_mightBeEligibleForGC = false;
}
void MessagePort::messageAvailable()
{
if (!m_scriptExecutionContext || m_scriptExecutionContext->activeDOMObjectsAreSuspended())
return;
m_scriptExecutionContext->processMessageWithMessagePortsSoon();
}
void MessagePort::start()
{
if (!isEntangled())
return;
registerLocalActivity();
ASSERT(m_scriptExecutionContext);
if (m_started)
return;
m_started = true;
m_scriptExecutionContext->processMessageWithMessagePortsSoon();
}
void MessagePort::close()
{
m_mightBeEligibleForGC = true;
if (m_closed)
return;
m_closed = true;
if (isMainThread())
MessagePortChannelProvider::singleton().messagePortClosed(m_identifier);
else {
callOnMainThread([identifier = m_identifier] {
MessagePortChannelProvider::singleton().messagePortClosed(identifier);
});
}
removeAllEventListeners();
}
void MessagePort::contextDestroyed()
{
ASSERT(m_scriptExecutionContext);
close();
m_scriptExecutionContext = nullptr;
}
void MessagePort::dispatchMessages()
{
ASSERT(started());
if (!m_scriptExecutionContext || m_scriptExecutionContext->activeDOMObjectsAreSuspended() || !isEntangled())
return;
auto messagesTakenHandler = [this, weakThis = makeWeakPtr(this)](Vector<MessageWithMessagePorts>&& messages, Function<void()>&& completionCallback) mutable {
auto scopeExit = makeScopeExit(WTFMove(completionCallback));
if (!weakThis)
return;
LOG(MessagePorts, "MessagePort %s (%p) dispatching %zu messages", m_identifier.logString().utf8().data(), this, messages.size());
if (!m_scriptExecutionContext)
return;
if (!messages.isEmpty())
registerLocalActivity();
ASSERT(m_scriptExecutionContext->isContextThread());
bool contextIsWorker = is<WorkerGlobalScope>(*m_scriptExecutionContext);
for (auto& message : messages) {
if (contextIsWorker && downcast<WorkerGlobalScope>(*m_scriptExecutionContext).isClosing())
return;
auto ports = MessagePort::entanglePorts(*m_scriptExecutionContext, WTFMove(message.transferredPorts));
queueTaskToDispatchEvent(*this, TaskSource::PostedMessageQueue, MessageEvent::create(WTFMove(ports), message.message.releaseNonNull()));
}
};
MessagePortChannelProvider::fromContext(*m_scriptExecutionContext).takeAllMessagesForPort(m_identifier, WTFMove(messagesTakenHandler));
}
void MessagePort::dispatchEvent(Event& event)
{
if (m_closed)
return;
if (is<WorkerGlobalScope>(*m_scriptExecutionContext) && downcast<WorkerGlobalScope>(*m_scriptExecutionContext).isClosing())
return;
EventTarget::dispatchEvent(event);
}
void MessagePort::updateActivity(MessagePortChannelProvider::HasActivity hasActivity)
{
bool hasHadLocalActivity = m_hasHadLocalActivitySinceLastCheck;
m_hasHadLocalActivitySinceLastCheck = false;
if (hasActivity == MessagePortChannelProvider::HasActivity::No && !hasHadLocalActivity)
m_isRemoteEligibleForGC = true;
if (hasActivity == MessagePortChannelProvider::HasActivity::Yes)
m_isRemoteEligibleForGC = false;
m_isAskingRemoteAboutGC = false;
}
bool MessagePort::virtualHasPendingActivity() const
{
m_mightBeEligibleForGC = true;
if (!m_scriptExecutionContext || m_closed)
return false;
if (!m_hasHadLocalActivitySinceLastCheck && m_isRemoteEligibleForGC)
return false;
if (!m_hasMessageEventListener)
return false;
if (!m_isAskingRemoteAboutGC) {
RefPtr<WorkerThread> workerThread;
if (is<WorkerGlobalScope>(*m_scriptExecutionContext))
workerThread = &downcast<WorkerGlobalScope>(*m_scriptExecutionContext).thread();
callOnMainThread([remoteIdentifier = m_remoteIdentifier, weakThis = makeWeakPtr(const_cast<MessagePort*>(this)), workerThread = WTFMove(workerThread)]() mutable {
MessagePortChannelProvider::singleton().checkRemotePortForActivity(remoteIdentifier, [weakThis = WTFMove(weakThis), workerThread = WTFMove(workerThread)](auto hasActivity) mutable {
if (!workerThread) {
if (weakThis)
weakThis->updateActivity(hasActivity);
return;
}
workerThread->runLoop().postTaskForMode([weakThis = WTFMove(weakThis), hasActivity](auto&) mutable {
if (weakThis)
weakThis->updateActivity(hasActivity);
}, WorkerRunLoop::defaultMode());
});
});
m_isAskingRemoteAboutGC = true;
}
return true;
}
bool MessagePort::isLocallyReachable() const
{
return !m_mightBeEligibleForGC;
}
MessagePort* MessagePort::locallyEntangledPort() const
{
return nullptr;
}
ExceptionOr<TransferredMessagePortArray> MessagePort::disentanglePorts(Vector<RefPtr<MessagePort>>&& ports)
{
if (ports.isEmpty())
return TransferredMessagePortArray { };
HashSet<MessagePort*> portSet;
for (auto& port : ports) {
if (!port || !port->m_entangled || !portSet.add(port.get()).isNewEntry)
return Exception { DataCloneError };
}
TransferredMessagePortArray portArray;
portArray.reserveInitialCapacity(ports.size());
for (auto& port : ports) {
portArray.uncheckedAppend({ port->identifier(), port->remoteIdentifier() });
port->disentangle();
}
return portArray;
}
Vector<RefPtr<MessagePort>> MessagePort::entanglePorts(ScriptExecutionContext& context, TransferredMessagePortArray&& transferredPorts)
{
LOG(MessagePorts, "Entangling %zu transferred ports to ScriptExecutionContext %s (%p)", transferredPorts.size(), context.url().string().utf8().data(), &context);
if (transferredPorts.isEmpty())
return { };
Vector<RefPtr<MessagePort>> ports;
ports.reserveInitialCapacity(transferredPorts.size());
for (auto& transferredPort : transferredPorts) {
auto port = MessagePort::create(context, transferredPort.first, transferredPort.second);
port->entangle();
ports.uncheckedAppend(WTFMove(port));
}
return ports;
}
bool MessagePort::addEventListener(const AtomString& eventType, Ref<EventListener>&& listener, const AddEventListenerOptions& options)
{
if (eventType == eventNames().messageEvent) {
if (listener->isAttribute())
start();
m_hasMessageEventListener = true;
registerLocalActivity();
}
return EventTargetWithInlineData::addEventListener(eventType, WTFMove(listener), options);
}
bool MessagePort::removeEventListener(const AtomString& eventType, EventListener& listener, const ListenerOptions& options)
{
auto result = EventTargetWithInlineData::removeEventListener(eventType, listener, options);
if (!hasEventListeners(eventNames().messageEvent))
m_hasMessageEventListener = false;
return result;
}
const char* MessagePort::activeDOMObjectName() const
{
return "MessagePort";
}
}