[Andoid] Threaded TestRunner creation and SetUp and TearDown calls.

Create TestRunners only once and only call SetUp and TearDown once.

BUG=176325,168889
TEST=pylib/utils/raiser_thread_unittest.py, pylib/base/shard_unittest.py, run_tests.py

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

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

9 files changed, 421 insertions, 144 deletions

diff --git a/build/android/pylib/base/new_base_test_runner.py b/build/android/pylib/base/new_base_test_runner.py
index 4e7d344..67a7bda 100644
--- a/build/android/pylib/base/new_base_test_runner.py
+++ b/build/android/pylib/base/new_base_test_runner.py
@@ -65,31 +65,23 @@ class BaseTestRunner(object):
                              '%d:%d' % (self.test_server_spawner_port,
                                         self.test_server_port))
 
-  def Run(self, test):
-    """Calls subclass functions to set up test, run it and tear it down.
+  def RunTest(self, test):
+    """Runs a test. Needs to be overridden.
 
     Args:
-      test: A Test to run.
+      test: A test to run.
 
     Returns:
-      Test results returned from RunTest(test).
+      Tuple containing: (test_result.TestResults, tests to rerun or None)
     """
-    self.SetUp()
-    try:
-      return self.RunTest(test)
-    finally:
-      self.TearDown()
+    raise NotImplementedError
 
   def SetUp(self):
-    """Called before tests run."""
+    """Run once before all tests are run."""
     Forwarder.KillDevice(self.adb, self.tool)
 
-  def RunTest(self, test):
-    """Runs the tests. Needs to be overridden."""
-    raise NotImplementedError
-
   def TearDown(self):
-    """Called when tests finish running."""
+    """Run once after all tests are run."""
     self.ShutdownHelperToolsForTestSuite()
 
   def CopyTestData(self, test_data_paths, dest_dir):
diff --git a/build/android/pylib/base/shard.py b/build/android/pylib/base/shard.py
index b4e1281..fdca524 100644
--- a/build/android/pylib/base/shard.py
+++ b/build/android/pylib/base/shard.py
@@ -5,61 +5,154 @@
 """Implements test sharding logic."""
 
 import logging
-import sys
 import threading
 
 from pylib import android_commands
 from pylib import forwarder
+from pylib.utils import reraiser_thread
 
 import test_result
 
 
-class _Worker(threading.Thread):
-  """Runs tests from the test_queue using the given runner in a separate thread.
+class _Test(object):
+  """Holds a test with additional metadata."""
+  def __init__(self, test, tries=0):
+    """Initializes the _Test object.
 
-  Places results in the out_results.
+    Args:
+      test: the test.
+      tries: number of tries so far.
+    """
+    self.test = test
+    self.tries = tries
+
+
+class _TestCollection(object):
+  """A threadsafe collection of tests.
+
+  Args:
+    tests: list of tests to put in the collection.
   """
-  def __init__(self, runner, test_queue, out_results, out_retry):
-    """Initializes the worker.
+  def __init__(self, tests=[]):
+    self._lock = threading.Lock()
+    self._tests = []
+    self._tests_in_progress = 0
+    # Used to signal that an item is avaliable or all items have been handled.
+    self._item_avaliable_or_all_done = threading.Event()
+    for t in tests:
+      self.add(t)
+
+  def _pop(self):
+    """Pop a test from the collection.
+
+    Waits until a test is avaliable or all tests have been handled.
+
+    Returns:
+      A test or None if all tests have been handled.
+    """
+    while True:
+      # Wait for a test to be avaliable or all tests to have been handled.
+      self._item_avaliable_or_all_done.wait()
+      with self._lock:
+        # Check which of the two conditions triggered the signal.
+        if self._tests_in_progress == 0:
+          return None
+        try:
+          return self._tests.pop()
+        except IndexError:
+          # Another thread beat us to the avaliable test, wait again.
+          self._item_avaliable_or_all_done.clear()
+
+  def add(self, test):
+    """Add an test to the collection.
 
     Args:
