Upstream the android python scripts changes that enable the new Forwarder2 to be used in our test framework.

BUG=146502

content/ changes are trivial so I am adding a TBR.
TBR=avi


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

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

build/all_android.gyp

@@ -51,7 +51,6 @@
         '../third_party/WebKit/Source/WebKit/chromium/All.gyp:*',
         '../tools/android/device_stats_monitor/device_stats_monitor.gyp:device_stats_monitor',
         '../tools/android/fake_dns/fake_dns.gyp:fake_dns',
-        '../tools/android/forwarder/forwarder.gyp:forwarder',
         '../tools/android/forwarder2/forwarder.gyp:forwarder2',
         '../tools/android/md5sum/md5sum.gyp:md5sum',
         '../ui/ui.gyp:ui_unittests',

build/android/pylib/base_test_runner.py

@@ -127,6 +127,7 @@ class BaseTestRunner(object):
     else:
       logging.critical('Failed to start http server')
     self.StartForwarderForHttpServer()
+    return (self._forwarder_device_port, self._http_server.port)
 
   def StartForwarder(self, port_pairs):
     """Starts TCP traffic forwarding for the given |port_pairs|.
@@ -134,10 +135,6 @@ class BaseTestRunner(object):
     Args:
       host_port_pairs: A list of (device_port, local_port) tuples to forward.
     """
