[Android] Support generic parallel execution across devices.

build/android/pylib/device/device_utils.py

Index: build/android/pylib/device/device_utils.py
diff --git a/build/android/pylib/device/device_utils.py b/build/android/pylib/device/device_utils.py
index 1440961aa5ec8f31b0a8dc88aab4a945a5dec3a5..c76c7965c37488614e74bb09b2c294b89e54ff2f 100644
--- a/build/android/pylib/device/device_utils.py
+++ b/build/android/pylib/device/device_utils.py
@@ -9,51 +9,17 @@ Eventually, this will be based on adb_wrapper.
 """
 # pylint: disable=W0613
 
-import multiprocessing
-import os
-import sys
-
 import pylib.android_commands
 from pylib.device import adb_wrapper
 from pylib.device import decorators
-
-CHROME_SRC_DIR = os.path.abspath(
-    os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
-sys.path.append(os.path.join(
-    CHROME_SRC_DIR, 'third_party', 'android_testrunner'))
-import errors
+from pylib.device import device_errors
+from pylib.utils import reraiser_thread
+from pylib.utils import watchdog_timer
 
 _DEFAULT_TIMEOUT = 30
 _DEFAULT_RETRIES = 3
 
 
-# multiprocessing map_async requires a top-level function for pickle library.
-def RebootDeviceSafe(device):
-  """Reboot a device, wait for it to start, and squelch timeout exceptions."""
-  try:
-    DeviceUtils(device).old_interface.Reboot(True)
-  except errors.DeviceUnresponsiveError as e:
-    return e
-
-
-def RebootDevices():
-  """Reboot all attached and online devices."""
-  devices = pylib.android_commands.GetAttachedDevices()
-  print 'Rebooting: %s' % devices
-  if devices:
-    pool = multiprocessing.Pool(len(devices))
-    results = pool.map_async(RebootDeviceSafe, devices).get(99999)
-
-    for device, result in zip(devices, results):
-      if result:
-        print '%s failed to startup.' % device
-
-    if any(results):
-      print 'RebootDevices() Warning: %s' % results
-    else:
-      print 'Reboots complete.'
-
-
 @decorators.WithExplicitTimeoutAndRetries(
     _DEFAULT_TIMEOUT, _DEFAULT_RETRIES)
 def GetAVDs():
@@ -76,8 +42,102 @@ def RestartServer():
   pylib.android_commands.AndroidCommands().RestartAdbServer()
 
 
+class _ParallelExecutor(object):

[frankf, 2014/05/16 17:47:26]

this should be in its own module with tests

[frankf, 2014/05/16 17:47:26]

Also think about how to organize various modes of execution (See 
test_dispatcher.py for a work queue model).

[jbudorick, 2014/05/16 17:52:50]

I debated about the separate module thing. Opted not to at the time since it is
currently specific to DeviceUtils, but I'm ok with moving it out. Tests are in
this CL.

[craigdh, 2014/05/16 21:59:35]

This is probably worth sticking in utils/, probably with a module level doc
showing some simple examples so people can figure out how to use it :)

It might also be useful to have a mixin class that just has a 'parallel' object,
instead of counting on objects to follow that convention.

[jbudorick, 2014/05/21 16:42:48]

Done.

+  """ Allows parallel execution across a provided set of devices.

[frankf, 2014/05/16 17:47:26]

"""First line of docstinrg.

Additional comments
"""

Please address this everywhere.

[jbudorick, 2014/05/16 17:52:50]

In general -- what if there are no additional comments ...?

"""Only line of docstring."""

or

"""Only line of docstring.
"""

?

[frankf, 2014/05/16 18:41:27]

The former for single line docstring.


On 2014/05/16 17:52:50, jbudorick wrote:

