diff --git a/CHANGELOG.md b/CHANGELOG.md index 933226147..0469789ef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,7 @@ Attention: The newest changes should be on top --> ### Added +- ENH: Reproducible Monte Carlo seeding, and an append that continues the same study [#1187](https://github.com/RocketPy-Team/RocketPy/pull/1187) [#1053](https://github.com/RocketPy-Team/RocketPy/issues/1053) [#1075](https://github.com/RocketPy-Team/RocketPy/issues/1075) - ENH: Support fixed-time parachute deployment triggers [#1133](https://github.com/RocketPy-Team/RocketPy/pull/1133) [#437](https://github.com/RocketPy-Team/RocketPy/issues/437) - DOC: Add SIL parachute ejection integration example [#1131](https://github.com/RocketPy-Team/RocketPy/pull/1131) [#524](https://github.com/RocketPy-Team/RocketPy/issues/524) - ENH: List NOAA atmosphere datasets and fetch latest [#1136](https://github.com/RocketPy-Team/RocketPy/pull/1136) [#660](https://github.com/RocketPy-Team/RocketPy/issues/660) diff --git a/docs/notebooks/monte_carlo_analysis/monte_carlo_class_usage.ipynb b/docs/notebooks/monte_carlo_analysis/monte_carlo_class_usage.ipynb index 8181c03ba..24c76df86 100644 --- a/docs/notebooks/monte_carlo_analysis/monte_carlo_class_usage.ipynb +++ b/docs/notebooks/monte_carlo_analysis/monte_carlo_class_usage.ipynb @@ -772,7 +772,9 @@ "Finally, let's simulate our flights. \n", "We can run the simulations using the method `MonteCarlo.simulate()`.\n", "\n", - "Set `append=False` to overwrite the previous results, or `append=True` to add the new results to the previous ones.\n" + "Set `append=False` to overwrite the previous results, or `append=True` to add the new results to the previous ones.\n", + "\n", + "An append carries on the study already in the file. Every input row records the seed root it was drawn from, so the seed does not have to be given again, and one that disagrees with the rows is refused rather than mixed in. A log written before this behaviour existed carries no root, so an append onto one is refused too: run the study again to write a log that can be continued.\n" ] }, { diff --git a/docs/user/stochastic.rst b/docs/user/stochastic.rst index 2a9e9ca8e..0dd0a9fe5 100644 --- a/docs/user/stochastic.rst +++ b/docs/user/stochastic.rst @@ -341,6 +341,23 @@ draws under a fixed seed. The rocket's own inputs, such as ``mass`` and one model. Sharing one between two components leaves each of them seeding it from their own child, and the last one to be reset decides what both draw. +.. note:: + A whole run is fixed by ``MonteCarlo.simulate(random_seed=...)`` rather than + by seeding these models yourself. Each simulation takes its seed from its own + index, so simulation 7 draws the same inputs whether the run was serial or + split over any number of workers, and whether it was reached first or last. + Every input row records the root it came from, so appending carries that + study on whether or not the seed is given again, and a different one is + refused rather than mixed in. Without a seed a run draws fresh entropy and + reproduces nothing. + + An index fixes the draw, not the object it is drawn around. Where the note + above says a value follows its deterministic object, moving that object + between runs still moves what is sampled, however the seed was set. A + ``CustomSampler`` that draws from the process-global ``numpy.random`` + rather than from the generator it is handed sits outside all of this, as + :ref:`custom_sampler` warns. + Conclusion ---------- diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py index 8c00c1385..a9f87e776 100644 --- a/rocketpy/simulation/monte_carlo.py +++ b/rocketpy/simulation/monte_carlo.py @@ -14,11 +14,14 @@ """ import csv +import hashlib import json import os +import threading import traceback import warnings from contextlib import suppress +from copy import deepcopy from numbers import Real from pathlib import Path from time import monotonic, time @@ -32,6 +35,7 @@ from rocketpy.prints.monte_carlo_prints import _MonteCarloPrints from rocketpy.simulation.flight import Flight from rocketpy.tools import ( + _seed_sequence_to_int, generate_monte_carlo_ellipses, generate_monte_carlo_ellipses_coordinates, import_optional_dependency, @@ -57,6 +61,198 @@ # and leaving at the end of the simulation in hand, not blocked on a dead lock. _REPORTED_FAILURE_GRACE_SECONDS = 60.0 +# Which root drew a row. An append reads it to continue the same stream. +_SIMULATION_ROOT_KEY = "run_root" + +# Told apart from a row that carries ``None``, which no run writes. +_NOTHING_READ_YET = object() + + +def _root_seed_sequence(random_seed): + """The immutable root a run derives every simulation's seed from. + + A ``SeedSequence`` is rebuilt from its full state rather than used as + given, since ``spawn`` advances a counter the caller still holds. A + ``Generator`` is refused rather than read, because using a consume-on-use + object as an immutable seed cannot mean what it says. + """ + if isinstance(random_seed, np.random.SeedSequence): + return np.random.SeedSequence(**random_seed.state) + if isinstance(random_seed, (np.random.Generator, np.random.BitGenerator)): + raise TypeError( + f"random_seed must be an int, a sequence of non-negative integers, " + f"or a numpy.random.SeedSequence, not a " + f"{type(random_seed).__name__}. Pass the seed the generator was " + f"built from." + ) + return np.random.SeedSequence(random_seed) + + +def _jsonable_entropy(entropy): + """``SeedSequence`` entropy as something ``json`` will take. + + It may be an int, a sequence or an ndarray; only the first survives. + """ + if entropy is None or isinstance(entropy, (int, np.integer)): + return None if entropy is None else int(entropy) + return [int(part) for part in np.asarray(entropy).ravel()] + + +def _root_written_into_a_row(root_state): + """The run's root as one JSON value, carried by every input row. + + In the rows because a file beside a log cannot be shown to belong to it. + """ + entropy, spawn_key, pool_size, base = root_state + return { + "entropy": _jsonable_entropy(entropy), + "spawn_key": [int(key) for key in spawn_key], + "pool_size": int(pool_size), + "n_children_spawned": int(base), + } + + +_ROOT_FIELDS = frozenset(("entropy", "spawn_key", "pool_size", "n_children_spawned")) + + +def _root_digest(root): + """A short stable name for a root, for rows that only have to match one. + + 64 bits of SHA-256, carried by the output rows instead of the root itself, + which would cost a hundred bytes a row on a study the input log already + records it for. Wide enough to tell studies apart, not a signature. + """ + canonical = json.dumps(root, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16] + + +def _a_whole_number(value): + """A non-negative int, and not a bool standing in for one.""" + return isinstance(value, int) and not isinstance(value, bool) and value >= 0 + + +def _whole_numbers(value, may_be_empty=False): + """A list of non-negative ints, empty only where that is a valid one.""" + return ( + isinstance(value, list) + and (may_be_empty or len(value) > 0) + and all(_a_whole_number(part) for part in value) + ) + + +def _root_state_a_row_records(root, path): + """The four values a root is rebuilt from, refused unless all are usable. + + Agreeing rows show the study is one study, not that what they agree on can + be resumed: ``SeedSequence(entropy=None)`` draws fresh entropy every time. + """ + entropy = root.get("entropy") if isinstance(root, dict) else None + usable = ( + isinstance(root, dict) + and set(root) == _ROOT_FIELDS + and (_a_whole_number(entropy) or _whole_numbers(entropy)) + and _whole_numbers(root.get("spawn_key"), may_be_empty=True) + and _a_whole_number(root.get("pool_size")) + and _a_whole_number(root.get("n_children_spawned")) + ) + if usable: + state = ( + entropy, + tuple(root["spawn_key"]), + root["pool_size"], + root["n_children_spawned"], + ) + try: + # Drawing one child is what proves the pool size numpy will take. + _seed_of_simulation(state, 0) + return state + except ValueError: + pass + raise ValueError( + f"cannot continue {path}: its rows record a root that no stream " + f"can be rebuilt from, so what they were drawn with is unknown." + ) + + +def _what_the_rows_say_drew_them(path): + """What every row agrees drew it, and the simulations it numbers. + + ``(None, [])`` means the log holds no rows, and nothing else does. + + ``None`` means the log holds no rows, and nothing else does. Rows that + carry no root are refused instead: they cannot be shown to be one study, + and reading them as an empty log would start a second one in the file. + A log whose rows disagree is refused for the same reason. + """ + first = _NOTHING_READ_YET + numbered = [] + with open(path, "r", encoding="utf-8") as recorded: + for line in recorded: + if not line.strip(): + continue + try: + row = json.loads(line) + except ValueError as error: + raise ValueError( + f"cannot continue {path}: a row cannot be read, so what " + f"produced it cannot be established." + ) from error + root = row.get(_SIMULATION_ROOT_KEY) if isinstance(row, dict) else None + if root is None: + raise ValueError( + f"cannot continue {path}: a row does not say which root " + f"drew it, which is how a study written before this " + f"release looks. Start a new one rather than continuing " + f"one whose rows cannot be checked." + ) + index = row.get("index") + if not _a_whole_number(index): + raise ValueError( + f"cannot continue {path}: a row does not number the " + f"simulation it holds, so where to carry on from cannot " + f"be established." + ) + numbered.append(index) + if first is _NOTHING_READ_YET: + first = root + elif root != first: + raise ValueError( + f"cannot continue {path}: its rows were not all drawn " + f"from one root, so it holds more than one study." + ) + return (None if first is _NOTHING_READ_YET else first), numbered + + +def _root_state_of(root): + """A root as the four picklable values a worker can rebuild it from. + + Sent to each worker instead of the object, and instead of the list of + children, so a run of a million simulations costs four values. The entropy + is copied because a sequence one is kept by reference all the way from the + caller, who could otherwise still move every child by editing their list. + """ + return ( + deepcopy(root.entropy), + tuple(root.spawn_key), + root.pool_size, + root.n_children_spawned, + ) + + +def _seed_of_simulation(root_state, sim_idx): + """The seed for one simulation index, without spawning the ones before it. + + ``spawn`` derives child ``i`` by appending ``n_children_spawned + i`` to + the parent spawn key, so rebuilding that one child directly reproduces it + and any index can be reached from the four values above alone. + """ + entropy, spawn_key, pool_size, base = root_state + return np.random.SeedSequence( + entropy=entropy, + spawn_key=(*spawn_key, base + sim_idx), + pool_size=pool_size, + ) + def _refuse_logs_this_run_cannot_write( input_file, output_file, error_file, export_config=None @@ -279,6 +475,8 @@ def simulate( append=False, parallel=False, n_workers=None, + *, + random_seed=None, **kwargs, ): """ @@ -298,6 +496,19 @@ def simulate( number of workers will be equal to the number of CPUs available. A minimum of 2 workers is required for parallel mode. Default is None. + random_seed : int, sequence of int or numpy.random.SeedSequence, optional + Fixes what every simulation draws. Simulation ``i`` takes the same + inputs whichever way the run was split up, so serial and parallel + results agree and the number of workers does not reach the + sampling. Keyword-only. Default is None, which draws fresh entropy + and reproduces nothing. + + Every input row carries the root it was drawn from, so an append + carries on from the study already in the file whether or not the + seed is given again. A different one is refused, not mixed in. + + A ``Generator`` or ``BitGenerator`` is refused rather than read. + Pass the seed it was built from. kwargs : dict Custom arguments for simulation export of the ``inputs`` file. Options are: @@ -338,13 +549,24 @@ def simulate( """ self._export_config = kwargs self.number_of_simulations = number_of_simulations - self._initial_sim_idx = self.num_of_loaded_sims if append else 0 - + self._initial_sim_idx = 0 + # Validated here, before __setup_files truncates anything, so an + # unusable seed cannot cost a previous run its results. Kept as four + # picklable values rather than as the object, since a worker rebuilds + # any index from them. + self.__root_state = _root_state_of(_root_seed_sequence(random_seed)) # Before anything is opened: __setup_files truncates for append=False. _refuse_logs_this_run_cannot_write( self.input_file, self.output_file, self.error_file, kwargs ) + # After that one, which says plainly that a .csv cannot be a working + # log. Reaching this first would report it as a row that cannot be read. + if append: + # From the checkpoint just validated, not the line count taken when + # this object was built: that one counts blank lines and goes stale. + self._initial_sim_idx = self.__continue_the_root_the_rows_carry(random_seed) + print("Starting Monte Carlo analysis") self.__setup_files(append) @@ -435,14 +657,18 @@ def __run_in_serial(self): n_simulations=self.number_of_simulations, start_time=time(), ) + sim_idx = sim_monitor.count try: - while sim_monitor.keep_simulating(): - sim_monitor.increment() + # Counted from zero, as the parallel path already does: the two + # used to name the same simulation 1, 2, 3 and 0, 1, 2. + while (claimed := sim_monitor.claim_next_index()) is not None: + sim_idx = claimed inputs_json, outputs_json = "", "" + self.__seed_this_simulation(sim_idx) flight = self.__run_single_simulation() - inputs_json = self.__evaluate_flight_inputs(sim_monitor.count) - outputs_json = self.__evaluate_flight_outputs(flight, sim_monitor.count) + inputs_json = self.__evaluate_flight_inputs(sim_idx) + outputs_json = self.__evaluate_flight_outputs(flight, sim_idx) self._append_simulation_record(inputs_json, outputs_json) @@ -456,7 +682,7 @@ def __run_in_serial(self): f.write(inputs_json) except Exception as error: - print(f"Error on iteration {sim_monitor.count}: {error}") + print(f"Error on iteration {sim_idx}: {error}") with open(self._error_file, "a", encoding="utf-8") as f: f.write(inputs_json) raise error @@ -491,14 +717,15 @@ def __run_in_parallel(self, n_workers=None): ) processes = [] - seeds = np.random.SeedSequence().spawn(n_workers) try: - for seed in seeds: + # No seed per worker any more: every simulation takes its own + # from its index, so the workers are interchangeable and how + # many there are does not reach the sampling. + for _ in range(n_workers): sim_producer = multiprocess.Process( target=self.__sim_producer, args=( - seed, sim_monitor, mutex, simulation_error_event, @@ -549,13 +776,11 @@ def __validate_number_of_workers(self, n_workers): raise ValueError("Number of workers must be at least 2 for parallel mode.") return n_workers - def __sim_producer(self, seed, sim_monitor, mutex, error_event): # pylint: disable=too-many-statements + def __sim_producer(self, sim_monitor, mutex, error_event): """Simulation producer to be used in parallel by multiprocessing. Parameters ---------- - seed : int - The seed to set the random number generator. sim_monitor : _SimMonitor The simulation monitor object to keep track of the simulations. mutex : multiprocess.Lock @@ -566,15 +791,10 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event): # pylint: disa # 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) - self.rocket._set_stochastic(seed) - self.flight._set_stochastic(seed) - - while sim_monitor.keep_simulating(): - sim_idx = sim_monitor.increment() - 1 + while (sim_idx := sim_monitor.claim_next_index()) is not None: inputs_json, outputs_json = "", "" + self.__seed_this_simulation(sim_idx) flight = self.__run_single_simulation() inputs_json = self.__evaluate_flight_inputs(sim_idx) outputs_json = self.__evaluate_flight_outputs(flight, sim_idx) @@ -644,6 +864,66 @@ def __report_a_failed_simulation(self, sim_idx, inputs_json, mutex, error_event) mutex.release() return announced + def __continue_the_root_the_rows_carry(self, random_seed): + """Take the root from the rows being appended to, or refuse to. + + Without this an append draws a second root into one file and nothing + afterwards can tell which simulation came from which. Reading it back + also means a fresh object can continue a study, which is the ordinary + way of resuming one. + """ + recorded, held = _what_the_rows_say_drew_them(self.input_file) + stamped, written = _what_the_rows_say_drew_them(self.output_file) + named = None if recorded is None else _root_digest(recorded) + if named != stamped: + raise ValueError( + f"cannot append to {self.input_file}: it and {self.output_file} " + f"were not drawn from the same root, so they are two studies " + f"rather than the two halves of one." + ) + if recorded is None: + return 0 + if held != written: + raise ValueError( + f"cannot append to {self.input_file}: it and {self.output_file} " + f"do not record the same simulations, so where to carry on " + f"from cannot be established." + ) + # Rows arrive in completion order, so these are not sorted. What has + # to hold is that between them they are the run's first len(held). + if sorted(held) != list(range(len(held))): + raise ValueError( + f"cannot append to {self.input_file}: the simulations it " + f"records are not the run's first {len(held)}, so where to " + f"carry on from cannot be established." + ) + state = _root_state_a_row_records(recorded, self.input_file) + if random_seed is None: + self.__root_state = state + return len(held) + if _root_written_into_a_row(self.__root_state) != recorded: + raise ValueError( + f"cannot append to {self.input_file}: its rows were drawn from " + f"a different root than random_seed gives. Continuing would put " + f"two studies in one file. Pass the seed the run started with, " + f"or leave random_seed out to carry on from the rows." + ) + return len(held) + + def __seed_this_simulation(self, sim_idx): + """Reseed the three models from this index's own child of the root. + + Per index rather than per worker, which is what makes a simulation's + inputs the same however the run was split up. The child is split three + ways so the environment, rocket and flight draw independently instead + of sharing one stream. + """ + child = _seed_of_simulation(self.__root_state, sim_idx) + environment, rocket, flight = child.spawn(3) + self.environment._set_stochastic(_seed_sequence_to_int(environment)) + self.rocket._set_stochastic(_seed_sequence_to_int(rocket)) + self.flight._set_stochastic(_seed_sequence_to_int(flight)) + def __run_single_simulation(self): """Runs a single simulation and returns the inputs and outputs. @@ -864,6 +1144,7 @@ def __evaluate_flight_inputs(self, sim_idx): for item in d.items() ) inputs_dict["index"] = sim_idx + inputs_dict[_SIMULATION_ROOT_KEY] = _root_written_into_a_row(self.__root_state) return ( json.dumps(inputs_dict, cls=RocketPyEncoder, **self._export_config) + "\n" ) @@ -887,8 +1168,6 @@ def __evaluate_flight_outputs(self, flight, sim_idx): export_item: getattr(flight, export_item) for export_item in self.export_list } - outputs_dict["index"] = sim_idx - if self.data_collector is not None: additional_exports = {} for key, callback in self.data_collector.items(): @@ -900,6 +1179,12 @@ def __evaluate_flight_outputs(self, flight, sim_idx): ) from e outputs_dict = outputs_dict | additional_exports + # After the collectors: these two say which run the row belongs to. + outputs_dict["index"] = sim_idx + outputs_dict[_SIMULATION_ROOT_KEY] = _root_digest( + _root_written_into_a_row(self.__root_state) + ) + return ( json.dumps(outputs_dict, cls=RocketPyEncoder, **self._export_config) + "\n" ) @@ -1054,6 +1339,13 @@ def _check_data_collector(self, data_collector): f"is written after the collectors run and cannot be " f"replaced by one." ) + if key == _SIMULATION_ROOT_KEY: + raise ValueError( + f"Invalid 'data_collector' key '{key}'! It is the root " + f"the row was drawn with, which is written after the " + f"collectors run, so a callback under that name would " + f"be run and then discarded." + ) if not callable(callback): raise ValueError( f"Invalid value in 'data_collector' for key '{key}'! " @@ -2068,13 +2360,20 @@ def __init__(self, initial_count, n_simulations, start_time): self.n_simulations = n_simulations self.start_time = start_time self.completed_count = 0 + self._claim = threading.Lock() # proxy calls run in the manager - def keep_simulating(self): - return self.count < self.n_simulations + def claim_next_index(self): + """The next index to run, or None once every one has been claimed. - def increment(self): - self.count += 1 - return self.count + One call, because a separate check and increment let two workers both + see the last slot free and then claim an index each past the end. + """ + with self._claim: + if self.count >= self.n_simulations: + return None + claimed = self.count + self.count += 1 + return claimed def print_update_status(self): """Prints a message on the same line as the previous one and replaces diff --git a/tests/unit/simulation/test_append_lineage.py b/tests/unit/simulation/test_append_lineage.py new file mode 100644 index 000000000..35f80844b --- /dev/null +++ b/tests/unit/simulation/test_append_lineage.py @@ -0,0 +1,391 @@ +import json + +import pytest + +from rocketpy.simulation.monte_carlo import ( + MonteCarlo, + _root_digest, + _SIMULATION_ROOT_KEY, +) + + +def _study(tmp_path, stem, models): + environment, rocket, flight = models + return MonteCarlo( + filename=str(tmp_path / stem), + environment=environment, + rocket=rocket, + flight=flight, + ) + + +def _drawn(analysis): + with open(analysis.input_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + return {row["index"]: row.get("mass") for row in rows} + + +@pytest.fixture(name="models") +def _models(stochastic_environment, stochastic_calisto, stochastic_flight): + return stochastic_environment, stochastic_calisto, stochastic_flight + + +def test_a_row_says_which_root_drew_it(models, tmp_path): + """Every input row carries the root, so the log describes itself.""" + analysis = _study(tmp_path, "study", models) + + analysis.simulate(2, append=False, random_seed=42) + + with open(analysis.input_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + assert all(row[_SIMULATION_ROOT_KEY]["entropy"] == 42 for row in rows) + + +def test_an_append_with_no_seed_carries_on_from_the_rows(models, tmp_path): + """The ordinary resume: no seed given, the stream continues anyway.""" + whole = _study(tmp_path, "whole", models) + whole.simulate(4, append=False, random_seed=42) + + part = _study(tmp_path, "part", models) + part.simulate(2, append=False, random_seed=42) + part.simulate(4, append=True) + + assert _drawn(part) == _drawn(whole) + + +def test_a_fresh_object_can_continue_a_study(models, tmp_path): + """A notebook restart is a new object over the same files.""" + first = _study(tmp_path, "study", models) + first.simulate(2, append=False, random_seed=42) + + second = _study(tmp_path, "study", models) + second.simulate(4, append=True) + + assert sorted(_drawn(second)) == [0, 1, 2, 3] + + +def test_appending_with_another_seed_is_refused(models, tmp_path): + """Two roots in one file is the thing this exists to prevent.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + + with pytest.raises(ValueError, match="different root"): + analysis.simulate(4, append=True, random_seed=7) + + +def test_appending_with_the_same_seed_is_allowed(models, tmp_path): + """The control. Saying the seed again is not a mismatch.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + + analysis.simulate(4, append=True, random_seed=42) + + assert sorted(_drawn(analysis)) == [0, 1, 2, 3] + + +def test_a_log_holding_two_studies_is_refused(models, tmp_path): + """Rows that disagree are two studies, and neither is safe to continue.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + with open(analysis.input_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + rows[1][_SIMULATION_ROOT_KEY]["entropy"] = 999 + with open(analysis.input_file, "w", encoding="utf-8") as rewritten: + for row in rows: + rewritten.write(json.dumps(row) + "\n") + + with pytest.raises(ValueError, match="more than one study"): + analysis.simulate(4, append=True) + + +def test_a_log_whose_rows_carry_no_root_is_refused(models, tmp_path): + """A study from before this release cannot be shown to be one study.""" + # Measured before this was refused: the append read the log as empty, + # started a root of its own, and left two lineages in the one file. + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + with open(analysis.input_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + for row in rows: + row.pop(_SIMULATION_ROOT_KEY) + with open(analysis.input_file, "w", encoding="utf-8") as rewritten: + for row in rows: + rewritten.write(json.dumps(row) + "\n") + + with pytest.raises(ValueError, match="does not say which root"): + analysis.simulate(4, append=True) + + +def _rewrite(path, replacement): + """Put ``replacement`` under the root key of every row of one log.""" + with open(path, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + for row in rows: + row[_SIMULATION_ROOT_KEY] = replacement(row[_SIMULATION_ROOT_KEY]) + with open(path, "w", encoding="utf-8") as rewritten: + for row in rows: + rewritten.write(json.dumps(row) + "\n") + + +def _rewrite_roots(analysis, damage): + """Damage the recorded root, leaving the two logs still naming one study. + + The output rows hold the root's digest, so they are restamped rather than + damaged: otherwise the two logs disagree and that is refused first. + """ + with open(analysis.input_file, "r", encoding="utf-8") as written: + first = json.loads(next(line for line in written if line.strip())) + damaged = damage(first[_SIMULATION_ROOT_KEY]) + _rewrite(analysis.input_file, lambda _root: damaged) + if isinstance(damaged, dict): + _rewrite(analysis.output_file, lambda _digest: _root_digest(damaged)) + + +def test_a_log_whose_root_is_null_is_refused(models, tmp_path): + """A null root is no more of a lineage than no root at all.""" + # It read as an empty log, so an append started a second root behind the + # damaged rows, which is the case this check exists to prevent. + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + + _rewrite_roots(analysis, lambda root: None) + + with pytest.raises(ValueError, match="does not say which root"): + analysis.simulate(4, append=True) + + +@pytest.mark.parametrize( + "damage", + [ + lambda root: {**root, "entropy": None}, + lambda root: {**root, "pool_size": 0}, + lambda root: {**root, "pool_size": 2}, + lambda root: {**root, "n_children_spawned": -1}, + lambda root: {**root, "spawn_key": [True]}, + lambda root: {key: value for key, value in root.items() if key != "entropy"}, + lambda root: {**root, "unexpected": 1}, + ], + ids=[ + "null entropy", + "no pool", + "pool numpy refuses", + "negative base", + "bool key", + "short", + "long", + ], +) +def test_a_root_that_no_stream_can_be_rebuilt_from_is_refused(models, tmp_path, damage): + """Rows agreeing on a root does not make it a root that can be resumed.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + + _rewrite_roots(analysis, damage) + + with pytest.raises(ValueError, match="rebuilt from"): + analysis.simulate(4, append=True) + + +def test_an_empty_log_is_not_a_log_that_cannot_be_checked(models, tmp_path): + """The control. Nothing recorded yet is a fresh start, not a refusal.""" + analysis = _study(tmp_path, "study", models) + + analysis.simulate(2, append=True, random_seed=42) + + assert sorted(_drawn(analysis)) == [0, 1] + + +def test_a_seed_given_as_a_sequence_is_recorded_as_one(models, tmp_path): + """A sequence is a documented seed, so the row has to carry it whole.""" + analysis = _study(tmp_path, "study", models) + + analysis.simulate(2, append=False, random_seed=[1, 2, 3]) + + with open(analysis.input_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + assert all(row[_SIMULATION_ROOT_KEY]["entropy"] == [1, 2, 3] for row in rows) + + +def test_a_sequence_seed_still_continues_the_same_study(models, tmp_path): + """And reads back, since a root that cannot be compared refuses the append.""" + whole = _study(tmp_path, "whole", models) + whole.simulate(4, append=False, random_seed=[1, 2, 3]) + + part = _study(tmp_path, "part", models) + part.simulate(2, append=False, random_seed=[1, 2, 3]) + part.simulate(4, append=True) + + assert _drawn(part) == _drawn(whole) + + +def test_blank_lines_between_rows_do_not_hide_the_root(models, tmp_path): + """A gap in the file is not a row, and not a study without a root either.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + with open(analysis.input_file, "r", encoding="utf-8") as written: + rows = written.read().splitlines() + with open(analysis.input_file, "w", encoding="utf-8") as spaced: + for row in rows: + spaced.write(row + "\n\n") + + analysis.simulate(4, append=True) + + assert sorted(_drawn(analysis)) == [0, 1, 2, 3] + + +def test_a_row_that_cannot_be_read_is_refused(models, tmp_path): + """A row that will not parse leaves nothing to check the root against.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + with open(analysis.input_file, "a", encoding="utf-8") as damaged: + damaged.write("{ this was cut off\n") + + with pytest.raises(ValueError, match="cannot be read"): + analysis.simulate(4, append=True) + + +def test_an_output_log_from_another_study_is_refused(models, tmp_path): + """Matching indices do not make two files the two halves of one run.""" + # Nothing in the indices tells them apart: both studies number their rows + # from zero, so the root is what says they belong together. + one = _study(tmp_path, "one", models) + one.simulate(2, append=False, random_seed=42) + other = _study(tmp_path, "other", models) + other.simulate(2, append=False, random_seed=7) + with open(other.output_file, "r", encoding="utf-8") as theirs: + stolen = theirs.read() + with open(one.output_file, "w", encoding="utf-8") as ours: + ours.write(stolen) + + with pytest.raises(ValueError, match="same root"): + one.simulate(4, append=True) + + +def test_a_checkpoint_whose_halves_disagree_is_refused(models, tmp_path): + """Where to carry on from is not established by one log alone.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + with open(analysis.output_file, "r", encoding="utf-8") as written: + rows = [line for line in written if line.strip()] + with open(analysis.output_file, "w", encoding="utf-8") as trimmed: + trimmed.write(rows[0]) + + with pytest.raises(ValueError, match="do not record the same simulations"): + analysis.simulate(4, append=True) + + +def test_a_collector_cannot_take_over_the_root(models, tmp_path): + """The root binds the two logs, so nothing outside the run writes it.""" + # Set past the constructor, so this pins the row and not the validation. + analysis = _study(tmp_path, "study", models) + analysis.data_collector = {_SIMULATION_ROOT_KEY: lambda _flight: "forged"} + + analysis.simulate(2, append=False, random_seed=42) + + with open(analysis.output_file, "r", encoding="utf-8") as written: + roots = { + json.dumps(json.loads(line)[_SIMULATION_ROOT_KEY], sort_keys=True) + for line in written + if line.strip() + } + assert roots != {'"forged"'} + assert len(roots) == 1 + + +def test_a_collector_named_after_the_root_is_refused(models, tmp_path): + """Overwriting it afterwards protects the log but discards the callback.""" + with pytest.raises(ValueError, match="run after the collectors|discarded"): + MonteCarlo( + filename=str(tmp_path / "study"), + environment=models[0], + rocket=models[1], + flight=models[2], + data_collector={_SIMULATION_ROOT_KEY: lambda _flight: 1}, + ) + + +def _rows_of(path): + with open(path, "r", encoding="utf-8") as written: + return [line for line in written if line.strip()] + + +def _rewrite_indices(path, numbers): + """Renumber a log's rows, leaving everything else in them alone.""" + rows = [json.loads(line) for line in _rows_of(path)] + for row, number in zip(rows, numbers): + row["index"] = number + with open(path, "w", encoding="utf-8") as rewritten: + for row in rows: + rewritten.write(json.dumps(row) + "\n") + + +def test_a_blank_line_in_the_output_log_does_not_move_the_continuation( + models, tmp_path +): + """Where to carry on from comes from the records, not the line count.""" + # A fresh object counts physical output lines when it opens the file, so + # one blank line made an append start at 3 and leave index 2 missing. + written = _study(tmp_path, "study", models) + written.simulate(2, append=False, random_seed=42) + with open(written.output_file, "a", encoding="utf-8") as padded: + padded.write("\n") + + resumed = _study(tmp_path, "study", models) + assert resumed.num_of_loaded_sims == 3 # the count this must not trust + resumed.simulate(4, append=True) + + assert sorted(_drawn(resumed)) == [0, 1, 2, 3] + + +def test_halves_that_number_the_same_count_differently_are_refused(models, tmp_path): + """Equal row counts do not make two logs the two halves of one run.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + _rewrite_indices(analysis.output_file, [0, 2]) + + with pytest.raises(ValueError, match="do not record the same simulations"): + analysis.simulate(4, append=True) + + +@pytest.mark.parametrize( + "numbers", [[0, 0], [0, 2], [1, 2]], ids=["duplicate", "hole", "no zero"] +) +def test_a_checkpoint_that_is_not_the_runs_first_simulations_is_refused( + models, tmp_path, numbers +): + """An append continues a run, so what it continues has to be its start.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + for path in (analysis.input_file, analysis.output_file): + _rewrite_indices(path, numbers) + + with pytest.raises(ValueError, match="not the run's first"): + analysis.simulate(4, append=True) + + +def test_the_order_a_parallel_run_finished_in_is_carried_on_from(models, tmp_path): + """Rows arrive in completion order, which is not sorted and not wrong.""" + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + for path in (analysis.input_file, analysis.output_file): + _rewrite_indices(path, [1, 0]) + + analysis.simulate(4, append=True) + + assert sorted(_drawn(analysis)) == [0, 1, 2, 3] + + +@pytest.mark.parametrize( + "number", [True, 1.0, "1", None, -1], ids=["bool", "float", "str", "null", "neg"] +) +def test_a_row_that_does_not_number_its_simulation_is_refused(models, tmp_path, number): + """Only a non-negative int says which simulation a row holds.""" + # True and 1.0 both equal 1, so either would pass for a record that is not + # there, and the count they contribute to decides where an append starts. + analysis = _study(tmp_path, "study", models) + analysis.simulate(2, append=False, random_seed=42) + for path in (analysis.input_file, analysis.output_file): + _rewrite_indices(path, [0, number]) + + with pytest.raises(ValueError, match="does not number"): + analysis.simulate(4, append=True) diff --git a/tests/unit/simulation/test_monte_carlo_seeding.py b/tests/unit/simulation/test_monte_carlo_seeding.py new file mode 100644 index 000000000..bd1f0cfa2 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_seeding.py @@ -0,0 +1,296 @@ +import json +import os +from time import time + +import multiprocess +import numpy as np +import pytest + +from rocketpy.stochastic import StochasticEnvironment +from rocketpy.simulation.monte_carlo import ( + MonteCarlo, + _root_seed_sequence, + _root_state_of, + _seed_of_simulation, + _seed_sequence_to_int, + _SimMonitor, +) + + +def _what_was_drawn(value): + """The same record without the parts that say which process wrote it. + + Two of the fields a serialized ``Function`` carries are properties of the + writer rather than of the draw. ``hash`` is the object's identity in that + process. A callable ``source`` is its pickle, and the same callable pickles + to different bytes under ``spawn``: measured on one drag curve, the parent + and a forked child agree and a spawned child does not, for a value the + fixture does not vary at all. Everything a run actually draws is numeric + and stays. + """ + if isinstance(value, dict): + return { + key: _what_was_drawn(item) + for key, item in value.items() + if key != "hash" and not (key == "source" and isinstance(item, str)) + } + if isinstance(value, list): + return [_what_was_drawn(item) for item in value] + return value + + +def _seed_for(index): + """The seed MonteCarlo gives one index, without running the others.""" + root = _root_state_of(_root_seed_sequence(42)) + return _seed_sequence_to_int(_seed_of_simulation(root, index)) + + +def _sampled_inputs(analysis): + with open(analysis.input_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + return {row["index"]: _what_was_drawn(row) for row in rows} + + +def _a_run(tmp_path, stem, models, *, parallel=False, workers=None, seed=None, count=4): + environment, rocket, flight = models + analysis = MonteCarlo( + filename=str(tmp_path / stem), + environment=environment, + rocket=rocket, + flight=flight, + ) + analysis.simulate( + number_of_simulations=count, + append=False, + parallel=parallel, + n_workers=workers, + random_seed=seed, + ) + return analysis + + +@pytest.fixture(name="models") +def _models(stochastic_environment, stochastic_calisto, stochastic_flight): + return stochastic_environment, stochastic_calisto, stochastic_flight + + +# --------------------------------------------------------------- the derivation + + +@pytest.mark.parametrize("seed", [42, None, [1, 2, 3], np.random.SeedSequence(7)]) +def test_a_simulation_gets_the_child_spawn_would_have_given_it(seed): + # The whole point of deriving one index directly: it has to be the same + # child, bit for bit, as spawning every index before it would produce. + root = _root_seed_sequence(seed) + state = _root_state_of(root) + spawned = np.random.SeedSequence(**root.state).spawn(6) + + for index, expected in enumerate(spawned): + assert np.array_equal( + expected.generate_state(4), + _seed_of_simulation(state, index).generate_state(4), + ) + + +def test_deriving_an_index_costs_nothing_for_the_ones_before_it(): + state = _root_state_of(_root_seed_sequence(42)) + + far = _seed_of_simulation(state, 1_000_000) + + assert far.spawn_key == (1_000_000,) + + +def test_two_indices_do_not_share_a_stream(): + state = _root_state_of(_root_seed_sequence(42)) + + first = _seed_of_simulation(state, 0).generate_state(4) + second = _seed_of_simulation(state, 1).generate_state(4) + + assert not np.array_equal(first, second) + + +# ------------------------------------------------------------------- the root + + +def test_a_caller_seed_sequence_is_not_consumed(): + given = np.random.SeedSequence(42) + + _root_seed_sequence(given).spawn(5) + + assert given.n_children_spawned == 0 + + +def test_a_caller_cannot_move_the_run_by_editing_the_list_it_passed(): + # SeedSequence keeps a sequence entropy by reference, so without a copy of + # its own the run would follow whatever the caller did to that list next. + given = [1, 2, 3] + state = _root_state_of(_root_seed_sequence(given)) + + before = _seed_of_simulation(state, 0).generate_state(4) + given[0] = 999 + + assert np.array_equal(_seed_of_simulation(state, 0).generate_state(4), before) + + +@pytest.mark.parametrize("given", [np.random.default_rng(42), np.random.PCG64(42)]) +def test_a_generator_is_refused_rather_than_read(given): + # Using a consume-on-use object as an immutable seed cannot mean what it + # says, so it is refused instead of quietly meaning something else. + with pytest.raises(TypeError, match="random_seed must be"): + _root_seed_sequence(given) + + +def test_an_unusable_seed_costs_the_previous_run_nothing(models, tmp_path): + analysis = _a_run(tmp_path, "study", models, seed=42, count=2) + kept = _sampled_inputs(analysis) + + with pytest.raises(TypeError): + analysis.simulate(2, append=False, random_seed=np.random.default_rng(1)) + + assert _sampled_inputs(analysis) == kept + + +# ------------------------------------------------------------- what a run gives + + +def test_one_seed_gives_one_set_of_inputs(models, tmp_path): + first = _a_run(tmp_path, "first", models, seed=42) + again = _a_run(tmp_path, "again", models, seed=42) + + assert _sampled_inputs(first) == _sampled_inputs(again) + + +def test_another_seed_gives_another_set(models, tmp_path): + # The control. Without this the test above passes on a run that ignores + # the seed entirely. + first = _a_run(tmp_path, "first", models, seed=42) + other = _a_run(tmp_path, "other", models, seed=7) + + assert _sampled_inputs(first) != _sampled_inputs(other) + + +@pytest.mark.skipif(os.cpu_count() < 4, reason="needs four workers to be four") +def test_an_index_gets_the_same_inputs_however_the_run_was_split(models, tmp_path): + # This is the guarantee. Splitting the work differently must not change + # what any one simulation drew. + serial = _a_run(tmp_path, "serial", models, seed=42) + two = _a_run(tmp_path, "two", models, parallel=True, workers=2, seed=42) + four = _a_run(tmp_path, "four", models, parallel=True, workers=4, seed=42) + + assert _sampled_inputs(serial) == _sampled_inputs(two) + assert _sampled_inputs(serial) == _sampled_inputs(four) + + +def test_an_index_drawn_last_draws_what_it_draws_alone(example_spaceport_env): + """A worker reaches an index without running the ones before it. + + The split test above cannot see this: every input in its fixture is given + an explicit nominal, and only one read off the wrapped object can move. + """ + # A bare standard deviation, so the nominal is read from the environment, + # which is also where create_object writes the draw back. Both are built + # before anything is drawn, the way every worker starts from one model. + walked = StochasticEnvironment(example_spaceport_env, elevation=10) + alone = StochasticEnvironment(example_spaceport_env, elevation=10) + + for index in range(4): + walked._set_stochastic(_seed_for(index)) + walked.create_object() + reached_last = walked.last_rnd_dict["elevation"] + + alone._set_stochastic(_seed_for(3)) + alone.create_object() + + assert reached_last == alone.last_rnd_dict["elevation"] + + +def test_the_indices_of_that_run_are_not_all_one_value(example_spaceport_env): + # The control. Without it the test above holds on a model that draws a + # constant, which is what an elevation of nominal zero would give. + walked = StochasticEnvironment(example_spaceport_env, elevation=10) + drawn = [] + for index in range(4): + walked._set_stochastic(_seed_for(index)) + walked.create_object() + drawn.append(walked.last_rnd_dict["elevation"]) + + assert len(set(drawn)) == len(drawn) + + +def test_an_appended_run_carries_the_same_stream_on(models, tmp_path): + whole = _a_run(tmp_path, "whole", models, seed=42, count=4) + + part = _a_run(tmp_path, "part", models, seed=42, count=2) + part.simulate(4, append=True, random_seed=42) + + assert _sampled_inputs(part) == _sampled_inputs(whole) + + +@pytest.fixture(name="spawned_workers") +def _spawned_workers(): + """Start workers the way Windows does, wherever the test happens to run.""" + was = multiprocess.get_start_method() + multiprocess.set_start_method("spawn", force=True) + yield + multiprocess.set_start_method(was, force=True) + + +@pytest.mark.usefixtures("spawned_workers") +@pytest.mark.skipif(os.cpu_count() < 4, reason="needs four workers to be four") +def test_an_index_keeps_its_inputs_when_the_workers_are_spawned(models, tmp_path): + """The same guarantee on the start method Windows uses. + + A spawned child rebuilds the models by unpickling rather than inheriting + them, so this is a different question from the one above and was worth + asking separately: the first version of these tests compared serialized + callables, which differ between processes for a value nothing varies. + """ + serial = _a_run(tmp_path, "serial", models, seed=42) + two = _a_run(tmp_path, "two", models, parallel=True, workers=2, seed=42) + four = _a_run(tmp_path, "four", models, parallel=True, workers=4, seed=42) + + assert _sampled_inputs(serial) == _sampled_inputs(two) + assert _sampled_inputs(serial) == _sampled_inputs(four) + + +class _ALockNobodyContends: + """The manager's mutex, with only this process to hand it to.""" + + def acquire(self): + pass + + def release(self): + pass + + +class _AnEventNobodySet: + """The failure event, which nothing in a clean run reaches for.""" + + def is_set(self): + return False + + def set(self): + raise AssertionError("the producer reported a failure") + + +def test_a_worker_loop_draws_the_study_serial_draws(models, tmp_path): + """The loop a worker runs, driven here, produces the run serial produces.""" + # Every other check of this starts a child process, where the same code + # runs and nothing in the test can watch it. + serial = _a_run(tmp_path, "serial", models, seed=42, count=4) + environment, rocket, flight = models + worker = MonteCarlo( + filename=str(tmp_path / "worker"), + environment=environment, + rocket=rocket, + flight=flight, + ) + worker.simulate(number_of_simulations=0, append=False, random_seed=42) + + worker._MonteCarlo__sim_producer( + _SimMonitor(initial_count=0, n_simulations=4, start_time=time()), + _ALockNobodyContends(), + _AnEventNobodySet(), + ) + + assert _sampled_inputs(worker) == _sampled_inputs(serial) diff --git a/tests/unit/simulation/test_monte_carlo_simulation_index.py b/tests/unit/simulation/test_monte_carlo_simulation_index.py new file mode 100644 index 000000000..4ac75b67f --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_simulation_index.py @@ -0,0 +1,170 @@ +import json +import threading +from time import time + +import pytest + +from rocketpy.simulation.monte_carlo import MonteCarlo, _SimMonitor + + +def _indices(analysis): + with open(analysis.output_file, "r", encoding="utf-8") as written: + return sorted(json.loads(line)["index"] for line in written if line.strip()) + + +def _a_study(tmp_path, stem, environment, rocket, flight): + return MonteCarlo( + filename=str(tmp_path / stem), + environment=environment, + rocket=rocket, + flight=flight, + ) + + +@pytest.mark.parametrize("parallel", [False, True]) +def test_a_run_numbers_its_simulations_from_zero( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path, parallel +): + # Serial used to write 1, 2, 3 while parallel wrote 0, 1, 2, so the same + # simulation had two names depending on how the run was started. + analysis = _a_study( + tmp_path, + f"study-{parallel}", + stochastic_environment, + stochastic_calisto, + stochastic_flight, + ) + + analysis.simulate( + number_of_simulations=3, + append=False, + parallel=parallel, + n_workers=2 if parallel else None, + ) + + assert _indices(analysis) == [0, 1, 2] + + +def test_both_modes_agree_on_what_a_simulation_is_called( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + one = _a_study( + tmp_path, + "serial", + stochastic_environment, + stochastic_calisto, + stochastic_flight, + ) + other = _a_study( + tmp_path, "para", stochastic_environment, stochastic_calisto, stochastic_flight + ) + + one.simulate(number_of_simulations=3, append=False, parallel=False) + other.simulate(number_of_simulations=3, append=False, parallel=True, n_workers=2) + + assert _indices(one) == _indices(other) + + +def test_an_appended_run_carries_on_from_the_last_index( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + analysis = _a_study( + tmp_path, "study", stochastic_environment, stochastic_calisto, stochastic_flight + ) + + analysis.simulate(number_of_simulations=2, append=False) + analysis.simulate(number_of_simulations=4, append=True) + + assert _indices(analysis) == [0, 1, 2, 3] + + +def test_a_serial_failure_names_the_simulation_the_way_parallel_would( + stochastic_environment, + stochastic_calisto, + stochastic_flight, + tmp_path, + monkeypatch, + capsys, +): + """A run that fails reports the index the other mode would have used.""" + # The message is the only place the numbering reaches whoever ran it, so + # it can disagree with the logs without anything else noticing. + analysis = _a_study( + tmp_path, + "failing", + stochastic_environment, + stochastic_calisto, + stochastic_flight, + ) + attempts = [] + ran = MonteCarlo._MonteCarlo__run_single_simulation + + def fails_on_the_second(self): + attempts.append(None) + if len(attempts) == 2: + raise RuntimeError("the flight would not run") + return ran(self) + + monkeypatch.setattr( + MonteCarlo, "_MonteCarlo__run_single_simulation", fails_on_the_second + ) + + with pytest.raises(RuntimeError, match="the flight would not run"): + analysis.simulate(number_of_simulations=3, append=False) + + assert "Error on iteration 1:" in capsys.readouterr().out + + +def test_claim_next_index_hands_out_each_index_once(): + """Claimers racing each other get range(n) between them and nothing past it.""" + n_simulations, n_claimers = 200, 8 + monitor = _SimMonitor(0, n_simulations, time()) + ready = threading.Barrier(n_claimers) + guard = threading.Lock() + claimed = [] + + def claim_until_empty(): + ready.wait() + mine = [] + while (index := monitor.claim_next_index()) is not None: + mine.append(index) + with guard: + claimed.extend(mine) + + workers = [threading.Thread(target=claim_until_empty) for _ in range(n_claimers)] + for worker in workers: + worker.start() + for worker in workers: + worker.join() + + assert len(claimed) == n_simulations + assert len(set(claimed)) == n_simulations + assert sorted(claimed) == list(range(n_simulations)) + assert max(claimed) < n_simulations + + +def test_claim_next_index_is_empty_once_the_target_is_reached(): + """The control: a checkpoint that already holds the target claims nothing.""" + monitor = _SimMonitor(3, 3, time()) + + assert monitor.claim_next_index() is None + assert monitor.count == 3 + + +def test_a_collector_cannot_take_over_the_simulation_index( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + """The index names the seed a row was drawn with, so nothing else sets it.""" + # Set past the constructor, so this pins the row and not the validation. + analysis = _a_study( + tmp_path, "study", stochastic_environment, stochastic_calisto, stochastic_flight + ) + analysis.export_list = [] + analysis._export_config = {} + # The row also carries which root drew it, which simulate() would have set. + analysis._MonteCarlo__root_state = (42, (), 4, 0) + analysis.data_collector = {"index": lambda _flight: 999} + + row = json.loads(analysis._MonteCarlo__evaluate_flight_outputs(None, 7)) + + assert row["index"] == 7 diff --git a/tests/unit/simulation/test_monte_carlo_worker_reporting.py b/tests/unit/simulation/test_monte_carlo_worker_reporting.py index 9c5b9c896..696823a86 100644 --- a/tests/unit/simulation/test_monte_carlo_worker_reporting.py +++ b/tests/unit/simulation/test_monte_carlo_worker_reporting.py @@ -86,27 +86,42 @@ def _a_worker(tmp_path, model, event=None): rocket=model, flight=model, ) + # simulate() settles the root before any worker starts, and the producer + # derives each index's seed from it, so a worker driven directly needs one. + study._MonteCarlo__root_state = mc_module._root_state_of( + mc_module._root_seed_sequence(42) + ) return study, event or _ErrorEvent() +def _claims(*indices): + """A monitor that hands out these indices and then says there are no more.""" + handing = iter(indices) + return SimpleNamespace( + claim_next_index=lambda: next(handing, None), + print_update_status=lambda: None, + ) + + 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) + study._MonteCarlo__sim_producer(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) +def test_a_worker_that_cannot_reseed_names_the_index_it_was_seeding(tmp_path, capsys): + """Seeding is per index, so a reseed failure belongs to that index.""" + # It used to be per worker and above the loop, where the only name for it + # was worker startup. The claim comes first now, so there is one to give. study, error_event = _a_worker(tmp_path, _refusing_model()) - _run(study, monitor, error_event) + _run(study, _claims(3), error_event) assert error_event.was_set reported = capsys.readouterr().out - assert "worker startup" in reported + assert "iteration 3" in reported assert "the models would not reseed" in reported @@ -117,7 +132,7 @@ 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) + monitor = SimpleNamespace(claim_next_index=refuse) study, error_event = _a_worker(tmp_path, model) _run(study, monitor, error_event) @@ -140,10 +155,9 @@ def refuse(_self): 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) + _run(study, _claims(7), error_event) assert error_event.was_set assert "iteration 7" in capsys.readouterr().out @@ -151,10 +165,15 @@ def refuse(_self): 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()) + def refuse(): + raise RuntimeError("the monitor would not hand out an index") + + monitor = SimpleNamespace(claim_next_index=refuse) + model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None) + study, error_event = _a_worker(tmp_path, model) _run(study, monitor, error_event) @@ -163,10 +182,10 @@ def test_a_startup_failure_is_written_down_and_not_only_printed(tmp_path): 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"] + assert "would not hand out an index" in rows[0]["error"] -@pytest.mark.parametrize("failing", ["_set_stochastic", "increment"]) +@pytest.mark.parametrize("failing", ["_set_stochastic", "claim_next_index"]) def test_a_worker_failure_never_raises_out_of_the_producer(tmp_path, failing): """A reported failure leaves the producer without an exception.""" @@ -180,8 +199,7 @@ def refuse(*_args): _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), + claim_next_index=refuse if failing == "claim_next_index" else (lambda: 0), ) study, error_event = _a_worker(tmp_path, model) @@ -205,7 +223,7 @@ def test_reporting_a_failure_never_keeps_the_mutex(tmp_path, monkeypatch, breaki mc_module._SimMonitor, "reprint", _raise_instead("no stdout") ) event = _ErrorEvent(refuse=breaking == "event") - monitor = SimpleNamespace(keep_simulating=lambda: True) + monitor = _claims(0) study, error_event = _a_worker(tmp_path, _refusing_model(), event) mutex = _Mutex() @@ -224,7 +242,7 @@ def test_a_reporting_failure_does_not_replace_the_simulation_failure( ): """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) + monitor = _claims(0) study, error_event = _a_worker(tmp_path, _refusing_model()) _run(study, monitor, error_event) @@ -254,15 +272,14 @@ def _committing_producer(monkeypatch): def _one_then_broken(): calls = {"count": 0} - def keep_simulating(): + def claim_next_index(): calls["count"] += 1 if calls["count"] == 1: - return True + return 0 raise RuntimeError("the monitor died between simulations") return SimpleNamespace( - keep_simulating=keep_simulating, - increment=lambda: 1, + claim_next_index=claim_next_index, print_update_status=lambda: None, ) @@ -307,7 +324,7 @@ def test_a_worker_that_cannot_announce_its_failure_does_not_exit_cleanly(tmp_pat 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) + _run(study, _claims(0), error_event) def test_a_worker_that_did_announce_its_failure_returns(tmp_path): @@ -316,7 +333,7 @@ def test_a_worker_that_did_announce_its_failure_returns(tmp_path): # 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) + _run(study, _claims(0), error_event) assert error_event.was_set @@ -330,7 +347,7 @@ def test_a_lock_the_reporter_cannot_take_does_not_stop_it( """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) + monitor = _claims(0) study, error_event = _a_worker(tmp_path, _refusing_model()) mutex = mutex_class() @@ -344,7 +361,7 @@ def test_a_lock_the_reporter_cannot_take_does_not_stop_it( 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) + monitor = _claims(0) study, error_event = _a_worker(tmp_path, _refusing_model()) _run(study, monitor, error_event, _MutexThatBreaksOnRelease()) @@ -363,7 +380,7 @@ def test_a_failure_after_the_inputs_were_drawn_still_records_why(tmp_path, monke "_MonteCarlo__evaluate_flight_outputs", _raise_instead("the outputs would not serialize"), ) - monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 1) + monitor = _claims(0) study, error_event = _a_worker( tmp_path, SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _s: None) ) @@ -393,7 +410,7 @@ def test_an_input_row_that_is_not_an_object_does_not_break_the_reporter( "_MonteCarlo__evaluate_flight_outputs", _raise_instead("the outputs would not serialize"), ) - monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 1) + monitor = _claims(0) study, error_event = _a_worker( tmp_path, SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _s: None) )