[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 processes.

[Ken Rockot(use gerrit already), 2016/06/01 15:40:48]

nit: processed*

[Anand Mistry (off Chromium), 2016/06/02 04:35:36]

Done.

+  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);
+      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;
 }

[Ken Rockot(use gerrit already), 2016/06/01 15:40:48]

Huh. It probably doesn't make much difference in practice, but I wonder if
DropAllPeers should also AcceptIncomingMessages.

[Anand Mistry (off Chromium), 2016/06/02 04:35:36]

There'a problem with running AcceptIncomingMessages() here.
AcceptIncomingMessages can trigger watcher wake-ups which will call back into
user code. I don't think we want to trigger watchers since this is on message
loop shutdown.

But given that we don't close any other ports, this is probably entirely
unnecessary. This might even be completely wrong and should be removed. WDYT?

 
 void NodeController::GenerateRandomPortName(ports::PortName* port_name) {
   GenerateRandomName(port_name);
 }
 
 void NodeController::AllocMessage(size_t num_header_bytes,
                                   ports::ScopedMessage* message) {
   message->reset(new PortsMessage(num_header_bytes, 0, 0, nullptr));
 }
 
@@ skipping 285 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