-      runner: A TestRunner object used to run the tests.
-      test_queue: A list from which to get tests to run.
-      out_results: A list to add TestResults to.
-      out_retry: A list to add tests to retry.
-    """
-    super(_Worker, self).__init__()
-    self.daemon = True
-    self._exc_info = None
-    self._runner = runner
-    self._test_queue = test_queue
-    self._out_results = out_results
-    self._out_retry = out_retry
-
-  #override
-  def run(self):
-    """Run tests from the queue in a seperate thread until it is empty.
-
-    Adds TestResults objects to the out_results list and may add tests to the
-    out_retry list.
+      item: A test to add.
     """
+    with self._lock:
+      self._tests.append(test)
+      self._item_avaliable_or_all_done.set()
+      self._tests_in_progress += 1
+
+  def test_completed(self):
+    """Indicate that a test has been fully handled."""
+    with self._lock:
+      self._tests_in_progress -= 1
+      if self._tests_in_progress == 0:
+        # All tests have been handled, signal all waiting threads.
+        self._item_avaliable_or_all_done.set()
+
+  def __iter__(self):
+    """Iterate through tests in the collection until all have been handled."""
+    while True:
+      r = self._pop()
+      if r is None:
+        break
+      yield r
+
+
+def _RunTestsFromQueue(runner, test_collection, out_results):
+  """Runs tests from the test_collection until empty using the given runner.
+
+  Adds TestResults objects to the out_results list and may add tests to the
+  out_retry list.
+
+  Args:
+    runner: A TestRunner object used to run the tests.
+    test_collection: A _TestCollection from which to get _Test objects to run.
+    out_results: A list to add TestResults to.
+  """
+  for test in test_collection:
     try:
-      while True:
-        test = self._test_queue.pop()
-        result, retry = self._runner.Run(test)
-        self._out_results.append(result)
-        if retry:
-          self._out_retry.append(retry)
-    except IndexError:
-      pass
+      if not android_commands.IsDeviceAttached(runner.device):
+        # Device is unresponsive, stop handling tests on this device.
+        msg = 'Device %s is unresponsive.' % runner.device
+        logging.warning(msg)
+        raise android_commands.errors.DeviceUnresponsiveError(msg)
+      result, retry = runner.RunTest(test.test)
+      test.tries += 1
+      if retry and test.tries <= 3:
+        # Retry non-passing results, only record passing results.
+        out_results.append(test_result.TestResults.FromRun(ok=result.ok))
+        logging.warning('****Retrying test, try #%s.' % test.tries)
+        test_collection.add(_Test(test=retry, tries=test.tries))
+      else:
+        # All tests passed or retry limit reached. Either way, record results.
+        out_results.append(result)
+    except android_commands.errors.DeviceUnresponsiveError:
+      # Device is unresponsive, stop handling tests on this device and ensure
+      # current test gets runs by another device. Don't reraise this exception
+      # on the main thread.
+      test_collection.add(test)
+      return
     except:
-      self._exc_info = sys.exc_info()
+      # An unhandleable exception, ensure tests get run by another device and
+      # reraise this exception on the main thread.
+      test_collection.add(test)
       raise
+    finally:
+      # Retries count as separate tasks so always mark the popped test as done.
+      test_collection.test_completed()
+
+
+def _SetUp(runner_factory, device, out_runners):
+  """Creates a test runner for each device and calls SetUp() in parallel.
 
-  def ReraiseIfException(self):
-    """Reraise exception if an exception was raised in the thread."""
-    if self._exc_info:
-      raise self._exc_info[0], self._exc_info[1], self._exc_info[2]
+  Note: if a device is unresponsive the corresponding TestRunner will not be
+    added to out_runners.
+
+  Args:
+    runner_factory: callable that takes a device and returns a TestRunner.
+    device: the device serial number to set up.
+    out_runners: list to add the successfully set up TestRunner object.
+  """
+  try:
+    logging.warning('*****Creating shard for %s.', device)
+    runner = runner_factory(device)
+    runner.SetUp()
+    out_runners.append(runner)
+  except android_commands.errors.DeviceUnresponsiveError as e:
+    logging.warning('****Failed to create shard for %s: [%s]', (device, e))
 
 
 def _RunAllTests(runners, tests):
@@ -70,28 +163,21 @@ def _RunAllTests(runners, tests):
     tests: a list of Tests to run using the given TestRunners.
 
   Returns:
-    Tuple: (list of TestResults, list of tests to retry)
+    A TestResults object.
   """
