Merge WebKit at r73109: Initial merge by git.

Change-Id: I61f1a66d9642e3d8405d3ac6ccab2a53421c75d8

10 files changed, 537 insertions, 156 deletions

diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/dump_render_tree_thread.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/dump_render_tree_thread.py
index 88f493d..fdb8da6 100644
--- a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/dump_render_tree_thread.py
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/dump_render_tree_thread.py
@@ -48,7 +48,11 @@ import sys
 import thread
 import threading
 import time
-import traceback
+
+
+from webkitpy.layout_tests.test_types import image_diff
+from webkitpy.layout_tests.test_types import test_type_base
+from webkitpy.layout_tests.test_types import text_diff
 
 import test_failures
 import test_output
@@ -58,23 +62,6 @@ _log = logging.getLogger("webkitpy.layout_tests.layout_package."
                          "dump_render_tree_thread")
 
 
-def find_thread_stack(id):
-    """Returns a stack object that can be used to dump a stack trace for
-    the given thread id (or None if the id is not found)."""
-    for thread_id, stack in sys._current_frames().items():
-        if thread_id == id:
-            return stack
-    return None
-
-
-def log_stack(stack):
-    """Log a stack trace to log.error()."""
-    for filename, lineno, name, line in traceback.extract_stack(stack):
-        _log.error('File: "%s", line %d, in %s' % (filename, lineno, name))
-        if line:
-            _log.error('  %s' % line.strip())
-
-
 def _expected_test_output(port, filename):
     """Returns an expected TestOutput object."""
     return test_output.TestOutput(port.expected_text(filename),
@@ -82,7 +69,7 @@ def _expected_test_output(port, filename):
                                   port.expected_checksum(filename))
 
 def _process_output(port, options, test_input, test_types, test_args,
-                    test_output):
+                    test_output, worker_name):
     """Receives the output from a DumpRenderTree process, subjects it to a
     number of tests, and returns a list of failure types the test produced.
 
@@ -94,6 +81,7 @@ def _process_output(port, options, test_input, test_types, test_args,
       test_types: list of test types to subject the output to
       test_args: arguments to be passed to each test
       test_output: a TestOutput object containing the output of the test
+      worker_name: worker name for logging
 
     Returns: a TestResult object
     """
@@ -104,20 +92,18 @@ def _process_output(port, options, test_input, test_types, test_args,
     if test_output.timeout:
         failures.append(test_failures.FailureTimeout())
 
+    test_name = port.relative_test_filename(test_input.filename)
     if test_output.crash:
-        _log.debug("Stacktrace for %s:\n%s" % (test_input.filename,
-                                               test_output.error))
-        # Strip off "file://" since RelativeTestFilename expects
-        # filesystem paths.
-        filename = os.path.join(options.results_directory,
-                                port.relative_test_filename(
-                                test_input.filename))
+        _log.debug("%s Stacktrace for %s:\n%s" % (worker_name, test_name,
+                                                  test_output.error))
+        filename = os.path.join(options.results_directory, test_name)
         filename = os.path.splitext(filename)[0] + "-stack.txt"
         port.maybe_make_directory(os.path.split(filename)[0])
         with codecs.open(filename, "wb", "utf-8") as file:
             file.write(test_output.error)
     elif test_output.error:
-        _log.debug("Previous test output stderr lines:\n%s" % test_output.error)
+        _log.debug("%s %s output stderr lines:\n%s" % (worker_name, test_name,
+                                                       test_output.error))
 
     expected_test_output = _expected_test_output(port, test_input.filename)
 
@@ -161,7 +147,7 @@ def _should_fetch_expected_checksum(options):
     return options.pixel_tests and not (options.new_baseline or options.reset_results)
 
 
-def _run_single_test(port, options, test_input, test_types, test_args, driver):
+def _run_single_test(port, options, test_input, test_types, test_args, driver, worker_name):
     # FIXME: Pull this into TestShellThread._run().
 
     # The image hash is used to avoid doing an image dump if the
@@ -169,23 +155,23 @@ def _run_single_test(port, options, test_input, test_types, test_args, driver):
     # are generating a new baseline.  (Otherwise, an image from a
     # previous run will be copied into the baseline."""
     if _should_fetch_expected_checksum(options):
-        image_hash_to_driver = port.expected_checksum(test_input.filename)
-    else:
-        image_hash_to_driver = None
-    uri = port.filename_to_uri(test_input.filename)
-    test_output = driver.run_test(uri, test_input.timeout, image_hash_to_driver)
+        test_input.image_hash = port.expected_checksum(test_input.filename)
+    test_output = driver.run_test(test_input)
     return _process_output(port, options, test_input, test_types, test_args,
-                           test_output)
+                           test_output, worker_name)
 
 
 class SingleTestThread(threading.Thread):
     """Thread wrapper for running a single test file."""
 
