Fix parallel-coordinator hang when summary pipe saturates (#248)

* Fix parallel-coordinator hang when summary pipe saturates

The coordinator drained the 'summary' queue only after joining all worker
processes. With enough queued data (or a single large testsFailed dict),
the summary-pipe buffer (~64 KiB on Linux) saturates and worker feeder
threads block in pipe_write, both inside on_timeout's join_thread() and
during Python's end-of-process queue finalization. This in turn hangs the
coordinator's p.join() indefinitely.

Introduce a module-level helper _join_workers_with_summary_drain that
joins workers while continuously draining 'summary' from a background
thread, and use it in execute(). Also correct the stale comment in the
on_timeout closure to describe the actual watcher-thread os._exit(1)
flow.

* Use SimpleQueue for summary in parallel coordinator

SimpleQueue.put is synchronous (no feeder thread, no internal buffer),
so a successful put() implies the bytes are already in the kernel pipe.
That removes the need for the summary.close() + summary.join_thread()
dance in on_timeout before the watcher's os._exit(1), and the comment
that explained it.

The coordinator-side drain thread is updated to a blocking get() driven
by a sentinel on shutdown, eliminating its busy-loop timeout too.

results stays a Queue because the progressbar liveness loop relies on
get(timeout=...), which SimpleQueue does not expose publicly.

* Address review: reorder puts in on_timeout, fix stale comment

In the on_timeout closure, put 'results' before 'summary'. The summary
queue is a SimpleQueue with a synchronous put(), and the coordinator
only starts draining it after every result is in. Putting summary first
risked blocking on a full summary pipe while the coordinator was still
waiting on this worker's result, which would have stalled the whole
results-collection loop. Putting results first guarantees the worker's
output reaches the coordinator unconditionally; the subsequent summary
put may briefly block but always unblocks once the coordinator moves to
the drain phase.

Also drop the stale 'feeder threads' wording near the call site: the
summary queue no longer has a feeder thread.

* Collapse parallel results+summary into a single queue

The parallel coordinator used two queues: 'results' for per-test output
(read by the progressbar loop) and 'summary' for per-worker aggregates
('done' count and the worker's full testsFailed dict, read after the
loop). Workers' summary.put could block on a full pipe because the
coordinator only drained summary in a second phase, after every results
message had been received.

Collapse to a single 'results' queue carrying one self-contained message
per test: { test_name, output, done, failures }. The worker resets
self.testsFailed = {} before each test so addFailure() writes into a
fresh dict that ships verbatim; the worker keeps no cumulative state.
The coordinator owns the canonical testsFailed via update() per message.

This eliminates the deadlock by construction: the only queue is drained
continuously by the coordinator's progressbar loop for the entire
lifetime of the workers, so worker put()s can never block on a full
pipe. Removes the SimpleQueue import, the _SUMMARY_DRAIN_STOP sentinel,
and the _join_workers_with_summary_drain helper. on_timeout shrinks to
a single put + close + join_thread.

The unit test is updated to exercise the new pattern: workers push many
large per-test messages while the main thread drains them live.

* Enhance parallel test execution: add shutdown messages and improve result handling

* Address review: ship shutdown sentinel from on_timeout; misc fixups

- on_timeout now ships both the per-test result (with shutdown=False)
  and the shutdown sentinel before the watcher thread calls os._exit(1),
  keeping the coordinator's bounded count of n_jobs + parallelism
  accurate when a worker dies on timeout. Also fixes a KeyError on the
  missing 'shutdown' key in the timeout payload.
- Grammar: 'no more processors is alive' -> 'are alive'.
- test_parallel_drain: use time.monotonic() for elapsed measurement.

Author: GuyAv46

RLTest/__main__.py

@@ -941,44 +941,60 @@ def on_timeout():
 
             self.takeEnvDown(fullShutDown=True)
 
-        def run_jobs(jobs, results, summary, port):
+        def run_jobs(jobs, results, port):
             Defaults.port = port
-            done = 0
             while True:
                 try:
                     test = jobs.get(timeout=0.1)
                 except Exception as e:
                     break
 
+                # Reset per-test: addFailure() in this worker writes into this
+                # dict, which is shipped as-is. The coordinator owns the
+                # cumulative testsFailed; the worker keeps no state across tests.
+                self.testsFailed = {}
                 output = io.StringIO()
                 with redirect_stdout(output):
                     def on_timeout():
-                        nonlocal done
                         try:
-                            done += 1
                             self.killEnvWithSegFault()
                             self.handleFailure(testFullName=test.name, testname=test.name, error_msg=Colors.Bred('Test timeout'))
                         except Exception as e:
                             self.handleFailure(testFullName=test.name, testname=test.name, error_msg=Colors.Bred('Exception on timeout function %s' % str(e)))
                         finally:
-                            results.put({'test_name': test.name, "output": output.getvalue()}, block=False)
-                            summary.put({'done': done, 'failures': self.testsFailed}, block=False)
-                            # After we return the processes will be killed, so we must make sure the queues are drained properly.
+                            # The watcher thread calls os._exit(1) right after
+                            # this returns, bypassing Python finalization and
+                            # the normal post-loop shutdown put. Ship both the
+                            # per-test result and the shutdown sentinel here so
+                            # the coordinator's bounded count of
+                            # n_jobs + parallelism remains accurate. close() +
+                            # join_thread() flushes both puts to the pipe first.
+                            results.put({'test_name': test.name, 'output': output.getvalue(),
+                                         'done': 1, 'failures': self.testsFailed,
+                                         'shutdown': False}, block=False)
+                            results.put({'test_name': '<worker shutdown>', 'output': '',
+                                         'done': 0, 'failures': {},
+                                         'shutdown': True}, block=False)
                             results.close()
-                            summary.close()
-                            summary.join_thread()
                             results.join_thread()
-                    done += self.run_single_test(test, on_timeout)
 
-                results.put({'test_name': test.name, "output": output.getvalue()}, block=False)
+                    done_delta = self.run_single_test(test, on_timeout)
 
-            self.takeEnvDown(fullShutDown=True)
+                results.put({'test_name': test.name, 'output': output.getvalue(),
+                             'done': done_delta, 'failures': self.testsFailed,
+                             'shutdown': False}, block=False)
 
-            # serialized the results back
-            summary.put({'done': done, 'failures': self.testsFailed}, block=False)
+            # Always ship one shutdown message per worker so the coordinator
+            # reads a known total of n_jobs + parallelism messages. Captures
+            # failures raised during final shutdown (e.g. "redis did not exit
+            # cleanly" when env_reuse=True).
+            self.testsFailed = {}
+            self.takeEnvDown(fullShutDown=True)
+            results.put({'test_name': '<worker shutdown>', 'output': '',
+                         'done': 0, 'failures': self.testsFailed,
+                         'shutdown': True}, block=False)
 
         results = Queue()
-        summary = Queue()
         # Open group for all tests at the start (parallel execution)
         self._openGitHubActionsTestsGroup()
         if self.parallelism == 1:
@@ -987,39 +1003,40 @@ def on_timeout():
             processes = []
             currPort = Defaults.port
             for i in range(self.parallelism):
-                p = Process(target=run_jobs, args=(jobs,results,summary,currPort))
+                p = Process(target=run_jobs, args=(jobs,results,currPort))
                 currPort += 30 # safe distance for cluster and replicas
                 processes.append(p)
                 p.start()
-            for _ in self.progressbar(n_jobs):
+            # Workers send exactly n_jobs per-test messages plus one shutdown
+            # message each, for a known total. The single shared queue does
+            # not preserve per-worker ordering, so a fast worker's shutdown
+            # may arrive before a slow worker's last test. We read every
+            # message in one bounded loop and tick the progressbar only on
+            # per-test ones. The has_live_processor guard turns a worker
+            # crash before it ships its shutdown message into a clean error
+            # instead of an indefinite hang.
+            def _get_result():
                 while True:
-                    # check if we have some lives executors
-                    has_live_processor = False
-                    for p in processes:
-                        if p.is_alive():
-                            has_live_processor = True
-                            break
                     try:
-                        res = results.get(timeout=1)
-                        break
-                    except Exception as e:
-                        if not has_live_processor:
-                            raise Exception('Failed to get job result and no more processors is alive')
-                output = res['output']
-                print('%s' % output, end="")
+                        return results.get(timeout=1)
+                    except Exception:
+                        if not any(p.is_alive() for p in processes):
+                            raise Exception('Failed to get job result and no more processors are alive')
+
+            bar_iter = iter(self.progressbar(n_jobs))
+            for _ in range(n_jobs + self.parallelism):
+                res = _get_result()
+                if res['output']:
+                    print('%s' % res['output'], end="")
+                done += res['done']
+                self.testsFailed.update(res['failures'])
+                if not res['shutdown']:
+                    next(bar_iter, None)
+            next(bar_iter, None)  # finalize bar.update(n_jobs)
 
             for p in processes:
                 p.join()
 
-            # join results
-            while True:
-                try:
-                    res = summary.get(timeout=1)
-                except Exception as e:
-                    break
-                done += res['done']
-                self.testsFailed.update(res['failures'])
-
         endTime = time.time()
 
         # Close group after all tests complete (parallel execution)

tests/unit/test_parallel_drain.py

@@ -0,0 +1,86 @@
+"""Regression test for a hang in the parallel test coordinator.
+
+The parallel coordinator uses a single ``results`` queue carrying one message
+per test. Workers push these messages while the coordinator drains them in its
+progressbar loop. If the coordinator ever stops draining before workers stop
+pushing, large per-test outputs saturate the pipe (~64 KiB on Linux), worker
+``put()`` calls block in ``pipe_write``, and ``p.join()`` hangs indefinitely.
+
+This test reproduces the saturation scenario by spawning workers that each
+push many large messages, then asserts that a coordinator that drains
+continuously throughout the workers' lifetime finishes promptly with every
+message accounted for and every worker cleanly exited.
+"""
+
+import multiprocessing as mp
+import sys
+import time
+from unittest import TestCase
+
+
+# ~32 KiB per message × 8 workers × 8 messages = 2 MiB total, well over the
+# typical 64 KiB pipe buffer on Linux, so writers will block on ``pipe_write``
+# unless the parent is actively reading throughout.
+_PAYLOAD_BYTES = 32 * 1024
+_NUM_WORKERS = 8
+_MSGS_PER_WORKER = 8
+_JOIN_TIMEOUT_SECS = 30.0
+
+
+def _worker_puts_many_results(results, n_msgs, payload_bytes):
+    # Queue.put is async (via a feeder thread), but at process exit the feeder
+    # must flush the buffered items to the pipe before the worker can exit. If
+    # the parent is not draining, that flush blocks forever.
+    payload = 'x' * payload_bytes
+    for i in range(n_msgs):
+        results.put({'test_name': 't%d' % i, 'output': payload,
+                     'done': 1, 'failures': {}})
+
+
+class TestParallelResultsDrain(TestCase):
+
+    def setUp(self):
+        if sys.platform == 'win32':
+            self.skipTest('fork start method is unavailable on Windows')
+        self._ctx = mp.get_context('fork')
+        self._procs = []
+
+    def tearDown(self):
+        # Safety net: if the test ever hangs despite the fix, make sure the
+        # pytest session can still exit cleanly.
+        for p in self._procs:
+            if p.is_alive():
+                p.kill()
+                p.join(timeout=5)
+
+    def test_continuous_drain_does_not_hang(self):
+        results = self._ctx.Queue()
+        self._procs = [
+            self._ctx.Process(
+                target=_worker_puts_many_results,
+                args=(results, _MSGS_PER_WORKER, _PAYLOAD_BYTES),
+            )
+            for _ in range(_NUM_WORKERS)
+        ]
+        for p in self._procs:
+            p.start()
+
+        # Mimic the coordinator's progressbar loop: drain every per-test
+        # message live, in the same thread, while workers are still running.
+        expected = _NUM_WORKERS * _MSGS_PER_WORKER
+        start = time.monotonic()
+        collected = [results.get(timeout=_JOIN_TIMEOUT_SECS) for _ in range(expected)]
+        for p in self._procs:
+            p.join(timeout=_JOIN_TIMEOUT_SECS)
+        elapsed = time.monotonic() - start
+
+        for p in self._procs:
+            self.assertFalse(
+                p.is_alive(),
+                'worker still alive after join; results pipe likely saturated',
+            )
+            self.assertEqual(p.exitcode, 0)
+        self.assertEqual(len(collected), expected)
+        # The drain should return well under its own timeout; we only assert a
+        # loose upper bound to avoid flakiness on slow machines.
+        self.assertLess(elapsed, _JOIN_TIMEOUT_SECS)