Source code
Revision control
Copy as Markdown
Other Tools
/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this file,
#include "mozilla/ipc/NodeController.h"
#include "MainThreadUtils.h"
#include "base/process_util.h"
#include "chrome/common/ipc_message.h"
#include "mojo/core/ports/name.h"
#include "mojo/core/ports/node.h"
#include "mojo/core/ports/port_locker.h"
#include "mozilla/AlreadyAddRefed.h"
#include "mozilla/RandomNum.h"
#include "mozilla/StaticPtr.h"
#include "mozilla/Assertions.h"
#include "mozilla/ToString.h"
#include "mozilla/ipc/BrowserProcessSubThread.h"
#include "mozilla/ipc/ProtocolUtils.h"
#include "mozilla/mozalloc.h"
#include "nsISerialEventTarget.h"
#include "nsTArray.h"
#include "nsXULAppAPI.h"
#include "nsPrintfCString.h"
#define PORTS_ALWAYS_OK(expr) MOZ_ALWAYS_TRUE(mojo::core::ports::OK == (expr))
namespace mozilla::ipc {
static StaticRefPtr<NodeController> gNodeController;
static LazyLogModule gNodeControllerLog{"NodeController"};
// Helper logger macro which includes the name of the `this` NodeController in
// the logged messages.
#define NODECONTROLLER_LOG(level_, fmt_, ...) \
MOZ_LOG(gNodeControllerLog, level_, \
("[%s]: " fmt_, ToString(mName).c_str(), ##__VA_ARGS__))
// Helper warning macro which both does logger logging and emits NS_WARNING logs
// under debug mode.
#ifdef DEBUG
# define NODECONTROLLER_WARNING(fmt_, ...) \
do { \
nsPrintfCString warning("[%s]: " fmt_, ToString(mName).c_str(), \
##__VA_ARGS__); \
NS_WARNING(warning.get()); \
MOZ_LOG(gNodeControllerLog, LogLevel::Debug, ("%s", warning.get())); \
} while (0)
#else
# define NODECONTROLLER_WARNING(fmt_, ...) \
NODECONTROLLER_LOG(LogLevel::Warning, fmt_, ##__VA_ARGS__)
#endif
NodeController::NodeController(const NodeName& aName)
: mName(aName), mNode(MakeUnique<Node>(aName, this)) {}
NodeController::~NodeController() {
auto state = mState.Lock();
MOZ_RELEASE_ASSERT(state->mPeers.IsEmpty(),
"Destroying NodeController before closing all peers");
MOZ_RELEASE_ASSERT(state->mInvites.IsEmpty(),
"Destroying NodeController before closing all invites");
};
// FIXME: Actually provide some way to create the thing.
/* static */ NodeController* NodeController::GetSingleton() {
MOZ_ASSERT(gNodeController);
return gNodeController;
}
std::pair<ScopedPort, ScopedPort> NodeController::CreatePortPair() {
PortRef port0, port1;
PORTS_ALWAYS_OK(mNode->CreatePortPair(&port0, &port1));
return {ScopedPort{std::move(port0), this},
ScopedPort{std::move(port1), this}};
}
auto NodeController::GetPort(const PortName& aName) -> PortRef {
PortRef port;
int rv = mNode->GetPort(aName, &port);
if (NS_WARN_IF(rv != mojo::core::ports::OK)) {
NODECONTROLLER_WARNING("Call to GetPort(%s) Failed",
ToString(aName).c_str());
return {};
}
return port;
}
void NodeController::SetPortObserver(const PortRef& aPort,
PortObserver* aObserver) {
PORTS_ALWAYS_OK(mNode->SetUserData(aPort, aObserver));
}
auto NodeController::GetStatus(const PortRef& aPort) -> Maybe<PortStatus> {
PortStatus status{};
int rv = mNode->GetStatus(aPort, &status);
if (rv != mojo::core::ports::OK) {
return Nothing();
}
return Some(status);
}
void NodeController::ClosePort(const PortRef& aPort) {
PORTS_ALWAYS_OK(mNode->ClosePort(aPort));
}
bool NodeController::GetMessage(const PortRef& aPort,
UniquePtr<IPC::Message>* aMessage) {
UniquePtr<UserMessageEvent> messageEvent;
int rv = mNode->GetMessage(aPort, &messageEvent, nullptr);
if (rv != mojo::core::ports::OK) {
if (rv == mojo::core::ports::ERROR_PORT_PEER_CLOSED) {
return false;
}
MOZ_CRASH("GetMessage on port in invalid state");
}
if (messageEvent) {
UniquePtr<IPC::Message> message = messageEvent->TakeMessage<IPC::Message>();
// If our UserMessageEvent has any ports directly attached to it, fetch them
// from our node and attach them to the IPC::Message we extracted.
//
// It's important to only do this if we have nonempty set of ports on the
// event, as we may have never serialized our IPC::Message's ports onto the
// event if it was routed in-process.
if (messageEvent->num_ports() > 0) {
nsTArray<ScopedPort> attachedPorts(messageEvent->num_ports());
for (size_t i = 0; i < messageEvent->num_ports(); ++i) {
attachedPorts.AppendElement(
ScopedPort{GetPort(messageEvent->ports()[i]), this});
}
message->SetAttachedPorts(std::move(attachedPorts));
}
*aMessage = std::move(message);
} else {
*aMessage = nullptr;
}
return true;
}
bool NodeController::SendUserMessage(const PortRef& aPort,
UniquePtr<IPC::Message> aMessage) {
auto messageEvent = MakeUnique<UserMessageEvent>(0);
messageEvent->AttachMessage(std::move(aMessage));
int rv = mNode->SendUserMessage(aPort, std::move(messageEvent));
if (rv == mojo::core::ports::OK) {
return true;
}
if (rv == mojo::core::ports::ERROR_PORT_PEER_CLOSED) {
NODECONTROLLER_LOG(LogLevel::Debug,
"Ignoring message to port %s as peer was closed",
ToString(aPort.name()).c_str());
return true;
}
NODECONTROLLER_WARNING("Failed to send message to port %s",
ToString(aPort.name()).c_str());
return false;
}
auto NodeController::SerializeEventMessage(UniquePtr<Event> aEvent,
const NodeName* aRelayTarget,
uint32_t aType)
-> UniquePtr<IPC::Message> {
UniquePtr<IPC::Message> message;
if (aEvent->type() == Event::kUserMessage) {
MOZ_DIAGNOSTIC_ASSERT(
aType == EVENT_MESSAGE_TYPE,
"Can only send a UserMessage in an EVENT_MESSAGE_TYPE");
message = static_cast<UserMessageEvent*>(aEvent.get())
->TakeMessage<IPC::Message>();
} else {
message = MakeUnique<IPC::Message>(MSG_ROUTING_CONTROL, aType);
}
message->set_relay(aRelayTarget != nullptr);
size_t length = aEvent->GetSerializedSize();
if (aRelayTarget) {
length += sizeof(NodeName);
}
// Use an intermediate buffer to serialize to avoid potential issues with the
// segmented `IPC::Message` bufferlist. This should be fairly cheap, as the
// majority of events are fairly small.
Vector<char, 256, InfallibleAllocPolicy> buffer;
(void)buffer.initLengthUninitialized(length);
if (aRelayTarget) {
memcpy(buffer.begin(), aRelayTarget, sizeof(NodeName));
aEvent->Serialize(buffer.begin() + sizeof(NodeName));
} else {
aEvent->Serialize(buffer.begin());
}
message->WriteFooter(buffer.begin(), buffer.length());
message->set_event_footer_size(buffer.length());
#ifdef DEBUG
// Debug-assert that we can read the same data back out of the buffer.
MOZ_ASSERT(message->event_footer_size() == length);
Vector<char, 256, InfallibleAllocPolicy> buffer2;
(void)buffer2.initLengthUninitialized(message->event_footer_size());
MOZ_ASSERT(message->ReadFooter(buffer2.begin(), buffer2.length(),
/* truncate */ false));
MOZ_ASSERT(!memcmp(buffer2.begin(), buffer.begin(), buffer.length()));
#endif
return message;
}
auto NodeController::DeserializeEventMessage(UniquePtr<IPC::Message> aMessage,
NodeName* aRelayTarget)
-> UniquePtr<Event> {
if (aMessage->is_relay() && !aRelayTarget) {
NODECONTROLLER_WARNING("Unexpected relay message '%s'", aMessage->name());
return nullptr;
}
Vector<char, 256, InfallibleAllocPolicy> buffer;
(void)buffer.initLengthUninitialized(aMessage->event_footer_size());
// Truncate the message when reading the footer, so that the extra footer data
// is no longer present in the message. This allows future code to eventually
// send the same `IPC::Message` to another process.
if (!aMessage->ReadFooter(buffer.begin(), buffer.length(),
/* truncate */ true)) {
NODECONTROLLER_WARNING("Call to ReadFooter for message '%s' Failed",
aMessage->name());
return nullptr;
}
aMessage->set_event_footer_size(0);
UniquePtr<Event> event;
if (aRelayTarget) {
MOZ_ASSERT(aMessage->is_relay());
if (buffer.length() < sizeof(NodeName)) {
NODECONTROLLER_WARNING(
"Insufficient space in message footer for message '%s'",
aMessage->name());
return nullptr;
}
memcpy(aRelayTarget, buffer.begin(), sizeof(NodeName));
event = Event::Deserialize(buffer.begin() + sizeof(NodeName),
buffer.length() - sizeof(NodeName));
} else {
event = Event::Deserialize(buffer.begin(), buffer.length());
}
if (!event) {
NODECONTROLLER_WARNING("Call to Event::Deserialize for message '%s' Failed",
aMessage->name());
return nullptr;
}
if (event->type() == Event::kUserMessage) {
static_cast<UserMessageEvent*>(event.get())
->AttachMessage(std::move(aMessage));
}
return event;
}
already_AddRefed<NodeChannel> NodeController::GetNodeChannel(
const NodeName& aName) {
auto state = mState.Lock();
return do_AddRef(state->mPeers.Get(aName));
}
void NodeController::DropPeer(NodeName aNodeName) {
AssertIOThread();
#ifdef FUZZING_SNAPSHOT
MOZ_FUZZING_IPC_DROP_PEER("NodeController::DropPeer");
#endif
Invite invite;
RefPtr<NodeChannel> channel;
nsTArray<PortRef> pendingMerges;
{
auto state = mState.Lock();
state->mPeers.Remove(aNodeName, &channel);
state->mPendingMessages.Remove(aNodeName);
state->mInvites.Remove(aNodeName, &invite);
state->mPendingMerges.Remove(aNodeName, &pendingMerges);
}
NODECONTROLLER_LOG(LogLevel::Info, "Dropping Peer %s (pid: %" PRIPID ")",
ToString(aNodeName).c_str(),
channel ? channel->OtherPid() : base::kInvalidProcessId);
if (channel) {
channel->Close();
}
if (invite.mChannel) {
invite.mChannel->Close();
}
if (invite.mToMerge.is_valid()) {
// Ignore any possible errors here.
(void)mNode->ClosePort(invite.mToMerge);
}
for (auto& port : pendingMerges) {
// Ignore any possible errors here.
(void)mNode->ClosePort(port);
}
mNode->LostConnectionToNode(aNodeName);
}
void NodeController::ContactRemotePeer(const NodeName& aNode,
UniquePtr<Event> aEvent) {
// On Windows and macOS, messages holding HANDLEs or mach ports must be
// relayed via the broker process so it can transfer ownership.
bool needsRelay = false;
#if defined(XP_WIN) || defined(XP_DARWIN)
if (aEvent && !IsBroker() && aNode != kBrokerNodeName &&
aEvent->type() == Event::kUserMessage) {
auto* userEvent = static_cast<UserMessageEvent*>(aEvent.get());
needsRelay =
userEvent->HasMessage() &&
userEvent->GetMessage<IPC::Message>()->num_relayed_attachments() > 0;
}
#endif
UniquePtr<IPC::Message> message;
if (aEvent) {
message =
SerializeEventMessage(std::move(aEvent), needsRelay ? &aNode : nullptr);
MOZ_ASSERT(message->is_relay() == needsRelay,
"Message relay status set incorrectly");
}
RefPtr<NodeChannel> peer;
RefPtr<NodeChannel> broker;
bool needsIntroduction = false;
bool needsBroker = needsRelay;
{
auto state = mState.Lock();
// Check if we know this peer. If we don't, we'll need to request an
// introduction.
peer = state->mPeers.Get(aNode);
if (!peer) {
// We don't know the peer, check if we've already requested an
// introduction, or if we need to request a new one.
auto& queue = state->mPendingMessages.LookupOrInsertWith(aNode, [&]() {
needsIntroduction = true;
needsBroker = true;
return Queue<UniquePtr<IPC::Message>, 64>{};
});
// If we aren't relaying, queue up the message to be sent.
if (message && !needsRelay) {
queue.Push(std::move(message));
}
}
if (needsBroker && !IsBroker()) {
broker = state->mPeers.Get(kBrokerNodeName);
}
}
if (needsBroker && !broker) {
NODECONTROLLER_WARNING(
"Dropping message '%s'; no connection to unknown peer %s",
message ? message->name() : "<null>", ToString(aNode).c_str());
if (needsIntroduction) {
// We have no broker and will never be able to be introduced to this node.
// Queue a task to clean up any ports connected to it.
XRE_GetIOMessageLoop()->PostTask(NewRunnableMethod<NodeName>(
"NodeController::DropPeer", this, &NodeController::DropPeer, aNode));
}
return;
}
if (needsIntroduction) {
NODECONTROLLER_LOG(LogLevel::Info, "Requesting introduction to peer %s",
ToString(aNode).c_str());
broker->RequestIntroduction(aNode);
}
if (message) {
if (needsRelay) {
NODECONTROLLER_LOG(LogLevel::Info,
"Relaying message '%s' for peer %s due to %" PRIu32
" attachments",
message->name(), ToString(aNode).c_str(),
message->num_relayed_attachments());
MOZ_ASSERT(message->num_relayed_attachments() > 0 && broker);
broker->SendEventMessage(std::move(message));
} else if (peer) {
peer->SendEventMessage(std::move(message));
}
}
}
void NodeController::ForwardEvent(const NodeName& aNode,
UniquePtr<Event> aEvent) {
MOZ_ASSERT(aEvent, "cannot forward null event");
if (aNode == mName) {
(void)mNode->AcceptEvent(mName, std::move(aEvent));
} else {
ContactRemotePeer(aNode, std::move(aEvent));
}
}
void NodeController::BroadcastEvent(UniquePtr<Event> aEvent) {
UniquePtr<IPC::Message> message =
SerializeEventMessage(std::move(aEvent), nullptr, BROADCAST_MESSAGE_TYPE);
if (IsBroker()) {
OnBroadcast(mName, std::move(message));
} else if (RefPtr<NodeChannel> broker = GetNodeChannel(kBrokerNodeName)) {
broker->Broadcast(std::move(message));
} else {
NODECONTROLLER_WARNING(
"Trying to broadcast event, but no connection to broker");
}
}
void NodeController::PortStatusChanged(const PortRef& aPortRef) {
RefPtr<UserData> userData;
int rv = mNode->GetUserData(aPortRef, &userData);
if (rv != mojo::core::ports::OK) {
NODECONTROLLER_WARNING("GetUserData call for port '%s' failed",
ToString(aPortRef.name()).c_str());
return;
}
if (userData) {
// All instances of `UserData` attached to ports in this node must be of
// type `PortObserver`, so we can call `OnPortStatusChanged` directly on
// them.
static_cast<PortObserver*>(userData.get())->OnPortStatusChanged();
}
}
void NodeController::ObserveRemoteNode(const NodeName& aNode) {
MOZ_ASSERT(aNode != mName);
ContactRemotePeer(aNode, nullptr);
}
void NodeController::OnEventMessage(const NodeName& aFromNode,
UniquePtr<IPC::Message> aMessage) {
AssertIOThread();
bool isRelay = aMessage->is_relay();
if (isRelay && aMessage->num_relayed_attachments() == 0) {
NODECONTROLLER_WARNING(
"Invalid relay message without relayed attachments from peer %s!",
ToString(aFromNode).c_str());
DropPeer(aFromNode);
return;
}
NodeName relayTarget;
UniquePtr<Event> event = DeserializeEventMessage(
std::move(aMessage), isRelay ? &relayTarget : nullptr);
if (!event) {
NODECONTROLLER_WARNING("Invalid EventMessage from peer %s!",
ToString(aFromNode).c_str());
DropPeer(aFromNode);
return;
}
NodeName fromNode = aFromNode;
#if defined(XP_WIN) || defined(XP_DARWIN)
if (isRelay) {
if (event->type() != Event::kUserMessage) {
NODECONTROLLER_WARNING(
"Unexpected relay of non-UserMessage event from peer %s!",
ToString(aFromNode).c_str());
DropPeer(aFromNode);
return;
}
// If we're the broker, then we'll need to forward this message on to the
// true recipient. To do this, we re-serialize the message, passing along
// the original source node, and send it to the final node.
if (IsBroker()) {
UniquePtr<IPC::Message> message =
SerializeEventMessage(std::move(event), &aFromNode);
if (!message) {
NODECONTROLLER_WARNING(
"Relaying EventMessage from peer %s failed to re-serialize!",
ToString(aFromNode).c_str());
DropPeer(aFromNode);
return;
}
MOZ_ASSERT(message->is_relay(), "Message stopped being a relay message?");
MOZ_ASSERT(message->num_relayed_attachments() > 0,
"Message doesn't have relayed attachments?");
NODECONTROLLER_LOG(
LogLevel::Info,
"Relaying message '%s' from peer %s to peer %s (%" PRIu32
" attachments)",
message->name(), ToString(aFromNode).c_str(),
ToString(relayTarget).c_str(), message->num_relayed_attachments());
RefPtr<NodeChannel> peer;
{
auto state = mState.Lock();
peer = state->mPeers.Get(relayTarget);
}
if (!peer) {
NODECONTROLLER_WARNING(
"Dropping relayed message from %s to unknown peer %s",
ToString(aFromNode).c_str(), ToString(relayTarget).c_str());
return;
}
peer->SendEventMessage(std::move(message));
return;
}
// Otherwise, we're the final recipient, so we can continue & process the
// message as usual.
if (aFromNode != kBrokerNodeName) {
NODECONTROLLER_WARNING(
"Unexpected relayed EventMessage from non-broker peer %s!",
ToString(aFromNode).c_str());
DropPeer(aFromNode);
return;
}
fromNode = relayTarget;
NODECONTROLLER_LOG(LogLevel::Info, "Got relayed message from peer %s",
ToString(fromNode).c_str());
}
#endif
// If we're getting a requested port merge from another process, check to make
// sure that we're expecting the request, and record that the merge has
// arrived so we don't try to close the port on error.
if (event->type() == Event::kMergePort) {
// Check that the target port for the merge actually exists.
auto targetPort = GetPort(event->port_name());
if (!targetPort.is_valid()) {
NODECONTROLLER_WARNING(
"Unexpected MergePortEvent from peer %s for unknown port %s",
ToString(fromNode).c_str(), ToString(event->port_name()).c_str());
DropPeer(fromNode);
return;
}
// Check if `targetPort` is in our pending merges entry for the given source
// node. If this makes the `mPendingMerges` entry empty, remove it.
bool expectingMerge = [&] {
auto state = mState.Lock();
auto pendingMerges = state->mPendingMerges.Lookup(aFromNode);
if (!pendingMerges) {
return false;
}
size_t removed = pendingMerges->RemoveElementsBy(
[&](auto& port) { return port.name() == targetPort.name(); });
if (removed != 0 && pendingMerges->IsEmpty()) {
pendingMerges.Remove();
}
return removed != 0;
}();
if (!expectingMerge) {
NODECONTROLLER_WARNING(
"Unexpected MergePortEvent from peer %s for port %s",
ToString(fromNode).c_str(), ToString(event->port_name()).c_str());
DropPeer(fromNode);
return;
}
}
(void)mNode->AcceptEvent(fromNode, std::move(event));
}
void NodeController::OnBroadcast(const NodeName& aFromNode,
UniquePtr<IPC::Message> aMessage) {
MOZ_DIAGNOSTIC_ASSERT(aMessage->type() == BROADCAST_MESSAGE_TYPE);
// NOTE: This method may be called off of the IO thread by the
// `BroadcastEvent` node callback.
if (!IsBroker()) {
NODECONTROLLER_WARNING("Broadcast request received by non-broker node");
return;
}
UniquePtr<Event> event = DeserializeEventMessage(std::move(aMessage));
if (!event) {
NODECONTROLLER_WARNING("Invalid broadcast message from peer");
return;
}
nsTArray<RefPtr<NodeChannel>> peers;
{
auto state = mState.Lock();
peers.SetCapacity(state->mPeers.Count());
for (const auto& peer : state->mPeers.Values()) {
peers.AppendElement(peer);
}
}
for (const auto& peer : peers) {
// NOTE: This `clone` operation is only supported for a limited number of
// message types by the ports API, which provides some extra security by
// only allowing those specific types of messages to be broadcasted.
// Messages which don't support `CloneForBroadcast` cannot be broadcast, and
// the ports library will not attempt to broadcast them.
auto clone = event->CloneForBroadcast();
if (!clone) {
NODECONTROLLER_WARNING("Attempt to broadcast unsupported message");
break;
}
peer->SendEventMessage(SerializeEventMessage(std::move(clone)));
}
}
void NodeController::OnIntroduce(const NodeName& aFromNode,
NodeChannel::Introduction aIntroduction) {
AssertIOThread();
if (aFromNode != kBrokerNodeName) {
NODECONTROLLER_WARNING("Introduction received from non-broker node");
DropPeer(aFromNode);
return;
}
MOZ_ASSERT(aIntroduction.mMyPid == base::GetCurrentProcId(),
"We're the wrong process to receive this?");
if (!aIntroduction.mHandle) {
NODECONTROLLER_WARNING("Could not be introduced to peer %s",
ToString(aIntroduction.mName).c_str());
mNode->LostConnectionToNode(aIntroduction.mName);
auto state = mState.Lock();
state->mPendingMessages.Remove(aIntroduction.mName);
return;
}
auto channel =
MakeUnique<IPC::Channel>(std::move(aIntroduction.mHandle),
aIntroduction.mMode, aIntroduction.mOtherPid);
auto nodeChannel = MakeRefPtr<NodeChannel>(
aIntroduction.mName, std::move(channel), this, aIntroduction.mOtherPid);
{
auto state = mState.Lock();
bool isNew = false;
state->mPeers.LookupOrInsertWith(aIntroduction.mName, [&]() {
isNew = true;
return nodeChannel;
});
if (!isNew) {
// We got a duplicate introduction. This can happen during normal
// execution if both sides request an introduction at the same time. We
// can just ignore the second one, as they'll arrive in the same order in
// both processes.
nodeChannel->Close();
return;
}
// Deliver any pending messages, then remove the entry from our table. We do
// this while `mState` is still held to ensure that these messages are
// all sent before another thread can observe the newly created channel.
// As the channel hasn't been `Connect()`-ed yet, this will only queue the
// messages up to be sent, so is OK to do with the mutex held. These
// messages will be processed to be sent during `Start()` below, which is
// performed outside of the lock.
if (auto pending = state->mPendingMessages.Lookup(aIntroduction.mName)) {
while (!pending->IsEmpty()) {
nodeChannel->SendEventMessage(pending->Pop());
}
pending.Remove();
}
}
// NodeChannel::Start must be called with the lock not held, as it may lead to
// callbacks being made into `OnChannelError` or `OnMessageReceived`, which
// will attempt to re-acquire our lock.
nodeChannel->Start();
}
void NodeController::OnRequestIntroduction(const NodeName& aFromNode,
const NodeName& aName) {
AssertIOThread();
if (NS_WARN_IF(!IsBroker())) {
return;
}
RefPtr<NodeChannel> peerA = GetNodeChannel(aFromNode);
if (!peerA || aName == mojo::core::ports::kInvalidNodeName) {
NODECONTROLLER_WARNING("Invalid OnRequestIntroduction message from node %s",
ToString(aFromNode).c_str());
DropPeer(aFromNode);
return;
}
RefPtr<NodeChannel> peerB = GetNodeChannel(aName);
IPC::Channel::ChannelHandle handleA, handleB;
if (!peerB || !IPC::Channel::CreateRawPipe(&handleA, &handleB)) {
NODECONTROLLER_WARNING(
"Rejecting introduction request from '%s' for unknown peer '%s'",
ToString(aFromNode).c_str(), ToString(aName).c_str());
// We don't know this peer, or ran into issues creating the descriptor! Send
// an invalid introduction to content to clean up any pending outbound
// messages.
NodeChannel::Introduction intro{aName, nullptr, IPC::Channel::MODE_SERVER,
peerA->OtherPid(), base::kInvalidProcessId};
peerA->Introduce(std::move(intro));
return;
}
NodeChannel::Introduction introA{aName, std::move(handleA),
IPC::Channel::MODE_SERVER, peerA->OtherPid(),
peerB->OtherPid()};
NodeChannel::Introduction introB{aFromNode, std::move(handleB),
IPC::Channel::MODE_CLIENT, peerB->OtherPid(),
peerA->OtherPid()};
peerA->Introduce(std::move(introA));
peerB->Introduce(std::move(introB));
}
void NodeController::OnAcceptInvite(const NodeName& aFromNode,
const NodeName& aRealName,
const PortName& aInitialPort) {
AssertIOThread();
if (!IsBroker()) {
NODECONTROLLER_WARNING("Ignoring AcceptInvite message as non-broker");
return;
}
if (aRealName == mojo::core::ports::kInvalidNodeName ||
aInitialPort == mojo::core::ports::kInvalidPortName) {
NODECONTROLLER_WARNING("Invalid name in AcceptInvite message");
DropPeer(aFromNode);
return;
}
bool inserted = false;
Invite invite;
{
auto state = mState.Lock();
MOZ_ASSERT(state->mPendingMessages.IsEmpty(),
"Shouldn't have pending messages in broker");
// Try to remove the source node from our invites list and insert it into
// our peers map under the new name.
if (state->mInvites.Remove(aFromNode, &invite)) {
MOZ_ASSERT(invite.mChannel && invite.mToMerge.is_valid());
state->mPeers.LookupOrInsertWith(aRealName, [&]() {
inserted = true;
return invite.mChannel;
});
}
}
if (!inserted) {
NODECONTROLLER_WARNING("Invalid AcceptInvite message from node %s",
ToString(aFromNode).c_str());
DropPeer(aFromNode);
return;
}
// Update the name of the node. This field is only accessed from the IO
// thread, so it's safe to update it without a lock held.
invite.mChannel->SetName(aRealName);
// Start the port merge to allow our existing initial port to begin
// communicating with the remote port.
PORTS_ALWAYS_OK(mNode->MergePorts(invite.mToMerge, aRealName, aInitialPort));
}
void NodeController::OnChannelError(const NodeName& aFromNode) {
AssertIOThread();
DropPeer(aFromNode);
}
static mojo::core::ports::NodeName RandomNodeName() {
return {RandomUint64OrDie(), RandomUint64OrDie()};
}
std::tuple<ScopedPort, RefPtr<NodeChannel>> NodeController::InviteChildProcess(
UniquePtr<IPC::Channel> aChannel,
GeckoChildProcessHost* aChildProcessHost) {
MOZ_ASSERT(IsBroker());
AssertIOThread();
// Create the peer with a randomly generated name, and store it in `mInvites`.
// This channel and name will be used for communication with the node until it
// sends us its' real name in an `AcceptInvite` message.
auto ports = CreatePortPair();
auto inviteName = RandomNodeName();
auto nodeChannel =
MakeRefPtr<NodeChannel>(inviteName, std::move(aChannel), this,
base::kInvalidProcessId, aChildProcessHost);
{
auto state = mState.Lock();
MOZ_DIAGNOSTIC_ASSERT(!state->mPeers.Contains(inviteName),
"UUID conflict?");
MOZ_DIAGNOSTIC_ASSERT(!state->mInvites.Contains(inviteName),
"UUID conflict?");
state->mInvites.InsertOrUpdate(inviteName,
Invite{nodeChannel, ports.second.Release()});
}
nodeChannel->Start();
return std::tuple{std::move(ports.first), std::move(nodeChannel)};
}
void NodeController::InitBrokerProcess() {
AssertIOThread();
MOZ_ASSERT(!gNodeController);
gNodeController = new NodeController(kBrokerNodeName);
}
ScopedPort NodeController::InitChildProcess(UniquePtr<IPC::Channel> aChannel,
base::ProcessId aParentPid) {
AssertIOThread();
MOZ_ASSERT(!gNodeController);
auto nodeName = RandomNodeName();
gNodeController = new NodeController(nodeName);
auto ports = gNodeController->CreatePortPair();
PortRef toMerge = ports.second.Release();
// Mark the port as expecting a pending merge. This is a duplicate of the
// information tracked by `mPendingMerges`, and was added by upstream
// chromium.
{
mojo::core::ports::SinglePortLocker locker(&toMerge);
locker.port()->pending_merge_peer = true;
}
auto nodeChannel = MakeRefPtr<NodeChannel>(
kBrokerNodeName, std::move(aChannel), gNodeController, aParentPid);
{
auto state = gNodeController->mState.Lock();
MOZ_DIAGNOSTIC_ASSERT(!state->mPeers.Contains(kBrokerNodeName));
state->mPeers.InsertOrUpdate(kBrokerNodeName, nodeChannel);
MOZ_DIAGNOSTIC_ASSERT(!state->mPendingMerges.Contains(kBrokerNodeName));
state->mPendingMerges.LookupOrInsert(kBrokerNodeName)
.AppendElement(toMerge);
}
nodeChannel->Start();
nodeChannel->AcceptInvite(nodeName, toMerge.name());
return std::move(ports.first);
}
void NodeController::CleanUp() {
AssertIOThread();
MOZ_ASSERT(gNodeController);
RefPtr<NodeController> nodeController = gNodeController;
gNodeController = nullptr;
// Collect all objects from our state which need to be cleaned up.
nsTArray<NodeName> lostConnections;
nsTArray<RefPtr<NodeChannel>> channelsToClose;
nsTArray<PortRef> portsToClose;
{
auto state = nodeController->mState.Lock();
for (const auto& chan : state->mPeers) {
lostConnections.AppendElement(chan.GetKey());
channelsToClose.AppendElement(chan.GetData());
}
for (const auto& pending : state->mPendingMessages.Keys()) {
lostConnections.AppendElement(pending);
}
for (const auto& invite : state->mInvites.Values()) {
channelsToClose.AppendElement(invite.mChannel);
portsToClose.AppendElement(invite.mToMerge);
}
for (const auto& pendingPorts : state->mPendingMerges.Values()) {
portsToClose.AppendElements(pendingPorts);
}
state->mPeers.Clear();
state->mPendingMessages.Clear();
state->mInvites.Clear();
state->mPendingMerges.Clear();
}
for (auto& nodeChannel : channelsToClose) {
nodeChannel->Close();
}
for (auto& port : portsToClose) {
nodeController->mNode->ClosePort(port);
}
for (auto& name : lostConnections) {
nodeController->mNode->LostConnectionToNode(name);
}
}
#undef NODECONTROLLER_LOG
#undef NODECONTROLLER_WARNING
} // namespace mozilla::ipc