-    # Sometimes the forwarder device port may be already used. We have to kill
-    # all forwarder processes to ensure that the forwarder can be started since
-    # currently we can not associate the specified port to related pid.
-    self.adb.KillAll('forwarder')
     if self._forwarder:
       self._forwarder.Close()
     self._forwarder = Forwarder(
@@ -168,7 +165,7 @@ class BaseTestRunner(object):
     if self._forwarder or self._spawner_forwarder:
       # Kill all forwarders on the device and then kill the process on the host
       # (if it exists)
-      self.adb.KillAll('forwarder')
+      self.adb.KillAll('device_forwarder')
       if self._forwarder:
         self._forwarder.Close()
       if self._spawner_forwarder:

build/android/pylib/forwarder.py

@@ -7,14 +7,20 @@ import os
 import pexpect
 import re
 import sys
+import time
 
 import android_commands
+import cmd_helper
 import constants
+import ports
 
 class Forwarder(object):
   """Class to manage port forwards from the device to the host."""
 
-  _FORWARDER_PATH = constants.TEST_EXECUTABLE_DIR + '/forwarder'
+  _DEVICE_FORWARDER_PATH = constants.TEST_EXECUTABLE_DIR + '/device_forwarder'
+
+  # Unix Abstract socket path:
+  _DEVICE_ADB_CONTROL_PORT = 'chrome_device_forwarder'
   _TIMEOUT_SECS = 30
 
   def __init__(self, adb, port_pairs, tool, host_name, build_type):
@@ -40,75 +46,147 @@ class Forwarder(object):
     """
     self._adb = adb
     self._host_to_device_port_map = dict()
-    self._process = None
+    self._host_process = None
+    self._device_process = None
+    self._adb_forward_process = None
+
+    self._host_adb_control_port = ports.AllocateTestServerPort()
+    if not self._host_adb_control_port:
+      raise Exception('Failed to allocate a TCP port in the host machine.')
     adb.PushIfNeeded(
-        os.path.join(constants.CHROME_DIR, 'out', build_type, 'forwarder'),
-        Forwarder._FORWARDER_PATH)
+        os.path.join(constants.CHROME_DIR, 'out', build_type,
+                     'device_forwarder'),
+        Forwarder._DEVICE_FORWARDER_PATH)
+    self._host_forwarder_path = os.path.join(constants.CHROME_DIR,
+                                             'out',
+                                             build_type,
+                                             'host_forwarder')
     forward_string = ['%d:%d:%s' %
                       (device, host, host_name) for device, host in port_pairs]
+    logging.info('Forwarding ports: %s', forward_string)
+    timeout_sec = 5
+    host_pattern = 'host_forwarder.*' + ' '.join(forward_string)
+    # TODO(felipeg): Rather than using a blocking kill() here, the device
+    # forwarder could try to bind the Unix Domain Socket until it succeeds or
+    # while it fails because the socket is already bound (with appropriate
+    # timeout handling obviously).
+    self._KillHostForwarderBlocking(host_pattern, timeout_sec)
+    self._KillDeviceForwarderBlocking(timeout_sec)
+    self._adb_forward_process = pexpect.spawn(
+        'adb', ['-s',
+                adb._adb.GetSerialNumber(),
+                'forward',
+                'tcp:%s' % self._host_adb_control_port,
+                'localabstract:%s' % Forwarder._DEVICE_ADB_CONTROL_PORT])
+    self._device_process = pexpect.spawn(
+        'adb', ['-s',
+                adb._adb.GetSerialNumber(),
+                'shell',
+                '%s %s -D --adb_sock=%s' % (
+                    tool.GetUtilWrapper(),
+                    Forwarder._DEVICE_FORWARDER_PATH,
+                    Forwarder._DEVICE_ADB_CONTROL_PORT)])
 
-    # Kill off any existing forwarders on conflicting non-dynamically allocated
-    # ports.
-    for device_port, _ in port_pairs:
-      if device_port != 0:
-        self._KillForwardersUsingDevicePort(device_port)
+    device_success_re = re.compile('Starting Device Forwarder.')
+    device_failure_re = re.compile('.*:ERROR:(.*)')
+    index = self._device_process.expect([device_success_re,
+                                         device_failure_re,
+                                         pexpect.EOF,
+                                         pexpect.TIMEOUT],
+                                        Forwarder._TIMEOUT_SECS)
+    if index == 1:
+      # Failure
+      error_msg = str(self._device_process.match.group(1))
+      logging.error(self._device_process.before)
+      self._CloseProcess()
+      raise Exception('Failed to start Device Forwarder with Error: %s' %
+                      error_msg)
+    elif index == 2:
+      logging.error(self._device_process.before)
+      self._CloseProcess()
+      raise Exception('Unexpected EOF while trying to start Device Forwarder.')
+    elif index == 3:
+      logging.error(self._device_process.before)
+      self._CloseProcess()
+      raise Exception('Timeout while trying start Device Forwarder')
 
-    logging.info('Forwarding ports: %s' % (forward_string))
-    process = pexpect.spawn(
-      'adb', ['-s', adb._adb.GetSerialNumber(),
-              'shell', '%s %s -D %s' % (
-          tool.GetUtilWrapper(), Forwarder._FORWARDER_PATH,
-          ' '.join(forward_string))])
+    self._host_process = pexpect.spawn(self._host_forwarder_path,
+                                       ['--adb_port=%s' % (
+                                           self._host_adb_control_port)] +
+                                       forward_string)
 
     # Read the output of the command to determine which device ports where
     # forwarded to which host ports (necessary if
-    success_re = re.compile('Forwarding device port (\d+) to host (\d+):')
-    failure_re = re.compile('Couldn\'t start forwarder server for port spec: '
-                            '(\d+):(\d+)')
+    host_success_re = re.compile('Forwarding device port (\d+) to host (\d+):')
+    host_failure_re = re.compile('Couldn\'t start forwarder server for port '
+                                 'spec: (\d+):(\d+)')
     for pair in port_pairs:
-      index = process.expect([success_re, failure_re, pexpect.EOF,
-                              pexpect.TIMEOUT],
-                             Forwarder._TIMEOUT_SECS)
+      index = self._host_process.expect([host_success_re,
+                                         host_failure_re,
+                                         pexpect.EOF,
+                                         pexpect.TIMEOUT],
+                                        Forwarder._TIMEOUT_SECS)
       if index == 0:
         # Success
-        device_port = int(process.match.group(1))
-        host_port = int(process.match.group(2))
+        device_port = int(self._host_process.match.group(1))
+        host_port = int(self._host_process.match.group(2))
         self._host_to_device_port_map[host_port] = device_port
         logging.info("Forwarding device port: %d to host port: %d." %
                      (device_port, host_port))
       elif index == 1:
         # Failure
-        device_port = int(process.match.group(1))
-        host_port = int(process.match.group(2))
-        process.close()
+        device_port = int(self._host_process.match.group(1))
+        host_port = int(self._host_process.match.group(2))
+        self._CloseProcess()
         raise Exception('Failed to forward port %d to %d' % (device_port,
                                                              host_port))
       elif index == 2:
-        logging.error(process.before)
-        process.close()
+        logging.error(self._host_process.before)
+        self._CloseProcess()
         raise Exception('Unexpected EOF while trying to forward ports %s' %
                         port_pairs)
       elif index == 3:
-        logging.error(process.before)
-        process.close()
+        logging.error(self._host_process.before)
+        self._CloseProcess()
         raise Exception('Timeout while trying to forward ports %s' % port_pairs)
 
-    self._process = process
+  def _KillHostForwarderBlocking(self, host_pattern, timeout_sec):
+    """Kills any existing host forwarders using the provided pattern.
 
-  def _KillForwardersUsingDevicePort(self, device_port):
-    """Check if the device port is in use and if it is try to terminate the
-       forwarder process (if any) that may already be forwarding it"""
-    processes = self._adb.ProcessesUsingDevicePort(device_port)
-    for pid, name in processes:
-      if name == 'forwarder':
-        logging.warning(
-            'Killing forwarder process with pid %d using device_port %d' % (
-                 pid, device_port))
-        self._adb.RunShellCommand('kill %d' % pid)
-      else:
-        logging.error(
-             'Not killing process with pid %d (%s) using device_port %d' % (
-                 pid, name, device_port))
+    Note that this waits until the process terminates.
+    """
+    cmd_helper.RunCmd(['pkill', '-f', host_pattern])
+    elapsed = 0
+    wait_period = 0.1
+    while not cmd_helper.RunCmd(['pgrep', '-f', host_pattern]) and (
+        elapsed < timeout_sec):
+      time.sleep(wait_period)
+      elapsed += wait_period
+    if elapsed >= timeout_sec:
+        raise Exception('Timed out while killing ' + host_pattern)
+
+  def _KillDeviceForwarderBlocking(self, timeout_sec):
+    """Kills any existing device forwarders.
+
+    Note that this waits until the process terminates.
+    """
+    processes_killed = self._adb.KillAllBlocking(
+        'device_forwarder', timeout_sec)
+    if not processes_killed:
+      pids = self._adb.ExtractPid('device_forwarder')
+      if pids:
+        raise Exception('Timed out while killing device_forwarder')
+
+  def _CloseProcess(self):
+    if self._host_process:
+      self._host_process.close()
+    if self._device_process:
+      self._device_process.close()
+    if self._adb_forward_process:
+      self._adb_forward_process.close()
+    self._host_process = None
+    self._device_process = None
+    self._adb_forward_process = None
 
   def DevicePortForHostPort(self, host_port):
     """Get the device port that corresponds to a given host port."""
@@ -116,6 +194,4 @@ class Forwarder(object):
 
   def Close(self):
     """Terminate the forwarder process."""
-    if self._process:
-      self._process.close()
-      self._process = None
+    self._CloseProcess()

build/android/pylib/ports.py

@@ -46,6 +46,7 @@ def AllocateTestServerPort():
     TEST_SERVER_PORT_LAST. Returning 0 means no more valid port can be used.
   """
   port = 0
+  ports_tried = []
   try:
     fp_lock = open(constants.TEST_SERVER_PORT_LOCKFILE, 'w')
     fcntl.flock(fp_lock, fcntl.LOCK_EX)
@@ -53,8 +54,10 @@ def AllocateTestServerPort():
     assert os.path.exists(constants.TEST_SERVER_PORT_FILE)
     with open(constants.TEST_SERVER_PORT_FILE, 'r+') as fp:
       port = int(fp.read())
+      ports_tried.append(port)
       while IsHostPortUsed(port):
         port += 1
+        ports_tried.append(port)
       if (port > constants.TEST_SERVER_PORT_LAST or
           port < constants.TEST_SERVER_PORT_FIRST):
         port = 0
@@ -67,7 +70,11 @@ def AllocateTestServerPort():
     if fp_lock:
       fcntl.flock(fp_lock, fcntl.LOCK_UN)
       fp_lock.close()
-  logging.info('Allocate port %d for test server.', port)
+  if port:
+    logging.info('Allocate port %d for test server.', port)
+  else:
+    logging.error('Could not allocate port for test server. '
+                  'List of ports tried: %s', str(ports_tried))
   return port
 
 

build/android/pylib/run_java_tests.py

@@ -129,8 +129,7 @@ class TestRunner(BaseTestRunner):
     self.ports_to_forward = ports_to_forward
 
     self.test_results = TestResults()
-    # List of forwarders created by this instance of TestRunner.
-    self.forwarders = []
+    self.forwarder = None
 
     if self.coverage:
       if os.path.exists(TestRunner._COVERAGE_MERGED_FILENAME):
@@ -269,26 +268,23 @@ class TestRunner(BaseTestRunner):
     # We give different default value to launch HTTP server based on shard index
     # because it may have race condition when multiple processes are trying to
     # launch lighttpd with same port at same time.
-    # This line *must* come before the forwarding below, as it nukes all
-    # the other forwarders. A more comprehensive fix might be to pull the
-    # forwarder-killing line up to here, but that might violate assumptions
-    # implicit in other places.
-    self.LaunchTestHttpServer(os.path.join(constants.CHROME_DIR),
-                              (constants.LIGHTTPD_RANDOM_PORT_FIRST +
-                               self.shard_index))
-
+    http_server_ports = self.LaunchTestHttpServer(
+        os.path.join(constants.CHROME_DIR),
+        (constants.LIGHTTPD_RANDOM_PORT_FIRST + self.shard_index))
     if self.ports_to_forward:
-      for port in self.ports_to_forward:
-        self.forwarders.append(Forwarder(
-            self.adb, [(port, port)], self.tool, '127.0.0.1', self.build_type))
+      port_pairs = [(port, port) for port in self.ports_to_forward]
+      # We need to remember which ports the HTTP server is using, since the
+      # forwarder will stomp on them otherwise.
+      port_pairs.append(http_server_ports)
+      self.forwarder = Forwarder(
+         self.adb, port_pairs, self.tool, '127.0.0.1', self.build_type)
     self.CopyTestFilesOnce()
     self.flags.AddFlags(['--enable-test-intents'])
 
   def TearDown(self):
     """Cleans up the test harness and saves outstanding data from test run."""
-    if self.forwarders:
-      for forwarder in self.forwarders:
-        forwarder.Close()
+    if self.forwarder:
+      self.forwarder.Close()
     self.GenerateCoverageReportIfNeeded()
     super(TestRunner, self).TearDown()
 

build/android/run_instrumentation_tests.py

@@ -12,10 +12,11 @@ import time
 
 from pylib import apk_info
 from pylib import buildbot_report
-from pylib import test_options_parser
+from pylib import ports
 from pylib import run_java_tests
 from pylib import run_python_tests
 from pylib import run_tests_helper
+from pylib import test_options_parser
 from pylib.test_result import TestResults
 
 
@@ -53,6 +54,10 @@ def DispatchInstrumentationTests(options):
   Returns:
     An integer representing the number of failing tests.
   """
+  # Reset the test port allocation. It's important to do it before starting
+  # to dispatch any tests.
+  if not ports.ResetTestServerPortAllocation():
+    raise Exception('Failed to reset test server port.')
   start_date = int(time.time() * 1000)
   java_results = TestResults()
   python_results = TestResults()

content/content_tests.gypi

@@ -910,7 +910,7 @@
             '../media/media.gyp:media_test_support',
             '../net/net.gyp:net_java',
             '../net/net.gyp:net_javatests',
-            '../tools/android/forwarder/forwarder.gyp:forwarder',
+            '../tools/android/forwarder2/forwarder.gyp:forwarder2',
           ],
           'variables': {
             'package_name': 'content_shell_test',

tools/android/forwarder2/device_forwarder_main.cc

@@ -26,7 +26,7 @@ const char kDefaultAdbSocket[] = "chrome_device_forwarder";
 void KillHandler(int /* unused */) {
   CHECK(g_notifier);
   if (!g_notifier->Notify())
-    exit(-1);
+    exit(1);
 }
 
 }  // namespace

tools/android/forwarder2/host_controller.cc

@@ -68,7 +68,9 @@ bool HostController::Connect() {
     return false;
   }
   // Send the command to the device start listening to the "device_forward_port"
-  SendCommand(command::LISTEN, device_port_, &adb_control_socket_);
+  bool send_command_success = SendCommand(
+      command::LISTEN, device_port_, &adb_control_socket_);
+  CHECK(send_command_success);
   int device_port_allocated;
   command::Type command;
   if (!ReadCommand(&adb_control_socket_, &device_port_allocated, &command) ||

tools/android/forwarder2/host_forwarder_main.cc

@@ -39,7 +39,7 @@ void KillHandler(int /* unused */) {
   // (non-nicely). This is useful when debugging.
   ++s_kill_handler_count;
   if (!g_notifier->Notify() || s_kill_handler_count > 2)
-    exit(-1);
+    exit(1);
 }
 
 // Format of arg: <Device port>[:<Forward to port>:<Forward to address>]
@@ -77,6 +77,11 @@ int main(int argc, char** argv) {
     printf("Could not parse adb port number: %s\n", adb_port_str.c_str());
     show_help = true;
   }
+  if (adb_port <= 0) {
+    printf("Invalid adb port number: %s. Adb port must be a "
+           "postivie integer.\n", adb_port_str.c_str());
+    show_help = true;
+  }
   CommandLine::StringVector forward_args = command_line.GetArgs();
   if (show_help || forward_args.empty()) {
     tools::ShowHelp(
@@ -89,7 +94,7 @@ int main(int argc, char** argv) {
             "  <Forward to port> default is <Device port>\n"
             "  <Forward to address> default is 127.0.0.1.",
             kDefaultAdbPort).c_str());
-    return 0;
+    return 1;
   }
 
   g_notifier = new forwarder2::PipeNotifier();

tools/android/forwarder2/socket.cc

@@ -36,6 +36,10 @@
 namespace {
 const int kNoTimeout = -1;
 const int kConnectTimeOut = 10;  // Seconds.
+
+bool FamilyIsTCP(int family) {
+  return family == AF_INET || family == AF_INET6;
+}
 }  // namespace
 
 namespace forwarder2 {
@@ -129,7 +133,6 @@ bool Socket::InitUnixSocket(const std::string& path, bool abstract) {
   abstract_ = abstract;
   family_ = PF_UNIX;
   addr_.addr_un.sun_family = family_;
-
   if (abstract) {
     // Copied from net/base/unix_domain_socket_posix.cc
     // Convert the path given into abstract socket name. It must start with
@@ -143,14 +146,12 @@ bool Socket::InitUnixSocket(const std::string& path, bool abstract) {
     memcpy(addr_.addr_un.sun_path, path.c_str(), path.size());
     addr_len_ = sizeof(sockaddr_un);
   }
-
   addr_ptr_ = reinterpret_cast<sockaddr*>(&addr_.addr_un);
   return InitSocketInternal();
 }
 
 bool Socket::InitTcpSocket(const std::string& host, int port) {
   port_ = port;
-
   if (host.empty()) {
     // Use localhost: INADDR_LOOPBACK
     family_ = AF_INET;
@@ -159,8 +160,7 @@ bool Socket::InitTcpSocket(const std::string& host, int port) {
   } else if (!Resolve(host)) {
     return false;
   }
-  CHECK(family_ == AF_INET || family_ == AF_INET6)
-      << "Invalid socket family.";
+  CHECK(FamilyIsTCP(family_)) << "Invalid socket family.";
   if (family_ == AF_INET) {
     addr_.addr4.sin_port = htons(port_);
     addr_ptr_ = reinterpret_cast<sockaddr*>(&addr_.addr4);
@@ -180,7 +180,7 @@ bool Socket::BindAndListen() {
     SetSocketError();
     return false;
   }
-  if (port_ == 0) {
+  if (port_ == 0 && FamilyIsTCP(family_)) {
     SockAddr addr;
     memset(&addr, 0, sizeof(addr));
     socklen_t addrlen = 0;
@@ -225,19 +225,25 @@ bool Socket::Accept(Socket* new_socket) {
 }
 
 bool Socket::Connect() {
-  // Set non-block because we use select.
-  fcntl(socket_, F_SETFL, fcntl(socket_, F_GETFL) | O_NONBLOCK);
+  // Set non-block because we use select for connect.
+  const int kFlags = fcntl(socket_, F_GETFL);
+  DCHECK(!(kFlags & O_NONBLOCK));
+  fcntl(socket_, F_SETFL, kFlags | O_NONBLOCK);
   errno = 0;
   if (HANDLE_EINTR(connect(socket_, addr_ptr_, addr_len_)) < 0 &&
       errno != EINPROGRESS) {
     SetSocketError();
+    PRESERVE_ERRNO_HANDLE_EINTR(fcntl(socket_, F_SETFL, kFlags));
     return false;
   }
   // Wait for connection to complete, or receive a notification.
   if (!WaitForEvent(WRITE, kConnectTimeOut)) {
     SetSocketError();
+    PRESERVE_ERRNO_HANDLE_EINTR(fcntl(socket_, F_SETFL, kFlags));
     return false;
   }
+  // Disable non-block since our code assumes blocking semantics.
+  fcntl(socket_, F_SETFL, kFlags);
   return true;
 }
 
@@ -271,7 +277,7 @@ bool Socket::Resolve(const std::string& host) {
 }
 
 int Socket::GetPort() {
-  if (family_ != AF_INET && family_ != AF_INET6) {
+  if (!FamilyIsTCP(family_)) {
     LOG(ERROR) << "Can't call GetPort() on an unix domain socket.";
     return 0;
   }