Upstream test sharder.

On chromium for android, we shard tests by splitting the set of tests
and running on multiple connected devices.

BUG=
TEST=build/android/run_tests.py


Review URL: http://codereview.chromium.org/9494007

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

5 files changed, 237 insertions, 55 deletions

diff --git a/build/android/base_test_runner.py b/build/android/base_test_runner.py
index bb0316b..33e0741 100644
--- a/build/android/base_test_runner.py
+++ b/build/android/base_test_runner.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2011 The Chromium Authors. All rights reserved.
+# 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.
 
@@ -19,12 +19,17 @@ TEST_SYNC_SERVER_PORT = 8003
 
 
 class BaseTestRunner(object):
-  """Base class for running tests on a single device."""
+  """Base class for running tests on a single device.
 
-  def __init__(self, device):
+  A subclass should implement RunTests() with no parameter, so that calling
+  the Run() method will set up tests, run them and tear them down.
+  """
+
+  def __init__(self, device, shard_index):
     """
       Args:
         device: Tests will run on the device of this ID.
+        shard_index: Index number of the shard on which the test suite will run.
     """
     self.device = device
     self.adb = android_commands.AndroidCommands(device=device)
@@ -41,17 +46,28 @@ class BaseTestRunner(object):
     self.forwarder_base_url = ('http://localhost:%d' %
         self._forwarder_device_port)
     self.flags = FlagChanger(self.adb)
+    self.shard_index = shard_index
 
-  def RunTests(self):
-    # TODO(bulach): this should actually do SetUp / RunTestsInternal / TearDown.
-    # Refactor the various subclasses to expose a RunTestsInternal without
-    # any params.
-    raise NotImplementedError
+  def Run(self):
+    """Calls subclass functions to set up tests, run them and tear them down.
+
+    Returns:
+      Test results returned from RunTests().
+    """
+    self.SetUp()
+    try:
+      return self.RunTests()
+    finally:
+      self.TearDown()
 
   def SetUp(self):
     """Called before tests run."""
     pass
 
+  def RunTests(self):
+    """Runs the tests. Need to be overridden."""
+    raise NotImplementedError
+
   def TearDown(self):
     """Called when tests finish running."""
     self.ShutdownHelperToolsForTestSuite()
diff --git a/build/android/base_test_sharder.py b/build/android/base_test_sharder.py
new file mode 100644
index 0000000..b7b1a34
--- /dev/null
+++ b/build/android/base_test_sharder.py
@@ -0,0 +1,108 @@
+#!/usr/bin/python
+# 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 logging
+import multiprocessing
+
+from test_result import *
+
+
+def _ShardedTestRunnable(test):
+  """Standalone function needed by multiprocessing.Pool."""
+  log_format = '[' + test.device + '] # %(asctime)-15s: %(message)s'
+  if logging.getLogger().handlers:
+    logging.getLogger().handlers[0].setFormatter(logging.Formatter(log_format))
+  else:
+    logging.basicConfig(format=log_format)
+  return test.Run()
+
+
+def SetTestsContainer(tests_container):
+  """Sets tests container.
+
+  multiprocessing.Queue can't be pickled across processes, so we need to set
+  this as a 'global', per process, via multiprocessing.Pool.
+  """
+  BaseTestSharder.tests_container = tests_container
+
+
+class BaseTestSharder(object):
+  """Base class for sharding tests across multiple devices.
+
+  Args:
+    attached_devices: A list of attached devices.
+  """
+  # See more in SetTestsContainer.
+  tests_container = None
+
+  def __init__(self, attached_devices):
+    self.attached_devices = attached_devices
+    self.retries = 1
+    self.tests = []
+
+  def CreateShardedTestRunner(self, device, index):
+    """Factory function to create a suite-specific test runner.
+
+    Args:
+      device: Device serial where this shard will run
+      index: Index of this device in the pool.
+
+    Returns:
+      An object of BaseTestRunner type (that can provide a "Run()" method).
+    """
+    pass
+
+  def SetupSharding(self, tests):
+    """Called before starting the shards."""
+    pass
+
+  def OnTestsCompleted(self, test_runners, test_results):
+    """Notifies that we completed the tests."""
+    pass
+
+  def RunShardedTests(self):
+    """Runs the tests in all connected devices.
+
+    Returns:
+      A TestResults object.
+    """
+    logging.warning('*' * 80)
+    logging.warning('Sharding in ' + str(len(self.attached_devices)) +
+                    ' devices.')
+    logging.warning('Note that the output is not synchronized.')
+    logging.warning('Look for the "Final result" banner in the end.')
+    logging.warning('*' * 80)
+    final_results = TestResults()
+    for retry in xrange(self.retries):
+      logging.warning('Try %d of %d' % (retry + 1, self.retries))
+      self.SetupSharding(self.tests)
+      test_runners = []
+      for index, device in enumerate(self.attached_devices):
+        logging.warning('*' * 80)
+        logging.warning('Creating shard %d for %s' % (index, device))
+        logging.warning('*' * 80)
+        test_runner = self.CreateShardedTestRunner(device, index)
+        test_runners += [test_runner]
+      logging.warning('Starting...')
+      pool = multiprocessing.Pool(len(self.attached_devices),
+                                  SetTestsContainer,
+                                  [BaseTestSharder.tests_container])
+      results_lists = pool.map(_ShardedTestRunnable, test_runners)
+      test_results = TestResults.FromTestResults(results_lists)
+      if retry == self.retries - 1:
+        all_passed = final_results.ok + test_results.ok
+        final_results = test_results
+        final_results.ok = all_passed
+        break
+      else:
+        final_results.ok += test_results.ok
+        self.tests = []
+        for t in test_results.GetAllBroken():
+          self.tests += [t.name]
+        if not self.tests:
+          break
+    self.OnTestsCompleted(test_runners, final_results)
+    return final_results
diff --git a/build/android/emulator.py b/build/android/emulator.py
index 5abfadc..19cfe03 100755
--- a/build/android/emulator.py
+++ b/build/android/emulator.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-# Copyright (c) 2011 The Chromium Authors. All rights reserved.
+# 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.
 
