diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py index c2dcd4030..9d982df10 100644 --- a/rocketpy/simulation/monte_carlo.py +++ b/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. diff --git a/rocketpy/stochastic/stochastic_model.py b/rocketpy/stochastic/stochastic_model.py index d42fb76c5..1dadb2f01 100644 --- a/rocketpy/stochastic/stochastic_model.py +++ b/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. diff --git a/rocketpy/tools.py b/rocketpy/tools.py index 0d7f1a74e..7f31f3e19 100644 --- a/rocketpy/tools.py +++ b/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. diff --git a/tests/unit/simulation/test_monte_carlo_parallel_runs.py b/tests/unit/simulation/test_monte_carlo_parallel_runs.py new file mode 100644 index 000000000..4ab0be440 --- /dev/null +++ b/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) 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..005eb31d6 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_run_completeness.py @@ -0,0 +1,123 @@ +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_a_run_that_lost_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): + inputs, outputs = _complete(tmp_path) + + _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3) + + +def test_a_missing_simulation_is_refused(tmp_path): + 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_a_run_that_lost_a_simulation(inputs, outputs, 3) + + +def test_a_simulation_recorded_twice_is_refused(tmp_path): + 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_a_run_that_lost_a_simulation(inputs, outputs, 3) + + +def test_a_row_that_cannot_be_read_is_refused(tmp_path): + 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_a_run_that_lost_a_simulation(inputs, outputs, 3) + + +def test_a_row_numbered_past_the_run_is_refused(tmp_path): + inputs, outputs = _complete(tmp_path) + _a_log(tmp_path, "ok.inputs.txt", [_row(0), _row(1), _row(2), _row(9)]) + + with pytest.raises(RuntimeError, match="past the run"): + _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3) + + +def test_logs_that_hold_different_simulations_are_refused(tmp_path): + 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_a_run_that_lost_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 +): + 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(): + # 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}" 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..d9ca0dc68 --- /dev/null +++ b/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 + ) 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..20ef8df22 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_worker_join.py @@ -0,0 +1,145 @@ +import pytest + +from rocketpy.simulation.monte_carlo import _join_the_workers + + +class _Worker: + """A process that stops after a set number of polls, or never. + + ``never`` stands in for one blocked on a lock its dead sibling was holding, + which is the case an unbounded join waits out forever. + """ + + def __init__(self, exitcode=0, alive_for=0, never=False): + self.exitcode = None + self._final_exitcode = exitcode + self._alive_for = alive_for + self._never = never + self.joins = 0 + self.terminated = False + + def is_alive(self): + return self.exitcode is None + + def join(self, timeout=None): # pylint: disable=unused-argument + self.joins += 1 + # 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 + self.exitcode = -15 + + +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 + + +class _BrokenEvent(_Event): + """A manager proxy that has gone away.""" + + def is_set(self): + raise OSError("the manager is gone") + + +def test_a_run_where_every_worker_finishes_is_left_alone(): + workers = [_Worker(alive_for=3), _Worker(alive_for=5)] + + _join_the_workers(workers, _Event(), grace_period=0) + + assert [worker.exitcode for worker in workers] == [0, 0] + assert not any(worker.terminated for worker in workers) + + +def test_a_worker_blocked_behind_a_dead_one_does_not_wait_forever(): + # The one that mattered. Without a bound this call never returns, so the + # parent never reaches the check that would have reported the failure. + died = _Worker(exitcode=-9, alive_for=1) + blocked = _Worker(never=True) + + _join_the_workers([died, blocked], _Event(), grace_period=0) + + assert blocked.terminated + + +def test_the_survivors_are_asked_before_they_are_ended(): + died = _Worker(exitcode=1, alive_for=1) + blocked = _Worker(never=True) + event = _Event() + + _join_the_workers([died, blocked], event, grace_period=0) + + assert event.was_set + + +def test_a_survivor_that_stops_on_its_own_is_not_terminated(): + died = _Worker(exitcode=1, alive_for=1) + cooperative = _Worker(alive_for=2) + + _join_the_workers([died, cooperative], _Event(), grace_period=0) + + assert not cooperative.terminated + assert cooperative.exitcode == 0 + + +@pytest.mark.parametrize("exitcode", [-9, 1, 2]) +def test_any_bad_exit_starts_the_shutdown(exitcode): + died = _Worker(exitcode=exitcode, alive_for=1) + blocked = _Worker(never=True) + + _join_the_workers([died, blocked], _Event(), grace_period=0) + + assert blocked.terminated + + +def test_a_slow_run_is_never_bounded(): + # Nothing here may act on how long a worker takes, only on it having died. + slow = _Worker(alive_for=50) + slower = _Worker(alive_for=80) + + _join_the_workers([slow, slower], _Event(), grace_period=0) + + assert not any(worker.terminated for worker in (slow, slower)) + assert slower.joins > 50 + + +def test_a_reported_failure_also_stops_a_sibling_that_never_returns(): + # A worker that fails the ordinary way is caught by the producer, reports + # through the event and returns, so it exits cleanly. Waiting only on exit + # codes leaves the parent sitting behind whichever sibling is stuck. + 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 + + +def test_a_clean_run_is_not_stopped_by_an_event_nobody_set(): + 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_an_event_that_cannot_be_read_does_not_stop_the_run(): + # The control on the control. If asking the manager raises, that is not + # evidence of failure and must not end a healthy run. + first, second = _Worker(alive_for=2), _Worker(alive_for=3) + + _join_the_workers([first, second], _BrokenEvent(), grace_period=0) + + assert not any(worker.terminated for worker in (first, second)) 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..cf9d51ebc --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_worker_reporting.py @@ -0,0 +1,189 @@ +import json +import os +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 + + def acquire(self): + self.acquired += 1 + self.held = True + + def release(self): + self.held = False + + +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): + 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): + 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 +): + # 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 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): + # 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 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 = _run(study, monitor, error_event) + + assert mutex.acquired == 1 + assert not mutex.held + + +def test_a_reporting_failure_does_not_replace_the_simulation_failure( + tmp_path, monkeypatch, capsys +): + 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) diff --git a/tests/unit/stochastic/test_seed_types.py b/tests/unit/stochastic/test_seed_types.py new file mode 100644 index 000000000..ba3d42583 --- /dev/null +++ b/tests/unit/stochastic/test_seed_types.py @@ -0,0 +1,81 @@ +import numpy as np +import pytest + +from rocketpy.stochastic.stochastic_model import ( + _names_as_spawn_key, + _sampler_seed, +) +from rocketpy.tools import _seed_sequence_to_int + + +def _a_worker_seed(index=0, workers=2): + # What MonteCarlo.__run_in_parallel spawns and hands to each worker, which + # passes it straight to environment/rocket/flight._set_stochastic. + return np.random.SeedSequence().spawn(workers)[index] + + +def test_the_seed_type_a_worker_is_handed_is_accepted(stochastic_calisto): + stochastic_calisto._set_stochastic(_a_worker_seed()) + + stochastic_calisto.create_object() + + +def test_a_parachute_derives_its_noise_seed_from_a_worker_seed( + stochastic_main_parachute, +): + stochastic_main_parachute._set_stochastic(_a_worker_seed()) + + assert stochastic_main_parachute.create_object().noise[2] is not None + + +def test_two_workers_do_not_share_a_sampler_stream(): + first, second = np.random.SeedSequence(7).spawn(2) + # They come off one root, so they carry the same entropy and differ only in + # spawn_key. Reading the entropy alone would put both on one stream. + assert first.entropy == second.entropy + + assert _sampler_seed(first, ("__list_choice__",)) != _sampler_seed( + second, ("__list_choice__",) + ) + + +def test_a_caller_seed_sequence_is_not_consumed(): + root = np.random.SeedSequence(42) + + _sampler_seed(root, ("__list_choice__",)) + + assert root.n_children_spawned == 0 + assert root.spawn(1)[0].spawn_key == (0,) + + +def test_the_same_seed_sequence_twice_gives_the_same_sampler_seed(): + root = np.random.SeedSequence(42) + + first = _sampler_seed(root, ("pressure_noise", "main")) + second = _sampler_seed(root, ("pressure_noise", "main")) + + assert first == second + + +@pytest.mark.parametrize("seed", [42, 7, [1, 2, 3]]) +@pytest.mark.parametrize("names", [("__list_choice__",), ("pressure_noise", "main")]) +def test_a_seed_that_is_not_a_sequence_reaches_numpy_untouched(seed, names): + # The control. Every fixed-seed baseline in the suite was recorded through + # this path, so anything but a SeedSequence has to arrive as it always did. + # Compared with the expression rather than with a recorded number, which + # would go red on a NumPy release instead of on a change of ours. + unchanged = np.random.SeedSequence( + entropy=seed, spawn_key=_names_as_spawn_key(tuple(sorted(names))) + ) + + assert _sampler_seed(seed, names) == _seed_sequence_to_int(unchanged) + + +def test_no_seed_still_means_no_seed(): + # None is left out above on purpose: it asks NumPy for fresh entropy, so + # two calls must not agree, and comparing one against another would be + # asserting the opposite of what an unseeded run promises. + first = _sampler_seed(None, ("__list_choice__",)) + second = _sampler_seed(None, ("__list_choice__",)) + + assert first != second