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():
@@ -496,15 +501,25 @@ def __run_in_parallel(self, n_workers=None):
                         "for more information."
                     )
 
+                # Last, and from the logs rather than from the workers: every
+                # check above reads how a process ended, and none of them can
+                # see a worker that left cleanly between claiming an index and
+                # recording it.
+                _refuse_a_run_that_lost_a_simulation(
+                    self.input_file, self.output_file, self.number_of_simulations
+                )
+
                 sim_monitor.print_final_status()
 
             # Handle error from the main process
             # pylint: disable=broad-except
             except (Exception, KeyboardInterrupt) as error:
-                simulation_error_event.set()
-
-                for sim_producer in processes:
-                    sim_producer.join()
+                # The same bounded teardown, which sets the event itself. An
+                # unbounded join here used to undo the bound above on exactly
+                # the stubborn worker it exists for.
+                _stop_the_workers_still_running(
+                    processes, simulation_error_event, _SHUTDOWN_GRACE_SECONDS
+                )
 
                 if not isinstance(error, KeyboardInterrupt):
                     raise error
@@ -531,6 +546,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 +587,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 +1790,148 @@ 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 _the_run_is_already_lost(processes, error_event):
+    """Whether anything says the run cannot finish.
+
+    A worker that fails the ordinary way reports through the event and returns,
+    so it exits cleanly and its exit code says nothing. Waiting only on exit
+    codes leaves the parent sitting behind a sibling that is stuck.
+    """
+    if any(_ended_badly(worker) for worker in processes):
+        return True
+    with suppress(Exception):
+        return bool(error_event.is_set())
+    return False
+
+
+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 or a reported failure says a worker has given up.
+    """
+    while any(worker.is_alive() for worker in processes):
+        for worker in processes:
+            worker.join(timeout=_JOIN_POLL_SECONDS)
+        if _the_run_is_already_lost(processes, error_event):
+            _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 _indices_a_log_holds(path):
+    """Every index a log records, in order, and ``None`` for a row it cannot."""
+    found = []
+    with open(path, "r", encoding="utf-8") as recorded:
+        for line in recorded:
+            if not line.strip():
+                continue
+            try:
+                found.append(json.loads(line)["index"])
+            except (ValueError, KeyError, TypeError):
+                found.append(None)
+    return found
+
+
+def _refuse_a_run_that_lost_a_simulation(input_file, output_file, target):
+    """Raise unless both logs hold every simulation the run was asked for.
+
+    An exit code says how a worker ended, never whether the index it had
+    already claimed reached the logs, and the monitor counts claims rather than
+    rows. A worker that leaves between the two is invisible to everything else
+    here, so the logs themselves are what the run is judged on.
+    """
+    wanted = set(range(target))
+    for label, path in (("input", input_file), ("output", output_file)):
+        found = _indices_a_log_holds(path)
+        held = set(found)
+        if None in held:
+            raise RuntimeError(
+                f"The run is incomplete: the {label} log has rows that cannot "
+                f"be read, so what it holds cannot be established."
+            )
+        if len(found) != len(held):
+            raise RuntimeError(
+                f"The run is incomplete: the {label} log records "
+                f"{len(found) - len(held)} simulation(s) more than once."
+            )
+        if held != wanted:
+            missing = sorted(wanted - held)
+            extra = sorted(held - wanted)
+            trouble = []
+            if missing:
+                trouble.append(
+                    f"{len(missing)} of {target} are missing, the first "
+                    f"being {missing[0]}"
+                )
+            if extra:
+                trouble.append(f"{len(extra)} are numbered past the run")
+            raise RuntimeError(
+                f"The run is incomplete: the {label} log does not hold every "
+                f"simulation that was asked for, {' and '.join(trouble)}."
+            )
+
+
+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)