diff --git a/openpilot/selfdrive/modeld/modeld.py b/openpilot/selfdrive/modeld/modeld.py index 91e25f547f..9e20539fc0 100755 --- a/openpilot/selfdrive/modeld/modeld.py +++ b/openpilot/selfdrive/modeld/modeld.py @@ -43,6 +43,7 @@ from openpilot.starpilot.lateral.lane_change_status_wire import ( from openpilot.starpilot.models.receipt import ModelReceiptOwner from openpilot.starpilot.models.status import ModelVariant from openpilot.starpilot.models.startup import report_load_timeout, wait_for_chestnut_power +from openpilot.starpilot.models.recovery import STARTUP_FALLBACK_EXIT from openpilot.starpilot.navigation.intent import TurnIntent, matching_turn_signal from openpilot.starpilot.models.catalog import BUNDLED_CURRENT, BY_ID, DEFAULT_SMALL, DEFAULT_SMALL_SHA256 from openpilot.starpilot.models.runner import CatalogModelState, load_verified_model @@ -379,7 +380,10 @@ def main(demo=False): model = big_model params.put_bool("ChestnutActive", model is not None) if model is None: - drop_chestnut() + # A timed-out loader can still mutate tinygrad's shared device state. + # The manager must confirm this process has exited before starting Small. + cloudlog.error("Chestnut startup failed; requesting isolated Small restart") + raise SystemExit(STARTUP_FALLBACK_EXIT) if small_model is None: small_model, small_prepared, selected_small_failed = load_small() diff --git a/openpilot/starpilot/models/recovery.py b/openpilot/starpilot/models/recovery.py index a8a6dec64e..bed174e770 100644 --- a/openpilot/starpilot/models/recovery.py +++ b/openpilot/starpilot/models/recovery.py @@ -11,6 +11,7 @@ STALL_NS = 2_000_000_000 CAMERA_MAX_AGE_NS = 200_000_000 INT_GRACE_NS = 500_000_000 KILL_GRACE_NS = 1_000_000_000 +STARTUP_FALLBACK_EXIT = 75 @dataclass(frozen=True) diff --git a/openpilot/starpilot/models/recovery_process.py b/openpilot/starpilot/models/recovery_process.py index 7514a71d9e..cecab75386 100644 --- a/openpilot/starpilot/models/recovery_process.py +++ b/openpilot/starpilot/models/recovery_process.py @@ -6,7 +6,7 @@ from multiprocessing import Process from openpilot.common.swaglog import cloudlog from openpilot.starpilot.models.receipt import process_start_ticks, read_receipt -from openpilot.starpilot.models.recovery import KILL_GRACE_NS, ModelRecoveryOwner, RecoveryEvidence +from openpilot.starpilot.models.recovery import KILL_GRACE_NS, STARTUP_FALLBACK_EXIT, ModelRecoveryOwner, RecoveryEvidence from openpilot.starpilot.models.status import ModelVariant from openpilot.system.manager.process import PythonProcess, join_process, launcher @@ -27,6 +27,7 @@ class ModeldProcess(PythonProcess): super().__init__(name, module, should_run, enabled=enabled) self.recovery = ModelRecoveryOwner() self.context_ready = False + self.launch_drive_id = 0 def observe(self, sm, started): now = time.monotonic_ns() @@ -50,6 +51,18 @@ class ModeldProcess(PythonProcess): device_fresh = (sm.seen['deviceState'] and sm.valid['deviceState'] and sm.alive['deviceState'] and 0 < int(sm.logMonoTime['deviceState']) <= now <= int(sm.logMonoTime['deviceState']) + 1_000_000_000) self.context_ready = bool(device_fresh and started) + if (self.context_ready and self.proc is not None and not alive and not self.shutting_down and + self.proc.exitcode == STARTUP_FALLBACK_EXIT and self.recovery.phase == 'monitor' and + not self.recovery.small_only and 0 < self.launch_drive_id == evidence.drive_id): + # Only our explicitly failed startup may request this one fresh process. + # Never reuse the old GPU context or retry an arbitrary Small crash. + self.proc.join(timeout=0) + self.proc = None + self.recovery.drive_id = evidence.drive_id + self.recovery.small_only = True + self.recovery.phase = 'small' + cloudlog.event('modeld.recovery', action='small-after-startup-exit', pid=pid, drive=evidence.drive_id) + return if not device_fresh and self.recovery.phase == 'monitor': self.recovery._withdraw() return @@ -106,5 +119,6 @@ class ModeldProcess(PythonProcess): return self.proc = Process(name=self.name, target=modeld_launcher, args=(self.module, self.name, self.recovery.small_only)) + self.launch_drive_id = self.recovery.drive_id self.proc.start() self.shutting_down = False diff --git a/openpilot/starpilot/models/tests/test_modeld_startup_fallback.py b/openpilot/starpilot/models/tests/test_modeld_startup_fallback.py index 25195305f0..eb49204c03 100644 --- a/openpilot/starpilot/models/tests/test_modeld_startup_fallback.py +++ b/openpilot/starpilot/models/tests/test_modeld_startup_fallback.py @@ -104,6 +104,8 @@ def test_remote_inference_failure_does_not_change_chestnut_status(): def execute(statements, env): env.setdefault("os", SimpleNamespace(getenv=lambda key: None)) env.setdefault("recovery_small_only", False) + from openpilot.starpilot.models.recovery import STARTUP_FALLBACK_EXIT + env.setdefault("STARTUP_FALLBACK_EXIT", STARTUP_FALLBACK_EXIT) exec(compile(ast.fix_missing_locations(ast.Module(body=statements, type_ignores=[])), str(MODEL_SOURCE), "exec"), env) @@ -148,9 +150,9 @@ def test_observed_availability_and_big_eligibility(monkeypatch, present, cable, params.put_bool.assert_called_once_with("ChestnutLoading", expected) -@pytest.mark.parametrize("timeout", [False, True]) +@pytest.mark.parametrize("timeout,worker_alive", [(False, False), (True, False), (True, True)]) @pytest.mark.parametrize("custom", [False, True]) -def test_big_worker_waits_before_load_and_timeout_preserves_small_fallback(custom, timeout): +def test_big_worker_waits_before_load_and_failure_requests_fresh_small_process(custom, timeout, worker_alive): body = main_body() start = next(i for i, node in enumerate(body) if isinstance(node, ast.Assign) and ast.unparse(node.targets[0]) == "st") end = next(i for i, node in enumerate(body) if ast.unparse(node).startswith("config_realtime_process(")) @@ -172,7 +174,7 @@ def test_big_worker_waits_before_load_and_timeout_preserves_small_fallback(custo events.append(("join", duration)) def is_alive(self): - return False + return worker_alive def wait(): assert worker @@ -201,6 +203,7 @@ def test_big_worker_waits_before_load_and_timeout_preserves_small_fallback(custo "CHESTNUT": True, "vipc_client_main": SimpleNamespace(width=1928, height=1208), "wait_for_chestnut": wait, "ModelState": state, "load_verified_model": verified, "modeld_pkl_path": lambda big: "bundled-big" if big else "bundled-small", "receipt_owner": receipt, "threading": SimpleNamespace(Thread=Thread), "BIG_MODEL_TIMEOUT": 30, + "report_load_timeout": Mock(), "params": params, "requested_model_id": "big", "ModelVariant": ModelVariant, "CP": SimpleNamespace(brand="mock"), "demo": False, "wait_for_chestnut_power": lambda CP, timeout: events.append("power"), @@ -209,6 +212,18 @@ def test_big_worker_waits_before_load_and_timeout_preserves_small_fallback(custo wrapper = ast.parse("def startup():\n pass").body[0] wrapper.body = body[start:end] + [ast.Return(value=ast.Call(func=ast.Name(id="locals", ctx=ast.Load()), args=[], keywords=[]))] execute([wrapper], env) + if timeout: + from openpilot.starpilot.models.recovery import STARTUP_FALLBACK_EXIT + with pytest.raises(SystemExit) as failed: + env["startup"]() + assert failed.value.code == STARTUP_FALLBACK_EXIT + env["drop_chestnut"].assert_not_called() + receipt.loaded.assert_not_called() + env["release_compile_pool"].assert_not_called() + bind.assert_not_called() + assert events.count("small") == int(custom) + params.put_bool.assert_called_once_with("ChestnutActive", False) + return env.update(env["startup"]()) bind.assert_not_called() assert env["jetlink_model"] is None diff --git a/openpilot/starpilot/models/tests/test_startup_recovery_process.py b/openpilot/starpilot/models/tests/test_startup_recovery_process.py new file mode 100644 index 0000000000..839477af68 --- /dev/null +++ b/openpilot/starpilot/models/tests/test_startup_recovery_process.py @@ -0,0 +1,104 @@ +"""An explicit failed GPU startup gets one new, isolated Small process.""" +from unittest.mock import Mock, patch + +import pytest + +from openpilot.starpilot.models.recovery import STARTUP_FALLBACK_EXIT +from openpilot.starpilot.models.recovery_process import ModeldProcess +from openpilot.starpilot.models.tests.test_recovery import BASE +from openpilot.starpilot.models.tests.test_recovery_process import Feed + + +@pytest.mark.parametrize('alive,exitcode,fresh,started,same_drive,shutting,small_only', [ + (False, STARTUP_FALLBACK_EXIT, True, True, True, False, False), + (True, STARTUP_FALLBACK_EXIT, True, True, True, False, False), + (False, 1, True, True, True, False, False), + (False, -9, True, True, True, False, False), + (False, None, True, True, True, False, False), + (False, STARTUP_FALLBACK_EXIT, False, True, True, False, False), + (False, STARTUP_FALLBACK_EXIT, True, False, True, False, False), + (False, STARTUP_FALLBACK_EXIT, True, True, False, False, False), + (False, STARTUP_FALLBACK_EXIT, True, True, True, True, False), + (False, STARTUP_FALLBACK_EXIT, True, True, True, False, True), +]) +def test_startup_restart_requires_owned_exit_and_current_drive(alive, exitcode, fresh, started, same_drive, shutting, small_only): + owner = ModeldProcess('modeld', 'module', lambda *args: True) + child = Mock(pid=123, exitcode=exitcode) + child.is_alive.return_value = alive + owner.proc = child + owner.launch_drive_id = BASE if same_drive else BASE - 1 + owner.recovery.drive_id = BASE + owner.recovery.small_only = small_only + owner.shutting_down = shutting + feed = Feed(2.) + if not fresh: + feed.logMonoTime['deviceState'] = BASE + with patch('openpilot.starpilot.models.recovery_process.time.monotonic_ns', return_value=BASE + 2_000_000_000), \ + patch('openpilot.starpilot.models.recovery_process.process_start_ticks', return_value=456), \ + patch('openpilot.starpilot.models.recovery_process.read_receipt', return_value=None): + owner.observe(feed, started) + qualifies = not alive and exitcode == STARTUP_FALLBACK_EXIT and fresh and started and same_drive and not shutting and not small_only + if qualifies: + child.join.assert_called_once_with(timeout=0) + assert owner.proc is None + assert owner.recovery.small_only + with patch('openpilot.starpilot.models.recovery_process.Process') as spawn: + owner.start() + assert spawn.call_args.kwargs['args'] == ('module', 'modeld', True) + spawn.return_value.start.assert_called_once_with() + owner.start() + spawn.assert_called_once() + else: + child.join.assert_not_called() + assert owner.proc is child + + +def test_launch_records_current_drive_and_never_retries_failed_small(): + owner = ModeldProcess('modeld', 'module', lambda *args: True) + owner.context_ready = True + owner.recovery.drive_id = BASE + owner.recovery.small_only = True + owner.recovery.phase = 'small' + with patch('openpilot.starpilot.models.recovery_process.Process') as spawn: + owner.start() + assert owner.launch_drive_id == BASE + child = spawn.return_value + child.pid = 123 + child.is_alive.return_value = False + child.exitcode = STARTUP_FALLBACK_EXIT + with patch('openpilot.starpilot.models.recovery_process.time.monotonic_ns', return_value=BASE + 2_000_000_000): + owner.observe(Feed(2.), True) + owner.start() + spawn.assert_called_once() + child.join.assert_not_called() + + +def _failed_gpu_startup(): + import threading + threading.Thread(target=threading.Event().wait, daemon=True).start() + raise SystemExit(STARTUP_FALLBACK_EXIT) + + +def test_real_failed_startup_exits_with_live_daemon_before_small_can_start(): + from multiprocessing import Process + child = Process(target=_failed_gpu_startup) + child.start() + try: + child.join(timeout=3) + assert not child.is_alive() + assert child.exitcode == STARTUP_FALLBACK_EXIT + owner = ModeldProcess('modeld', 'module', lambda *args: True) + owner.proc = child + owner.launch_drive_id = BASE + owner.recovery.drive_id = BASE + with patch('openpilot.starpilot.models.recovery_process.time.monotonic_ns', return_value=BASE + 2_000_000_000), \ + patch('openpilot.starpilot.models.recovery_process.Process') as spawn: + owner.observe(Feed(2.), True) + owner.start() + assert spawn.call_args.kwargs['args'] == ('module', 'modeld', True) + spawn.assert_called_once() + finally: + if child.is_alive(): + child.kill() + child.join(timeout=3) + child.close()