Reland r212020: Move Python setup/tear down logic into Forwarder ...

Forwarder used to be a pain to setup/tear down across all the various
harnesses.
This CL should hopefully solve these issues by hiding these
implementation details. The host daemon is now killed once the first time that
the Forwarder class is used and the daemon running on the devices is also
killed the first time a port is forwarded for a specific device.

TBR=bulach@chromium.org
BUG=242846

Review URL: https://codereview.chromium.org/19550004

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

8 files changed, 244 insertions, 131 deletions

diff --git a/build/android/adb_reverse_forwarder.py b/build/android/adb_reverse_forwarder.py
index e465ea7..ee38332 100755
--- a/build/android/adb_reverse_forwarder.py
+++ b/build/android/adb_reverse_forwarder.py
@@ -17,7 +17,6 @@ import time
 
 from pylib import android_commands, forwarder
 from pylib.utils import run_tests_helper
-from pylib.valgrind_tools import CreateTool
 
 
 def main(argv):
@@ -51,17 +50,14 @@ def main(argv):
     sys.exit(1)
 
   adb = android_commands.AndroidCommands(options.device)
-  tool = CreateTool(None, adb)
-  forwarder_instance = forwarder.Forwarder(adb, options.build_type)
   try:
-    forwarder_instance.Run(port_pairs, tool)
+    forwarder.Forwarder.Map(port_pairs, adb, options.build_type)
     while True:
       time.sleep(60)
   except KeyboardInterrupt:
     sys.exit(0)
   finally:
-    forwarder_instance.Close()
-
+    forwarder.Forwarder.UnmapAllDevicePorts(adb)
 
 if __name__ == '__main__':
   main(sys.argv)
diff --git a/build/android/bb_run_sharded_steps.py b/build/android/bb_run_sharded_steps.py
index afd354a..9b768a9 100755
--- a/build/android/bb_run_sharded_steps.py
+++ b/build/android/bb_run_sharded_steps.py
@@ -165,7 +165,6 @@ def _PrintAllStepsOutput(steps):
 
 def _KillPendingServers():
   for retry in range(5):
-    forwarder.Forwarder.KillHost()
     for server in ['lighttpd', 'web-page-replay']:
       pids = cmd_helper.GetCmdOutput(['pgrep', '-f', server])
       pids = [pid.strip() for pid in pids.split('\n') if pid.strip()]
@@ -201,6 +200,8 @@ def main(argv):
   # previous runs.
   _KillPendingServers()
 
+  forwarder.Forwarder.UseMultiprocessing()
+
   # Reset the test port allocation. It's important to do it before starting
   # to dispatch any step.
   if not ports.ResetTestServerPortAllocation():
diff --git a/build/android/pylib/base/base_test_runner.py b/build/android/pylib/base/base_test_runner.py
index df04f9f..9173b72 100644
--- a/build/android/pylib/base/base_test_runner.py
+++ b/build/android/pylib/base/base_test_runner.py
@@ -44,14 +44,12 @@ class BaseTestRunner(object):
     self.adb = android_commands.AndroidCommands(device=device)
     self.tool = CreateTool(tool, self.adb)
     self._http_server = None
-    self._forwarder = None
     self._forwarder_device_port = 8000
     self.forwarder_base_url = ('http://localhost:%d' %
         self._forwarder_device_port)
     self.flags = FlagChanger(self.adb)
     self.flags.AddFlags(['--disable-fre'])
     self._spawning_server = None
-    self._spawner_forwarder = None
     # We will allocate port for test server spawner when calling method
     # LaunchChromeTestServerSpawner and allocate port for test server when
     # starting it in TestServerThread.
@@ -90,7 +88,6 @@ class BaseTestRunner(object):
 
   def SetUp(self):
     """Run once before all tests are run."""
-    Forwarder.KillDevice(self.adb, self.tool)
     self.InstallTestPackage()
     push_size_before = self.adb.GetPushSizeInfo()
     if self._push_deps:
@@ -128,49 +125,47 @@ class BaseTestRunner(object):
                    self._http_server.port)
     else:
       logging.critical('Failed to start http server')
-    self.StartForwarderForHttpServer()
+    self._ForwardPortsForHttpServer()
     return (self._forwarder_device_port, self._http_server.port)
 
-  def _ForwardPort(self, port_pairs):
-    """Creates a forwarder instance if needed and forward a port."""
-    if not self._forwarder:
-      self._forwarder = Forwarder(self.adb, self.build_type)
-    self._forwarder.Run(port_pairs, self.tool)
+  def _ForwardPorts(self, port_pairs):
+    """Forwards a port."""
+    Forwarder.Map(port_pairs, self.adb, self.build_type, self.tool)
 
+  def _UnmapPorts(self, port_pairs):
+    """Unmap previously forwarded ports."""
+    for (device_port, _) in port_pairs:
+      Forwarder.UnmapDevicePort(device_port, self.adb)
+
+  # Deprecated: Use ForwardPorts instead.
   def StartForwarder(self, port_pairs):
     """Starts TCP traffic forwarding for the given |port_pairs|.
 
     Args:
       host_port_pairs: A list of (device_port, local_port) tuples to forward.
     """
-    self._ForwardPort(port_pairs)
+    self._ForwardPorts(port_pairs)
 
-  def StartForwarderForHttpServer(self):
+  def _ForwardPortsForHttpServer(self):
     """Starts a forwarder for the HTTP server.
 
     The forwarder forwards HTTP requests and responses between host and device.
     """
-    self._ForwardPort([(self._forwarder_device_port, self._http_server.port)])
+    self._ForwardPorts([(self._forwarder_device_port, self._http_server.port)])
 
-  def RestartHttpServerForwarderIfNecessary(self):
+  def _RestartHttpServerForwarderIfNecessary(self):
     """Restarts the forwarder if it's not open."""
     # Checks to see if the http server port is being used.  If not forwards the
     # request.
     # TODO(dtrainor): This is not always reliable because sometimes the port
     # will be left open even after the forwarder has been killed.
-    if not ports.IsDevicePortUsed(self.adb,
-        self._forwarder_device_port):
-      self.StartForwarderForHttpServer()
+    if not ports.IsDevicePortUsed(self.adb, self._forwarder_device_port):
+      self._ForwardPortsForHttpServer()
 
   def ShutdownHelperToolsForTestSuite(self):
     """Shuts down the server and the forwarder."""
-    # Forwarders should be killed before the actual servers they're forwarding
-    # to as they are clients potentially with open connections and to allow for
-    # proper hand-shake/shutdown.
-    Forwarder.KillDevice(self.adb, self.tool)
-    if self._forwarder:
-      self._forwarder.Close()
     if self._http_server:
+      self._UnmapPorts([(self._forwarder_device_port, self._http_server.port)])
       self._http_server.ShutdownHttpServer()
     if self._spawning_server:
       self._spawning_server.Stop()
@@ -195,12 +190,11 @@ class BaseTestRunner(object):
     # Try 3 times to launch test spawner server.
     for i in xrange(0, 3):
       self.test_server_spawner_port = ports.AllocateTestServerPort()
-      self._ForwardPort(
+      self._ForwardPorts(
           [(self.test_server_spawner_port, self.test_server_spawner_port)])
       self._spawning_server = SpawningServer(self.test_server_spawner_port,
                                              self.adb,
                                              self.tool,
-                                             self._forwarder,
                                              self.build_type)
       self._spawning_server.Start()
       server_ready, error_msg = ports.IsHttpServerConnectable(
diff --git a/build/android/pylib/base/shard.py b/build/android/pylib/base/shard.py
index 8c429f7..af02eed 100644
--- a/build/android/pylib/base/shard.py
+++ b/build/android/pylib/base/shard.py
@@ -9,7 +9,6 @@ import threading
 
 from pylib import android_commands
 from pylib import constants
-from pylib import forwarder
 from pylib.utils import reraiser_thread
 from pylib.utils import watchdog_timer
 
@@ -293,7 +292,6 @@ def ShardAndRunTests(runner_factory, devices, tests, build_type='Debug',
     return (base_test_result.TestRunResults(), constants.ERROR_EXIT_CODE)
 
   logging.info('Will run %d tests: %s', len(tests), str(tests))
-  forwarder.Forwarder.KillHost(build_type)
   runners = _CreateRunners(runner_factory, devices, setup_timeout)
   try:
     return _RunAllTests(runners, tests, num_retries, test_timeout)
@@ -302,5 +300,3 @@ def ShardAndRunTests(runner_factory, devices, tests, build_type='Debug',
       _TearDownRunners(runners, setup_timeout)
     except android_commands.errors.DeviceUnresponsiveError as e:
       logging.warning('Device unresponsive during TearDown: [%s]', e)
-    finally:
-      forwarder.Forwarder.KillHost(build_type)
diff --git a/build/android/pylib/chrome_test_server_spawner.py b/build/android/pylib/chrome_test_server_spawner.py
index 4fea07a..f9fcd30 100644
--- a/build/android/pylib/chrome_test_server_spawner.py
+++ b/build/android/pylib/chrome_test_server_spawner.py
@@ -23,6 +23,7 @@ import urlparse
 import constants
 import ports
 
+from pylib.forwarder import Forwarder
 
 # Path that are needed to import necessary modules when launching a testserver.
 os.environ['PYTHONPATH'] = os.environ.get('PYTHONPATH', '') + (':%s:%s:%s:%s:%s'
@@ -99,7 +100,7 @@ def _GetServerTypeCommandLine(server_type):
 class TestServerThread(threading.Thread):
   """A thread to run the test server in a separate process."""
 
-  def __init__(self, ready_event, arguments, adb, tool, forwarder, build_type):
+  def __init__(self, ready_event, arguments, adb, tool, build_type):
     """Initialize TestServerThread with the following argument.
 
     Args:
@@ -107,7 +108,6 @@ class TestServerThread(threading.Thread):
       arguments: dictionary of arguments to run the test server.
       adb: instance of AndroidCommands.
       tool: instance of runtime error detection tool.
-      forwarder: instance of Forwarder.
       build_type: 'Release' or 'Debug'.
     """
     threading.Thread.__init__(self)
@@ -122,7 +122,6 @@ class TestServerThread(threading.Thread):
     self.is_ready = False
     self.host_port = self.arguments['port']
     assert isinstance(self.host_port, int)
-    self._test_server_forwarder = forwarder
     # The forwarder device port now is dynamically allocated.
     self.forwarder_device_port = 0
     # Anonymous pipe in order to get port info from test server.
@@ -240,11 +239,10 @@ class TestServerThread(threading.Thread):
       else:
         self.is_ready = _CheckPortStatus(self.host_port, True)
     if self.is_ready:
-      self._test_server_forwarder.Run([(0, self.host_port)], self.tool)
+      Forwarder.Map([(0, self.host_port)], self.adb, self.build_type, self.tool)
       # Check whether the forwarder is ready on the device.
       self.is_ready = False
-      device_port = self._test_server_forwarder.DevicePortForHostPort(
-          self.host_port)
+      device_port = Forwarder.DevicePortForHostPort(self.host_port)
       if device_port and _CheckDevicePortStatus(self.adb, device_port):
         self.is_ready = True
         self.forwarder_device_port = device_port
@@ -254,7 +252,7 @@ class TestServerThread(threading.Thread):
     _WaitUntil(lambda: self.stop_flag, max_attempts=sys.maxint)
     if self.process.poll() is None:
       self.process.kill()
-    self._test_server_forwarder.UnmapDevicePort(self.forwarder_device_port)
+    Forwarder.UnmapDevicePort(self.forwarder_device_port, self.adb)
     self.process = None
     self.is_ready = False
     if self.pipe_out:
@@ -324,7 +322,6 @@ class SpawningServerRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
         json.loads(test_server_argument_json),
         self.server.adb,
         self.server.tool,
-        self.server.forwarder,
         self.server.build_type)
     self.server.test_server_instance.setDaemon(True)
     self.server.test_server_instance.start()
@@ -392,14 +389,12 @@ class SpawningServerRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
 class SpawningServer(object):
   """The class used to start/stop a http server."""
 
-  def __init__(self, test_server_spawner_port, adb, tool, forwarder,
-               build_type):
+  def __init__(self, test_server_spawner_port, adb, tool, build_type):
     logging.info('Creating new spawner on port: %d.', test_server_spawner_port)
     self.server = BaseHTTPServer.HTTPServer(('', test_server_spawner_port),
                                             SpawningServerRequestHandler)
     self.server.adb = adb
     self.server.tool = tool
-    self.server.forwarder = forwarder
     self.server.test_server_instance = None
     self.server.build_type = build_type
 
diff --git a/build/android/pylib/forwarder.py b/build/android/pylib/forwarder.py
index 5bdb1cd..adf538c 100644
--- a/build/android/pylib/forwarder.py
+++ b/build/android/pylib/forwarder.py
@@ -2,24 +2,49 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
+import fcntl
 import logging
 import os
+import psutil
 import re
 import sys
-import threading
 import time
 
 import android_commands
 import cmd_helper
 import constants
 
-from pylib import pexpect
+from pylib import valgrind_tools
 
 
 def _MakeBinaryPath(build_type, binary_name):
   return os.path.join(cmd_helper.OutDirectory.get(), build_type, binary_name)
 
 
+def _GetProcessStartTime(pid):
+  return psutil.Process(pid).create_time
+
+
+class _FileLock(object):
+  """With statement-aware implementation of a file lock.
+
+  File locks are needed for cross-process synchronization when the
+  multiprocessing Python module is used.
+  """
+  def __init__(self, path):
+    self._path = path
+
+  def __enter__(self):
+    self._fd = os.open(self._path, os.O_RDONLY | os.O_CREAT)
+    if self._fd < 0:
+      raise Exception('Could not open file %s for reading' % self._path)
+    fcntl.flock(self._fd, fcntl.LOCK_EX)
+
+  def __exit__(self, type, value, traceback):
+    fcntl.flock(self._fd, fcntl.LOCK_UN)
+    os.close(self._fd)
+
+
 class Forwarder(object):
   """Thread-safe class to manage port forwards from the device to the host."""
 
@@ -28,28 +53,18 @@ class Forwarder(object):
   _DEVICE_FORWARDER_PATH = (constants.TEST_EXECUTABLE_DIR +
                             '/forwarder/device_forwarder')
   _LD_LIBRARY_PATH = 'LD_LIBRARY_PATH=%s' % _DEVICE_FORWARDER_FOLDER
+  _LOCK_PATH = '/tmp/chrome.forwarder.lock'
+  _MULTIPROCESSING_ENV_VAR = 'CHROME_FORWARDER_USE_MULTIPROCESSING'
 
-  def __init__(self, adb, build_type):
-    """Forwards TCP ports on the device back to the host.
-
-    Works like adb forward, but in reverse.
+  _instance = None
 
-    Args:
-      adb: Instance of AndroidCommands for talking to the device.
-      build_type: 'Release' or 'Debug'.
-    """
-    assert build_type in ('Release', 'Debug')
-    self._adb = adb
-    self._device_to_host_port_map = dict()
-    self._host_to_device_port_map = dict()
-    self._device_initialized = False
-    self._host_adb_control_port = 0
-    self._lock = threading.Lock()
-    self._host_forwarder_path = _MakeBinaryPath(build_type, 'host_forwarder')
-    self._device_forwarder_path_on_host = os.path.join(
-        cmd_helper.OutDirectory.get(), build_type, 'forwarder_dist')
+  @staticmethod
+  def UseMultiprocessing():
+    """Tells the forwarder that multiprocessing is used."""
+    os.environ[Forwarder._MULTIPROCESSING_ENV_VAR] = '1'
 
-  def Run(self, port_pairs, tool):
+  @staticmethod
+  def Map(port_pairs, adb, build_type='Debug', tool=None):
     """Runs the forwarder.
 
     Args:
@@ -58,24 +73,29 @@ class Forwarder(object):
                  port will by dynamically assigned on the device. You can
                  get the number of the assigned port using the
                  DevicePortForHostPort method.
+      adb: An AndroidCommands instance.
       tool: Tool class to use to get wrapper, if necessary, for executing the
             forwarder (see valgrind_tools.py).
 
     Raises:
       Exception on failure to forward the port.
     """
-    with self._lock:
-      self._InitDeviceLocked(tool)
-      host_name = '127.0.0.1'
+    if not tool:
+      tool = valgrind_tools.CreateTool(None, adb)
+    with _FileLock(Forwarder._LOCK_PATH):
+      instance = Forwarder._GetInstanceLocked(build_type, tool)
+      instance._InitDeviceLocked(adb, tool)
+
+      device_serial = adb.Adb().GetSerialNumber()
       redirection_commands = [
-          ['--serial-id=' + self._adb.Adb().GetSerialNumber(), '--map',
-           str(device), str(host)] for device, host in port_pairs]
+          ['--serial-id=' + device_serial, '--map', str(device),
+           str(host)] for device, host in port_pairs]
       logging.info('Forwarding using commands: %s', redirection_commands)
 
       for redirection_command in redirection_commands:
         try:
           (exit_code, output) = cmd_helper.GetCmdStatusAndOutput(
-              [self._host_forwarder_path] + redirection_command)
+              [instance._host_forwarder_path] + redirection_command)
         except OSError as e:
           if e.errno == 2:
             raise Exception('Unable to start host forwarder. Make sure you have'
@@ -83,83 +103,203 @@ class Forwarder(object):
           else: raise
         if exit_code != 0:
           raise Exception('%s exited with %d:\n%s' % (
-              self._host_forwarder_path, exit_code, '\n'.join(output)))
+              instance._host_forwarder_path, exit_code, '\n'.join(output)))
         tokens = output.split(':')
         if len(tokens) != 2:
           raise Exception(('Unexpected host forwarder output "%s", ' +
                           'expected "device_port:host_port"') % output)
         device_port = int(tokens[0])
         host_port = int(tokens[1])
-        self._device_to_host_port_map[device_port] = host_port
-        self._host_to_device_port_map[host_port] = device_port
+        serial_with_port = (device_serial, device_port)
+        instance._device_to_host_port_map[serial_with_port] = host_port
+        instance._host_to_device_port_map[host_port] = serial_with_port
         logging.info('Forwarding device port: %d to host port: %d.',
                      device_port, host_port)
 
-  def _InitDeviceLocked(self, tool):
-    """Initializes the device forwarder process (only once)."""
-    if self._device_initialized:
-      return
-    self._adb.PushIfNeeded(
-        self._device_forwarder_path_on_host,
-        Forwarder._DEVICE_FORWARDER_FOLDER)
-    (exit_code, output) = self._adb.GetShellCommandStatusAndOutput(
-        '%s %s %s' % (Forwarder._LD_LIBRARY_PATH, tool.GetUtilWrapper(),
-                      Forwarder._DEVICE_FORWARDER_PATH))
-    if exit_code != 0:
-      raise Exception(
-          'Failed to start device forwarder:\n%s' % '\n'.join(output))
-    self._device_initialized = True
-
-  def UnmapDevicePort(self, device_port):
+  @staticmethod
+  def UnmapDevicePort(device_port, adb):
     """Unmaps a previously forwarded device port.
 
     Args:
-      device_port: A previously forwarded port (through Run()).
+      adb: An AndroidCommands instance.
+      device_port: A previously forwarded port (through Map()).
+    """
+    with _FileLock(Forwarder._LOCK_PATH):
+      Forwarder._UnmapDevicePortLocked(device_port, adb)
+
+  @staticmethod
+  def UnmapAllDevicePorts(adb):
+    """Unmaps all the previously forwarded ports for the provided device.
+
+    Args:
+      adb: An AndroidCommands instance.
+      port_pairs: A list of tuples (device_port, host_port) to unmap.
+    """
+    with _FileLock(Forwarder._LOCK_PATH):
+      port_map = Forwarder._GetInstanceLocked(
+          None, None)._device_to_host_port_map
+      adb_serial = adb.Adb().GetSerialNumber()
+      for ((device_serial, device_port), _) in port_map:
+        if adb_serial == device_serial:
+          Forwarder._UnmapDevicePortLocked(device_port, adb)
+
+  @staticmethod
+  def DevicePortForHostPort(host_port):
+    """Returns the device port that corresponds to a given host port."""
+    with _FileLock(Forwarder._LOCK_PATH):
+      (device_serial, device_port) = Forwarder._GetInstanceLocked(
+          None, None)._host_to_device_port_map.get(host_port)
+      return device_port
+
+  @staticmethod
+  def _GetInstanceLocked(build_type, tool):
+    """Returns the singleton instance.
+
+    Note that the global lock must be acquired before calling this method.
+
+    Args:
+      build_type: 'Release' or 'Debug'
+      tool: Tool class to use to get wrapper, if necessary, for executing the
+            forwarder (see valgrind_tools.py).
     """
-    with self._lock:
-      self._UnmapDevicePortInternalLocked(device_port)
+    if not Forwarder._instance:
+      Forwarder._instance = Forwarder(build_type, tool)
+    return Forwarder._instance
+
+  def __init__(self, build_type, tool):
+    """Constructs a new instance of Forwarder.
 
-  def _UnmapDevicePortInternalLocked(self, device_port):
-    if not device_port in self._device_to_host_port_map:
+    Note that Forwarder is a singleton therefore this constructor should be
+    called only once.
+
+    Args:
+      build_type: 'Release' or 'Debug'
+      tool: Tool class to use to get wrapper, if necessary, for executing the
+            forwarder (see valgrind_tools.py).
+    """
+    assert not Forwarder._instance
+    self._build_type = build_type
+    self._tool = tool
+    self._initialized_devices = set()
+    self._device_to_host_port_map = dict()
+    self._host_to_device_port_map = dict()
+    self._host_forwarder_path = _MakeBinaryPath(
+        self._build_type, 'host_forwarder')
+    if not os.path.exists(self._host_forwarder_path):
+      self._build_type = 'Release' if self._build_type == 'Debug' else 'Debug'
+      self._host_forwarder_path = _MakeBinaryPath(
+          self._build_type, 'host_forwarder')
+      assert os.path.exists(
+          self._host_forwarder_path), 'Please build forwarder2'
+    self._device_forwarder_path_on_host = os.path.join(
+        cmd_helper.OutDirectory.get(), self._build_type, 'forwarder_dist')
+    self._InitHostLocked()
+
+  @staticmethod
+  def _UnmapDevicePortLocked(device_port, adb):
+    """Internal method used by UnmapDevicePort().
+
+    Note that the global lock must be acquired before calling this method.
+    """
+    instance = Forwarder._GetInstanceLocked(None, None)
+    serial = adb.Adb().GetSerialNumber()
+    serial_with_port = (serial, device_port)
+    if not serial_with_port in instance._device_to_host_port_map:
+      logging.error('Trying to unmap non-forwarded port %d' % device_port)
       return
-    redirection_command = [
-        '--serial-id=' + self._adb.Adb().GetSerialNumber(), '--unmap',
-        str(device_port)]
+    redirection_command = ['--serial-id=' + serial, '--unmap', str(device_port)]
     (exit_code, output) = cmd_helper.GetCmdStatusAndOutput(
-        [self._host_forwarder_path] + redirection_command)
+        [instance._host_forwarder_path] + redirection_command)
     if exit_code != 0:
       logging.error('%s exited with %d:\n%s' % (
-          self._host_forwarder_path, exit_code, '\n'.join(output)))
-    host_port = self._device_to_host_port_map[device_port]
-    del self._device_to_host_port_map[device_port]
-    del self._host_to_device_port_map[host_port]
+          instance._host_forwarder_path, exit_code, '\n'.join(output)))
+    host_port = instance._device_to_host_port_map[serial_with_port]
+    del instance._device_to_host_port_map[serial_with_port]
+    del instance._host_to_device_port_map[host_port]
 
   @staticmethod
-  def KillHost(build_type='Debug'):
-    """Kills the forwarder process running on the host.
+  def _GetPidForLock():
+    """Returns the PID used for host_forwarder initialization.
+
+    In case multi-process sharding is used, the PID of the "sharder" is used.
+    The "sharder" is the initial process that forks that is the parent process.
+    By default, multi-processing is not used. In that case the PID of the
+    current process is returned.
+    """
+    use_multiprocessing = Forwarder._MULTIPROCESSING_ENV_VAR in os.environ
+    return os.getppid() if use_multiprocessing else os.getpid()
+
+  def _InitHostLocked(self):
+    """Initializes the host forwarder daemon.
+
+    Note that the global lock must be acquired before calling this method. This
+    method kills any existing host_forwarder process that could be stale.
+    """
+    # See if the host_forwarder daemon was already initialized by a concurrent
+    # process or thread (in case multi-process sharding is not used).
+    pid_for_lock = Forwarder._GetPidForLock()
+    fd = os.open(Forwarder._LOCK_PATH, os.O_RDWR | os.O_CREAT)
+    with os.fdopen(fd, 'r+') as pid_file:
+      pid_with_start_time = pid_file.readline()
+      if pid_with_start_time:
+        (pid, process_start_time) = pid_with_start_time.split(':')
+        if pid == str(pid_for_lock):
+          if process_start_time == str(_GetProcessStartTime(pid_for_lock)):
+            return
+      self._KillHostLocked()
+      pid_file.seek(0)
+      pid_file.write(
+          '%s:%s' % (pid_for_lock, str(_GetProcessStartTime(pid_for_lock))))
+
+  def _InitDeviceLocked(self, adb, tool):
+    """Initializes the device_forwarder daemon for a specific device (once).
+
+    Note that the global lock must be acquired before calling this method. This
+    method kills any existing device_forwarder daemon on the device that could
+    be stale, pushes the latest version of the daemon (to the device) and starts
+    it.
 
     Args:
-      build_type: 'Release' or 'Debug' (default='Debug')
+      adb: An AndroidCommands instance.
+      tool: Tool class to use to get wrapper, if necessary, for executing the
+            forwarder (see valgrind_tools.py).
+    """
+    device_serial = adb.Adb().GetSerialNumber()
+    if device_serial in self._initialized_devices:
+      return
+    Forwarder._KillDeviceLocked(adb, tool)
+    adb.PushIfNeeded(
+        self._device_forwarder_path_on_host,
+        Forwarder._DEVICE_FORWARDER_FOLDER)
+    (exit_code, output) = adb.GetShellCommandStatusAndOutput(
+        '%s %s %s' % (Forwarder._LD_LIBRARY_PATH, tool.GetUtilWrapper(),
+                      Forwarder._DEVICE_FORWARDER_PATH))
+    if exit_code != 0:
+      raise Exception(
+          'Failed to start device forwarder:\n%s' % '\n'.join(output))
+    self._initialized_devices.add(device_serial)
+
+  def _KillHostLocked(self):
+    """Kills the forwarder process running on the host.
+
+    Note that the global lock must be acquired before calling this method.
     """
     logging.info('Killing host_forwarder.')
-    host_forwarder_path = _MakeBinaryPath(build_type, 'host_forwarder')
-    if not os.path.exists(host_forwarder_path):
-      host_forwarder_path = _MakeBinaryPath(
-          'Release' if build_type == 'Debug' else 'Debug', 'host_forwarder')
-    assert os.path.exists(host_forwarder_path), 'Please build forwarder2'
     (exit_code, output) = cmd_helper.GetCmdStatusAndOutput(
-        [host_forwarder_path, '--kill-server'])
+        [self._host_forwarder_path, '--kill-server'])
     if exit_code != 0:
       (exit_code, output) = cmd_helper.GetCmdStatusAndOutput(
-          ['pkill', 'host_forwarder'])
+          ['pkill', '-9', 'host_forwarder'])
       if exit_code != 0:
         raise Exception('%s exited with %d:\n%s' % (
-              host_forwarder_path, exit_code, '\n'.join(output)))
+              self._host_forwarder_path, exit_code, '\n'.join(output)))
 
   @staticmethod
-  def KillDevice(adb, tool):
+  def _KillDeviceLocked(adb, tool):
     """Kills the forwarder process running on the device.
 
+    Note that the global lock must be acquired before calling this method.
+
     Args:
       adb: Instance of AndroidCommands for talking to the device.
       tool: Wrapper tool (e.g. valgrind) that can be used to execute the device
@@ -180,14 +320,3 @@ class Forwarder(object):
       pids = adb.ExtractPid('device_forwarder')
       if pids:
         raise Exception('Timed out while killing device_forwarder')
-
-  def DevicePortForHostPort(self, host_port):
-    """Returns the device port that corresponds to a given host port."""
-    with self._lock:
-      return self._host_to_device_port_map.get(host_port)
-
-  def Close(self):
-    """Releases the previously forwarded ports."""
-    with self._lock:
-      for device_port in self._device_to_host_port_map.copy():
-        self._UnmapDevicePortInternalLocked(device_port)
diff --git a/build/android/pylib/host_driven/python_test_sharder.py b/build/android/pylib/host_driven/python_test_sharder.py
index cc74687..7ea3f6c 100644
--- a/build/android/pylib/host_driven/python_test_sharder.py
+++ b/build/android/pylib/host_driven/python_test_sharder.py
@@ -122,7 +122,7 @@ class PythonTestSharder(object):
     final_results = base_test_result.TestRunResults()
     tests_to_run = self.tests
 
-    Forwarder.KillHost()
+    Forwarder.UseMultiprocessing()
 
     for retry in xrange(self.retries):
       logging.warning('Try %d of %d', retry + 1, self.retries)
