BUG: make a failing Monte Carlo worker say so

rocketpy/simulation/monte_carlo.py

@@ -18,6 +18,7 @@
 import os
 import traceback
 import warnings
+from contextlib import suppress
 from numbers import Real
 from pathlib import Path
 from time import time
@@ -485,8 +486,12 @@ 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
+                # run would report the simulations it never wrote as done.
+                _refuse_a_worker_that_did_not_finish(processes)
 
                 # Handle error from the child processes
                 if simulation_error_event.is_set():
@@ -531,6 +536,10 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
         error_event : multiprocess.Event
             Event signaling an error occurred during the simulation.
         """
+        # Bound before the try: the handler below reads both, and a failure in
+        # the seeding, or in the claim that opens the loop, reaches it with
+        # neither of them assigned.
+        sim_idx, inputs_json = None, ""
         try:
             # Ensure Processes generate different random numbers
             self.environment._set_stochastic(seed)
@@ -568,16 +577,32 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
                     mutex.release()
 
         except Exception:  # pylint: disable=broad-except
-            mutex.acquire()
-            with open(self.error_file, "a", encoding="utf-8") as f:
-                f.write(inputs_json)
+            self.__report_a_failed_simulation(sim_idx, inputs_json, mutex, error_event)
 
-            # See note above: must use print() to remain visible from a
-            # multiprocessing worker process.
-            _SimMonitor.reprint(
-                f"Error on iteration {sim_idx}:\n{traceback.format_exc()}"
-            )
+    def __report_a_failed_simulation(self, sim_idx, inputs_json, mutex, error_event):
+        """Write down and announce a simulation this worker could not finish.
+
+        The event goes first and from outside the lock, since a worker that
+        cannot write its diagnostics still has to be able to stop the others.
+        Each step under the lock is suppressed on its own: a full disk would
+        otherwise replace the failure being reported, and the lock is a
+        manager's, so ending while holding it leaves the next worker waiting
+        on a process that no longer exists.
+        """
+        details = traceback.format_exc()
+        where = "worker startup" if sim_idx is None else f"iteration {sim_idx}"
+        with suppress(Exception):
             error_event.set()
+
+        mutex.acquire()
+        try:
+            with suppress(Exception):
+                with open(self.error_file, "a", encoding="utf-8") as f:
+                    f.write(inputs_json or _worker_failure_record(where, details))
+            with suppress(Exception):
+                # Must use print() to remain visible from a worker process.
+                _SimMonitor.reprint(f"Error on {where}:\n{details}")
+        finally:
             mutex.release()
 
     def __run_single_simulation(self):
@@ -1755,6 +1780,81 @@ 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.
+
+    Written because the caller is told to read the error file, and a traceback
+    a worker printed is not there to be read once its output is redirected.
+    """
+    return json.dumps({"index": None, "stage": where, "error": details}) + "\n"
+
+
+def _refuse_a_worker_that_did_not_finish(processes):
+    """Raise if any worker left without exiting cleanly.
+
+    The workers report their own failures through an event, which one that was
+    killed never reaches, so what is left of it is its exit code. A negative
+    one is the signal that ended it, and ``None`` is one still running.
+    """
+    unfinished = [
+        f"worker {position} with exit code {process.exitcode}"
+        for position, process in enumerate(processes)
+        if process.exitcode != 0
+    ]
+    if not unfinished:
+        return
+    raise RuntimeError(
+        f"The run is incomplete: {', '.join(unfinished)}. A worker that ends "
+        "this way records nothing and cannot say why, so the simulations it "
+        "held are missing from the results."
+    )
+
+
 def _import_multiprocess():
     """Import the necessary modules and submodules for the
     multiprocess library.

rocketpy/stochastic/stochastic_model.py

@@ -8,7 +8,7 @@
 from rocketpy.mathutils.function import Function
 from rocketpy.stochastic.custom_sampler import CustomSampler
 
-from ..tools import get_distribution
+from ..tools import _seed_sequence_to_int, get_distribution
 
 
 def _names_as_spawn_key(input_names):
@@ -41,6 +41,18 @@ def _format_number(value):
     return f"array of shape {np.shape(value)}"
 
 
+def _seed_as_entropy(seed):
+    """A seed as something ``SeedSequence`` will take as entropy.
+
+    A parallel run is handed a ``SeedSequence``, which it will not take. Any
+    other seed goes through untouched, so the stream an int reaches stays where
+    it was.
+    """
+    if not isinstance(seed, np.random.SeedSequence):
+        return seed
+    return _seed_sequence_to_int(seed)
+
+
 def _sampler_seed(seed, input_names):
     """Derive a seed for one sampler, or for one group that shares a generator.
 
@@ -54,10 +66,10 @@ def _sampler_seed(seed, input_names):
     # Sorted here rather than trusting the caller, so a future call site cannot
     # give one group two different seeds by listing its members another way.
     root = np.random.SeedSequence(
-        entropy=seed, spawn_key=_names_as_spawn_key(tuple(sorted(input_names)))
+        entropy=_seed_as_entropy(seed),
+        spawn_key=_names_as_spawn_key(tuple(sorted(input_names))),
     )
