[mojo-edk] Close pending merges if the parent channel dies.

mojo/edk/system/node_controller.cc

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "mojo/edk/system/node_controller.h"
 
 #include <algorithm>
 #include <limits>
 
 #include "base/bind.h"
@@ skipping 292 matching lines @@
       reserved_ports_.erase(it);
       was_merged = true;
     }
   }
   if (was_merged) {
     AcceptIncomingMessages();
     return;
   }
 
   scoped_refptr<NodeChannel> parent;
+  bool reject_merge = false;
   {
     // Hold |pending_port_merges_lock_| while getting |parent|. Otherwise,
     // there is a race where the parent can be set, and |pending_port_merges_|
     // be processed between retrieving |parent| and adding the merge to
     // |pending_port_merges_|.
     base::AutoLock lock(pending_port_merges_lock_);
     parent = GetParentChannel();
-    if (!parent) {
+    if (reject_pending_merges_) {
+      reject_merge = true;
+    } else if (!parent) {
       pending_port_merges_.push_back(std::make_pair(token, port));
       return;
     }
   }
+  if (reject_merge) {
+    node_->ClosePort(port);
+    DVLOG(2) << "Rejecting port merge for token " << token
+             << " due to closed parent channel.";
+    AcceptIncomingMessages();
+    return;
+  }
+
   parent->RequestPortMerge(port.name(), token);
 }
 
 int NodeController::MergeLocalPorts(const ports::PortRef& port0,
                                     const ports::PortRef& port1) {
   int rv = node_->MergeLocalPorts(port0, port1);
   AcceptIncomingMessages();
   return rv;
 }
 
@@ skipping 153 matching lines @@
   if (start_channel)
     channel->Start();
 
   // Flush any queued message we need to deliver to this node.
   while (!pending_messages.empty()) {
     channel->PortsMessage(std::move(pending_messages.front()));
     pending_messages.pop();
   }
 }
 
