Skip to content

Commit fa81b7f

Browse files
committed
BUG: notice a worker that never finished
The workers say they failed by setting an event, and the parent joins them and reads it. A worker that is killed runs no handler, so the event stays clear, the join returns because the process is gone, and the run reports the simulations it never wrote as done. Measured with a SIGKILL in the second simulation of six, two workers: simulate() returned normally with two rows on disk. Its exit code is what is left of a worker that ends this way, so the parent reads that too. Anything other than zero is refused, None included, since that is a worker that has not finished at all. Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
1 parent 318bf46 commit fa81b7f

2 files changed

Lines changed: 93 additions & 0 deletions

File tree

‎rocketpy/simulation/monte_carlo.py‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -488,6 +488,11 @@ def __run_in_parallel(self, n_workers=None):
488488
for sim_producer in processes:
489489
sim_producer.join()
490490

491+
# Before the event: a worker that was killed, or that died
492+
# before its own handler could set it, leaves it clear, and the
493+
# run would report the simulations it never wrote as done.
494+
_refuse_a_worker_that_did_not_finish(processes)
495+
491496
# Handle error from the child processes
492497
if simulation_error_event.is_set():
493498
raise RuntimeError(
@@ -1758,6 +1763,27 @@ def export_errors_to_json(self, filename):
17581763
self._write_log_to_json(self.errors_log, filename)
17591764

17601765

1766+
def _refuse_a_worker_that_did_not_finish(processes):
1767+
"""Raise if any worker left without exiting cleanly.
1768+
1769+
The workers report their own failures through an event, which one that was
1770+
killed never reaches, so what is left of it is its exit code. A negative
1771+
one is the signal that ended it, and ``None`` is one still running.
1772+
"""
1773+
unfinished = [
1774+
f"worker {position} with exit code {process.exitcode}"
1775+
for position, process in enumerate(processes)
1776+
if process.exitcode != 0
1777+
]
1778+
if not unfinished:
1779+
return
1780+
raise RuntimeError(
1781+
f"The run is incomplete: {', '.join(unfinished)}. A worker that ends "
1782+
"this way records nothing and cannot say why, so the simulations it "
1783+
"held are missing from the results."
1784+
)
1785+
1786+
17611787
def _import_multiprocess():
17621788
"""Import the necessary modules and submodules for the
17631789
multiprocess library.
Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
import os
2+
import signal
3+
from types import SimpleNamespace
4+
5+
import pytest
6+
7+
from rocketpy.simulation import monte_carlo as mc_module
8+
from rocketpy.simulation.monte_carlo import _refuse_a_worker_that_did_not_finish
9+
10+
11+
def _worker(exitcode):
12+
return SimpleNamespace(exitcode=exitcode)
13+
14+
15+
def test_workers_that_all_exited_cleanly_are_accepted():
16+
_refuse_a_worker_that_did_not_finish([_worker(0), _worker(0)])
17+
18+
19+
def test_a_worker_killed_by_a_signal_is_refused():
20+
with pytest.raises(RuntimeError, match=r"worker 1 with exit code -9"):
21+
_refuse_a_worker_that_did_not_finish([_worker(0), _worker(-9)])
22+
23+
24+
def test_a_worker_that_exited_nonzero_is_refused():
25+
with pytest.raises(RuntimeError, match=r"worker 0 with exit code 1"):
26+
_refuse_a_worker_that_did_not_finish([_worker(1), _worker(0)])
27+
28+
29+
def test_every_unfinished_worker_is_named():
30+
with pytest.raises(RuntimeError) as raised:
31+
_refuse_a_worker_that_did_not_finish([_worker(-9), _worker(0), _worker(3)])
32+
33+
assert "worker 0" in str(raised.value)
34+
assert "worker 2" in str(raised.value)
35+
assert "worker 1" not in str(raised.value)
36+
37+
38+
@pytest.mark.parametrize("exitcode", [None, -15, 2])
39+
def test_anything_but_a_clean_exit_is_refused(exitcode):
40+
with pytest.raises(RuntimeError):
41+
_refuse_a_worker_that_did_not_finish([_worker(exitcode), _worker(0)])
42+
43+
44+
_REAL_SIMULATION = mc_module.MonteCarlo._MonteCarlo__run_single_simulation
45+
46+
47+
def test_a_killed_worker_does_not_pass_as_a_finished_run(monte_carlo_calisto, tmp_path):
48+
# The event the workers report through is set by their own handler, and a
49+
# signal runs no handler, so the run used to return as though it had done
50+
# every simulation it was asked for.
51+
calls = {"count": 0}
52+
53+
def die_on_the_second(self):
54+
calls["count"] += 1
55+
if calls["count"] == 2:
56+
os.kill(os.getpid(), signal.SIGKILL)
57+
return _REAL_SIMULATION(self)
58+
59+
mc_module.MonteCarlo._MonteCarlo__run_single_simulation = die_on_the_second
60+
monte_carlo_calisto.filename = os.path.join(str(tmp_path), "study")
61+
try:
62+
with pytest.raises(RuntimeError, match="incomplete"):
63+
monte_carlo_calisto.simulate(
64+
number_of_simulations=6, append=False, parallel=True, n_workers=2
65+
)
66+
finally:
67+
mc_module.MonteCarlo._MonteCarlo__run_single_simulation = _REAL_SIMULATION

0 commit comments

Comments
 (0)