mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-11-06 20:43:46 +08:00
Hold RFComm for a moment, Better Traces.
This commit is contained in:
@@ -27,8 +27,8 @@ MAX_FAILURES = 5 # distinct failures detailed per drive report
|
||||
MAX_BREADCRUMBS = 100
|
||||
TRACE_QUIET = frozenset({"nav_guidance_sent", "nav_focus", "nav_focus_requested", "nav_status_state", "media_key", "media_audio_state",
|
||||
"media_audio_focus", "media_status_state", "hfp_command", "video_focus"})
|
||||
TRACE_SESSION_BYTES = 40 << 10 # per failed session; the oldest events go first
|
||||
TRACE_TOTAL_BYTES = 160 << 10 # all traces in one report, well under the sender's event cap
|
||||
TRACE_SESSION_BYTES = 64 << 10 # per failed session; the middle goes first
|
||||
TRACE_TOTAL_BYTES = 200 << 10 # all traces in one report, well under the sender's event cap
|
||||
EVENT_BUDGET = 400 << 10 # a report is shed down to this before sending; the sender refuses more than 512 KB
|
||||
MAX_ATTENTION = 6 # sessions per report that get a trace and a device slice
|
||||
SERIES_KEYS = ("fps", "encode_ms", "frame_age_p95_ms", "wifi_dbm", "wifi_link", "wifi_noise", "load1", "mem_available_mb",
|
||||
@@ -151,15 +151,19 @@ def quiet_counts(events: list[dict]) -> dict:
|
||||
|
||||
|
||||
def bounded_trace(trace: list[dict], limit: int) -> dict:
|
||||
"""The newest events that fit in ``limit`` bytes of JSON, with how many older ones were left out."""
|
||||
kept, size = [], 0
|
||||
for event in reversed(trace):
|
||||
size += len(json.dumps(event, default=str)) + 2
|
||||
if size > limit:
|
||||
break
|
||||
kept.append(event)
|
||||
kept.reverse()
|
||||
return {"events": kept, "omitted": len(trace) - len(kept)}
|
||||
"""The first quarter and the newest events that fit in ``limit`` bytes of JSON, with how many in between were left out."""
|
||||
sizes = [len(json.dumps(event, default=str)) + 2 for event in trace]
|
||||
if sum(sizes) <= limit:
|
||||
return {"events": list(trace), "omitted": 0}
|
||||
head_end, used = 0, 0
|
||||
while head_end < len(trace) and used + sizes[head_end] <= limit // 4:
|
||||
used += sizes[head_end]
|
||||
head_end += 1
|
||||
tail_start = len(trace)
|
||||
while tail_start > head_end and used + sizes[tail_start - 1] <= limit:
|
||||
tail_start -= 1
|
||||
used += sizes[tail_start]
|
||||
return {"events": trace[:head_end] + trace[tail_start:], "omitted": tail_start - head_end, "head": head_end}
|
||||
|
||||
|
||||
MAX_OFFERED_MODES = 16
|
||||
|
||||
@@ -30,7 +30,9 @@ ERRORS = {1: "invalid SDP version", 2: "invalid service record handle", 3: "inva
|
||||
|
||||
|
||||
class SdpError(RuntimeError):
|
||||
pass
|
||||
def __init__(self, message: str, raw: bytes = b""):
|
||||
super().__init__(message)
|
||||
self.raw = raw # what the car sent, for the log when the record is not usable
|
||||
|
||||
|
||||
def build_request(continuation: bytes, service: uuid.UUID = AA_WIRELESS_UUID, transaction_id: int = 0) -> bytes:
|
||||
@@ -107,5 +109,5 @@ def query_channel(sock, timeout: float = 10.0) -> int:
|
||||
raise SdpError("Head unit does not advertise Android Auto Wireless (SDP record missing)")
|
||||
channel = rfcomm_channel(bytes(collected))
|
||||
if channel is None:
|
||||
raise SdpError("Android Auto Wireless SDP record has no RFCOMM channel")
|
||||
raise SdpError("Android Auto Wireless SDP record has no RFCOMM channel", bytes(collected))
|
||||
return channel
|
||||
|
||||
@@ -61,6 +61,8 @@ ENCODER_RECOVERIES = 3 # hardware encoder reopens allowed per window befor
|
||||
ENCODER_RECOVERY_WINDOW = 10.0
|
||||
CAR_LINK_HOLD = 10.0 # a hands-free connection from the car counts as "car present" this long
|
||||
HFP_FRESH = 3.0 # a hands-free link from the car this recent still means "the car is reaching out now"
|
||||
COMMA_PAGE_WINDOW = 6.0 # a hands-free link this soon after we paged the car is ours; later is the car's
|
||||
RFCOMM_BUSY_WAITS = (1.0, 2.0) # the kernel can still hold the last session's channel for a moment
|
||||
CAR_FIRST_WAIT = 8.0 # at ignition, give the car this long to open hands-free before the comma pages it
|
||||
NO_STARPORT_WAIT = 8.0 # a paired device must offer wireless projection within this long
|
||||
USB_HANDSHAKE_WAIT = 10.0 # the car connected but sent no AOA START this long -> present as an accessory directly
|
||||
@@ -429,6 +431,8 @@ class Supervisor:
|
||||
self._pairing_attempt: PairingAttempt | None = None
|
||||
self._pairing_seen: tuple = ()
|
||||
self._car_seen_at = -CAR_LINK_HOLD
|
||||
self._paged_at = -COMMA_PAGE_WINDOW # when this comma last asked BlueZ to connect to the car
|
||||
self._attempt_ended_at: float | None = None
|
||||
self._hfp_link = threading.Event() # set when the chosen car opens the hands-free link
|
||||
self._modern_reply: dict[str, bool] = {} # car address -> answer its v4+ version request like a current phone
|
||||
self._ladder = setup_ladder.SetupLadder(identity_store.CONFIG_PATH.with_name("setup_state.json")) # which handshake to try for each car, and which one last worked
|
||||
@@ -1110,7 +1114,10 @@ class Supervisor:
|
||||
|
||||
def _hfp_connected(self, address: str) -> None:
|
||||
if address.upper() in (self.config["receiver_address"].upper(), self.config.get("companion_address", "").upper()):
|
||||
self._car_seen_at = time.monotonic()
|
||||
now = time.monotonic()
|
||||
since_page = now - self._paged_at
|
||||
self.log("hfp_initiator", initiator="comma" if since_page < COMMA_PAGE_WINDOW else "car", since_page_s=round(since_page, 1))
|
||||
self._car_seen_at = now
|
||||
self._hfp_link.set()
|
||||
|
||||
def _release_phone(self) -> None:
|
||||
@@ -1327,6 +1334,7 @@ class Supervisor:
|
||||
car_first = True # the car pages right back once its link drops; let it, rather than crossing it again
|
||||
finally:
|
||||
self._close_sockets()
|
||||
self._attempt_ended_at = time.monotonic()
|
||||
try:
|
||||
lease.release(restore=False)
|
||||
except Exception as error:
|
||||
@@ -1402,6 +1410,7 @@ class Supervisor:
|
||||
if delay := car_profiles.tuning(config)["connect_delay_s"]:
|
||||
self.log("car_profile_connect_delay", seconds=delay)
|
||||
self._wait(delay)
|
||||
self._paged_at = time.monotonic()
|
||||
bluez.connect_device(address)
|
||||
self._wait(1.0)
|
||||
|
||||
@@ -1426,7 +1435,7 @@ class Supervisor:
|
||||
channel_type=strategy.channel_type, patience=strategy.patience, experiments=strategy.experiments)
|
||||
try:
|
||||
self._stage("rfcomm", f"channel {channel}")
|
||||
rfcomm = self._track(bt_sockets.connect_rfcomm(address, channel))
|
||||
rfcomm = self._connect_rfcomm(address, channel)
|
||||
boot = WirelessBootstrap(rfcomm, self._bootstrap_log, device_serial=config["device_name"],
|
||||
version_status=int(config.get("version_status", 0)),
|
||||
experiment="all" if car_profiles.tuning(config)["bootstrap_experiments"] or strategy.experiments else "",
|
||||
@@ -1465,6 +1474,33 @@ class Supervisor:
|
||||
finally:
|
||||
keepalive_stop.set()
|
||||
|
||||
def _connect_rfcomm(self, address: str, channel: int):
|
||||
from openpilot.starpilot.system.starport import bt_sockets
|
||||
ended = self._attempt_ended_at
|
||||
since_end = round(time.monotonic() - ended, 1) if ended is not None else None
|
||||
waited = 0.0
|
||||
for tries, delay in enumerate(RFCOMM_BUSY_WAITS, 1):
|
||||
try:
|
||||
sock = self._track(bt_sockets.connect_rfcomm(address, channel))
|
||||
except OSError as error:
|
||||
if error.errno != errno.EBUSY:
|
||||
raise
|
||||
self.log("rfcomm_busy_retry", channel=channel, wait_s=delay, since_last_attempt_s=since_end)
|
||||
self._wait(delay)
|
||||
waited += delay
|
||||
continue
|
||||
if tries > 1:
|
||||
self.log("rfcomm_busy_cleared", channel=channel, tries=tries, waited_s=waited, since_last_attempt_s=since_end)
|
||||
return sock
|
||||
try:
|
||||
sock = self._track(bt_sockets.connect_rfcomm(address, channel))
|
||||
except OSError as error:
|
||||
if error.errno == errno.EBUSY:
|
||||
self.log("rfcomm_busy_gave_up", channel=channel, waited_s=waited, since_last_attempt_s=since_end)
|
||||
raise
|
||||
self.log("rfcomm_busy_cleared", channel=channel, tries=len(RFCOMM_BUSY_WAITS) + 1, waited_s=waited, since_last_attempt_s=since_end)
|
||||
return sock
|
||||
|
||||
def _discover_channel(self, address: str) -> int:
|
||||
from openpilot.starpilot.system.starport import bt_sockets, sdp
|
||||
last_error: Exception | None = None
|
||||
@@ -1478,7 +1514,7 @@ class Supervisor:
|
||||
if _stray_link(error):
|
||||
raise StrayLinkError(f"Bluetooth link to the car is stuck: {error}") from error # retrying cannot clear it
|
||||
last_error = error
|
||||
self.log("sdp_retry", error=str(error), settle=settle)
|
||||
self.log("sdp_retry", error=str(error), settle=settle, **({"record_hex": error.raw.hex()[:400]} if getattr(error, "raw", b"") else {}))
|
||||
self._wait(0.35)
|
||||
raise RuntimeError(f"Could not find the car's Android Auto service: {last_error}")
|
||||
|
||||
|
||||
@@ -515,3 +515,15 @@ def test_report_says_how_the_rfcomm_channel_was_found(tmp_path):
|
||||
events.insert(2, (0, 2, "rfcomm_channel", {"channel": 8, "source": "sdp"}))
|
||||
assert br.session_record(write_session(tmp_path, 1, events))["sdp"] == {"channel": 8, "source": "sdp", "retries": 1}
|
||||
assert br.session_record(write_session(tmp_path, 2, STREAMED))["sdp"] == {}
|
||||
|
||||
|
||||
def test_trace_keeps_how_the_session_started_as_well_as_how_it_ended(tmp_path):
|
||||
noisy = [(0, 0, "session_start", {}), (0, 1, "hfp_initiator", {"initiator": "car"})]
|
||||
noisy += [(i // 60, i % 60, "sdp_retry", {"error": "x" * 250, "n": i}) for i in range(2, 400)]
|
||||
noisy.append((9, 0, "attempt_failed", {"stage": "discovering", "error": "gone", "kind": "RuntimeError"}))
|
||||
event = br.drive_event([br.session_record(write_session(tmp_path, 1, noisy))])
|
||||
trace = event["contexts"]["starport_trace_session_1"]
|
||||
names = [e["event"] for e in trace["events"]]
|
||||
assert names[:2] == ["session_start", "hfp_initiator"] and names[-2:] == ["attempt_failed", "session_stop"]
|
||||
assert trace["omitted"] > 0 and 0 < trace["head"] < len(names)
|
||||
assert len(json.dumps(trace)) <= br.TRACE_SESSION_BYTES + 1024
|
||||
|
||||
@@ -179,3 +179,90 @@ def test_disconnect_stray_link_without_a_handle_only_logs(monkeypatch):
|
||||
phone, commands, events = _phone_with(monkeypatch, {"journalctl": ("", 0)})
|
||||
assert phone.disconnect_stray_link() is None
|
||||
assert len(commands) == 1 and events == ["stray_link_not_found"]
|
||||
|
||||
|
||||
def _rfcomm_attempts(sup, monkeypatch, outcomes):
|
||||
from openpilot.starpilot.system.starport import bt_sockets
|
||||
calls = []
|
||||
|
||||
def connect(*args, **kwargs):
|
||||
calls.append(args)
|
||||
outcome = outcomes[min(len(calls), len(outcomes)) - 1]
|
||||
if isinstance(outcome, int):
|
||||
raise OSError(outcome, "boom")
|
||||
return outcome
|
||||
|
||||
monkeypatch.setattr(bt_sockets, "connect_rfcomm", connect)
|
||||
return calls
|
||||
|
||||
|
||||
def test_busy_rfcomm_channel_is_retried_in_the_same_attempt(identity, tmp_path, monkeypatch):
|
||||
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
|
||||
_virtual_time(sup, monkeypatch)
|
||||
good = socket.socketpair()[0]
|
||||
calls = _rfcomm_attempts(sup, monkeypatch, [errno.EBUSY, good])
|
||||
assert sup._connect_rfcomm(CAR, 6) is good
|
||||
assert len(calls) == 2 and len(_events(sup, "rfcomm_busy_retry")) == 1
|
||||
|
||||
|
||||
def test_busy_rfcomm_channel_gives_up_after_the_waits(identity, tmp_path, monkeypatch):
|
||||
from openpilot.starpilot.system.starport import supervisor
|
||||
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
|
||||
_virtual_time(sup, monkeypatch)
|
||||
calls = _rfcomm_attempts(sup, monkeypatch, [errno.EBUSY])
|
||||
with pytest.raises(OSError) as raised:
|
||||
sup._connect_rfcomm(CAR, 6)
|
||||
assert raised.value.errno == errno.EBUSY and len(calls) == len(supervisor.RFCOMM_BUSY_WAITS) + 1
|
||||
|
||||
|
||||
def test_other_rfcomm_errors_are_not_retried(identity, tmp_path, monkeypatch):
|
||||
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
|
||||
_virtual_time(sup, monkeypatch)
|
||||
calls = _rfcomm_attempts(sup, monkeypatch, [errno.ECONNRESET])
|
||||
with pytest.raises(OSError):
|
||||
sup._connect_rfcomm(CAR, 6)
|
||||
assert len(calls) == 1 and not _events(sup, "rfcomm_busy_retry")
|
||||
|
||||
|
||||
def test_a_retried_busy_channel_logs_how_it_ended(identity, tmp_path, monkeypatch):
|
||||
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
|
||||
now = _virtual_time(sup, monkeypatch)
|
||||
sup._attempt_ended_at = now[0] - 4.0
|
||||
_rfcomm_attempts(sup, monkeypatch, [errno.EBUSY, errno.EBUSY, socket.socketpair()[0]])
|
||||
sup._connect_rfcomm(CAR, 6)
|
||||
assert _events(sup, "rfcomm_busy_cleared")[0]["tries"] == 3
|
||||
assert _events(sup, "rfcomm_busy_retry")[0]["since_last_attempt_s"] == 4.0
|
||||
|
||||
|
||||
def test_a_channel_that_stays_busy_logs_that_it_gave_up(identity, tmp_path, monkeypatch):
|
||||
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
|
||||
_virtual_time(sup, monkeypatch)
|
||||
_rfcomm_attempts(sup, monkeypatch, [errno.EBUSY])
|
||||
with pytest.raises(OSError):
|
||||
sup._connect_rfcomm(CAR, 6)
|
||||
assert _events(sup, "rfcomm_busy_gave_up")[0]["waited_s"] == 3.0
|
||||
|
||||
|
||||
def test_hands_free_logs_who_started_the_link(identity, tmp_path, monkeypatch):
|
||||
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
|
||||
now = _virtual_time(sup, monkeypatch)
|
||||
sup._paged_at = now[0] - 1.0
|
||||
sup._hfp_connected(CAR)
|
||||
sup._paged_at = now[0] - 60.0
|
||||
sup._hfp_connected(CAR)
|
||||
assert [e["initiator"] for e in _events(sup, "hfp_initiator")] == ["comma", "car"]
|
||||
|
||||
|
||||
def test_unusable_sdp_record_is_logged_with_what_the_car_sent(identity, tmp_path, monkeypatch):
|
||||
from openpilot.starpilot.system.starport import bt_sockets, sdp
|
||||
sup, _ = make_supervisor(identity, tmp_path, monkeypatch, lambda *a, **k: socket.socketpair()[0])
|
||||
_virtual_time(sup, monkeypatch)
|
||||
monkeypatch.setattr(bt_sockets, "connect_l2cap", lambda *a, **k: socket.socketpair()[0])
|
||||
|
||||
def query(_):
|
||||
raise sdp.SdpError("Android Auto Wireless SDP record has no RFCOMM channel", b"\x35\x03\x19\x11\x1f")
|
||||
|
||||
monkeypatch.setattr(sdp, "query_channel", query)
|
||||
with pytest.raises(RuntimeError):
|
||||
sup._discover_channel(CAR)
|
||||
assert _events(sup, "sdp_retry")[0]["record_hex"] == "350319111f"
|
||||
|
||||
Reference in New Issue
Block a user