-  tests_queue = list(tests)
-  workers = []
+  logging.warning('****Running %s tests with %s test runners.' %
+                  (len(tests), len(runners)))
+  tests_collection = _TestCollection([_Test(t) for t in tests])
   results = []
-  retry = []
-  for r in runners:
-    worker = _Worker(r, tests_queue, results, retry)
-    worker.start()
-    workers.append(worker)
-  while workers:
-    for w in workers[:]:
-      # Allow the main thread to periodically check for keyboard interrupts.
-      w.join(0.1)
-      if not w.isAlive():
-        w.ReraiseIfException()
-        workers.remove(w)
-  return (results, retry)
+  workers = reraiser_thread.ReraiserThreadGroup([reraiser_thread.ReraiserThread(
+      _RunTestsFromQueue, [r, tests_collection, results]) for r in runners])
+  workers.StartAll()
+  workers.JoinAll()
+  return test_result.TestResults.FromTestResults(results)
 
 
 def _CreateRunners(runner_factory, devices):
-  """Creates a test runner for each device.
+  """Creates a test runner for each device and calls SetUp() in parallel.
 
   Note: if a device is unresponsive the corresponding TestRunner will not be
     included in the returned list.
@@ -103,20 +189,29 @@ def _CreateRunners(runner_factory, devices):
   Returns:
     A list of TestRunner objects.
   """
+  logging.warning('****Creating %s test runners.' % len(devices))
   test_runners = []
-  for index, device in enumerate(devices):
-    logging.warning('*' * 80)
-    logging.warning('Creating shard %d for %s', index, device)
-    logging.warning('*' * 80)
-    try:
-      test_runners.append(runner_factory(device))
-    except android_commands.errors.DeviceUnresponsiveError as e:
-      logging.warning('****Failed to create a shard: [%s]', e)
+  threads = reraiser_thread.ReraiserThreadGroup(
+      [reraiser_thread.ReraiserThread(_SetUp, [runner_factory, d, test_runners])
+       for d in devices])
+  threads.StartAll()
+  threads.JoinAll()
   return test_runners
 
 
-def ShardAndRunTests(runner_factory, devices, tests, build_type='Debug',
-                     tries=3):
+def _TearDownRunners(runners):
+  """Calls TearDown() for each test runner in parallel.
+  Args:
+    runners: a list of TestRunner objects.
+  """
+  threads = reraiser_thread.ReraiserThreadGroup(
+      [reraiser_thread.ReraiserThread(runner.TearDown)
+       for runner in runners])
+  threads.StartAll()
+  threads.JoinAll()
+
+
+def ShardAndRunTests(runner_factory, devices, tests, build_type='Debug'):
   """Run all tests on attached devices, retrying tests that don't pass.
 
   Args:
@@ -124,34 +219,18 @@ def ShardAndRunTests(runner_factory, devices, tests, build_type='Debug',
     devices: list of attached device serial numbers as strings.
     tests: list of tests to run.
     build_type: either 'Debug' or 'Release'.
-    tries: number of tries before accepting failure.
 
   Returns:
     A test_result.TestResults object.
   """
-  final_results = test_result.TestResults()
-  results = test_result.TestResults()
   forwarder.Forwarder.KillHost(build_type)
-  try_count = 0
-  while tests:
-    devices = set(devices).intersection(android_commands.GetAttachedDevices())
-    if not devices:
-      # There are no visible devices attached, this is unrecoverable.
-      msg = 'No devices attached and visible to run tests!'
-      logging.critical(msg)
-      raise Exception(msg)
-    if try_count >= tries:
-      # We've retried too many times, return the TestResults up to this point.
-      results.ok = final_results.ok
-      final_results = results
-      break
-    try_count += 1
-    runners = _CreateRunners(runner_factory, devices)
+  runners = _CreateRunners(runner_factory, devices)
+  try:
+    return _RunAllTests(runners, tests)
+  finally:
     try:
-      results_list, tests = _RunAllTests(runners, tests)
-      results = test_result.TestResults.FromTestResults(results_list)
-      final_results.ok += results.ok
+      _TearDownRunners(runners)
     except android_commands.errors.DeviceUnresponsiveError as e:
-      logging.warning('****Failed to run test: [%s]', e)
-  forwarder.Forwarder.KillHost(build_type)
-  return final_results
+      logging.warning('****Device unresponsive during TearDown: [%s]', e)
+    finally:
+      forwarder.Forwarder.KillHost(build_type)
diff --git a/build/android/pylib/base/shard_unittest.py b/build/android/pylib/base/shard_unittest.py
index 0448d83..79aa5b2 100644
--- a/build/android/pylib/base/shard_unittest.py
+++ b/build/android/pylib/base/shard_unittest.py
@@ -10,7 +10,10 @@ import unittest
 
 sys.path.append(os.path.join(os.path.dirname(os.path.realpath(__file__)),
                 os.pardir, os.pardir))
+
+# Mock out android_commands.GetAttachedDevices().
 from pylib import android_commands
+android_commands.GetAttachedDevices = lambda: ['0', '1']
 
 import shard
 import test_result
@@ -24,68 +27,106 @@ class MockRunner(object):
   """A mock TestRunner."""
   def __init__(self, device='0'):
     self.device = device
+    self.setups = 0
+    self.teardowns = 0
 
-  def Run(self, test):
+  def RunTest(self, test):
     return (test_result.TestResults.FromRun(
                 ok=[test_result.BaseTestResult(test, '')]),
             None)
 
+  def SetUp(self):
+    self.setups += 1
+
+  def TearDown(self):
+    self.teardowns += 1
+
 
-class MockRunnerRetry(MockRunner):
-  def Run(self, test):
+class MockRunnerFail(MockRunner):
+  def RunTest(self, test):
     return (test_result.TestResults.FromRun(
                 failed=[test_result.BaseTestResult(test, '')]),
             test)
 
 
+class MockRunnerFailTwice(MockRunner):
+  def __init__(self, device='0'):
+    super(MockRunnerFailTwice, self).__init__(device)
+    self._fails = 0
+
+  def RunTest(self, test):
+    self._fails += 1
+    if self._fails <= 2:
+      return (test_result.TestResults.FromRun(
+                  failed=[test_result.BaseTestResult(test, '')]),
+              test)
+    else:
+      return (test_result.TestResults.FromRun(
+                  ok=[test_result.BaseTestResult(test, '')]),
+            None)
+
+
 class MockRunnerException(MockRunner):
-  def Run(self, test):
+  def RunTest(self, test):
     raise TestException
 
 
-class TestWorker(unittest.TestCase):
-  """Tests for shard._Worker."""
+class TestFunctions(unittest.TestCase):
+  """Tests for shard._RunTestsFromQueue."""
   @staticmethod
-  def _RunRunner(mock_runner, tests):
+  def _RunTests(mock_runner, tests):
     results = []
-    retry = []
-    worker = shard._Worker(mock_runner, tests, results, retry)
-    worker.start()
-    worker.join()
-    worker.ReraiseIfException()
-    return (results, retry)
+    tests = shard._TestCollection([shard._Test(t) for t in tests])
+    shard._RunTestsFromQueue(mock_runner, tests, results)
+    return test_result.TestResults.FromTestResults(results)
 
-  def testRun(self):
-    results, retry = TestWorker._RunRunner(MockRunner(), ['a', 'b'])
-    self.assertEqual(len(results), 2)
-    self.assertEqual(len(retry), 0)
+  def testRunTestsFromQueue(self):
+    results = TestFunctions._RunTests(MockRunner(), ['a', 'b'])
+    self.assertEqual(len(results.ok), 2)
+    self.assertEqual(len(results.GetAllBroken()), 0)
 
-  def testRetry(self):
-    results, retry = TestWorker._RunRunner(MockRunnerRetry(), ['a', 'b'])
-    self.assertEqual(len(results), 2)
-    self.assertEqual(len(retry), 2)
+  def testRunTestsFromQueueRetry(self):
+    results = TestFunctions._RunTests(MockRunnerFail(), ['a', 'b'])
+    self.assertEqual(len(results.ok), 0)
+    self.assertEqual(len(results.failed), 2)
 
-  def testReraise(self):
-    with self.assertRaises(TestException):
-      TestWorker._RunRunner(MockRunnerException(), ['a', 'b'])
+  def testRunTestsFromQueueFailTwice(self):
+    results = TestFunctions._RunTests(MockRunnerFailTwice(), ['a', 'b'])
+    self.assertEqual(len(results.ok), 2)
+    self.assertEqual(len(results.GetAllBroken()), 0)
+
+  def testSetUp(self):
+    runners = []
+    shard._SetUp(MockRunner, '0', runners)
+    self.assertEqual(len(runners), 1)
+    self.assertEqual(runners[0].setups, 1)
 
 
-class TestRunAllTests(unittest.TestCase):
+class TestThreadGroupFunctions(unittest.TestCase):
   """Tests for shard._RunAllTests and shard._CreateRunners."""
   def setUp(self):
     self.tests = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
 
-  def testRun(self):
+  def testCreate(self):
     runners = shard._CreateRunners(MockRunner, ['0', '1'])
-    results, retry = shard._RunAllTests(runners, self.tests)
-    self.assertEqual(len(results), len(self.tests))
-    self.assertEqual(len(retry), 0)
+    for runner in runners:
+      self.assertEqual(runner.setups, 1)
+
+  def testRun(self):
+    runners = [MockRunner('0'), MockRunner('1')]
+    results = shard._RunAllTests(runners, self.tests)
+    self.assertEqual(len(results.ok), len(self.tests))
+
+  def testTearDown(self):
+    runners = [MockRunner('0'), MockRunner('1')]
+    shard._TearDownRunners(runners)
+    for runner in runners:
+      self.assertEqual(runner.teardowns, 1)
 
   def testRetry(self):
-    runners = shard._CreateRunners(MockRunnerRetry, ['0', '1'])
-    results, retry = shard._RunAllTests(runners, self.tests)
-    self.assertEqual(len(results), len(self.tests))
-    self.assertEqual(len(retry), len(self.tests))
+    runners = shard._CreateRunners(MockRunnerFail, ['0', '1'])
+    results = shard._RunAllTests(runners, self.tests)
+    self.assertEqual(len(results.failed), len(self.tests))
 
   def testReraise(self):
     runners = shard._CreateRunners(MockRunnerException, ['0', '1'])
@@ -97,17 +138,14 @@ class TestShard(unittest.TestCase):
   """Tests for shard.Shard."""
   @staticmethod
   def _RunShard(runner_factory):
-    devices = ['0', '1']
-    # Mock out android_commands.GetAttachedDevices().
-    android_commands.GetAttachedDevices = lambda: devices
-    return shard.ShardAndRunTests(runner_factory, devices, ['a', 'b', 'c'])
+    return shard.ShardAndRunTests(runner_factory, ['0', '1'], ['a', 'b', 'c'])
 
   def testShard(self):
     results = TestShard._RunShard(MockRunner)
     self.assertEqual(len(results.ok), 3)
 
   def testFailing(self):
-    results = TestShard._RunShard(MockRunnerRetry)
+    results = TestShard._RunShard(MockRunnerFail)
     self.assertEqual(len(results.ok), 0)
     self.assertEqual(len(results.failed), 3)
 
diff --git a/build/android/pylib/gtest/test_package.py b/build/android/pylib/gtest/test_package.py
index cde3b39..c1172c8 100644
--- a/build/android/pylib/gtest/test_package.py
+++ b/build/android/pylib/gtest/test_package.py
@@ -44,6 +44,10 @@ class TestPackage(object):
       timeout = timeout * 2
     self.timeout = timeout * self.tool.GetTimeoutScale()
 
+  def ClearApplicationState(self):
+    """Clears the application state."""
+    raise NotImplementedError('Method must be overriden.')
+
   def GetDisabledPrefixes(self):
     return ['DISABLED_', 'FLAKY_', 'FAILS_']
 
diff --git a/build/android/pylib/gtest/test_package_apk.py b/build/android/pylib/gtest/test_package_apk.py
index 3d9a926..f18d87c 100644
--- a/build/android/pylib/gtest/test_package_apk.py
+++ b/build/android/pylib/gtest/test_package_apk.py
@@ -73,6 +73,10 @@ class TestPackageApk(TestPackage):
     args += ['shell', 'cat', self._GetFifo()]
     return pexpect.spawn('adb', args, timeout=timeout, logfile=logfile)
 
+  def ClearApplicationState(self):
+    """Clear the application state."""
+    self.adb.ClearApplicationState(self._apk_package_name)
+
   def GetAllTests(self):
     """Returns a list of all tests available in the test suite."""
     self._CreateTestRunnerScript('--gtest_list_tests')
diff --git a/build/android/pylib/gtest/test_package_executable.py b/build/android/pylib/gtest/test_package_executable.py
index cb15092..690c101 100644
--- a/build/android/pylib/gtest/test_package_executable.py
+++ b/build/android/pylib/gtest/test_package_executable.py
@@ -76,6 +76,10 @@ class TestPackageExecutable(TestPackage):
     export_string += 'export GCOV_PREFIX_STRIP=%s\n' % depth
     return export_string
 
+  def ClearApplicationState(self):
+    """Clear the application state."""
+    self.adb.KillAllBlocking(self.test_suite_basename, 30)
+
   def GetAllTests(self):
     """Returns a list of all tests available in the test suite."""
     all_tests = self.adb.RunShellCommand(
diff --git a/build/android/pylib/gtest/test_runner.py b/build/android/pylib/gtest/test_runner.py
index 0ecc9c5a..c86e4fe 100644
--- a/build/android/pylib/gtest/test_runner.py
+++ b/build/android/pylib/gtest/test_runner.py
@@ -240,6 +240,7 @@ class TestRunner(base_test_runner.BaseTestRunner):
           gtest_filter_base_path + '_emulator_additional_disabled'))
     return disabled_tests
 
+  #override
   def RunTest(self, test):
     """Runs a test on a single device.
 
@@ -254,6 +255,7 @@ class TestRunner(base_test_runner.BaseTestRunner):
       return test_results, None
 
     try:
+      self.test_package.ClearApplicationState()
       self.test_package.CreateTestRunnerScript(test, self._test_arguments)
       test_results = self.test_package.RunTestsAndListResults()
     except errors.DeviceUnresponsiveError as e:
@@ -261,7 +263,6 @@ class TestRunner(base_test_runner.BaseTestRunner):
       logging.warning(e)
       if android_commands.IsDeviceAttached(self.device):
         raise
-      test_results.device_exception = device_exception
     # Calculate unknown test results.
     # TODO(frankf): Do not break TestResults encapsulation.
     all_tests = set(test.split(':'))
@@ -272,15 +273,16 @@ class TestRunner(base_test_runner.BaseTestRunner):
     retry = ':'.join([t.name for t in test_results.GetAllBroken()])
     return test_results, retry
 
+  #override
   def SetUp(self):
     """Sets up necessary test enviroment for the test suite."""
     super(TestRunner, self).SetUp()
-    self.adb.ClearApplicationState(constants.CHROME_PACKAGE)
     self.StripAndCopyFiles()
     if _TestSuiteRequiresMockTestServer(self.test_package.test_suite_basename):
       self.LaunchChromeTestServerSpawner()
     self.tool.SetupEnvironment()
 
+  #override
   def TearDown(self):
     """Cleans up the test enviroment for the test suite."""
     self.tool.CleanUpEnvironment()
diff --git a/build/android/pylib/utils/reraiser_thread.py b/build/android/pylib/utils/reraiser_thread.py
new file mode 100644
index 0000000..95c3f74
--- /dev/null
+++ b/build/android/pylib/utils/reraiser_thread.py
@@ -0,0 +1,74 @@
+# Copyright 2013 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.
+
+"""Thread and ThreadGroup that reraise exceptions on the main thread."""
+
+import sys
+import threading
+
+
+class ReraiserThread(threading.Thread):
+  """Thread class that can reraise exceptions."""
+  def __init__(self, func, args=[], kwargs={}):
+    super(ReraiserThread, self).__init__()
+    self.daemon = True
+    self._func = func
+    self._args = args
+    self._kwargs = kwargs
+    self._exc_info = None
+
+  def ReraiseIfException(self):
+    """Reraise exception if an exception was raised in the thread."""
+    if self._exc_info:
+      raise self._exc_info[0], self._exc_info[1], self._exc_info[2]
+
+  #override
+  def run(self):
+    """Overrides Thread.run() to add support for reraising exceptions."""
+    try:
+      self._func(*self._args, **self._kwargs)
+    except:
+      self._exc_info = sys.exc_info()
+      raise
+
+
+class ReraiserThreadGroup(object):
+  """A group of ReraiserThread objects."""
+  def __init__(self, threads=[]):
+    """Initialize thread group.
+
+    Args:
+      threads: a list of ReraiserThread objects; defaults to empty.
+    """
+    self._threads = threads
+
+  def Add(self, thread):
+    """Add a thread to the group.
+
+    Args:
+      thread: a ReraiserThread object.
+    """
+    self._threads.append(thread)
+
+  def StartAll(self):
+    """Start all threads."""
+    for thread in self._threads:
+      thread.start()
+
+  def JoinAll(self):
+    """Join all threads.
+
+    Reraises exceptions raised by the child threads and supports
+    breaking immediately on exceptions raised on the main thread.
+    """
+    alive_threads = self._threads[:]
+    while alive_threads:
+      for thread in alive_threads[:]:
+        # Allow the main thread to periodically check for interrupts.
+        thread.join(0.1)
+        if not thread.isAlive():
+          alive_threads.remove(thread)
+    # All threads are allowed to complete before reraising exceptions.
+    for thread in self._threads:
+      thread.ReraiseIfException()
diff --git a/build/android/pylib/utils/reraiser_thread_unittest.py b/build/android/pylib/utils/reraiser_thread_unittest.py
new file mode 100644
index 0000000..9228829
--- /dev/null
+++ b/build/android/pylib/utils/reraiser_thread_unittest.py
@@ -0,0 +1,80 @@
+# Copyright 2013 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.
+
+"""Unittests for reraiser_thread.py."""
+
+import unittest
+
+import reraiser_thread
+
+
+class TestException(Exception):
+  pass
+
+
+class TestReraiserThread(unittest.TestCase):
+  """Tests for reraiser_thread.ReraiserThread."""
+  def testNominal(self):
+    result = [None, None]
+
+    def f(a, b=None):
+      result[0] = a
+      result[1] = b
+
+    thread = reraiser_thread.ReraiserThread(f, [1], {'b': 2})
+    thread.start()
+    thread.join()
+    self.assertEqual(result[0], 1)
+    self.assertEqual(result[1], 2)
+
+  def testRaise(self):
+    def f():
+      raise TestException
+
+    thread = reraiser_thread.ReraiserThread(f)
+    thread.start()
+    thread.join()
+    with self.assertRaises(TestException):
+      thread.ReraiseIfException()
+
+
+class TestReraiserThreadGroup(unittest.TestCase):
+  """Tests for reraiser_thread.ReraiserThreadGroup."""
+  def testInit(self):
+    ran = [False] * 5
+    def f(i):
+      ran[i] = True
+
+    group = reraiser_thread.ReraiserThreadGroup(
+      [reraiser_thread.ReraiserThread(f, args=[i]) for i in range(5)])
+    group.StartAll()
+    group.JoinAll()
+    for v in ran:
+      self.assertTrue(v)
+
+  def testAdd(self):
+    ran = [False] * 5
+    def f(i):
+      ran[i] = True
+
+    group = reraiser_thread.ReraiserThreadGroup()
+    for i in xrange(5):
+      group.Add(reraiser_thread.ReraiserThread(f, args=[i]))
+    group.StartAll()
+    group.JoinAll()
+    for v in ran:
+      self.assertTrue(v)
+
+  def testJoinRaise(self):
+    def f():
+      raise TestException
+    group = reraiser_thread.ReraiserThreadGroup(
+      [reraiser_thread.ReraiserThread(f) for _ in xrange(5)])
+    group.StartAll()
+    with self.assertRaises(TestException):
+      group.JoinAll()
+
+
+if __name__ == '__main__':
+  unittest.main()