Don't let a slow queue feeder make the runner skip tests

RLTest/__main__.py

@@ -994,20 +994,14 @@ def execute(self):
             print(Colors.Bred('only one test can be run on interactive-debugger use -t'))
             sys.exit(1)
 
-        jobs = Queue()
-        n_jobs = 0
-        for test in self.loader:
-            jobs.put(test, block=False)
-            n_jobs += 1
+        tests = list(self.loader)
+        n_jobs = len(tests)
 
-        def run_jobs_main_thread(jobs):
+        def run_jobs_main_thread(tests):
             nonlocal done
             bar = self.progressbar(n_jobs)
-            for _ in bar:
-                try:
-                    test = jobs.get(timeout=0.1)
-                except Exception as e:
-                    break
+            for i in bar:
+                test = tests[i]
 
                 def on_timeout():
                     nonlocal done
@@ -1027,9 +1021,8 @@ def on_timeout():
         def run_jobs(jobs, results, port):
             Defaults.port = port
             while True:
-                try:
-                    test = jobs.get(timeout=0.1)
-                except Exception as e:
+                test = jobs.get()
+                if test is None:
                     break
 
                 # Reset per-test: addFailure() in this worker writes into this
@@ -1081,8 +1074,13 @@ def on_timeout():
         # Open group for all tests at the start (parallel execution)
         self._openGitHubActionsTestsGroup()
         if self.parallelism == 1:
-            run_jobs_main_thread(jobs)
+            run_jobs_main_thread(tests)
         else :
+            # A timed get() can see an empty queue before the feeder thread
+            # flushes, so workers block on get() until their None sentinel.
+            jobs = Queue()
+            for test in tests + [None] * self.parallelism:
+                jobs.put(test, block=False)
             processes = []
             currPort = Defaults.port
             for i in range(self.parallelism):

tests/unit/test_serial_runner_queue.py

@@ -0,0 +1,70 @@
+"""Regression test: the serial runner must run every test even when a
+multiprocessing.Queue's feeder thread has not flushed yet.
+
+It used to read tests back from a multiprocessing.Queue with
+get(timeout=0.1) and stop on Empty, so a starved feeder thread made it run
+zero tests and exit 0.
+"""
+
+import argparse
+import multiprocessing as mp
+import queue
+import sys
+from types import SimpleNamespace
+from unittest import TestCase
+from unittest.mock import patch
+
+import RLTest.__main__ as rlmain
+from RLTest.__main__ import RLTest
+
+
+class _StarvedQueue:
+    """A queue whose feeder thread never gets to flush."""
+
+    def put(self, obj, block=True, timeout=None):
+        pass
+
+    def get(self, block=True, timeout=None):
+        raise queue.Empty
+
+
+def _make_rltest(tests, parallelism, run_single_test):
+    rl = RLTest.__new__(RLTest)
+    rl.args = argparse.Namespace(env_only=False, interactive_debugger=False,
+                                 no_output_catch=True, no_progress=True,
+                                 failed_tests_file=None)
+    rl.loader = tests
+    rl.parallelism = parallelism
+    rl.currEnv = None
+    rl.testsFailed = {}
+    rl.github_actions_group_open = False
+    rl.run_single_test = run_single_test
+    return rl
+
+
+class TestRunnerQueue(TestCase):
+
+    def test_serial_runs_every_test_with_a_starved_queue(self):
+        ran = []
+
+        def run_single_test(test, on_timeout):
+            ran.append(test)
+            return 1
+
+        rl = _make_rltest(['t1', 't2', 't3'], 1, run_single_test)
+        with patch.object(rlmain, 'Queue', _StarvedQueue):
+            rl.execute()
+        self.assertEqual(ran, ['t1', 't2', 't3'])
+
+    def test_parallel_runs_every_test(self):
+        if sys.platform == 'win32':
+            self.skipTest('fork start method is unavailable on Windows')
+        ctx = mp.get_context('fork')
+        tests = [SimpleNamespace(name='t%d' % i) for i in range(20)]
+        rl = _make_rltest(tests, 3, lambda test, on_timeout: 1)
+        with patch.object(rlmain, 'Queue', ctx.Queue), \
+             patch.object(rlmain, 'Process', ctx.Process), \
+             patch('sys.stdout') as out:
+            rl.execute()
+        summary = ''.join(c.args[0] for c in out.write.call_args_list)
+        self.assertIn('Total Tests Run: 20,', summary)