[lit] Enqueue tests on a separate thread to not hit limits on parallel queues

Summary:
The multiprocessing.Queue.put() call can hang if we try queueing all the
tests before starting to take them out of the queue.
The current implementation hangs if tests exceed 2^^15, on Mac OS X.
This might happen with a ninja check-all if one has a bunch of llvm
projects.

Reviewers: delcypher, bkramer

Subscribers: llvm-commits

Differential Revision: http://reviews.llvm.org/D17609

llvm-svn: 263731

diff --git a/llvm/utils/lit/lit/run.py b/llvm/utils/lit/lit/run.py
index 27c414d..bbf644e 100644 (file)
--- a/llvm/utils/lit/lit/run.py
+++ b/llvm/utils/lit/lit/run.py
@@ -44,11 +44,13 @@ class LockedValue(object):
     value = property(_get_value, _set_value)
 
 class TestProvider(object):
-    def __init__(self, tests, num_jobs, queue_impl, canceled_flag):
+    def __init__(self, queue_impl, canceled_flag):
         self.canceled_flag = canceled_flag
 
         # Create a shared queue to provide the test indices.
         self.queue = queue_impl()
+
+    def queue_tests(self, tests, num_jobs):
         for i in range(len(tests)):
             self.queue.put(i)
         for i in range(num_jobs):
@@ -229,7 +231,15 @@ class Run(object):
             consumer = ThreadResultsConsumer(display)
 
         # Create the test provider.
-        provider = TestProvider(self.tests, jobs, queue_impl, canceled_flag)
+        provider = TestProvider(queue_impl, canceled_flag)
+
+        # Queue the tests outside the main thread because we can't guarantee
+        # that we can put() all the tests without blocking:
+        # https://docs.python.org/2/library/multiprocessing.html
+        # e.g: On Mac OS X, we will hang if we put 2^15 elements in the queue
+        # without taking any out.
+        queuer = task_impl(target=provider.queue_tests, args=(self.tests, jobs))
+        queuer.start()
 
         # Install a console-control signal handler on Windows.
         if win32api is not None:
@@ -252,6 +262,8 @@ class Run(object):
             # Otherwise, execute the tests in parallel
             self._execute_tests_in_parallel(task_impl, provider, consumer, jobs)
 
+        queuer.join()
+
         # Cancel the timeout handler.
         if max_time is not None:
             timeout_timer.cancel()