Implement clustering support in run_test_cases.py.

run_test_cases.py

 #!/usr/bin/env 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.
 
 """Runs each test cases as a single shard, single process execution.
 
 Similar to sharding_supervisor.py but finer grained. It runs each test case
 individually instead of running per shard. Runs multiple instances in parallel.
 """
@@ skipping 652 matching lines @@
     # times in total.
     self.retries = retries
     self.decider = decider
     self.verbose = verbose
     self.add_task = add_task
     self.add_serial_task = add_serial_task
     # It is important to remove the shard environment variables since it could
     # conflict with --gtest_filter.
     self.env = setup_gtest_env()
 
-  def map(self, priority, test_case, try_count):
+  def map(self, priority, test_cases, try_count):
     """Traces a single test case and returns its output.
 
     try_count is 0 based, the original try is 0.
     """
     if self.decider.should_stop():
       return []
 
+    cmd = self.cmd + ['--gtest_filter=%s' % ':'.join(test_cases)]
+    if '--gtest_print_time' not in cmd:
+      cmd.append('--gtest_print_time')
     start = time.time()
     output, returncode = call_with_timeout(
-        self.cmd + ['--gtest_filter=%s' % test_case],
+        cmd,
         self.timeout,
         cwd=self.cwd_dir,
         stderr=subprocess.STDOUT,
         env=self.env)
     duration = time.time() - start
-    data = {
-      'test_case': test_case,
-      'returncode': returncode,
-      'duration': duration,
-      # It needs to be valid utf-8 otherwise it can't be store.
-      'output': output.decode('ascii', 'ignore').encode('utf-8'),
-    }
-    if '[ RUN      ]' not in output:
-      # Can't find gtest marker, mark it as invalid.
-      returncode = returncode or 1
-    self.decider.got_result(not bool(returncode))
+
+    # It needs to be valid utf-8 otherwise it can't be stored.
+    # TODO(maruel): Be more intelligent than decoding to ascii.
+    utf8_output = output.decode('ascii', 'ignore').encode('utf-8')
+
+    if len(test_cases) > 1:
+      data = process_output(utf8_output, test_cases, duration, returncode)
+    else:
+      if '[ RUN      ]' not in output:
+        # Can't find gtest marker, mark it as invalid.
+        returncode = returncode or 1
+      data = [
+        {
+          'test_case': test_cases[0],
+          'returncode': returncode,
+          'duration': duration,
+          'output': utf8_output,
+        }
+      ]
+
     if sys.platform == 'win32':
       output = output.replace('\r\n', '\n')
-    need_to_retry = returncode and try_count < self.retries
 
-    if try_count:
-      line = '%s (%.2fs) - retry #%d' % (test_case, duration, try_count)
-    else:
-      line = '%s (%.2fs)' % (test_case, duration)
-    if self.verbose or returncode or try_count > 0:
-      # Print output in one of three cases:
-      #   --verbose was specified.
-      #   The test failed.
-      #   The wasn't the first attempt (this is needed so the test parser can
-      #       detect that a test has been successfully retried).
-      line += '\n' + output
-    self.progress.update_item(line, True, need_to_retry)
+    for i in data:
+      self.decider.got_result(i['returncode'] == 0)

[M-A Ruel, 2013/03/11 18:43:22]

I got caught by the fact that i['returncode'] == None is failure. So I changed
the conditions to be more explicit.

+      need_to_retry = i['returncode'] != 0 and try_count < self.retries
+      if try_count:
+        line = '%s (%.2fs) - retry #%d' % (
+            i['test_case'], i['duration'] or 0, try_count)
+      else:
+        line = '%s (%.2fs)' % (i['test_case'], i['duration'] or 0)
+      if self.verbose or i['returncode'] != 0 or try_count > 0:
+        # Print output in one of three cases:
+        #   --verbose was specified.
+        #   The test failed.
+        #   The wasn't the first attempt (this is needed so the test parser can
+        #       detect that a test has been successfully retried).
+        line += '\n' + i['output']
+      self.progress.update_item(line, True, need_to_retry)
 
