tools/android/forwarder2/forwarders_manager.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "tools/android/forwarder2/forwarders_manager.h"
   6
   7 #include <sys/select.h>
   8 #include <unistd.h>
   9
  10 #include <algorithm>
  11
  12 #include "base/basictypes.h"
  13 #include "base/bind.h"
  14 #include "base/callback_helpers.h"
  15 #include "base/location.h"
  16 #include "base/logging.h"
  17 #include "base/posix/eintr_wrapper.h"
  18 #include "tools/android/forwarder2/forwarder.h"
  19 #include "tools/android/forwarder2/socket.h"
  20
  21 namespace forwarder2 {
  22
  23 ForwardersManager::ForwardersManager() : thread_("ForwardersManagerThread") {
  24   thread_.Start();
  25   WaitForEventsOnInternalThreadSoon();
  26 }
  27
  28
  29 ForwardersManager::~ForwardersManager() {
  30   deletion_notifier_.Notify();
  31 }
  32
  33 void ForwardersManager::CreateAndStartNewForwarder(scoped_ptr<Socket> socket1,
  34                                                    scoped_ptr<Socket> socket2) {
  35   // Note that the internal Forwarder vector is populated on the internal thread
  36   // which is the only thread from which it's accessed.
  37   thread_.task_runner()->PostTask(
  38       FROM_HERE,
  39       base::Bind(&ForwardersManager::CreateNewForwarderOnInternalThread,
  40                  base::Unretained(this), base::Passed(&socket1),
  41                  base::Passed(&socket2)));
  42
  43   // Guarantees that the CreateNewForwarderOnInternalThread callback posted to
  44   // the internal thread gets executed immediately.
  45   wakeup_notifier_.Notify();
  46 }
  47
  48 void ForwardersManager::CreateNewForwarderOnInternalThread(
  49     scoped_ptr<Socket> socket1,
  50     scoped_ptr<Socket> socket2) {
  51   DCHECK(thread_.task_runner()->RunsTasksOnCurrentThread());
  52   forwarders_.push_back(new Forwarder(socket1.Pass(), socket2.Pass()));
  53 }
  54
  55 void ForwardersManager::WaitForEventsOnInternalThreadSoon() {
  56   thread_.task_runner()->PostTask(
  57       FROM_HERE,
  58       base::Bind(&ForwardersManager::WaitForEventsOnInternalThread,
  59                  base::Unretained(this)));
  60 }
  61
  62 void ForwardersManager::WaitForEventsOnInternalThread() {
  63   DCHECK(thread_.task_runner()->RunsTasksOnCurrentThread());
  64   fd_set read_fds;
  65   fd_set write_fds;
  66
  67   FD_ZERO(&read_fds);
  68   FD_ZERO(&write_fds);
  69
  70   // Populate the file descriptor sets.
  71   int max_fd = -1;
  72   for (ScopedVector<Forwarder>::iterator it = forwarders_.begin();
  73        it != forwarders_.end(); ++it) {
  74     Forwarder* const forwarder = *it;
  75     forwarder->RegisterFDs(&read_fds, &write_fds, &max_fd);
  76   }
  77
  78   const int notifier_fds[] = {
  79     wakeup_notifier_.receiver_fd(),
  80     deletion_notifier_.receiver_fd(),
  81   };
  82
  83   for (size_t i = 0; i < arraysize(notifier_fds); ++i) {
  84     const int notifier_fd = notifier_fds[i];
  85     DCHECK_GT(notifier_fd, -1);
  86     FD_SET(notifier_fd, &read_fds);
  87     max_fd = std::max(max_fd, notifier_fd);
  88   }
  89
  90   const int ret = HANDLE_EINTR(
  91       select(max_fd + 1, &read_fds, &write_fds, NULL, NULL));
  92   if (ret < 0) {
  93     PLOG(ERROR) << "select";
  94     return;
  95   }
  96
  97   const bool must_shutdown = FD_ISSET(
  98       deletion_notifier_.receiver_fd(), &read_fds);
  99   if (must_shutdown && forwarders_.empty())
 100     return;
 101
 102   base::ScopedClosureRunner wait_for_events_soon(
 103       base::Bind(&ForwardersManager::WaitForEventsOnInternalThreadSoon,
 104                  base::Unretained(this)));
 105
 106   if (FD_ISSET(wakeup_notifier_.receiver_fd(), &read_fds)) {
 107     // Note that the events on FDs other than the wakeup notifier one, if any,
 108     // will be processed upon the next select().
 109     wakeup_notifier_.Reset();
 110     return;
 111   }
 112
 113   // Notify the Forwarder instances and remove the ones that are closed.
 114   for (size_t i = 0; i < forwarders_.size(); ) {
 115     Forwarder* const forwarder = forwarders_[i];
 116     forwarder->ProcessEvents(read_fds, write_fds);
 117
 118     if (must_shutdown)
 119       forwarder->Shutdown();
 120
 121     if (!forwarder->IsClosed()) {
 122       ++i;
 123       continue;
 124     }
 125
 126     std::swap(forwarders_[i], forwarders_.back());
 127     forwarders_.pop_back();
 128   }
 129 }
 130
 131 }  // namespace forwarder2