diff --git a/src/lib.rs b/src/lib.rs index 6ca8ba1..d1c03cb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,4 +3,5 @@ pub mod config; pub mod hyperhdr; +pub mod pacing; pub mod slot; diff --git a/src/pacing.rs b/src/pacing.rs new file mode 100644 index 0000000..9ec02f2 --- /dev/null +++ b/src/pacing.rs @@ -0,0 +1,380 @@ +//! Pure scheduling logic for the HyperHDR connection: send pacing, reconnect backoff +//! and log rate limiting. Nothing here does I/O or reads the clock; `now` is injected. + +use std::time::{Duration, Instant}; + +/// What the network thread should do next. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Action { + /// Send the newest frame, which has not been sent yet. + SendFrame, + /// Re-send the last frame so the server does not time out the connection. + SendKeepalive, + /// Clear our priority: no new frames for `idle_clear`. + Clear, + /// Keep the connection open while cleared, without showing an image. + Heartbeat, + /// Nothing to do until the given instant, or until the next event if `None`. + Wait(Option), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PacingConfig { + /// `1 / max_fps`. + pub min_interval: Duration, + pub keepalive: Duration, + pub idle_clear: Option, +} + +/// Send rules (spec 4.3.6), evaluated as a function of `now` and the recorded events. +#[derive(Debug)] +pub struct Pacer { + cfg: PacingConfig, + has_frame: bool, + unsent: bool, + cleared: bool, + last_frame_at: Option, + last_image_at: Option, + last_message_at: Option, +} + +impl Pacer { + pub fn new(cfg: PacingConfig) -> Self { + Self { + cfg, + has_frame: false, + unsent: false, + cleared: false, + last_frame_at: None, + last_image_at: None, + last_message_at: None, + } + } + + /// A new frame arrived. Returns `true` if it replaces a frame that was never sent. + pub fn on_new_frame(&mut self, now: Instant) -> bool { + let superseded = self.unsent; + self.has_frame = true; + self.unsent = true; + self.cleared = false; + self.last_frame_at = Some(now); + superseded + } + + /// The source disappeared: forget the last frame. + pub fn on_source_gone(&mut self) { + self.has_frame = false; + self.unsent = false; + self.cleared = false; + self.last_frame_at = None; + } + + /// A new connection was established (and registered). + pub fn on_connected(&mut self) { + self.last_image_at = None; + self.last_message_at = None; + if self.has_frame && !self.cleared { + self.unsent = true; + } + } + + /// `action` was written successfully at `now`. + pub fn on_sent(&mut self, action: Action, now: Instant) { + match action { + Action::SendFrame | Action::SendKeepalive => { + self.unsent = false; + self.last_image_at = Some(now); + self.last_message_at = Some(now); + } + Action::Clear => { + self.cleared = true; + self.last_message_at = Some(now); + } + Action::Heartbeat => self.last_message_at = Some(now), + Action::Wait(_) => {} + } + } + + /// Whether a connection to HyperHDR should exist while disconnected. + /// Marks the source as cleared if the idle timeout expired in the meantime. + pub fn wants_connection(&mut self, now: Instant) -> bool { + if self.idle_expired(now) { + self.cleared = true; + self.unsent = false; + } + self.has_frame && !self.cleared + } + + pub fn is_cleared(&self) -> bool { + self.cleared + } + + fn idle_expired(&self, now: Instant) -> bool { + match (self.cfg.idle_clear, self.last_frame_at) { + (Some(idle), Some(at)) => !self.unsent && now >= at + idle, + _ => false, + } + } + + /// Next action for a connected client. + pub fn poll(&self, now: Instant) -> Action { + let image_allowed_at = self.last_image_at.map(|t| t + self.cfg.min_interval); + + if self.unsent { + return match image_allowed_at { + Some(at) if now < at => Action::Wait(Some(at)), + _ => Action::SendFrame, + }; + } + if !self.has_frame { + return Action::Wait(None); + } + + let keepalive_at = self.last_message_at.map_or(now, |t| t + self.cfg.keepalive); + if self.cleared { + return if now >= keepalive_at { + Action::Heartbeat + } else { + Action::Wait(Some(keepalive_at)) + }; + } + + let idle_at = match (self.cfg.idle_clear, self.last_frame_at) { + (Some(idle), Some(at)) => Some(at + idle), + _ => None, + }; + if idle_at.is_some_and(|at| now >= at) { + return Action::Clear; + } + + let keepalive_at = image_allowed_at.map_or(keepalive_at, |at| at.max(keepalive_at)); + if now >= keepalive_at { + return Action::SendKeepalive; + } + Action::Wait(Some( + idle_at.map_or(keepalive_at, |at| at.min(keepalive_at)), + )) + } +} + +/// Exponential reconnect backoff (spec 4.3.7). +#[derive(Debug, Clone)] +pub struct Backoff { + initial: Duration, + max: Duration, + reset_after: Duration, + next: Duration, +} + +impl Backoff { + pub fn new(initial: Duration, max: Duration, reset_after: Duration) -> Self { + Self { + initial, + max, + reset_after, + next: initial, + } + } + + /// Delay before the next attempt. `session` is how long the failed connection had been + /// up, or `None` if connecting itself failed. + pub fn on_failure(&mut self, session: Option) -> Duration { + if session.is_some_and(|s| s >= self.reset_after) { + self.next = self.initial; + } + let delay = self.next; + self.next = (self.next * 2).min(self.max); + delay + } + + pub fn reset(&mut self) { + self.next = self.initial; + } +} + +/// Lets one event through per `interval` and counts the suppressed ones. +#[derive(Debug, Clone)] +pub struct RateLimit { + interval: Duration, + last: Option, + suppressed: u64, +} + +impl RateLimit { + pub fn new(interval: Duration) -> Self { + Self { + interval, + last: None, + suppressed: 0, + } + } + + /// Returns `Some(suppressed since the last emitted event)` if this event should be logged. + pub fn check(&mut self, now: Instant) -> Option { + match self.last { + Some(last) if now < last + self.interval => { + self.suppressed += 1; + None + } + _ => { + self.last = Some(now); + Some(std::mem::take(&mut self.suppressed)) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const MS: Duration = Duration::from_millis(1); + + fn pacer(idle_clear: Option) -> (Pacer, Instant) { + let cfg = PacingConfig { + min_interval: 100 * MS, + keepalive: 1000 * MS, + idle_clear, + }; + (Pacer::new(cfg), Instant::now()) + } + + #[test] + fn nothing_without_frames() { + let (p, t0) = pacer(None); + assert_eq!(p.poll(t0), Action::Wait(None)); + } + + #[test] + fn fps_ceiling_and_freshest_wins() { + let (mut p, t0) = pacer(None); + assert!(!p.on_new_frame(t0)); + assert_eq!(p.poll(t0), Action::SendFrame); + p.on_sent(Action::SendFrame, t0); + + // Too early: wait for the slot, newer frames replace the pending one. + assert!(!p.on_new_frame(t0 + 10 * MS)); + assert_eq!(p.poll(t0 + 10 * MS), Action::Wait(Some(t0 + 100 * MS))); + assert!(p.on_new_frame(t0 + 20 * MS)); + assert_eq!(p.poll(t0 + 99 * MS), Action::Wait(Some(t0 + 100 * MS))); + assert_eq!(p.poll(t0 + 100 * MS), Action::SendFrame); + p.on_sent(Action::SendFrame, t0 + 100 * MS); + + // Late frame goes out immediately. + p.on_new_frame(t0 + 350 * MS); + assert_eq!(p.poll(t0 + 350 * MS), Action::SendFrame); + } + + #[test] + fn keepalive_repeats_last_frame() { + let (mut p, t0) = pacer(None); + p.on_new_frame(t0); + p.on_sent(Action::SendFrame, t0); + assert_eq!(p.poll(t0 + 500 * MS), Action::Wait(Some(t0 + 1000 * MS))); + assert_eq!(p.poll(t0 + 1000 * MS), Action::SendKeepalive); + p.on_sent(Action::SendKeepalive, t0 + 1000 * MS); + assert_eq!(p.poll(t0 + 1001 * MS), Action::Wait(Some(t0 + 2000 * MS))); + } + + #[test] + fn keepalive_respects_fps_ceiling() { + let cfg = PacingConfig { + min_interval: 500 * MS, + keepalive: 100 * MS, + idle_clear: None, + }; + let mut p = Pacer::new(cfg); + let t0 = Instant::now(); + p.on_new_frame(t0); + p.on_sent(Action::SendFrame, t0); + assert_eq!(p.poll(t0 + 100 * MS), Action::Wait(Some(t0 + 500 * MS))); + assert_eq!(p.poll(t0 + 500 * MS), Action::SendKeepalive); + } + + #[test] + fn idle_clear_and_resume() { + let (mut p, t0) = pacer(Some(3000 * MS)); + p.on_new_frame(t0); + p.on_sent(Action::SendFrame, t0); + for s in 1..3 { + let t = t0 + s * 1000 * MS; + assert_eq!(p.poll(t), Action::SendKeepalive); + p.on_sent(Action::SendKeepalive, t); + } + assert_eq!(p.poll(t0 + 2500 * MS), Action::Wait(Some(t0 + 3000 * MS))); + let t_clear = t0 + 3000 * MS; + assert_eq!(p.poll(t_clear), Action::Clear); + p.on_sent(Action::Clear, t_clear); + assert!(p.is_cleared()); + + // No more frame keepalives, only heartbeats. + assert_eq!( + p.poll(t_clear + 500 * MS), + Action::Wait(Some(t_clear + 1000 * MS)) + ); + assert_eq!(p.poll(t_clear + 1000 * MS), Action::Heartbeat); + p.on_sent(Action::Heartbeat, t_clear + 1000 * MS); + + // A new frame resumes sending immediately. + let t_new = t_clear + 1200 * MS; + p.on_new_frame(t_new); + assert!(!p.is_cleared()); + assert_eq!(p.poll(t_new), Action::SendFrame); + } + + #[test] + fn pending_frame_is_not_idle() { + let (mut p, t0) = pacer(Some(1000 * MS)); + p.on_new_frame(t0); + assert!(p.wants_connection(t0 + 5000 * MS)); + assert_eq!(p.poll(t0 + 5000 * MS), Action::SendFrame); + } + + #[test] + fn connection_wanted_only_with_fresh_frames() { + let (mut p, t0) = pacer(Some(2000 * MS)); + assert!(!p.wants_connection(t0)); + p.on_new_frame(t0); + p.on_sent(Action::SendFrame, t0); + assert!(p.wants_connection(t0 + 1000 * MS)); + assert!(!p.wants_connection(t0 + 2000 * MS)); + p.on_new_frame(t0 + 3000 * MS); + assert!(p.wants_connection(t0 + 3000 * MS)); + p.on_source_gone(); + assert!(!p.wants_connection(t0 + 3000 * MS)); + } + + #[test] + fn reconnect_resends_last_frame() { + let (mut p, t0) = pacer(None); + p.on_new_frame(t0); + p.on_sent(Action::SendFrame, t0); + p.on_connected(); + assert_eq!(p.poll(t0 + MS), Action::SendFrame); + } + + #[test] + fn backoff_sequence_and_reset() { + let mut b = Backoff::new(500 * MS, 10_000 * MS, 30_000 * MS); + let delays: Vec<_> = (0..7).map(|_| b.on_failure(None).as_millis()).collect(); + assert_eq!(delays, [500, 1000, 2000, 4000, 8000, 10000, 10000]); + // Short session does not reset. + assert_eq!(b.on_failure(Some(29_000 * MS)).as_millis(), 10000); + // Long session does. + assert_eq!(b.on_failure(Some(31_000 * MS)).as_millis(), 500); + assert_eq!(b.on_failure(None).as_millis(), 1000); + b.reset(); + assert_eq!(b.on_failure(None).as_millis(), 500); + } + + #[test] + fn rate_limit() { + let mut r = RateLimit::new(10_000 * MS); + let t0 = Instant::now(); + assert_eq!(r.check(t0), Some(0)); + assert_eq!(r.check(t0 + 1000 * MS), None); + assert_eq!(r.check(t0 + 9999 * MS), None); + assert_eq!(r.check(t0 + 10_000 * MS), Some(2)); + assert_eq!(r.check(t0 + 10_001 * MS), None); + } +}