Roll gtest-parallel cac6d27..a5a21a7

gtest-parallel

 #!/usr/bin/env python2
 # Copyright 2013 Google Inc. All rights reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
@@ skipping 29 matching lines @@
 # wait(p) will call p.terminate() and raise ProcessWasInterrupted.
 class SigintHandler(object):
   class ProcessWasInterrupted(Exception): pass
   sigint_returncodes = {-signal.SIGINT,  # Unix
                         -1073741510,     # Windows
                         }
   def __init__(self):
     self.__lock = threading.Lock()
     self.__processes = set()
     self.__got_sigint = False
-    signal.signal(signal.SIGINT, self.__sigint_handler)
+    signal.signal(signal.SIGINT, lambda signal_num, frame: self.interrupt())
   def __on_sigint(self):
     self.__got_sigint = True
     while self.__processes:
       try:
         self.__processes.pop().terminate()
       except OSError:
         pass
-  def __sigint_handler(self, signal_num, frame):
+  def interrupt(self):
     with self.__lock:
       self.__on_sigint()
   def got_sigint(self):
     with self.__lock:
       return self.__got_sigint
   def wait(self, p):
     with self.__lock:
       if self.__got_sigint:
         p.terminate()
       self.__processes.add(p)
@@ skipping 51 matching lines @@
   if sys.stdout.isatty():
     # stdout needs to be unbuffered since the output is interactive.
     sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)
 
   out = Outputter(sys.stdout)
   total_tests = 0
   finished_tests = 0
 
   tests = {}
   outputs = {}
+  started = []
+  finished = []
   failures = []
 
+  def print_tests(self, message, test_ids):
+    if test_ids:
+      self.out.permanent_line("%s (%s/%s):" %
+                              (message, len(test_ids), self.total_tests))
+      test_ids = sorted(test_ids, key=lambda test_id: self.tests[test_id])
+      for test_id in test_ids:
+        self.out.permanent_line(" %s: %s" % self.tests[test_id])
+
   def print_test_status(self, last_finished_test, time_ms):
     self.out.transient_line("[%d/%d] %s (%d ms)"
                             % (self.finished_tests, self.total_tests,
                                last_finished_test, time_ms))
 
   def handle_meta(self, job_id, args):
     (command, arg) = args.split(' ', 1)
     if command == "TEST":
       (binary, test) = arg.split(' ', 1)
       self.tests[job_id] = (binary, test.strip())
+    elif command == "START":
+      self.started.append(job_id)
     elif command == "EXIT":
       (exit_code, time_ms) = [int(x) for x in arg.split(' ', 1)]
       self.finished_tests += 1
+      self.finished.append(job_id)
       (binary, test) = self.tests[job_id]
       self.print_test_status(test, time_ms)
       if exit_code != 0:
-        self.failures.append(self.tests[job_id])
+        self.failures.append(job_id)
         with open(self.outputs[job_id]) as f:
           for line in f.readlines():
             self.out.permanent_line(line.rstrip())
         self.out.permanent_line(
           "[%d/%d] %s returned/aborted with exit code %d (%d ms)"
           % (self.finished_tests, self.total_tests, test, exit_code, time_ms))
     elif command == "TESTCNT":
       self.total_tests = int(arg.split(' ', 1)[1])
       self.out.transient_line("[0/%d] Running tests..." % self.total_tests)
 
   def logfile(self, job_id, name):
     self.outputs[job_id] = name
 
   def log(self, line):
     stdout_lock.acquire()
     (prefix, output) = line.split(' ', 1)
 
     assert prefix[-1] == ':'
     self.handle_meta(int(prefix[:-1]), output)
     stdout_lock.release()
 
   def end(self):
-    if self.failures:
-      self.out.permanent_line("FAILED TESTS (%d/%d):"
-                              % (len(self.failures), self.total_tests))
-      for (binary, test) in self.failures:
-        self.out.permanent_line(" " + binary + ": " + test)
+    self.print_tests("FAILED TESTS", self.failures)
+    interruptions = set(self.started) - set(self.finished)
+    self.print_tests("INTERRUPTED TESTS", interruptions)
     self.out.flush_transient_output()
 
 class RawFormat:
   def log(self, line):
     stdout_lock.acquire()
     sys.stdout.write(line + "\n")
     sys.stdout.flush()
     stdout_lock.release()
   def logfile(self, job_id, name):
     with open(name) as f:
@@ skipping 31 matching lines @@
   def dump_to_file_and_close(self):
     json.dump(self.test_results, self.json_dump_file)
     self.json_dump_file.close()
 
 class IgnoreTestResults(object):
   def log(self, test, result):
     pass
   def dump_to_file_and_close(self):
     pass
 
