//! Frame pacing and a hysteretic, configurable thermal load limit. //! Times are monotonic microseconds so the scheduling policy is host-testable. pub struct Pacer { period_us: u64, next_due_us: u64, } impl Pacer { pub fn new(fps: u32) -> Self { Self::with_gap_us(period(fps)) } pub fn with_gap_us(period_us: u64) -> Self { Self { period_us, next_due_us: 0, } } pub fn delay_us(&self, now_us: u64) -> u64 { self.next_due_us.saturating_sub(now_us) } pub fn started(&mut self, now_us: u64) { // An overrun starts a new period; never burst to repay missed slots. self.next_due_us = now_us.saturating_add(self.period_us); } pub fn set_fps(&mut self, fps: u32, now_us: u64) { self.period_us = period(fps); self.started(now_us); } } pub const MAX_SEND_HOLD_US: u64 = 25_000; pub const MAX_CAPTURE_AGE_US: u64 = 100_000; pub const DEFAULT_RETRY_BUDGET_US: u64 = 8_000; /// One current frame, no pending-send FIFO. A short retry budget is shared by /// all fragments/peers; success never renews the frame's absolute deadline. pub struct FrameBudget { deadline_us: u64, retry_left_us: u64, } impl FrameBudget { pub fn new(now_us: u64, captured_us: u64, retry_us: u64) -> Self { Self { deadline_us: now_us .saturating_add(MAX_SEND_HOLD_US) .min(captured_us.min(now_us).saturating_add(MAX_CAPTURE_AGE_US)), retry_left_us: retry_us.min(MAX_SEND_HOLD_US), } } pub fn expired(&self, now_us: u64) -> bool { now_us >= self.deadline_us } pub fn allow_retry(&mut self, now_us: u64, wait_us: u64) -> bool { if wait_us == 0 || wait_us > self.retry_left_us || now_us.saturating_add(wait_us) >= self.deadline_us { return false; } self.retry_left_us -= wait_us; true } } fn period(fps: u32) -> u64 { if fps == 0 { 0 } else { 1_000_000u64.div_ceil(fps as u64) } } /// Adjust JPEG compression from five-second submission windows. The sensor's /// quality number increases with compression; restore detail more slowly than /// we shed load, and do not interpret absent subscribers as a healthy link. pub struct QualityController { base: u8, ceiling: u8, current: u8, healthy_windows: u8, } impl QualityController { pub fn new(base: u8, ceiling: u8) -> Self { Self { base, ceiling, current: base, healthy_windows: 0, } } pub fn observe(&mut self, peer_frames: u64, dropped: u32, send_ms: f32) -> Option { if peer_frames < 5 || !send_ms.is_finite() { self.healthy_windows = 0; return None; } let previous = self.current; if u64::from(dropped) * 100 >= peer_frames.saturating_mul(2) || send_ms >= 8.0 { self.current = self.current.saturating_add(3).min(self.ceiling); self.healthy_windows = 0; } else if dropped == 0 && send_ms < 4.0 { self.healthy_windows += 1; if self.healthy_windows >= 3 { self.current = self.current.saturating_sub(1).max(self.base); self.healthy_windows = 0; } } else { self.healthy_windows = 0; } (self.current != previous).then_some(self.current) } } pub struct ThermalLimiter { normal_fps: u32, limit_c: f32, limited: bool, } impl ThermalLimiter { pub fn new(normal_fps: u32, limit_c: f32) -> Self { Self { normal_fps, limit_c, limited: false, } } pub fn observe(&mut self, temperature: Option) -> u32 { if self.limit_c > 0.0 { if let Some(t) = temperature.filter(|t| t.is_finite()) { if t >= self.limit_c { self.limited = true; } else if t <= self.limit_c - 5.0 { self.limited = false; } } } self.fps() } pub fn fps(&self) -> u32 { if !self.limited { self.normal_fps } else if self.normal_fps == 0 { 8 } else { self.normal_fps.div_ceil(2) } } pub fn limited(&self) -> bool { self.limited } } #[cfg(test)] mod tests { use super::*; #[test] fn pacing_caps_rate_without_catching_up_after_an_overrun() { let mut pacer = Pacer::new(15); assert_eq!(pacer.delay_us(100), 0); pacer.started(100); assert_eq!(pacer.delay_us(101), 66_666); assert_eq!(pacer.delay_us(300_000), 0); pacer.started(300_000); assert_eq!(pacer.delay_us(300_001), 66_666); } #[test] fn unlimited_and_runtime_rate_changes_do_not_burst() { let mut pacer = Pacer::new(0); pacer.started(1_000); assert_eq!(pacer.delay_us(1_001), 0); pacer.set_fps(10, 1_001); assert_eq!(pacer.delay_us(1_002), 99_999); } #[test] fn thermal_limit_has_five_degree_hysteresis() { let mut limit = ThermalLimiter::new(15, 75.0); assert_eq!(limit.observe(Some(74.9)), 15); assert_eq!(limit.observe(Some(75.0)), 8); assert_eq!(limit.observe(Some(72.0)), 8); assert_eq!(limit.observe(None), 8); assert_eq!(limit.observe(Some(f32::NAN)), 8); assert_eq!(limit.observe(Some(70.0)), 15); } #[test] fn thermal_guard_can_be_disabled_or_used_with_uncapped_streams() { assert_eq!(ThermalLimiter::new(15, 0.0).observe(Some(90.0)), 15); assert_eq!(ThermalLimiter::new(0, 75.0).observe(Some(80.0)), 8); assert_eq!(ThermalLimiter::new(1, 75.0).observe(Some(80.0)), 1); } #[test] fn packet_spacing_starts_from_actual_submission_without_catch_up() { let mut pacer = Pacer::with_gap_us(1000); pacer.started(100); assert_eq!(pacer.delay_us(900), 200); assert_eq!(pacer.delay_us(5000), 0); pacer.started(5000); assert_eq!(pacer.delay_us(5001), 999); } #[test] fn frame_deadline_is_not_renewed_by_successful_fragments_or_other_peers() { let frame = FrameBudget::new(50_000, 20_000, 8_000); assert!(!frame.expired(74_999)); assert!(frame.expired(75_000)); } #[test] fn an_old_capture_cannot_gain_a_fresh_send_lifetime() { let frame = FrameBudget::new(99_000, 0, 8_000); assert!(!frame.expired(99_999)); assert!(frame.expired(100_000)); assert!(FrameBudget::new(100_001, 0, 8_000).expired(100_001)); } #[test] fn retry_allowance_is_shared_by_all_fragments_and_stops_before_deadline() { let mut frame = FrameBudget::new(50_000, 20_000, 2_000); assert!(frame.allow_retry(51_000, 1000)); assert!(frame.allow_retry(52_000, 1000)); assert!(!frame.allow_retry(53_000, 1000)); let mut late = FrameBudget::new(50_000, 20_000, 8_000); assert!(!late.allow_retry(74_000, 1000)); } #[test] fn zero_retry_mode_still_allows_immediate_sends_and_abandons_backpressure() { let mut frame = FrameBudget::new(50_000, 20_000, 0); assert!(!frame.expired(50_001)); assert!(!frame.allow_retry(50_001, 1000)); } #[test] fn compression_sheds_congestion_load_quickly_and_stays_in_bounds() { let mut control = QualityController::new(18, 24); assert_eq!(control.observe(100, 2, 3.0), Some(21)); assert_eq!(control.observe(100, 0, 8.0), Some(24)); assert_eq!(control.observe(100, 20, 20.0), None); let mut disabled = QualityController::new(18, 18); assert_eq!(disabled.observe(100, 50, 20.0), None); } #[test] fn compression_recovers_only_after_three_consecutive_healthy_windows() { let mut control = QualityController::new(18, 24); control.observe(100, 10, 10.0); assert_eq!(control.observe(100, 0, 2.0), None); assert_eq!(control.observe(100, 0, 2.0), None); assert_eq!(control.observe(100, 1, 5.0), None); assert_eq!(control.observe(100, 0, 2.0), None); assert_eq!(control.observe(100, 0, 2.0), None); assert_eq!(control.observe(100, 0, 2.0), Some(20)); for _ in 0..20 { control.observe(100, 0, 2.0); } assert_eq!(control.current, 18); } #[test] fn missing_traffic_or_invalid_timing_does_not_restore_quality() { let mut control = QualityController::new(18, 24); control.observe(100, 10, 10.0); control.observe(100, 0, 2.0); control.observe(100, 0, 2.0); assert_eq!(control.observe(0, 0, 0.0), None); assert_eq!(control.observe(100, 0, f32::NAN), None); assert_eq!(control.observe(100, 0, 2.0), None); assert_eq!(control.current, 21); } }