From bc3dce2c0c88bbb5025f99f908717bbea90e8c58 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Thu, 29 Oct 2026 16:12:56 -0500 Subject: [PATCH] Keep manager shutdown nonblocking and preserve paused AOL intent --- .../aol/tests/test_repeated_warning_join.py | 123 ++++++++++++++++++ openpilot/starpilot/aol/transport_pause.py | 29 +++-- .../starpilot/models/recovery_process.py | 2 +- openpilot/system/manager/process.py | 13 +- .../manager/test/test_nonblocking_stop.py | 76 +++++++++++ .../manager/test/test_process_recovery.py | 5 + prebuilt.json | 2 +- 7 files changed, 235 insertions(+), 15 deletions(-) create mode 100644 openpilot/starpilot/aol/tests/test_repeated_warning_join.py create mode 100644 openpilot/system/manager/test/test_nonblocking_stop.py diff --git a/openpilot/starpilot/aol/tests/test_repeated_warning_join.py b/openpilot/starpilot/aol/tests/test_repeated_warning_join.py new file mode 100644 index 0000000000..6e1aba7973 --- /dev/null +++ b/openpilot/starpilot/aol/tests/test_repeated_warning_join.py @@ -0,0 +1,123 @@ +from contextlib import ExitStack + +import pytest + +from opendbc.car.hyundai.values import CAR + +from openpilot.cereal import messaging +from openpilot.common.prefix import OpenpilotPrefix +from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture + + +@pytest.mark.parametrize('profile', [False, CAR.HYUNDAI_IONIQ_5_PE, CAR.KIA_EV9], ids=['ioniq6', 'ioniq5pe', 'ev9']) +@pytest.mark.parametrize('order', ['event_first', 'axis_first', 'together']) +def test_repeated_communication_warning_keeps_existing_latch(order, profile): + case = fixture.IoniqAvailabilityTests(methodName='runTest') + with ExitStack() as cleanup: + cleanup.callback(case.doCleanups) + cleanup.enter_context(OpenpilotPrefix()) + sd, _, step, card, receive = case._ioniq_availability_callers(full_op=not profile, stock_angle=profile) + for tick in range(2, 65): + axis, _, command = step(tick, car_output_valid=False, native_request_long=False) + assert not axis.lateralActive and not command.latActive and not command.longActive + assert receive(10_000_000_000 + tick * 10_000_000, axis=sd.pm.messages['aolAxisState'], + event=sd.pm.messages['onroadEvents']) + assert card.aol_transport_feedback.ordinary_warning_retained + axis, _, command = step(65, car_output_valid=False, native_request_long=False) + proof = sd.pm.messages['aolAxisState'].as_builder() + event = sd.pm.messages['onroadEvents'].as_builder() + now = 10_650_000_000 + event.logMonoTime = now + proof.aolAxisState.faultEventMonoTime = now + frames = [{'event': event}, {'axis': proof}] if order == 'event_first' else ( + [{'axis': proof}, {'event': event}] if order == 'axis_first' else [{'axis': proof, 'event': event}]) + for index, frame in enumerate(frames): + assert receive(now + index * 1_000_000, **frame), 'REPEATED_SOFT_WARNING_CLEARED_LATCH' + assert not card.aol_transport_feedback.failed + assert not command.latActive and not command.longActive and not sd.enabled + recovered, _, command = step(66, native_request_long=False) + assert recovered.lateralActive and command.latActive + assert not command.longActive and not sd.enabled + assert receive(10_660_000_000, axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents']) + + +@pytest.mark.parametrize('order', ['axis_first', 'event_first']) +@pytest.mark.parametrize('repeat', [False, True]) +def test_repeated_warning_missing_partner_keeps_original_deadline(order, repeat): + case = fixture.IoniqAvailabilityTests(methodName='runTest') + with ExitStack() as cleanup: + cleanup.callback(case.doCleanups) + cleanup.enter_context(OpenpilotPrefix()) + sd, _, step, card, receive = case._ioniq_availability_callers() + for tick in (2, 3): + step(tick, car_output_valid=False, native_request_long=False) + assert receive(10_000_000_000 + tick * 10_000_000, axis=sd.pm.messages['aolAxisState'], + event=sd.pm.messages['onroadEvents']) + step(4, car_output_valid=False, native_request_long=False) + now = 10_040_000_000 + proof = sd.pm.messages['aolAxisState'].as_builder() + event = sd.pm.messages['onroadEvents'].as_builder() + proof.aolAxisState.faultEventMonoTime = event.logMonoTime = now + half = {'axis': proof} if order == 'axis_first' else {'event': event} + assert receive(now, **half), 'REPEATED_SOFT_WARNING_CLEARED_LATCH' + original_deadline = card.aol_transport_feedback.pending_event + assert original_deadline == now + def fresh_half(stamp): + updated = messaging.new_message('aolAxisState', valid=True, logMonoTime=stamp) + updated.aolAxisState = card.sm['aolAxisState'] + updated.aolAxisState.sequence += 1 + updated.aolAxisState.observedMonoTime = updated.aolAxisState.sourceCarStateMonoTime = stamp + updated.aolAxisState.validUntilMonoTime = stamp + 30_000_000 + return {**half, 'axis': updated} + + if repeat: + assert receive(now + 10_000_000, **fresh_half(now + 10_000_000)) + assert card.aol_transport_feedback.pending_event == original_deadline + expired = now + 30_000_001 + assert not receive(expired, **(fresh_half(expired) if repeat else half)) + assert card.aol_transport_feedback.failed and not card.aol_card_intent.allowed_latch + + +@pytest.mark.parametrize('fault', ['producer', 'epoch', 'session', 'fault_session', 'stale_source', + 'future_source', 'invalid_native', 'can_invalid', 'permanent_eps', 'critical_event']) +def test_repeated_warning_join_preserves_denials(fault): + case = fixture.IoniqAvailabilityTests(methodName='runTest') + with ExitStack() as cleanup: + cleanup.callback(case.doCleanups) + cleanup.enter_context(OpenpilotPrefix()) + sd, _, step, card, receive = case._ioniq_availability_callers() + for tick in (2, 3): + step(tick, car_output_valid=False, native_request_long=False) + assert receive(10_000_000_000 + tick * 10_000_000, axis=sd.pm.messages['aolAxisState'], + event=sd.pm.messages['onroadEvents']) + step(4, car_output_valid=False, native_request_long=False) + now = 10_040_000_000 + proof = sd.pm.messages['aolAxisState'].as_builder() + proof.aolAxisState.faultEventMonoTime = now + observed = fixture.car_state() + frame = {'axis': proof} + if fault == 'producer': + proof.aolAxisState.intentProducerSessionId = 'other-card' + elif fault == 'epoch': + proof.aolAxisState.intentIntegrityEpoch += 1 + elif fault == 'session': + proof.aolAxisState.sessionId = 'other-session' + elif fault == 'fault_session': + proof.aolAxisState.faultSessionId = 'other-session' + elif fault == 'stale_source': + proof.aolAxisState.sourceCarStateMonoTime = now - 30_000_001 + elif fault == 'future_source': + proof.aolAxisState.sourceCarStateMonoTime = now + 1 + elif fault == 'invalid_native': + sd.sm.valid['aolSafetyWire'] = False + elif fault == 'can_invalid': + observed.canValid = False + elif fault == 'permanent_eps': + observed.steerFaultPermanent = True + else: + event = messaging.new_message('onroadEvents', 1, valid=True, logMonoTime=now) + event.onroadEvents[0].name = 'canError' + event.onroadEvents[0].immediateDisable = True + frame['event'] = event + assert not receive(now, state=observed, **frame) + assert not card.aol_card_intent.allowed_latch diff --git a/openpilot/starpilot/aol/transport_pause.py b/openpilot/starpilot/aol/transport_pause.py index e9fecee79c..e09ca17069 100644 --- a/openpilot/starpilot/aol/transport_pause.py +++ b/openpilot/starpilot/aol/transport_pause.py @@ -440,9 +440,6 @@ class TransportPauseFeedback: return False warning = any(event.softDisable for event in events) recovering = bool(getattr(self, 'ordinary_warning_retained', False) and not warning) - if pending and (axis.faultEventMonoTime != self.pending_event or - not 0 < self.pending_event <= now_ns or now_ns - self.pending_event > AXIS_MAX_AGE_NS): - return False stock_user = stock_lateral_user_event(cp) # Family button/pedal latch clear: no rearm. if any(event.userDisable and event.name != stock_user and not ( @@ -452,6 +449,20 @@ class TransportPauseFeedback: now_ns - event_ns <= SAFETY_MAX_AGE_NS)) for event in events): return False prior = pending_baseline if pending else self.signature + repeated_join = bool(warning and self.ordinary_warning_retained and prior is not None and + prior[5] == 'availabilityPause' and not any((prior[9], prior[10], prior[12])) and + prior[14:16] == (producer, self.availability_epoch) and + str(axis.faultReason) == 'availabilityPause' and not axis.lateralActive and + not axis.longitudinalActive and not axis.desiredLongitudinal and + (axis.desiredLateral, axis.nativeAcknowledged) == (prior[11], prior[13]) and + 0 < axis.faultEventMonoTime <= stamp and + 0 < prior[6] == min(axis.faultEventMonoTime, event_ns) < max(axis.faultEventMonoTime, event_ns) <= now_ns and + now_ns - max(axis.faultEventMonoTime, event_ns) <= AXIS_MAX_AGE_NS) + if pending and (not 0 < self.pending_event <= now_ns or + now_ns - self.pending_event > AXIS_MAX_AGE_NS or + (max(axis.faultEventMonoTime, event_ns) != self.pending_event if repeated_join else + axis.faultEventMonoTime != self.pending_event)): + return False # Pre-event SOFT: intent only, fixed original 30ms join; no newer/duplicate withdrawal reanchoring. axis_first = bool(str(axis.faultReason) == 'availabilityPause' and not (axis.lateralActive or axis.longitudinalActive or axis.desiredLateral or @@ -471,21 +482,21 @@ class TransportPauseFeedback: not axis.desiredLongitudinal and (warning and str(axis.faultReason) == 'availabilityPause' and axis.faultEventMonoTime == event_ns <= stamp or recovering and str(axis.faultReason) == 'none')) - if not active_join and not withdrawn and not axis_first: + if not active_join and not withdrawn and not axis_first and not repeated_join: return False - if pending and not axis_first and not (withdrawn and warning and event_ns == self.pending_event): + if pending and not axis_first and not repeated_join and not (withdrawn and warning and event_ns == self.pending_event): return False native = _configured_transport_native(sm, cp, self.session, now_ns, - allow_restricted_lateral=withdrawn or axis_first or bool(cs.steerFaultTemporary), - require_longitudinal=not (withdrawn or axis_first)) + allow_restricted_lateral=withdrawn or axis_first or repeated_join or bool(cs.steerFaultTemporary), + require_longitudinal=not (withdrawn or axis_first or repeated_join)) if (native is None or not native.pandaInventory or (native.pandaSerial, tuple(slot.hardwareSerial for slot in native.pandaInventory)) != self.lateral_inventory or native.observedMonoTime < self.lateral_native_ns or active_join and not (native.requestedLateral and native.lateralAllowed)): return False - if axis_first: + if axis_first or repeated_join: if not pending: - self.pending_event = int(axis.faultEventMonoTime) + self.pending_event = max(int(axis.faultEventMonoTime), event_ns) self.ordinary_warning_pending_baseline = prior self.awaiting = True return True diff --git a/openpilot/starpilot/models/recovery_process.py b/openpilot/starpilot/models/recovery_process.py index cecab75386..8c3e72df05 100644 --- a/openpilot/starpilot/models/recovery_process.py +++ b/openpilot/starpilot/models/recovery_process.py @@ -114,7 +114,7 @@ class ModeldProcess(PythonProcess): if not self.context_ready or self.recovery.phase in ('sigint', 'sigkill', 'failed'): return if self.shutting_down: - self.stop() + self.stop(block=False) if self.proc is not None: return self.proc = Process(name=self.name, target=modeld_launcher, diff --git a/openpilot/system/manager/process.py b/openpilot/system/manager/process.py index b8050b249c..78a0e6bb2b 100644 --- a/openpilot/system/manager/process.py +++ b/openpilot/system/manager/process.py @@ -67,6 +67,7 @@ class ManagerProcess(ABC): enabled = True name = "" shutting_down = False + stop_requested_at: float | None = None @abstractmethod def start(self) -> None: @@ -83,9 +84,12 @@ class ManagerProcess(ABC): sig = signal.SIGKILL if self.sigkill else signal.SIGINT self.signal(sig) self.shutting_down = True + self.stop_requested_at = time.monotonic() - if not block: - return None + if not block: + if retry and self.stop_requested_at is not None and time.monotonic() - self.stop_requested_at >= 5: + self.signal(signal.SIGKILL) + return None join_process(self.proc, 5) @@ -100,6 +104,7 @@ class ManagerProcess(ABC): if self.proc.exitcode is not None: self.shutting_down = False + self.stop_requested_at = None self.proc = None return ret @@ -143,7 +148,7 @@ class NativeProcess(ManagerProcess): def start(self) -> None: # In case we only tried a non blocking stop we need to stop it before restarting if self.shutting_down: - self.stop() + self.stop(block=False) if self.proc is not None: return @@ -195,7 +200,7 @@ class PythonProcess(ManagerProcess): def start(self) -> None: # In case we only tried a non blocking stop we need to stop it before restarting if self.shutting_down: - self.stop() + self.stop(block=False) if self.proc is not None: return diff --git a/openpilot/system/manager/test/test_nonblocking_stop.py b/openpilot/system/manager/test/test_nonblocking_stop.py new file mode 100644 index 0000000000..7d6c54ceab --- /dev/null +++ b/openpilot/system/manager/test/test_nonblocking_stop.py @@ -0,0 +1,76 @@ +import signal +from unittest.mock import Mock, call, patch + +import pytest + +from openpilot.starpilot.models.recovery_process import ModeldProcess +from openpilot.system.manager import process + + +def owner(kind): + if kind == 'native': + return process.NativeProcess('worker', '.', ['unused'], lambda *_: True) + if kind == 'modeld': + result = ModeldProcess('modeld', 'unused', lambda *_: True) + result.context_ready = True + return result + return process.PythonProcess('starportd', 'unused', lambda *_: True) + + +@pytest.mark.parametrize('kind', ['python', 'native', 'modeld']) +@pytest.mark.parametrize('restart', [False, True]) +def test_polling_stop_and_restart_never_wait_for_a_live_child(kind, restart): + p = owner(kind) + child = Mock(pid=123, exitcode=None) + p.proc = child + now = [100.] + with patch.object(process.time, 'monotonic', side_effect=lambda: now[0]), \ + patch.object(process.os, 'kill') as kill, \ + patch.object(process, 'join_process', side_effect=AssertionError('NONBLOCKING_STOP_WAITED')), \ + patch.object(process, 'Process', side_effect=AssertionError('OVERLAPPING_CHILD')): + p.stop(block=False) + for elapsed in (0.5, 1., 4.99): + now[0] = 100. + elapsed + if restart: + p.start() + else: + p.stop(block=False) + assert p.proc is child and p.shutting_down + assert p.stop_requested_at == 100. + assert kill.call_args_list == [call(123, signal.SIGINT)] + now[0] = 105. + p.stop(block=False) + assert kill.call_args_list == [call(123, signal.SIGINT), call(123, signal.SIGKILL)] + child.join.assert_not_called() + child.exitcode = -signal.SIGKILL + assert p.stop(block=False) == -signal.SIGKILL + assert p.proc is None and not p.shutting_down and p.stop_requested_at is None + + +def test_nonblocking_stop_retry_false_preserves_original_escalation_deadline(): + p = owner('python') + p.proc = Mock(pid=123, exitcode=None) + now = [100.] + with patch.object(process.time, 'monotonic', side_effect=lambda: now[0]), \ + patch.object(process.os, 'kill') as kill, \ + patch.object(process, 'join_process', side_effect=AssertionError('NONBLOCKING_STOP_WAITED')): + p.stop(retry=False, block=False) + now[0] = 106. + p.stop(retry=False, block=False) + assert kill.call_count == 1 + p.stop(block=False) + assert kill.call_args.args == (123, signal.SIGKILL) + + +@pytest.mark.parametrize('retry', [False, True]) +def test_explicit_blocking_shutdown_keeps_wait_and_kill_behavior(retry): + p = owner('python') + child = Mock(pid=123, exitcode=None) + p.proc = child + child.join.side_effect = lambda: setattr(child, 'exitcode', -signal.SIGKILL) + with patch.object(process.os, 'kill') as kill, patch.object(process, 'join_process') as wait: + result = p.stop(retry=retry) + wait.assert_called_once_with(child, 5) + assert kill.call_args.args == (123, signal.SIGKILL if retry else signal.SIGINT) + assert child.join.call_count == int(retry) + assert result == (-signal.SIGKILL if retry else None) diff --git a/openpilot/system/manager/test/test_process_recovery.py b/openpilot/system/manager/test/test_process_recovery.py index a53360172b..cecf003b71 100644 --- a/openpilot/system/manager/test/test_process_recovery.py +++ b/openpilot/system/manager/test/test_process_recovery.py @@ -173,6 +173,11 @@ class TestProcessRecovery(unittest.TestCase): self.assertTrue(old.joined) self.step(p, started=False) self.assertTrue(p.shutting_down) + stopping = p.proc + self.step(p, started=False) + self.assertIs(p.proc, stopping) + assert isinstance(stopping, Child) + stopping.exitcode = 0 self.step(p, started=False) self.assertIsNone(p.proc) self.now += 60 diff --git a/prebuilt.json b/prebuilt.json index de548aa3ee..c9e0adef42 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "8b02ec20dd62df5e56337f2378dc6ee87aba456cc486b14ad99f4b25f0992034", "schema_version": 1} +{"protected_inventory_sha256": "6ced2806b5125a6965c378b25c736e83d0187cd1fa2c0770baeed339d5ac0887", "schema_version": 1}