WebKitTools/Scripts/webkitpy/layout_tests/layout_package/message_broker.py

   1 # Copyright (C) 2010 Google Inc. All rights reserved.
   2 #
   3 # Redistribution and use in source and binary forms, with or without
   4 # modification, are permitted provided that the following conditions are
   5 # met:
   6 #
   7 #     * Redistributions of source code must retain the above copyright
   8 # notice, this list of conditions and the following disclaimer.
   9 #     * Redistributions in binary form must reproduce the above
  10 # copyright notice, this list of conditions and the following disclaimer
  11 # in the documentation and/or other materials provided with the
  12 # distribution.
  13 #     * Neither the name of Google Inc. nor the names of its
  14 # contributors may be used to endorse or promote products derived from
  15 # this software without specific prior written permission.
  16 #
  17 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  18 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  19 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  20 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  21 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  24 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  25 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  26 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  27 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  28
  29 """Module for handling messages and concurrency for run-webkit-tests.
  30
  31 Testing is accomplished by having a manager (TestRunner) gather all of the
  32 tests to be run, and sending messages to a pool of workers (TestShellThreads)
  33 to run each test. Each worker communicates with one driver (usually
  34 DumpRenderTree) to run one test at a time and then compare the output against
  35 what we expected to get.
  36
  37 This modules provides a message broker that connects the manager to the
  38 workers: it provides a messaging abstraction and message loops, and
  39 handles launching threads and/or processes depending on the
  40 requested configuration.
  41 """
  42
  43 import logging
  44 import sys
  45 import time
  46 import traceback
  47
  48 import dump_render_tree_thread
  49
  50 _log = logging.getLogger(__name__)
  51
  52
  53 def get(port, options):
  54     """Return an instance of a WorkerMessageBroker."""
  55     worker_model = options.worker_model
  56     if worker_model == 'inline':
  57         return _InlineBroker(port, options)
  58     if worker_model == 'threads':
  59         return _MultiThreadedBroker(port, options)
  60     raise ValueError('unsupported value for --worker-model: %s' % worker_model)
  61
  62
  63 class _WorkerMessageBroker(object):
  64     def __init__(self, port, options):
  65         self._port = port
  66         self._options = options
  67         self._num_workers = int(self._options.child_processes)
  68
  69         # This maps worker names to their TestShellThread objects.
  70         self._threads = {}
  71
  72     def start_workers(self, test_runner):
  73         """Starts up the pool of workers for running the tests.
  74
  75         Args:
  76             test_runner: a handle to the manager/TestRunner object
  77         """
  78         self._test_runner = test_runner
  79         for worker_number in xrange(self._num_workers):
  80             thread = self.start_worker(worker_number)
  81             self._threads[thread.name()] = thread
  82         return self._threads.values()
  83
  84     def start_worker(self, worker_number):
  85         # FIXME: Replace with something that isn't a thread.
  86         # Note: Don't start() the thread! If we did, it would actually
  87         # create another thread and start executing it, and we'd no longer
  88         # be single-threaded.
  89         return dump_render_tree_thread.TestShellThread(self._port,
  90             self._options, worker_number,
  91             self._test_runner._current_filename_queue,
  92             self._test_runner._result_queue)
  93
  94     def run_message_loop(self):
  95         """Loop processing messages until done."""
  96         raise NotImplementedError
  97
  98     def cancel_workers(self):
  99         """Cancel/interrupt any workers that are still alive."""
 100         pass
 101
 102     def cleanup(self):
 103         """Perform any necessary cleanup on shutdown."""
 104         pass
 105
 106
 107 class _InlineBroker(_WorkerMessageBroker):
 108     def run_message_loop(self):
 109         thread = self._threads.values()[0]
 110         thread.run_in_main_thread(self._test_runner,
 111                                   self._test_runner._current_result_summary)
 112         self._test_runner.update()
 113
 114
 115 class _MultiThreadedBroker(_WorkerMessageBroker):
 116     def start_worker(self, worker_number):
 117         thread = _WorkerMessageBroker.start_worker(self, worker_number)
 118         thread.start()
 119         return thread
 120
 121     def run_message_loop(self):
 122         # Loop through all the threads waiting for them to finish.
 123         some_thread_is_alive = True
 124         while some_thread_is_alive:
 125             some_thread_is_alive = False
 126             t = time.time()
 127             for thread in self._threads.values():
 128                 exception_info = thread.exception_info()
 129                 if exception_info is not None:
 130                     # Re-raise the thread's exception here to make it
 131                     # clear that testing was aborted. Otherwise,
 132                     # the tests that did not run would be assumed
 133                     # to have passed.
 134                     raise exception_info[0], exception_info[1], exception_info[2]
 135
 136                 if thread.isAlive():
 137                     some_thread_is_alive = True
 138                     next_timeout = thread.next_timeout()
 139                     if next_timeout and t > next_timeout:
 140                         log_wedged_worker(thread.name(), thread.id())
 141                         thread.clear_next_timeout()
 142
 143             self._test_runner.update()
 144
 145             if some_thread_is_alive:
 146                 time.sleep(0.01)
 147
 148     def cancel_workers(self):
 149         for thread in self._threads.values():
 150             thread.cancel()
 151
 152
 153 def log_wedged_worker(name, id):
 154     """Log information about the given worker state."""
 155     stack = _find_thread_stack(id)
 156     assert(stack is not None)
 157     _log.error("")
 158     _log.error("%s (tid %d) is wedged" % (name, id))
 159     _log_stack(stack)
 160     _log.error("")
 161
 162
 163 def _find_thread_stack(id):
 164     """Returns a stack object that can be used to dump a stack trace for
 165     the given thread id (or None if the id is not found)."""
 166     for thread_id, stack in sys._current_frames().items():
 167         if thread_id == id:
 168             return stack
 169     return None
 170
 171
 172 def _log_stack(stack):
 173     """Log a stack trace to log.error()."""
 174     for filename, lineno, name, line in traceback.extract_stack(stack):
 175         _log.error('File: "%s", line %d, in %s' % (filename, lineno, name))
 176         if line:
 177             _log.error('  %s' % line.strip())