[dosep] Run tests in a more parallel fashion

[labath]

Summary

Due to in-tree builds, we were parallelizing the tests at the directory
level. Now that the tests are built out-of-tree, we can remove this
limitation and paralelize at file level instead.

This decreases test suite time by about 10% for me, which is not
world-shattering, but it makes the code slightly simpler and will also
allow us to merge tests which were artificially spread over multiple
folders (TestConcurrentEvents...) to work-around this limitation.

To make this work, I've also needed to include the test file name in the
build directory name, as just the test method name is not unique enough
(plenty of tests have a test method called "test" or similar).

While doing this, I've found a couple of tests that are taking waaay longer then
they ought to (TestBreakpointCaseSensitivity -- 90 seconds), which I plan to
look into in the future.

[aprantl]

This is very cool! I imagine that this will help a lot with TestIntegerTypes and other really long-running ones. Thanks.

[jingham]

Very nice! Thanks for working on this!

lldb/trunk/packages/Python/lldbsuite/test/dosep.py

send_inferior_post_run_events(
         command,
         worker_index,
         process_driver,
         test_filename)
 
     return process_driver.results
 
 
-def process_dir(root, files, dotest_argv, inferior_pid_events):
-    """Examine a directory for tests, and invoke any found within it."""
+def process_file(test_file, dotest_argv, inferior_pid_events):
+    """Run tests in the specified file in a subprocess and gather the results."""
     results = []
-    for (base_name, full_test_path) in files:
-        import __main__ as main
-        global dotest_options
-        if dotest_options.p and not re.search(dotest_options.p, base_name):
-            continue
+    base_name = os.path.basename(test_file)
+
+        return
 
-        script_file = main.__file__
-        command = ([sys.executable, script_file] +
-                   dotest_argv +
-                   ["-S", dotest_options.session_file_format] +
-                   ["--inferior", "-p", base_name, root])
+               ["--inferior", "-p", base_name, os.path.dirname(test_file)])
 
-        timeout_name = os.path.basename(os.path.splitext(base_name)[0]).upper()
 
-        timeout = (os.getenv("LLDB_%s_TIMEOUT" % timeout_name) or
-                   getDefaultTimeout(dotest_options.lldb_platform_name))
 
-        results.append(call_with_timeout(
-            command, timeout, base_name, inferior_pid_events, full_test_path))
+        command, timeout, base_name, inferior_pid_events, test_file))
 
     # result = (name, status, passes, failures, unexpected_successes)
     timed_out = [name for name, status, _, _, _ in results
                  if status == eTimedOut]
     passed = [name for name, status, _, _, _ in results
               if status == ePassed]
     failed = [name for name, status, _, _, _ in results
               if status != ePassed]
