Remove Thread wrapper class in forwarder2.

tools/android/forwarder2/device_controller.cc

Index: tools/android/forwarder2/device_controller.cc
diff --git a/tools/android/forwarder2/device_controller.cc b/tools/android/forwarder2/device_controller.cc
index e9ce9cb7c37eb9f331b6ed670a3d432daae1816b..20e1ed9d15b70597b3fe5321fe7e6dd59a9ec0d1 100644
--- a/tools/android/forwarder2/device_controller.cc
+++ b/tools/android/forwarder2/device_controller.cc
@@ -7,9 +7,12 @@
 #include <errno.h>
 #include <stdlib.h>
 
+#include "base/bind.h"
+#include "base/bind_helpers.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
-#include "base/safe_strerror_posix.h"
+#include "base/message_loop/message_loop_proxy.h"
+#include "base/single_thread_task_runner.h"
 #include "tools/android/forwarder2/command.h"
 #include "tools/android/forwarder2/device_listener.h"
 #include "tools/android/forwarder2/socket.h"
@@ -17,38 +20,17 @@
 namespace forwarder2 {
 
 DeviceController::DeviceController(int exit_notifier_fd)
-    : exit_notifier_fd_(exit_notifier_fd) {
+    : exit_notifier_fd_(exit_notifier_fd),
+      listeners_deletion_thread_(base::MessageLoopProxy::current()) {
   kickstart_adb_socket_.AddEventFd(exit_notifier_fd);
 }
 
-DeviceController::~DeviceController() {
-  KillAllListeners();
-  CleanUpDeadListeners();
-  CHECK_EQ(0, listeners_.size());
-}
-
-void DeviceController::CleanUpDeadListeners() {
-  // Clean up dead listeners.
-  for (ListenersMap::iterator it(&listeners_); !it.IsAtEnd(); it.Advance()) {
-    if (!it.GetCurrentValue()->is_alive())
-      // Remove deletes the listener.
-      listeners_.Remove(it.GetCurrentKey());
-  }
-}
-
-void DeviceController::KillAllListeners() {
-  for (ListenersMap::iterator it(&listeners_); !it.IsAtEnd(); it.Advance())
-    it.GetCurrentValue()->ForceExit();
-  for (ListenersMap::iterator it(&listeners_); !it.IsAtEnd(); it.Advance()) {
-    it.GetCurrentValue()->Join();
-    CHECK(!it.GetCurrentValue()->is_alive());
-  }
-}
+DeviceController::~DeviceController() {}
 
 bool DeviceController::Init(const std::string& adb_unix_socket) {
   if (!kickstart_adb_socket_.BindUnix(adb_unix_socket)) {
-    LOG(ERROR) << "Could not BindAndListen DeviceController socket on port "
-               << adb_unix_socket << ": " << safe_strerror(errno);
+    PLOG(ERROR) << "Could not BindAndListen DeviceController socket on port "
+                << adb_unix_socket << ": ";
     return false;
   }
   LOG(INFO) << "Listening on Unix Domain Socket " << adb_unix_socket;
@@ -56,87 +38,103 @@ bool DeviceController::Init(const std::string& adb_unix_socket) {
 }
 
 void DeviceController::Start() {
-  while (true) {
-    CleanUpDeadListeners();
-    scoped_ptr<Socket> socket(new Socket);
-    if (!kickstart_adb_socket_.Accept(socket.get())) {
-      if (!kickstart_adb_socket_.DidReceiveEvent()) {
-        LOG(ERROR) << "Could not Accept DeviceController socket: "
-                   << safe_strerror(errno);
-      } else {
-        LOG(INFO) << "Received exit notification";
+  AcceptClientSoon();
+}
+
+void DeviceController::AcceptClientSoon() {

[digit1, 2013/07/18 20:08:39]

nit: I think AcceptHostCommandSoon() / AcceptHostCommand() would be more
descriptive names here

[Philippe, 2013/07/22 15:16:14]

Indeed.

+  base::MessageLoopProxy::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&DeviceController::AcceptClientInternal,
+                 base::Unretained(this)));
+}
+
+void DeviceController::AcceptClientInternal() {
+  scoped_ptr<Socket> socket(new Socket);
+  if (!kickstart_adb_socket_.Accept(socket.get())) {
+    if (!kickstart_adb_socket_.DidReceiveEvent()) {
+      PLOG(ERROR) << "Could not Accept DeviceController socket";
+    } else {
+      LOG(INFO) << "Received exit notification";
+    }
+    return;
+  }
+  base::ScopedClosureRunner accept_next_client(
+      base::Bind(&DeviceController::AcceptClientSoon, base::Unretained(this)));
+  // So that |socket| doesn't block on read if it has notifications.
+  socket->AddEventFd(exit_notifier_fd_);
+  int port;
+  command::Type command;
+  if (!ReadCommand(socket.get(), &port, &command)) {
+    LOG(ERROR) << "Invalid command received.";
+    return;
+  }
+  DeviceListener* listener = listeners_.Lookup(port);
+  switch (command) {
+    case command::LISTEN: {
+      if (listener != NULL) {
+        LOG(WARNING) << "Already forwarding port " << port
+                     << ". Attempting to restart the listener.\n";
+        // Remove deletes the listener object.
+        listeners_.Remove(port);
       }
+      scoped_ptr<DeviceListener> new_listener(
+          DeviceListener::Create(
+              socket.Pass(), port, base::Bind(&DeviceController::DeleteListener,
+                                              base::Unretained(this))));
+      if (!new_listener)
+        return;
+      new_listener->Start();
+      // |port| can be zero, to allow dynamically allocated port, so instead, we
+      // call DeviceListener::listener_port() to retrieve the currently
+      // allocated port to this new listener, which has been set by the
+      // BindListenerSocket() method in case of success.
+      const int listener_port = new_listener->listener_port();
+      // |new_listener| is now owned by listeners_ map.
+      listeners_.AddWithID(new_listener.release(), listener_port);
+      LOG(INFO) << "Forwarding device port " << listener_port << " to host.";
       break;
     }
-    // So that |socket| doesn't block on read if it has notifications.
-    socket->AddEventFd(exit_notifier_fd_);
-    int port;
-    command::Type command;
-    if (!ReadCommand(socket.get(), &port, &command)) {
-      LOG(ERROR) << "Invalid command received.";
-      continue;
-    }
-    DeviceListener* listener = listeners_.Lookup(port);
-    switch (command) {
-      case command::LISTEN: {
-        if (listener != NULL) {
-          LOG(WARNING) << "Already forwarding port " << port
-                       << ". Attempting to restart the listener.\n";
-          listener->ForceExit();
-          listener->Join();
-          CHECK(!listener->is_alive());
-          // Remove deletes the listener object.
-          listeners_.Remove(port);
-        }
-        scoped_ptr<DeviceListener> new_listener(
-            new DeviceListener(socket.Pass(), port));
-        if (!new_listener->BindListenerSocket())
-          continue;
-        new_listener->Start();
-        // |port| can be zero, to allow dynamically allocated port, so instead,
-        // we call DeviceListener::listener_port() to retrieve the currently
-        // allocated port to this new listener, which has been set by the
-        // BindListenerSocket() method in case of success.
-        const int listener_port = new_listener->listener_port();
-        // |new_listener| is now owned by listeners_ map.
-        listeners_.AddWithID(new_listener.release(), listener_port);
-        LOG(INFO) << "Forwarding device port " << listener_port << " to host.";
+    case command::DATA_CONNECTION:
+      if (listener == NULL) {
+        LOG(ERROR) << "Data Connection command received, but "
+                   << "listener has not been set up yet for port " << port;
+        // After this point it is assumed that, once we close our Adb Data
+        // socket, the Adb forwarder command will propagate the closing of
+        // sockets all the way to the host side.
         break;
       }
-      case command::DATA_CONNECTION:
-        if (listener == NULL) {
-          LOG(ERROR) << "Data Connection command received, but "
-                     << "listener has not been set up yet for port " << port;
-          // After this point it is assumed that, once we close our Adb Data
-          // socket, the Adb forwarder command will propagate the closing of
-          // sockets all the way to the host side.
-          continue;
-        } else if (!listener->SetAdbDataSocket(socket.Pass())) {
-          LOG(ERROR) << "Could not set Adb Data Socket for port: " << port;
-          // Same assumption as above, but in this case the socket is closed
-          // inside SetAdbDataSocket.
-          continue;
-        }
-        break;
-      case command::UNMAP_PORT:
-        if (!listener) {
-          SendCommand(command::UNMAP_PORT_ERROR, port, socket.get());
-          break;
-        }
-        listener->ForceExit();
-        listener->Join();
-        CHECK(!listener->is_alive());
-        listeners_.Remove(port);
-        SendCommand(command::UNMAP_PORT_SUCCESS, port, socket.get());
+      listener->SetAdbDataSocket(socket.Pass());

[digit1, 2013/07/18 20:08:39]

Since you changed the interaction between DeviceController and DeviceListener,
shouldn't you check that the listener is effectively waiting for a new socket
here? Otherwise things could get hairy if the host forwarder sends this command
by mistake.

[Philippe, 2013/07/22 15:16:14]

I believe it should not be too risky as it is currently. The DeviceListener
instances are created and destroyed on the same thread this function runs on. If
|listener| is non-null then the listener should be accepting/have accepted a
connection. In case it fails while doing that (on its own internal thread), then
it self-deletes on the controller thread (see DeleteListener() below) which is
also the thread this function is running on. In the worst case the
OnAdbDataSocketReceivedOnInternalThread() callback is queued and executed after
a socket error. This would cause an error when this callback sends a command to
the host but I don't see more risks like use after free (not sure if this is the
kind of things you meant by "hairy" :)). Sorry, not sure this is very clear :)

What do you think?

+      break;
+    case command::UNMAP_PORT:

[digit1, 2013/07/18 20:08:39]

nit:  This should really be called UNLISTEN :-) Let's keep it for another CL
though.

[Philippe, 2013/07/22 15:16:14]

Yeah indeed :) I made the change in this CL since it's a simple change.

+      if (!listener) {
+        SendCommand(command::UNMAP_PORT_ERROR, port, socket.get());
         break;
-      default:
-        // TODO(felipeg): add a KillAllListeners command.
-        LOG(ERROR) << "Invalid command received. Port: " << port
-                   << " Command: " << command;
-    }
+      }
+      listeners_.Remove(port);
+      SendCommand(command::UNMAP_PORT_SUCCESS, port, socket.get());
+      break;
+    default:
+      // TODO(felipeg): add a KillAllListeners command.
+      LOG(ERROR) << "Invalid command received. Port: " << port
+                 << " Command: " << command;
+  }
+}
+
+void DeviceController::DeleteListener(int listener_port) {
+  if (!listeners_deletion_thread_->RunsTasksOnCurrentThread()) {
+    listeners_deletion_thread_->PostTask(
+        FROM_HERE,
+        base::Bind(&DeviceController::DeleteListenerOnTaskRunner,
+                   base::Unretained(this), listener_port));
+    return;
   }
-  KillAllListeners();
-  CleanUpDeadListeners();
+  DeleteListenerOnTaskRunner(listener_port);
+}
+
+void DeviceController::DeleteListenerOnTaskRunner(int listener_port) {
+  DCHECK(listeners_deletion_thread_->RunsTasksOnCurrentThread());
+  if (listeners_.Lookup(listener_port))
+    listeners_.Remove(listener_port);
 }
 
 }  // namespace forwarder