Changeset 78502 in webkit

Timestamp:

Feb 14, 2011 1:48:25 PM (13 years ago)

Author:

dpranke@chromium.org

Message:

2011-02-14 Dirk Pranke <​dpranke@chromium.org>

Reviewed by Tony Chang.

update the NRWT multiprocessing code to spawn multiple workers
instead of just using one.

​https://bugs.webkit.org/show_bug.cgi?id=54071

• Scripts/webkitpy/layout_tests/layout_package/test_runner2.py:

Location:

trunk/Tools

Files:

• ChangeLog (modified) (1 diff)
• Scripts/webkitpy/layout_tests/layout_package/test_runner2.py (modified) (5 diffs)

trunk/Tools/ChangeLog

@@ +1 @@
+2011-02-14  Dirk Pranke  <dpranke@chromium.org>
+
+        Reviewed by Tony Chang.
+
+        update the NRWT multiprocessing code to spawn multiple workers
+        instead of just using one.
+
+        https://bugs.webkit.org/show_bug.cgi?id=54071
+
+        * Scripts/webkitpy/layout_tests/layout_package/test_runner2.py:
+
 2011-02-12  Chang Shu  <cshu@webkit.org>
 

trunk/Tools/Scripts/webkitpy/layout_tests/layout_package/test_runner2.py

@@ -35 +35 @@
 
 import logging
+import time
 
+from webkitpy.tool import grammar
 
 from webkitpy.layout_tests.layout_package import manager_worker_broker
@@ -44 +46 @@
 
 
+class _WorkerState(object):
+    """A class for the TestRunner/manager to use to track the current state
+    of the workers."""
+    def __init__(self, number, worker_connection):
+        self.worker_connection = worker_connection
+        self.number = number
+        self.done = False
+
+    def __repr__(self):
+        return "_WorkerState(" + str(self.__dict__) + ")"
+
+
 class TestRunner2(test_runner.TestRunner):
     def __init__(self, port, options, printer):
@@ -50 +64 @@
         self._group_stats = {}
         self._current_result_summary = None
-        self._done = False
+
+        # This maps worker names to the state we are tracking for each of them.
+        self._worker_states = {}
 
     def is_done(self):
-        return self._done
+        worker_states = self._worker_states.values()
+        return worker_states and all(self._worker_is_done(worker_state) for worker_state in worker_states)
+
+    def _worker_is_done(self, worker_state):
+        # FIXME: check if the worker is wedged.
+        return worker_state.done
 
     def name(self):
@@ -73 +94 @@
         """
         self._current_result_summary = result_summary
+        self._all_results = []
+        self._worker_states = {}
 
-        # FIXME: shard properly.
+        num_workers = self._num_workers()
 
-        # FIXME: should shard_tests return a list of objects rather than tuples?
-        test_lists = self._shard_tests(file_list, False)
+        self._printer.print_update('Sharding tests ...')
+        test_lists = self._shard_tests(file_list,
+            num_workers > 1 and not self._options.experimental_fully_parallel)
+        _log.debug("Using %d shards" % len(test_lists))
 
-        manager_connection = manager_worker_broker.get(self._port, self._options, self, worker.Worker)
+        manager_connection = manager_worker_broker.get(self._port, self._options,
+                                                       self, worker.Worker)
 
-        # FIXME: start all of the workers.
-        manager_connection.start_worker(0)
+        self._printer.print_update('Starting %s ...' %
+                                   grammar.pluralize('worker', num_workers))
+        for worker_number in xrange(num_workers):
+            worker_connection = manager_connection.start_worker(worker_number)
+            worker_state = _WorkerState(worker_number, worker_connection)
+            self._worker_states[worker_connection.name] = worker_state
+
+            # FIXME: If we start workers up too quickly, DumpRenderTree appears
+            # to thrash on something and time out its first few tests. Until
+            # we can figure out what's going on, sleep a bit in between
+            # workers.
+            time.sleep(0.1)
 
         for test_list in test_lists:
             manager_connection.post_message('test_list', test_list[0], test_list[1])
 
-        manager_connection.post_message('stop')
+        # We post one 'stop' message for each worker. Because the stop message
+        # are sent after all of the tests, and because each worker will stop
+        # reading messsages after receiving a stop, we can be sure each
+        # worker will get a stop message and hence they will all shut down.
+        for i in xrange(num_workers):
+            manager_connection.post_message('stop')
 
         keyboard_interrupted = False
         interrupted = False
         if not self._options.dry_run:
-            while not self._check_if_done():
+            while not self.is_done():
+                # We loop with a timeout in order to be able to detect wedged threads.
                 manager_connection.run_message_loop(delay_secs=1.0)
 
+        # FIXME: handle exceptions, interrupts.
+
         # FIXME: implement stats.
+
         thread_timings = []
 
@@ -102 +147 @@
                 self._group_stats, self._all_results)
 
-    def _check_if_done(self):
-        """Returns true iff all the workers have either completed or wedged."""
-        # FIXME: implement to check for wedged workers.
-        return self._done
-
-    def handle_started_test(self, src, test_info, hang_timeout):
+    def handle_started_test(self, source, test_info, hang_timeout):
         # FIXME: implement
         pass
 
-    def handle_done(self, src):
-        # FIXME: implement properly to handle multiple workers.
-        self._done = True
-        pass
+    def handle_done(self, source):
+        worker_state = self._worker_states[source]
+        worker_state.done = True
 
-    def handle_exception(self, src, exception_info):
+    def handle_exception(self, source, exception_info):
         raise exception_info
 
-    def handle_finished_list(self, src, list_name, num_tests, elapsed_time):
+    def handle_finished_list(self, source, list_name, num_tests, elapsed_time):
         # FIXME: update stats
         pass
 
-    def handle_finished_test(self, src, result, elapsed_time):
+    def handle_finished_test(self, source, result, elapsed_time):
         self._update_summary_with_result(self._current_result_summary, result)