Add mojo::edk::ClosePeerConnection.

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 209 matching lines @@
       reserved_ports_.erase(token);
   }
 
   for (const auto& port : ports_to_close)
     node_->ClosePort(port);
 
   // Ensure local port closure messages are processed.
   AcceptIncomingMessages();
 }
 
+void NodeController::ClosePeerConnection(const std::string& peer_token) {
+  io_task_runner_->PostTask(
+      FROM_HERE, base::Bind(&NodeController::ClosePeerConnectionOnIOThread,
+                            base::Unretained(this), peer_token));
+}
+
 void NodeController::ConnectToParent(ScopedPlatformHandle platform_handle) {
 #if !defined(OS_MACOSX) && !defined(OS_NACL_SFI)
   // Use the bootstrap channel for the broker and receive the node's channel
   // synchronously as the first message from the broker.
   base::ElapsedTimer timer;
   broker_.reset(new Broker(std::move(platform_handle)));
   platform_handle = broker_->GetParentPlatformHandle();
   UMA_HISTOGRAM_TIMES("Mojo.System.GetParentPlatformHandleSyncTime",
                       timer.Elapsed());
 
   if (!platform_handle.is_valid()) {
     // Most likely the browser side of the channel has already been closed and
     // the broker was unable to negotiate a NodeChannel pipe. In this case we
     // can cancel parent connection.
     DVLOG(1) << "Cannot connect to invalid parent channel.";
     CancelPendingPortMerges();
     return;
   }
 #endif
 
   io_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&NodeController::ConnectToParentOnIOThread,
                  base::Unretained(this),
                  base::Passed(&platform_handle)));
 }
 
 void NodeController::ConnectToPeer(ScopedPlatformHandle handle,
-                                   const ports::PortRef& port) {
+                                   const ports::PortRef& port,
+                                   const std::string& peer_token) {
   ports::NodeName node_name;
   GenerateRandomName(&node_name);
   io_task_runner_->PostTask(
       FROM_HERE, base::Bind(&NodeController::ConnectToPeerOnIOThread,
                             base::Unretained(this), base::Passed(&handle),
-                            node_name, port));
+                            node_name, port, peer_token));
 }
 
 void NodeController::SetPortObserver(
     const ports::PortRef& port,
     const scoped_refptr<PortObserver>& observer) {
   node_->SetUserData(port, observer);
 }
 
 void NodeController::ClosePort(const ports::PortRef& port) {
   SetPortObserver(port, nullptr);
@@ skipping 165 matching lines @@
     // Relying on message pipes to be closed is not enough because the parent
     // may see the message pipe closure before the child is dead, causing the
     // child process to be unexpectedly SIGKILL'd.
     bootstrap_parent_channel_->LeakHandleOnShutdown();
   }
   bootstrap_parent_channel_->Start();
 }
 
 void NodeController::ConnectToPeerOnIOThread(ScopedPlatformHandle handle,
                                              ports::NodeName token,
-                                             ports::PortRef port) {
+                                             ports::PortRef port,
+                                             const std::string& peer_token) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   scoped_refptr<NodeChannel> channel =
       NodeChannel::Create(this, std::move(handle), io_task_runner_, {});
-  pending_peers_.insert({token, {channel, port}});
+  peer_connections_.insert(
+      {token, PeerConnection{channel, port, peer_token}});
+  peers_by_token_.insert({peer_token, token});
 
   channel->SetRemoteNodeName(token);
   channel->Start();
 
   channel->AcceptPeer(name_, token, port.name());
 }
 
+void NodeController::ClosePeerConnectionOnIOThread(
+    const std::string& peer_token) {
+  RequestContext request_context(RequestContext::Source::SYSTEM);
+  auto peer = peers_by_token_.find(peer_token);
+  // The connection may already be closed.
+  if (peer == peers_by_token_.end())
+    return;
+
+  // |peer| may be removed so make a copy of |name|.
+  ports::NodeName name = peer->second;
+  DropPeer(name, nullptr);
+}
+
 scoped_refptr<NodeChannel> NodeController::GetPeerChannel(
     const ports::NodeName& name) {
   base::AutoLock lock(peers_lock_);
   auto it = peers_.find(name);
   if (it == peers_.end())
     return nullptr;
   return it->second;
 }
 
 scoped_refptr<NodeChannel> NodeController::GetParentChannel() {
@@ skipping 112 matching lines @@
     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)
     CancelPendingPortMerges();
 
