Remove Thread wrapper class in forwarder2.

tools/android/forwarder2/forwarder.cc

Index: tools/android/forwarder2/forwarder.cc
diff --git a/tools/android/forwarder2/forwarder.cc b/tools/android/forwarder2/forwarder.cc
index 506857270bd47618a6e911bc915e42366ede9b10..df4c29cf9ff3ff01ef4ebca1209839900e920e08 100644
--- a/tools/android/forwarder2/forwarder.cc
+++ b/tools/android/forwarder2/forwarder.cc
@@ -4,18 +4,15 @@
 
 #include "tools/android/forwarder2/forwarder.h"
 
-#include <errno.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-
+#include "base/basictypes.h"
+#include "base/bind.h"
 #include "base/logging.h"
+#include "base/memory/ref_counted.h"
 #include "base/posix/eintr_wrapper.h"
-#include "base/safe_strerror_posix.h"
+#include "base/single_thread_task_runner.h"
 #include "tools/android/forwarder2/socket.h"
 
 namespace forwarder2 {
-
 namespace {
 
 // Helper class to buffer reads and writes from one socket to another.
@@ -87,62 +84,88 @@ class BufferedCopier {
   DISALLOW_COPY_AND_ASSIGN(BufferedCopier);
 };
 
-}  // namespace
-
-Forwarder::Forwarder(scoped_ptr<Socket> socket1, scoped_ptr<Socket> socket2)
-    : socket1_(socket1.Pass()),
-      socket2_(socket2.Pass()) {
-  DCHECK(socket1_.get());
-  DCHECK(socket2_.get());
-}
-
-Forwarder::~Forwarder() {
-  Detach();
-}
-
-void Forwarder::Run() {
-  const int nfds = Socket::GetHighestFileDescriptor(*socket1_, *socket2_) + 1;
-  fd_set read_fds;
-  fd_set write_fds;
-
-  // Copy from socket1 to socket2
-  BufferedCopier buffer1(socket1_.get(), socket2_.get());
+// Internal class that wraps a helper thread to forward traffic between
+// |socket1| and |socket2|. After creating a new instance, call its Start()
+// method to launch operations. Thread stops automatically if one of the socket
+// disconnects, but ensures that all buffered writes to the other, still alive,
+// socket, are written first. When this happens, the instance will delete itself
+// automatically.
+// Note that the instance will always be destroyed on the same thread that
+// created it.
+class Forwarder {
+ public:
+  Forwarder(scoped_ptr<Socket> socket1, scoped_ptr<Socket> socket2)
+      : socket1_(socket1.Pass()),
+        socket2_(socket2.Pass()),
+        destructor_runner_(base::MessageLoopProxy::current()),
+        thread_("ForwarderThread") {
+  }
 
-  // Copy from socket2 to socket1
-  BufferedCopier buffer2(socket2_.get(), socket1_.get());
+  void Start() {
+    thread_.Start();
+    thread_.message_loop_proxy()->PostTask(
+        FROM_HERE,
+        base::Bind(&Forwarder::ThreadHandler, base::Unretained(this)));
+  }
 
-  bool run = true;
-  while (run) {
-    FD_ZERO(&read_fds);
-    FD_ZERO(&write_fds);
+ private:
+  void ThreadHandler() {
+    const int nfds = Socket::GetHighestFileDescriptor(*socket1_, *socket2_) + 1;
+    fd_set read_fds;
+    fd_set write_fds;
+
+    // Copy from socket1 to socket2
+    BufferedCopier buffer1(socket1_.get(), socket2_.get());
+    // Copy from socket2 to socket1
+    BufferedCopier buffer2(socket2_.get(), socket1_.get());
+
+    bool run = true;
+    while (run) {
+      FD_ZERO(&read_fds);
+      FD_ZERO(&write_fds);
+
+      buffer1.AddToReadSet(&read_fds);
+      buffer2.AddToReadSet(&read_fds);
+      buffer1.AddToWriteSet(&write_fds);
+      buffer2.AddToWriteSet(&write_fds);
+
+      if (HANDLE_EINTR(select(nfds, &read_fds, &write_fds, NULL, NULL)) <= 0) {

[bulach, 2013/07/30 11:11:37]

there are too many layers, so apologies for what could be a silly question :)
I think we always start roughly in this order:
WPR
forwarder
chrome

however, if chrome for some reason hangs, or say, didn't finish the test, the
next bot run will do the above, and afaict, it'd kill each one of the above
individually...

once we kill wpr, or the host forwarder, would this be propagated all the way
down here?

or in another way: is there a case where both read and write fds here are valid
and we'd not receive an EINTR, so that this thread would never be able to join?

[pliard, 2013/07/30 11:50:20]

This is a non-trivial question so not silly at all :)

I'm not sure I fully understand it but I will do my best: select() should return
as soon as there is any event on at least one of the sockets.
So any side doing a close() would trigger the self-deletion of this class.Note
that this class runs both on the host and the device.

The global shutdown situation (through a kill) here could be improved by using a
"PipeNotifier" as we do in other places. Currently if one of the two processes
(host_forwarder or device_forwarder) is killed, we are (rather unintentionally)
relying on the fact that the file descriptors are automatically closed. The
other side would see a close on the corresponding socket. I will send a follow
up CL to improve that (by using a PipeNotifier here). I don't expect it to cause
any behavioral change though.

What do you think?

Are you referring to the bug where you could see ~16 threads blocked here on
select()?
I will try to reproduce the hang with this CL.

[bulach, 2013/07/30 13:11:44]

thanks for the detailed explanation!
agree with going ahead with this patch, just some more details on the case I was
thinking:
1) run a test where chrome loads a URL, but for whatever reason, it doesn't
complete (or perhaps that request had a Keep-Alive)

2) way, way higher up would eventually fail the test, like a timeout on
buildbot, or telemetry, etc... at this stage, all components (WPR, forwarder,
chrome), are left behind in an unknown state... afaict, both forwarders would be
blocked at this select() here.

3) next run comes along. at this point, I think we'd first try to kill the WPR,
then forwarder, then chrome... so if I understand correctly, the flow at this
point is:
3.1) kill WPR
3.2) close the socket.
3.3) unblock here, close the host forwarder
3.4) which would unblock the device forwarder
4) start the forwarder
5) start chrome, run the test