+class DummyTimer(object):
+  def start(self):
+    pass
+  def cancel(self):
+    pass
+
 # Record of test runtimes. Has built-in locking.
 class TestTimes(object):
   def __init__(self, save_file):
     "Create new object seeded with saved test times from the given file."
     self.__times = {}  # (test binary, test name) -> runtime in ms
 
     # Protects calls to record_test_time(); other calls are not
     # expected to be made concurrently.
     self.__lock = threading.Lock()
 
@@ skipping 70 matching lines @@
 parser.add_option('--shard_count', type='int', default=1,
                   help='total number of shards (for sharding test execution '
                        'between multiple machines)')
 parser.add_option('--shard_index', type='int', default=0,
                   help='zero-indexed number identifying this shard (for '
                        'sharding test execution between multiple machines)')
 parser.add_option('--dump_json_test_results', type='string', default=None,
                   help='Saves the results of the tests as a JSON machine-'
                        'readable file. The format of the file is specified at '
                        'https://www.chromium.org/developers/the-json-test-results-format')
+parser.add_option('--timeout', type='int', default=None,
+                  help='Interrupt all remaining processes after the given '
+                       'time (in seconds).')
 
 (options, binaries) = parser.parse_args()
 
 if binaries == []:
   parser.print_usage()
   sys.exit(1)
 
 logger = RawFormat()
 if options.format == 'raw':
   pass
 elif options.format == 'filter':
   logger = FilterFormat()
 else:
   parser.error("Unknown output format: " + options.format)
 
 if options.shard_count < 1:
   parser.error("Invalid number of shards: %d. Must be at least 1." %
                options.shard_count)
 if not (0 <= options.shard_index < options.shard_count):
   parser.error("Invalid shard index: %d. Must be between 0 and %d "
                "(less than the number of shards)." %
                (options.shard_index, options.shard_count - 1))
 
+timeout = (DummyTimer() if options.timeout is None
+           else threading.Timer(options.timeout, sigint_handler.interrupt))
+
 test_results = (IgnoreTestResults() if options.dump_json_test_results is None
                 else CollectTestResults(options.dump_json_test_results))
 
 # Find tests.
 save_file = os.path.join(os.path.expanduser("~"), ".gtest-parallel-times")
 times = TestTimes(save_file)
 tests = []
 for test_binary in binaries:
   command = [test_binary]
   if options.gtest_also_run_disabled_tests:
@@ skipping 61 matching lines @@
 # Remove files from old test runs.
 for logfile in os.listdir(options.output_dir):
   os.remove(os.path.join(options.output_dir, logfile))
 
 # Run the specified job. Return the elapsed time in milliseconds if
 # the job succeeds, or None if the job fails. (This ensures that
 # failing tests will run first the next time.)
 def run_job((command, job_id, test, test_index)):
   begin = time.time()
 
+  logger.log("%s: START " % job_id)
   test_name = re.sub('[^A-Za-z0-9]', '_', test) + '-' + str(test_index) + '.log'
   with open(os.path.join(options.output_dir, test_name), 'w') as log:
     sub = subprocess.Popen(command + ['--gtest_filter=' + test] +
                              ['--gtest_color=' + options.gtest_color],
                            stdout=log, stderr=log)
     try:
       code = sigint_handler.wait(sub)
     except sigint_handler.ProcessWasInterrupted:
       thread.exit()
     runtime_ms = int(1000 * (time.time() - begin))
@@ skipping 25 matching lines @@
     if job is None:
       return
     times.record_test_time(test_binary, test, run_job(job))
 
 def start_daemon(func):
   t = threading.Thread(target=func)
   t.daemon = True
   t.start()
   return t
 
-workers = [start_daemon(worker) for i in range(options.workers)]
+try:
+  timeout.start()
+  workers = [start_daemon(worker) for i in range(options.workers)]
+  [t.join() for t in workers]
+finally:
+  timeout.cancel()
 
-[t.join() for t in workers]
 logger.end()
 times.write_to_file(save_file)
 if options.print_test_times:
   ts = sorted((times.get_test_time(test_binary, test), test_binary, test)
               for (_, test_binary, test, _) in tests
               if times.get_test_time(test_binary, test) is not None)
   for (time_ms, test_binary, test) in ts:
     print "%8s %s" % ("%dms" % time_ms, test)
 
 test_results.dump_to_file_and_close()
 sys.exit(-signal.SIGINT if sigint_handler.got_sigint() else exit_code)