Add Thread and Forwarder Class to the Forwarder2.

Third CL of many, to implement Forwarder2.
The big picture CL can be seem here: https://chromiumcodereview.appspot.com/10918057/

BUG=146502


Review URL: https://chromiumcodereview.appspot.com/10905186

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

5 files changed, 264 insertions, 0 deletions

diff --git a/tools/android/forwarder2/forwarder.cc b/tools/android/forwarder2/forwarder.cc
new file mode 100644
index 0000000..c4e645c
--- /dev/null
+++ b/tools/android/forwarder2/forwarder.cc
@@ -0,0 +1,153 @@
+// Copyright (c) 2012 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 "tools/android/forwarder2/forwarder.h"
+
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "base/eintr_wrapper.h"
+#include "base/logging.h"
+#include "base/safe_strerror_posix.h"
+#include "tools/android/forwarder2/socket.h"
+
+namespace forwarder2 {
+
+namespace {
+
+// Helper class to buffer reads and writes from one socket to another.
+class BufferedCopier {
+ public:
+  // Does NOT own the pointers.
+  BufferedCopier(Socket* socket_from,
+                 Socket* socket_to)
+      : socket_from_(socket_from),
+        socket_to_(socket_to),
+        bytes_read_(0),
+        write_offset_(0) {
+  }
+
+  bool AddToReadSet(fd_set* read_fds) {
+    if (bytes_read_ == 0)
+      return socket_from_->AddFdToSet(read_fds);
+    return false;
+  }
+
+  bool AddToWriteSet(fd_set* write_fds) {
+    if (write_offset_ < bytes_read_)
+      return socket_to_->AddFdToSet(write_fds);
+    return false;
+  }
+
+  bool TryRead(const fd_set& read_fds) {
+    if (!socket_from_->IsFdInSet(read_fds))
+      return false;
+    if (bytes_read_ != 0)  // Can't read.
+      return false;
+    int ret = socket_from_->Read(buffer_, kBufferSize);
+    if (ret > 0) {
+      bytes_read_ = ret;
+      return true;
+    }
+    return false;
+  }
+
+  bool TryWrite(const fd_set& write_fds) {
+    if (!socket_to_->IsFdInSet(write_fds))
+      return false;
+    if (write_offset_ >= bytes_read_)  // Nothing to write.
+      return false;
+    int ret = socket_to_->Write(buffer_ + write_offset_,
+                                bytes_read_ - write_offset_);
+    if (ret > 0) {
+      write_offset_ += ret;
+      if (write_offset_ == bytes_read_) {
+        write_offset_ = 0;
+        bytes_read_ = 0;
+      }
+      return true;
+    }
+    return false;
+  }
+
+ private:
+  // Not owned.
+  Socket* socket_from_;
+  Socket* socket_to_;
+
+  // A big buffer to let our file-over-http bridge work more like real file.
+  static const int kBufferSize = 1024 * 128;
+  int bytes_read_;
+  int write_offset_;
+  char buffer_[kBufferSize];
+
+  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());
+  // 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();
+}
+
+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());
+
+  // 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) {
+      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);
+  }
+
+  delete this;
+}
+
+void Forwarder::Join() {
+  NOTREACHED() << "Can't Join a Forwarder thread.";
+}
+
+}  // namespace forwarder
diff --git a/tools/android/forwarder2/forwarder.gyp b/tools/android/forwarder2/forwarder.gyp
index 7b62e1f..5bff33e 100644
--- a/tools/android/forwarder2/forwarder.gyp
+++ b/tools/android/forwarder2/forwarder.gyp
@@ -39,7 +39,9 @@
       'sources': [
         'command.cc',
         'device_forwarder_main.cc',
+        'forwarder.cc',
         'socket.cc',
+        'thread.cc',
       ],
     },
     {
@@ -55,8 +57,10 @@
       ],
       'sources': [
         'command.cc',
+        'forwarder.cc',
         'host_forwarder_main.cc',
         'socket.cc',
+        'thread.cc',
       ],
     },
   ],