-    def __init__(self, port, options, test_input, test_types, test_args):
+    def __init__(self, port, options, worker_number, worker_name,
+                 test_input, test_types, test_args):
         """
         Args:
           port: object implementing port-specific hooks
           options: command line argument object from optparse
+          worker_number: worker number for tests
+          worker_name: for logging
           test_input: Object containing the test filename and timeout
           test_types: A list of TestType objects to run the test output
               against.
@@ -199,6 +185,8 @@ class SingleTestThread(threading.Thread):
         self._test_types = test_types
         self._test_args = test_args
         self._driver = None
+        self._worker_number = worker_number
+        self._name = worker_name
 
     def run(self):
         self._covered_run()
@@ -206,12 +194,12 @@ class SingleTestThread(threading.Thread):
     def _covered_run(self):
         # FIXME: this is a separate routine to work around a bug
         # in coverage: see http://bitbucket.org/ned/coveragepy/issue/85.
-        self._driver = self._port.create_driver(self._test_args.png_path,
-                                                self._options)
+        self._driver = self._port.create_driver(self._worker_number)
         self._driver.start()
         self._test_result = _run_single_test(self._port, self._options,
                                              self._test_input, self._test_types,
-                                             self._test_args, self._driver)
+                                             self._test_args, self._driver,
+                                             self._name)
         self._driver.stop()
 
     def get_test_result(self):
@@ -254,29 +242,28 @@ class WatchableThread(threading.Thread):
 
 
 class TestShellThread(WatchableThread):
-    def __init__(self, port, options, filename_list_queue, result_queue,
-                 test_types, test_args):
+    def __init__(self, port, options, worker_number, worker_name,
+                 filename_list_queue, result_queue):
         """Initialize all the local state for this DumpRenderTree thread.
 
         Args:
           port: interface to port-specific hooks
           options: command line options argument from optparse
+          worker_number: identifier for a particular worker thread.
+          worker_name: for logging.
           filename_list_queue: A thread safe Queue class that contains lists
               of tuples of (filename, uri) pairs.
           result_queue: A thread safe Queue class that will contain
               serialized TestResult objects.
-          test_types: A list of TestType objects to run the test output
-              against.
-          test_args: A TestArguments object to pass to each TestType.
         """
         WatchableThread.__init__(self)
         self._port = port
         self._options = options
+        self._worker_number = worker_number
+        self._name = worker_name
         self._filename_list_queue = filename_list_queue
         self._result_queue = result_queue
         self._filename_list = []
-        self._test_types = test_types
-        self._test_args = test_args
         self._driver = None
         self._test_group_timing_stats = {}
         self._test_results = []
@@ -287,6 +274,12 @@ class TestShellThread(WatchableThread):
         self._http_lock_wait_begin = 0
         self._http_lock_wait_end = 0
 
+        self._test_types = []
+        for cls in self._get_test_type_classes():
+            self._test_types.append(cls(self._port,
+                                        self._options.results_directory))
+        self._test_args = self._get_test_args(worker_number)
+
         # Current group of tests we're running.
         self._current_group = None
         # Number of tests in self._current_group.
@@ -294,6 +287,20 @@ class TestShellThread(WatchableThread):
         # Time at which we started running tests from self._current_group.
         self._current_group_start_time = None
 
+    def _get_test_args(self, worker_number):
+        """Returns the tuple of arguments for tests and for DumpRenderTree."""
+        test_args = test_type_base.TestArguments()
+        test_args.new_baseline = self._options.new_baseline
+        test_args.reset_results = self._options.reset_results
+
+        return test_args
+
+    def _get_test_type_classes(self):
+        classes = [text_diff.TestTextDiff]
+        if self._options.pixel_tests:
+            classes.append(image_diff.ImageDiff)
+        return classes
+
     def get_test_group_timing_stats(self):
         """Returns a dictionary mapping test group to a tuple of
         (number of tests in that group, time to run the tests)"""
@@ -417,9 +424,9 @@ class TestShellThread(WatchableThread):
             batch_count += 1
             self._num_tests += 1
             if self._options.run_singly:
-                result = self._run_test_singly(test_input)
+                result = self._run_test_in_another_thread(test_input)
             else:
-                result = self._run_test(test_input)
+                result = self._run_test_in_this_thread(test_input)
 
             filename = test_input.filename
             tests_run_file.write(filename + "\n")
@@ -449,7 +456,7 @@ class TestShellThread(WatchableThread):
             if test_runner:
                 test_runner.update_summary(result_summary)
 
-    def _run_test_singly(self, test_input):
+    def _run_test_in_another_thread(self, test_input):
         """Run a test in a separate thread, enforcing a hard time limit.
 
         Since we can only detect the termination of a thread, not any internal
