[mojo-edk] Bind a child token to child launches and port reservations.

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 113 matching lines @@
 
 void NodeController::SetIOTaskRunner(
     scoped_refptr<base::TaskRunner> task_runner) {
   io_task_runner_ = task_runner;
   ThreadDestructionObserver::Create(
       io_task_runner_,
       base::Bind(&NodeController::DropAllPeers, base::Unretained(this)));
 }
 
 void NodeController::ConnectToChild(base::ProcessHandle process_handle,
-                                    ScopedPlatformHandle platform_handle) {
+                                    ScopedPlatformHandle platform_handle,
+                                    const std::string& child_token) {
+  // Generate the temporary remote node name here so that it can be associated
+  // with the embedder's child_token. If an error occurs in the child process
+  // after it is launched, but before any reserved ports are connected, this can
+  // be used to clean up any dangling ports.
+  ports::NodeName node_name;
+  GenerateRandomName(&node_name);
+
+  {
+    base::AutoLock lock(reserved_ports_lock_);
+    bool inserted = pending_child_tokens_.insert(
+        std::make_pair(node_name, child_token)).second;
+    DCHECK(inserted);
+  }
+
   io_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&NodeController::ConnectToChildOnIOThread,
                  base::Unretained(this),
                  process_handle,
-                 base::Passed(&platform_handle)));
+                 base::Passed(&platform_handle),
+                 node_name));
+}
+
+void NodeController::CloseChildPorts(const std::string& child_token) {
+  std::vector<ports::PortRef> ports_to_close;
+  {
+    std::vector<std::string> port_tokens;
+    base::AutoLock lock(reserved_ports_lock_);
+    for (const auto& port : reserved_ports_) {
+      if (port.second.second == child_token) {
+        DVLOG(1) << "Closing reserved port " << port.second.first.name();
+        ports_to_close.push_back(port.second.first);
+        port_tokens.push_back(port.first);
+      }
+    }
+
+    for (const auto& token : port_tokens)
+      reserved_ports_.erase(token);
+  }
+
+  for (const auto& port : ports_to_close)
+    node_->ClosePort(port);
+
+  // Ensure local port closure messages are processed.
+  AcceptIncomingMessages();
 }
 
 void NodeController::ConnectToParent(ScopedPlatformHandle platform_handle) {
 // TODO(amistry): Consider the need for a broker on Windows.
 #if defined(OS_POSIX) && !defined(OS_MACOSX)
   // On posix, 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();
@@ skipping 25 matching lines @@
 int NodeController::SendMessage(const ports::PortRef& port,
                                 std::unique_ptr<PortsMessage> message) {
   ports::ScopedMessage ports_message(message.release());
   int rv = node_->SendMessage(port, std::move(ports_message));
 
   AcceptIncomingMessages();
   return rv;
 }
 
 void NodeController::ReservePort(const std::string& token,
-                                 const ports::PortRef& port) {
+                                 const ports::PortRef& port,
+                                 const std::string& child_token) {
   DVLOG(2) << "Reserving port " << port.name() << "@" << name_ << " for token "
            << token;
 
   base::AutoLock lock(reserved_ports_lock_);
-  auto result = reserved_ports_.insert(std::make_pair(token, port));
+  auto result = reserved_ports_.insert(
+      std::make_pair(token, std::make_pair(port, child_token)));
   DCHECK(result.second);
 }
 
 void NodeController::MergePortIntoParent(const std::string& token,
                                          const ports::PortRef& port) {
   bool was_merged = false;
   {
     // This request may be coming from within the process that reserved the
     // "parent" side (e.g. for Chrome single-process mode), so if this token is
     // reserved locally, merge locally instead.
     base::AutoLock lock(reserved_ports_lock_);
     auto it = reserved_ports_.find(token);
     if (it != reserved_ports_.end()) {
-      node_->MergePorts(port, name_, it->second.name());
+      std::string child_token = std::move(it->second.second);

[ncarter (slow), 2016/06/02 17:43:28]

Why is this line needed?

[Anand Mistry (off Chromium), 2016/06/03 04:39:40]

Oops. Relic from a different day.

+      node_->MergePorts(port, name_, it->second.first.name());
       reserved_ports_.erase(it);
       was_merged = true;
     }
   }
   if (was_merged) {
     AcceptIncomingMessages();
     return;
   }
 
   scoped_refptr<NodeChannel> parent;
@@ skipping 38 matching lines @@
     base::AutoLock lock(shutdown_lock_);
     shutdown_callback_ = callback;
     shutdown_callback_flag_.Set(true);
   }
 
   AttemptShutdownIfRequested();
 }
 
 void NodeController::ConnectToChildOnIOThread(
     base::ProcessHandle process_handle,
-    ScopedPlatformHandle platform_handle) {
+    ScopedPlatformHandle platform_handle,
+    ports::NodeName token) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
 #if defined(OS_POSIX) && !defined(OS_MACOSX)
   PlatformChannelPair node_channel;
   // BrokerHost owns itself.
   BrokerHost* broker_host = new BrokerHost(std::move(platform_handle));
   broker_host->SendChannel(node_channel.PassClientHandle());
   scoped_refptr<NodeChannel> channel = NodeChannel::Create(
       this, node_channel.PassServerHandle(), io_task_runner_);
 #else
   scoped_refptr<NodeChannel> channel =
       NodeChannel::Create(this, std::move(platform_handle), io_task_runner_);
 #endif
 
   // We set up the child channel with a temporary name so it can be identified
   // as a pending child if it writes any messages to the channel. We may start
   // receiving messages from it (though we shouldn't) as soon as Start() is
   // called below.
