BUG: make a failing Monte Carlo worker say so (#1182)

Six ways a worker that failed got away without saying so. It died inside its
own handler on an unbound name, so the manager lock it was holding was never
given back and the run waited for good. A worker that was killed ran no
handler, set no event, and left only an exit code nobody read, so simulate()
returned normally with rows missing. Reporting a failure could itself block on
a lock a dead sibling still held.

Both names are bound before the try now, the handler reports through a bounded
lock and releases it from a finally, and the parent reads exit codes and the
failure event instead of joining unbounded. A run that is only slow is still
never cut short: how a worker ended decides that, not how long it took.

An exit code cannot show a worker that left between claiming an index and
recording it, so a run is checked against its own logs at the end. Both must
hold exactly the simulations asked for, none twice, none unreadable.

Scope is the parallel producer. __run_in_serial has the same unbound name and
belongs to #1177.

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

Author: thc1006

CHANGELOG.md

@@ -62,6 +62,7 @@ Attention: The newest changes should be on top -->
 ### Fixed
 
 - BUG: Correct the gravity sign an `Accelerometer` applies when `consider_gravity=True`. The gravitational field was added to the inertial acceleration instead of subtracted from it, so the sensor reported the negative of the proper acceleration along the vertical: one at rest read -g rather than +g. Recorded accelerometer data taken with `consider_gravity=True` changes sign in that term. [#1175](https://github.com/RocketPy-Team/RocketPy/pull/1175)
+- BUG: Report a Monte Carlo worker that fails instead of hanging or passing for a finished run [#1182](https://github.com/RocketPy-Team/RocketPy/pull/1182)
 - BUG: Sample `StochasticFlight` inputs once per simulation [#1126](https://github.com/RocketPy-Team/RocketPy/pull/1126) [#1090](https://github.com/RocketPy-Team/RocketPy/issues/1090)
 - BUG: Fix spurious `ValueError` from floating-point roundoff at exact tank depletion [#1166](https://github.com/RocketPy-Team/RocketPy/pull/1166)
 - BUG: Draw each declared eccentricity once per simulation [#1168](https://github.com/RocketPy-Team/RocketPy/pull/1168)

rocketpy/simulation/monte_carlo.py

@@ -18,9 +18,10 @@
 import os
 import traceback
 import warnings
+from contextlib import suppress
 from numbers import Real
 from pathlib import Path
-from time import time
+from time import monotonic, time
 
 import numpy as np
 import simplekml
@@ -43,6 +44,19 @@
 # this is the only format it can both resume from and overwrite safely.
 _SIMULATION_LOG_SUFFIX = ".txt"
 
+# Which simulation a row belongs to. Every check on a finished run reads it.
+_SIMULATION_INDEX_KEY = "index"
+
+# How a manager that has gone away answers a proxy call.
+_MANAGER_IS_GONE = (OSError, EOFError)
+
+# Bounded, so a lock its dead holder never released cannot pin this worker.
+_REPORT_LOCK_SECONDS = 5.0
+
+# Longer than the exit-code path: a worker that only read the event is healthy
+# and leaving at the end of the simulation in hand, not blocked on a dead lock.
+_REPORTED_FAILURE_GRACE_SECONDS = 60.0
+
 
 def _refuse_logs_this_run_cannot_write(
     input_file, output_file, error_file, export_config=None
@@ -300,6 +314,14 @@ def simulate(
         -------
         None
 
+        Raises
+        ------
+        RuntimeError
+            If a parallel run does not finish. A worker that ends badly, one
+            that reports a failure, and logs that do not hold every simulation
+            asked for are each refused, since a run that lost work must not be
+            reported as one that completed.
+
         Notes
         -----
         If you need to stop the simulations after starting them, you can
@@ -471,22 +493,26 @@ def __run_in_parallel(self, n_workers=None):
             processes = []
             seeds = np.random.SeedSequence().spawn(n_workers)
 
-            for seed in seeds:
-                sim_producer = multiprocess.Process(
-                    target=self.__sim_producer,
-                    args=(
-                        seed,
-                        sim_monitor,
-                        mutex,
-                        simulation_error_event,
-                    ),
-                )
-                processes.append(sim_producer)
-                sim_producer.start()
-
             try:
-                for sim_producer in processes:
-                    sim_producer.join()
+                for seed in seeds:
+                    sim_producer = multiprocess.Process(
+                        target=self.__sim_producer,
+                        args=(
+                            seed,
+                            sim_monitor,
+                            mutex,
+                            simulation_error_event,
+                        ),
+                    )
+                    sim_producer.start()
+                    # Started first: one that never did cannot be joined, and
+                    # a later start failing still has to bring these down.
+                    processes.append(sim_producer)
+
+                _join_the_workers(processes, simulation_error_event)
+
+                # Before the event: a killed worker never sets it.
+                _refuse_a_worker_that_did_not_finish(processes)
 
                 # Handle error from the child processes
                 if simulation_error_event.is_set():
@@ -496,15 +522,21 @@ def __run_in_parallel(self, n_workers=None):
                         "for more information."
                     )
 
+                # An exit code cannot show a worker that left between
+                # claiming an index and recording it.
+                _refuse_logs_missing_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()
+                # Bounded here too. An unbounded join undid the bound above.
+                _stop_the_workers_still_running(
+                    processes, simulation_error_event, _SHUTDOWN_GRACE_SECONDS
+                )
 
                 if not isinstance(error, KeyboardInterrupt):
                     raise error
@@ -531,6 +563,8 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
         error_event : multiprocess.Event
             Event signaling an error occurred during the simulation.
         """
+        # The handler reads both, and a failure above the loop precedes them.
+        sim_idx, inputs_json = None, ""
         try:
             # Ensure Processes generate different random numbers
             self.environment._set_stochastic(seed)
@@ -567,18 +601,48 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
                 finally:
                     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)
+                # Nothing is in flight between two simulations, nor are these.
+                sim_idx, inputs_json = None, ""
 
-            # 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()}"
-            )
+        except Exception:  # pylint: disable=broad-except
+            if not self.__report_a_failed_simulation(
+                sim_idx, inputs_json, mutex, error_event
+            ):
+                # The event could not be set; the exit code is what is left.
+                raise
+
+    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}"
+        announced = False
+        with suppress(_MANAGER_IS_GONE):
             error_event.set()
-            mutex.release()
+            announced = True
+
+        held = False
+        with suppress(*_MANAGER_IS_GONE):
+            held = mutex.acquire(timeout=_REPORT_LOCK_SECONDS)
+        try:
+            with suppress(OSError):
+                with open(self.error_file, "a", encoding="utf-8") as f:
+                    f.write(_worker_failure_record(where, details, inputs_json))
+            with suppress(OSError, ValueError):
+                # Must use print() to remain visible from a worker process.
+                _SimMonitor.reprint(f"Error on {where}:\n{details}")
+        finally:
+            if held:
+                with suppress(*_MANAGER_IS_GONE):
+                    mutex.release()
+        return announced
 
     def __run_single_simulation(self):
         """Runs a single simulation and returns the inputs and outputs.
@@ -983,6 +1047,13 @@ def _check_data_collector(self, data_collector):
                         "Invalid 'data_collector' key! "
                         f"Variable names overwrites 'export_list' key '{key}'."
                     )
+                if key == _SIMULATION_INDEX_KEY:
+                    raise ValueError(
+                        f"Invalid 'data_collector' key '{key}'! It is the "
+                        f"number of the simulation the row belongs to, which "
+                        f"is written after the collectors run and cannot be "
+                        f"replaced by one."
+                    )
                 if not callable(callback):
                     raise ValueError(
                         f"Invalid value in 'data_collector' for key '{key}'! "
@@ -1755,6 +1826,192 @@ def export_errors_to_json(self, filename):
         self._write_log_to_json(self.errors_log, filename)
 
 
+# Prompt enough to notice a dead worker, cheap enough over a run of 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 _a_failure_was_reported(error_event):
+    """Whether a worker has said it failed, false if it cannot be asked."""
+    with suppress(*_MANAGER_IS_GONE):
+        return error_event.is_set()
+    return False
+
+
+def _wait_for_the_workers(processes, seconds):
+    """Join every worker against one shared deadline, not one each.
+
+    Monotonic, since a clock correction would move a wall-clock deadline.
+    """
+    deadline = monotonic() + seconds
+    for worker in processes:
+        worker.join(timeout=max(0.0, deadline - monotonic()))
+
+
+def _stop_the_workers_still_running(processes, error_event, grace_period):
+    """Ask the rest to stop, end what cannot, kill what outlives that.
+
+    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. Terminate runs no handlers, so it comes second,
+    and a worker can still ignore it.
+    """
+    with suppress(_MANAGER_IS_GONE):
+        error_event.set()
+    _wait_for_the_workers(processes, grace_period)
+
+    for worker in processes:
+        if worker.is_alive():
+            worker.terminate()
+    _wait_for_the_workers(processes, grace_period)
+
+    for worker in processes:
+        if worker.is_alive():
+            worker.kill()
+    _wait_for_the_workers(processes, grace_period)
+
+
+def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECONDS):
+    """Wait for the workers, and stop once one of them has failed.
+
+    A reported failure ends the wait as well as a bad exit code, since a
+    worker that reports one leaves cleanly and says nothing through its exit
+    status. Its siblings read the event between simulations, but one blocked
+    on a lock nobody owns never reaches that check, and the run is already
+    short a simulation either way, so the wait is bounded here rather than
+    left to them. The reported path gets the longer grace: those siblings are
+    working, not stuck.
+
+    Slowness alone ends nothing. With no failure reported a healthy worker is
+    given as long as it needs.
+    """
+    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
+        if _a_failure_was_reported(error_event):
+            _stop_the_workers_still_running(
+                processes, error_event, _REPORTED_FAILURE_GRACE_SECONDS
+            )
+            return
+
+
+def _worker_failure_record(where, details, inputs_json=""):
+    """A row saying what failed, and what the simulation had drawn so far.
+
+    The inputs alone left the error file with no stage and no traceback, which
+    is what the caller is sent there to read.
+    """
+    record = {"index": None, "stage": where, "error": details}
+    with suppress(ValueError):
+        drawn = json.loads(inputs_json)
+        if isinstance(drawn, dict):
+            record["index"] = drawn.get("index")
+            record["inputs"] = drawn
+    return json.dumps(record) + "\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:
+                index = json.loads(line)["index"]
+            except (ValueError, KeyError, TypeError):
+                found.append(None)
+                continue
+            usable = (
+                isinstance(index, int) and not isinstance(index, bool) and index >= 0
+            )
+            found.append(index if usable else None)
+    return found
+
+
+def _refuse_logs_missing_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.
+
+    Rows numbered past the target are left alone: an append given a smaller
+    target than the checkpoint already holds is an append question, not a lost
+    simulation. What each log holds still has to be the consecutive run it
+    claims to be, so its indices are required to be exactly as many as its
+    rows, which refuses a stray number and a hole without needing to be told
+    how long the checkpoint was. The two logs must also agree row for row,
+    since a record goes into both under one lock. Streamed rather than read
+    through ``_read_log_file``, which would hold every row in memory.
+    """
+    wanted = set(range(target))
+    recorded = {}
+    for label, path in (("input", input_file), ("output", output_file)):
+        found = _indices_a_log_holds(path)
+        recorded[label] = found
+        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."
+            )
+        missing = sorted(wanted - held)
+        if missing:
+            raise RuntimeError(
+                f"The run is incomplete: the {label} log is missing "
+                f"{len(missing)} of {target} simulations, the first being "
+                f"{missing[0]}."
+            )
+        strays = sorted(held - set(range(len(found))))
+        if strays:
+            raise RuntimeError(
+                f"The run is incomplete: the {label} log numbers a simulation "
+                f"{strays[0]}, past the {len(found)} it holds, so what it "
+                f"records is not one run of consecutive simulations."
+            )
+
+    if recorded["input"] != recorded["output"]:
+        raise RuntimeError(
+            "The run is incomplete: the input and output logs do not record "
+            "the same simulations in the same order. A record is written to "
+            "both under one lock, so they hold two different runs."
+        )
+
+
+def _refuse_a_worker_that_did_not_finish(processes):
+    """Raise if any worker left without exiting cleanly.
+
+    A negative code is the signal that ended it, ``None`` 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.