-void NodeController::DropPeer(const ports::NodeName& name) {
+void NodeController::DropPeer(const ports::NodeName& name,
+                              NodeChannel* channel) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   {
     base::AutoLock lock(peers_lock_);
     auto it = peers_.find(name);
 
     if (it != peers_.end()) {
       ports::NodeName peer = it->first;
       peers_.erase(it);
       DVLOG(1) << "Dropped peer " << peer;
@@ skipping 26 matching lines @@
       // We have to erase reserved ports in a two-step manner because the usual
       // manner of using the returned iterator from map::erase isn't technically
       // valid in C++11 (although it is in C++14).
       for (const auto& token : port_tokens)
         reserved_ports_.erase(token);
 
       pending_child_tokens_.erase(it);
     }
   }
 
+  bool is_parent;
+  {
+    base::AutoLock lock(parent_lock_);
+    is_parent = (name == parent_name_ || channel == bootstrap_parent_channel_);
+  }
+  // If the error comes from the parent channel, we also need to cancel any
+  // port merge requests, so that errors can be propagated to the message
+  // pipes.
+  if (is_parent) {
+    base::AutoLock lock(pending_port_merges_lock_);
+    reject_pending_merges_ = true;
+
+    for (const auto& port : pending_port_merges_)
+      ports_to_close.push_back(port.second);
+    pending_port_merges_.clear();
+  }
+
   for (const auto& port : ports_to_close)
     node_->ClosePort(port);
 
   node_->LostConnectionToNode(name);
 
   AcceptIncomingMessages();
 }
 
 void NodeController::SendPeerMessage(const ports::NodeName& name,
                                      ports::ScopedMessage message) {
@@ skipping 216 matching lines @@
   {
     base::AutoLock lock(parent_lock_);
     if (bootstrap_parent_channel_ && parent_name_ == ports::kInvalidNodeName) {
       parent_name_ = parent_name;
       parent = bootstrap_parent_channel_;
     }
   }
 
   if (!parent) {
     DLOG(ERROR) << "Unexpected AcceptChild message from " << from_node;
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   parent->SetRemoteNodeName(parent_name);
   parent->AcceptParent(token, name_);
 
   // NOTE: The child does not actually add its parent as a peer until
   // receiving an AcceptBrokerClient message from the broker. The parent
   // will request that said message be sent upon receiving AcceptParent.
 
   DVLOG(1) << "Child " << name_ << " accepting parent " << parent_name;
 }
 
 void NodeController::OnAcceptParent(const ports::NodeName& from_node,
                                     const ports::NodeName& token,
                                     const ports::NodeName& child_name) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   auto it = pending_children_.find(from_node);
   if (it == pending_children_.end() || token != from_node) {
     DLOG(ERROR) << "Received unexpected AcceptParent message from "
                 << from_node;
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   scoped_refptr<NodeChannel> channel = it->second;
   pending_children_.erase(it);
 
   DCHECK(channel);
 
   DVLOG(1) << "Parent " << name_ << " accepted child " << child_name;
 
@@ skipping 36 matching lines @@
       ScopedPlatformHandle(PlatformHandle(process_handle));
 #endif
   scoped_refptr<NodeChannel> sender = GetPeerChannel(from_node);
   if (!sender) {
     DLOG(ERROR) << "Ignoring AddBrokerClient from unknown sender.";
     return;
   }
 
   if (GetPeerChannel(client_name)) {
     DLOG(ERROR) << "Ignoring AddBrokerClient for known client.";
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   PlatformChannelPair broker_channel;
   scoped_refptr<NodeChannel> client = NodeChannel::Create(
       this, broker_channel.PassServerHandle(), io_task_runner_,
       ProcessErrorCallback());
 
 #if defined(OS_WIN)
   // The broker must have a working handle to the client process in order to
@@ skipping 110 matching lines @@
 
 void NodeController::OnPortsMessage(const ports::NodeName& from_node,
                                     Channel::MessagePtr channel_message) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   void* data;
   size_t num_data_bytes, num_header_bytes, num_payload_bytes, num_ports_bytes;
   if (!ParsePortsMessage(channel_message.get(), &data, &num_data_bytes,
                          &num_header_bytes, &num_payload_bytes,
                          &num_ports_bytes)) {
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   CHECK(channel_message);
   std::unique_ptr<PortsMessage> ports_message(
       new PortsMessage(num_header_bytes,
                        num_payload_bytes,
                        num_ports_bytes,
                        std::move(channel_message)));
   ports_message->set_source_node(from_node);
@@ skipping 30 matching lines @@
 }
 
 void NodeController::OnRequestIntroduction(const ports::NodeName& from_node,
                                            const ports::NodeName& name) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   scoped_refptr<NodeChannel> requestor = GetPeerChannel(from_node);
   if (from_node == name || name == ports::kInvalidNodeName || !requestor) {
     DLOG(ERROR) << "Rejecting invalid OnRequestIntroduction message from "
                 << from_node;
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   scoped_refptr<NodeChannel> new_friend = GetPeerChannel(name);
   if (!new_friend) {
     // We don't know who they're talking about!
     requestor->Introduce(name, ScopedPlatformHandle());
   } else {
     PlatformChannelPair new_channel;
     requestor->Introduce(name, new_channel.PassServerHandle());
@@ skipping 25 matching lines @@
 
 void NodeController::OnBroadcast(const ports::NodeName& from_node,
                                  Channel::MessagePtr message) {
   DCHECK(!message->has_handles());
 
   void* data;
   size_t num_data_bytes, num_header_bytes, num_payload_bytes, num_ports_bytes;
   if (!ParsePortsMessage(message.get(), &data, &num_data_bytes,
                          &num_header_bytes, &num_payload_bytes,
                          &num_ports_bytes)) {
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   // Broadcast messages must not contain ports.
   if (num_ports_bytes > 0) {
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   base::AutoLock lock(peers_lock_);
   for (auto& iter : peers_) {
     // Copy and send the message to each known peer.
     Channel::MessagePtr peer_message(
         new Channel::Message(message->payload_size(), 0));
     memcpy(peer_message->mutable_payload(), message->payload(),
            message->payload_size());
     iter.second->PortsMessage(std::move(peer_message));
   }
 }
 
 #if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
 void NodeController::OnRelayPortsMessage(const ports::NodeName& from_node,
                                          base::ProcessHandle from_process,
                                          const ports::NodeName& destination,
                                          Channel::MessagePtr message) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   if (GetBrokerChannel()) {
     // Only the broker should be asked to relay a message.
     LOG(ERROR) << "Non-broker refusing to relay message.";
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   // The parent should always know which process this came from.
   DCHECK(from_process != base::kNullProcessHandle);
 
 #if defined(OS_WIN)
   // Rewrite the handles to this (the parent) process. If the message is
   // destined for another child process, the handles will be rewritten to that
   // process before going out (see NodeChannel::WriteChannelMessage).
@@ skipping 40 matching lines @@
     peer->PortsMessageFromRelay(from_node, std::move(message));
   else
     DLOG(ERROR) << "Dropping relay message for unknown node " << destination;
 }
 
 void NodeController::OnPortsMessageFromRelay(const ports::NodeName& from_node,
                                              const ports::NodeName& source_node,
                                              Channel::MessagePtr message) {
   if (GetPeerChannel(from_node) != GetBrokerChannel()) {
     LOG(ERROR) << "Refusing relayed message from non-broker node.";
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
   OnPortsMessage(source_node, std::move(message));
 }
 #endif
 
-void NodeController::OnChannelError(const ports::NodeName& from_node) {
+void NodeController::OnChannelError(const ports::NodeName& from_node,
+                                    NodeChannel* channel) {
   if (io_task_runner_->RunsTasksOnCurrentThread()) {
-    DropPeer(from_node);
+    DropPeer(from_node, channel);
     // DropPeer may have caused local port closures, so be sure to process any
     // pending local messages.
     AcceptIncomingMessages();
   } else {
     io_task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&NodeController::OnChannelError, base::Unretained(this),
-                   from_node));
+                   from_node, channel));
   }
 }
 
 #if defined(OS_MACOSX) && !defined(OS_IOS)
 MachPortRelay* NodeController::GetMachPortRelay() {
   {
     base::AutoLock lock(parent_lock_);
     // Return null if we're not the root.
     if (bootstrap_parent_channel_ || parent_name_ != ports::kInvalidNodeName)
       return nullptr;
@@ skipping 27 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo