diff --git a/CHANGELOG.md b/CHANGELOG.md index 31e66e333..933226147 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py index c2dcd4030..8c00c1385 100644 --- a/rocketpy/simulation/monte_carlo.py +++ b/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. diff --git a/tests/unit/simulation/test_monte_carlo_parallel_runs.py b/tests/unit/simulation/test_monte_carlo_parallel_runs.py index 4ab0be440..21ac90589 100644 --- a/tests/unit/simulation/test_monte_carlo_parallel_runs.py +++ b/tests/unit/simulation/test_monte_carlo_parallel_runs.py @@ -7,6 +7,7 @@ def test_a_monte_carlo_run_finishes( stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path, parallel ): + """A real run completes and records every simulation, both modes.""" # 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. diff --git a/tests/unit/simulation/test_monte_carlo_run_completeness.py b/tests/unit/simulation/test_monte_carlo_run_completeness.py new file mode 100644 index 000000000..d98f51390 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_run_completeness.py @@ -0,0 +1,272 @@ +import ast +import inspect +import json +import os + +import pytest + +from rocketpy.simulation import monte_carlo as mc_module +from rocketpy.simulation.monte_carlo import ( + MonteCarlo, + _refuse_logs_missing_a_simulation, +) + + +def _a_log(tmp_path, name, rows): + path = tmp_path / name + path.write_text("".join(rows), encoding="utf-8") + return str(path) + + +def _row(index): + return json.dumps({"index": index, "mass": 1.0}) + "\n" + + +def _complete(tmp_path, count=3, name="ok"): + rows = [_row(index) for index in range(count)] + return ( + _a_log(tmp_path, f"{name}.inputs.txt", rows), + _a_log(tmp_path, f"{name}.outputs.txt", rows), + ) + + +def test_a_run_that_recorded_everything_is_accepted(tmp_path): + """Logs holding every index the run asked for raise nothing.""" + inputs, outputs = _complete(tmp_path) + + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_blank_lines_between_rows_are_not_simulations(tmp_path): + """A blank line is skipped rather than counted as an unreadable row.""" + # An interrupted write leaves them, and reading one as a row would report + # a damaged log for a run that lost nothing. + rows = [_row(0), "\n", _row(1), " \n", _row(2)] + inputs = _a_log(tmp_path, "gappy.inputs.txt", rows) + outputs = _a_log(tmp_path, "gappy.outputs.txt", rows) + + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_a_missing_simulation_is_refused(tmp_path): + """A gap in the output log names the first index that is missing.""" + inputs, outputs = _complete(tmp_path) + _a_log(tmp_path, "ok.outputs.txt", [_row(0), _row(2)]) + + with pytest.raises(RuntimeError, match=r"output log.*missing.*being 1"): + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_a_simulation_recorded_twice_is_refused(tmp_path): + """A duplicated index is refused, since a set alone would hide it.""" + inputs, outputs = _complete(tmp_path) + _a_log(tmp_path, "ok.inputs.txt", [_row(0), _row(1), _row(1), _row(2)]) + + with pytest.raises(RuntimeError, match="more than once"): + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_a_row_that_cannot_be_read_is_refused(tmp_path): + """A torn row means the log's contents cannot be established.""" + inputs, outputs = _complete(tmp_path) + _a_log(tmp_path, "ok.outputs.txt", [_row(0), "{half a row\n", _row(2)]) + + with pytest.raises(RuntimeError, match="cannot be read"): + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_rows_numbered_past_the_run_are_left_alone(tmp_path): + """An append below what a checkpoint already holds loses no simulation.""" + # Refusing these said rows were missing when they were extra, and moved + # what append means inside a change about worker failure. + rows = [_row(index) for index in range(4)] + inputs = _a_log(tmp_path, "big.inputs.txt", rows) + outputs = _a_log(tmp_path, "big.outputs.txt", rows) + + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +def test_logs_that_hold_different_simulations_are_refused(tmp_path): + """The input and output logs have to hold the same indices.""" + inputs = _a_log(tmp_path, "a.inputs.txt", [_row(0), _row(1)]) + outputs = _a_log(tmp_path, "a.outputs.txt", [_row(0), _row(2)]) + + with pytest.raises(RuntimeError): + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +@pytest.mark.parametrize("index", [True, False, 1.0, -1, [], "1", None]) +def test_an_index_that_is_not_a_whole_number_is_refused(tmp_path, index): + """Only a non-negative int names a simulation, whatever compares equal.""" + # True and 1.0 both equal 1, so either could stand in for a simulation that + # was never run. An unhashable one used to escape as a raw TypeError. + rows = [_row(0), _row(index)] + inputs = _a_log(tmp_path, "odd.inputs.txt", rows) + outputs = _a_log(tmp_path, "odd.outputs.txt", rows) + + with pytest.raises(RuntimeError, match="cannot be read"): + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +def test_a_stray_index_past_the_rows_is_refused(tmp_path): + """A number no run of this length could have produced is not a simulation.""" + # A worker that claimed one index too many writes exactly this, and the + # target alone cannot tell it from a checkpoint that is legitimately longer. + rows = [_row(0), _row(1), _row(99)] + inputs = _a_log(tmp_path, "stray.inputs.txt", rows) + outputs = _a_log(tmp_path, "stray.outputs.txt", rows) + + with pytest.raises(RuntimeError, match="past the 3 it holds"): + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +def test_the_order_a_parallel_run_finishes_in_is_not_a_hole(tmp_path): + """Rows arrive in completion order, which is not sorted and not wrong.""" + rows = [_row(1), _row(0), _row(2)] + inputs = _a_log(tmp_path, "para.inputs.txt", rows) + outputs = _a_log(tmp_path, "para.outputs.txt", rows) + + _refuse_logs_missing_a_simulation(inputs, outputs, 3) + + +def test_logs_that_disagree_on_the_order_are_refused(tmp_path): + """A record goes into both logs under one lock, so the order is the same.""" + inputs = _a_log(tmp_path, "order.inputs.txt", [_row(0), _row(1)]) + outputs = _a_log(tmp_path, "order.outputs.txt", [_row(1), _row(0)]) + + with pytest.raises(RuntimeError, match="same order"): + _refuse_logs_missing_a_simulation(inputs, outputs, 2) + + +def _leave_cleanly_without_recording(_flight): + # A worker that ends the way an out-of-memory kill ends it, but with the + # status of one that finished. Nothing about the process says otherwise. + os._exit(0) + + +def test_a_worker_that_leaves_cleanly_without_recording_is_not_a_success( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + """A zero exit with no row written makes ``simulate`` raise.""" + analysis = MonteCarlo( + filename=str(tmp_path / "study"), + environment=stochastic_environment, + rocket=stochastic_calisto, + flight=stochastic_flight, + data_collector={"leave": _leave_cleanly_without_recording}, + ) + + with pytest.raises(RuntimeError, match="incomplete"): + analysis.simulate( + number_of_simulations=6, append=False, parallel=True, n_workers=2 + ) + + +def test_no_failure_path_waits_on_a_worker_without_a_bound(): + """No ``join`` in the parallel path is called without a timeout.""" + # An unbounded join anywhere in the parallel path puts back the hang that + # the bounded teardown exists to end, and it does so where it is hardest + # to notice: only when a worker is already stuck. + tree = ast.parse(inspect.getsource(mc_module)) + run_in_parallel = next( + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef) and node.name == "__run_in_parallel" + ) + + unbounded = [ + node.lineno + for node in ast.walk(run_in_parallel) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "join" + and not node.args + and not node.keywords + ] + + assert not unbounded, f"join() with no timeout at lines {unbounded}" + + +def test_starting_a_worker_happens_inside_the_cleanup_scope(): + """A start that fails has to leave the workers before it accounted for.""" + # Structural for the same reason the unbounded-join check is: it only shows + # when a start has already failed, which a unit test cannot make a real + # process do. Outside the try, an interrupt there left children running. + tree = ast.parse(inspect.getsource(mc_module)) + run_in_parallel = next( + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef) and node.name == "__run_in_parallel" + ) + guarded = [ + node + for handler in ast.walk(run_in_parallel) + if isinstance(handler, ast.Try) + for node in ast.walk(handler) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "start" + ] + + assert guarded, "Process.start() is not inside a try in __run_in_parallel" + + +def test_a_worker_is_recorded_only_once_it_has_started(): + """A process that never started cannot be joined or terminated.""" + tree = ast.parse(inspect.getsource(mc_module)) + run_in_parallel = next( + node + for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef) and node.name == "__run_in_parallel" + ) + lines = {"start": None, "append": None} + for node in ast.walk(run_in_parallel): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + if node.func.attr == "start": + lines["start"] = node.lineno + if node.func.attr == "append": + lines["append"] = node.lineno + + assert lines["start"] is not None and lines["append"] is not None + assert lines["start"] < lines["append"], "appended before it started" + + +def test_a_collector_cannot_take_over_the_simulation_index( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + """A collector key called index is refused before the run touches a file.""" + # Measured before this was refused: every row was written with the + # collector's value, so the log said 999 twice for a two-simulation run + # and every check that reads an index was reading the wrong thing. + # Refused when the collector is handed over, which is before any file + # is opened, rather than at the end of a run that is already spoilt. + with pytest.raises(ValueError, match="index"): + MonteCarlo( + filename=str(tmp_path / "study"), + environment=stochastic_environment, + rocket=stochastic_calisto, + flight=stochastic_flight, + data_collector={"index": lambda flight: 999}, + ) + + +def test_a_collector_key_of_its_own_is_still_welcome( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + """The control. Only the one reserved name is refused.""" + analysis = MonteCarlo( + filename=str(tmp_path / "ok"), + environment=stochastic_environment, + rocket=stochastic_calisto, + flight=stochastic_flight, + data_collector={"apogee_twice": lambda flight: 2 * flight.apogee}, + ) + + analysis.simulate(number_of_simulations=1, append=False) + + with open(analysis.output_file, "r", encoding="utf-8") as written: + row = json.loads(next(line for line in written if line.strip())) + # Not the value of the index: how a run numbers its simulations is + # settled elsewhere, and pinning it here would tie this to that. + assert "index" in row + assert "apogee_twice" in row diff --git a/tests/unit/simulation/test_monte_carlo_worker_exit.py b/tests/unit/simulation/test_monte_carlo_worker_exit.py new file mode 100644 index 000000000..5b44bd352 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_worker_exit.py @@ -0,0 +1,78 @@ +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(): + """A fleet that all exited zero raises nothing.""" + _refuse_a_worker_that_did_not_finish([_worker(0), _worker(0)]) + + +def test_a_worker_killed_by_a_signal_is_refused(): + """A negative exit code names the worker and the signal that ended it.""" + 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(): + """A positive exit code is refused the same way a signal is.""" + 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(): + """The message names each unfinished worker and leaves the clean ones out.""" + 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): + """``None`` counts as unfinished, not as finished.""" + with pytest.raises(RuntimeError): + _refuse_a_worker_that_did_not_finish([_worker(exitcode), _worker(0)]) + + +def _leave_without_recording(_flight): + """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 +): + """A worker leaving through ``os._exit`` makes ``simulate`` raise.""" + # 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 + ) diff --git a/tests/unit/simulation/test_monte_carlo_worker_join.py b/tests/unit/simulation/test_monte_carlo_worker_join.py new file mode 100644 index 000000000..fc674ba75 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_worker_join.py @@ -0,0 +1,184 @@ +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, ignores_terminate=False): + self.exitcode = None + self._final_exitcode = exitcode + self._alive_for = alive_for + self._never = never + self._ignores_terminate = ignores_terminate + self.joins = 0 + self.timeouts = [] + self.terminated = False + self.killed = False + + def is_alive(self): + return self.exitcode is None + + def join(self, timeout=None): + self.joins += 1 + self.timeouts.append(timeout) + # A real worker that never returns makes the caller hang, which is the + # bug. Reproducing that here would hang CI instead of reporting, so the + # stand-in gives up and says so. + assert self.joins < 200, "the join loop never stopped waiting" + if self._never or self.joins <= self._alive_for: + return + self.exitcode = self._final_exitcode + + def terminate(self): + self.terminated = True + if not self._ignores_terminate: + self.exitcode = -15 + + def kill(self): + self.killed = True + self.exitcode = -9 + + +class _Event: + def __init__(self, already_set=False): + self.was_set = already_set + + def is_set(self): + return self.was_set + + def set(self): + self.was_set = True + + +def test_a_run_where_every_worker_finishes_is_left_alone(): + """A healthy fleet is joined to completion and never terminated.""" + 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(): + """One bad exit ends the wait for a sibling that never returns.""" + # 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(): + """The event is set before anything is terminated.""" + 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(): + """A worker that leaves during the grace period is left alone.""" + 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): + """Signals and non-zero codes both start the shutdown.""" + 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(): + """Elapsed time is not evidence: a slow fleet is polled, never stopped.""" + # 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 + + +def test_a_reported_failure_ends_the_wait(): + """A worker that reported leaves cleanly, so its exit status says nothing.""" + # Left alone this waits on the sibling for good, and the run is already + # short a simulation whichever way that goes. The sibling is asked first + # and gets the longer grace, since one that only read the event is working + # rather than blocked on a lock nobody owns. + reported = _Worker(exitcode=0, alive_for=1) + stuck = _Worker(never=True) + + _join_the_workers([reported, stuck], _Event(already_set=True), grace_period=0) + + assert stuck.terminated or stuck.killed + assert max(t for t in stuck.timeouts if t is not None) >= 1.0 + + +def test_a_sibling_that_finishes_inside_the_grace_is_not_ended(): + """Asked first, and a worker that leaves on its own is left to do it.""" + reported = _Worker(exitcode=0, alive_for=1) + finishing = _Worker(alive_for=1) + + _join_the_workers([reported, finishing], _Event(already_set=True), grace_period=0) + + assert not finishing.terminated + assert finishing.exitcode == 0 + + +def test_a_clean_run_is_not_stopped_by_an_event_nobody_set(): + """An unset event leaves a healthy run running.""" + first, second = _Worker(alive_for=2), _Worker(alive_for=3) + + _join_the_workers([first, second], _Event(), grace_period=0) + + assert not any(worker.terminated for worker in (first, second)) + + +def test_a_worker_that_ignores_terminate_is_killed(): + """Terminate can be ignored; the fleet still has to come down.""" + died = _Worker(exitcode=-9, alive_for=1) + stubborn = _Worker(never=True, ignores_terminate=True) + + _join_the_workers([died, stubborn], _Event(), grace_period=0) + + assert stubborn.terminated + assert stubborn.killed + + +def test_the_fleet_comes_down_on_one_deadline_not_one_each(): + """A stage gives the fleet one grace period between them, not each.""" + # Observed through what each worker is offered: with a deadline of its own + # every worker is given the whole grace, so a fleet of thirty takes thirty + # times as long to give up on. + died = _Worker(exitcode=-9, alive_for=1) + stuck = [_Worker(never=True) for _ in range(4)] + + _join_the_workers([died, *stuck], _Event(), grace_period=0.05) + + offered = [t for t in stuck[-1].timeouts if t is not None] + assert offered + assert min(offered) < 0.05 diff --git a/tests/unit/simulation/test_monte_carlo_worker_reporting.py b/tests/unit/simulation/test_monte_carlo_worker_reporting.py new file mode 100644 index 000000000..9c5b9c896 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_worker_reporting.py @@ -0,0 +1,406 @@ +import json +import os +from contextlib import suppress +from types import SimpleNamespace + +import pytest + +from rocketpy.simulation import monte_carlo as mc_module +from rocketpy.simulation.monte_carlo import MonteCarlo + + +class _Mutex: + def __init__(self): + self.held = False + self.acquired = 0 + self.blocking = None + self.timeout = None + + def acquire(self, blocking=True, timeout=None): # the proxy's signature + self.acquired += 1 + self.blocking, self.timeout = blocking, timeout + self.held = True + return True + + def release(self): + self.held = False + + +class _MutexThatCannotBeTaken(_Mutex): + """A lock a dead holder never gave back: acquire waits out its bound.""" + + def acquire(self, blocking=True, timeout=None): + super().acquire(blocking, timeout) + self.held = False + return False + + +class _MutexThatBreaksOnAcquire(_Mutex): + """The manager is gone, so asking for the lock raises instead.""" + + def acquire(self, blocking=True, timeout=None): + super().acquire(blocking, timeout) + self.held = False + raise OSError("the manager is gone") + + +class _MutexThatBreaksOnRelease(_Mutex): + """The manager goes while the lock is held, so giving it back raises.""" + + def release(self): + raise OSError("the manager is gone") + + +class _ErrorEvent: + def __init__(self, refuse=False): + self.was_set = False + self.refuse = refuse + + def is_set(self): + return self.was_set + + def set(self): + if self.refuse: + raise OSError("the manager is gone") + self.was_set = True + + +def _raise_instead(message): + def refuse(*_args, **_kwargs): + raise OSError(message) + + return refuse + + +def _refusing_model(): + def refuse(_seed): + raise RuntimeError("the models would not reseed") + + return SimpleNamespace(last_rnd_dict={}, _set_stochastic=refuse) + + +def _a_worker(tmp_path, model, event=None): + study = MonteCarlo( + filename=str(tmp_path / "study"), + environment=model, + rocket=model, + flight=model, + ) + return study, event or _ErrorEvent() + + +def _run(study, monitor, error_event, mutex=None): + # Name-mangled: the producer is what each worker process runs, and nothing + # else in the suite calls it. + mutex = mutex or _Mutex() + study._MonteCarlo__sim_producer(42, monitor, mutex, error_event) + return mutex + + +def test_a_worker_that_fails_before_seeding_finishes_says_so(tmp_path, capsys): + """A failure above the loop is reported against worker startup.""" + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, monitor, error_event) + + assert error_event.was_set + reported = capsys.readouterr().out + assert "worker startup" in reported + assert "the models would not reseed" in reported + + +def test_a_worker_that_fails_before_claiming_an_index_says_so(tmp_path, capsys): + """A failed claim is startup too, since no index was taken.""" + + def refuse(): + raise RuntimeError("the monitor would not hand out an index") + + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + monitor = SimpleNamespace(keep_simulating=lambda: True, increment=refuse) + study, error_event = _a_worker(tmp_path, model) + + _run(study, monitor, error_event) + + assert error_event.was_set + assert "worker startup" in capsys.readouterr().out + + +def test_a_worker_that_fails_inside_a_simulation_names_the_index( + tmp_path, capsys, monkeypatch +): + """A failure after a claim is reported against that index.""" + + # The control. An index is claimed and the simulation then fails, which is + # the path that already worked, so the report still has to name it. + def refuse(_self): + raise RuntimeError("the simulation would not run") + + monkeypatch.setattr( + MonteCarlo, "_MonteCarlo__run_single_simulation", refuse, raising=True + ) + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 8) + study, error_event = _a_worker(tmp_path, model) + + _run(study, monitor, error_event) + + assert error_event.was_set + assert "iteration 7" in capsys.readouterr().out + + +def test_a_startup_failure_is_written_down_and_not_only_printed(tmp_path): + """The error log gets a row even when no inputs were drawn.""" + # The caller is told to read the error file, and a traceback the worker + # printed is not there to be read once its output has been redirected. + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, monitor, error_event) + + with open(study.error_file, "r", encoding="utf-8") as recorded: + rows = [json.loads(line) for line in recorded if line.strip()] + assert len(rows) == 1 + assert rows[0]["index"] is None + assert rows[0]["stage"] == "worker startup" + assert "the models would not reseed" in rows[0]["error"] + + +@pytest.mark.parametrize("failing", ["_set_stochastic", "increment"]) +def test_a_worker_failure_never_raises_out_of_the_producer(tmp_path, failing): + """A reported failure leaves the producer without an exception.""" + + # The handler used to reach for names the loop had not bound yet, so the + # process died with UnboundLocalError and the parent waited forever. + def refuse(*_args): + raise RuntimeError("boom") + + model = SimpleNamespace( + last_rnd_dict={}, + _set_stochastic=refuse if failing == "_set_stochastic" else lambda _s: None, + ) + monitor = SimpleNamespace( + keep_simulating=lambda: True, + increment=refuse if failing == "increment" else (lambda: 1), + ) + study, error_event = _a_worker(tmp_path, model) + + _run(study, monitor, error_event) + + assert error_event.was_set + + +@pytest.mark.parametrize("breaking", ["error_file", "reprint", "event"]) +def test_reporting_a_failure_never_keeps_the_mutex(tmp_path, monkeypatch, breaking): + """The manager lock is released however the reporting goes.""" + # The mutex is the manager's, so a worker that ends while holding it leaves + # the next one waiting on a process that is gone, and the parent never + # reaches the join that would have noticed. + if breaking == "error_file": + monkeypatch.setattr( + mc_module, "_worker_failure_record", _raise_instead("no disk") + ) + if breaking == "reprint": + monkeypatch.setattr( + mc_module._SimMonitor, "reprint", _raise_instead("no stdout") + ) + event = _ErrorEvent(refuse=breaking == "event") + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model(), event) + mutex = _Mutex() + + # A worker that could not announce its failure re-raises on the way out, so + # that its exit code carries what the event could not. The lock still has + # to be back either way, which is what this is about. + with suppress(RuntimeError): + _run(study, monitor, error_event, mutex) + + assert mutex.acquired == 1 + assert not mutex.held + + +def test_a_reporting_failure_does_not_replace_the_simulation_failure( + tmp_path, monkeypatch, capsys +): + """An unwritable log does not hide what actually failed.""" + monkeypatch.setattr(mc_module, "_worker_failure_record", _raise_instead("no disk")) + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, monitor, error_event) + + assert error_event.was_set + assert "the models would not reseed" in capsys.readouterr().out + assert not os.path.getsize(study.error_file) + + +def _committing_producer(monkeypatch): + """Make one simulation run start to finish without a real flight.""" + monkeypatch.setattr( + MonteCarlo, "_MonteCarlo__run_single_simulation", lambda self: None + ) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_inputs", + lambda self, index: json.dumps({"index": index, "committed": True}) + "\n", + ) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_outputs", + lambda self, flight, index: json.dumps({"index": index}) + "\n", + ) + + +def _one_then_broken(): + calls = {"count": 0} + + def keep_simulating(): + calls["count"] += 1 + if calls["count"] == 1: + return True + raise RuntimeError("the monitor died between simulations") + + return SimpleNamespace( + keep_simulating=keep_simulating, + increment=lambda: 1, + print_update_status=lambda: None, + ) + + +def test_a_failure_between_simulations_is_not_blamed_on_the_last_one( + tmp_path, capsys, monkeypatch +): + """A failure after a committed row is not reported against it.""" + # Simulation 0 finishes and its row is committed. The next claim then + # fails, which is not simulation 0's doing and must not be recorded as it. + _committing_producer(monkeypatch) + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + study, error_event = _a_worker(tmp_path, model) + + _run(study, _one_then_broken(), error_event) + + assert "worker startup" in capsys.readouterr().out + + +def test_a_committed_row_is_not_written_to_the_error_log_as_well(tmp_path, monkeypatch): + """A row that succeeded appears in one log, not in both.""" + _committing_producer(monkeypatch) + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + study, error_event = _a_worker(tmp_path, model) + + _run(study, _one_then_broken(), error_event) + + with open(study.output_file, "r", encoding="utf-8") as written: + committed = [json.loads(line) for line in written if line.strip()] + with open(study.error_file, "r", encoding="utf-8") as recorded: + errored = [json.loads(line) for line in recorded if line.strip()] + + assert committed == [{"index": 0}] + assert all(row.get("committed") is None for row in errored) + + +def test_a_worker_that_cannot_announce_its_failure_does_not_exit_cleanly(tmp_path): + """With the event unreachable the producer raises, so the exit is not zero.""" + # The event is how a worker reaches the parent. With it unreachable, the + # only signal left is how the process ends, so it must not end well. + model = _refusing_model() + study, error_event = _a_worker(tmp_path, model, _ErrorEvent(refuse=True)) + + with pytest.raises(RuntimeError, match="the models would not reseed"): + _run(study, SimpleNamespace(keep_simulating=lambda: True), error_event) + + +def test_a_worker_that_did_announce_its_failure_returns(tmp_path): + """With the event delivered the producer returns on purpose.""" + # The control. With the event delivered the parent already knows, so the + # producer returns and the process exits cleanly on purpose. + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, SimpleNamespace(keep_simulating=lambda: True), error_event) + + assert error_event.was_set + + +@pytest.mark.parametrize( + "mutex_class", [_MutexThatCannotBeTaken, _MutexThatBreaksOnAcquire] +) +def test_a_lock_the_reporter_cannot_take_does_not_stop_it( + tmp_path, capsys, mutex_class +): + """A lock that times out or is gone still leaves the failure announced.""" + # Asking for it without a bound is how a worker whose sibling died holding + # the lock waits forever, with nothing recorded and no exit code to read. + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + mutex = mutex_class() + + _run(study, monitor, error_event, mutex) + + assert error_event.was_set + assert "the models would not reseed" in capsys.readouterr().out + assert not mutex.held + assert mutex.timeout is not None # asked for with a bound + + +def test_a_lock_that_breaks_on_release_does_not_hide_the_failure(tmp_path, capsys): + """Giving the lock back can raise, and must not replace what failed.""" + monitor = SimpleNamespace(keep_simulating=lambda: True) + study, error_event = _a_worker(tmp_path, _refusing_model()) + + _run(study, monitor, error_event, _MutexThatBreaksOnRelease()) + + assert error_event.was_set + assert "the models would not reseed" in capsys.readouterr().out + + +def test_a_failure_after_the_inputs_were_drawn_still_records_why(tmp_path, monkeypatch): + """The error file is where the caller is sent, so it has to say what broke.""" + # Writing the input row on its own left no stage and no traceback there, + # for every failure past the point the inputs had been built. + _committing_producer(monkeypatch) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_outputs", + _raise_instead("the outputs would not serialize"), + ) + monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 1) + study, error_event = _a_worker( + tmp_path, SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _s: None) + ) + + _run(study, monitor, error_event) + + with open(study.error_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + assert len(rows) == 1 + assert "the outputs would not serialize" in rows[0]["error"] + assert rows[0]["stage"] == "iteration 0" + assert rows[0]["inputs"]["committed"] is True + + +def test_an_input_row_that_is_not_an_object_does_not_break_the_reporter( + tmp_path, monkeypatch +): + """Reading the row is best effort: the failure being reported comes first.""" + _committing_producer(monkeypatch) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_inputs", + lambda self, index: json.dumps([index]) + "\n", + ) + monkeypatch.setattr( + MonteCarlo, + "_MonteCarlo__evaluate_flight_outputs", + _raise_instead("the outputs would not serialize"), + ) + monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 1) + study, error_event = _a_worker( + tmp_path, SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _s: None) + ) + + _run(study, monitor, error_event) + + with open(study.error_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + assert "the outputs would not serialize" in rows[0]["error"] + assert "inputs" not in rows[0]