-    words = root.generate_state(4, dtype=np.uint32)
-    return sum(int(word) << (32 * position) for position, word in enumerate(words))
+    return _seed_sequence_to_int(root)
 
 
 # TODO: Stop using assert in production code. Use exceptions instead.

rocketpy/tools.py

@@ -1377,6 +1377,17 @@ def euler313_to_quaternions(phi, theta, psi):
     return e0, e1, e2, e3
 
 
+def _seed_sequence_to_int(seed_sequence):
+    """Returns a ``SeedSequence`` as the 128-bit ``int`` it can be rebuilt from.
+
+    Folded through ``generate_state`` rather than read off ``entropy``, since
+    the children of one root differ only by ``spawn_key``, and combined by
+    value so it does not depend on byte order.
+    """
+    words = seed_sequence.generate_state(4, dtype=np.uint32)
+    return sum(int(word) << (32 * position) for position, word in enumerate(words))
+
+
 def get_matplotlib_supported_file_endings():
     """Gets the file endings supported by matplotlib.
 

tests/unit/simulation/test_monte_carlo_parallel_runs.py

@@ -0,0 +1,32 @@
+import pytest
+
+from rocketpy.simulation.monte_carlo import MonteCarlo
+
+
+@pytest.mark.parametrize("parallel", [False, True])
+def test_a_monte_carlo_run_finishes(
+    stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path, parallel
+):
+    # The parallel path hands each worker a SeedSequence rather than an int, and
+    # nothing else in the suite exercises that. A worker that dies on it is not
+    # reported, so this reads as a hang rather than as a failure.
+    #
+    # Built here rather than taken from the monte_carlo_calisto fixture, whose
+    # own filename is fixed, since `filename` is a plain attribute and the three
+    # working paths are settled when the object is constructed.
+    analysis = MonteCarlo(
+        filename=str(tmp_path / "study"),
+        environment=stochastic_environment,
+        rocket=stochastic_calisto,
+        flight=stochastic_flight,
+    )
+
+    analysis.simulate(
+        number_of_simulations=2,
+        append=False,
+        parallel=parallel,
+        n_workers=2 if parallel else None,
+    )
+
+    assert analysis.num_of_loaded_sims == 2
+    assert str(tmp_path) in str(analysis.output_file)

tests/unit/simulation/test_monte_carlo_worker_exit.py

@@ -0,0 +1,72 @@
+import os
+from types import SimpleNamespace
+
+import pytest
+
+from rocketpy.simulation.monte_carlo import (
+    MonteCarlo,
+    _refuse_a_worker_that_did_not_finish,
+)
+
+
+def _worker(exitcode):
+    return SimpleNamespace(exitcode=exitcode)
+
+
+def test_workers_that_all_exited_cleanly_are_accepted():
+    _refuse_a_worker_that_did_not_finish([_worker(0), _worker(0)])
+
+
+def test_a_worker_killed_by_a_signal_is_refused():
+    with pytest.raises(RuntimeError, match=r"worker 1 with exit code -9"):
+        _refuse_a_worker_that_did_not_finish([_worker(0), _worker(-9)])
+
+
+def test_a_worker_that_exited_nonzero_is_refused():
+    with pytest.raises(RuntimeError, match=r"worker 0 with exit code 1"):
+        _refuse_a_worker_that_did_not_finish([_worker(1), _worker(0)])
+
+
+def test_every_unfinished_worker_is_named():
+    with pytest.raises(RuntimeError) as raised:
+        _refuse_a_worker_that_did_not_finish([_worker(-9), _worker(0), _worker(3)])
+
+    assert "worker 0" in str(raised.value)
+    assert "worker 2" in str(raised.value)
+    assert "worker 1" not in str(raised.value)
+
+
+@pytest.mark.parametrize("exitcode", [None, -15, 2])
+def test_anything_but_a_clean_exit_is_refused(exitcode):
+    with pytest.raises(RuntimeError):
+        _refuse_a_worker_that_did_not_finish([_worker(exitcode), _worker(0)])
+
+
+def _leave_without_recording(flight):  # pylint: disable=unused-argument
+    """Ends the worker the way a kill or an out-of-memory exit does.
+
+    ``os._exit`` rather than a signal, since ``SIGKILL`` is POSIX-only, and
+    reached through the data collector rather than a patched method, since a
+    ``spawn`` platform re-imports the module and would not see the patch.
+    """
+    os._exit(1)
+
+
+def test_a_worker_that_leaves_early_does_not_pass_as_a_finished_run(
+    stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path
+):
+    # The event the workers report through is set by their own handler, and
+    # this one leaves without running it, so the run used to return as though
+    # it had done every simulation it was asked for.
+    analysis = MonteCarlo(
+        filename=str(tmp_path / "study"),
+        environment=stochastic_environment,
+        rocket=stochastic_calisto,
+        flight=stochastic_flight,
+        data_collector={"leave": _leave_without_recording},
+    )
+
+    with pytest.raises(RuntimeError, match="incomplete"):
+        analysis.simulate(
+            number_of_simulations=6, append=False, parallel=True, n_workers=2
+        )