-    if need_to_retry:
-      if try_count + 1 < self.retries:
-        # The test failed and needs to be retried normally.
-        # Leave a buffer of ~40 test cases before retrying.
-        priority += 40
-        self.add_task(priority, self.map, priority, test_case, try_count + 1)
-      else:
-        # This test only has one retry left, so the final retry should be
-        # done serially.
-        self.add_serial_task(priority, self.map, priority, test_case,
-                             try_count + 1)
-
-    return [data]
+      if need_to_retry:
+        if try_count + 1 < self.retries:
+          # The test failed and needs to be retried normally.
+          # Leave a buffer of ~40 test cases before retrying.
+          priority += 40
+          self.add_task(
+              priority, self.map, priority, [i['test_case']], try_count + 1)
+        else:
+          # This test only has one retry left, so the final retry should be
+          # done serially.
+          self.add_serial_task(
+              priority, self.map, priority, [i['test_case']], try_count + 1)
+    return data
 
 
 def get_test_cases(cmd, cwd, whitelist, blacklist, index, shards, seed):
   """Returns the filtered list of test cases.
 
   This is done synchronously.
   """
   try:
     tests = list_test_cases(
         cmd,
@@ skipping 202 matching lines @@
 def run_test_cases(
     cmd, cwd, test_cases, jobs, timeout, clusters, retries, run_all,
     max_failures, no_cr, gtest_output, result_file, verbose):
   """Runs test cases in parallel.
 
   Arguments:
     - cmd: command to run.
     - cwd: working directory.
     - test_cases: list of preprocessed test cases to run.
     - jobs: number of parallel execution threads to do.
-    - timeout: individual test case timeout.
+    - timeout: individual test case timeout. Modulated when used with
+      clustering.
     - clusters: number of test cases to lump together in a single execution. 0
       means the default automatic value which depends on len(test_cases) and
       jobs. Capped to len(test_cases) / jobs.
     - retries: number of times a test case can be retried.
     - run_all: If true, do not early return even if all test cases fail.
     - max_failures is the absolute maximum number of tolerated failures or None.
     - no_cr: makes output friendly to piped logs.
     - gtest_output: saves results as xml.
     - result_file: saves results as json.
     - verbose: print more details.
 
   It may run a subset of the test cases if too many test cases failed, as
   determined with max_failures, retries and run_all.
   """
   assert 0 <= retries <= 100000
   if not test_cases:
     return 0
   if run_all:
     decider = RunAll()
   else:
     # If 10% of test cases fail, just too bad.
     decider = RunSome(len(test_cases), retries, 2, 0.1, max_failures)
 
   if not clusters:
     clusters = calc_cluster_default(len(test_cases), jobs)
   else:
     # Limit the value.
     clusters = min(clusters, len(test_cases) / jobs)
 
+  logging.debug('%d test cases with clusters of %d', len(test_cases), clusters)
+
   if gtest_output:
     gtest_output = gen_gtest_output_dir(cwd, gtest_output)
   progress = Progress(len(test_cases))
   serial_tasks = QueueWithProgress(0)
   serial_tasks.set_progress(progress)
 
   def add_serial_task(priority, func, *args, **kwargs):
     """Adds a serial task, to be executed later."""
     assert isinstance(priority, int)
     assert callable(func)
     serial_tasks.put((priority, func, args, kwargs))
 
   with ThreadPool(progress, jobs, jobs, len(test_cases)) as pool:
     runner = Runner(
         cmd, cwd, timeout, progress, retries, decider, verbose,
         pool.add_task, add_serial_task)
     function = runner.map
-    logging.debug('Adding tests to ThreadPool')
     progress.use_cr_only = not no_cr
-    for i, test_case in enumerate(test_cases):
-      pool.add_task(i, function, i, test_case, 0)
-    logging.debug('All tests added to the ThreadPool')
+    # Cluster the test cases right away.
+    for i in xrange((len(test_cases) + clusters - 1) / clusters):
+      cluster = test_cases[i*clusters : (i+1)*clusters]
+      pool.add_task(i, function, i, cluster, 0)
     results = pool.join()
 
     # Retry any failed tests serially.
     if not serial_tasks.empty():
       progress.update_item('\n'.join(running_serial_warning()), index=False,
                                       size=False)
 
       while not serial_tasks.empty():
         _priority, func, args, kwargs = serial_tasks.get()
         results.append(func(*args, **kwargs))
         serial_tasks.task_done()
 
       # Call join since that is a standard call once a queue has been emptied.
       serial_tasks.join()
 
     duration = time.time() - pool.tasks.progress.start
 
   cleaned = {}
-  for item in results:
-    if item:
-      cleaned.setdefault(item[0]['test_case'], []).extend(item)
+  for map_run in results:
+    for test_case_result in map_run:
+      cleaned.setdefault(test_case_result['test_case'], []).append(
+          test_case_result)
   results = cleaned
 
   # Total time taken to run each test case.
   test_case_duration = dict(
-      (test_case, sum(i.get('duration', 0) for i in item))
+      (test_case, sum((i.get('duration') or 0) for i in item))
       for test_case, item in results.iteritems())
 
   # Classify the results
   success = []
   flaky = []
   fail = []
   nb_runs = 0
   for test_case in sorted(results):
     items = results[test_case]
     nb_runs += len(items)
-    if not any(not i['returncode'] for i in items):
+    if not any(i['returncode'] == 0 for i in items):
       fail.append(test_case)
-    elif len(items) > 1 and any(not i['returncode'] for i in items):
+    elif len(items) > 1 and any(i['returncode'] == 0 for i in items):
       flaky.append(test_case)
     elif len(items) == 1 and items[0]['returncode'] == 0:
       success.append(test_case)
     else:
+      # The test never ran.
       assert False, items
   missing = list(set(test_cases) - set(success) - set(flaky) - set(fail))
 
   saved = {
     'test_cases': results,
     'expected': len(test_cases),
     'success': success,
     'flaky': flaky,
     'fail': fail,
     'missing': missing,
@@ skipping 303 matching lines @@
         options.gtest_output,
         result_file,
         options.verbose)
   except Failure as e:
     print >> sys.stderr, e.args[0]
     return 1
 
 
 if __name__ == '__main__':
   sys.exit(main(sys.argv[1:]))