+  """
+
+  def __init__(self, device_list=None):
+    if not device_list:
+      self._devices = [
+          DeviceUtils(s) for s in
+          pylib.android_commands.GetAttachedDevices()]
+    else:
+      device_list_type = self._typeOfContents(device_list)
+      if issubclass(device_list_type, (basestring,
+                                       pylib.android_commands.AndroidCommands,
+                                       adb_wrapper.AdbWrapper)):
+        self._devices = [DeviceUtils(d) for d in device_list]
+      else:
+        self._devices = device_list
+    if not len(self._devices):
+      # TODO(jbudorick) Should this be legal?
+      raise ValueError('No devices connected.')

[craigdh, 2014/05/16 21:59:35]

I don't think ValueError is the right exception to raise here.

[jbudorick, 2014/05/21 16:42:48]

What would you suggest? IndexError? A separate exception class?

[craigdh, 2014/05/21 16:50:13]

Maybe RuntimeError. Or just do an assert. ValueError would only really make
sense if a device was passed in that wasn't actually attached.

+    self._objs = self._devices
+    self._emulated_type = self._typeOfContents(self._objs)
+
+  @staticmethod
+  def _typeOfContents(objs):
+    assert(all([type(objs[0]) == type(o) for o in objs]))
+    return type(objs[0]) if len(objs) else type(None)
+
+  def __getattr__(self, name):
+    """ Emulate getting the |name| attribute of |self|.
+
+    Args:
+      name: The name of the attribute to retrieve.
+    """
+    r = _ParallelExecutor(self._devices)
+    r._objs = [getattr(o, name) for o in self._objs]
+
+    # pylint: disable=W0212
+    emulated_type = self._typeOfContents(r._objs)
+    if all([callable(o) for o in r._objs]):

[craigdh, 2014/05/16 21:59:35]

prefer generator expressions over list comprehensions if you just want an
iterable

[jbudorick, 2014/05/21 16:42:48]

Done.

+      r._emulated_type = self._emulated_type
+    else:
+      r._emulated_type = emulated_type
+    # pylint: enable=W0212
+    return r
+
+  def __getitem__(self, index):
+    """ Emulate getting the value of |self| at |index|.
+    """
+    r = _ParallelExecutor(self._devices)
+    r._objs = [o[index] for o in self._objs]
+    return r
+
+  def __call__(self, *args, **kwargs):
+    """ Emulate calling |self| with |args| and |kwargs|.
+    """
+    if not self._objs:
+      raise AttributeError('Nothing to call.')
+    for o in self._objs:
+      if not callable(o):
+        raise AttributeError("'%s' is not callable" % o.__name__)
+
+    r = _ParallelExecutor(self._devices)
+    r._objs = reraiser_thread.ReraiserThreadGroup(
+        [reraiser_thread.ReraiserThread(
+            o, args=args, kwargs=kwargs,
+            name='%s.%s' % (str(d), o.__name__))
+         for d, o in zip(self._devices, self._objs)])
+    r._objs.StartAll() # pylint: disable=W0212
+    return r
+
+  def get(self, timeout):
+    """ Get the return values from the most recent call.
+
+    Args:
+      timeout: The maximum number of seconds to wait for an individual
+               result to return.
+    Returns:
+      A list of the results, in order of the provided devices.
+    Raises:
+      Any exception raised by any of the called functions, or
+      CommandTimeoutError on timeout.
+    """
+    try:
+      if isinstance(self._objs, reraiser_thread.ReraiserThreadGroup):
+        self._objs.JoinAll()
+        self._objs = self._objs.Get(watchdog_timer.WatchdogTimer(timeout))
+      return self._objs
+    except reraiser_thread.TimeoutError as e:
+      raise device_errors.CommandTimeoutError(str(e))
+
+
 class DeviceUtils(object):
 
+  parallel = _ParallelExecutor
+
   def __init__(self, device):
     self.old_interface = None
     if isinstance(device, basestring):
@@ -89,3 +149,6 @@ class DeviceUtils(object):
     elif not device:
       self.old_interface = pylib.android_commands.AndroidCommands()
 
+  def __str__(self):
+    return self.old_interface.GetDevice()
+