def process_dir_worker_multiprocessing(
     setup_global_variables(
         a_output_lock, a_test_counter, a_total_tests, a_test_name_len,
         a_dotest_options, worker_index_map)
 
     # Keep grabbing entries from the queue until done.
     while not job_queue.empty():
         try:
             job = job_queue.get(block=False)
-            result = process_dir(job[0], job[1], job[2],
+            result = process_file(job[0], job[1], job[2],
                                  inferior_pid_events)
             result_queue.put(result)
         except queue.Empty:
             # Fine, we're done.
             pass
 
 
-def process_dir_worker_multiprocessing_pool(args):
-    return process_dir(*args)
+def process_file_worker_multiprocessing_pool(args):
+    return process_file(*args)
 
 
-def process_dir_worker_threading(job_queue, result_queue, inferior_pid_events):
+def process_file_worker_threading(job_queue, result_queue, inferior_pid_events):
     """Worker thread main loop when in threading mode.
 
     This one supports the hand-rolled pooling support.
 
     Takes one directory specification at a time and works on it."""
 
     # Keep grabbing entries from the queue until done.
     while not job_queue.empty():
         try:
             job = job_queue.get(block=False)
-            result = process_dir(job[0], job[1], job[2],
-                                 inferior_pid_events)
+            result = process_file(job[0], job[1], inferior_pid_events)
             result_queue.put(result)
         except queue.Empty:
             # Fine, we're done.
             pass
 
 
-def process_dir_worker_threading_pool(args):
-    return process_dir(*args)
+def process_file_worker_threading_pool(args):
+    return process_file(*args)
 
 
-def process_dir_mapper_inprocess(args):
+def process_file_mapper_inprocess(args):
     """Map adapter for running the subprocess-based, non-threaded test runner.
 
     @param args the process work item tuple
     @return the test result tuple
     """
-    return process_dir(*args)
+    return process_file(*args)
 
 
 def collect_active_pids_from_pid_events(event_queue):
     """
     Returns the set of what should be active inferior pids based on
     the event stream.
 
     @param event_queue a multiprocessing.Queue containing events of the
def kill_all_worker_threads(workers, inferior_pid_events):
 
     # We don't have a way to nuke the threads.  However, since we killed
     # all the inferiors, and we drained the job queue, this will be
     # good enough.  Wait cleanly for each worker thread to wrap up.
     for worker in workers:
         worker.join()
 
 
-def find_test_files_in_dir_tree(dir_root, found_func):
-    """Calls found_func for all the test files in the given dir hierarchy.
+def find_test_files_in_dir_tree(dir_root):
+    """Returns all the test files in the given dir hierarchy.
 
     @param dir_root the path to the directory to start scanning
     for test files.  All files in this directory and all its children
     directory trees will be searched.
-
-    @param found_func a callable object that will be passed
-    the parent directory (relative to dir_root) and the list of
-    test files from within that directory.
     """
     for root, _, files in os.walk(dir_root, topdown=False):
         def is_test_filename(test_dir, base_filename):
             """Returns True if the given filename matches the test name format.
 
             @param test_dir the directory to check.  Should be absolute or
             relative to current working directory.
 
             @param base_filename the base name of the filename to check for a
             dherence to the python test case filename format.
 
             @return True if name matches the python test case filename format.
             """
             # Not interested in symbolically linked files.
             if os.path.islink(os.path.join(test_dir, base_filename)):
                 return False
             # Only interested in test files with the "Test*.py" naming pattern.
             return (base_filename.startswith("Test") and
                     base_filename.endswith(".py"))
 
-        tests = [
-            (filename, os.path.join(root, filename))
-            for filename in files
-            if is_test_filename(root, filename)]
-        if tests:
-            found_func(root, tests)
+        for f in files:
+            if is_test_filename(root, f):
+                yield os.path.join(root, f)
 
 
 def initialize_global_vars_common(num_threads, test_work_items, session_dir,
                                   runner_context):
     global g_session_dir, g_runner_context, total_tests, test_counter
     global test_name_len
 
-    total_tests = sum([len(item[1]) for item in test_work_items])
+    total_tests = len(test_work_items)
     test_counter = multiprocessing.Value('i', 0)
     test_name_len = multiprocessing.Value('i', 0)
     g_session_dir = session_dir
     g_runner_context = runner_context
     if not (RESULTS_FORMATTER and RESULTS_FORMATTER.is_using_terminal()):
         print(
             "Testing: %d test suites, %d thread%s" %
             (total_tests,
def multiprocessing_test_runner(num_threads, test_work_items, session_dir,
     manager = multiprocessing.Manager()
     worker_index_map = manager.dict()
 
     # Create workers.  We don't use multiprocessing.Pool due to
     # challenges with handling ^C keyboard interrupts.
     workers = []
     for _ in range(num_threads):
         worker = multiprocessing.Process(
-            target=process_dir_worker_multiprocessing,
+            target=process_file_worker_multiprocessing,
             args=(output_lock,
                   test_counter,
                   total_tests,
                   test_name_len,
                   dotest_options,
                   job_queue,
                   result_queue,
                   inferior_pid_events,
def multiprocessing_test_runner_pool(num_threads, test_work_items, session_dir,
     pool = multiprocessing.Pool(
         num_threads,
         initializer=setup_global_variables,
         initargs=(output_lock, test_counter, total_tests, test_name_len,
                   dotest_options, worker_index_map))
 
     # Start the map operation (async mode).
     map_future = pool.map_async(
-        process_dir_worker_multiprocessing_pool, test_work_items)
+        process_file_worker_multiprocessing_pool, test_work_items)
     return map_async_run_loop(
         map_future, RUNNER_PROCESS_ASYNC_MAP, RESULTS_LISTENER_CHANNEL)
 
 
 def threading_test_runner(num_threads, test_work_items, session_dir,
                           runner_context):
     """Provides hand-wrapped pooling threading-based test runner adapter
     with Ctrl-C support.
"""
     # child processes they spawn.
     inferior_pid_events = queue.Queue()
 
     # Create workers. We don't use multiprocessing.pool.ThreadedPool
     # due to challenges with handling ^C keyboard interrupts.
     workers = []
     for _ in range(num_threads):
         worker = threading.Thread(
-            target=process_dir_worker_threading,
+            target=process_file_worker_threading,
             args=(job_queue,
                   result_queue,
                   inferior_pid_events))
         worker.start()
         workers.append(worker)
 
     # Main loop: wait for all workers to finish and wait for
     # the socket handlers to wrap up.

 def threading_test_runner_pool(num_threads, test_work_items, session_dir,
                                runner_context):
     # Initialize our global state.
     initialize_global_vars_threading(num_threads, test_work_items, session_dir,
                                      runner_context)
 
     pool = multiprocessing.pool.ThreadPool(num_threads)
     map_future = pool.map_async(
-        process_dir_worker_threading_pool, test_work_items)
+        process_file_worker_threading_pool, test_work_items)
 
     return map_async_run_loop(
         map_future, RUNNER_PROCESS_ASYNC_MAP, RESULTS_LISTENER_CHANNEL)
 
 
 def asyncore_run_loop(channel_map):
     try:
         asyncore.loop(None, False, channel_map)
def inprocess_exec_test_runner(test_work_items, session_dir, runner_context):
     # global RUNNER_PROCESS_ASYNC_MAP
     global RESULTS_LISTENER_CHANNEL
     if RESULTS_LISTENER_CHANNEL is not None:
         socket_thread = threading.Thread(
             target=lambda: asyncore_run_loop(RUNNER_PROCESS_ASYNC_MAP))
         socket_thread.start()
 
     # Do the work.
-    test_results = list(map(process_dir_mapper_inprocess, test_work_items))
+    test_results = list(map(process_file_mapper_inprocess, test_work_items))
 
     # If we have a listener channel, shut it down here.
     if RESULTS_LISTENER_CHANNEL is not None:
         # Close down the channel.
         RESULTS_LISTENER_CHANNEL.close()
         RESULTS_LISTENER_CHANNEL = None
 
         # Wait for the listener and handlers to complete.
         socket_thread.join()
 
     return test_results
 
 
 def walk_and_invoke(test_files, dotest_argv, num_workers, test_runner_func):
     """Invokes the test runner on each test file specified by test_files.
 
-    @param test_files a list of (test_subdir, list_of_test_files_in_dir)
+    @param test_files a list of (test_file, full_path_to_test_file)
     @param num_workers the number of worker queues working on these test files
     @param test_runner_func the test runner configured to run the tests
 
     @return a tuple of results from the running of the specified tests,
     of the form (timed_out, passed, failed, unexpected_successes, pass_count,
     fail_count)
     """
     # The async_map is important to keep all thread-related asyncore
if RESULTS_FORMATTER is not None:
             dotest_argv[port_value_index] = listener_port
         except ValueError:
             # --results-port doesn't exist (yet), add it
             dotest_argv.append("--results-port")
             dotest_argv.append(listener_port)
 
     # Build the test work items out of the (dir, file_list) entries passed in.
     test_work_items = []
-    for entry in test_files:
-        test_work_items.append((entry[0], entry[1], dotest_argv, None))
+    for test_file in test_files:
+        test_work_items.append((test_file, dotest_argv, None))
 
     # Convert test work items into test results using whatever
     # was provided as the test run function.
     test_results = test_runner_func(test_work_items)
 
     # Summarize the results and return to caller.
     timed_out = sum([result[0] for result in test_results], [])
     passed = sum([result[1] for result in test_results], [])
def main(num_threads, test_subdir, test_runner_name, results_formatter):
 
     if test_runner_name not in runner_strategies_by_name:
         raise Exception(
             "specified testrunner name '{}' unknown. Valid choices: {}".format(
                 test_runner_name,
                 list(runner_strategies_by_name.keys())))
     test_runner_func = runner_strategies_by_name[test_runner_name]
 
-    # Collect the files on which we'll run the first test run phase.
-    test_files = []
-    find_test_files_in_dir_tree(
-        test_subdir, lambda tdir, tfiles: test_files.append(
-            (test_subdir, tfiles)))
-
     # Do the first test run phase.
     summary_results = walk_and_invoke(
-        test_files,
+        find_test_files_in_dir_tree(test_subdir),
         dotest_argv,
         num_threads,
         test_runner_func)
 
     (timed_out, passed, failed, unexpected_successes, pass_count,
      fail_count) = summary_results
 
     # Check if we have any tests to rerun as phase 2.

lldb/trunk/packages/Python/lldbsuite/test/lldbtest.py

def setPlatformWorkingDir(self):
                 "rm %s/*" % remote_test_dir)
             lldb.remote_platform.Run(shell_cmd)
         self.addTearDownHook(clean_working_directory)
 
     def getSourceDir(self):
         """Return the full path to the current test."""
         return os.path.join(os.environ["LLDB_TEST"], self.mydir)
 
+    def getBuildDirBasename(self):
+        return self.__class__.__module__ + "." + self.testMethodName
+
     def getBuildDir(self):
         """Return the full path to the current test."""
         variant = self.getDebugInfo()
         if variant is None:
             variant = 'default'
         return os.path.join(os.environ["LLDB_BUILD"], self.mydir,
-                            self.testMethodName)
+                            self.getBuildDirBasename())
     
      
     def makeBuildDir(self):
         """Create the test-specific working directory."""
         # See also dotest.py which sets up ${LLDB_BUILD}.
         lldbutil.mkdir_p(self.getBuildDir())
  
     def getBuildArtifact(self, name="a.out"):
class Base(unittest2.TestCase):
     def buildDefault(
             self,
             architecture=None,
             compiler=None,
             dictionary=None,
             clean=True):
         """Platform specific way to build the default binaries."""
         testdir = self.mydir
-        testname = self.testMethodName
+        testname = self.getBuildDirBasename()
         if self.getDebugInfo():
             raise Exception("buildDefault tests must set NO_DEBUG_INFO_TESTCASE")
         module = builder_module()
         self.makeBuildDir()
         dictionary = lldbplatformutil.finalize_build_dictionary(dictionary)
         if not module.buildDefault(self, architecture, compiler,
                                    dictionary, clean, testdir, testname):
             raise Exception("Don't know how to build default binary")
 
     def buildDsym(
             self,
             architecture=None,
             compiler=None,
             dictionary=None,
             clean=True):
         """Platform specific way to build binaries with dsym info."""
         testdir = self.mydir
-        testname = self.testMethodName
+        testname = self.getBuildDirBasename()
         if self.getDebugInfo() != "dsym":
             raise Exception("NO_DEBUG_INFO_TESTCASE must build with buildDefault")
 
         module = builder_module()
         dictionary = lldbplatformutil.finalize_build_dictionary(dictionary)
         if not module.buildDsym(self, architecture, compiler,
                                 dictionary, clean, testdir, testname):
             raise Exception("Don't know how to build binary with dsym")
 
     def buildDwarf(
             self,
             architecture=None,
             compiler=None,
             dictionary=None,
             clean=True):
         """Platform specific way to build binaries with dwarf maps."""
         testdir = self.mydir
-        testname = self.testMethodName
+        testname = self.getBuildDirBasename()
         if self.getDebugInfo() != "dwarf":
             raise Exception("NO_DEBUG_INFO_TESTCASE must build with buildDefault")
 
         module = builder_module()
         dictionary = lldbplatformutil.finalize_build_dictionary(dictionary)
         if not module.buildDwarf(self, architecture, compiler,
                                    dictionary, clean, testdir, testname):
             raise Exception("Don't know how to build binary with dwarf")
 
     def buildDwo(
             self,
             architecture=None,
             compiler=None,
             dictionary=None,
             clean=True):
         """Platform specific way to build binaries with dwarf maps."""
         testdir = self.mydir
-        testname = self.testMethodName
+        testname = self.getBuildDirBasename()
         if self.getDebugInfo() != "dwo":
             raise Exception("NO_DEBUG_INFO_TESTCASE must build with buildDefault")
 
         module = builder_module()
         dictionary = lldbplatformutil.finalize_build_dictionary(dictionary)
         if not module.buildDwo(self, architecture, compiler,
                                    dictionary, clean, testdir, testname):
             raise Exception("Don't know how to build binary with dwo")
 
     def buildGModules(
             self,
             architecture=None,
             compiler=None,
             dictionary=None,
             clean=True):
         """Platform specific way to build binaries with gmodules info."""
         testdir = self.mydir
-        testname = self.testMethodName
+        testname = self.getBuildDirBasename()
         if self.getDebugInfo() != "gmodules":
             raise Exception("NO_DEBUG_INFO_TESTCASE must build with buildDefault")
 
         module = builder_module()
         dictionary = lldbplatformutil.finalize_build_dictionary(dictionary)
         if not module.buildGModules(self, architecture, compiler,
                                     dictionary, clean, testdir, testname):
             raise Exception("Don't know how to build binary with gmodules")