Android: improves test sharding reliability.

build/android/pylib/base_test_sharder.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 android_commands
 import logging
 import multiprocessing
 
+from android_commands import errors
 from test_result import TestResults
 
 
 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)
@@ skipping 16 matching lines @@
   """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
+    # Worst case scenario: a device will drop offline per run, so we need
+    # to retry until we're out of devices.
+    self.retries = len(self.attached_devices)
     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:
@@ skipping 19 matching lines @@
     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]
+
+      # Try to create N shards, and retrying on failure.
+      try:
+        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]
+      except errors.DeviceUnresponsiveError as e:
+        logging.critical('****Failed to create a shard: [%s]', e)
+        self.attached_devices.remove(device)
+        continue
+
       logging.warning('Starting...')
       pool = multiprocessing.Pool(len(self.attached_devices),
                                   SetTestsContainer,
                                   [BaseTestSharder.tests_container])
       # map can't handle KeyboardInterrupt exception. It's a python bug.
       # So use map_async instead.
       async_results = pool.map_async(_ShardedTestRunnable, test_runners)
-      results_lists = async_results.get(999999)
-
+      try:
+        results_lists = async_results.get(999999)
+      except errors.DeviceUnresponsiveError as e:
+        logging.critical('****Failed to run test: [%s]', e)
+        self.attached_devices = android_commands.GetAttachedDevices()
+        continue

[yongsheng, 2012/11/01 02:22:49]

If one device raises an exception, this might block other test runners.
what about catching up exceptions in "_ShardedTestRunnable" so that once one is
failed, others can continue to run? then in the next round, we just need to try
those failures?

[bulach, 2012/11/01 11:41:06]

hmm, sorry, I'm not sure what do you mean by "block other test runners"...
I just did a quick test, and if you run "foo.py 4", it blocks for ~10secs.
running "foo.py 5" returns immediately, i.e., the exception thrown by any of the
shards immediately bubbles up...

so in our case here, we'd immediately retry on the remaining runners, and as
mentioned to yaron above, they wouldn't be blocked...

I appreciate we could potentially optimize this :) and just retry the failures.
however, there's currently no simple way to communicate that back from
"_ShardedTestRunnable" (we'd need more logic there to set all tests as broken
with a device exception so that 121 and 132-134 would work...)

if that's ok, would you be fine with doing this as a possible follow up?

I'd like to gather data first to understand what are all failure scenarios in
the bots.


===foo.py===

import multiprocessing
import sys
import time

def sleep_or_raise(i):
  if i < 4:
    time.sleep(10)
  else:
    raise Exception('Raised!!!')

n = int(sys.argv[1])
pool = multiprocessing.Pool(n)
r = pool.map_async(sleep_or_raise, xrange(n))
r.get(9999999)

[yongsheng, 2012/11/02 00:55:11]

you're right. That's what i mean: one exception is raised from one device and
other devices running cases will be interrupted. yes, that's just a
optimization.
 
yes, please.
oh, yes, Yaron is right. we can help improve this later. Please go ahead.

       test_results = TestResults.FromTestResults(results_lists)
       # Re-check the attached devices for some devices may
       # become offline
       retry_devices = set(android_commands.GetAttachedDevices())
       # Remove devices that had exceptions.
       retry_devices -= TestResults.DeviceExceptions(results_lists)
       # Retry on devices that didn't have any exception.
       self.attached_devices = list(retry_devices)
       if (retry == self.retries - 1 or
           len(self.attached_devices) == 0):
         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
+    else:
+      # We ran out retries, possibly out of healthy devices.
+      # There's no recovery at this point.
+      raise Exception('Unrecoverable error while retrying test runs.')
     self.OnTestsCompleted(test_runners, final_results)
     return final_results