Changeset 78398 in webkit

Timestamp:

Feb 11, 2011 5:42:00 PM (13 years ago)

Author:

dpranke@chromium.org

Message:

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

Reviewed by Tony Chang.

This patch adds to NRWT most of the support needed to run the new
message-based workers in separate threads or processes. The code
isn't fully complete yet because we don't support cancel() or
is_alive().

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

• Scripts/webkitpy/layout_tests/layout_package/manager_worker_broker.py:
• Scripts/webkitpy/layout_tests/layout_package/manager_worker_broker_unittest.py:
• Scripts/webkitpy/layout_tests/layout_package/test_runner2.py:
• Scripts/webkitpy/layout_tests/port/base.py:
• Scripts/webkitpy/layout_tests/port/mock_drt.py:
• Scripts/webkitpy/layout_tests/run_webkit_tests_unittest.py:

Location:

trunk/Tools

Files:

• ChangeLog (modified) (1 diff)
• Scripts/webkitpy/layout_tests/layout_package/manager_worker_broker.py (modified) (5 diffs)
• Scripts/webkitpy/layout_tests/layout_package/manager_worker_broker_unittest.py (modified) (5 diffs)
• Scripts/webkitpy/layout_tests/layout_package/test_runner2.py (modified) (1 diff)
• Scripts/webkitpy/layout_tests/port/base.py (modified) (1 diff)
• Scripts/webkitpy/layout_tests/port/mock_drt.py (modified) (2 diffs)
• Scripts/webkitpy/layout_tests/run_webkit_tests_unittest.py (modified) (1 diff)

trunk/Tools/ChangeLog

@@ +1 @@
+2011-02-11  Dirk Pranke  <dpranke@chromium.org>
+
+        Reviewed by Tony Chang.
+
+        This patch adds to NRWT most of the support needed to run the new
+        message-based workers in separate threads or processes. The code
+        isn't fully complete yet because we don't support cancel() or
+        is_alive().
+
+        https://bugs.webkit.org/show_bug.cgi?id=54070
+
+        * Scripts/webkitpy/layout_tests/layout_package/manager_worker_broker.py:
+        * Scripts/webkitpy/layout_tests/layout_package/manager_worker_broker_unittest.py:
+        * Scripts/webkitpy/layout_tests/layout_package/test_runner2.py:
+        * Scripts/webkitpy/layout_tests/port/base.py:
+        * Scripts/webkitpy/layout_tests/port/mock_drt.py:
+        * Scripts/webkitpy/layout_tests/run_webkit_tests_unittest.py:
+
 2011-02-11  Sailesh Agrawal  <sail@chromium.org>
 

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

@@ -47 +47 @@
 import optparse
 import Queue
+import thread
 import threading
+import time
+
 
 # Handle Python < 2.6 where multiprocessing isn't available.
+#
+# _Multiprocessing_Process is needed so that _MultiProcessWorker
+# can be defined with or without multiprocessing.
 try:
     import multiprocessing
+    _Multiprocessing_Process = multiprocessing.Process
 except ImportError:
     multiprocessing = None
-
-
+    _Multiprocessing_Process = threading.Thread
+
+
+from webkitpy.layout_tests import port
 from webkitpy.layout_tests.layout_package import message_broker2
 
@@ -165 +174 @@
 
     def start_worker(self, worker_number):
-        self._inline_worker = _InlineWorker(self._broker, self._port, self._client,
-            self._worker_class, worker_number)
+        self._inline_worker = _InlineWorkerConnection(self._broker, self._port,
+            self._client, self._worker_class, worker_number)
         return self._inline_worker
 
@@ -178 +187 @@
 class _ThreadedManager(_ManagerConnection):
     def __init__(self, broker, port, options, client, worker_class):
-        raise NotImplementedError
+        _ManagerConnection.__init__(self, broker, options, client, worker_class)
+        self._port = port
+
+    def start_worker(self, worker_number):
+        worker_connection = _ThreadedWorkerConnection(self._broker, self._port,
+            self._worker_class, worker_number)
+        worker_connection.start()
+        return worker_connection
 
 
 class _MultiProcessManager(_ManagerConnection):
     def __init__(self, broker, port, options, client, worker_class):
-        raise NotImplementedError
+        # Note that this class does not keep a handle to the actual port
+        # object, because it isn't Picklable. Instead it keeps the port
+        # name and recreates the port in the child process from the name
+        # and options.
+        _ManagerConnection.__init__(self, broker, options, client, worker_class)
+        self._platform_name = port.real_name()
+
+    def start_worker(self, worker_number):
+        worker_connection = _MultiProcessWorkerConnection(self._broker, self._platform_name,
+            self._worker_class, worker_number, self._options)
+        worker_connection.start()
+        return worker_connection
 
 
@@ -193 +220 @@
                                                   ANY_WORKER_TOPIC, MANAGER_TOPIC)
 
