285b2e7b1fd49029f238ae85d54a250875dedd1f
[WebKit.git] / 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())