mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-11-02 20:13:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

Render alert audio without the Python GIL

Browse Source
This commit is contained in:
firestar5683
2026-10-14 16:33:46 -05:00
parent fb227e68a1
commit 4ba8d6d891
9 changed files with 389 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SConstruct
+1
View File
@@ -328,6 +328,7 @@ SConscript([
'openpilot/selfdrive/locationd/SConscript',
'openpilot/selfdrive/modeld/SConscript',
'openpilot/selfdrive/ui/SConscript',
'openpilot/starpilot/audio/native/SConscript',
])
# Replay is also the isolated on-device viewer engine.
openpilot/selfdrive/ui/soundd.py
+28 -148
View File
@@ -1,11 +1,6 @@
import math
import json
import os
import numpy as np
import time
from time import monotonic_ns
from threading import get_native_id
from collections import deque
from pathlib import Path
@@ -16,7 +11,7 @@ from openpilot.common.utils import retry
from openpilot.common.swaglog import cloudlog
from openpilot.system import micd
from openpilot.common.hardware import HARDWARE
from openpilot.common.hardware import HARDWARE, PC
from openpilot.common.params import Params
from openpilot.starpilot.aol.wire import INTENT_SERVICE
from openpilot.starpilot.audio.alert_volume import AUTO, VOLUMES, effective_volume, read_volume
@@ -85,93 +80,6 @@ def check_selfdrive_timeout_alert(sm):
return False
def _native_audio_time(info, key):
try:
value = float(getattr(info, key))
return value if math.isfinite(value) else None
except (AttributeError, TypeError, ValueError):
return None
def _callback_thread_snapshot(tid):
snapshot = {"tid": tid}
try:
directory = Path(f"/proc/self/task/{tid}")
with (directory / "stat").open() as source:
fields = source.read(4096).rsplit(")", 1)[1].split()
with (directory / "schedstat").open() as source:
runtime_ns, runqueue_ns, timeslices = map(int, source.read(4096).split())
snapshot.update(state=fields[0], priority=int(fields[15]), nice=int(fields[16]), processor=int(fields[36]),
runtime_ns=runtime_ns, runqueue_ns=runqueue_ns, timeslices=timeslices,
policy=os.sched_getscheduler(tid), affinity=sorted(os.sched_getaffinity(tid)))
except (OSError, ValueError, IndexError, AttributeError):
snapshot["unavailable"] = True
return snapshot
def _read_callback_schedstat(tid):
with Path(f"/proc/self/task/{tid}/schedstat").open() as source:
values = tuple(map(int, source.read(128).split()))
if len(values) != 3 or any(value < 0 for value in values):
raise ValueError("invalid callback schedstat")
return values
class _CallbackSchedulerHistory:
MAX_GAP_NS = 100_000_000
def __init__(self):
self.tid = None
self.samples = deque(maxlen=128)
def sample(self, tid):
if tid != self.tid:
self.samples.clear()
self.tid = tid
if tid is None:
return
try:
begin = monotonic_ns()
counters = _read_callback_schedstat(tid)
end = monotonic_ns()
except (OSError, ValueError, IndexError, AttributeError):
self.samples.clear()
return
if end < begin or end - begin > self.MAX_GAP_NS:
self.samples.clear()
return
if self.samples and (begin < self.samples[-1][1] or
any(new < old for new, old in zip(counters, self.samples[-1][2], strict=True))):
self.samples.clear()
self.samples.append((begin, end, counters))
def bracket(self, tid, event_end, event_gap):
unknown = {"available": False}
if tid != self.tid or event_gap is None or event_gap <= 0:
return unknown
event_start = event_end - event_gap
before = [i for i, sample in enumerate(self.samples) if sample[1] <= event_start]
after = [i for i, sample in enumerate(self.samples) if sample[0] >= event_end]
if not before or not after:
return unknown
window = list(self.samples)[before[-1]:after[0] + 1]
if len(window) < 2:
return unknown
first, last = window[0], window[-1]
max_gap = max(right[1] - left[0] for left, right in zip(window, window[1:], strict=False))
if (max_gap > self.MAX_GAP_NS or event_start - first[1] > self.MAX_GAP_NS or
last[0] - event_end > self.MAX_GAP_NS):
return unknown
return {"available": True, "tid": tid,
"event_gap_start_ns": event_start, "event_gap_end_ns": event_end,
"sample_before_begin_ns": first[0], "sample_before_end_ns": first[1],
"sample_after_begin_ns": last[0], "sample_after_end_ns": last[1],
"runtime_delta_ns": last[2][0] - first[2][0],
"runqueue_delta_ns": last[2][1] - first[2][1],
"timeslices_delta": last[2][2] - first[2][2],
"max_sample_gap_ns": max_gap, "sample_count": len(window)}
class Soundd:
def __init__(self, params=None, pack_root=PACK_ROOT):
self.volume_params = params if params is not None else Params()
@@ -196,12 +104,7 @@ class Soundd:
self.output_underflow_count = 0
self.reported_output_underflows = 0
self.stream_report_at = None
self.last_callback_start_ns = None
self.last_callback_work_ns = None
self.last_native_audio_time = None
self.callback_tid = None
self.callback_scheduler_history = _CallbackSchedulerHistory()
self.underflow_timing = None
self.native_audio = None
self.axis_alerts = AxisAlerts()
self.spl_filter_weighted = FirstOrderFilter(0, 2.5, FILTER_DT, initialized=False)
@@ -256,33 +159,11 @@ class Soundd:
self.saved_volumes = {key: read_volume(self.volume_params, key).value for key, _, _ in VOLUMES}
def callback(self, data_out: np.ndarray, frames: int, time, status) -> None:
started_ns = monotonic_ns()
previous_ns = getattr(self, "last_callback_start_ns", None)
previous_work_ns = getattr(self, "last_callback_work_ns", None)
previous_native_time = getattr(self, "last_native_audio_time", None)
native_time = _native_audio_time(time, "currentTime")
dac_time = _native_audio_time(time, "outputBufferDacTime")
native_gap = (native_time - previous_native_time if native_time is not None and
previous_native_time is not None and native_time >= previous_native_time else None)
if getattr(self, "callback_tid", None) is None:
self.callback_tid = get_native_id()
self.last_callback_start_ns = started_ns
self.last_native_audio_time = native_time
try:
data_out[:frames, 0] = self.get_sound_data(frames)
finally:
finished_ns = monotonic_ns()
self.last_callback_work_ns = finished_ns - started_ns
if status:
underflow = bool(getattr(status, "output_underflow", False))
if underflow:
self.underflow_timing = (self.output_underflow_count + 1, started_ns,
None if previous_ns is None else started_ns - previous_ns,
finished_ns - started_ns, frames, previous_work_ns,
native_time, dac_time, native_gap, self.callback_tid)
self.pending_stream_status = status
self.stream_status_count += 1
self.output_underflow_count += int(underflow)
data_out[:frames, 0] = self.get_sound_data(frames)
if status:
self.pending_stream_status = status
self.stream_status_count += 1
self.output_underflow_count += int(bool(getattr(status, "output_underflow", False)))
def log_pending_stream_status(self, stream=None) -> None:
status = self.pending_stream_status
@@ -296,27 +177,6 @@ class Soundd:
underflows = self.output_underflow_count
message = (f"soundd stream diagnostics: status={status} status_callbacks={self.stream_status_count} "
+ f"output_underflows={underflows}")
timing = getattr(self, "underflow_timing", None)
if timing is not None and timing[0] == underflows:
_, started_ns, gap_ns, work_ns, frames, previous_work_ns, native_time, dac_time, native_gap, tid = timing
gap_ms = "unknown" if gap_ns is None else f"{gap_ns / 1e6:.3f}"
message += (f" underflow_callback_mono={started_ns / 1e9:.9f} callback_gap_ms={gap_ms}"
+ f" callback_work_ms={work_ns / 1e6:.3f}"
+ f" callback_period_ms={frames / SAMPLE_RATE * 1e3:.3f}")
previous_work_ms = "unknown" if previous_work_ns is None else f"{previous_work_ns / 1e6:.3f}"
native_text = "unknown" if native_time is None else f"{native_time:.9f}"
dac_text = "unknown" if dac_time is None else f"{dac_time:.9f}"
native_gap_ms = "unknown" if native_gap is None else f"{native_gap * 1e3:.3f}"
message += (f" previous_callback_work_ms={previous_work_ms} native_current_time={native_text}"
+ f" native_output_dac_time={dac_text} native_callback_gap_ms={native_gap_ms} callback_tid={tid}")
if underflows > self.reported_output_underflows:
snapshot = json.dumps(_callback_thread_snapshot(tid), separators=(",", ":"), sort_keys=True)
message += f" callback_thread_snapshot_current={snapshot} snapshot_mono={now:.9f}"
history = getattr(self, "callback_scheduler_history", None)
bracket = history.bracket(tid, started_ns, gap_ns) if history is not None else {"available": False}
message += " callback_scheduler_bracket=" + json.dumps(bracket, separators=(",", ":"), sort_keys=True)
if stream is not None:
message += f" latency={stream.latency} cpu_load={stream.cpu_load}"
# Output dropouts must reach errorLogMessage, which is retained in qlog.
if underflows > self.reported_output_underflows:
cloudlog.error(message)
@@ -381,6 +241,23 @@ class Soundd:
# reload sounddevice to reinitialize portaudio
sd._terminate()
sd._initialize()
if not PC:
from openpilot.starpilot.audio.native_playback import NativePlayback
try:
playback = NativePlayback(self, sd)
except (OSError, AttributeError, ValueError, MemoryError):
pass
else:
try:
stream = playback.stream()
except (AttributeError, TypeError, ValueError):
playback.close()
except BaseException:
playback.close()
raise
else:
self.native_audio = playback
return stream
return sd.OutputStream(channels=1, samplerate=SAMPLE_RATE, callback=self.callback, blocksize=SAMPLE_BUFFER)
def soundd_thread(self):
@@ -396,7 +273,8 @@ class Soundd:
cloudlog.info(f"soundd stream started: {stream.samplerate=} {stream.channels=} {stream.dtype=} {stream.device=}, {stream.blocksize=}, {stream.latency=}")
while True:
sm.update(0)
self.callback_scheduler_history.sample(self.callback_tid)
if self.native_audio is not None:
self.native_audio.sync_owner()
self.log_pending_stream_status(stream)
self.refresh_saved_volumes(time.monotonic())
self.load_sounds()
@@ -410,6 +288,8 @@ class Soundd:
self.get_audible_alert(sm)
self.update_critical_sound(time.monotonic())
if self.native_audio is not None:
self.native_audio.publish()
rk.keep_time()
openpilot/selfdrive/ui/tests/test_soundd.py
-110
View File
@@ -1,5 +1,4 @@
import threading
from unittest.mock import mock_open
import numpy as np
from openpilot.common.test import OpenpilotTestCase
@@ -12,8 +11,6 @@ AudibleAlert = log.SelfdriveState.AudibleAlert
class TestSoundd(OpenpilotTestCase):
def test_callback_defers_and_bounds_stream_status_logging(self, mocker):
scheduler_read = mocker.patch("openpilot.selfdrive.ui.soundd._read_callback_schedstat", side_effect=AssertionError("callback I/O"))
thread_snapshot = mocker.patch("openpilot.selfdrive.ui.soundd._callback_thread_snapshot", side_effect=AssertionError("callback I/O"))
soundd = Soundd.__new__(Soundd)
soundd.current_alert = AudibleAlert.none
soundd.current_volume = 0.1
@@ -24,8 +21,6 @@ class TestSoundd(OpenpilotTestCase):
soundd.output_underflow_count = 0
soundd.reported_output_underflows = 0
soundd.stream_report_at = None
soundd.last_callback_start_ns = None
soundd.underflow_timing = None
warning = mocker.patch("openpilot.selfdrive.ui.soundd.cloudlog.warning")
error = mocker.patch("openpilot.selfdrive.ui.soundd.cloudlog.error")
output = np.empty((8, 1), dtype=np.float32)
@@ -35,8 +30,6 @@ class TestSoundd(OpenpilotTestCase):
assert np.array_equal(output, np.zeros_like(output))
warning.assert_not_called()
error.assert_not_called()
scheduler_read.assert_not_called()
thread_snapshot.assert_not_called()
soundd.log_pending_stream_status()
warning.assert_called_once_with("soundd stream diagnostics: status=latest underflow status_callbacks=2 output_underflows=0")
@@ -65,106 +58,3 @@ class TestSoundd(OpenpilotTestCase):
assert not check_selfdrive_timeout_alert(sm)
# TODO: add test with micd for checking that soundd actually outputs sounds
def scheduler_sample(history, mocker, begin, counters, *, tid=7, duration=1):
mocker.patch("openpilot.selfdrive.ui.soundd.monotonic_ns", side_effect=(begin, begin + duration))
mocker.patch("openpilot.selfdrive.ui.soundd._read_callback_schedstat", return_value=counters)
history.sample(tid)
class TestSounddSchedulerHistory(OpenpilotTestCase):
def test_scheduler_history_reports_only_explicit_bracket(self, mocker):
from openpilot.selfdrive.ui.soundd import _CallbackSchedulerHistory
history = _CallbackSchedulerHistory()
for tick in range(8):
scheduler_sample(history, mocker, tick * 50_000_000, (tick * 1_000_000, tick * 3_000_000, tick))
bracket = history.bracket(7, 320_000_000, 270_000_000)
assert bracket == {"available": True, "tid": 7,
"event_gap_start_ns": 50_000_000, "event_gap_end_ns": 320_000_000,
"sample_before_begin_ns": 0, "sample_before_end_ns": 1,
"sample_after_begin_ns": 350_000_000, "sample_after_end_ns": 350_000_001,
"runtime_delta_ns": 7_000_000, "runqueue_delta_ns": 21_000_000,
"timeslices_delta": 7, "max_sample_gap_ns": 50_000_001, "sample_count": 8}
assert history.bracket(8, 320_000_000, 270_000_000) == {"available": False}
assert history.bracket(7, 320_000_000, None) == {"available": False}
assert history.bracket(7, 320_000_000, 0) == {"available": False}
assert history.bracket(7, 400_000_000, 270_000_000) == {"available": False}
assert history.bracket(7, 320_000_000, 400_000_000) == {"available": False}
def test_scheduler_history_never_bridges_missing_or_reset_samples(self, mocker):
for failure in ("gap", "slow_read", "read_error", "counter_reset", "tid_reset"):
with self.subTest(failure=failure):
from openpilot.selfdrive.ui.soundd import _CallbackSchedulerHistory
history = _CallbackSchedulerHistory()
scheduler_sample(history, mocker, 0, (100, 100, 1))
if failure == "gap":
scheduler_sample(history, mocker, 200_000_000, (200, 200, 2))
elif failure == "slow_read":
scheduler_sample(history, mocker, 50_000_000, (200, 200, 2), duration=101_000_000)
elif failure == "read_error":
mocker.patch("openpilot.selfdrive.ui.soundd.monotonic_ns", return_value=50_000_000)
mocker.patch("openpilot.selfdrive.ui.soundd._read_callback_schedstat", side_effect=OSError("missing proc"))
history.sample(7)
elif failure == "counter_reset":
scheduler_sample(history, mocker, 50_000_000, (1, 1, 1))
else:
scheduler_sample(history, mocker, 50_000_000, (200, 200, 2), tid=8)
scheduler_sample(history, mocker, 250_000_000, (300, 300, 3), tid=8 if failure == "tid_reset" else 7)
assert history.bracket(7, 220_000_000, 210_000_000) == {"available": False}
def test_scheduler_history_is_capped_and_has_no_regular_logging(self, mocker):
from openpilot.selfdrive.ui.soundd import _CallbackSchedulerHistory
warning = mocker.patch("openpilot.selfdrive.ui.soundd.cloudlog.warning")
error = mocker.patch("openpilot.selfdrive.ui.soundd.cloudlog.error")
history = _CallbackSchedulerHistory()
for tick in range(140):
scheduler_sample(history, mocker, tick * 50_000_000, (tick, tick, tick))
assert len(history.samples) == 128
warning.assert_not_called()
error.assert_not_called()
history.sample(None)
assert not history.samples
def test_scheduler_reader_rejects_malformed_counters(self, mocker):
for raw in ("1 2", "1 2 3 4", "-1 2 3", "1 bad 3"):
with self.subTest(raw=raw):
from openpilot.selfdrive.ui.soundd import _read_callback_schedstat
opened = mocker.patch("pathlib.Path.open", mock_open(read_data=raw))
with self.assertRaises(ValueError):
_read_callback_schedstat(7)
opened().read.assert_called_once_with(128)
def test_underflow_log_contains_scheduler_bracket_without_callback_io(self, mocker):
from types import SimpleNamespace
from openpilot.selfdrive.ui.soundd import _CallbackSchedulerHistory
owner = Soundd.__new__(Soundd)
owner.current_alert, owner.current_volume = AudibleAlert.none, 0.1
owner.loaded_sounds, owner.saved_volumes = {}, {}
owner.pending_stream_status = None
owner.stream_status_count = owner.output_underflow_count = owner.reported_output_underflows = 0
owner.stream_report_at = owner.underflow_timing = None
owner.last_callback_start_ns = 25_000_000
owner.callback_tid = 7
owner.callback_scheduler_history = _CallbackSchedulerHistory()
for tick in range(4):
scheduler_sample(owner.callback_scheduler_history, mocker, tick * 50_000_000, (tick, tick * 10, tick))
reader = mocker.patch("openpilot.selfdrive.ui.soundd._read_callback_schedstat", side_effect=AssertionError("callback I/O"))
snapshot = mocker.patch("openpilot.selfdrive.ui.soundd._callback_thread_snapshot", return_value={})
error = mocker.patch("openpilot.selfdrive.ui.soundd.cloudlog.error")
mocker.patch("openpilot.selfdrive.ui.soundd.monotonic_ns", side_effect=(125_000_000, 125_001_000))
mocker.patch("openpilot.selfdrive.ui.soundd.time.monotonic", return_value=0.2)
output = np.empty((8, 1), dtype=np.float32)
owner.callback(output, 8, None, SimpleNamespace(output_underflow=True))
assert np.array_equal(output, np.zeros_like(output))
reader.assert_not_called()
snapshot.assert_not_called()
error.assert_not_called()
owner.log_pending_stream_status()
error.assert_called_once()
message = error.call_args.args[0]
assert 'callback_scheduler_bracket={"available":true,' in message
assert '"runqueue_delta_ns":30' in message
assert '"event_gap_start_ns":25000000' in message
owner.log_pending_stream_status()
error.assert_called_once()
openpilot/starpilot/audio/native/SConscript
+6
View File
@@ -0,0 +1,6 @@
Import('env', 'arch')
if arch == "comma_arm64":
native_audio = env.Clone()
native_audio.Append(CFLAGS=['-std=c11'])
native_audio.SharedLibrary('audio_mixer', ['audio_mixer.c'])
openpilot/starpilot/audio/native/audio_mixer.c
+115
View File
@@ -0,0 +1,115 @@
// Bounded mono float32 alert renderer. PCM allocation/publication/reclamation occur on the control thread only.
#define _GNU_SOURCE
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>
#define SLOTS 16
#define MAX_PCM 1440000U
#define MAX_FRAMES 4096U
_Static_assert(ATOMIC_LONG_LOCK_FREE == 2 && ATOMIC_LLONG_LOCK_FREE == 2 && ATOMIC_POINTER_LOCK_FREE == 2, "Audio requires lock-free atomics");
struct pa_time { double input_adc, current, output_dac; };
struct bank { uint64_t revision; unsigned lengths[SLOTS]; float *pcm[SLOTS]; };
struct audio {
struct bank *active;
_Atomic(struct bank *) pending, retired;
_Atomic uint64_t command_seq, command[8], status_seq, status[10];
uint64_t applied_seq, restart, frame, stop_frame;
uint64_t status_count, underflows, last_flags;
unsigned alert, sound;
int loops;
float gain;
bool pending_stop;
atomic_flag rendering;
};
static void free_bank(struct bank *b) { if (b) { for(unsigned i=0;i<SLOTS;i++) free(b->pcm[i]); free(b); } }
unsigned audio_mixer_abi(void) { return 3; }
struct audio *audio_create(void) { struct audio *a=calloc(1,sizeof(*a)); if(a) { *a=(struct audio){.rendering=ATOMIC_FLAG_INIT,.loops=-1}; } return a; }
void audio_collect(struct audio *a) { free_bank(atomic_exchange(&a->retired,NULL)); }
int audio_can_publish(struct audio *a) { return a && atomic_load(&a->pending)==NULL; }
int audio_publish_bank(struct audio *a,const float *const *pcm,const unsigned *lengths,unsigned count,uint64_t revision) {
if(!a || !revision || count>SLOTS || atomic_load(&a->pending)) return 0;
audio_collect(a);
struct bank *b=calloc(1,sizeof(*b)); if(!b) return 0; b->revision=revision;
for(unsigned i=0;i<count;i++) {
if(!lengths[i] || lengths[i]>MAX_PCM || !pcm[i]) { free_bank(b); return 0; }
b->pcm[i]=malloc((size_t)lengths[i]*sizeof(float)); if(!b->pcm[i]) {free_bank(b);return 0;}
memcpy(b->pcm[i],pcm[i],(size_t)lengths[i]*sizeof(float)); b->lengths[i]=lengths[i];
}
atomic_store(&a->pending,b); return 1;
}
uint64_t audio_command(struct audio *a,unsigned alert,unsigned sound,int loops,int stop,float gain,uint64_t restart,uint64_t bank_revision,uint64_t stop_frame) {
if(!a || sound>=SLOTS || loops < -1 || loops==0 || !isfinite(gain) || gain<0 || gain>1) return 0;
uint32_t bits;memcpy(&bits,&gain,sizeof(bits));
uint64_t seq=atomic_load(&a->command_seq);atomic_store(&a->command_seq,seq+1);
uint64_t values[8]={alert,sound,(uint64_t)(int64_t)loops,stop!=0,bits,restart,bank_revision,stop_frame};
for(unsigned i=0;i<8;i++) atomic_store(&a->command[i],values[i]);
atomic_store(&a->command_seq,seq+2); return seq+2;
}
static void accept_command(struct audio *a) {
uint64_t seq=atomic_load(&a->command_seq),v[8];
if(seq==a->applied_seq || (seq&1)) return;
for(unsigned i=0;i<8;i++) v[i]=atomic_load(&a->command[i]);
if(seq!=atomic_load(&a->command_seq)) return;
if(!a->active || v[6]!=a->active->revision) {
struct bank *next=atomic_load(&a->pending);
if(!next || next->revision!=v[6] || atomic_load(&a->retired)) return;
next=atomic_exchange(&a->pending,NULL);
struct bank *old=a->active;a->active=next;atomic_store(&a->retired,old);
}
if(v[5]!=a->restart) { a->frame=0;a->restart=v[5]; }
a->alert=(unsigned)v[0];a->sound=(unsigned)v[1];a->loops=(int)(int64_t)v[2];
a->pending_stop=v[3]!=0;a->stop_frame=v[7];
// The main thread may publish dismissal after this cursor has already crossed
// its observed loop endpoint. Never replay an additional loop in that case.
if(a->pending_stop && a->frame>=a->stop_frame) { a->alert=0;a->pending_stop=false; }
uint32_t bits=(uint32_t)v[4];memcpy(&a->gain,&bits,sizeof(bits));a->applied_seq=seq;
}
// Initialization only: caller has not yet opened/started a PortAudio stream.
// Admit the initial bank without rendering samples or publishing fake timing.
int audio_prepare(struct audio *a) {
if(!a) return 0;
accept_command(a);
return a->active!=NULL && a->applied_seq!=0;
}
static int audio_callback(const void *input,void *output,unsigned long frames,const struct pa_time *t,unsigned long flags,void *user) {
(void)input;(void)t;struct audio *a=user;
if(!a || !output || frames>MAX_FRAMES || atomic_flag_test_and_set(&a->rendering)) return 2;
accept_command(a);
float *out=output;memset(out,0,frames*sizeof(float));
unsigned length=a->active?a->active->lengths[a->sound]:0;
if(a->alert && !length) { atomic_flag_clear(&a->rendering);return 2; }
unsigned written=0;
while(a->alert && written<frames && (a->loops<0 || a->frame/length<(uint64_t)a->loops)) {
unsigned pos=(unsigned)(a->frame%length),n=length-pos;
if(n>frames-written)n=(unsigned)frames-written;
memcpy(out+written,a->active->pcm[a->sound]+pos,n*sizeof(float));written+=n;a->frame+=n;
if(a->pending_stop && a->frame%length==0) { a->alert=0;a->pending_stop=false;break; }
}
for(unsigned i=0;i<frames;i++) out[i]*=a->gain;
// Only detached stream health and playback cursor cross the callback boundary.
uint64_t status_seq=atomic_load(&a->status_seq);atomic_store(&a->status_seq,status_seq+1);
if(flags) { a->status_count++;a->last_flags=flags; }
if(flags&4U) a->underflows++;
atomic_store(&a->status[0],a->frame);atomic_store(&a->status[1],a->alert);atomic_store(&a->status[2],a->pending_stop);
atomic_store(&a->status[3],a->status_count);atomic_store(&a->status[4],a->underflows);atomic_store(&a->status[5],a->last_flags);
atomic_store(&a->status[6],1U);atomic_store(&a->status[7],a->applied_seq);
atomic_store(&a->status[8],a->restart);atomic_store(&a->status[9],a->sound);
atomic_store(&a->status_seq,status_seq+2);
atomic_flag_clear(&a->rendering);return 0;
}
uintptr_t audio_callback_address(void) { return (uintptr_t)&audio_callback; }
int audio_status(struct audio *a,uint64_t *out) {
if(!a || !out) return 0;
for(unsigned attempt=0;attempt<3;attempt++) {
uint64_t seq=atomic_load(&a->status_seq);if(seq&1)continue;
for(unsigned i=0;i<10;i++)out[i]=atomic_load(&a->status[i]);
if(seq==atomic_load(&a->status_seq))return 1;
}
return 0;
}
// Caller MUST stop/close PortAudio before destroying this stream-owned mixer.
void audio_destroy(struct audio *a) { if(a) {free_bank(a->active);free_bank(atomic_load(&a->pending));free_bank(atomic_load(&a->retired));free(a);} }
openpilot/starpilot/audio/native/libaudio_mixer.so
Executable
BIN
View File
Binary file not shown.
openpilot/starpilot/audio/native_playback.py
+216
View File
@@ -0,0 +1,216 @@
"""Stream-owned native sample delivery; Python retains all alert and settings decisions."""
from __future__ import annotations
import ctypes
from pathlib import Path
import numpy as np
from openpilot.starpilot.audio.alert_volume import effective_volume
LIBRARY = Path(__file__).with_name('native') / 'libaudio_mixer.so'
# sounddevice marks its pointer closed even when Pa_CloseStream returns an error.
# Keep code, userdata and PCM alive until process exit if that fence fails.
_unclosed_streams = []
class NativePlayback:
def __init__(self, owner, sd, library=LIBRARY):
from openpilot.selfdrive.ui.soundd import ALERT_VOLUME_KEYS, AudibleAlert, SAMPLE_BUFFER, SAMPLE_RATE, STARPILOT_AUTO_MIN, sound_list
self.owner, self.sd = owner, sd
self.volume_keys, self.specs = ALERT_VOLUME_KEYS, sound_list
self.critical_alerts = (AudibleAlert.warningSoft, AudibleAlert.warningImmediate)
self.buffer, self.rate, self.auto_min = SAMPLE_BUFFER, SAMPLE_RATE, STARPILOT_AUTO_MIN
self.keys = tuple(sound_list)
if len(self.keys) > 16:
raise ValueError('Too many native sound slots')
self.slots = {key: i for i, key in enumerate(self.keys)}
self.lib = ctypes.CDLL(str(library))
self.lib.audio_mixer_abi.restype = ctypes.c_uint
if self.lib.audio_mixer_abi() != 3:
raise ValueError('Native audio ABI mismatch')
self.lib.audio_create.restype = ctypes.c_void_p
self.lib.audio_callback_address.restype = ctypes.c_size_t
self.lib.audio_can_publish.argtypes = [ctypes.c_void_p]
self.lib.audio_can_publish.restype = ctypes.c_int
self.lib.audio_collect.argtypes = [ctypes.c_void_p]
self.lib.audio_prepare.argtypes = [ctypes.c_void_p]
self.lib.audio_prepare.restype = ctypes.c_int
self.lib.audio_publish_bank.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.POINTER(ctypes.c_float)),
ctypes.POINTER(ctypes.c_uint), ctypes.c_uint, ctypes.c_uint64]
self.lib.audio_publish_bank.restype = ctypes.c_int
self.lib.audio_command.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint, ctypes.c_int,
ctypes.c_int, ctypes.c_float, ctypes.c_uint64, ctypes.c_uint64, ctypes.c_uint64]
self.lib.audio_command.restype = ctypes.c_uint64
self.lib.audio_status.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint64)]
self.lib.audio_status.restype = ctypes.c_int
self.lib.audio_destroy.argtypes = [ctypes.c_void_p]
self.handle = self.lib.audio_create()
if not self.handle:
raise MemoryError('Native audio allocation failed')
self._close_failed = False
self.signature = None
self.sounds = {}
self.builtin = {}
self.bank_sequence = 0
self.restart = self.command_sequence = 0
self.last_command = None
self.selection_sequence = 0
self.stop_frame = 0
self.previous_alert = self.previous_sound = None
self.status_count = 0
try:
if not self.publish():
raise ValueError('Native sounds were not admitted')
if not self.lib.audio_prepare(self.handle):
raise ValueError('Native initial bank was not admitted')
except BaseException:
self.close()
raise
def publish(self):
owner = self.owner
# Publish a newly selected safety alert using the last admitted bank before
# optional validation/copying of a replacement pack.
if self.sounds:
self.publish_command()
signature = tuple(id(owner.loaded_sounds[key]) for key in self.keys)
if signature != self.signature and self.lib.audio_can_publish(self.handle):
arrays = [np.ascontiguousarray(owner.loaded_sounds[key], dtype=np.float32) for key in self.keys]
if any(a.ndim != 1 or not 0 < len(a) <= 1_440_000 or not np.isfinite(a).all() for a in arrays):
raise ValueError('Invalid native waveform')
pointers = (ctypes.POINTER(ctypes.c_float) * len(arrays))(
*(a.ctypes.data_as(ctypes.POINTER(ctypes.c_float)) for a in arrays))
lengths = (ctypes.c_uint * len(arrays))(*(len(a) for a in arrays))
if self.lib.audio_publish_bank(self.handle, pointers, lengths, len(arrays), self.bank_sequence + 1):
self.bank_sequence += 1
self.signature = signature
self.builtin = {key: owner.pack_loader.is_builtin(owner.loaded_sounds[key]) for key in self.keys}
self.sounds = dict(owner.loaded_sounds)
if not self.sounds:
return False
self.lib.audio_collect(self.handle)
self.publish_command()
return True
def publish_command(self):
owner = self.owner
alert = owner.current_alert
sound = owner.current_sound
if (alert != self.previous_alert or sound != self.previous_sound) and owner.current_sound_frame == 0:
self.restart += 1
self.previous_alert, self.previous_sound = alert, sound
slot = self.slots.get(sound, 0)
loops = self.specs[alert][1] if alert else None
gain = owner.current_volume
samples = self.sounds.get(sound)
if samples is not None and self.builtin.get(sound, False):
gain = max(gain, self.auto_min)
key = self.volume_keys.get(alert)
if key is not None:
gain = effective_volume(key, owner.saved_volumes.get(key), gain,
immediate_ramp=owner.current_volume if alert in self.critical_alerts else None)
length = len(samples) if samples is not None else 1
if owner.pending_stop:
if not self.stop_frame or (self.last_command is not None and self.last_command[6] != self.bank_sequence):
self.stop_frame = (owner.current_sound_frame // length + 1) * length
else:
self.stop_frame = 0
command = (int(alert), slot, -1 if loops is None else loops, int(owner.pending_stop),
gain, self.restart, self.bank_sequence, self.stop_frame)
if command == self.last_command:
return
sequence = self.lib.audio_command(self.handle, *command)
if not sequence:
raise ValueError('Invalid native alert command')
self.command_sequence = sequence
if self.last_command is None or command[:4] + command[5:] != self.last_command[:4] + self.last_command[5:]:
self.selection_sequence = sequence
self.last_command = command
def snapshot(self):
values = (ctypes.c_uint64 * 10)()
return list(values) if self.lib.audio_status(self.handle, values) else None
def sync_owner(self):
value = self.snapshot()
if value is None or not value[6]:
return
owner = self.owner
# Volume-only publications must not hide cursor progress. A new selection
# epoch still prevents an old callback from overwriting a newer alert.
if value[8] == self.restart and value[9] == self.slots.get(owner.current_sound, 0):
owner.current_sound_frame = value[0]
if value[7] >= self.selection_sequence:
owner.current_alert, owner.pending_stop = value[1], bool(value[2])
# Native dismissal completion is a selection transition too.
self.previous_alert = owner.current_alert
if value[3] != self.status_count:
owner.pending_stream_status = self.sd.CallbackFlags(value[5])
self.status_count = value[3]
owner.stream_status_count, owner.output_underflow_count = value[3], value[4]
def stream(self):
# Public OutputStream/RawOutputStream wrap callbacks in Python; only the existing base CData path avoids the GIL.
callback = self.sd._ffi.cast('PaStreamCallback*', self.lib.audio_callback_address())
userdata = self.sd._ffi.cast('void*', self.handle)
base = self.sd._OutputStreamBase
stream = base.__new__(base)
wrapper = NativeStream(stream, self)
try:
base.__init__(stream, kind='output', wrap_callback=None, channels=1, dtype='float32',
samplerate=self.rate, blocksize=self.buffer, callback=callback, userdata=userdata)
except BaseException as constructor_error:
pointer = getattr(stream, '_ptr', self.sd._ffi.NULL)
# Before a successful open, sounddevice holds PaStream**; after open it
# holds PaStream*. Only the latter requires the explicit close fence.
if self.sd._ffi.typeof(pointer) == self.sd._ffi.typeof('PaStream*') and pointer:
try:
wrapper.close()
except BaseException as close_error:
constructor_error.add_note(f'Native audio close also failed: {close_error}')
raise
return wrapper
def close(self):
if self.handle and not self._close_failed:
self.lib.audio_destroy(self.handle)
self.handle = None
class NativeStream:
def __init__(self, stream, playback):
self.stream, self.playback = stream, playback
def __getattr__(self, name):
return getattr(self.stream, name)
def __enter__(self):
try:
self.stream.start()
except BaseException as start_error:
try:
self.close()
except Exception as close_error:
# The original start failure remains the caller-visible error; record
# the close failure on that exception without freeing stream memory.
start_error.add_note(f'Native audio close also failed: {close_error}')
raise
return self
def __exit__(self, *args):
try:
self.stream.stop(ignore_errors=False)
finally:
self.close()
return False
def close(self):
try:
self.stream.close(ignore_errors=False)
except BaseException:
self.playback._close_failed = True
_unclosed_streams.append(self)
raise
else:
self.playback.close()
openpilot/starpilot/audio/tests/test_stream_diagnostics.py
+22 -59
View File
@@ -14,11 +14,6 @@ def make_daemon():
daemon.output_underflow_count = 0
daemon.reported_output_underflows = 0
daemon.stream_report_at = None
daemon.last_callback_start_ns = None
daemon.last_callback_work_ns = None
daemon.last_native_audio_time = None
daemon.callback_tid = 42
daemon.underflow_timing = None
samples = np.array([0.5, -0.5, 0., 0.25], dtype=np.float32)
daemon.get_sound_data = Mock(side_effect=lambda frames: samples[:frames])
return daemon, samples
@@ -29,12 +24,8 @@ def test_underflow_counters_preserve_signal_and_log_on_service_thread():
output = np.empty((4, 1), dtype=np.float32)
underflow = SimpleNamespace(output_underflow=True)
other = SimpleNamespace(output_underflow=False)
clocks = [1_000_000_000, 1_001_000_000, 1_085_333_333, 1_087_333_333,
1_170_666_666, 1_171_666_666, 1_256_000_000, 1_257_000_000]
with patch.object(soundd.cloudlog, "warning") as warning, patch.object(soundd.cloudlog, "error") as error, \
patch.object(soundd, "monotonic_ns", side_effect=clocks), \
patch.object(soundd.time, "monotonic", return_value=2.), \
patch.object(soundd, "_callback_thread_snapshot", return_value={"tid": 42}) as snapshot:
patch.object(soundd.time, "monotonic", return_value=2.):
for status in (underflow, underflow, other, None):
daemon.callback(output, 4, None, status)
np.testing.assert_array_equal(output[:, 0], samples)
@@ -42,17 +33,10 @@ def test_underflow_counters_preserve_signal_and_log_on_service_thread():
error.assert_not_called()
assert daemon.stream_status_count == 3
assert daemon.output_underflow_count == 2
snapshot.assert_not_called()
daemon.log_pending_stream_status(SimpleNamespace(latency=0.2, cpu_load=0.01))
warning.assert_not_called()
error.assert_called_once_with(
f"soundd stream diagnostics: status={other} status_callbacks=3 output_underflows=2"
+ " underflow_callback_mono=1.085333333 callback_gap_ms=85.333 callback_work_ms=2.000"
+ " callback_period_ms=0.083 previous_callback_work_ms=1.000 native_current_time=unknown"
+ " native_output_dac_time=unknown native_callback_gap_ms=unknown callback_tid=42"
+ ' callback_thread_snapshot_current={"tid":42} snapshot_mono=2.000000000'
+ ' callback_scheduler_bracket={"available":false} latency=0.2 cpu_load=0.01')
snapshot.assert_called_once_with(42)
f"soundd stream diagnostics: status={other} status_callbacks=3 output_underflows=2")
daemon.log_pending_stream_status(SimpleNamespace(latency=0.2, cpu_load=0.01))
assert error.call_count == 1
@@ -96,77 +80,56 @@ def test_other_status_stays_a_warning_and_underflow_counter_survives_status_clea
assert "output_underflows=1" in error.call_args.args[0]
def test_late_callback_and_slow_work_are_distinct_without_changing_samples():
def test_status_callbacks_preserve_samples_and_defer_health_reports():
daemon, short_samples = make_daemon()
samples = np.tile(short_samples, 1024)
daemon.get_sound_data = Mock(side_effect=lambda frames: samples[:frames])
output = np.empty((4096, 1), dtype=np.float32)
underflow = SimpleNamespace(output_underflow=True)
clocks = [1_000_000_000, 1_001_000_000, 1_200_000_000, 1_201_000_000,
1_285_333_333, 1_385_333_333, 1_500_000_000, 1_501_000_000]
with patch.object(soundd, "monotonic_ns", side_effect=clocks), \
patch.object(soundd.time, "monotonic", side_effect=[0., 5.]), \
with patch.object(soundd.time, "monotonic", side_effect=[0., 5.]), \
patch("pathlib.Path.open", side_effect=AssertionError("callback file IO")), \
patch.object(soundd, "_callback_thread_snapshot", return_value={"tid": 42}) as snapshot, \
patch.object(soundd.cloudlog, "error") as error, patch.object(soundd.cloudlog, "warning") as warning:
for status, native in ((None, 10.), (underflow, 10.085333333)):
daemon.callback(output, 4096, SimpleNamespace(currentTime=native, outputBufferDacTime=native + .085333333), status)
for status in (None, underflow):
daemon.callback(output, 4096, None, status)
assert output[:, 0].tobytes() == samples.tobytes()
error.assert_not_called()
warning.assert_not_called()
snapshot.assert_not_called()
daemon.log_pending_stream_status()
assert "native_callback_gap_ms=85.333" in error.call_args.args[0]
assert "previous_callback_work_ms=1.000" in error.call_args.args[0]
assert "callback_gap_ms=200.000 callback_work_ms=1.000 callback_period_ms=85.333" in error.call_args.args[0]
for status, native in ((underflow, 10.170666666), (None, 10.256)):
daemon.callback(output, 4096, SimpleNamespace(currentTime=native, outputBufferDacTime=native + .085333333), status)
assert "output_underflows=1" in error.call_args.args[0]
for status in (underflow, None):
daemon.callback(output, 4096, None, status)
assert output[:, 0].tobytes() == samples.tobytes()
assert error.call_count == 1
warning.assert_not_called()
daemon.log_pending_stream_status()
assert error.call_count == 2
message = error.call_args.args[0]
assert "output_underflows=2 underflow_callback_mono=1.285333333" in message
assert "callback_gap_ms=85.333 callback_work_ms=100.000 callback_period_ms=85.333" in message
assert "status_callbacks=2 output_underflows=2" in error.call_args.args[0]
def test_native_clock_discontinuity_does_not_change_output_or_reuse_old_time():
def test_native_clock_metadata_does_not_change_output_or_health_counts():
daemon, samples = make_daemon()
output = np.empty((4, 1), dtype=np.float32)
underflow = SimpleNamespace(output_underflow=True)
with patch.object(soundd, "_callback_thread_snapshot", return_value={"tid": 42}), \
patch.object(soundd.time, "monotonic", side_effect=[0., 5., 10., 15., 20.]), patch.object(soundd.cloudlog, "error") as error:
for native in (10., float("nan"), 9., 8., -1.):
with patch.object(soundd.time, "monotonic", side_effect=[0., 5., 10., 15., 20.]), \
patch.object(soundd.cloudlog, "error") as error:
for count, native in enumerate((10., float("nan"), 9., 8., -1.), 1):
daemon.callback(output, 4, SimpleNamespace(currentTime=native, outputBufferDacTime=float("inf")), underflow)
assert output[:, 0].tobytes() == samples.tobytes()
assert daemon.output_underflow_count == count
daemon.log_pending_stream_status()
assert "native_callback_gap_ms=unknown" in error.call_args.args[0]
assert "native_output_dac_time=unknown" in error.call_args.args[0]
if native == 9.:
assert "native_current_time=9.000000000" in error.call_args.args[0]
elif native == 8.:
assert "native_current_time=8.000000000" in error.call_args.args[0]
assert "native_current_time=-1.000000000" in error.call_args.args[0]
assert f"output_underflows={count}" in error.call_args.args[0]
assert error.call_count == 5
def test_thread_snapshot_is_post_event_and_proc_failure_keeps_underflow_report(tmp_path):
fields = ["S"] + ["0"] * 36
fields[15], fields[16], fields[36] = "20", "0", "3"
(tmp_path / "stat").write_text("42 (callback thread) " + " ".join(fields))
(tmp_path / "schedstat").write_text("100 200 3")
with patch.object(soundd, "Path", return_value=tmp_path), \
patch.object(soundd.os, "sched_getscheduler", return_value=0, create=True), \
patch.object(soundd.os, "sched_getaffinity", return_value={1, 3}, create=True):
result = soundd._callback_thread_snapshot(42)
assert result == {"tid": 42, "state": "S", "priority": 20, "nice": 0, "processor": 3,
"runtime_ns": 100, "runqueue_ns": 200, "timeslices": 3, "policy": 0, "affinity": [1, 3]}
def test_proc_failure_does_not_affect_underflow_report_or_samples():
daemon, samples = make_daemon()
output = np.empty((4, 1), dtype=np.float32)
with patch("pathlib.Path.open", side_effect=FileNotFoundError), patch.object(soundd.cloudlog, "error") as error:
with patch("pathlib.Path.open", side_effect=FileNotFoundError) as file_open, \
patch.object(soundd.cloudlog, "error") as error:
daemon.callback(output, 4, None, SimpleNamespace(output_underflow=True))
assert output[:, 0].tobytes() == samples.tobytes()
error.assert_not_called()
daemon.log_pending_stream_status()
file_open.assert_not_called()
assert "output_underflows=1" in error.call_args.args[0]
assert 'callback_thread_snapshot_current={"tid":42,"unavailable":true}' in error.call_args.args[0]
prebuilt.json
+1 -1
View File
@@ -1 +1 @@
{"protected_inventory_sha256": "a93ce0e776ab0abe390ade88f83bed71fd98d9c71c2976bb29d7e660d44347e5", "schema_version": 1}
{"protected_inventory_sha256": "3f389147d8db51d535b549a991bd0ece6acd750d3d407a6884b075767a7f60d4", "schema_version": 1}