diff --git a/build/android/pylib/instrumentation/test_runner.py b/build/android/pylib/instrumentation/test_runner.py
index a2692fa..5fa1135 100644
--- a/build/android/pylib/instrumentation/test_runner.py
+++ b/build/android/pylib/instrumentation/test_runner.py
@@ -153,11 +153,13 @@ class TestRunner(base_test_runner.BaseTestRunner):
     http_server_ports = self.LaunchTestHttpServer(
         os.path.join(constants.DIR_SOURCE_ROOT), self._lighttp_port)
     if self.ports_to_forward:
-      self.StartForwarder([(port, port) for port in self.ports_to_forward])
+      self.ForwardPorts([(port, port) for port in self.ports_to_forward])
     self.flags.AddFlags(['--enable-test-intents'])
 
   def TearDown(self):
     """Cleans up the test harness and saves outstanding data from test run."""
+    if self.ports_to_forward:
+      self._UnmapPortPairs(self.ports_to_forward)
     super(TestRunner, self).TearDown()
 
   def TestSetup(self, test):
@@ -171,7 +173,7 @@ class TestRunner(base_test_runner.BaseTestRunner):
     self.tool.SetupEnvironment()
 
     # Make sure the forwarder is still running.
-    self.RestartHttpServerForwarderIfNecessary()
+    self._RestartHttpServerForwarderIfNecessary()
 
   def _IsPerfTest(self, test):
     """Determines whether a test is a performance test.