Add an additional content_shell per worker for running virtual tests.

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._default_driver_handler = None
+        self._virtual_driver_handler = 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
 
+        self._default_driver_handler = DriverHandler(
+            'default', self._worker_number, self._name, self._port)
+        self._virtual_driver_handler = DriverHandler(
+            'virtual', self._worker_number, self._name, self._port)
+
     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)
@@ skipping 13 matching lines @@
         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)
+        self._default_driver_handler.prepare()
+        is_virtual = self._port.is_virtual_test(test_input.test_name)
+        if is_virtual:
+            # Lazily initialize or restart virtual driver.
+            self._virtual_driver_handler.prepare()
 
-        if not self._driver:
+        if (not self._default_driver_handler.is_prepared() or
+                (is_virtual and not self._virtual_driver_handler.is_prepared())):
             # 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 = single_test_runner.run_single_test(
+            self._port, self._options, self._results_directory, self._name,
+            self._default_driver_handler.driver,
+            self._virtual_driver_handler.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()
+        if self._default_driver_handler:
+            self._default_driver_handler.kill()
+        if self._virtual_driver_handler:
+            self._virtual_driver_handler.kill()
 
     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()
+                self._default_driver_handler.kill()
+                if self._port.is_virtual_test(test_input.test_name):
+                    # FIXME: Need more information in failure reporting so
+                    # we know which driver needs to be restarted for virtual
+                    # tests. For now we restart both drivers.
+                    self._virtual_driver_handler.kill()
+
                 # 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 DriverHandler(object):
+    """Maintains the lifecycle of a driver for a given worker.

[Dirk Pranke, 2015/06/01 22:06:20]

I'm not seeing what the value of this class is; it seems like it is just
duplicating the logic that already exists in the Driver class?

In particular, the contract for the driver class says that driver.run_test() is
responsible for starting a new content_shell if and when it is needed, so the
logic you have here in prepare() seems unneeded?

I'm probably missing something in your thinking?

[joelo, 2015/06/02 00:07:04]

There's actually no new logic here -- the worker class above now needs to keep
track of the state of 2 drivers, so instead of duplicating the logic I
encapsulated it in a separate class.

prepare() also takes care of restarting the driver/content_shell if it has
crashed. According to it's docstring, driver.run_test() can leave the driver in
an 'indeterminate state' if a test crashes/times out. I also peeked at the code
and AFAICT it doesn't appear to handle restarting the driver.. 

[Dirk Pranke, 2015/06/02 19:06:04]

Oh, I see the confusion.

The docstring is wrong :( run_test() does not leave the child process in an
indeterminate state; if the test completed normally, the child is still running.
If the test crashed, the child is gone. If the test times out, the driver kills
the child (so it is also gone). (see driver.py:182).

run_test() is responsible for starting a child if need be, so I think prepare()
is doing nothing useful. Then, assuming that's true, your wrapper class doesn't
really do much except move lines around in the file. I don't think it's
encapsulating enough to be useful at that point.

Does that make sense?

(Also, please update the docstring to match reality as a part of this).

[joelo, 2015/06/02 23:49:17]

Ah, I see how the driver gets reset now. Let me remove the prepare logic and
make sure that it doesn't break anything (since it was already there before).

+
+    Attributes:
+        driver: The current Driver object, can be None.
+        label: Label for this driver, used for logging purposes.
+    """
+
+    def __init__(self, label, worker_number, worker_name, port):
+        self.driver = None
+        self.label = label
+        self._worker_number = worker_number
+        self._worker_name = worker_name
+        self._port = port
+
+    def _create(self):
+        # Create a driver if one doesn't exist.
+        if not self.driver:
+            self.driver = self._port.create_driver(self._worker_number)
+
+    def is_prepared(self):
+        """Returns true if the driver is initialized and in a good state."""
+        return self.driver and not self.driver.has_crashed()
+
+    def prepare(self):
+        """Prepares the driver for testing.
+
+        Makes sure the driver is properly initialized and in a good state,
+        restarting it if it has crashed.
+        """
+        if self.driver and self.driver.has_crashed():
+            self.kill()
+        if not self.driver:
+            self.driver = self._port.create_driver(self._worker_number)
+
+    def kill(self):
+        """Shuts down the driver."""
+        # Be careful about how and when we kill the driver; if driver.stop()
+        # raises an exception, this routine may get re-entered via __del__.
+        if self.driver:
+            driver = self.driver
+            self.driver = None
+            _log.debug(
+                "%s killing driver: %s" % (self._worker_name, self.label))
+            driver.stop()
+
+
 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)
@@ skipping 142 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