@@ -136,12 +136,14 @@ class Emulator(object):
     port = _GetAvailablePort()
     return ('emulator-%d' % port, port)
 
-  def Launch(self):
-    """Launches the emulator and waits for package manager to startup.
+  def Launch(self, kill_all_emulators):
+    """Launches the emulator asynchronously. Call ConfirmLaunch() to ensure the
+    emulator is ready for use.
 
     If fails, an exception will be raised.
     """
-    _KillAllEmulators()  # just to be sure
+    if kill_all_emulators:
+      _KillAllEmulators()  # just to be sure
     if not self.fast_and_loose:
       self._AggressiveImageCleanup()
     (self.device, port) = self._DeviceName()
@@ -166,7 +168,6 @@ class Emulator(object):
     self.popen = subprocess.Popen(args=emulator_command,
                                   stderr=subprocess.STDOUT)
     self._InstallKillHandler()
-    self._ConfirmLaunch()
 
   def _AggressiveImageCleanup(self):
     """Aggressive cleanup of emulator images.
@@ -186,7 +187,7 @@ class Emulator(object):
         logging.info('Deleting emulator image %s', full_name)
         os.unlink(full_name)
 
-  def _ConfirmLaunch(self, wait_for_boot=False):
+  def ConfirmLaunch(self, wait_for_boot=False):
     """Confirm the emulator launched properly.
 
     Loop on a wait-for-device with a very small timeout.  On each
diff --git a/build/android/run_tests.py b/build/android/run_tests.py
index f945bee..fd98afe 100755
--- a/build/android/run_tests.py
+++ b/build/android/run_tests.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-# Copyright (c) 2011 The Chromium Authors. All rights reserved.
+# 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.
 
@@ -47,7 +47,9 @@ loaded. We don't care about the rare testcases which succeeded on emuatlor, but
 failed on device.
 """
 
+import fnmatch
 import logging
+import multiprocessing
 import os
 import re
 import subprocess
@@ -55,6 +57,7 @@ import sys
 import time
 
 import android_commands
+from base_test_sharder import BaseTestSharder
 import cmd_helper
 import debug_info
 import emulator
