diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ac64cc..fd1f825 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `RLTaskEnv.step` publishes the *terminal* observation in `info["observations"]["raw"]["obs"]` instead of the episode's first one (off-policy truncation bootstraps in clean_rl SAC/TD3 and FastTD3 read it). - IsaacGym and PyBullet reported `joint_pos_target` in native DoF order while `joint_pos` is in sorted-name order; both now use `get_joint_names(sort=True)` (completes #12). +- `ParallelSimWrapper`: a worker that died during handler construction or `launch` surfaced as a bare `EOFError`/`ConnectionResetError` from the handshake and left the other workers running; the handshake now raises the worker's own traceback, `close()` tolerates dead workers, and a failed constructor tears the pool down. - MuJoCo: `` is reserved by default; humanoid + mesh scenes no longer die with `mj_stackAlloc: out of memory` (get_started/10_mount_camera.py). - `hf_util`: a symlinked `roboverse_data` is no longer refused as path traversal; concurrent diff --git a/metasim/sim/parallel.py b/metasim/sim/parallel.py index 6be950f..c3fd4cb 100644 --- a/metasim/sim/parallel.py +++ b/metasim/sim/parallel.py @@ -5,6 +5,7 @@ import multiprocessing as mp import platform import sys +import time import traceback from copy import deepcopy from functools import partial @@ -151,11 +152,15 @@ def __init__(self, scenario: ScenarioCfg, extra_spec: dict[str, BaseQueryType] | self.processes.append(process) work_remote.close() - # To make sure environments are initialized in all workers - for remote in self.remotes: - remote.send(("handshake", (None,))) - for remote in self.remotes: - remote.recv() + # Every worker must reach its command loop before the wrapper is usable. A worker whose + # handler construction raised has already exited, so the handshake reads through + # ``_recv_or_surface`` and raises its real traceback (from ``error_queue``) instead of + # a bare ``EOFError``; the surviving workers are torn down so nothing is left behind. + try: + self._handshake() + except BaseException: + self._terminate_workers() + raise def _check_error(self): """Drain the error queue and detect dead workers. @@ -195,6 +200,7 @@ def _recv_or_surface(self, remote_idx: int): return self.remotes[remote_idx].recv() except (EOFError, ConnectionResetError, BrokenPipeError) as err: # Worker died — _check_error will raise with the real message. + self._wait_for_error_report() self._check_error() raise RuntimeError( f"Parallel worker {remote_idx} closed its pipe without reporting an error. " @@ -203,23 +209,58 @@ def _recv_or_surface(self, remote_idx: int): def launch(self): for remote in self.remotes: - remote.send(("launch", (None,))) + self._send(remote, ("launch", (None,))) self.waiting = False - # ``launch`` sends no reply; handshake round-trips so every - # worker has finished launching before the wrapper returns. - for remote in self.remotes: - remote.send(("handshake", (None,))) - for remote in self.remotes: - remote.recv() - self._check_error() + # ``launch`` sends no reply; the handshake round-trips so every worker has finished + # launching before the wrapper returns, and surfaces a worker that died in ``launch``. + self._handshake() def close(self): if self.closed: return for remote in self.remotes: - remote.send(("close", (None,))) + self._send(remote, ("close", (None,))) + self._terminate_workers() + + @staticmethod + def _send(remote: Connection, msg) -> None: + """``send`` that tolerates a dead worker: the following ``recv`` / ``_check_error`` + reports the worker's real error, which a ``BrokenPipeError`` here would only hide.""" + try: + remote.send(msg) + except (BrokenPipeError, ConnectionResetError, EOFError, OSError): + pass + + def _handshake(self) -> None: + """Round-trip every worker; raises the worker's own error if one has died.""" + for remote in self.remotes: + self._send(remote, ("handshake", (None,))) + for idx in range(len(self.remotes)): + self._recv_or_surface(idx) + self._check_error() + + def _wait_for_error_report(self, timeout: float = 2.0) -> None: + """Give a dying worker's queue feeder thread time to deliver its traceback. + + ``mp.Queue.put`` hands the item to a background thread; the pipe end can close before + the report is readable on this side, which would turn a real traceback into + "died without reporting an error". + """ + deadline = time.monotonic() + timeout + while self.error_queue.empty() and time.monotonic() < deadline: + if any(p.is_alive() for p in self.processes): + time.sleep(0.01) + else: + break + + def _terminate_workers(self, join_timeout: float = 10.0) -> None: + """Join every worker, force-terminating any that does not exit in time.""" + for process in self.processes: + process.join(timeout=join_timeout) for process in self.processes: - process.join() + if process.is_alive(): + process.terminate() + process.join(timeout=join_timeout) self.closed = True def _set_states(self, states: list[DictEnvState], env_ids: list[int] | None = None) -> None: diff --git a/metasim/test/test_parallel_error_handling_general.py b/metasim/test/test_parallel_error_handling_general.py index bcfea51..7265105 100644 --- a/metasim/test/test_parallel_error_handling_general.py +++ b/metasim/test/test_parallel_error_handling_general.py @@ -100,6 +100,12 @@ def __init__(self, alive: bool = True, exitcode: int | None = None): def is_alive(self) -> bool: return self._alive + def join(self, timeout: float | None = None) -> None: + pass + + def terminate(self) -> None: + self._alive = False + def _make_parallel_stub_handler(remotes, processes, error_queue=None) -> Any: """Build a ``ParallelHandler`` instance that bypasses ``__init__`` so @@ -217,3 +223,53 @@ def test_set_seed_forwards_to_every_worker(): cmd, payload = remote.sent[-1] assert cmd == "set_seed", f"worker {i} got {cmd!r}, expected 'set_seed'" assert payload == (123,), f"worker {i} got payload {payload!r}, expected (123,)" + + +@pytest.mark.general +def test_launch_surfaces_worker_error_instead_of_eof(): + """A worker that dies inside ``launch`` used to surface as a bare ``EOFError`` from the + handshake ``recv``; the real traceback in ``error_queue`` must be what the caller sees.""" + err_queue = _SyncQueue() + err_queue.put(("FileNotFoundError", "missing.usd", ["worker traceback\n"])) + handler = _make_parallel_stub_handler( + remotes=[_FakeRemote(recv_raises=EOFError)], + processes=[_FakeProcess(alive=False, exitcode=1)], + error_queue=err_queue, + ) + with pytest.raises(RuntimeError, match=r"Parallel worker error \(FileNotFoundError\): missing\.usd"): + handler.launch() + + +@pytest.mark.general +def test_close_tolerates_dead_worker(): + """``close`` must not raise ``BrokenPipeError`` on a worker that already died.""" + + class _DeadRemote(_FakeRemote): + def send(self, msg): + raise BrokenPipeError + + handler = _make_parallel_stub_handler( + remotes=[_DeadRemote()], + processes=[_FakeProcess(alive=False, exitcode=1)], + ) + handler.close() + assert handler.closed + + +class _BoomHandler(_StubBaseHandler): + """Handler whose construction fails, as an asset-loading error inside a worker would.""" + + def __init__(self, *_a, **_k): + raise RuntimeError("asset missing: boom.urdf") + + +@pytest.mark.general +def test_constructor_reports_worker_construction_error(): + """End-to-end with real worker processes: a handler that raises in ``__init__`` used to + surface as ``EOFError`` from the constructor handshake, with the traceback lost.""" + from metasim.scenario.scenario import ScenarioCfg + + scenario = ScenarioCfg(num_envs=2) + with pytest.raises(RuntimeError, match=r"asset missing: boom\.urdf") as excinfo: + ParallelSimWrapper(_BoomHandler)(scenario) + assert "EOFError" not in str(excinfo.value)