@@ -461,10 +468,11 @@ class TestShellThread(WatchableThread):
 
         Returns:
           A TestResult
-
         """
         worker = SingleTestThread(self._port,
                                   self._options,
+                                  self._worker_number,
+                                  self._name,
                                   test_input,
                                   self._test_types,
                                   self._test_args)
@@ -496,11 +504,11 @@ class TestShellThread(WatchableThread):
             _log.error('Cannot get results of test: %s' %
                        test_input.filename)
             result = test_results.TestResult(test_input.filename, failures=[],
-                test_run_time=0, total_time_for_all_diffs=0, time_for_diffs=0)
+                test_run_time=0, total_time_for_all_diffs=0, time_for_diffs={})
 
         return result
 
-    def _run_test(self, test_input):
+    def _run_test_in_this_thread(self, test_input):
         """Run a single test file using a shared DumpRenderTree process.
 
         Args:
@@ -514,7 +522,7 @@ class TestShellThread(WatchableThread):
         self._next_timeout = time.time() + thread_timeout
         test_result = _run_single_test(self._port, self._options, test_input,
                                        self._test_types, self._test_args,
-                                       self._driver)
+                                       self._driver, self._name)
         self._test_results.append(test_result)
         return test_result
 
@@ -527,9 +535,8 @@ class TestShellThread(WatchableThread):
         """
         # poll() is not threadsafe and can throw OSError due to:
         # http://bugs.python.org/issue1731717
-        if (not self._driver or self._driver.poll() is not None):
-            self._driver = self._port.create_driver(self._test_args.png_path,
-                                                    self._options)
+        if not self._driver or self._driver.poll() is not None:
+            self._driver = self._port.create_driver(self._worker_number)
             self._driver.start()
 
     def _start_servers_with_lock(self):
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/dump_render_tree_thread_unittest.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/dump_render_tree_thread_unittest.py
deleted file mode 100644
index 63f86d9..0000000
--- a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/dump_render_tree_thread_unittest.py
+++ /dev/null
@@ -1,49 +0,0 @@
-# Copyright (C) 2010 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
-# in the documentation and/or other materials provided with the
-# distribution.
-#    * Neither the name of Google Inc. nor the names of its
-# contributors may be used to endorse or promote products derived from
-# this software without specific prior written permission.
-#
-# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
-# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
-# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
-# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
-# 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.
-
-""""Tests code paths not covered by the regular unit tests."""
-
-import sys
-import unittest
-
-import dump_render_tree_thread
-
-
-class Test(unittest.TestCase):
-    def test_find_thread_stack_found(self):
-        id, stack = sys._current_frames().items()[0]
-        found_stack = dump_render_tree_thread.find_thread_stack(id)
-        self.assertNotEqual(found_stack, None)
-
-    def test_find_thread_stack_not_found(self):
-        found_stack = dump_render_tree_thread.find_thread_stack(0)
-        self.assertEqual(found_stack, None)
-
-
-if __name__ == '__main__':
-    unittest.main()
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_layout_results_generator.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_layout_results_generator.py
index 101d30b..b054c5b 100644
--- a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_layout_results_generator.py
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_layout_results_generator.py
@@ -129,6 +129,10 @@ class JSONLayoutResultsGenerator(json_results_generator.JSONResultsGeneratorBase
         return self.PASS_RESULT
 
     # override
+    def _get_result_char(self, test_name):
+        return self._get_modifier_char(test_name)
+
+    # override
     def _convert_json_to_current_version(self, results_json):
         archive_version = None
         if self.VERSION_KEY in results_json:
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_results_generator.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_results_generator.py
index 3267718..331e330 100644
--- a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_results_generator.py
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_results_generator.py
@@ -80,7 +80,7 @@ class TestResult(object):
 class JSONResultsGeneratorBase(object):
     """A JSON results generator for generic tests."""
 
-    MAX_NUMBER_OF_BUILD_RESULTS_TO_LOG = 1500
+    MAX_NUMBER_OF_BUILD_RESULTS_TO_LOG = 750
     # Min time (seconds) that will be added to the JSON.
     MIN_TIME = 1
     JSON_PREFIX = "ADD_RESULTS("
@@ -303,6 +303,23 @@ class JSONResultsGeneratorBase(object):
 
         return JSONResultsGenerator.PASS_RESULT
 
+    def _get_result_char(self, test_name):
+        """Returns a single char (e.g. SKIP_RESULT, FAIL_RESULT,
+        PASS_RESULT, NO_DATA_RESULT, etc) that indicates the test result
+        for the given test_name.
+        """
+        if test_name not in self._test_results_map:
+            return JSONResultsGenerator.NO_DATA_RESULT
+
+        test_result = self._test_results_map[test_name]
+        if test_result.modifier == TestResult.DISABLED:
+            return JSONResultsGenerator.SKIP_RESULT
+
+        if test_result.failed:
+            return JSONResultsGenerator.FAIL_RESULT
+
+        return JSONResultsGenerator.PASS_RESULT
+
     # FIXME: Callers should use scm.py instead.
     # FIXME: Identify and fix the run-time errors that were observed on Windows
     # chromium buildbot when we had updated this code to use scm.py once before.
@@ -484,7 +501,7 @@ class JSONResultsGeneratorBase(object):
           tests: Dictionary containing test result entries.
         """
 
-        result = self._get_modifier_char(test_name)
+        result = self._get_result_char(test_name)
         time = self._get_test_timing(test_name)
 
         if test_name not in tests:
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_results_generator_unittest.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_results_generator_unittest.py
index 606a613..d6275ee 100644
--- a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_results_generator_unittest.py
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/json_results_generator_unittest.py
@@ -56,15 +56,6 @@ class JSONGeneratorTest(unittest.TestCase):
         self._FLAKY_tests = set([])
         self._FAILS_tests = set([])
 
-    def _get_test_modifier(self, test_name):
-        if test_name.startswith('DISABLED_'):
-            return json_results_generator.JSONResultsGenerator.SKIP_RESULT
-        elif test_name.startswith('FLAKY_'):
-            return json_results_generator.JSONResultsGenerator.FLAKY_RESULT
-        elif test_name.startswith('FAILS_'):
-            return json_results_generator.JSONResultsGenerator.FAIL_RESULT
-        return json_results_generator.JSONResultsGenerator.PASS_RESULT
-
     def _test_json_generation(self, passed_tests_list, failed_tests_list):
         tests_set = set(passed_tests_list) | set(failed_tests_list)
 
@@ -74,9 +65,9 @@ class JSONGeneratorTest(unittest.TestCase):
                            if t.startswith('FLAKY_')])
         FAILS_tests = set([t for t in tests_set
                            if t.startswith('FAILS_')])
-        PASS_tests = tests_set ^ (DISABLED_tests | FLAKY_tests | FAILS_tests)
+        PASS_tests = tests_set - (DISABLED_tests | FLAKY_tests | FAILS_tests)
 
-        passed_tests = set(passed_tests_list) ^ DISABLED_tests
+        passed_tests = set(passed_tests_list) - DISABLED_tests
         failed_tests = set(failed_tests_list)
 
         test_timings = {}
@@ -180,10 +171,10 @@ class JSONGeneratorTest(unittest.TestCase):
                 test = tests[test_name]
 
                 failed = 0
-                modifier = self._get_test_modifier(test_name)
                 for result in test[JRG.RESULTS]:
-                    if result[1] == modifier:
+                    if result[1] == JRG.FAIL_RESULT:
                         failed = result[0]
+
                 self.assertEqual(1, failed)
 
                 timing_count = 0
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/message_broker.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/message_broker.py
new file mode 100644
index 0000000..e520a9c
--- /dev/null
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/message_broker.py
@@ -0,0 +1,197 @@
+# Copyright (C) 2010 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
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# 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.
+
+"""Module for handling messages, threads, processes, and concurrency for run-webkit-tests.
+
+Testing is accomplished by having a manager (TestRunner) gather all of the
+tests to be run, and sending messages to a pool of workers (TestShellThreads)
+to run each test. Each worker communicates with one driver (usually
+DumpRenderTree) to run one test at a time and then compare the output against
+what we expected to get.
+
+This modules provides a message broker that connects the manager to the
+workers: it provides a messaging abstraction and message loops, and
+handles launching threads and/or processes depending on the
+requested configuration.
+"""
+
+import logging
+import sys
+import time
+import traceback
+
+import dump_render_tree_thread
+
+_log = logging.getLogger(__name__)
+
+
+def get(port, options):
+    """Return an instance of a WorkerMessageBroker."""
+    worker_model = options.worker_model
+    if worker_model == 'inline':
+        return InlineBroker(port, options)
+    if worker_model == 'threads':
+        return MultiThreadedBroker(port, options)
+    raise ValueError('unsupported value for --worker-model: %s' % worker_model)
+
+
+class _WorkerState(object):
+    def __init__(self, name):
+        self.name = name
+        self.thread = None
+
+
+class WorkerMessageBroker(object):
+    def __init__(self, port, options):
+        self._port = port
+        self._options = options
+        self._num_workers = int(self._options.child_processes)
+
+        # This maps worker names to their _WorkerState values.
+        self._workers = {}
+
+    def _threads(self):
+        return tuple([w.thread for w in self._workers.values()])
+
+    def start_workers(self, test_runner):
+        """Starts up the pool of workers for running the tests.
+
+        Args:
+            test_runner: a handle to the manager/TestRunner object
+        """
+        self._test_runner = test_runner
+        for worker_number in xrange(self._num_workers):
+            worker = _WorkerState('worker-%d' % worker_number)
+            worker.thread = self._start_worker(worker_number, worker.name)
+            self._workers[worker.name] = worker
+        return self._threads()
+
+    def _start_worker(self, worker_number, worker_name):
+        raise NotImplementedError
+
+    def run_message_loop(self):
+        """Loop processing messages until done."""
+        raise NotImplementedError
+
+    def cancel_workers(self):
+        """Cancel/interrupt any workers that are still alive."""
+        pass
+
+    def cleanup(self):
+        """Perform any necessary cleanup on shutdown."""
+        pass
+
+
+class InlineBroker(WorkerMessageBroker):
+    def _start_worker(self, worker_number, worker_name):
+        # FIXME: Replace with something that isn't a thread.
+        thread = dump_render_tree_thread.TestShellThread(self._port,
+            self._options, worker_number, worker_name,
+            self._test_runner._current_filename_queue,
+            self._test_runner._result_queue)
+        # Note: Don't start() the thread! If we did, it would actually
+        # create another thread and start executing it, and we'd no longer
+        # be single-threaded.
+        return thread
+
+    def run_message_loop(self):
+        thread = self._threads()[0]
+        thread.run_in_main_thread(self._test_runner,
+                                  self._test_runner._current_result_summary)
+        self._test_runner.update()
+
+
+class MultiThreadedBroker(WorkerMessageBroker):
+    def _start_worker(self, worker_number, worker_name):
+        thread = dump_render_tree_thread.TestShellThread(self._port,
+            self._options, worker_number, worker_name,
+            self._test_runner._current_filename_queue,
+            self._test_runner._result_queue)
+        thread.start()
+        return thread
+
+    def run_message_loop(self):
+        threads = self._threads()
+
+        # Loop through all the threads waiting for them to finish.
+        some_thread_is_alive = True
+        while some_thread_is_alive:
+            some_thread_is_alive = False
+            t = time.time()
+            for thread in threads:
+                exception_info = thread.exception_info()
+                if exception_info is not None:
+                    # Re-raise the thread's exception here to make it
+                    # clear that testing was aborted. Otherwise,
+                    # the tests that did not run would be assumed
+                    # to have passed.
+                    raise exception_info[0], exception_info[1], exception_info[2]
+
+                if thread.isAlive():
+                    some_thread_is_alive = True
+                    next_timeout = thread.next_timeout()
+                    if next_timeout and t > next_timeout:
+                        log_wedged_worker(thread.getName(), thread.id())
+                        thread.clear_next_timeout()
+
+            self._test_runner.update()
+
+            if some_thread_is_alive:
+                time.sleep(0.01)
+
+    def cancel_workers(self):
+        threads = self._threads()
+        for thread in threads:
+            thread.cancel()
+
+
+def log_wedged_worker(name, id):
+    """Log information about the given worker state."""
+    stack = _find_thread_stack(id)
+    assert(stack is not None)
+    _log.error("")
+    _log.error("%s (tid %d) is wedged" % (name, id))
+    _log_stack(stack)
+    _log.error("")
+
+
+def _find_thread_stack(id):
+    """Returns a stack object that can be used to dump a stack trace for
+    the given thread id (or None if the id is not found)."""
+    for thread_id, stack in sys._current_frames().items():
+        if thread_id == id:
+            return stack
+    return None
+
+
+def _log_stack(stack):
+    """Log a stack trace to log.error()."""
+    for filename, lineno, name, line in traceback.extract_stack(stack):
+        _log.error('File: "%s", line %d, in %s' % (filename, lineno, name))
+        if line:
+            _log.error('  %s' % line.strip())
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/message_broker_unittest.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/message_broker_unittest.py
new file mode 100644
index 0000000..6f04fd3
--- /dev/null
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/message_broker_unittest.py
@@ -0,0 +1,183 @@
+# Copyright (C) 2010 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
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# 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 Queue
+import sys
+import thread
+import threading
+import time
+import unittest
+
+from webkitpy.common import array_stream
+from webkitpy.common.system import outputcapture
+from webkitpy.tool import mocktool
+
+from webkitpy.layout_tests import run_webkit_tests
+
+import message_broker
+
+
+class TestThread(threading.Thread):
+    def __init__(self, started_queue, stopping_queue):
+        threading.Thread.__init__(self)
+        self._thread_id = None
+        self._started_queue = started_queue
+        self._stopping_queue = stopping_queue
+        self._timeout = False
+        self._timeout_queue = Queue.Queue()
+        self._exception_info = None
+
+    def id(self):
+        return self._thread_id
+
+    def getName(self):
+        return "worker-0"
+
+    def run(self):
+        self._covered_run()
+
+    def _covered_run(self):
+        # FIXME: this is a separate routine to work around a bug
+        # in coverage: see http://bitbucket.org/ned/coveragepy/issue/85.
+        self._thread_id = thread.get_ident()
+        try:
+            self._started_queue.put('')
+            msg = self._stopping_queue.get()
+            if msg == 'KeyboardInterrupt':
+                raise KeyboardInterrupt
+            elif msg == 'Exception':
+                raise ValueError()
+            elif msg == 'Timeout':
+                self._timeout = True
+                self._timeout_queue.get()
+        except:
+            self._exception_info = sys.exc_info()
+
+    def exception_info(self):
+        return self._exception_info
+
+    def next_timeout(self):
+        if self._timeout:
+            self._timeout_queue.put('done')
+            return time.time() - 10
+        return time.time()
+
+    def clear_next_timeout(self):
+        self._next_timeout = None
+
+class TestHandler(logging.Handler):
+    def __init__(self, astream):
+        logging.Handler.__init__(self)
+        self._stream = astream
+
+    def emit(self, record):
+        self._stream.write(self.format(record))
+
+
+class MultiThreadedBrokerTest(unittest.TestCase):
+    class MockTestRunner(object):
+        def __init__(self):
+            pass
+
+        def __del__(self):
+            pass
+
+        def update(self):
+            pass
+
+    def run_one_thread(self, msg):
+        runner = self.MockTestRunner()
+        port = None
+        options = mocktool.MockOptions(child_processes='1')
+        starting_queue = Queue.Queue()
+        stopping_queue = Queue.Queue()
+        broker = message_broker.MultiThreadedBroker(port, options)
+        broker._test_runner = runner
+        child_thread = TestThread(starting_queue, stopping_queue)
+        broker._workers['worker-0'] = message_broker._WorkerState('worker-0')
+        broker._workers['worker-0'].thread = child_thread
+        child_thread.start()
+        started_msg = starting_queue.get()
+        stopping_queue.put(msg)
+        return broker.run_message_loop()
+
+    def test_basic(self):
+        interrupted = self.run_one_thread('')
+        self.assertFalse(interrupted)
+
+    def test_interrupt(self):
+        self.assertRaises(KeyboardInterrupt, self.run_one_thread, 'KeyboardInterrupt')
+
+    def test_timeout(self):
+        oc = outputcapture.OutputCapture()
+        oc.capture_output()
+        interrupted = self.run_one_thread('Timeout')
+        self.assertFalse(interrupted)
+        oc.restore_output()
+
+    def test_exception(self):
+        self.assertRaises(ValueError, self.run_one_thread, 'Exception')
+
+
+class Test(unittest.TestCase):
+    def test_find_thread_stack_found(self):
+        id, stack = sys._current_frames().items()[0]
+        found_stack = message_broker._find_thread_stack(id)
+        self.assertNotEqual(found_stack, None)
+
+    def test_find_thread_stack_not_found(self):
+        found_stack = message_broker._find_thread_stack(0)
+        self.assertEqual(found_stack, None)
+
+    def test_log_wedged_worker(self):
+        oc = outputcapture.OutputCapture()
+        oc.capture_output()
+        logger = message_broker._log
+        astream = array_stream.ArrayStream()
+        handler = TestHandler(astream)
+        logger.addHandler(handler)
+
+        starting_queue = Queue.Queue()
+        stopping_queue = Queue.Queue()
+        child_thread = TestThread(starting_queue, stopping_queue)
+        child_thread.start()
+        msg = starting_queue.get()
+
+        message_broker.log_wedged_worker(child_thread.getName(),
+                                         child_thread.id())
+        stopping_queue.put('')
+        child_thread.join(timeout=1.0)
+
+        self.assertFalse(astream.empty())
+        self.assertFalse(child_thread.isAlive())
+        oc.restore_output()
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/printing.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/printing.py
index fb9fe6d..7a6aad1 100644
--- a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/printing.py
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/printing.py
@@ -126,7 +126,6 @@ def print_options():
     ]
 
 
-
 def parse_print_options(print_options, verbose, child_processes,
                         is_fully_parallel):
     """Parse the options provided to --print and dedup and rank them.
@@ -182,8 +181,8 @@ def _configure_logging(stream, verbose):
     log_datefmt = '%y%m%d %H:%M:%S'
     log_level = logging.INFO
     if verbose:
-        log_fmt = ('%(asctime)s %(process)d %(filename)s:%(lineno)-4d %(levelname)s'
-                   '%(message)s')
+        log_fmt = ('%(asctime)s %(process)d %(filename)s:%(lineno)d '
+                   '%(levelname)s %(message)s')
         log_level = logging.DEBUG
 
     root = logging.getLogger()
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/printing_unittest.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/printing_unittest.py
index 9a0f4ee..27a6a29 100644
--- a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/printing_unittest.py
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/printing_unittest.py
@@ -78,8 +78,9 @@ class TestUtilityFunctions(unittest.TestCase):
         self.assertTrue(options is not None)
 
     def test_parse_print_options(self):
-        def test_switches(args, verbose, child_processes, is_fully_parallel,
-                          expected_switches_str):
+        def test_switches(args, expected_switches_str,
+                          verbose=False, child_processes=1,
+                          is_fully_parallel=False):
             options, args = get_options(args)
             if expected_switches_str:
                 expected_switches = set(expected_switches_str.split(','))
@@ -92,28 +93,23 @@ class TestUtilityFunctions(unittest.TestCase):
             self.assertEqual(expected_switches, switches)
 
         # test that we default to the default set of switches
-        test_switches([], False, 1, False,
-                      printing.PRINT_DEFAULT)
+        test_switches([], printing.PRINT_DEFAULT)
 
         # test that verbose defaults to everything
-        test_switches([], True, 1, False,
-                      printing.PRINT_EVERYTHING)
+        test_switches([], printing.PRINT_EVERYTHING, verbose=True)
 
         # test that --print default does what it's supposed to
-        test_switches(['--print', 'default'], False, 1, False,
-                      printing.PRINT_DEFAULT)
+        test_switches(['--print', 'default'], printing.PRINT_DEFAULT)
 
         # test that --print nothing does what it's supposed to
-        test_switches(['--print', 'nothing'], False, 1, False,
-                      None)
+        test_switches(['--print', 'nothing'], None)
 
         # test that --print everything does what it's supposed to
-        test_switches(['--print', 'everything'], False, 1, False,
-                      printing.PRINT_EVERYTHING)
+        test_switches(['--print', 'everything'], printing.PRINT_EVERYTHING)
 
         # this tests that '--print X' overrides '--verbose'
-        test_switches(['--print', 'actual'], True, 1, False,
-                      'actual')
+        test_switches(['--print', 'actual'], 'actual', verbose=True)
+
 
 
 class  Testprinter(unittest.TestCase):
diff --git a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/test_results_uploader.py b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/test_results_uploader.py
index 680b848..033c8c6 100644
--- a/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/test_results_uploader.py
+++ b/WebKitTools/Scripts/webkitpy/layout_tests/layout_package/test_results_uploader.py
@@ -27,45 +27,81 @@
 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+from __future__ import with_statement
+
+import codecs
 import mimetypes
 import socket
+import urllib2
 
 from webkitpy.common.net.networktransaction import NetworkTransaction
-from webkitpy.thirdparty.autoinstalled.mechanize import Browser
-
 
 def get_mime_type(filename):
-    return mimetypes.guess_type(filename)[0] or "text/plain"
+    return mimetypes.guess_type(filename)[0] or 'application/octet-stream'
+
+
+def _encode_multipart_form_data(fields, files):
+    """Encode form fields for multipart/form-data.
+
+    Args:
+      fields: A sequence of (name, value) elements for regular form fields.
+      files: A sequence of (name, filename, value) elements for data to be
+             uploaded as files.
+    Returns:
+      (content_type, body) ready for httplib.HTTP instance.
+
+    Source:
+      http://code.google.com/p/rietveld/source/browse/trunk/upload.py
+    """
+    BOUNDARY = '-M-A-G-I-C---B-O-U-N-D-A-R-Y-'
+    CRLF = '\r\n'
+    lines = []
+
+    for key, value in fields:
+        lines.append('--' + BOUNDARY)
+        lines.append('Content-Disposition: form-data; name="%s"' % key)
+        lines.append('')
+        if isinstance(value, unicode):
+            value = value.encode('utf-8')
+        lines.append(value)
+
+    for key, filename, value in files:
+        lines.append('--' + BOUNDARY)
+        lines.append('Content-Disposition: form-data; name="%s"; filename="%s"' % (key, filename))
+        lines.append('Content-Type: %s' % get_mime_type(filename))
+        lines.append('')
+        if isinstance(value, unicode):
+            value = value.encode('utf-8')
+        lines.append(value)
+
+    lines.append('--' + BOUNDARY + '--')
+    lines.append('')
+    body = CRLF.join(lines)
+    content_type = 'multipart/form-data; boundary=%s' % BOUNDARY
+    return content_type, body
 
 
 class TestResultsUploader:
     def __init__(self, host):
         self._host = host
-        self._browser = Browser()
 
     def _upload_files(self, attrs, file_objs):
-        self._browser.open("http://%s/testfile/uploadform" % self._host)
-        self._browser.select_form("test_result_upload")
-        for (name, data) in attrs:
-            self._browser[name] = str(data)
-
-        for (filename, handle) in file_objs:
-            self._browser.add_file(handle, get_mime_type(filename), filename,
-                "file")
-
-        self._browser.submit()
+        url = "http://%s/testfile/upload" % self._host
+        content_type, data = _encode_multipart_form_data(attrs, file_objs)
+        headers = {"Content-Type": content_type}
+        request = urllib2.Request(url, data, headers)
+        urllib2.urlopen(request)
 
     def upload(self, params, files, timeout_seconds):
-        orig_timeout = socket.getdefaulttimeout()
         file_objs = []
-        try:
-            file_objs = [(filename, open(path, "rb")) for (filename, path)
-                in files]
+        for filename, path in files:
+            with codecs.open(path, "rb") as file:
+                file_objs.append(('file', filename, file.read()))
 
+        orig_timeout = socket.getdefaulttimeout()
+        try:
             socket.setdefaulttimeout(timeout_seconds)
             NetworkTransaction(timeout_seconds=timeout_seconds).run(
                 lambda: self._upload_files(params, file_objs))
         finally:
             socket.setdefaulttimeout(orig_timeout)
-            for (filename, handle) in file_objs:
-                handle.close()