-    def run(self):
-        raise NotImplementedError
-
     def yield_to_broker(self):
         pass
 
 
-class _InlineWorker(_WorkerConnection):
+class _InlineWorkerConnection(_WorkerConnection):
     def __init__(self, broker, port, manager_client, worker_class, worker_number):
         _WorkerConnection.__init__(self, broker, worker_class, worker_number, port._options)
@@ -211 +235 @@
     def yield_to_broker(self):
         self._broker.run_all_pending(MANAGER_TOPIC, self._manager_client)
+
+
+class _Thread(threading.Thread):
+    def __init__(self, worker_connection, port, client):
+        threading.Thread.__init__(self)
+        self._worker_connection = worker_connection
+        self._port = port
+        self._client = client
+
+    def run(self):
+        # FIXME: We can remove this once everyone is on 2.6.
+        if not hasattr(self, 'ident'):
+            self.ident = thread.get_ident()
+        self._client.run(self._port)
+
+
+class _ThreadedWorkerConnection(_WorkerConnection):
+    def __init__(self, broker, port, worker_class, worker_number):
+        _WorkerConnection.__init__(self, broker, worker_class, worker_number, port._options)
+        self._thread = _Thread(self, port, self._client)
+
+    def start(self):
+        self._thread.start()
+
+
+class _Process(_Multiprocessing_Process):
+    def __init__(self, worker_connection, platform_name, options, client):
+        _Multiprocessing_Process.__init__(self)
+        self._worker_connection = worker_connection
+        self._platform_name = platform_name
+        self._options = options
+        self._client = client
+
+    def run(self):
+        logging.basicConfig()
+        port_obj = port.get(self._platform_name, self._options)
+        self._client.run(port_obj)
+
+
+class _MultiProcessWorkerConnection(_WorkerConnection):
+    def __init__(self, broker, platform_name, worker_class, worker_number, options):
+        _WorkerConnection.__init__(self, broker, worker_class, worker_number, options)
+        self._proc = _Process(self, platform_name, options, self._client)
+
+    def start(self):
+        self._proc.start()

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

@@ -28 +28 @@
 
 import optparse
+import Queue
 import sys
 import unittest
@@ -44 +45 @@
 
 
-class TestWorker(object):
-    pass
+class TestWorker(manager_worker_broker.AbstractWorker):
+    def __init__(self, broker_connection, worker_number, options):
+        self._broker_connection = broker_connection
+        self._options = options
+        self._worker_number = worker_number
+        self._name = 'TestWorker/%d' % worker_number
+        self._stopped = False
+
+    def handle_stop(self, src):
+        self._stopped = True
+
+    def handle_test(self, src, an_int, a_str):
+        assert an_int == 1
+        assert a_str == "hello, world"
+        self._broker_connection.post_message('test', 2, 'hi, everybody')
+
+    def is_done(self):
+        return self._stopped
+
+    def name(self):
+        return self._name
+
+    def start(self):
+        pass
+
+    def run(self, port):
+        try:
+            self._broker_connection.run_message_loop()
+            self._broker_connection.yield_to_broker()
+            self._broker_connection.post_message('done')
+        except Exception, e:
+            self._broker_connection.post_message('exception', (type(e), str(e), None))
 
 
@@ -66 +97 @@
 
     def test_get__threads(self):
-        self.assertRaises(NotImplementedError, make_broker, self, 'threads')
+        self.assertTrue(make_broker(self, 'threads') is not None)
 
     def test_get__processes(self):
         if multiprocessing:
-            self.assertRaises(NotImplementedError, make_broker, self, 'processes')
+            self.assertTrue(make_broker(self, 'processes') is not None)
         else:
             self.assertRaises(ValueError, make_broker, self, 'processes')
@@ -77 +108 @@
         self.assertRaises(ValueError, make_broker, self, 'unknown')
 
