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

mojo/edk/system/node_controller.cc

Index: mojo/edk/system/node_controller.cc
diff --git a/mojo/edk/system/node_controller.cc b/mojo/edk/system/node_controller.cc
index a7f247a094387f86039b35db49690c0b9828381f..6d261888e0c6a8e70f45c21510b0231c10b99720 100644
--- a/mojo/edk/system/node_controller.cc
+++ b/mojo/edk/system/node_controller.cc
@@ -310,6 +310,7 @@ void NodeController::MergePortIntoParent(const std::string& token,
   }
 
   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_|
@@ -317,11 +318,21 @@ void NodeController::MergePortIntoParent(const std::string& token,
     // |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);
 }
 
@@ -495,7 +506,8 @@ void NodeController::AddPeer(const ports::NodeName& name,
   }
 }
 
-void NodeController::DropPeer(const ports::NodeName& name) {
+void NodeController::DropPeer(const ports::NodeName& name,
+                              NodeChannel* channel) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   {
@@ -542,6 +554,23 @@ void NodeController::DropPeer(const ports::NodeName& name) {
     }
   }
 
+  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);
 
@@ -778,7 +807,7 @@ void NodeController::OnAcceptChild(const ports::NodeName& from_node,
 
   if (!parent) {
     DLOG(ERROR) << "Unexpected AcceptChild message from " << from_node;
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
@@ -801,7 +830,7 @@ void NodeController::OnAcceptParent(const ports::NodeName& 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;
   }
 
@@ -858,7 +887,7 @@ void NodeController::OnAddBrokerClient(const ports::NodeName& from_node,
 
   if (GetPeerChannel(client_name)) {
     DLOG(ERROR) << "Ignoring AddBrokerClient for known client.";
-    DropPeer(from_node);
+    DropPeer(from_node, nullptr);
     return;
   }
 
@@ -989,7 +1018,7 @@ void NodeController::OnPortsMessage(const ports::NodeName& from_node,
   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;
   }
 
@@ -1040,7 +1069,7 @@ void NodeController::OnRequestIntroduction(const ports::NodeName& 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;
   }
 
@@ -1086,13 +1115,13 @@ void NodeController::OnBroadcast(const ports::NodeName& from_node,
   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;
   }
 
@@ -1117,7 +1146,7 @@ void NodeController::OnRelayPortsMessage(const ports::NodeName& from_node,
   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;
   }
 
@@ -1178,7 +1207,7 @@ void NodeController::OnPortsMessageFromRelay(const ports::NodeName& from_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;
   }
 
@@ -1186,9 +1215,10 @@ void NodeController::OnPortsMessageFromRelay(const ports::NodeName& from_node,
 }
 #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();
@@ -1196,7 +1226,7 @@ void NodeController::OnChannelError(const ports::NodeName& from_node) {
     io_task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&NodeController::OnChannelError, base::Unretained(this),
-                   from_node));
+                   from_node, channel));
   }
 }