So I guess it will fit together nicely, right? :)

...unless.. (3.4) and (4) are async, and potentially 4 happens before... I guess
with the other change making the forwarder API synchronized at the python level
this shouldn't happen, but anyways..

+        PLOG(ERROR) << "select";
+        break;
+      }
+      // When a socket in the read set closes the connection, select() returns
+      // with that socket descriptor set as "ready to read".  When we call
+      // TryRead() below, it will return false, but the while loop will continue
+      // to run until all the write operations are finished, to make sure the
+      // buffers are completely flushed out.
+
+      // Keep running while we have some operation to do.
+      run = buffer1.TryRead(read_fds);
+      run = run || buffer2.TryRead(read_fds);
+      run = run || buffer1.TryWrite(write_fds);
+      run = run || buffer2.TryWrite(write_fds);
+    }
 
-    buffer1.AddToReadSet(&read_fds);
-    buffer2.AddToReadSet(&read_fds);
-    buffer1.AddToWriteSet(&write_fds);
-    buffer2.AddToWriteSet(&write_fds);
+    // Note that the thread that |destruction_runner_| runs tasks on could be
+    // temporarily blocked on I/O (e.g. select()) therefore it is safer to close
+    // the sockets now rather than relying on the destructor.
+    socket1_.reset();
+    socket2_.reset();
 
-    if (HANDLE_EINTR(select(nfds, &read_fds, &write_fds, NULL, NULL)) <= 0) {
-      LOG(ERROR) << "Select error: " << safe_strerror(errno);
-      break;
-    }
-    // When a socket in the read set closes the connection, select() returns
-    // with that socket descriptor set as "ready to read".  When we call
-    // TryRead() below, it will return false, but the while loop will continue
-    // to run until all the write operations are finished, to make sure the
-    // buffers are completely flushed out.
-
-    // Keep running while we have some operation to do.
-    run = buffer1.TryRead(read_fds);
-    run = run || buffer2.TryRead(read_fds);
-    run = run || buffer1.TryWrite(write_fds);
-    run = run || buffer2.TryWrite(write_fds);
+    // Note that base::Thread must be destroyed on the thread it was created on.
+    destructor_runner_->DeleteSoon(FROM_HERE, this);
   }
 
-  delete this;
-}
+  scoped_ptr<Socket> socket1_;
+  scoped_ptr<Socket> socket2_;
+  scoped_refptr<base::SingleThreadTaskRunner> destructor_runner_;
+  base::Thread thread_;
+};
+
+}  // namespace
 
-void Forwarder::Join() {
-  NOTREACHED() << "Can't Join a Forwarder thread.";
+void StartForwarder(scoped_ptr<Socket> socket1, scoped_ptr<Socket> socket2) {
+  (new Forwarder(socket1.Pass(), socket2.Pass()))->Start();
 }
 
-}  // namespace forwarder
+}  // namespace forwarder2