+
+class _TestsMixin(object):
+    """Mixin class that implements a series of tests to enforce the
+    contract all implementations must follow."""
+
+    #
+    # Methods to implement the Manager side of the ClientInterface
+    #
+    def name(self):
+        return 'Tester'
+
+    def is_done(self):
+        return self._done
+
+    #
+    # Handlers for the messages the TestWorker may send.
+    #
+    def handle_done(self, src):
+        self._done = True
+
+    def handle_test(self, src, an_int, a_str):
+        self._an_int = an_int
+        self._a_str = a_str
+
+    def handle_exception(self, src, exc_info):
+        self._exception = exc_info
+        self._done = True
+
+    #
+    # Testing helper methods
+    #
+    def setUp(self):
+        self._an_int = None
+        self._a_str = None
+        self._broker = None
+        self._done = False
+        self._exception = None
+        self._worker_model = None
+
+    def make_broker(self):
+        self._broker = make_broker(self, self._worker_model)
+
+    #
+    # Actual unit tests
+    #
+    def test_done(self):
+        if not self._worker_model:
+            return
+        self.make_broker()
+        worker = self._broker.start_worker(0)
+        self._broker.post_message('test', 1, 'hello, world')
+        self._broker.post_message('stop')
+        self._broker.run_message_loop()
+        self.assertTrue(self.is_done())
+        self.assertEqual(self._an_int, 2)
+        self.assertEqual(self._a_str, 'hi, everybody')
+
+    def test_unknown_message(self):
+        if not self._worker_model:
+            return
+        self.make_broker()
+        worker = self._broker.start_worker(0)
+        self._broker.post_message('unknown')
+        self._broker.run_message_loop()
+
+        self.assertTrue(self.is_done())
+        self.assertEquals(self._exception[0], ValueError)
+        self.assertEquals(self._exception[1],
+            "TestWorker/0: received message 'unknown' it couldn't handle")
+
+
+class InlineBrokerTests(_TestsMixin, unittest.TestCase):
+    def setUp(self):
+        _TestsMixin.setUp(self)
+        self._worker_model = 'inline'
+
+
+class MultiProcessBrokerTests(_TestsMixin, unittest.TestCase):
+    def setUp(self):
+        _TestsMixin.setUp(self)
+        if multiprocessing:
+            self._worker_model = 'processes'
+        else:
+            self._worker_model = None
+
+    def queue(self):
+        return multiprocessing.Queue()
+
+
+class ThreadedBrokerTests(_TestsMixin, unittest.TestCase):
+    def setUp(self):
+        _TestsMixin.setUp(self)
+        self._worker_model = 'threads'
+
+
+class FunctionsTest(unittest.TestCase):
     def test_runtime_options(self):
         option_list = manager_worker_broker.runtime_options()
@@ -83 +210 @@
         self.assertTrue(options)
 
-# FIXME: Add in unit tests for the managers, coverage tests for the interfaces.
+
+class InterfaceTest(unittest.TestCase):
+    # These tests mostly exist to pacify coverage.
+
+    # FIXME: There must be a better way to do this and also verify
+    # that classes do implement every abstract method in an interface.
+    def test_managerconnection_is_abstract(self):
+        # Test that all the base class methods are abstract and have the
+        # signature we expect.
+        broker = make_broker(self, 'inline')
+        obj = manager_worker_broker._ManagerConnection(broker._broker, None, self, None)
+        self.assertRaises(NotImplementedError, obj.start_worker, 0)
 
 

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

@@ -46 +46 @@
 class TestRunner2(test_runner.TestRunner):
     def __init__(self, port, options, printer):
-        if options.worker_model in ('threads', 'processes'):
-            raise ValueError('--worker-model=%s not supported yet' % options.worker_model)
-
         test_runner.TestRunner.__init__(self, port, options, printer)
         self._all_results = []

trunk/Tools/Scripts/webkitpy/layout_tests/port/base.py

@@ -468 +468 @@
         return 'cpu'
 
+    def real_name(self):
+        """Returns the actual name of the port, not the delegate's."""
+        return self.name()
+
     def get_option(self, name, default_value=None):
         # FIXME: Eventually we should not have to do a test for

trunk/Tools/Scripts/webkitpy/layout_tests/port/mock_drt.py

@@ -54 +54 @@
             kwargs['port_name'] = kwargs['port_name'][len(prefix):]
         self.__delegate = factory.get(**kwargs)
+        self.__real_name = prefix + self.__delegate.name()
+
+    def real_name(self):
+        return self.__real_name
 
     def __getattr__(self, name):
@@ -272 +276 @@
 
 
-
 if __name__ == '__main__':
     fs = filesystem.FileSystem()

trunk/Tools/Scripts/webkitpy/layout_tests/run_webkit_tests_unittest.py

@@ -468 +468 @@
 
     def test_worker_model__processes(self):
-        self.assertRaises(ValueError, logging_run, ['--worker-model', 'processes'])
+        if compare_version(sys, '2.6')[0] >= 0:
+            self.assertTrue(passing_run(['--worker-model', 'processes']))
 
     def test_worker_model__threads(self):
-        self.assertRaises(ValueError, logging_run, ['--worker-model', 'threads'])
+        self.assertTrue(passing_run(['--worker-model', 'threads']))
 
     def test_worker_model__unknown(self):