+  auto peer = peer_connections_.find(name);
+  if (peer != peer_connections_.end()) {
+    peers_by_token_.erase(peer->second.peer_token);
+    ports_to_close.push_back(peer->second.local_port);
+    peer_connections_.erase(peer);
+  }
+
   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 130 matching lines @@
 
   {
     base::AutoLock lock(peers_lock_);
     for (const auto& peer : peers_)
       all_peers.push_back(peer.second);
     for (const auto& peer : pending_children_)
       all_peers.push_back(peer.second);
     peers_.clear();
     pending_children_.clear();
     pending_peer_messages_.clear();
-    pending_peers_.clear();
+    peer_connections_.clear();
   }
 
   for (const auto& peer : all_peers)
     peer->ShutDown();
 
   if (destroy_on_io_thread_shutdown_)
     delete this;
 }
 
 void NodeController::GenerateRandomPortName(ports::PortName* port_name) {
@@ skipping 494 matching lines @@
   OnPortsMessage(source_node, std::move(message));
 }
 #endif
 
 void NodeController::OnAcceptPeer(const ports::NodeName& from_node,
                                   const ports::NodeName& token,
                                   const ports::NodeName& peer_name,
                                   const ports::PortName& port_name) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
-  auto it = pending_peers_.find(from_node);
-  if (it == pending_peers_.end()) {
+  auto it = peer_connections_.find(from_node);
+  if (it == peer_connections_.end()) {
     DLOG(ERROR) << "Received unexpected AcceptPeer message from " << from_node;
     DropPeer(from_node, nullptr);
     return;
   }
 
-  scoped_refptr<NodeChannel> channel = it->second.first;
-  ports::PortRef local_port = it->second.second;
-  pending_peers_.erase(it);
+  scoped_refptr<NodeChannel> channel = std::move(it->second.channel);
+  ports::PortRef local_port = it->second.local_port;
+  std::string peer_token = std::move(it->second.peer_token);
+  peer_connections_.erase(it);
   DCHECK(channel);
 
   // If the peer connection is a self connection (which is used in tests),
   // drop the channel to it and skip straight to merging the ports.
-  if (name_ != peer_name) {
+  if (name_ == peer_name) {
+    peers_by_token_.erase(peer_token);
+  } else {
+    peers_by_token_[peer_token] = peer_name;
+    peer_connections_.insert(
+        {peer_name, PeerConnection{nullptr, local_port, peer_token}});
     DVLOG(1) << "Node " << name_ << " accepted peer " << peer_name;
 
     AddPeer(peer_name, channel, false /* start_channel */);
   }
 
   // We need to choose one side to initiate the port merge. It doesn't matter
   // who does it as long as they don't both try. Simple solution: pick the one
   // with the "smaller" port name.
   if (local_port.name() < port_name) {
     node()->MergePorts(local_port, peer_name, port_name);
@@ skipping 66 matching lines @@
     callback = shutdown_callback_;
     shutdown_callback_.Reset();
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
+NodeController::PeerConnection::PeerConnection() = default;
+
+NodeController::PeerConnection::PeerConnection(
+    const PeerConnection& other) = default;
+
+NodeController::PeerConnection::PeerConnection(
+    PeerConnection&& other) = default;
+
+NodeController::PeerConnection::PeerConnection(
+    const scoped_refptr<NodeChannel>& channel,
+    const ports::PortRef& local_port,
+    const std::string& peer_token)
+    : channel(channel), local_port(local_port), peer_token(peer_token) {}
+
+NodeController::PeerConnection::~PeerConnection() = default;
+
+NodeController::PeerConnection& NodeController::PeerConnection::
+operator=(const PeerConnection& other) = default;
+
+NodeController::PeerConnection& NodeController::PeerConnection::
+operator=(PeerConnection&& other) = default;
+
 }  // namespace edk
 }  // namespace mojo