BUG: stop waiting on workers once one of them has died

The lock the workers share belongs to the manager and is not released
when the process holding it is killed. A sibling then blocks on a lock
nobody owns, and the parent, joining without a timeout, waits with it.
The exit code check the previous commit added is never reached, so the
one case it exists for is the one it cannot see.

The join polls now, and acts only when a worker has actually ended
badly. A run that is merely slow is never bounded: an exit code, not a
duration, is what says a worker is gone. The survivors are asked
through the event first, since one between simulations leaves with its
logs intact, and only the ones still running after that are ended.

Undoing this leaves every test in the new file red. Deciding on how
long a worker has taken instead of on how it ended leaves exactly one
red, which is the test that says a slow run must be left alone.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>

Author: thc1006

rocketpy/simulation/monte_carlo.py

@@ -486,8 +486,7 @@ def __run_in_parallel(self, n_workers=None):
                 sim_producer.start()
 
             try:
-                for sim_producer in processes:
-                    sim_producer.join()
+                _join_the_workers(processes, simulation_error_event)
 
                 # Before the event: a worker that was killed, or that died
                 # before its own handler could set it, leaves it clear, and the
@@ -1777,6 +1776,51 @@ def export_errors_to_json(self, filename):
         self._write_log_to_json(self.errors_log, filename)
 
 
+# Short enough that a dead worker is noticed promptly, long enough that the
+# polling costs nothing over a run that takes hours.
+_JOIN_POLL_SECONDS = 0.2
+_SHUTDOWN_GRACE_SECONDS = 5.0
+
+
+def _ended_badly(worker):
+    """Whether a worker has stopped, and stopped for the wrong reason."""
+    return worker.exitcode not in (None, 0)
+
+
+def _stop_the_workers_still_running(processes, error_event, grace_period):
+    """Ask the rest to stop, then end the ones that cannot.
+
+    Asked first because a worker between simulations reads the event and leaves
+    with its logs intact. One blocked on a lock its dead sibling was holding
+    never reaches that check, and only ending it frees the run.
+    """
+    with suppress(Exception):
+        error_event.set()
+    deadline = time() + grace_period
+    for worker in processes:
+        worker.join(timeout=max(0.0, deadline - time()))
+    for worker in processes:
+        if worker.is_alive():
+            worker.terminate()
+            worker.join(timeout=grace_period)
+
+
+def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECONDS):
+    """Wait for the workers, and stop waiting once one of them has died badly.
+
+    The lock the workers share belongs to the manager and is not released when
+    its holder is killed, so a sibling can block on a lock nobody owns while an
+    unbounded join waits with it. Nothing here bounds a run that is merely
+    slow: only an exit code says a worker has died.
+    """
+    while any(worker.is_alive() for worker in processes):
+        for worker in processes:
+            worker.join(timeout=_JOIN_POLL_SECONDS)
+        if any(_ended_badly(worker) for worker in processes):
+            _stop_the_workers_still_running(processes, error_event, grace_period)
+            return
+
+
 def _worker_failure_record(where, details):
     """A row for a worker that failed before it drew anything.
 

tests/unit/simulation/test_monte_carlo_worker_join.py

@@ -0,0 +1,101 @@
+import pytest
+
+from rocketpy.simulation.monte_carlo import _join_the_workers
+
+
+class _Worker:
+    """A process that stops after a set number of polls, or never.
+
+    ``never`` stands in for one blocked on a lock its dead sibling was holding,
+    which is the case an unbounded join waits out forever.
+    """
+
+    def __init__(self, exitcode=0, alive_for=0, never=False):
+        self.exitcode = None
+        self._final_exitcode = exitcode
+        self._alive_for = alive_for
+        self._never = never
+        self.joins = 0
+        self.terminated = False
+
+    def is_alive(self):
+        return self.exitcode is None
+
+    def join(self, timeout=None):  # pylint: disable=unused-argument
+        self.joins += 1
+        if self._never or self.joins <= self._alive_for:
+            return
+        self.exitcode = self._final_exitcode
+
+    def terminate(self):
+        self.terminated = True
+        self.exitcode = -15
+
+
+class _Event:
+    def __init__(self):
+        self.was_set = False
+
+    def set(self):
+        self.was_set = True
+
+
+def test_a_run_where_every_worker_finishes_is_left_alone():
+    workers = [_Worker(alive_for=3), _Worker(alive_for=5)]
+
+    _join_the_workers(workers, _Event(), grace_period=0)
+
+    assert [worker.exitcode for worker in workers] == [0, 0]
+    assert not any(worker.terminated for worker in workers)
+
+
+def test_a_worker_blocked_behind_a_dead_one_does_not_wait_forever():
+    # The one that mattered. Without a bound this call never returns, so the
+    # parent never reaches the check that would have reported the failure.
+    died = _Worker(exitcode=-9, alive_for=1)
+    blocked = _Worker(never=True)
+
+    _join_the_workers([died, blocked], _Event(), grace_period=0)
+
+    assert blocked.terminated
+
+
+def test_the_survivors_are_asked_before_they_are_ended():
+    died = _Worker(exitcode=1, alive_for=1)
+    blocked = _Worker(never=True)
+    event = _Event()
+
+    _join_the_workers([died, blocked], event, grace_period=0)
+
+    assert event.was_set
+
+
+def test_a_survivor_that_stops_on_its_own_is_not_terminated():
+    died = _Worker(exitcode=1, alive_for=1)
+    cooperative = _Worker(alive_for=2)
+
+    _join_the_workers([died, cooperative], _Event(), grace_period=0)
+
+    assert not cooperative.terminated
+    assert cooperative.exitcode == 0
+
+
+@pytest.mark.parametrize("exitcode", [-9, 1, 2])
+def test_any_bad_exit_starts_the_shutdown(exitcode):
+    died = _Worker(exitcode=exitcode, alive_for=1)
+    blocked = _Worker(never=True)
+
+    _join_the_workers([died, blocked], _Event(), grace_period=0)
+
+    assert blocked.terminated
+
+
+def test_a_slow_run_is_never_bounded():
+    # Nothing here may act on how long a worker takes, only on it having died.
+    slow = _Worker(alive_for=50)
+    slower = _Worker(alive_for=80)
+
+    _join_the_workers([slow, slower], _Event(), grace_period=0)
+
+    assert not any(worker.terminated for worker in (slow, slower))
+    assert slower.joins > 50