Move Python setup/tear down logic into Forwarder itself.

build/android/pylib/forwarder.py

 # 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.
 
+import fcntl
 import logging
 import os
 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)
 
 
+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._file = open(path, 'w')
+
+  def __enter__(self):
+    fcntl.flock(self._file, fcntl.LOCK_EX)

[bulach, 2013/07/12 11:48:26]

hmm, I think this needs:
fcntl.LOCK_EX | fcntl.LOCK_NB

the scenario I'm thinking is:
1) first shard starts, acquires the ex-lock.
2) second shard starts, blocks until first shard ends.
I guess it works fine, but entirely breaks the idea of sharding :)

unfortunately, I think we need to:
- shard N comes along, tries to acquire the lock.
-- if it does NOT acquire the lock, life is good: a live-shard has already
initialized, happily proceed!
-- it it DOES acquire the lock
---- check the content of the file.
---- if it's NOT the same PPID, proceed with the first-time initialization.
---- if it IS the same PPID, stat the file and the ppid, and check:
-------- file AFTER ppid, life is good: a sibling shard has initialized, just
proceed.
-------- file BEFORE ppid: unluckiest case! but doing first-time initialization
here should be safe.

[frankf, 2013/07/12 21:56:56]

Are you suggesting the first process holds the lock during its entire lifetime?
Why can't you do a blocking exclusive lock, but then release it after you update
the file?

On 2013/07/12 11:48:26, bulach wrote:

[bulach, 2013/07/15 09:32:30]

afaict, 48 below does hold the EX lock during the entire lifetime, no? because
it's blocking, it serializes the shards. :)

but yeah, holding an EX lock for shorter is indeed much better: on the cases I
mentioned above, the very first one would fail if the first shard is alive and
has acquired the lock but didn't fully initialize yet :(

So something like:
- acquire the EX lock
-- read the ppid
-- if empty, or ppid not the same, or older than ppid:
---- proceed with first-time initialization
-- write the ppid
- release the EX lock

This would protect the shards from:
1) first shard initializes correctly, even if previous runs have failed.
2) other shards won't reset anything (even if the first one terminates earlier).
3) two shards will be serialized only during forwarder initialization.


would that work?

[pliard, 2013/07/15 09:44:29]

Wouldn't the lock be held only in a with statement? Line 48 should only open the
file (that was my intention at least :)). The critical sections should be fast
and I would not expect any contention. Therefore I don't think this would impact
sharding performance especially if we don't forward thousands of ports every
second :) What do you think?

[bulach, 2013/07/15 10:20:03]

of course! you're right, the lock would only take place on "with".. my bad,
sorry... :)

however... what happens once the Nth shard comes along after the first one (that
acquired the lock and killed the host) has terminated? wouldn't it kill the
forwarder, despite it still being potentially in use for [1, N-1] other shards?

[pliard, 2013/07/15 10:51:11]

If we use the locked file just as you said (i.e. including the ppid and the
process start time), we would be covered, right? :)

To be clear, the current patch set doesn't contain yet a working version (since
it doesn't deal with ppid...) but this is planned :)

+
+  def __exit__(self, type, value, traceback):
+    fcntl.flock(self._file, fcntl.LOCK_UN)
+
+
 class Forwarder(object):
   """Thread-safe class to manage port forwards from the device to the host."""
 
   _DEVICE_FORWARDER_FOLDER = (constants.TEST_EXECUTABLE_DIR +
                               '/forwarder/')
   _DEVICE_FORWARDER_PATH = (constants.TEST_EXECUTABLE_DIR +
                             '/forwarder/device_forwarder')
   _LD_LIBRARY_PATH = 'LD_LIBRARY_PATH=%s' % _DEVICE_FORWARDER_FOLDER
 
-  def __init__(self, adb, build_type):
-    """Forwards TCP ports on the device back to the host.
+  _lock = _FileLock('/tmp/chrome.forwarder.lock')
+  _instance = None
 
-    Works like adb forward, but in reverse.
-
-    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')
-
-  def Run(self, port_pairs, tool):
+  @staticmethod
+  def Map(port_pairs, adb, build_type='Debug', tool=None):
     """Runs the forwarder.
 
     Args:
       port_pairs: A list of tuples (device_port, host_port) to forward. Note
                  that you can specify 0 as a device_port, in which case a
                  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 Forwarder._lock:
+      instance = Forwarder._GetInstanceLocked(build_type, tool)
+      instance._InitDeviceOnceLocked(adb, tool)
+
       redirection_commands = [
-          ['--serial-id=' + self._adb.Adb().GetSerialNumber(), '--map',
+          ['--serial-id=' + adb.Adb().GetSerialNumber(), '--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'
                             ' built host_forwarder.')
           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
+        instance._device_to_host_port_map[device_port] = host_port
+        instance._host_to_device_port_map[host_port] = device_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:
+  @staticmethod
+  def UnmapDevicePort(device_port, adb):
+    """Unmaps a previously forwarded device port.
+
+    Args:
+      adb: An AndroidCommands instance.
+      device_port: A previously forwarded port (through Map()).
+    """
+    with Forwarder._lock:
+      instance = Forwarder._GetInstanceLocked(None, None)
+      if not device_port in instance._device_to_host_port_map:
+        return
+      redirection_command = [
+          '--serial-id=' + adb.Adb().GetSerialNumber(), '--unmap',
+          str(device_port)]
+      (exit_code, output) = cmd_helper.GetCmdStatusAndOutput(
+          [instance._host_forwarder_path] + redirection_command)
+      if exit_code != 0:
+        logging.error('%s exited with %d:\n%s' % (
+            instance._host_forwarder_path, exit_code, '\n'.join(output)))
+      host_port = instance._device_to_host_port_map[device_port]
+      del instance._device_to_host_port_map[device_port]
+      del instance._host_to_device_port_map[host_port]
+
+  @staticmethod
+  def UnmapPorts(port_pairs, adb):
+    """Unmaps previously forwarded ports.
+
+    Args:
+      adb: An AndroidCommands instance.
+      port_pairs: A list of tuples (device_port, host_port) to unmap.
+    """
+    for (device_port, _) in port_pairs:
+      Forwarder.UnmapDevicePort(device_port, adb)
+
+  @staticmethod
+  def DevicePortForHostPort(host_port):
+    """Returns the device port that corresponds to a given host port."""
+    with Forwarder._lock:
+      return Forwarder._GetInstanceLocked(
+          None, None)._host_to_device_port_map.get(host_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).
+    """
+    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.
+
+    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._InitHostOnceLocked()
+    self._device_forwarder_path_on_host = os.path.join(
+        cmd_helper.OutDirectory.get(), self._build_type, 'forwarder_dist')
+
+  def _InitHostOnceLocked(self):
+    """Initializes the host forwarder daemon.
+
+    Note that the global lock must be acquired before calling this method. This
+    method determines the host_forwarder binary location and kills any existing
+    host_forwarder process that could be stale.
+    """
+    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._KillHostLocked()
+
+  def _InitDeviceOnceLocked(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:
+      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
-    self._adb.PushIfNeeded(
+    Forwarder._KillDeviceLocked(adb, tool)
+    adb.PushIfNeeded(
         self._device_forwarder_path_on_host,
         Forwarder._DEVICE_FORWARDER_FOLDER)
-    (exit_code, output) = self._adb.GetShellCommandStatusAndOutput(
+    (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._device_initialized = True
+    self._initialized_devices.add(device_serial)
 
-  def UnmapDevicePort(self, device_port):
-    """Unmaps a previously forwarded device port.
+  def _KillHostLocked(self):
+    """Kills the forwarder process running on the host.
 
-    Args:
-      device_port: A previously forwarded port (through Run()).
+    Note that the global lock must be acquired before calling this method.
     """
-    with self._lock:
-      self._UnmapDevicePortInternalLocked(device_port)
-
-  def _UnmapDevicePortInternalLocked(self, device_port):
-    if not device_port in self._device_to_host_port_map:
-      return
-    redirection_command = [
-        '--serial-id=' + self._adb.Adb().GetSerialNumber(), '--unmap',
-        str(device_port)]
+    logging.info('Killing host_forwarder.')
     (exit_code, output) = cmd_helper.GetCmdStatusAndOutput(
-        [self._host_forwarder_path] + redirection_command)
+        [self._host_forwarder_path, '--kill-server'])
     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]
+      (exit_code, output) = cmd_helper.GetCmdStatusAndOutput(
+          ['pkill', '-9', 'host_forwarder'])
+      if exit_code != 0:
+        raise Exception('%s exited with %d:\n%s' % (
+              self._host_forwarder_path, exit_code, '\n'.join(output)))
 
   @staticmethod
-  def KillHost(build_type='Debug'):
-    """Kills the forwarder process running on the host.
+  def _KillDeviceLocked(adb, tool):
+    """Kills the forwarder process running on the device.
 
-    Args:
-      build_type: 'Release' or 'Debug' (default='Debug')
-    """
-    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'])
-    if exit_code != 0:
-      (exit_code, output) = cmd_helper.GetCmdStatusAndOutput(
-          ['pkill', 'host_forwarder'])
-      if exit_code != 0:
-        raise Exception('%s exited with %d:\n%s' % (
-              host_forwarder_path, exit_code, '\n'.join(output)))
-
-  @staticmethod
-  def KillDevice(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
             forwarder (see valgrind_tools.py).
     """
     logging.info('Killing device_forwarder.')
     if not adb.FileExistsOnDevice(Forwarder._DEVICE_FORWARDER_PATH):
       return
     (exit_code, output) = adb.GetShellCommandStatusAndOutput(
         '%s %s --kill-server' % (tool.GetUtilWrapper(),
                                  Forwarder._DEVICE_FORWARDER_PATH))
     # TODO(pliard): Remove the following call to KillAllBlocking() when we are
     # sure that the old version of device_forwarder (not supporting
     # 'kill-server') is not running on the bots anymore.
     timeout_sec = 5
     processes_killed = adb.KillAllBlocking('device_forwarder', timeout_sec)
     if not processes_killed:
       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)