From 5426a047d5e9af2ef317c7fb179518830fbfc2bc Mon Sep 17 00:00:00 2001 From: 1okko <15377594951@189.cn> Date: Tue, 29 Sep 2026 22:25:34 +0800 Subject: [PATCH] fix(qnn): soft-reset temporal history without reconnecting --- .../modeld_v2/external_host/faults.py | 4 ++ .../modeld_v2/external_host/opif/transport.py | 3 +- .../modeld_v2/external_host/opif_client.py | 6 +- .../modeld_v2/external_host/qnn_hotstandby.py | 13 ++++- .../modeld_v2/external_host/qnn_model.py | 13 ++--- .../modeld_v2/external_host/session.py | 57 +++++++++++++++---- 6 files changed, 71 insertions(+), 25 deletions(-) diff --git a/openpilot/sunnypilot/modeld_v2/external_host/faults.py b/openpilot/sunnypilot/modeld_v2/external_host/faults.py index 29b78cd9d..b96ec67c1 100644 --- a/openpilot/sunnypilot/modeld_v2/external_host/faults.py +++ b/openpilot/sunnypilot/modeld_v2/external_host/faults.py @@ -11,6 +11,10 @@ class HistoryResetRequired(HostError): """An input was not executed on the required sequence/cadence.""" +class TransientBackpressure(HostError): + """Local admission is full; drop this frame without resetting the transport.""" + + class FatalModelFault(HostError): """Model identity, schema, output or protocol integrity was not valid.""" diff --git a/openpilot/sunnypilot/modeld_v2/external_host/opif/transport.py b/openpilot/sunnypilot/modeld_v2/external_host/opif/transport.py index 7458a8e01..8352d31e5 100644 --- a/openpilot/sunnypilot/modeld_v2/external_host/opif/transport.py +++ b/openpilot/sunnypilot/modeld_v2/external_host/opif/transport.py @@ -20,6 +20,7 @@ class Frame: action_t: tuple[float, float] = (0., 0.) timestamp_ns: int = 0 context: object = None + reset_temporal: bool = False @dataclass @@ -73,7 +74,7 @@ class Pipeline: if len(frame.tensor) != self.session.tensor_bytes: raise ValueError("incorrect current6 tensor size") prefix = wire.INFERENCE_PREFIX_STRUCT.pack( - int(self.session.layout), 0, frame.desire, frame.traffic, int(self.next_id == 1), + int(self.session.layout), 0, frame.desire, frame.traffic, int(self.next_id == 1 or frame.reset_temporal), *frame.action_t, frame.timestamp_ns, len(frame.tensor), len(frame.tensor)) request = wire.encode_message(wire.MessageType.INFERENCE_REQUEST, self.session.session_id, self.next_id, prefix + frame.tensor, self.session.maximum_datagram_bytes) diff --git a/openpilot/sunnypilot/modeld_v2/external_host/opif_client.py b/openpilot/sunnypilot/modeld_v2/external_host/opif_client.py index 2539dfa95..8ef55bb5a 100644 --- a/openpilot/sunnypilot/modeld_v2/external_host/opif_client.py +++ b/openpilot/sunnypilot/modeld_v2/external_host/opif_client.py @@ -112,7 +112,7 @@ class OpifLink: self.close() raise classify(error) from error - def submit(self, frame_id, eof, pixels, scalars): + def submit(self, frame_id, eof, pixels, scalars, reset_temporal=False): if self.error is not None: raise self.error if not self.ready or self.stopped.is_set(): @@ -140,7 +140,7 @@ class OpifLink: self.sent[seq] = (int(frame_id), int(eof), captured) if self.stage_enabled: self.stages[seq] = {'t_link_submit_ns': time.clock_gettime_ns(time.CLOCK_BOOTTIME)} - frame = Frame(bytes(pixels), captured, desire, traffic, tuple(values[10:]), int(eof), (seq, int(frame_id), int(eof))) + frame = Frame(bytes(pixels), captured, desire, traffic, tuple(values[10:]), int(eof), (seq, int(frame_id), int(eof)), bool(reset_temporal)) try: self.pending.put_nowait(frame) except queue.Full as e: @@ -180,7 +180,7 @@ class OpifLink: 'runtime_identity_verified': False, 'result_schema': self.result_schema, 'opif_session_id': self.pipeline.session.session_id, 'state_epoch': 1, 'opif_state_epoch': 1, 'request_id': seq, 'opif_request_id': seq, - 'frame_id': fid, 'eof_ns': eof, + 'frame_id': fid, 'eof_ns': eof, 'soft_reset': bool(frame.reset_temporal), 'terminal': 'expired' if time.monotonic() - frame.captured >= self.frame_timeout else 'executed'} if self.stage_enabled: stats.update(self.stages.pop(seq, {})) diff --git a/openpilot/sunnypilot/modeld_v2/external_host/qnn_hotstandby.py b/openpilot/sunnypilot/modeld_v2/external_host/qnn_hotstandby.py index 7078b9ec6..332dfeae4 100644 --- a/openpilot/sunnypilot/modeld_v2/external_host/qnn_hotstandby.py +++ b/openpilot/sunnypilot/modeld_v2/external_host/qnn_hotstandby.py @@ -201,8 +201,17 @@ class QnnHotStandby: identity = (event.get('opif_session_id'), event.get('state_epoch')) if self.history_identity is None: self.history_identity = identity - if identity != self.history_identity: - raise HostError('Remote execution history epoch changed without requalification') + elif identity != self.history_identity: + if event.get('soft_reset') is not True: + raise HostError('Remote execution history epoch changed without requalification') + self.history_identity = identity + if event.get('soft_reset') is True: + # The client-side admission window requested RESET_TEMPORAL on the + # existing TCP session. Accept that sanctioned reset while preserving + # the transport and restarting only continuity/qualification. + self.last_remote_seq = self.last_remote_frame = None + self.good_frames = 0 + self.last_eligible_at = None seq, frame = event['seq'], event['frame_id'] if event['terminal'] not in ('executed', 'expired'): raise HostError('Remote input execution was not confirmed') diff --git a/openpilot/sunnypilot/modeld_v2/external_host/qnn_model.py b/openpilot/sunnypilot/modeld_v2/external_host/qnn_model.py index a62185478..6804469f1 100644 --- a/openpilot/sunnypilot/modeld_v2/external_host/qnn_model.py +++ b/openpilot/sunnypilot/modeld_v2/external_host/qnn_model.py @@ -8,7 +8,7 @@ import numpy as np from .session import AsyncSession from .wire import HostError -from .faults import TransportUnavailable, HistoryResetRequired +from .faults import TransportUnavailable, HistoryResetRequired, TransientBackpressure from .qnn_reconnect import RecoveryWorker SOURCE_SHA = '404a18cfd86d29637d20c697dfde245bb47c666ae016730ab674c65f4d1e1aa4' @@ -295,13 +295,10 @@ def create_model(base_class, config_path, cam_w, cam_h, progress=None): raise HostError('QNN preprocessing exceeded the capture deadline') try: self.async_session.submit(meta.frame_id, meta.timestamp_eof, pixels.tobytes(), scalars, context) - except HistoryResetRequired: - # A full local admission window is transient backpressure, not a - # transport failure. Keep the TCP session alive and drop this frame; - # the next frame can use the freed sequence slot without making the - # phone UI report a disconnect/reconnect cycle. - if self.async_session.error is not None or self.async_session.stopped: - raise + except TransientBackpressure: + # Local admission pressure is not a transport fault. The session + # keeps its TCP connection and schedules a RESET_TEMPORAL bit on the + # first frame submitted after the old in-flight window drains. if after_enqueue: after_enqueue() return None diff --git a/openpilot/sunnypilot/modeld_v2/external_host/session.py b/openpilot/sunnypilot/modeld_v2/external_host/session.py index e8e572225..f10d0bcd5 100644 --- a/openpilot/sunnypilot/modeld_v2/external_host/session.py +++ b/openpilot/sunnypilot/modeld_v2/external_host/session.py @@ -5,7 +5,7 @@ import time from collections import deque from .wire import HostError from .priority import background_thread -from .faults import classify, HistoryResetRequired +from .faults import classify, HistoryResetRequired, TransientBackpressure class AsyncSession: """Frame identities and their capture-time context stay paired until publication.""" @@ -28,6 +28,8 @@ class AsyncSession: self.terminal_events = deque(maxlen=256) self.terminal_history = deque(maxlen=256) self.state_identity = None + self.soft_reset_pending = False + self.awaiting_soft_reset_epoch = False self.threads=[threading.Thread(target=self._send,daemon=True),threading.Thread(target=self._receive,daemon=True)] for thread in self.threads:thread.start() @@ -52,16 +54,33 @@ class AsyncSession: def submit(self, frame_id,eof,pixels,scalars,context): self.check() + reset_temporal = False with self.condition: - if len(self.pending)>=self.capacity: - error_type=HistoryResetRequired if self.soft_expiry else HostError - raise error_type(f'External model has {self.capacity} outstanding frames') + if self.soft_reset_pending: + # Let every old request complete before the client-side temporal reset + # bit is sent. This keeps one TCP session alive through backpressure. + if self.pending or not self.outgoing.empty(): + raise TransientBackpressure('External model is draining for a soft history reset') + self.soft_reset_pending = False + self.awaiting_soft_reset_epoch = True + reset_temporal = True + elif len(self.pending)>=self.capacity or self.outgoing.full(): + if self.soft_expiry: + self.soft_reset_pending = True + raise TransientBackpressure(f'External model has {self.capacity} outstanding frames') + raise HostError('External result exceeded the local admission window') seq=self.next_seq;self.next_seq+=1 self.pending[seq]=(time.monotonic(),context,int(frame_id),int(eof)) - try:self.outgoing.put_nowait((seq,frame_id,eof,pixels,scalars)) - except queue.Full as error: - error_type=HistoryResetRequired if self.soft_expiry else HostError - raise error_type('External send queue is full') from error + try: + item=(seq,frame_id,eof,pixels,scalars,reset_temporal) if self.soft_expiry else (seq,frame_id,eof,pixels,scalars) + self.outgoing.put_nowait(item) + except queue.Full as error: + self.pending.pop(seq,None) + self.next_seq -= 1 + if self.soft_expiry: + self.soft_reset_pending = True + raise TransientBackpressure('External send queue is full') from error + raise HostError('External send queue is full') from error def poll(self,timeout=0): self.check() @@ -90,7 +109,14 @@ class AsyncSession: identity=(stats.get('opif_session_id'),stats.get('state_epoch')) if any(type(value) is not int or value<=0 for value in identity):raise HostError('Missing OPIF state identity') if self.state_identity is None:self.state_identity=identity - if identity!=self.state_identity:raise HostError('OPIF state epoch changed inside an active session') + if self.state_identity is None: + self.state_identity=identity + elif identity!=self.state_identity: + if not self.awaiting_soft_reset_epoch or stats.get('soft_reset') is not True: + raise HostError('OPIF state epoch changed inside an active session') + self.state_identity=identity + if stats.get('soft_reset') is True: + self.awaiting_soft_reset_epoch=False if stats.get('request_id')!=seq:raise HostError('OPIF request identity differs from session sequence') expired=source_age>=self.maximum_age or stats.get('terminal')=='expired' terminal='expired' if expired else 'executed' @@ -98,6 +124,7 @@ class AsyncSession: received_age = (received_ns-eof)/1e9 if type(received_ns) is int else None event={'seq':seq,'request_id':seq,'frame_id':fid,'eof_ns':eof,'context':context, 'opif_session_id':identity[0],'state_epoch':identity[1], + 'soft_reset':bool(stats.get('soft_reset',False)), 'opif_state_epoch':identity[1],'opif_request_id':seq, 'terminal':terminal,'received_age_seconds':received_age,'consumed_age_seconds':source_age} if len(self.terminal_events)>=256:raise HostError('Unconsumed OPIF terminal event queue is full') @@ -125,9 +152,17 @@ class AsyncSession: background_thread() try: while not self.stopped: - try:seq,fid,eof,pixels,scalars=self.outgoing.get(timeout=.05) + try:item=self.outgoing.get(timeout=.05) except queue.Empty:continue - actual=self.link.submit(fid,eof,pixels,scalars) + if len(item)==6: + seq,fid,eof,pixels,scalars,reset_temporal=item + else: + seq,fid,eof,pixels,scalars=item + reset_temporal=False + if reset_temporal: + actual=self.link.submit(fid,eof,pixels,scalars,reset_temporal=True) + else: + actual=self.link.submit(fid,eof,pixels,scalars) if actual!=seq:raise HostError('Sender sequence diverged') with self.condition:self.condition.notify_all() except Exception as error:self._fail(error)