Add extra content_shell per worker for virtual tests, except on Android.

Tools/Scripts/webkitpy/layout_tests/controllers/layout_test_runner.py

 # 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
@@ skipping 10 matching lines @@
 # 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
 
@@ skipping 159 matching lines @@
         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._primary_driver = None
+        self._secondary_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._primary_driver = self._port.create_driver(self._worker_number)
+
+        if self._port.max_drivers_per_process() > 1:
+            self._secondary_driver = self._port.create_driver(self._worker_number)
 
         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
 
+        # Kill the secondary driver at the end of each test shard.
+        self._kill_driver(self._secondary_driver, 'secondary')
+
         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
-
         _log.debug("%s %s started" % (self._name, test_input.test_name))
         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 = single_test_runner.run_single_test(
+            self._port, self._options, self._results_directory, self._name,
+            self._primary_driver, self._secondary_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()
+        self._kill_driver(self._primary_driver, "primary")
+        self._kill_driver(self._secondary_driver, "secondary")
 
     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):
+    def _kill_driver(self, driver, label):
         # 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)
+            _log.debug("%s killing %s driver" % (self._name, label))
             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()
+                # FIXME: Need more information in failure reporting so
+                # we know which driver needs to be restarted. For now
+                # we kill both drivers.
+                self._kill_driver(self._primary_driver, "primary")
+                self._kill_driver(self._secondary_driver, "secondary")
+
                 # 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:
@@ skipping 155 matching lines @@
         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