diff --git a/tools/android/forwarder2/forwarder.h b/tools/android/forwarder2/forwarder.h
new file mode 100644
index 0000000..1ba09232
--- /dev/null
+++ b/tools/android/forwarder2/forwarder.h
@@ -0,0 +1,39 @@
+// Copyright (c) 2012 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.
+
+#ifndef TOOLS_ANDROID_FORWARDER2_FORWARDER_H_
+#define TOOLS_ANDROID_FORWARDER2_FORWARDER_H_
+
+#include "base/compiler_specific.h"
+#include "base/logging.h"
+#include "base/memory/scoped_ptr.h"
+#include "tools/android/forwarder2/socket.h"
+#include "tools/android/forwarder2/thread.h"
+
+namespace forwarder2 {
+
+class Forwarder : public Thread {
+ public:
+  Forwarder(scoped_ptr<Socket> socket1, scoped_ptr<Socket> socket2);
+  virtual ~Forwarder();
+
+  // This object self deletes after running, so one cannot join.
+  // Thread:
+  virtual void Join() OVERRIDE;
+
+ protected:
+  // Thread:
+  // This object self deletes after running.
+  virtual void Run() OVERRIDE;
+
+ private:
+  scoped_ptr<Socket> socket1_;
+  scoped_ptr<Socket> socket2_;
+
+  DISALLOW_COPY_AND_ASSIGN(Forwarder);
+};
+
+}  // namespace forwarder
+
+#endif  // TOOLS_ANDROID_FORWARDER2_FORWARDER_H_
diff --git a/tools/android/forwarder2/thread.cc b/tools/android/forwarder2/thread.cc
new file mode 100644
index 0000000..aa5da4b
--- /dev/null
+++ b/tools/android/forwarder2/thread.cc
@@ -0,0 +1,34 @@
+// Copyright (c) 2012 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 "tools/android/forwarder2/thread.h"
+
+#include "base/logging.h"
+
+namespace forwarder2 {
+
+Thread::Thread() : thread_(static_cast<pthread_t>(-1)) {}
+
+Thread::~Thread() {}
+
+void Thread::Start() {
+  int ret = pthread_create(&thread_, NULL, &ThreadCallback, this);
+  CHECK_EQ(0, ret);
+}
+
+void Thread::Join() {
+  CHECK_NE(static_cast<pthread_t>(-1), thread_);
+  int ret = pthread_join(thread_, NULL);
+  CHECK_EQ(0, ret);
+}
+
+// static
+void* Thread::ThreadCallback(void* arg) {
+  CHECK(arg);
+  Thread* obj = reinterpret_cast<Thread*>(arg);
+  obj->Run();
+  return NULL;
+}
+
+}  // namespace forwarder
diff --git a/tools/android/forwarder2/thread.h b/tools/android/forwarder2/thread.h
new file mode 100644
index 0000000..2b5d480
--- /dev/null
+++ b/tools/android/forwarder2/thread.h
@@ -0,0 +1,34 @@
+// Copyright (c) 2012 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.
+
+#ifndef TOOLS_ANDROID_FORWARDER2_THREAD_H_
+#define TOOLS_ANDROID_FORWARDER2_THREAD_H_
+
+#include <pthread.h>
+
+namespace forwarder2 {
+
+// Helper abstract class to create a new thread and curry the
+// call to the Run() method in an object from the new thread.
+class Thread {
+ public:
+  Thread();
+  virtual ~Thread();
+
+  // Start thread calling Run().
+  void Start();
+  virtual void Join();
+
+ protected:
+  virtual void Run() = 0;
+
+ private:
+  static void* ThreadCallback(void* arg);
+
+  pthread_t thread_;
+};
+
+}  // namespace forwarder
+
+#endif  // TOOLS_ANDROID_FORWARDER2_THREAD_H_