forwarder2: Make the Forwarder instances operate on a single thread.

tools/android/forwarder2/forwarders_manager.cc

 // Copyright 2013 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/forwarders_manager.h"
 
+#include <sys/select.h>
+#include <unistd.h>
+
 #include <algorithm>
 
 #include "base/basictypes.h"
 #include "base/bind.h"
+#include "base/callback_helpers.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop_proxy.h"
+#include "base/posix/eintr_wrapper.h"
 #include "tools/android/forwarder2/forwarder.h"
 #include "tools/android/forwarder2/socket.h"
 
 namespace forwarder2 {
 
-ForwardersManager::ForwardersManager() : delegate_(new Delegate()) {}
+ForwardersManager::ForwardersManager() : thread_("ForwardersManagerThread") {
+  thread_.Start();
+  WaitForEventsOnInternalThreadSoon();
+}
+
 
 ForwardersManager::~ForwardersManager() {
-  delegate_->Clear();
+  deletion_notifier_.Notify();
 }
 
 void ForwardersManager::CreateAndStartNewForwarder(scoped_ptr<Socket> socket1,
                                                    scoped_ptr<Socket> socket2) {
-  delegate_->CreateAndStartNewForwarder(socket1.Pass(), socket2.Pass());
+  // Note that the internal Forwarder vector is populated on the internal thread
+  // which is the only thread from which it's accessed.
+  thread_.message_loop_proxy()->PostTask(
+      FROM_HERE,
+      base::Bind(&ForwardersManager::CreateNewForwarderOnInternalThread,
+                 base::Unretained(this), base::Passed(&socket1),
+                 base::Passed(&socket2)));
+
+  // Guarantees that the CreateNewForwarderOnInternalThread callback posted to
+  // the internal thread gets executed immediately.
+  wakeup_notifier_.Notify();
 }
 
-ForwardersManager::Delegate::Delegate() {}
-
-ForwardersManager::Delegate::~Delegate() {
-  // The forwarder instances should already have been deleted on their
-  // construction thread. Deleting them here would be unsafe since we don't know
-  // which thread this destructor is called on.
-  DCHECK(forwarders_.empty());
+void ForwardersManager::CreateNewForwarderOnInternalThread(
+    scoped_ptr<Socket> socket1,
+    scoped_ptr<Socket> socket2) {
+  DCHECK(thread_.message_loop_proxy()->RunsTasksOnCurrentThread());
+  forwarders_.push_back(new Forwarder(socket1.Pass(), socket2.Pass()));
 }
 
-void ForwardersManager::Delegate::Clear() {
-  if (!forwarders_constructor_runner_) {
-    DCHECK(forwarders_.empty());
+void ForwardersManager::WaitForEventsOnInternalThreadSoon() {
+  thread_.message_loop_proxy()->PostTask(
+      FROM_HERE,
+      base::Bind(&ForwardersManager::WaitForEventsOnInternalThread,
+                 base::Unretained(this)));
+}
+
+void ForwardersManager::WaitForEventsOnInternalThread() {
+  DCHECK(thread_.message_loop_proxy()->RunsTasksOnCurrentThread());
+  fd_set read_fds;
+  fd_set write_fds;
+
+  FD_ZERO(&read_fds);
+  FD_ZERO(&write_fds);
+
+  // Populate the file descriptor sets.
+  int max_fd = -1;
+  for (ScopedVector<Forwarder>::iterator it = forwarders_.begin();
+       it != forwarders_.end(); ++it) {
+    Forwarder* const forwarder = *it;
+    forwarder->RegisterFDs(&read_fds, &write_fds, &max_fd);
+  }
+
+  const int notifier_fds[] = {
+    wakeup_notifier_.receiver_fd(),
+    deletion_notifier_.receiver_fd(),
+  };
+
+  for (int i = 0; i < arraysize(notifier_fds); ++i) {
+    const int notifier_fd = notifier_fds[i];
+    DCHECK_GT(notifier_fd, -1);
+    FD_SET(notifier_fd, &read_fds);
+    max_fd = std::max(max_fd, notifier_fd);
+  }
+
+  const int ret = HANDLE_EINTR(
+      select(max_fd + 1, &read_fds, &write_fds, NULL, NULL));
+  if (ret < 0) {
+    PLOG(ERROR) << "select";
     return;
   }
-  if (forwarders_constructor_runner_->RunsTasksOnCurrentThread()) {
-    ClearOnForwarderConstructorThread();
+
+  const bool must_shutdown = FD_ISSET(
+      deletion_notifier_.receiver_fd(), &read_fds);
+  if (must_shutdown && forwarders_.empty())
+    return;
+
+  base::ScopedClosureRunner wait_for_events_soon(

[digit1, 2014/01/28 17:28:33]

Can you avoid using this? a base::Bind() is pretty heavy and allocates heap
memory on each loop iteration (which gets released when the function returns).
This looks like un-needed CPU/heap churn to me.

Also you're binding to a function that will do its own base::Bind() to re-run
the current function, this is also very heavy in each loop iteration. I wonder
if a simple loop here would not be more appropriate, with a different way to
communicate additions / deletions to the thread.

[qsr, 2014/01/29 09:30:17]

 Given that this is only called when select returns and so, before some I/O,
this is far from being a busy loop. Starting to use a custom thread and custom
message passing for this seem to be a lot of effort for a gain that is not
really evident at this point.

+      base::Bind(&ForwardersManager::WaitForEventsOnInternalThreadSoon,
+                 base::Unretained(this)));
+
+  if (FD_ISSET(wakeup_notifier_.receiver_fd(), &read_fds)) {
+    // Note that the events on FDs other than the wakeup notifier one, if any,
+    // will be processed upon the next select().
+    wakeup_notifier_.Reset();
     return;
   }
-  forwarders_constructor_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(
-          &ForwardersManager::Delegate::ClearOnForwarderConstructorThread,
-          this));
-}
 
-void ForwardersManager::Delegate::CreateAndStartNewForwarder(
-    scoped_ptr<Socket> socket1,
-    scoped_ptr<Socket> socket2) {
-  const scoped_refptr<base::SingleThreadTaskRunner> current_task_runner(
-      base::MessageLoopProxy::current());
-  DCHECK(current_task_runner);
-  if (forwarders_constructor_runner_) {
-    DCHECK_EQ(current_task_runner, forwarders_constructor_runner_);
-  } else {
-    forwarders_constructor_runner_ = current_task_runner;
+  // Notify the Forwarder instances and remove the ones that are closed.
+  for (size_t i = 0; i < forwarders_.size(); ) {
+    Forwarder* const forwarder = forwarders_[i];
+    forwarder->ProcessEvents(read_fds, write_fds);
+
+    if (must_shutdown)
+      forwarder->Shutdown();
+
+    if (!forwarder->IsClosed()) {

[pasko, 2014/01/27 16:59:02]

I think if you replace one loop with three, it would get easier to read (LOL),
since the reader now has to keep in mind that Shutdown() makes the forwarder
closed. Also no modifying of the index variable in the loop.

proposed pseudocode:

for (size_t i = 0; i < forwarders_.size(); ) {
  forwarders_[i]->ProcessEvents(read_fds, write_fds);
}
if (must_shutdown) {
  for (auto forwarder : forwarders_) {
    forwarder->Shutdown();
  }
  forwarders_.clear();
} else {
  // Remove forwarders that are closed.
}

This may change the order of events, so I am not even sure it would work. So
even though it repeats looping three times, the code looks more linear and more
obvious. The code would not be much longer.

Up to you.

[Philippe, 2014/01/28 08:59:09]

You slightly cheated by using a foreach loop :) So the code would be even
longer.

The thing which is non-obvious here too is the fact that a Forwarder instance is
not necessarily immediately closed after Shutdown() was called therefore we
can't call |forwarders_.clear| as you did. If one of the internal Forwarder's
BufferCopier still has data in its buffer then it would go to the 'CLOSING'
state (as opposed to 'CLOSED') after a call to Shutdown() in which it flushes
its buffer.

So the code would rather be:
for (ScopedVector<Forwarder>::iterator it = forwarders_.begin();
         it != forwarders_.end(); ++it) {
  (*it)->ProcessEvents(read_fds, write_fds);
} 

if (must_shutdown) {
  for (ScopedVector<Forwarder>::iterator it = forwarders_.begin();
         it != forwarders_.end(); ++it) {
    (*it)->Shutdown();
  } 
}

for (size_t i = 0; i < forwarders_.size(); ) {
  Forwarder* const forwarder = forwarders_[i];
  if (!forwarder->IsClosed()) {
    ++i;
    continue;
  }
  std::swap(forwarders_[i], forwarders_.back());
  forwarders_.resize();
}
  
Which is much shorter, right? :)

It's a shame BTW that base::Callback doesn't provide an operator() so that we
can use it with STL algorithms.

[pasko, 2014/01/28 10:48:56]

Oh, I did not realize that. It can get straight to closed or to closing.
Wonderful.
Thanks for explanation.
 
absolutely)

To me the most irritating was understanding how the index variable gets modified
in the loop and why we are not using end(), and such, so a more compact loop
helps here, but making process and shutdown loops separate does not in fact make
sense if we cannot do clear(). How about this:

====================
for (ScopedVector<Forwarder>::iterator it = forwarders_.begin();
     it != forwarders_.end(); ++it) {
  Forwarder* forwarder = (*it);
  forwarder->ProcessEvents(read_fds, write_fds);
  if (must_shutdown)
    forwarder->Shutdown();
}

for (size_t i = 0; i < forwarders_.size(); ) {
  Forwarder* const forwarder = forwarders_[i];
  if (!forwarder->IsClosed()) {
    ++i;
    continue;
  }
  std::swap(forwarders_[i], forwarders_.back());
  forwarders_.pop_back();
}
====================

This is roughly +3 lines, but is easier for me to read. I won't insist though.
It is already taking too much time.
+1, though imagine you give this freedom to certain colleagues, they'd say it's
only a half solution, instead we should switch directly to LISP.

+      ++i;
+      continue;
+    }
+
+    std::swap(forwarders_[i], forwarders_.back());
+    forwarders_.pop_back();
   }
-  forwarders_.push_back(
-      new Forwarder(socket1.Pass(), socket2.Pass(),
-                    &deletion_notifier_,
-                    base::Bind(&ForwardersManager::Delegate::OnForwarderError,
-                               this)));
-  forwarders_.back()->Start();
-}
-
-void ForwardersManager::Delegate::OnForwarderError(
-    scoped_ptr<Forwarder> forwarder) {
-  DCHECK(forwarders_constructor_runner_->RunsTasksOnCurrentThread());
-  const ScopedVector<Forwarder>::iterator it = std::find(
-      forwarders_.begin(), forwarders_.end(), forwarder.get());
-  DCHECK(it != forwarders_.end());
-  std::swap(*it, forwarders_.back());
-  forwarders_.pop_back();
-  ignore_result(forwarder.release());  // Deleted by the pop_back() above.
-}
-
-void ForwardersManager::Delegate::ClearOnForwarderConstructorThread() {
-  DCHECK(forwarders_constructor_runner_->RunsTasksOnCurrentThread());
-  deletion_notifier_.Notify();
-  forwarders_.clear();
 }
 
 }  // namespace forwarder2