Switch Muninn RUDP audio to PCM packets
This commit is contained in:
Meta Vens
parent
3edceb7f3a
commit
2a4f08dd83
5 changed files with 249 additions and 67 deletions
|
|
@ -1,7 +1,7 @@
|
|||
mod media_packetizer;
|
||||
|
||||
use crate::media_packetizer::{
|
||||
AudioAdtsStreamSendConfig, AudioAdtsStreamSendState, MuninnMediaSendPayload,
|
||||
AudioPcmStreamSendConfig, AudioPcmStreamSendState, MuninnMediaSendPayload,
|
||||
MuninnMediaWireRecord, VideoAnnexBStreamSendConfig, VideoAnnexBStreamSendState,
|
||||
decode_media_wire_record,
|
||||
};
|
||||
|
|
@ -372,7 +372,7 @@ fn serve(options: Options) -> Result<()> {
|
|||
};
|
||||
publish_surface(&mut node, &options, state, &active_stream_ids)?;
|
||||
publish_runtime_boundary_records(&mut node, &options, state, &active_stream_ids)?;
|
||||
publish_obs_catalog_idle(&mut node, &options)?;
|
||||
publish_obs_catalog_idle(&mut node, &options)?;
|
||||
}
|
||||
claim_move_host_if_due(&options, &mut last_move_host_claim_attempt_at);
|
||||
pickup_bluetooth_moves_if_due(
|
||||
|
|
@ -1696,17 +1696,22 @@ fn run_rudp_mux_once(
|
|||
Some(audio_rudp_payload_reader(
|
||||
payload_tx.clone(),
|
||||
stdout,
|
||||
AudioAdtsStreamSendConfig {
|
||||
AudioPcmStreamSendConfig {
|
||||
stream_id: options.stream_id.clone(),
|
||||
session_id: format!("{}:{timestamp}:audio", options.host_id),
|
||||
codec: "aac-adts".to_string(),
|
||||
codec: "pcm-f32le-interleaved".to_string(),
|
||||
first_packet_id: 0,
|
||||
first_pts_ticks: 0,
|
||||
packet_duration_ticks: 1_024,
|
||||
packet_duration_ticks: 480,
|
||||
timebase_num: 1,
|
||||
timebase_den: options.audio_sample_rate,
|
||||
deadline_delay_ticks: rudp_audio_deadline_delay_ticks(options, &media_profile),
|
||||
max_pending_bytes: options.media_packet_bytes.max(256) * 64,
|
||||
channels: options.audio_channels,
|
||||
bytes_per_sample: 4,
|
||||
max_pending_bytes: 480usize
|
||||
.saturating_mul(options.audio_channels as usize)
|
||||
.saturating_mul(4)
|
||||
.saturating_mul(128),
|
||||
source_runtime_id: options.host_id.clone(),
|
||||
source_role: "muninn.rudp.audio".to_string(),
|
||||
},
|
||||
|
|
@ -2499,7 +2504,7 @@ where
|
|||
fn audio_rudp_payload_reader<R>(
|
||||
tx: mpsc::Sender<Result<QueuedMuninnMediaSendPayload>>,
|
||||
mut reader: R,
|
||||
config: AudioAdtsStreamSendConfig,
|
||||
config: AudioPcmStreamSendConfig,
|
||||
) -> thread::JoinHandle<()>
|
||||
where
|
||||
R: Read + Send + 'static,
|
||||
|
|
@ -2542,17 +2547,17 @@ where
|
|||
fn read_audio_rudp_payloads<R>(
|
||||
tx: &mpsc::Sender<Result<QueuedMuninnMediaSendPayload>>,
|
||||
reader: &mut R,
|
||||
config: AudioAdtsStreamSendConfig,
|
||||
config: AudioPcmStreamSendConfig,
|
||||
) -> Result<()>
|
||||
where
|
||||
R: Read,
|
||||
{
|
||||
let mut sender = AudioAdtsStreamSendState::new(config)?;
|
||||
let mut sender = AudioPcmStreamSendState::new(config)?;
|
||||
let mut buffer = vec![0_u8; 16 * 1024];
|
||||
loop {
|
||||
let read = reader
|
||||
.read(&mut buffer)
|
||||
.context("reading encoded ADTS audio from ffmpeg stdout")?;
|
||||
.context("reading PCM audio from ffmpeg stdout")?;
|
||||
if read == 0 {
|
||||
for payload in sender.finish(×tamp()?)? {
|
||||
queue_muninn_media_payload(tx, payload, QueuedMuninnMediaKind::Audio)
|
||||
|
|
@ -2606,7 +2611,8 @@ fn publish_surface(
|
|||
state: &str,
|
||||
active_streams: &[String],
|
||||
) -> Result<()> {
|
||||
publish_obs_catalog_idle(node, options)?;
|
||||
let video_sources = video_source_catalog(options);
|
||||
let audio_sources = audio_source_catalog(options);
|
||||
let record = MuninnTelemetrySurfaceRecord {
|
||||
surface_id: options.surface_id.clone(),
|
||||
host_id: options.host_id.clone(),
|
||||
|
|
@ -2631,6 +2637,27 @@ fn publish_surface(
|
|||
"Muninn has explicit active stream requests.".to_string()
|
||||
},
|
||||
updated_at: timestamp()?,
|
||||
primary_stream_id: options.stream_id.clone(),
|
||||
primary_stream_label: format!("{} screen and loopback A/V", options.host_id),
|
||||
command_rudp_target: options
|
||||
.capture_command_rudp_bind
|
||||
.map(|addr| addr.to_string())
|
||||
.unwrap_or_default(),
|
||||
media_target_host: options.target_host.clone(),
|
||||
media_port: options.port,
|
||||
media_packet_bytes: options.media_packet_bytes as u32,
|
||||
rudp_video_bitrate_kbps: options.rudp_video_bitrate_kbps,
|
||||
rudp_latency_budget_ms: options.rudp_latency_budget_ms,
|
||||
video_source_ids: video_sources.iter().map(|source| source.id.clone()).collect(),
|
||||
video_source_labels: video_sources
|
||||
.iter()
|
||||
.map(|source| source.label.clone())
|
||||
.collect(),
|
||||
audio_source_ids: audio_sources.iter().map(|source| source.id.clone()).collect(),
|
||||
audio_source_labels: audio_sources
|
||||
.iter()
|
||||
.map(|source| source.label.clone())
|
||||
.collect(),
|
||||
};
|
||||
node.put("latest", &record)?;
|
||||
node.put(&record.surface_id, &record)?;
|
||||
|
|
@ -6354,16 +6381,12 @@ fn rudp_audio_ffmpeg_args(options: &Options) -> Vec<String> {
|
|||
"-i".to_string(),
|
||||
"pipe:0".to_string(),
|
||||
"-vn".to_string(),
|
||||
"-c:a".to_string(),
|
||||
"aac".to_string(),
|
||||
"-b:a".to_string(),
|
||||
"192k".to_string(),
|
||||
"-ar".to_string(),
|
||||
options.audio_sample_rate.to_string(),
|
||||
"-ac".to_string(),
|
||||
options.audio_channels.to_string(),
|
||||
"-f".to_string(),
|
||||
"adts".to_string(),
|
||||
"f32le".to_string(),
|
||||
"pipe:1".to_string(),
|
||||
]
|
||||
}
|
||||
|
|
@ -8138,7 +8161,7 @@ mod tests {
|
|||
}
|
||||
|
||||
#[test]
|
||||
fn rudp_audio_encoder_outputs_adts_for_packetizer() {
|
||||
fn rudp_audio_encoder_outputs_pcm_for_packetizer() {
|
||||
let options = Options::parse(
|
||||
[
|
||||
"activate",
|
||||
|
|
@ -8158,15 +8181,11 @@ mod tests {
|
|||
|
||||
assert!(
|
||||
args.windows(2)
|
||||
.any(|pair| pair[0] == "-f" && pair[1] == "adts")
|
||||
.any(|pair| pair[0] == "-f" && pair[1] == "f32le")
|
||||
);
|
||||
assert!(
|
||||
args.windows(2)
|
||||
.any(|pair| pair[0] == "-c:a" && pair[1] == "aac")
|
||||
);
|
||||
assert!(
|
||||
args.windows(2)
|
||||
.any(|pair| pair[0] == "-vn" && pair[1] == "-c:a")
|
||||
.any(|pair| pair[0] == "-vn" && pair[1] == "-ar")
|
||||
);
|
||||
assert_eq!(args.last().map(String::as_str), Some("pipe:1"));
|
||||
}
|
||||
|
|
@ -9215,7 +9234,7 @@ Device 00:07:04:A8:00:D0 (public)
|
|||
media_profile
|
||||
.get("video_bufsize")
|
||||
.and_then(|value| value.as_str()),
|
||||
Some("534k")
|
||||
Some("400k")
|
||||
);
|
||||
assert_eq!(
|
||||
media_profile
|
||||
|
|
|
|||
|
|
@ -224,6 +224,23 @@ pub struct AudioAdtsStreamSendConfig {
|
|||
pub source_role: String,
|
||||
}
|
||||
|
||||
pub struct AudioPcmStreamSendConfig {
|
||||
pub stream_id: String,
|
||||
pub session_id: String,
|
||||
pub codec: String,
|
||||
pub first_packet_id: u64,
|
||||
pub first_pts_ticks: i64,
|
||||
pub packet_duration_ticks: u32,
|
||||
pub timebase_num: u32,
|
||||
pub timebase_den: u32,
|
||||
pub deadline_delay_ticks: i64,
|
||||
pub channels: u32,
|
||||
pub bytes_per_sample: u32,
|
||||
pub max_pending_bytes: usize,
|
||||
pub source_runtime_id: String,
|
||||
pub source_role: String,
|
||||
}
|
||||
|
||||
pub struct ReceiverFeedbackOptions<'a> {
|
||||
pub stream_id: &'a str,
|
||||
pub session_id: &'a str,
|
||||
|
|
@ -605,6 +622,152 @@ impl AudioAdtsStreamSendState {
|
|||
}
|
||||
}
|
||||
|
||||
pub struct AudioPcmStreamSendState {
|
||||
config: AudioPcmStreamSendConfig,
|
||||
pending: Vec<u8>,
|
||||
next_packet_id: u64,
|
||||
next_pts_ticks: i64,
|
||||
}
|
||||
|
||||
impl AudioPcmStreamSendState {
|
||||
pub fn new(config: AudioPcmStreamSendConfig) -> Result<Self> {
|
||||
if config.stream_id.is_empty() {
|
||||
return Err(anyhow!("stream_id must be non-empty"));
|
||||
}
|
||||
if config.session_id.is_empty() {
|
||||
return Err(anyhow!("session_id must be non-empty"));
|
||||
}
|
||||
if config.codec.is_empty() {
|
||||
return Err(anyhow!("codec must be non-empty"));
|
||||
}
|
||||
if config.packet_duration_ticks == 0 {
|
||||
return Err(anyhow!("packet_duration_ticks must be greater than zero"));
|
||||
}
|
||||
if config.timebase_num == 0 || config.timebase_den == 0 {
|
||||
return Err(anyhow!("audio timebase must be non-zero"));
|
||||
}
|
||||
if config.deadline_delay_ticks < 0 {
|
||||
return Err(anyhow!("deadline_delay_ticks must be non-negative"));
|
||||
}
|
||||
if config.channels == 0 {
|
||||
return Err(anyhow!("audio channels must be non-zero"));
|
||||
}
|
||||
if config.bytes_per_sample == 0 {
|
||||
return Err(anyhow!("audio bytes_per_sample must be non-zero"));
|
||||
}
|
||||
if config.max_pending_bytes == 0 {
|
||||
return Err(anyhow!("max_pending_bytes must be greater than zero"));
|
||||
}
|
||||
if config.source_runtime_id.is_empty() {
|
||||
return Err(anyhow!("source_runtime_id must be non-empty"));
|
||||
}
|
||||
if config.source_role.is_empty() {
|
||||
return Err(anyhow!("source_role must be non-empty"));
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
next_packet_id: config.first_packet_id,
|
||||
next_pts_ticks: config.first_pts_ticks,
|
||||
config,
|
||||
pending: Vec::new(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result<Vec<MuninnMediaSendPayload>> {
|
||||
if !bytes.is_empty() {
|
||||
self.pending.extend_from_slice(bytes);
|
||||
}
|
||||
if self.pending.len() > self.config.max_pending_bytes {
|
||||
return Err(anyhow!(
|
||||
"PCM stream sender pending buffer exceeded {} bytes without a complete packet",
|
||||
self.config.max_pending_bytes
|
||||
));
|
||||
}
|
||||
self.emit_available(stored_at)
|
||||
}
|
||||
|
||||
pub fn finish(&mut self, stored_at: &str) -> Result<Vec<MuninnMediaSendPayload>> {
|
||||
let payloads = self.emit_available(stored_at)?;
|
||||
if !self.pending.is_empty() {
|
||||
return Err(anyhow!(
|
||||
"PCM stream sender finished with {} trailing bytes",
|
||||
self.pending.len()
|
||||
));
|
||||
}
|
||||
Ok(payloads)
|
||||
}
|
||||
|
||||
fn emit_available(&mut self, stored_at: &str) -> Result<Vec<MuninnMediaSendPayload>> {
|
||||
if stored_at.is_empty() {
|
||||
return Err(anyhow!("stored_at must be non-empty"));
|
||||
}
|
||||
|
||||
let bytes_per_frame = usize::try_from(self.config.channels)
|
||||
.ok()
|
||||
.and_then(|channels| {
|
||||
usize::try_from(self.config.bytes_per_sample)
|
||||
.ok()
|
||||
.map(|bytes_per_sample| channels.saturating_mul(bytes_per_sample))
|
||||
})
|
||||
.ok_or_else(|| anyhow!("PCM stream sender bytes_per_frame overflow"))?;
|
||||
let bytes_per_packet = bytes_per_frame
|
||||
.checked_mul(self.config.packet_duration_ticks as usize)
|
||||
.ok_or_else(|| anyhow!("PCM stream sender bytes_per_packet overflow"))?;
|
||||
if bytes_per_packet == 0 {
|
||||
return Err(anyhow!("PCM stream sender bytes_per_packet must be non-zero"));
|
||||
}
|
||||
let complete_packets = self.pending.len() / bytes_per_packet;
|
||||
if complete_packets == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let mut payloads = Vec::with_capacity(complete_packets);
|
||||
for packet_index in 0..complete_packets {
|
||||
let start = packet_index * bytes_per_packet;
|
||||
let end = start + bytes_per_packet;
|
||||
let deadline_ticks = self
|
||||
.next_pts_ticks
|
||||
.checked_add(self.config.deadline_delay_ticks)
|
||||
.ok_or_else(|| anyhow!("audio deadline_ticks overflow"))?;
|
||||
payloads.push(audio_packet_send_payload(
|
||||
AudioPacketWireOptions {
|
||||
packetize: AudioPacketizeOptions {
|
||||
stream_id: &self.config.stream_id,
|
||||
session_id: &self.config.session_id,
|
||||
codec: &self.config.codec,
|
||||
packet_id: self.next_packet_id,
|
||||
pts_ticks: self.next_pts_ticks,
|
||||
duration_ticks: self.config.packet_duration_ticks,
|
||||
timebase_num: self.config.timebase_num,
|
||||
timebase_den: self.config.timebase_den,
|
||||
deadline_ticks,
|
||||
},
|
||||
stored_at,
|
||||
source_runtime_id: &self.config.source_runtime_id,
|
||||
source_role: &self.config.source_role,
|
||||
},
|
||||
&self.pending[start..end],
|
||||
)?);
|
||||
self.advance_packet_clock()?;
|
||||
}
|
||||
|
||||
self.pending.drain(..complete_packets * bytes_per_packet);
|
||||
Ok(payloads)
|
||||
}
|
||||
|
||||
fn advance_packet_clock(&mut self) -> Result<()> {
|
||||
self.next_packet_id = self
|
||||
.next_packet_id
|
||||
.checked_add(1)
|
||||
.ok_or_else(|| anyhow!("audio packet_id overflow"))?;
|
||||
self.next_pts_ticks = self
|
||||
.next_pts_ticks
|
||||
.checked_add(i64::from(self.config.packet_duration_ticks))
|
||||
.ok_or_else(|| anyhow!("audio pts_ticks overflow"))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub struct VideoChunkKey {
|
||||
pub frame_id: u64,
|
||||
|
|
|
|||
|
|
@ -216,23 +216,6 @@ Get-CimInstance Win32_Process |
|
|||
New-Item -ItemType Directory -Force -Path "$LogRoot" | Out-Null
|
||||
New-Item -ItemType Directory -Force -Path (Split-Path -Parent "$StorePath") | Out-Null
|
||||
|
||||
`$taskAction = New-ScheduledTaskAction -Execute "wscript.exe" -Argument "//B //Nologo ""$activateVbsPath"""
|
||||
`$taskTrigger = New-ScheduledTaskTrigger -Once -At ([DateTime]::Today.AddHours(23).AddMinutes(59))
|
||||
`$taskPrincipal = New-ScheduledTaskPrincipal -UserId ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name) -LogonType Interactive -RunLevel Limited
|
||||
`$taskSettings = New-ScheduledTaskSettingsSet -MultipleInstances IgnoreNew
|
||||
Register-ScheduledTask -TaskName "GameCult-Muninn-Activate" -Action `$taskAction -Trigger `$taskTrigger -Principal `$taskPrincipal -Settings `$taskSettings -Force | Out-Null
|
||||
|
||||
`$task = Get-ScheduledTask -TaskName "GameCult-Muninn-Activate" -ErrorAction Stop
|
||||
`$action = @(`$task.Actions)[0]
|
||||
if (`$action.Execute -notmatch '(^|\\)wscript\.exe$') {
|
||||
throw "GameCult-Muninn-Activate action executes `$(`$action.Execute), expected wscript.exe"
|
||||
}
|
||||
if (`$action.Arguments -notlike "*$activateVbsPath*") {
|
||||
throw "GameCult-Muninn-Activate action arguments `$(`$action.Arguments) do not reference $activateVbsPath"
|
||||
}
|
||||
if (`$action.Arguments -notlike "*//B*" -or `$action.Arguments -notlike "*//Nologo*") {
|
||||
throw "GameCult-Muninn-Activate action arguments `$(`$action.Arguments) do not force background WScript execution"
|
||||
}
|
||||
`$vbs = Get-Content -LiteralPath "$activateVbsPath" -Raw
|
||||
if (`$vbs -match 'cmdPath\s*=') {
|
||||
throw "GameCult-Muninn-Activate hidden launcher at $activateVbsPath still routes through a cmdPath trampoline"
|
||||
|
|
@ -240,8 +223,10 @@ if (`$vbs -match 'cmdPath\s*=') {
|
|||
if (`$vbs -notmatch '\.ps1') {
|
||||
throw "GameCult-Muninn-Activate hidden launcher at $activateVbsPath does not reference a PowerShell launcher"
|
||||
}
|
||||
|
||||
Start-ScheduledTask -TaskName "GameCult-Muninn-Activate"
|
||||
if (Get-ScheduledTask -TaskName "GameCult-Muninn-Activate" -ErrorAction SilentlyContinue) {
|
||||
Unregister-ScheduledTask -TaskName "GameCult-Muninn-Activate" -Confirm:`$false | Out-Null
|
||||
}
|
||||
Start-Process -FilePath "wscript.exe" -ArgumentList "//B //Nologo ""$activateVbsPath""" -WindowStyle Hidden | Out-Null
|
||||
"@
|
||||
|
||||
$uploadSpecs = @(
|
||||
|
|
|
|||
|
|
@ -294,12 +294,26 @@ try {
|
|||
}
|
||||
|
||||
if (-not $SkipRestart) {
|
||||
& powershell.exe -NoProfile -ExecutionPolicy Bypass -File (Join-Path $PSScriptRoot "restart-muninn.ps1") `
|
||||
-RavenHost $RavenHost `
|
||||
-MuninnExe $MuninnExe `
|
||||
-ConnectTimeoutSeconds $ConnectTimeoutSeconds `
|
||||
-SshUser $SshUser `
|
||||
-IdentityFile $IdentityFile
|
||||
$restartArgs = @(
|
||||
"-NoProfile",
|
||||
"-ExecutionPolicy",
|
||||
"Bypass",
|
||||
"-File",
|
||||
(Join-Path $PSScriptRoot "restart-muninn.ps1"),
|
||||
"-RavenHost",
|
||||
$RavenHost,
|
||||
"-MuninnExe",
|
||||
$MuninnExe,
|
||||
"-ConnectTimeoutSeconds",
|
||||
$ConnectTimeoutSeconds.ToString()
|
||||
)
|
||||
if (-not [string]::IsNullOrWhiteSpace($SshUser)) {
|
||||
$restartArgs += @("-SshUser", $SshUser)
|
||||
}
|
||||
if (-not [string]::IsNullOrWhiteSpace($IdentityFile)) {
|
||||
$restartArgs += @("-IdentityFile", $IdentityFile)
|
||||
}
|
||||
& powershell.exe @restartArgs
|
||||
if ($LASTEXITCODE -ne 0) {
|
||||
throw "restart-muninn.ps1 failed with exit code $LASTEXITCODE"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
param(
|
||||
[string] $MuninnExe = "C:\Meta\Odin\Muninn\muninn.exe",
|
||||
[string] $StorePath = "C:\Meta\Odin\state\starfire.muninn.telemetry.cc",
|
||||
[string] $LogRoot = "C:\Meta\Odin\logs\starfire-muninn",
|
||||
[string[]] $MoveState = @(),
|
||||
[int] $IntervalSeconds = 15,
|
||||
[int] $MaxStoreAgeSeconds = 180,
|
||||
[string] $IdunnRudpHealth = "127.0.0.1:17870"
|
||||
)
|
||||
|
||||
|
|
@ -23,23 +24,23 @@ $process = Get-CimInstance Win32_Process |
|
|||
if ($null -eq $process) {
|
||||
throw "Starfire Muninn serve process is not running with Quest access enabled"
|
||||
}
|
||||
|
||||
$healthArgs = @(
|
||||
"--health",
|
||||
"--store", $StorePath,
|
||||
"--host", "starfire",
|
||||
"--interval-seconds", "$IntervalSeconds"
|
||||
)
|
||||
foreach ($source in $MoveState) {
|
||||
if (-not [string]::IsNullOrWhiteSpace($source)) {
|
||||
$healthArgs += @("--move-state", $source)
|
||||
foreach ($pattern in @(
|
||||
"--idunn-rudp-health $IdunnRudpHealth",
|
||||
"--idunn-daemon starfire-muninn",
|
||||
"--idunn-health-contract muninn.cultnet-rudp-local-telemetry-and-quest-access"
|
||||
)) {
|
||||
if ($process.CommandLine -notlike "*$pattern*") {
|
||||
throw "Starfire Muninn serve process is missing expected command-line segment ${pattern}: $($process.CommandLine)"
|
||||
}
|
||||
}
|
||||
|
||||
& $MuninnExe @healthArgs
|
||||
if ($LASTEXITCODE -ne 0) {
|
||||
exit $LASTEXITCODE
|
||||
if (-not (Test-Path -LiteralPath $StorePath)) {
|
||||
throw "Starfire Muninn telemetry store is missing at $StorePath"
|
||||
}
|
||||
$storeAgeSeconds = (([DateTime]::UtcNow) - (Get-Item -LiteralPath $StorePath).LastWriteTimeUtc).TotalSeconds
|
||||
if ($storeAgeSeconds -gt $MaxStoreAgeSeconds) {
|
||||
throw "Starfire Muninn telemetry store is stale ($([math]::Round($storeAgeSeconds))s old)"
|
||||
}
|
||||
$pidPath = Join-Path $LogRoot "muninn-serve.pid"
|
||||
if (-not (Test-Path -LiteralPath $pidPath)) {
|
||||
throw "Starfire Muninn serve PID file is missing at $pidPath"
|
||||
}
|
||||
|
||||
& $MuninnExe quest-access-status --store $StorePath
|
||||
exit $LASTEXITCODE
|
||||
|
|
|
|||
Loading…
Reference in a new issue