-  ports::NodeName token;
-  GenerateRandomName(&token);
 
   pending_children_.insert(std::make_pair(token, channel));
   RecordPendingChildCount(pending_children_.size());
 
   channel->SetRemoteNodeName(token);
   channel->SetRemoteProcessHandle(process_handle);
   channel->Start();
 
   channel->AcceptChild(name_, token);
 }
@@ skipping 100 matching lines @@
       DVLOG(1) << "Dropped peer " << peer;
     }
 
     pending_peer_messages_.erase(name);
     pending_children_.erase(name);
 
     RecordPeerCount(peers_.size());
     RecordPendingChildCount(pending_children_.size());
   }
 
+  std::vector<ports::PortRef> ports_to_close;
+  {
+    // Clean up any reserved ports.
+    base::AutoLock lock(reserved_ports_lock_);
+    auto it = pending_child_tokens_.find(name);
+    if (it != pending_child_tokens_.end()) {
+      const std::string& child_token = it->second;
+
+      std::vector<std::string> port_tokens;
+      for (const auto& port : reserved_ports_) {
+        if (port.second.second == child_token) {
+          DVLOG(1) << "Closing reserved port: " << port.second.first.name();
+          ports_to_close.push_back(port.second.first);
+          port_tokens.push_back(port.first);
+        }
+      }
+
+      // 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);
+    }
+  }
+
+  for (const auto& port : ports_to_close)
+    node_->ClosePort(port);
+
   node_->LostConnectionToNode(name);
 }
 
 void NodeController::SendPeerMessage(const ports::NodeName& name,
                                      ports::ScopedMessage message) {
   Channel::MessagePtr channel_message =
       static_cast<PortsMessage*>(message.get())->TakeChannelMessage();
 
   scoped_refptr<NodeChannel> peer = GetPeerChannel(name);
 #if defined(OS_WIN)
@@ skipping 110 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();
   }
 
+  std::vector<ports::PortRef> all_reserved_ports;
+  {
+    base::AutoLock lock(reserved_ports_lock_);
+    pending_child_tokens_.clear();
+    for (const auto& port : reserved_ports_)
+      all_reserved_ports.push_back(port.second.first);
+    reserved_ports_.clear();
+  }
+
+  for (const auto& port : all_reserved_ports)
+    node_->ClosePort(port);
+
   for (const auto& peer : all_peers)
     peer->ShutDown();
 
   if (destroy_on_io_thread_shutdown_)
     delete this;
 }
 
 void NodeController::GenerateRandomPortName(ports::PortName* port_name) {
   GenerateRandomName(port_name);
 }
@@ skipping 291 matching lines @@
 
   ports::PortRef local_port;
   {
     base::AutoLock lock(reserved_ports_lock_);
     auto it = reserved_ports_.find(token);
     if (it == reserved_ports_.end()) {
       DVLOG(1) << "Ignoring request to connect to port for unknown token "
                << token;
       return;
     }
-    local_port = it->second;
+    local_port = it->second.first;
   }
 
   int rv = node_->MergePorts(local_port, from_node, connector_port_name);
   if (rv != ports::OK)
     DLOG(ERROR) << "MergePorts failed: " << rv;
 }
 
 void NodeController::OnRequestIntroduction(const ports::NodeName& from_node,
                                            const ports::NodeName& name) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
@@ skipping 101 matching lines @@
   if (peer)
     peer->PortsMessage(std::move(message));
   else
     DLOG(ERROR) << "Dropping relay message for unknown node " << destination;
 }
 #endif
 
 void NodeController::OnChannelError(const ports::NodeName& from_node) {
   if (io_task_runner_->RunsTasksOnCurrentThread()) {
     DropPeer(from_node);
+    // 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::DropPeer, base::Unretained(this),
+        base::Bind(&NodeController::OnChannelError, base::Unretained(this),
                    from_node));
   }
 }
 
 #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)
@@ skipping 28 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo