From 4fd747977a7d6ba8e8650cd314290d80e61c14b7 Mon Sep 17 00:00:00 2001 From: Vixea Date: Wed, 21 Oct 2026 21:58:53 -0500 Subject: [PATCH 1/2] feat(connection): add decoder configuration handling and notification mechanism --- alvr/client_core/src/connection.rs | 61 ++++++++++++++++++++++++------ alvr/client_core/src/lib.rs | 5 +-- alvr/packets/src/lib.rs | 3 +- alvr/server_core/src/lib.rs | 7 ++++ 4 files changed, 60 insertions(+), 16 deletions(-) diff --git a/alvr/client_core/src/connection.rs b/alvr/client_core/src/connection.rs index 2141a90d15..2568915cb5 100644 --- a/alvr/client_core/src/connection.rs +++ b/alvr/client_core/src/connection.rs @@ -15,8 +15,9 @@ use alvr_common::{ }; use alvr_packets::{ AUDIO, ClientConnectionResult, ClientControlPacket, ClientStatistics, ConnectionAcceptedInfo, - HAPTICS, Haptics, STATISTICS, ServerControlPacket, StreamConfigPacket, TRACKING, TrackingData, - VIDEO, VideoPacketHeader, VideoStreamingCapabilities, VideoStreamingCapabilitiesExt, + DecoderInitializationConfig, HAPTICS, Haptics, STATISTICS, ServerControlPacket, + StreamConfigPacket, TRACKING, TrackingData, VIDEO, VideoPacketHeader, + VideoStreamingCapabilities, VideoStreamingCapabilitiesExt, }; use alvr_session::{SocketProtocol, settings_schema::Switch}; use alvr_sockets::{ @@ -64,7 +65,9 @@ pub struct ConnectionContext { pub tracking_sender: Mutex>>, pub statistics_sender: Mutex>>, pub statistics_manager: Mutex>, + pub decoder_config: Mutex>, pub decoder_callback: Mutex>>, + pub decoder_callback_notifier: Condvar, pub video_frame_metadata_queue: Mutex>, pub max_prediction: RwLock, } @@ -82,6 +85,24 @@ fn set_hud_message(event_queue: &Mutex>, message: &str .push_back(ClientCoreEvent::UpdateHudMessage(message)); } +fn notify_decoder_config( + ctx: &ConnectionContext, + event_queue: &Mutex>, + config: DecoderInitializationConfig, +) { + let mut current_config = ctx.decoder_config.lock(); + if current_config.as_ref() != Some(&config) { + *current_config = Some(config.clone()); + *ctx.decoder_callback.lock() = None; + event_queue + .lock() + .push_back(ClientCoreEvent::DecoderConfig { + codec: config.codec, + config_nal: config.config_buffer, + }); + } +} + fn is_streaming(ctx: &ConnectionContext) -> bool { *ctx.state.read() == ConnectionState::Streaming } @@ -273,9 +294,12 @@ fn connection_pipeline( let statistics_sender = stream_socket.request_stream(STATISTICS); ctx.video_frame_metadata_queue.lock().clear(); + *ctx.decoder_config.lock() = None; + *ctx.decoder_callback.lock() = None; let video_receive_thread = thread::spawn({ let ctx = Arc::clone(&ctx); + let event_queue = Arc::clone(&event_queue); move || { let mut stream_corrupted = true; while is_streaming(&ctx) { @@ -292,6 +316,10 @@ fn connection_pipeline( stats.report_video_packet_received(header.timestamp); } + if let Some(config) = header.decoder_config { + notify_decoder_config(&ctx, &event_queue, config); + } + if header.is_idr { stream_corrupted = false; } else if data.had_packet_loss() { @@ -319,12 +347,25 @@ fn connection_pipeline( } } - let submitted = ctx - .decoder_callback - .lock() + let mut decoder_callback_guard = ctx.decoder_callback.lock(); + if decoder_callback_guard.is_none() && is_streaming(&ctx) { + let deadline = Instant::now() + STREAMING_RECV_TIMEOUT; + while decoder_callback_guard.is_none() && is_streaming(&ctx) { + let now = Instant::now(); + if now >= deadline { + break; + } + ctx.decoder_callback_notifier + .wait_for(&mut decoder_callback_guard, deadline - now); + } + } + + let submitted = decoder_callback_guard .as_mut() .is_some_and(|callback| callback(header.timestamp, nal)); + drop(decoder_callback_guard); + if !submitted { stream_corrupted = true; if let Some(sender) = &mut *ctx.control_sender.lock() { @@ -481,12 +522,7 @@ fn connection_pipeline( match maybe_packet { Ok(ServerControlPacket::DecoderConfig(config)) => { - event_queue - .lock() - .push_back(ClientCoreEvent::DecoderConfig { - codec: config.codec, - config_nal: config.config_buffer, - }); + notify_decoder_config(&ctx, &event_queue, config); } Ok(ServerControlPacket::Restarting) => { info!("{SERVER_RESTART_MESSAGE}"); @@ -570,6 +606,9 @@ fn connection_pipeline( *ctx.tracking_sender.lock() = None; *ctx.statistics_sender.lock() = None; *LOG_CHANNEL_SENDER.lock() = None; + *ctx.decoder_config.lock() = None; + *ctx.decoder_callback.lock() = None; + ctx.decoder_callback_notifier.notify_all(); event_queue .lock() diff --git a/alvr/client_core/src/lib.rs b/alvr/client_core/src/lib.rs index e87d416dba..fbc579d87d 100644 --- a/alvr/client_core/src/lib.rs +++ b/alvr/client_core/src/lib.rs @@ -262,10 +262,7 @@ impl ClientCoreContext { dbg_client_core!("set_decoder_input_callback"); *self.connection_context.decoder_callback.lock() = Some(callback); - - if let Some(sender) = &mut *self.connection_context.control_sender.lock() { - sender.send(&ClientControlPacket::RequestIdr).ok(); - } + self.connection_context.decoder_callback_notifier.notify_all(); } pub fn report_frame_decoded(&self, timestamp: Duration) { diff --git a/alvr/packets/src/lib.rs b/alvr/packets/src/lib.rs index 210d96a473..68f6a82a9f 100644 --- a/alvr/packets/src/lib.rs +++ b/alvr/packets/src/lib.rs @@ -149,7 +149,7 @@ impl StreamConfigPacket { } } -#[derive(Serialize, Deserialize, Clone)] +#[derive(Serialize, Deserialize, Clone, PartialEq, Eq, Debug)] pub struct DecoderInitializationConfig { pub codec: CodecType, pub config_buffer: Vec, // e.g. SPS + PPS NALs @@ -247,6 +247,7 @@ pub struct VideoPacketHeader { /// Centers used to encode this frame, already aligned. Normally absent when FFR is disabled. pub foveation_center_shifts: Option<[[f32; 2]; 2]>, pub is_idr: bool, + pub decoder_config: Option, } #[derive(Serialize, Deserialize)] diff --git a/alvr/server_core/src/lib.rs b/alvr/server_core/src/lib.rs index 9a2f931701..f38a1972b0 100644 --- a/alvr/server_core/src/lib.rs +++ b/alvr/server_core/src/lib.rs @@ -452,12 +452,19 @@ impl ServerCoreContext { file.write_all(&nal_buffer).ok(); } + let decoder_config = if is_idr { + self.connection_context.decoder_config.lock().clone() + } else { + None + }; + let sender_result = sender.try_send(VideoPacket { header: VideoPacketHeader { timestamp, global_view_params, foveation_center_shifts, is_idr, + decoder_config, }, payload: nal_buffer, }); From 3cd716617ec3685969b3e00847a2274713e736a9 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci-lite[bot]" <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com> Date: Sun, 25 Oct 2026 20:21:48 +0000 Subject: [PATCH 2/2] style: apply automatic formatting fixes --- alvr/client_core/src/lib.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/alvr/client_core/src/lib.rs b/alvr/client_core/src/lib.rs index fbc579d87d..d3fde1aa70 100644 --- a/alvr/client_core/src/lib.rs +++ b/alvr/client_core/src/lib.rs @@ -262,7 +262,9 @@ impl ClientCoreContext { dbg_client_core!("set_decoder_input_callback"); *self.connection_context.decoder_callback.lock() = Some(callback); - self.connection_context.decoder_callback_notifier.notify_all(); + self.connection_context + .decoder_callback_notifier + .notify_all(); } pub fn report_frame_decoded(&self, timestamp: Duration) {