diff --git a/Cargo.lock b/Cargo.lock index 0e4f668ba5..72f93222cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -276,6 +276,7 @@ dependencies = [ "anyhow", "backtrace", "glam", + "libc", "log", "parking_lot", "pastey", diff --git a/alvr/client_core/src/connection.rs b/alvr/client_core/src/connection.rs index 2141a90d15..8e1d837c3c 100644 --- a/alvr/client_core/src/connection.rs +++ b/alvr/client_core/src/connection.rs @@ -9,14 +9,15 @@ use crate::{ }; use alvr_common::{ ALVR_VERSION, AnyhowToCon, ConResult, ConnectionError, ConnectionState, LifecycleState, - dbg_connection, debug, error, info, + TimeSyncEstimator, dbg_connection, debug, error, info, monotonic_ns, parking_lot::{Condvar, Mutex, RwLock}, wait_rwlock, warn, }; use alvr_packets::{ AUDIO, ClientConnectionResult, ClientControlPacket, ClientStatistics, ConnectionAcceptedInfo, - HAPTICS, Haptics, STATISTICS, ServerControlPacket, StreamConfigPacket, TRACKING, TrackingData, - VIDEO, VideoPacketHeader, VideoStreamingCapabilities, VideoStreamingCapabilitiesExt, + HAPTICS, Haptics, STATISTICS, ServerControlPacket, StreamConfigPacket, TRACKING, TimeSyncPacket, + TrackingData, VIDEO, VideoPacketHeader, VideoStreamingCapabilities, + VideoStreamingCapabilitiesExt, }; use alvr_session::{SocketProtocol, settings_schema::Switch}; use alvr_sockets::{ @@ -50,6 +51,7 @@ const SOCKET_INIT_RETRY_INTERVAL: Duration = Duration::from_millis(500); const CONNECTION_RETRY_INTERVAL: Duration = Duration::from_secs(1); const HANDSHAKE_ACTION_TIMEOUT: Duration = Duration::from_secs(2); const STREAMING_RECV_TIMEOUT: Duration = Duration::from_millis(500); +const TIME_SYNC_INTERVAL: Duration = Duration::from_secs(15); const MAX_UNREAD_PACKETS: usize = 10; // Applies per stream const VIDEO_FRAME_METADATA_HISTORY_SIZE: usize = 128; @@ -64,6 +66,7 @@ pub struct ConnectionContext { pub tracking_sender: Mutex>>, pub statistics_sender: Mutex>>, pub statistics_manager: Mutex>, + pub time_sync_estimator: Mutex, pub decoder_callback: Mutex>>, pub video_frame_metadata_queue: Mutex>, pub max_prediction: RwLock, @@ -426,6 +429,7 @@ fn connection_pipeline( let disconnect_notif = Arc::clone(&disconnect_notif); move || { let mut keepalive_deadline = Instant::now(); + let mut timesync_deadline = Instant::now(); #[cfg(target_os = "android")] let mut battery_deadline = Instant::now(); @@ -449,6 +453,22 @@ fn connection_pipeline( keepalive_deadline = Instant::now() + KEEPALIVE_INTERVAL; } + if Instant::now() > timesync_deadline + && let Some(sender) = &mut *ctx.control_sender.lock() + { + let offset = ctx.time_sync_estimator.lock().get_offset(); + sender + .send(&ClientControlPacket::TimeSync(TimeSyncPacket { + client_send_ns: monotonic_ns(), + server_recv_ns: 0, + server_send_ns: 0, + client_to_server_offset_ns: offset, + })) + .ok(); + + timesync_deadline = Instant::now() + TIME_SYNC_INTERVAL; + } + #[cfg(target_os = "android")] if Instant::now() > battery_deadline { let (gauge_value, is_plugged) = alvr_system_info::get_battery_status(); @@ -502,6 +522,48 @@ fn connection_pipeline( error!("Unexpected StartStream paceket"); } Ok(ServerControlPacket::KeepAlive) => (), + Ok(ServerControlPacket::TimeSync(packet)) => { + let client_recv_ns = monotonic_ns(); + let mut estimator = ctx.time_sync_estimator.lock(); + + let t1 = packet.client_send_ns as i128; + let t2 = packet.server_recv_ns as i128; + let t3 = packet.server_send_ns as i128; + let t4 = client_recv_ns as i128; + + let raw_sample_offset = (((t2 - t1) + (t3 - t4)) / 2) as i64; + let rtt_ns = (t4 - t1).saturating_sub(t3 - t2); + + // Observe drift BEFORE applying correction + if let Some(predicted_offset) = estimator.predict_offset_at(client_recv_ns) { + let drift_ns = raw_sample_offset - predicted_offset; + let drift_ppm = (drift_ns as f64 / 15e9) * 1e6; + + info!( + "[TIME_SYNC 15s] RTT: {:.2}ms | Measured: {:+.3}ms | Model Pred: {:+.3}ms | Drift over 15s: {:+.3}ms ({:+.1} ppm) | Learned Skew: {:+.1} ppm", + rtt_ns as f64 / 1e6, + raw_sample_offset as f64 / 1e6, + predicted_offset as f64 / 1e6, + drift_ns as f64 / 1e6, + drift_ppm, + estimator.get_skew_ppm(), + ); + } else { + info!( + "[TIME_SYNC Init] Baseline offset established: {:+.3}ms | RTT: {:.2}ms", + raw_sample_offset as f64 / 1e6, + rtt_ns as f64 / 1e6, + ); + } + + // Apply correction + estimator.push_sample( + packet.client_send_ns, + packet.server_recv_ns, + packet.server_send_ns, + client_recv_ns, + ); + } Ok( ServerControlPacket::Reserved(_) | ServerControlPacket::ReservedBuffer(_), ) => {} diff --git a/alvr/client_core/src/lib.rs b/alvr/client_core/src/lib.rs index e87d416dba..38b7c8c612 100644 --- a/alvr/client_core/src/lib.rs +++ b/alvr/client_core/src/lib.rs @@ -257,6 +257,17 @@ impl ClientCoreContext { } } + pub fn client_time_to_server_ns(&self, client_time_ns: u64) -> Option { + self.connection_context + .time_sync_estimator + .lock() + .client_to_server_ns(client_time_ns) + } + + pub fn get_time_sync_offset(&self) -> Option { + self.connection_context.time_sync_estimator.lock().get_offset() + } + /// The callback should return true if the frame was successfully submitted to the decoder pub fn set_decoder_input_callback(&self, callback: Box) { dbg_client_core!("set_decoder_input_callback"); diff --git a/alvr/common/Cargo.toml b/alvr/common/Cargo.toml index 304658f5bf..e561d9abb7 100644 --- a/alvr/common/Cargo.toml +++ b/alvr/common/Cargo.toml @@ -16,3 +16,6 @@ semver = { version = "1", features = ["serde"] } serde = { version = "1", features = ["derive"] } settings-schema = { git = "https://github.com/alvr-org/settings-schema-rs", rev = "676185f" } # settings-schema = { path = "../../../../settings-schema-rs/settings-schema" } + +[target.'cfg(unix)'.dependencies] +libc = "0.2" diff --git a/alvr/common/src/lib.rs b/alvr/common/src/lib.rs index 7ec96d4974..4b148165f6 100644 --- a/alvr/common/src/lib.rs +++ b/alvr/common/src/lib.rs @@ -4,6 +4,7 @@ mod connection_result; pub mod inputs; mod logging; mod primitives; +pub mod time; mod version; use parking_lot::{Condvar, Mutex, RwLockWriteGuard}; @@ -24,6 +25,7 @@ pub use inputs::*; pub use log::{debug, error, info, warn}; pub use logging::*; pub use primitives::*; +pub use time::*; pub use version::*; pub const ALVR_NAME: &str = "ALVR"; diff --git a/alvr/common/src/time.rs b/alvr/common/src/time.rs new file mode 100644 index 0000000000..cb9b280f4a --- /dev/null +++ b/alvr/common/src/time.rs @@ -0,0 +1,227 @@ +use std::collections::VecDeque; + +const WINDOW_SIZE: usize = 16; +const EMA_ALPHA: f64 = 0.1; +const SKEW_ALPHA: f64 = 0.05; +const MAX_SKEW_PPM: f64 = 200.0 * 1e-6; // Clamp to +/- 200 ppm + +pub fn monotonic_ns() -> u64 { + #[cfg(unix)] + { + let mut ts = libc::timespec { tv_sec: 0, tv_nsec: 0 }; + unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) }; + (ts.tv_sec as u64) * 1_000_000_000 + (ts.tv_nsec as u64) + } + #[cfg(windows)] + { + // On Windows: QueryPerformanceCounter + extern "system" { + fn QueryPerformanceCounter(lpPerformanceCount: *mut i64) -> i32; + fn QueryPerformanceFrequency(lpFrequency: *mut i64) -> i32; + } + static FREQ: std::sync::LazyLock = std::sync::LazyLock::new(|| { + let mut freq = 0; + unsafe { QueryPerformanceFrequency(&mut freq) }; + freq + }); + let mut count = 0; + unsafe { QueryPerformanceCounter(&mut count) }; + ((count as u128 * 1_000_000_000) / (*FREQ as u128)) as u64 + } + } + + +struct Sample { + rtt_ns: i128, + offset_ns: i64, + client_time_ns: u64, +} +pub struct TimeSyncEstimator { + samples: VecDeque, + base_client_time_ns: u64, + base_offset_ns: Option, + skew: f64, // ppm ratio + last_skew_client_time_ns: u64, + last_skew_offset_ns: f64, +} + +impl Default for TimeSyncEstimator { + fn default() -> Self { + Self::new() + } +} + +impl TimeSyncEstimator { + pub fn new() -> Self { + Self { + samples: VecDeque::with_capacity(WINDOW_SIZE), + base_client_time_ns: 0, + base_offset_ns: None, + skew: 0.0, + last_skew_client_time_ns: 0, + last_skew_offset_ns: 0.0, + } + } + + pub fn push_sample(&mut self, client_send_ns: u64, server_recv_ns: u64, server_send_ns: u64, client_recv_ns: u64) { + let rtt_ns = (client_recv_ns as i128 - client_send_ns as i128) - (server_send_ns as i128 - server_recv_ns as i128); + let sample_offset = (((server_recv_ns as i128 - client_send_ns as i128) + (server_send_ns as i128 - client_recv_ns as i128)) / 2) as i64; + + if self.samples.len() >= WINDOW_SIZE { + self.samples.pop_front(); + } + self.samples.push_back(Sample { rtt_ns, offset_ns: sample_offset, client_time_ns: client_send_ns }); + + if let Some(best) = self.samples.iter().min_by_key(|s| s.rtt_ns) { + match self.base_offset_ns { + None => { + self.base_offset_ns = Some(best.offset_ns as f64); + self.base_client_time_ns = best.client_time_ns; + self.last_skew_client_time_ns = best.client_time_ns; + self.last_skew_offset_ns = best.offset_ns as f64; + } + Some(prev_offset) => { + let dt = (best.client_time_ns as f64) - (self.base_client_time_ns as f64); + // Project previous offset forward by current skew rate + let projected_offset = prev_offset + self.skew * dt; + // Blend residual error with EMA + let updated_offset = projected_offset + EMA_ALPHA * ((best.offset_ns as f64) - projected_offset); + + // Skew estimation over multi-second baseline + let skew_dt = best.client_time_ns.saturating_sub(self.last_skew_client_time_ns); + if skew_dt >= 2_000_000_000 { + let raw_skew = ((best.offset_ns as f64) - self.last_skew_offset_ns) / (skew_dt as f64); + let clamped_skew = raw_skew.clamp(-MAX_SKEW_PPM, MAX_SKEW_PPM); + if self.skew == 0.0 { + self.skew = clamped_skew; + } else { + self.skew = SKEW_ALPHA * clamped_skew + (1.0 - SKEW_ALPHA) * self.skew; + } + self.last_skew_client_time_ns = best.client_time_ns; + self.last_skew_offset_ns = best.offset_ns as f64; + } + + self.base_offset_ns = Some(updated_offset); + self.base_client_time_ns = best.client_time_ns; + } + } + } + } + + pub fn client_to_server_ns(&self, client_time_ns: u64) -> Option { + self.base_offset_ns.map(|theta_0| { + let dt = client_time_ns as f64 - self.base_client_time_ns as f64; + let current_offset = theta_0 + self.skew * dt; + client_time_ns.saturating_add_signed(current_offset.round() as i64) + }) + } + + pub fn get_offset(&self) -> Option { + self.base_offset_ns.map(|o| o.round() as i64) + } + + /// + pub fn get_skew_ppm(&self) -> f64 { + self.skew * 1e6 + } + + /// Predicts what the offset should be at `client_time_ns` based on current base offset and skew + /// For allowing extrapolation of the offset of the server clock without having to ping the server(debugging/tooling) + pub fn predict_offset_at(&self, client_time_ns: u64) -> Option { + self.base_offset_ns.map(|theta_0| { + let dt = client_time_ns as f64 - self.base_client_time_ns as f64; + (theta_0 + self.skew * dt).round() as i64 + }) + } +} + +#[test] +fn test_time_sync_ground_truth() { + let mut estimator = TimeSyncEstimator::new(); + let true_offset: i128 = 15_000_000_000; // Server is 15s ahead + let one_way_delay: i128 = 2_000_000; // 2ms network flight time + let server_hold: i128 = 500_000; // 0.5ms turnaround + + for i in 0..16 { + let t1 = 1_000_000_000 + i * 200_000_000; // Ping every 200ms + let t2 = t1 + true_offset + one_way_delay; + let t3 = t2 + server_hold; + let t4 = t3 - true_offset + one_way_delay; + + estimator.push_sample(t1 as u64, t2 as u64, t3 as u64, t4 as u64); + } + + let estimated_offset = estimator.get_offset().unwrap(); + // Must converge to within 1 microsecond of ground truth + assert!((estimated_offset - true_offset as i64).abs() < 1_000); +} + +#[test] +fn test_time_sync_negative_offset() { + let mut estimator = TimeSyncEstimator::new(); + // Client booted 24h ago, Server booted 1h ago: offset is -23 hours + let true_offset: i128 = -23 * 3600 * 1_000_000_000; + let delay: i128 = 3_000_000; // 3ms + + let t1 = 86_400_000_000_000u64; // 24 hours + let t2 = (t1 as i128 + true_offset + delay) as u64; + let t3 = t2 + 100_000; + let t4 = (t3 as i128 - true_offset + delay) as u64; + + estimator.push_sample(t1, t2, t3, t4); + assert_eq!(estimator.get_offset().unwrap(), true_offset as i64); +} + +#[test] +fn test_wi_fi_spike_rejection() { + let mut estimator = TimeSyncEstimator::new(); + let true_offset: i128 = 5_000_000_000; + + for i in 0..20 { + let t1 = 10_000_000_000 + i * 200_000_000; + // Inject 60ms spikes on samples 5 and 12, 2ms baseline otherwise + let delay = if i == 5 || i == 12 { 30_000_000 } else { 1_000_000 }; + let t2 = t1 + true_offset + delay; + let t3 = t2 + 200_000; + let t4 = t3 - true_offset + delay; + + estimator.push_sample(t1 as u64, t2 as u64, t3 as u64, t4 as u64); + } + + let diff = (estimator.get_offset().unwrap() - true_offset as i64).abs(); + // Offset must not get pulled by the 60ms spikes (error must stay under 0.1ms) + assert!(diff < 100_000, "Spike leaked into estimator! Error: {diff}ns"); +} + +#[test] +fn test_time_sync_skew_correction() { + let mut estimator = TimeSyncEstimator::new(); + let initial_offset: i128 = 10_000_000_000; // 10s + let skew_ppm: f64 = 40.0; // Server ticks 40 ppm faster + let skew_ratio = skew_ppm * 1e-6; + let delay: i128 = 1_500_000; // 1.5ms one-way + + // Simulate 30 pings every 500ms (15 seconds total) + for i in 0..30 { + let t1 = 1_000_000_000 + i * 500_000_000; + let drift = (t1 as f64 * skew_ratio) as i128; + let current_true_offset = initial_offset + drift; + + let t2 = t1 + current_true_offset + delay; + let t3 = t2 + 100_000; + let t4 = t3 - current_true_offset + delay; + + estimator.push_sample(t1 as u64, t2 as u64, t3 as u64, t4 as u64); + } + + // Extrapolate 3 seconds into the future without any new pings + let future_client_time = 1_000_000_000u64 + 18 * 1_000_000_000u64; + let future_true_offset = initial_offset + (future_client_time as f64 * skew_ratio) as i128; + let true_server_time = (future_client_time as i128 + future_true_offset) as u64; + + let predicted_server_time = estimator.client_to_server_ns(future_client_time as u64).unwrap(); + let error_ns = (predicted_server_time as i128 - true_server_time as i128).abs(); + + // With skew correction, error stays under 50 microseconds (0.05ms) even when extrapolating! + assert!(error_ns < 50_000, "Error was too large: {error_ns}ns"); +} \ No newline at end of file diff --git a/alvr/packets/src/lib.rs b/alvr/packets/src/lib.rs index 210d96a473..cfe76929ff 100644 --- a/alvr/packets/src/lib.rs +++ b/alvr/packets/src/lib.rs @@ -31,6 +31,14 @@ pub struct VideoStreamingCapabilitiesExt { // Nothing for now } +#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)] +pub struct TimeSyncPacket { + pub client_send_ns: u64, + pub server_recv_ns: u64, + pub server_send_ns: u64, + pub client_to_server_offset_ns: Option, +} + #[derive(Serialize, Deserialize, Clone)] pub struct VideoStreamingCapabilities { pub default_view_resolution: UVec2, @@ -163,6 +171,7 @@ pub enum ServerControlPacket { Restarting, KeepAlive, RealTimeConfig(RealTimeConfig), + TimeSync(TimeSyncPacket), Reserved(String), ReservedBuffer(Vec), } @@ -205,6 +214,7 @@ pub enum ClientControlPacket { message: String, }, ProximityState(bool), + TimeSync(TimeSyncPacket), Reserved(String), ReservedBuffer(Vec), } diff --git a/alvr/server_core/src/connection.rs b/alvr/server_core/src/connection.rs index 9a68b5350c..970adbcdac 100644 --- a/alvr/server_core/src/connection.rs +++ b/alvr/server_core/src/connection.rs @@ -38,7 +38,7 @@ use std::{ hash::{Hash, Hasher}, net::{IpAddr, Ipv4Addr}, process::Command, - sync::{Arc, mpsc::RecvTimeoutError}, + sync::{Arc, atomic::Ordering, mpsc::RecvTimeoutError}, thread, time::{Duration, Instant}, }; @@ -1376,6 +1376,17 @@ fn connection_pipeline( .send(ServerCoreEvent::ProximityState(headset_is_worn)) .ok(); } + ClientControlPacket::TimeSync(mut packet) => { + packet.server_recv_ns = alvr_common::time::monotonic_ns(); + + // Store the client's converged offset on the server + if let Some(offset) = packet.client_to_server_offset_ns { + ctx.client_to_server_offset_ns.store(offset, Ordering::Relaxed); + } + + packet.server_send_ns = alvr_common::time::monotonic_ns(); + control_sender.lock().send(&ServerControlPacket::TimeSync(packet)).ok(); + } ClientControlPacket::Reserved(_) | ClientControlPacket::ReservedBuffer(_) => (), } diff --git a/alvr/server_core/src/lib.rs b/alvr/server_core/src/lib.rs index 9a2f931701..70076fa999 100644 --- a/alvr/server_core/src/lib.rs +++ b/alvr/server_core/src/lib.rs @@ -43,7 +43,7 @@ use std::{ io::Write, sync::{ Arc, LazyLock, OnceLock, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicI64, Ordering}, mpsc::{self, SyncSender, TrySendError}, }, thread::{self, JoinHandle}, @@ -116,6 +116,7 @@ pub struct ConnectionContext { clients_to_be_removed: Mutex>, video_channel_sender: Mutex>>, haptics_sender: Mutex>>, + pub client_to_server_offset_ns: AtomicI64, } pub fn create_recording_file(connection_context: &ConnectionContext, settings: &Settings) { @@ -234,6 +235,7 @@ impl ServerCoreContext { clients_to_be_removed: Mutex::new(HashSet::new()), video_channel_sender: Mutex::new(None), haptics_sender: Mutex::new(None), + client_to_server_offset_ns: AtomicI64::new(i64::MIN), }); let webserver_runtime = Runtime::new().unwrap(); @@ -540,6 +542,16 @@ impl ServerCoreContext { .as_mut() .map(|stats| stats.duration_until_next_vsync()) } + + pub fn client_time_to_server_ns(&self, client_time_ns: u64) -> Option { + let offset = self.connection_context.client_to_server_offset_ns.load(Ordering::Relaxed); + (offset != i64::MIN).then(|| client_time_ns.saturating_add_signed(offset)) + } + + pub fn get_time_sync_offset(&self) -> Option { + let offset = self.connection_context.client_to_server_offset_ns.load(Ordering::Relaxed); + (offset != i64::MIN).then_some(offset) + } } impl Drop for ServerCoreContext {