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

[filcab]

Summary

The current implementation just hangs if tests exceed 2^^15, at least
on Mac OS X. This might happen with a ninja check-all if one has a bunch of
llvm projects (especially sanitizers + iOS support)

[chapuni]

Interesting. I didn't know such a limit.

[silvas]

This turns out to be simpler than I was imagining. Nice to finally have this implemented! (I still remember when I initially ran into this when doing crazy things with lit, but thought it would be a bit nastier to fix)

[filcab]

ping

[meadori]

I can't speak for the technical changes in this patch, but I will say that I too was experiencing a deadlock on OS X and this patch fixes the problem.

[filcab]

Anyone else wants to chime in or suggest changes?

[bkramer]

This at least needs a comment explaining the limit. Ideally with a python bug number if it is a python bug.

[filcab]

No bug, and there's no specific limit in the docs (well, you can see
how it's implemented and check the limits of the Queue's innards).

multiprocessing.Queue is allowed to behave like this, though:
https://docs.python.org/2/library/multiprocessing.html#multiprocessing.Queue.put

"put(obj[, block[, timeout]])
Put obj into the queue. If the optional argument block is True (the
default) and timeout is None (the default), block if necessary until a
free slot is available."

Do you want me to add this to the commit message? I don't think it's
the most useful thing to have in a comment there, reminding people to
look at the docs. :-)

Thank you,

Filipe

[filcab]

Update with review feedback.

[silvas]

LGTM

llvm/trunk/utils/lit/lit/run.py

def _set_value(self, value):
         try:
             self._value = value
         finally:
             self.lock.release()
 
     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):
             self.queue.put(None)
 
     def cancel(self):
         self.canceled_flag.value = 1
 
def execute_tests(self, display, jobs, max_time=None,
                 consumer = None
         if not consumer:
             task_impl = threading.Thread
             queue_impl = queue.Queue
             canceled_flag = LockedValue(0)
             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:
             def console_ctrl_handler(type):
                 provider.cancel()
                 return True
             win32api.SetConsoleCtrlHandler(console_ctrl_handler, True)
 
         # Install a timeout handler, if requested.
         if max_time is not None:
             def timeout_handler():
                 provider.cancel()
             timeout_timer = threading.Timer(max_time, timeout_handler)
             timeout_timer.start()
 
         # If not using multiple tasks, just run the tests directly.
         if jobs == 1:
             run_one_tester(self, provider, consumer)
         else:
             # 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()
 
         # Update results for any tests which weren't run.
         for test in self.tests:
             if test.result is None:
                 test.setResult(lit.Test.Result(lit.Test.UNRESOLVED, '', 0.0))