# Copyright (C) 2011 Google Inc. All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are # met: # # * Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # * Redistributions in binary form must reproduce the above # copyright notice, this list of conditions and the following disclaimer # in the documentation and/or other materials provided with the # distribution. # * Neither the name of Google Inc. nor the names of its # contributors may be used to endorse or promote products derived from # this software without specific prior written permission. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import logging import math import threading import time from webkitpy.common import message_pool from webkitpy.layout_tests.controllers import single_test_runner from webkitpy.layout_tests.models.test_run_results import TestRunResults from webkitpy.layout_tests.models import test_expectations from webkitpy.layout_tests.models import test_failures from webkitpy.layout_tests.models import test_results from webkitpy.tool import grammar _log = logging.getLogger(__name__) TestExpectations = test_expectations.TestExpectations # Export this so callers don't need to know about message pools. WorkerException = message_pool.WorkerException class TestRunInterruptedException(Exception): """Raised when a test run should be stopped immediately.""" def __init__(self, reason): Exception.__init__(self) self.reason = reason self.msg = reason def __reduce__(self): return self.__class__, (self.reason,) class LayoutTestRunner(object): def __init__(self, options, port, printer, results_directory, test_is_slow_fn): self._options = options self._port = port self._printer = printer self._results_directory = results_directory self._test_is_slow = test_is_slow_fn self._sharder = Sharder(self._port.split_test, self._options.max_locked_shards) self._filesystem = self._port.host.filesystem self._expectations = None self._test_inputs = [] self._retrying = False self._current_run_results = None def run_tests(self, expectations, test_inputs, tests_to_skip, num_workers, retrying): self._expectations = expectations self._test_inputs = test_inputs self._retrying = retrying self._shards_to_redo = [] # FIXME: rename all variables to test_run_results or some such ... run_results = TestRunResults(self._expectations, len(test_inputs) + len(tests_to_skip)) self._current_run_results = run_results self._printer.num_tests = len(test_inputs) self._printer.num_completed = 0 if not retrying: self._printer.print_expected(run_results, self._expectations.get_tests_with_result_type) for test_name in set(tests_to_skip): result = test_results.TestResult(test_name) result.type = test_expectations.SKIP run_results.add(result, expected=True, test_is_slow=self._test_is_slow(test_name)) self._printer.write_update('Sharding tests ...') locked_shards, unlocked_shards = self._sharder.shard_tests(test_inputs, int(self._options.child_processes), self._options.fully_parallel, self._options.run_singly or (self._options.batch_size == 1)) # We don't have a good way to coordinate the workers so that they don't # try to run the shards that need a lock. The easiest solution is to # run all of the locked shards first. all_shards = locked_shards + unlocked_shards num_workers = min(num_workers, len(all_shards)) self._printer.print_workers_and_shards(num_workers, len(all_shards), len(locked_shards)) if self._options.dry_run: return run_results self._printer.write_update('Starting %s ...' % grammar.pluralize('worker', num_workers)) start_time = time.time() try: with message_pool.get(self, self._worker_factory, num_workers, self._port.host) as pool: pool.run(('test_list', shard.name, shard.test_inputs) for shard in all_shards) if self._shards_to_redo: num_workers -= len(self._shards_to_redo) if num_workers > 0: with message_pool.get(self, self._worker_factory, num_workers, self._port.host) as pool: pool.run(('test_list', shard.name, shard.test_inputs) for shard in self._shards_to_redo) except TestRunInterruptedException, e: _log.warning(e.reason) run_results.interrupted = True except KeyboardInterrupt: self._printer.flush() self._printer.writeln('Interrupted, exiting ...') run_results.keyboard_interrupted = True except Exception, e: _log.debug('%s("%s") raised, exiting' % (e.__class__.__name__, str(e))) raise finally: run_results.run_time = time.time() - start_time return run_results def _worker_factory(self, worker_connection): results_directory = self._results_directory if self._retrying: self._filesystem.maybe_make_directory(self._filesystem.join(self._results_directory, 'retries')) results_directory = self._filesystem.join(self._results_directory, 'retries') return Worker(worker_connection, results_directory, self._options) def _mark_interrupted_tests_as_skipped(self, run_results): for test_input in self._test_inputs: if test_input.test_name not in run_results.results_by_name: result = test_results.TestResult(test_input.test_name, [test_failures.FailureEarlyExit()]) # FIXME: We probably need to loop here if there are multiple iterations. # FIXME: Also, these results are really neither expected nor unexpected. We probably # need a third type of result. run_results.add(result, expected=False, test_is_slow=self._test_is_slow(test_input.test_name)) def _interrupt_if_at_failure_limits(self, run_results): # Note: The messages in this method are constructed to match old-run-webkit-tests # so that existing buildbot grep rules work. def interrupt_if_at_failure_limit(limit, failure_count, run_results, message): if limit and failure_count >= limit: message += " %d tests run." % (run_results.expected + run_results.unexpected) self._mark_interrupted_tests_as_skipped(run_results) raise TestRunInterruptedException(message) interrupt_if_at_failure_limit( self._options.exit_after_n_failures, run_results.unexpected_failures, run_results, "Exiting early after %d failures." % run_results.unexpected_failures) interrupt_if_at_failure_limit( self._options.exit_after_n_crashes_or_timeouts, run_results.unexpected_crashes + run_results.unexpected_timeouts, run_results, # This differs from ORWT because it does not include WebProcess crashes. "Exiting early after %d crashes and %d timeouts." % (run_results.unexpected_crashes, run_results.unexpected_timeouts)) def _update_summary_with_result(self, run_results, result): expected = self._expectations.matches_an_expected_result(result.test_name, result.type, self._options.pixel_tests or result.reftest_type, self._options.enable_sanitizer) exp_str = self._expectations.get_expectations_string(result.test_name) got_str = self._expectations.expectation_to_string(result.type) if result.device_failed: self._printer.print_finished_test(result, False, exp_str, "Aborted") return run_results.add(result, expected, self._test_is_slow(result.test_name)) self._printer.print_finished_test(result, expected, exp_str, got_str) self._interrupt_if_at_failure_limits(run_results) def handle(self, name, source, *args): method = getattr(self, '_handle_' + name) if method: return method(source, *args) raise AssertionError('unknown message %s received from %s, args=%s' % (name, source, repr(args))) def _handle_started_test(self, worker_name, test_input, test_timeout_sec): self._printer.print_started_test(test_input.test_name) def _handle_finished_test_list(self, worker_name, list_name): pass def _handle_finished_test(self, worker_name, result, log_messages=[]): self._update_summary_with_result(self._current_run_results, result) def _handle_device_failed(self, worker_name, list_name, remaining_tests): _log.warning("%s has failed" % worker_name) if remaining_tests: self._shards_to_redo.append(TestShard(list_name, remaining_tests)) class Worker(object): def __init__(self, caller, results_directory, options): self._caller = caller self._worker_number = caller.worker_number self._name = caller.name self._results_directory = results_directory self._options = options # The remaining fields are initialized in start() self._host = None self._port = None self._batch_size = None self._batch_count = None self._filesystem = None self._driver = None self._num_tests = 0 def __del__(self): self.stop() def start(self): """This method is called when the object is starting to be used and it is safe for the object to create state that does not need to be pickled (usually this means it is called in a child process).""" self._host = self._caller.host self._filesystem = self._host.filesystem self._port = self._host.port_factory.get(self._options.platform, self._options) self._batch_count = 0 self._batch_size = self._options.batch_size or 0 def handle(self, name, source, test_list_name, test_inputs): assert name == 'test_list' for i, test_input in enumerate(test_inputs): device_failed = self._run_test(test_input, test_list_name) if device_failed: self._caller.post('device_failed', test_list_name, test_inputs[i:]) self._caller.stop_running() return self._caller.post('finished_test_list', test_list_name) def _update_test_input(self, test_input): if test_input.reference_files is None: # Lazy initialization. test_input.reference_files = self._port.reference_files(test_input.test_name) if test_input.reference_files: test_input.should_run_pixel_test = True else: test_input.should_run_pixel_test = self._port.should_run_as_pixel_test(test_input) def _run_test(self, test_input, shard_name): self._batch_count += 1 stop_when_done = False if self._batch_size > 0 and self._batch_count >= self._batch_size: self._batch_count = 0 stop_when_done = True self._update_test_input(test_input) test_timeout_sec = self._timeout(test_input) start = time.time() device_failed = False if self._driver and self._driver.has_crashed(): self._kill_driver() if not self._driver: self._driver = self._port.create_driver(self._worker_number) if not self._driver: # FIXME: Is this the best way to handle a device crashing in the middle of the test, or should we create # a new failure type? device_failed = True return device_failed self._caller.post('started_test', test_input, test_timeout_sec) result = single_test_runner.run_single_test(self._port, self._options, self._results_directory, self._name, self._driver, test_input, stop_when_done) result.shard_name = shard_name result.worker_name = self._name result.total_run_time = time.time() - start result.test_number = self._num_tests self._num_tests += 1 self._caller.post('finished_test', result) self._clean_up_after_test(test_input, result) return result.device_failed def stop(self): _log.debug("%s cleaning up" % self._name) self._kill_driver() def _timeout(self, test_input): """Compute the appropriate timeout value for a test.""" # The driver watchdog uses 2.5x the timeout; we want to be # larger than that. We also add a little more padding if we're # running tests in a separate thread. # # Note that we need to convert the test timeout from a # string value in milliseconds to a float for Python. # FIXME: Can we just return the test_input.timeout now? driver_timeout_sec = 3.0 * float(test_input.timeout) / 1000.0 def _kill_driver(self): # Be careful about how and when we kill the driver; if driver.stop() # raises an exception, this routine may get re-entered via __del__. driver = self._driver self._driver = None if driver: _log.debug("%s killing driver" % self._name) driver.stop() def _clean_up_after_test(self, test_input, result): test_name = test_input.test_name if result.failures: # Check and kill the driver if we need to. if any([f.driver_needs_restart() for f in result.failures]): self._kill_driver() # Reset the batch count since the shell just bounced. self._batch_count = 0 # Print the error message(s). _log.debug("%s %s failed:" % (self._name, test_name)) for f in result.failures: _log.debug("%s %s" % (self._name, f.message())) elif result.type == test_expectations.SKIP: _log.debug("%s %s skipped" % (self._name, test_name)) else: _log.debug("%s %s passed" % (self._name, test_name)) class TestShard(object): """A test shard is a named list of TestInputs.""" def __init__(self, name, test_inputs): self.name = name self.test_inputs = test_inputs self.requires_lock = test_inputs[0].requires_lock def __repr__(self): return "TestShard(name='%s', test_inputs=%s, requires_lock=%s'" % (self.name, self.test_inputs, self.requires_lock) def __eq__(self, other): return self.name == other.name and self.test_inputs == other.test_inputs class Sharder(object): def __init__(self, test_split_fn, max_locked_shards): self._split = test_split_fn self._max_locked_shards = max_locked_shards def shard_tests(self, test_inputs, num_workers, fully_parallel, run_singly): """Groups tests into batches. This helps ensure that tests that depend on each other (aka bad tests!) continue to run together as most cross-tests dependencies tend to occur within the same directory. Return: Two list of TestShards. The first contains tests that must only be run under the server lock, the second can be run whenever. """ # FIXME: Move all of the sharding logic out of manager into its # own class or module. Consider grouping it with the chunking logic # in prepare_lists as well. if num_workers == 1: return self._shard_in_two(test_inputs) elif fully_parallel: return self._shard_every_file(test_inputs, run_singly) return self._shard_by_directory(test_inputs) def _shard_in_two(self, test_inputs): """Returns two lists of shards, one with all the tests requiring a lock and one with the rest. This is used when there's only one worker, to minimize the per-shard overhead.""" locked_inputs = [] unlocked_inputs = [] for test_input in test_inputs: if test_input.requires_lock: locked_inputs.append(test_input) else: unlocked_inputs.append(test_input) locked_shards = [] unlocked_shards = [] if locked_inputs: locked_shards = [TestShard('locked_tests', locked_inputs)] if unlocked_inputs: unlocked_shards = [TestShard('unlocked_tests', unlocked_inputs)] return locked_shards, unlocked_shards def _shard_every_file(self, test_inputs, run_singly): """Returns two lists of shards, each shard containing a single test file. This mode gets maximal parallelism at the cost of much higher flakiness.""" locked_shards = [] unlocked_shards = [] virtual_inputs = [] for test_input in test_inputs: # Note that we use a '.' for the shard name; the name doesn't really # matter, and the only other meaningful value would be the filename, # which would be really redundant. if test_input.requires_lock: locked_shards.append(TestShard('.', [test_input])) elif test_input.test_name.startswith('virtual') and not run_singly: # This violates the spirit of sharding every file, but in practice, since the # virtual test suites require a different commandline flag and thus a restart # of content_shell, it's too slow to shard them fully. virtual_inputs.append(test_input) else: unlocked_shards.append(TestShard('.', [test_input])) locked_virtual_shards, unlocked_virtual_shards = self._shard_by_directory(virtual_inputs) # The locked shards still need to be limited to self._max_locked_shards in order to not # overload the http server for the http tests. return (self._resize_shards(locked_virtual_shards + locked_shards, self._max_locked_shards, 'locked_shard'), unlocked_virtual_shards + unlocked_shards) def _shard_by_directory(self, test_inputs): """Returns two lists of shards, each shard containing all the files in a directory. This is the default mode, and gets as much parallelism as we can while minimizing flakiness caused by inter-test dependencies.""" locked_shards = [] unlocked_shards = [] unlocked_slow_shards = [] tests_by_dir = {} # FIXME: Given that the tests are already sorted by directory, # we can probably rewrite this to be clearer and faster. for test_input in test_inputs: directory = self._split(test_input.test_name)[0] tests_by_dir.setdefault(directory, []) tests_by_dir[directory].append(test_input) for directory, test_inputs in tests_by_dir.iteritems(): shard = TestShard(directory, test_inputs) if test_inputs[0].requires_lock: locked_shards.append(shard) # In practice, virtual test suites are slow to run. It's a bit hacky, but # put them first since they're the long-tail of test runtime. elif directory.startswith('virtual'): unlocked_slow_shards.append(shard) else: unlocked_shards.append(shard) # Sort the shards by directory name. locked_shards.sort(key=lambda shard: shard.name) unlocked_slow_shards.sort(key=lambda shard: shard.name) unlocked_shards.sort(key=lambda shard: shard.name) # Put a ceiling on the number of locked shards, so that we # don't hammer the servers too badly. # FIXME: For now, limit to one shard or set it # with the --max-locked-shards. After testing to make sure we # can handle multiple shards, we should probably do something like # limit this to no more than a quarter of all workers, e.g.: # return max(math.ceil(num_workers / 4.0), 1) return (self._resize_shards(locked_shards, self._max_locked_shards, 'locked_shard'), unlocked_slow_shards + unlocked_shards) def _resize_shards(self, old_shards, max_new_shards, shard_name_prefix): """Takes a list of shards and redistributes the tests into no more than |max_new_shards| new shards.""" # This implementation assumes that each input shard only contains tests from a # single directory, and that tests in each shard must remain together; as a # result, a given input shard is never split between output shards. # # Each output shard contains the tests from one or more input shards and # hence may contain tests from multiple directories. def divide_and_round_up(numerator, divisor): return int(math.ceil(float(numerator) / divisor)) def extract_and_flatten(shards): test_inputs = [] for shard in shards: test_inputs.extend(shard.test_inputs) return test_inputs def split_at(seq, index): return (seq[:index], seq[index:]) num_old_per_new = divide_and_round_up(len(old_shards), max_new_shards) new_shards = [] remaining_shards = old_shards while remaining_shards: some_shards, remaining_shards = split_at(remaining_shards, num_old_per_new) new_shards.append(TestShard('%s_%d' % (shard_name_prefix, len(new_shards) + 1), extract_and_flatten(some_shards))) return new_shards