Add device port unmapping support to forwarder2.

There used not to be any way to unmap a previously forwarded port. Since
the host and device forwarder processes are kept alive during a whole test
suite, both the host and device ended up with an unnecessarily high amount of
unused open sockets. That is source of flakiness (e.g. too many opened file
descriptors).

This adds an UnmapDevicePort() method to the Python Forwarder wrapper class and
modifies the native forwarder2 to accept a negative port on the command line.
When a negative port is provided, the host forwarder tries to unmap that port
(that was mapped previously).

BUG=229685,239014
R=bulach@chromium.org, digit@chromium.org

Review URL: https://codereview.chromium.org/15008004

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@200257 0039d316-1c4b-4281-b951-d872f2087c98

10 files changed, 160 insertions, 75 deletions

diff --git a/tools/android/forwarder2/daemon.cc b/tools/android/forwarder2/daemon.cc
index 0936ef4..c17cbc7 100644
--- a/tools/android/forwarder2/daemon.cc
+++ b/tools/android/forwarder2/daemon.cc
@@ -56,7 +56,7 @@ bool RunServerAcceptLoop(const std::string& welcome_message,
   for (;;) {
     scoped_ptr<Socket> client_socket(new Socket());
     if (!server_socket->Accept(client_socket.get())) {
-      if (server_socket->exited())
+      if (server_socket->DidReceiveEvent())
         break;
       PError("Accept()");
       failed = true;
@@ -207,7 +207,7 @@ bool Daemon::SpawnIfNeeded() {
           exit(1);
         }
         server_delegate_->Init();
-        command_socket.set_exit_notifier_fd(get_exit_fd_callback_());
+        command_socket.AddEventFd(get_exit_fd_callback_());
         exit(!RunServerAcceptLoop(identifier_, &command_socket,
                                   server_delegate_));
       }
diff --git a/tools/android/forwarder2/device_controller.cc b/tools/android/forwarder2/device_controller.cc
index 212b0db..7338d5a 100644
--- a/tools/android/forwarder2/device_controller.cc
+++ b/tools/android/forwarder2/device_controller.cc
@@ -18,7 +18,7 @@ namespace forwarder2 {
 
 DeviceController::DeviceController(int exit_notifier_fd)
     : exit_notifier_fd_(exit_notifier_fd) {
-  kickstart_adb_socket_.set_exit_notifier_fd(exit_notifier_fd);
+  kickstart_adb_socket_.AddEventFd(exit_notifier_fd);
 }
 
 DeviceController::~DeviceController() {
@@ -60,7 +60,7 @@ void DeviceController::Start() {
     CleanUpDeadListeners();
     scoped_ptr<Socket> socket(new Socket);
     if (!kickstart_adb_socket_.Accept(socket.get())) {
-      if (!kickstart_adb_socket_.exited()) {
+      if (!kickstart_adb_socket_.DidReceiveEvent()) {
         LOG(ERROR) << "Could not Accept DeviceController socket: "
                    << safe_strerror(errno);
       } else {
@@ -69,7 +69,7 @@ void DeviceController::Start() {
       break;
     }
     // So that |socket| doesn't block on read if it has notifications.
-    socket->set_exit_notifier_fd(exit_notifier_fd_);
+    socket->AddEventFd(exit_notifier_fd_);
     int port;
     command::Type command;
     if (!ReadCommand(socket.get(), &port, &command)) {
diff --git a/tools/android/forwarder2/device_forwarder_main.cc b/tools/android/forwarder2/device_forwarder_main.cc
index c6fae82..5d674cf 100644
--- a/tools/android/forwarder2/device_forwarder_main.cc
+++ b/tools/android/forwarder2/device_forwarder_main.cc
@@ -62,7 +62,7 @@ class ServerDelegate : public Daemon::ServerDelegate {
     char buf[kBufSize];
     const int bytes_read = client_socket->Read(buf, sizeof(buf));
     if (bytes_read <= 0) {
-      if (client_socket->exited())
+      if (client_socket->DidReceiveEvent())
         return;
       PError("Read()");
       return;
diff --git a/tools/android/forwarder2/device_listener.cc b/tools/android/forwarder2/device_listener.cc
index d67151d..4b6265d 100644
--- a/tools/android/forwarder2/device_listener.cc
+++ b/tools/android/forwarder2/device_listener.cc
@@ -30,8 +30,8 @@ DeviceListener::DeviceListener(scoped_ptr<Socket> adb_control_socket, int port)
       is_alive_(false),
       must_exit_(false) {
   CHECK(adb_control_socket_.get());
-  adb_control_socket_->set_exit_notifier_fd(exit_notifier_.receiver_fd());
-  listener_socket_.set_exit_notifier_fd(exit_notifier_.receiver_fd());
+  adb_control_socket_->AddEventFd(exit_notifier_.receiver_fd());
+  listener_socket_.AddEventFd(exit_notifier_.receiver_fd());
   pthread_mutex_init(&adb_data_socket_mutex_, NULL);
   pthread_cond_init(&adb_data_socket_cond_, NULL);
 }
@@ -106,7 +106,7 @@ void DeviceListener::RunInternal() {
   while (!must_exit_) {
     scoped_ptr<Socket> device_data_socket(new Socket);
     if (!listener_socket_.Accept(device_data_socket.get())) {
-      if (listener_socket_.exited()) {
+      if (listener_socket_.DidReceiveEvent()) {
         LOG(INFO) << "Received exit notification, stopped accepting clients.";
         break;
       }
diff --git a/tools/android/forwarder2/forwarder.cc b/tools/android/forwarder2/forwarder.cc
index f1aef15..d82bb0c 100644
--- a/tools/android/forwarder2/forwarder.cc
+++ b/tools/android/forwarder2/forwarder.cc
@@ -94,16 +94,9 @@ Forwarder::Forwarder(scoped_ptr<Socket> socket1, scoped_ptr<Socket> socket2)
       socket2_(socket2.Pass()) {
   DCHECK(socket1_.get());
   DCHECK(socket2_.get());
-  // The forwarder thread doesn't need to listen to notifications. It can be
-  // keept alive until either the conenction is broken or the program exit.
-  socket1_->reset_exit_notifier_fd();
-  socket2_->reset_exit_notifier_fd();
 }
 
-Forwarder::~Forwarder() {
-  socket1_->Close();
-  socket2_->Close();
-}
+Forwarder::~Forwarder() {}
 
 void Forwarder::Run() {
   const int nfds = Socket::GetHighestFileDescriptor(*socket1_, *socket2_) + 1;
diff --git a/tools/android/forwarder2/host_controller.cc b/tools/android/forwarder2/host_controller.cc
index b95d986..02662d2 100644
--- a/tools/android/forwarder2/host_controller.cc
+++ b/tools/android/forwarder2/host_controller.cc
@@ -23,20 +23,26 @@ HostController::HostController(int device_port,
       forward_to_host_(forward_to_host),
       forward_to_host_port_(forward_to_host_port),
       adb_port_(adb_port),
-      exit_notifier_fd_(exit_notifier_fd),
+      global_exit_notifier_fd_(exit_notifier_fd),
       ready_(false) {
-  adb_control_socket_.set_exit_notifier_fd(exit_notifier_fd);
+  adb_control_socket_.AddEventFd(global_exit_notifier_fd_);
+  adb_control_socket_.AddEventFd(delete_controller_notifier_.receiver_fd());
 }
 
 HostController::~HostController() {
+  delete_controller_notifier_.Notify();
+  Join();
+  // Note that the Forwarder instance (that also received a delete notification)
+  // might still be running on its own thread at this point. This is not a
+  // problem since it will self-delete once the socket that it is operating on
+  // is closed.
 }
 
 void HostController::StartForwarder(
     scoped_ptr<Socket> host_server_data_socket) {
-  scoped_ptr<Socket> adb_data_socket(new Socket);
+  scoped_ptr<Socket> adb_data_socket(CreateSocket());
   if (!adb_data_socket->ConnectTcp("", adb_port_)) {
-    LOG(ERROR) << "Could not connect AdbDataSocket on port: "
-               << adb_port_;
+    LOG(ERROR) << "Could not connect AdbDataSocket on port: " << adb_port_;
     return;
   }
   // Open the Adb data connection, and send a command with the
@@ -61,6 +67,13 @@ void HostController::StartForwarder(
   forwarder->Start();
 }
 
+scoped_ptr<Socket> HostController::CreateSocket() {
+  scoped_ptr<Socket> socket(new Socket());
+  socket->AddEventFd(global_exit_notifier_fd_);
+  socket->AddEventFd(delete_controller_notifier_.receiver_fd());
+  return socket.Pass();
+}
+
 bool HostController::Connect() {
   if (!adb_control_socket_.ConnectTcp("", adb_port_)) {
     LOG(ERROR) << "Could not connect HostController socket on port: "
@@ -89,8 +102,15 @@ bool HostController::Connect() {
 void HostController::Run() {
   CHECK(ready_) << "HostController not ready. Must call Connect() first.";
   while (true) {
-    if (!ReceivedCommand(command::ACCEPT_SUCCESS,
-                         &adb_control_socket_)) {
+    if (!ReceivedCommand(command::ACCEPT_SUCCESS, &adb_control_socket_)) {
+      if (adb_control_socket_.DidReceiveEventOnFd(
+              delete_controller_notifier_.receiver_fd())) {
+        // The instance is being deleted. The control socket will be closed and
+        // the device controller will see that as an error (that it will recover
+        // from properly). TODO(pliard): tell the device controller that this is
+        // a graceful (i.e. user-intended) shutdown rather than an error.
+        break;
+      }
       // TODO(pliard): This can also happen if device_forwarder was
       // intentionally killed before host_forwarder. In that case,
       // device_forwarder should send a notification to the host.  Currently the
@@ -101,8 +121,7 @@ void HostController::Run() {
       break;
     }
     // Try to connect to host server.
-    scoped_ptr<Socket> host_server_data_socket(new Socket);
-    host_server_data_socket->set_exit_notifier_fd(exit_notifier_fd_);
+    scoped_ptr<Socket> host_server_data_socket(CreateSocket());
     if (!host_server_data_socket->ConnectTcp(
             forward_to_host_, forward_to_host_port_)) {
       LOG(ERROR) << "Could not Connect HostServerData socket on port: "
diff --git a/tools/android/forwarder2/host_controller.h b/tools/android/forwarder2/host_controller.h
index b4409eb..aa7c0e2 100644
--- a/tools/android/forwarder2/host_controller.h
+++ b/tools/android/forwarder2/host_controller.h
@@ -10,6 +10,7 @@
 #include "base/basictypes.h"
 #include "base/compiler_specific.h"
 #include "base/memory/scoped_ptr.h"
+#include "tools/android/forwarder2/pipe_notifier.h"
 #include "tools/android/forwarder2/socket.h"
 #include "tools/android/forwarder2/thread.h"
 
@@ -43,14 +44,21 @@ class HostController : public Thread {
  private:
   void StartForwarder(scoped_ptr<Socket> host_server_data_socket_ptr);
 
+  // Helper method that creates a socket and adds the appropriate event file
+  // descriptors.
+  scoped_ptr<Socket> CreateSocket();
+
   Socket adb_control_socket_;
   int device_port_;
   const std::string forward_to_host_;
   const int forward_to_host_port_;
   const int adb_port_;
 
-  // Used to notify the controller to exit.
-  const int exit_notifier_fd_;
+  // Used to notify the controller when the process is killed.
+  const int global_exit_notifier_fd_;
+  // Used to cancel the pending blocking IO operations when the host controller
+  // instance is deleted.
+  PipeNotifier delete_controller_notifier_;
 
   bool ready_;
 
diff --git a/tools/android/forwarder2/host_forwarder_main.cc b/tools/android/forwarder2/host_forwarder_main.cc
index f145832..31b7629 100644
--- a/tools/android/forwarder2/host_forwarder_main.cc
+++ b/tools/android/forwarder2/host_forwarder_main.cc
@@ -9,13 +9,16 @@
 #include <cstdio>
 #include <cstring>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "base/command_line.h"
 #include "base/compiler_specific.h"
 #include "base/file_util.h"
 #include "base/files/file_path.h"
+#include "base/hash_tables.h"
 #include "base/logging.h"
+#include "base/memory/linked_ptr.h"
 #include "base/memory/scoped_vector.h"
 #include "base/posix/eintr_wrapper.h"
 #include "base/safe_strerror_posix.h"
@@ -122,7 +125,7 @@ class ServerDelegate : public Daemon::ServerDelegate {
     char buf[kBufSize];
     const int bytes_read = client_socket->Read(buf, sizeof(buf));
     if (bytes_read <= 0) {
-      if (client_socket->exited())
+      if (client_socket->DidReceiveEvent())
         return;
       PError("Read()");
       has_failed_ = true;
@@ -137,35 +140,53 @@ class ServerDelegate : public Daemon::ServerDelegate {
         command, &adb_port, &device_port, &forward_to_host, &forward_to_port);
     if (!succeeded) {
       has_failed_ = true;
-      client_socket->WriteString(
-          base::StringPrintf("ERROR: Could not parse forward command '%s'",
-                             command.c_str()));
+      const std::string msg = base::StringPrintf(
+          "ERROR: Could not parse forward command '%s'", command.c_str());
+      SendMessage(msg, client_socket.get());
       return;
     }
+    if (device_port < 0) {
+      // Remove the previously created host controller.
+      const std::string controller_key = MakeHostControllerMapKey(
+          adb_port, -device_port);
+      const HostControllerMap::size_type removed_elements = controllers_.erase(
+          controller_key);
+      SendMessage(
+          !removed_elements ? "ERROR: could not unmap port" : "OK",
+          client_socket.get());
+      return;
+    }
+    // Create a new host controller.
     scoped_ptr<HostController> host_controller(
         new HostController(device_port, forward_to_host, forward_to_port,
                            adb_port, GetExitNotifierFD()));
     if (!host_controller->Connect()) {
       has_failed_ = true;
-      client_socket->WriteString("ERROR: Connection to device failed.");
+      SendMessage("ERROR: Connection to device failed.", client_socket.get());
       return;
     }
     // Get the current allocated port.
     device_port = host_controller->device_port();
     LOG(INFO) << "Forwarding device port " << device_port << " to host "
               << forward_to_host << ":" << forward_to_port;
-    if (!client_socket->WriteString(
-        base::StringPrintf("%d:%d", device_port, forward_to_port))) {
-      has_failed_ = true;
+    const std::string msg = base::StringPrintf(
+        "%d:%d", device_port, forward_to_port);
+    if (!SendMessage(msg, client_socket.get()))
       return;
-    }
     host_controller->Start();
-    controllers_.push_back(host_controller.release());
+    const std::string controller_key = MakeHostControllerMapKey(
+        adb_port, device_port);
+    controllers_.insert(
+        std::make_pair(controller_key,
+                       linked_ptr<HostController>(host_controller.release())));
   }
 
   virtual void OnServerExited() OVERRIDE {
-    for (int i = 0; i < controllers_.size(); ++i)
-      controllers_[i]->Join();
+    for (HostControllerMap::iterator it = controllers_.begin();
+         it != controllers_.end(); ++it) {
+      linked_ptr<HostController> host_controller = it->second;
+      host_controller->Join();
+    }
     if (controllers_.size() == 0) {
       LOG(ERROR) << "No forwarder servers could be started. Exiting.";
       has_failed_ = true;
@@ -173,7 +194,22 @@ class ServerDelegate : public Daemon::ServerDelegate {
   }
 
  private:
-  ScopedVector<HostController> controllers_;
+  typedef base::hash_map<
+      std::string, linked_ptr<HostController> > HostControllerMap;
+
+  static std::string MakeHostControllerMapKey(int adb_port, int device_port) {
+    return base::StringPrintf("%d:%d", adb_port, device_port);
+  }
+
+  bool SendMessage(const std::string& msg, Socket* client_socket) {
+    bool result = client_socket->WriteString(msg);
+    DCHECK(result);
+    if (!result)
+      has_failed_ = true;
+    return result;
+  }
+
+  HostControllerMap controllers_;
   bool has_failed_;
 
   DISALLOW_COPY_AND_ASSIGN(ServerDelegate);
@@ -213,8 +249,10 @@ class ClientDelegate : public Daemon::ClientDelegate {
 };
 
 void PrintUsage(const char* program_name) {
-  LOG(ERROR) << program_name << " adb_port:from_port:to_port:to_host\n"
-      "<adb port> is the TCP port Adb is configured to forward to.";
+  LOG(ERROR) << program_name
+             << " adb_port:from_port:to_port:to_host\n"
+                "<adb port> is the TCP port Adb is configured to forward to.\n"
+                "Note that <from_port> can be unmapped by making it negative.";
 }
 
 int RunHostForwarder(int argc, char** argv) {
diff --git a/tools/android/forwarder2/socket.cc b/tools/android/forwarder2/socket.cc
index 12729b4..242af40 100644
--- a/tools/android/forwarder2/socket.cc
+++ b/tools/android/forwarder2/socket.cc
@@ -73,9 +73,7 @@ Socket::Socket()
       socket_error_(false),
       family_(AF_INET),
       addr_ptr_(reinterpret_cast<sockaddr*>(&addr_.addr4)),
-      addr_len_(sizeof(sockaddr)),
-      exit_notifier_fd_(-1),
-      exited_(false) {
+      addr_len_(sizeof(sockaddr)) {
   memset(&addr_, 0, sizeof(addr_));
 }
 
@@ -330,6 +328,27 @@ int Socket::WriteString(const std::string& buffer) {
   return WriteNumBytes(buffer.c_str(), buffer.size());
 }
 
+void Socket::AddEventFd(int event_fd) {
+  Event event;
+  event.fd = event_fd;
+  event.was_fired = false;
+  events_.push_back(event);
+}
+
+bool Socket::DidReceiveEventOnFd(int fd) const {
+  for (size_t i = 0; i < events_.size(); ++i)
+    if (events_[i].fd == fd)
+      return events_[i].was_fired;
+  return false;
+}
+
+bool Socket::DidReceiveEvent() const {
+  for (size_t i = 0; i < events_.size(); ++i)
+    if (events_[i].was_fired)
+      return true;
+  return false;
+}
+
 int Socket::WriteNumBytes(const void* buffer, size_t num_bytes) {
   int bytes_written = 0;
   int ret = 1;
@@ -343,9 +362,8 @@ int Socket::WriteNumBytes(const void* buffer, size_t num_bytes) {
 }
 
 bool Socket::WaitForEvent(EventType type, int timeout_secs) {
-  if (exit_notifier_fd_ == -1 || socket_ == -1)
+  if (events_.empty() || socket_ == -1)
     return true;
-  const int nfds = std::max(socket_, exit_notifier_fd_) + 1;
   fd_set read_fds;
   fd_set write_fds;
   FD_ZERO(&read_fds);
@@ -354,8 +372,8 @@ bool Socket::WaitForEvent(EventType type, int timeout_secs) {
     FD_SET(socket_, &read_fds);
   else
     FD_SET(socket_, &write_fds);
-  FD_SET(exit_notifier_fd_, &read_fds);
-
+  for (size_t i = 0; i < events_.size(); ++i)
+    FD_SET(events_[i].fd, &read_fds);
   timeval tv = {};
   timeval* tv_ptr = NULL;
   if (timeout_secs > 0) {
@@ -363,13 +381,22 @@ bool Socket::WaitForEvent(EventType type, int timeout_secs) {
     tv.tv_usec = 0;
     tv_ptr = &tv;
   }
-  if (HANDLE_EINTR(select(nfds, &read_fds, &write_fds, NULL, tv_ptr)) <= 0)
-    return false;
-  if (FD_ISSET(exit_notifier_fd_, &read_fds)) {
-    exited_ = true;
+  int max_fd = socket_;
+  for (size_t i = 0; i < events_.size(); ++i)
+    if (events_[i].fd > max_fd)
+      max_fd = events_[i].fd;
+  if (HANDLE_EINTR(
+          select(max_fd + 1, &read_fds, &write_fds, NULL, tv_ptr)) <= 0) {
     return false;
   }
-  return true;
+  bool event_was_fired = false;
+  for (size_t i = 0; i < events_.size(); ++i) {
+    if (FD_ISSET(events_[i].fd, &read_fds)) {
+      events_[i].was_fired = true;
+      event_was_fired = true;
+    }
+  }
+  return !event_was_fired;
 }
 
 // static
diff --git a/tools/android/forwarder2/socket.h b/tools/android/forwarder2/socket.h
index 9057f74..b86fd9e 100644
--- a/tools/android/forwarder2/socket.h
+++ b/tools/android/forwarder2/socket.h
@@ -11,6 +11,7 @@
 #include <sys/un.h>
 
 #include <string>
+#include <vector>
 
 #include "base/basictypes.h"
 
@@ -69,25 +70,27 @@ class Socket {
 
   bool has_error() const { return socket_error_; }
 
-  // |exit_notifier_fd| must be a valid pipe file descriptor created from the
+  // |event_fd| must be a valid pipe file descriptor created from the
   // PipeNotifier and must live (not be closed) at least as long as this socket
   // is alive.
-  void set_exit_notifier_fd(int exit_notifier_fd) {
-    exit_notifier_fd_ = exit_notifier_fd;
-  }
-  // Unset the |exit_notifier_fd_| so that it will not receive notifications
-  // anymore.
-  void reset_exit_notifier_fd() { exit_notifier_fd_ = -1; }
+  void AddEventFd(int event_fd);
 
   // Returns whether Accept() or Connect() was interrupted because the socket
-  // received an exit notification.
-  bool exited() const { return exited_; }
+  // received an external event fired through the provided fd.
+  bool DidReceiveEventOnFd(int fd) const;
+
+  bool DidReceiveEvent() const;
 
   static int GetHighestFileDescriptor(const Socket& s1, const Socket& s2);
 
   static pid_t GetUnixDomainSocketProcessOwner(const std::string& path);
 
  private:
+  enum EventType {
+    READ,
+    WRITE
+  };
+
   union SockAddr {
     // IPv4 sockaddr
     sockaddr_in addr4;
@@ -97,6 +100,11 @@ class Socket {
     sockaddr_un addr_un;
   };
 
+  struct Event {
+    int fd;
+    bool was_fired;
+  };
+
   // If |host| is empty, use localhost.
   bool InitTcpSocket(const std::string& host, int port);
   bool InitUnixSocket(const std::string& path);
@@ -107,11 +115,6 @@ class Socket {
   bool InitSocketInternal();
   void SetSocketError();
 
-  enum EventType {
-    READ,
-    WRITE
-  };
-
   // Waits until either the Socket or the |exit_notifier_fd_| has received an
   // event.
   bool WaitForEvent(EventType type, int timeout_secs);
@@ -130,12 +133,9 @@ class Socket {
   // Length of one of the members of the above union depending on the family.
   socklen_t addr_len_;
 
-  // File descriptor from PipeNotifier (see pipe_notifier.h) to send application
-  // exit notifications before calling socket blocking operations such as Read
-  // and Accept.
-  int exit_notifier_fd_;
-
-  bool exited_;
+  // Used to listen for external events (e.g. process received a SIGTERM) while
+  // blocking on I/O operations.
+  std::vector<Event> events_;
 
   DISALLOW_COPY_AND_ASSIGN(Socket);
 };