@@ -140,7 +143,7 @@ class Xvfb(object):
 
 def RunTests(device, test_suite, gtest_filter, test_arguments, rebaseline,
              timeout, performance_test, cleanup_test_files, tool,
-             log_dump_name, fast_and_loose=False, annotate=False):
+             log_dump_name, annotate=False):
   """Runs the tests.
 
   Args:
@@ -154,8 +157,6 @@ def RunTests(device, test_suite, gtest_filter, test_arguments, rebaseline,
     cleanup_test_files: Whether or not to cleanup test files on device.
     tool: Name of the Valgrind tool.
     log_dump_name: Name of log dump file.
-    fast_and_loose: should we go extra-fast but sacrifice stability
-      and/or correctness?  Intended for quick cycle testing; not for bots!
     annotate: should we print buildbot-style annotations?
 
   Returns:
@@ -183,9 +184,8 @@ def RunTests(device, test_suite, gtest_filter, test_arguments, rebaseline,
       print '@@@BUILD_STEP Test suite %s@@@' % os.path.basename(t)
     test = SingleTestRunner(device, t, gtest_filter, test_arguments,
                             timeout, rebaseline, performance_test,
-                            cleanup_test_files, tool, not not log_dump_name,
-                            fast_and_loose=fast_and_loose)
-    test.RunTests()
+                            cleanup_test_files, tool, 0, not not log_dump_name)
+    test.Run()
 
     results += [test.test_results]
     # Collect debug info.
@@ -211,6 +211,61 @@ def RunTests(device, test_suite, gtest_filter, test_arguments, rebaseline,
   return TestResults.FromTestResults(results)
 
 
+class TestSharder(BaseTestSharder):
+  """Responsible for sharding the tests on the connected devices."""
+
+  def __init__(self, attached_devices, test_suite, gtest_filter,
+               test_arguments, timeout, rebaseline, performance_test,
+               cleanup_test_files, tool):
+    BaseTestSharder.__init__(self, attached_devices)
+    self.test_suite = test_suite
+    self.test_suite_basename = os.path.basename(test_suite)
+    self.gtest_filter = gtest_filter
+    self.test_arguments = test_arguments
+    self.timeout = timeout
+    self.rebaseline = rebaseline
+    self.performance_test = performance_test
+    self.cleanup_test_files = cleanup_test_files
+    self.tool = tool
+    test = SingleTestRunner(self.attached_devices[0], test_suite, gtest_filter,
+                            test_arguments, timeout, rebaseline,
+                            performance_test, cleanup_test_files, tool, 0)
+    all_tests = test.test_package.GetAllTests()
+    if not rebaseline:
+      disabled_list = test.GetDisabledTests()
+      # Only includes tests that do not have any match in the disabled list.
+      all_tests = filter(lambda t:
+                         not any([fnmatch.fnmatch(t, disabled_pattern)
+                                  for disabled_pattern in disabled_list]),
+                         all_tests)
+    self.tests = all_tests
+
+  def CreateShardedTestRunner(self, device, index):
+    """Creates a suite-specific test runner.
+
+    Args:
+      device: Device serial where this shard will run.
+      index: Index of this device in the pool.
+
+    Returns:
+      A SingleTestRunner object.
+    """
+    shard_size = len(self.tests) / len(self.attached_devices)
+    shard_test_list = self.tests[index * shard_size : (index + 1) * shard_size]
+    test_filter = ':'.join(shard_test_list)
+    return SingleTestRunner(device, self.test_suite,
+                            test_filter, self.test_arguments, self.timeout,
+                            self.rebaseline, self.performance_test,
+                            self.cleanup_test_files, self.tool, index)
+
+  def OnTestsCompleted(self, test_runners, test_results):
+    """Notifies that we completed the tests."""
+    test_results.LogFull()
+    if test_results.failed and self.rebaseline:
+      test_runners[0].UpdateFilter(test_results.failed)
+
+
+
 def _RunATestSuite(options):
   """Run a single test suite.
 
