Android: improves test sharding reliability.

build/android/pylib/base_test_sharder.py

Index: build/android/pylib/base_test_sharder.py
diff --git a/build/android/pylib/base_test_sharder.py b/build/android/pylib/base_test_sharder.py
index 5306769155050ba032364650daeecaa1581e609b..b8d03c8300dd170c6a9b46b17c1af16504feacf5 100644
--- a/build/android/pylib/base_test_sharder.py
+++ b/build/android/pylib/base_test_sharder.py
@@ -7,6 +7,7 @@ import android_commands
 import logging
 import multiprocessing
 
+from android_commands import errors
 from test_result import TestResults
 
 
@@ -43,7 +44,9 @@ class BaseTestSharder(object):
 
   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):
@@ -83,12 +86,20 @@ class BaseTestSharder(object):
       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,
@@ -96,8 +107,12 @@ class BaseTestSharder(object):
       # 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
@@ -119,5 +134,9 @@ class BaseTestSharder(object):
           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