@@ -225,14 +280,19 @@ def _RunATestSuite(options):
     0 if successful, number of failing tests otherwise.
   """
   attached_devices = []
-  buildbot_emulator = None
+  buildbot_emulators = []
 
   if options.use_emulator:
-    t = TimeProfile('Emulator launch')
-    buildbot_emulator = emulator.Emulator(options.fast_and_loose)
-    buildbot_emulator.Launch()
-    t.Stop()
-    attached_devices.append(buildbot_emulator.device)
+    for n in range(options.use_emulator):
+      t = TimeProfile('Emulator launch %d' % n)
+      buildbot_emulator = emulator.Emulator(options.fast_and_loose)
+      buildbot_emulator.Launch(kill_all_emulators=n == 0)
+      t.Stop()
+      buildbot_emulators.append(buildbot_emulator)
+      attached_devices.append(buildbot_emulator.device)
+    # Wait for all emulators to become available.
+    map(lambda buildbot_emulator:buildbot_emulator.ConfirmLaunch(),
+        buildbot_emulators)
   else:
     attached_devices = android_commands.GetAttachedDevices()
 
@@ -240,16 +300,24 @@ def _RunATestSuite(options):
     logging.critical('A device must be attached and online.')
     return 1
 
-  test_results = RunTests(attached_devices[0], options.test_suite,
+  if (len(attached_devices) > 1 and options.test_suite and
+      not options.gtest_filter and not options.performance_test):
+    sharder = TestSharder(attached_devices, options.test_suite,
                           options.gtest_filter, options.test_arguments,
-                          options.rebaseline, options.timeout,
+                          options.timeout, options.rebaseline,
                           options.performance_test,
-                          options.cleanup_test_files, options.tool,
-                          options.log_dump,
-                          fast_and_loose=options.fast_and_loose,
-                          annotate=options.annotate)
-
-  if buildbot_emulator:
+                          options.cleanup_test_files, options.tool)
+    test_results = sharder.RunShardedTests()
+  else:
+    test_results = RunTests(attached_devices[0], options.test_suite,
+                            options.gtest_filter, options.test_arguments,
+                            options.rebaseline, options.timeout,
+                            options.performance_test,
+                            options.cleanup_test_files, options.tool,
+                            options.log_dump,
+                            annotate=options.annotate)
+
+  for buildbot_emulator in buildbot_emulators:
     buildbot_emulator.Shutdown()
 
   # Another chance if we timed out?  At this point It is safe(r) to
@@ -330,8 +398,8 @@ def main(argv):
                            'in where the test_suite exists.')
   option_parser.add_option('-e', '--emulator', dest='use_emulator',
                            help='Run tests in a new instance of emulator',
-                           action='store_true',
-                           default=False)
+                           type='int',
+                           default=0)
   option_parser.add_option('-x', '--xvfb', dest='use_xvfb',
                            action='store_true', default=False,
                            help='Use Xvfb around tests (ignored if not Linux)')
diff --git a/build/android/single_test_runner.py b/build/android/single_test_runner.py
index fa5195b..ef9971a 100644
--- a/build/android/single_test_runner.py
+++ b/build/android/single_test_runner.py
@@ -26,14 +26,15 @@ class SingleTestRunner(BaseTestRunner):
     performance_test: Whether or not performance test(s).
     cleanup_test_files: Whether or not to cleanup test files on device.
     tool: Name of the Valgrind tool.
+    shard_index: index number of the shard on which the test suite will run.
     dump_debug_info: Whether or not to dump debug information.
   """
 
   def __init__(self, device, test_suite, gtest_filter, test_arguments, timeout,
                rebaseline, performance_test, cleanup_test_files, tool,
-               dump_debug_info=False,
+               shard_index, dump_debug_info=False,
                fast_and_loose=False):
-    BaseTestRunner.__init__(self, device)
+    BaseTestRunner.__init__(self, device, shard_index)
     self._running_on_emulator = self.device.startswith('emulator')
     self._gtest_filter = gtest_filter
     self._test_arguments = test_arguments
@@ -265,8 +266,8 @@ class SingleTestRunner(BaseTestRunner):
                                                          failed_results),
                                                     list(failed_results))
 
-  def _RunTestsForSuiteInternal(self):
-    """Runs all tests (in rebaseline mode, run each test in isolation).
+  def RunTests(self):
+    """Runs all tests (in rebaseline mode, runs each test in isolation).
 
     Returns:
       A TestResults object.
@@ -279,6 +280,7 @@ class SingleTestRunner(BaseTestRunner):
                              ':'.join(['*.' + x + '*' for x in
                                      self.test_package.GetDisabledPrefixes()]))
       self.RunTestsWithFilter()
+    return self.test_results
 
   def SetUp(self):
     """Sets up necessary test enviroment for the test suite."""
@@ -296,7 +298,6 @@ class SingleTestRunner(BaseTestRunner):
 
   def TearDown(self):
     """Cleans up the test enviroment for the test suite."""
-    super(SingleTestRunner, self).TearDown()
     self.test_package.tool.CleanUpEnvironment()
     if self.test_package.cleanup_test_files:
       self.adb.RemovePushedFiles()
@@ -304,16 +305,4 @@ class SingleTestRunner(BaseTestRunner):
       self.dump_debug_info.StopRecordingLog()
     if self.test_package.performance_test:
       self.adb.TearDownPerformanceTest()
-
-  def RunTests(self):
-    """Runs the tests and cleans up the files once finished.
-
-    Returns:
-      A TestResults object.
-    """
-    self.SetUp()
-    try:
-      self._RunTestsForSuiteInternal()
-    finally:
-      self.TearDown()
-    return self.test_results
+    super(SingleTestRunner, self).TearDown()