Skip to content
rachel/esp32-radio
File

Blob: firmware/vendor/str0m/src/bwe/delay/rate_control.rs

rust525 lines
1use std::fmt;
2use std::time::{Duration, Instant};
3 
4use super::super::macros::log_rate_control_applied_change;
5use super::super::macros::log_rate_control_observed_bitrate;
6use super::super::macros::log_rate_control_state;
7use crate::rtp_::Bitrate;
8use crate::util::MovingAverage;
9 
10use super::super::BandwidthUsage;
11 
12// Recommended values from https://datatracker.ietf.org/doc/html/draft-ietf-rmcat-gcc-02#section-5
13/// Smoothing factor applied to moving stats for observed bitrates when we are in the decreasing
14/// state.
15const OBSERVED_BIT_RATE_SMOOTHING_FACTOR: f64 = 0.95;
16/// The ratio of current estimated bandwidth to use when decreasing the rate.
17const BETA: f64 = 0.85;
18/// The coefficient used for multiplicative rate increase.
19const MULTIPLICATIVE_INCREASE_COEF: f64 = 1.08;
20/// The maximal ratio of the observed bitrate that we allow estimating in a single increase.
21const MAX_ESTIMATE_RATIO: f64 = 1.5;
22/// Default backoff time added to RTT for response time calculation (kDefaultBackoffTimeInMs in WebRTC).
23const DEFAULT_BACKOFF_TIME: Duration = Duration::from_millis(100);
24/// Number of standard deviations below mean to reset observed bitrate average.
25const OBSERVED_BITRATE_RESET_THRESHOLD_STD: f64 = 3.0;
26 
27/// A type used to estimates a suitable send bitrate.
28///
29/// Inputs to the rate controller are:
30/// * The observed received bitrate(via TWCC feddback).
31/// * RTT.
32/// * Congestion estimates from the delay controller.
33pub struct RateControl {
34    state: State,
35 
36    estimated_bitrate: Bitrate,
37    min_bitrate: Bitrate,
38    max_bitrate: Bitrate,
39 
40    /// The last observed bitrate calculated based on TWCC data.
41    last_observed_bitrate: Option<Bitrate>,
42    /// The averaged observed bitrate when we have been in the decrease state.
43    averaged_observed_bitrate: MovingAverage,
44    /// The last time we updated the estimated bitrate.
45    last_estimate_update: Option<Instant>,
46    // Last RTT estimate in micro-seconds
47    last_rtt: Option<Duration>,
48}
49 
50impl RateControl {
51    pub fn new(start_bitrate: Bitrate, min_bitrate: Bitrate, max_bitrate: Bitrate) -> Self {
52        log_rate_control_state!(State::Increase as i8);
53 
54        Self {
55            state: State::Increase,
56 
57            estimated_bitrate: start_bitrate,
58            min_bitrate,
59            max_bitrate,
60 
61            last_observed_bitrate: None,
62            averaged_observed_bitrate: MovingAverage::new(OBSERVED_BIT_RATE_SMOOTHING_FACTOR),
63            last_estimate_update: None,
64            last_rtt: None,
65        }
66    }
67 
68    /// Update with input from the delay controller.
69    pub fn update(
70        &mut self,
71        signal: Signal,
72        observed_bitrate: Bitrate,
73        rtt: Option<Duration>,
74        now: Instant,
75    ) {
76        self.last_observed_bitrate = Some(observed_bitrate);
77        if let Some(rtt) = rtt {
78            self.last_rtt = Some(rtt);
79        }
80 
81        self.state = self.state.transition(signal);
82        log_rate_control_observed_bitrate!(
83            observed_bitrate.as_f64(),
84            self.averaged_observed_bitrate
85                .get()
86                .map(|avg| avg.to_string())
87                .unwrap_or_default()
88        );
89 
90        match self.state {
91            State::Increase => {
92                self.increase(observed_bitrate, now);
93            }
94            State::Decrease => {
95                // Maintain observed bitrate statistics while we are in the decrease state.
96                //
97                // This must NOT be gated by time_to_reduce_further. That function is
98                // intended to gate *applying another reduction*, not collecting throughput stats.
99                self.update_observed_bitrate(observed_bitrate);
100 
101                // Only apply decrease if enough time has passed since last bitrate change
102                // or if throughput is critically low (< 50% of estimate)
103                if self.time_to_reduce_further(now, observed_bitrate) {
104                    self.decrease(observed_bitrate, now);
105                }
106            }
107            State::Hold => {
108                // Do nothing
109            }
110        }
111    }
112 
113    fn update_observed_bitrate(&mut self, observed_bitrate: Bitrate) {
114        if self
115            .averaged_observed_bitrate
116            .lower_range(OBSERVED_BITRATE_RESET_THRESHOLD_STD)
117            .map(|lower| observed_bitrate.as_f64() < lower)
118            .unwrap_or(false)
119        {
120            self.averaged_observed_bitrate.reset();
121        }
122        self.averaged_observed_bitrate
123            .update(observed_bitrate.as_f64());
124    }
125 
126    /// Check if it's time to reduce the bitrate further.
127    ///
128    /// This implements WebRTC's TimeToReduceFurther logic which prevents
129    /// rapid successive decreases, especially after probe results.
130    ///
131    /// Returns true if:
132    /// 1. Enough time (1 RTT, clamped to 10-200ms) has passed since last change, OR
133    /// 2. Throughput is critically low (< 50% of current estimate)
134    fn time_to_reduce_further(&self, now: Instant, observed_bitrate: Bitrate) -> bool {
135        let Some(last_change) = self.last_estimate_update else {
136            return true; // No previous change, allow decrease
137        };
138 
139        // WebRTC uses: clamp(rtt, 10ms, 200ms)
140        let rtt = self.last_rtt.unwrap_or(DEFAULT_BACKOFF_TIME);
141        let reduction_interval = rtt.clamp(Duration::from_millis(10), Duration::from_millis(200));
142 
143        let time_since_change = now.saturating_duration_since(last_change);
144 
145        if time_since_change >= reduction_interval {
146            return true;
147        }
148 
149        // If throughput is critically low (< 50% of estimate), allow immediate decrease
150        let threshold = self.estimated_bitrate * 0.5;
151        if observed_bitrate < threshold {
152            return true;
153        }
154 
155        false
156    }
157 
158    /// The current estimated bitrate.
159    pub fn estimated_bitrate(&self) -> Bitrate {
160        self.estimated_bitrate
161    }
162 
163    /// Set a probe result indicating discovered capacity.
164    ///
165    /// When a probe succeeds, it means the network can handle at least this bitrate.
166    /// We use this to quickly increase our estimate without waiting for gradual ramp-up.
167    ///
168    /// Apply a probe result directly to the estimate.
169    ///
170    /// This matches WebRTC's behavior where probe results are accepted unconditionally
171    /// (subject only to min/max clamping), regardless of whether they're higher or lower
172    /// than the current estimate. The timestamp is updated to prevent the next regular
173    /// update from immediately overriding the probe result.
174    ///
175    /// WebRTC does NOT change the rate control state when applying a probe - the state
176    /// remains unchanged to avoid triggering unintended AIMD behavior.
177    pub fn set_probe_result(&mut self, probe_bitrate: Bitrate, now: Instant) {
178        // WebRTC calls SetEstimate() directly without filtering by current estimate
179        // or changing the rate control state. Accept the probe result unconditionally
180        // (update_estimate handles clamping to min/max bounds).
181        self.update_estimate(probe_bitrate, now);
182 
183        // Do NOT change self.state - keep current state to match WebRTC behavior.
184        // Changing state here could trigger unintended AIMD decrease/increase logic.
185    }
186 
187    fn increase(&mut self, observed_bitrate: Bitrate, now: Instant) {
188        // WebRTC limits increases to 1.5x observed throughput to avoid unlimited growth
189        // when we're already above what we're actually sending
190        // See: aimd_rate_control.cc line 251-252
191        let increase_limit = observed_bitrate * 1.5 + Bitrate::kbps(10);
192 
193        if self.estimated_bitrate >= increase_limit {
194            // WebRTC updates time_last_bitrate_change_ even when skipping increase
195            // (see aimd_rate_control.cc line 281, which is outside the increase check)
196            // This prevents stale timestamps from allowing premature decreases
197            self.last_estimate_update = Some(now);
198            return;
199        }
200 
201        // Initialize timestamp if this is the first increase call, otherwise use existing value.
202        // Note: In practice, last_estimate_update is always Some here because either:
203        // 1) We returned early above and set it, or
204        // 2) A previous call to decrease() or increase() already set it
205        let last_estimate_update = self.last_estimate_update.unwrap_or(now);
206        if self.last_estimate_update.is_none() {
207            self.last_estimate_update = Some(now);
208        }
209 
210        if self
211            .averaged_observed_bitrate
212            .upper_range(3.0)
213            .map(|upper| observed_bitrate.as_f64() > upper)
214            .unwrap_or(false)
215        {
216            self.averaged_observed_bitrate.reset();
217        }
218 
219        let since_last_update = now - last_estimate_update;
220        assert!(since_last_update >= Duration::ZERO);
221        let near_convergence = self.is_near_convergence();
222 
223        let mut new_estimate = if near_convergence {
224            // Additive increase
225            log_rate_control_applied_change!("increase_additive");
226            let response_time = self.last_rtt.unwrap_or(Duration::ZERO) + DEFAULT_BACKOFF_TIME;
227 
228            let alpha =
229                0.5 * (since_last_update.as_secs_f64() / response_time.as_secs_f64()).min(1.0);
230            let expected_packet_size = self.estimated_packet_size();
231            self.estimated_bitrate.as_f64() + (alpha * expected_packet_size).max(1000.0)
232        } else {
233            // Multiplicative increase
234            log_rate_control_applied_change!("increase_multiplicative");
235            let eta = MULTIPLICATIVE_INCREASE_COEF.powf(since_last_update.as_secs_f64().min(1.0));
236            let increase = ((eta - 1.0) * self.estimated_bitrate.as_f64()).max(1_000.0);
237 
238            self.estimated_bitrate.as_f64() + increase
239        };
240 
241        // Cap at the increase limit (and observed bitrate ratio)
242        let max = observed_bitrate.as_f64() * MAX_ESTIMATE_RATIO;
243        new_estimate = max.min(new_estimate).min(increase_limit.as_f64());
244 
245        self.update_estimate(new_estimate.into(), now);
246    }
247 
248    fn decrease(&mut self, observed_bitrate: Bitrate, now: Instant) {
249        log_rate_control_applied_change!("decrease");
250        let mut new_estimate = observed_bitrate * BETA;
251 
252        if self.estimated_bitrate < new_estimate {
253            // Avoid increasing the bitrate on overuse
254            new_estimate = self.estimated_bitrate;
255        }
256 
257        #[allow(unused)]
258        if let Some(observed_average) = self.averaged_observed_bitrate.get() {
259            log_rate_control_observed_bitrate!(
260                observed_bitrate.as_u64(),
261                observed_average.round() as u64
262            );
263        }
264        // According to https://datatracker.ietf.org/doc/html/draft-ietf-rmcat-gcc-02#section-6 we
265        // should wait until this happens as consequence of the delay control, but libWebRTC does
266        // it immediately.
267        self.state = State::Hold;
268        log_rate_control_state!(self.state as i8);
269        self.update_estimate(new_estimate, now);
270    }
271 
272    fn is_near_convergence(&self) -> bool {
273        // Not near convergence until we have valid statistics
274        if !self.averaged_observed_bitrate.valid() {
275            return false;
276        }
277        let Some(last_observed_bitrate) = self.last_observed_bitrate else {
278            return false;
279        };
280 
281        // Near convergence if the observed bandwidth is within 3 standard deviations of
282        // the moving average when we have been in the decrease state.
283        self.averaged_observed_bitrate
284            .within_std(last_observed_bitrate.as_f64(), 3.0)
285    }
286 
287    fn update_estimate(&mut self, bitrate: Bitrate, now: Instant) {
288        self.estimated_bitrate = bitrate.clamp(self.min_bitrate, self.max_bitrate);
289        self.last_estimate_update = Some(now);
290    }
291 
292    fn estimated_packet_size(&self) -> f64 {
293        // Assume 30 FPS video dominates the send rate
294        let bits_per_frame = self.estimated_bitrate.as_f64() / 30.0;
295        let packets_per_frame = (bits_per_frame / (1200.0 / 8.0)).ceil();
296 
297        bits_per_frame / packets_per_frame
298    }
299}
300 
301#[derive(Debug, Clone, Copy, PartialEq, Eq)]
302pub enum Signal {
303    Overuse,
304    Underuse,
305    Normal,
306}
307 
308#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
309enum State {
310    Increase = 1,
311    Hold = 0,
312    #[default]
313    Decrease = -1,
314}
315 
316impl State {
317    fn transition(&self, signal: Signal) -> Self {
318        let new_state = match (self, signal) {
319            (_, Signal::Overuse) => Self::Decrease,
320            (_, Signal::Underuse) => Self::Hold,
321            // https://datatracker.ietf.org/doc/html/draft-ietf-rmcat-gcc-02#section-6 says to
322            // transition to Hold here, but libWebRTC stays in decrease. We will eventually
323            // transition to Hold on Underuse.
324            (Self::Decrease, Signal::Normal) => Self::Hold,
325            (Self::Hold | Self::Increase, Signal::Normal) => Self::Increase,
326        };
327 
328        if new_state != *self {
329            log_rate_control_state!(new_state as i8);
330        }
331 
332        new_state
333    }
334}
335 
336impl From<BandwidthUsage> for Signal {
337    fn from(value: BandwidthUsage) -> Self {
338        match value {
339            BandwidthUsage::Overuse => Signal::Overuse,
340            BandwidthUsage::Normal => Signal::Normal,
341            BandwidthUsage::Underuse => Signal::Underuse,
342        }
343    }
344}
345 
346impl fmt::Display for State {
347    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
348        match self {
349            State::Hold => write!(f, "hold"),
350            State::Increase => write!(f, "increase"),
351            State::Decrease => write!(f, "decrease"),
352        }
353    }
354}
355 
356impl fmt::Display for Signal {
357    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
358        match self {
359            Signal::Overuse => write!(f, "overuse"),
360            Signal::Underuse => write!(f, "underuse"),
361            Signal::Normal => write!(f, "normal"),
362        }
363    }
364}
365 
366#[cfg(test)]
367mod test {
368    use std::time::Duration;
369 
370    use super::{RateControl, Signal, State};
371 
372    mod state {
373        use super::{Signal, State};
374 
375        #[test]
376        fn test_state_transitions() {
377            // Tests based on the table in
378            // https://datatracker.ietf.org/doc/html/draft-ietf-rmcat-gcc-02#section-5
379 
380            // Hold
381            let hold = State::Hold;
382            assert_eq!(hold.transition(Signal::Overuse), State::Decrease);
383            assert_eq!(hold.transition(Signal::Normal), State::Increase);
384            assert_eq!(hold.transition(Signal::Underuse), State::Hold);
385 
386            // Increase
387            let increase = State::Increase;
388            assert_eq!(increase.transition(Signal::Overuse), State::Decrease);
389            assert_eq!(increase.transition(Signal::Normal), State::Increase);
390            assert_eq!(increase.transition(Signal::Underuse), State::Hold);
391 
392            // Decrease
393            let decrease = State::Decrease;
394            assert_eq!(decrease.transition(Signal::Overuse), State::Decrease);
395            assert_eq!(decrease.transition(Signal::Normal), State::Hold);
396            assert_eq!(decrease.transition(Signal::Underuse), State::Hold);
397        }
398    }
399 
400    mod rate_controller {
401        use std::time::Instant;
402 
403        use super::{RateControl, Signal, duration_ms};
404 
405        fn make_control(estimated_bitrate: u64) -> RateControl {
406            RateControl::new(estimated_bitrate.into(), 10_000.into(), 50_000_000.into())
407        }
408 
409        #[test]
410        fn test_initial_estimate() {
411            let rate_controller = make_control(100_000);
412 
413            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 100_000);
414        }
415 
416        #[test]
417        fn test_normal_yields_multiplicative_increase() {
418            let now = Instant::now();
419            let mut rate_controller = make_control(100_000);
420            // Seed last estimate value
421            rate_controller.update(Signal::Normal, 85_000.into(), None, now);
422            assert_eq!(
423                rate_controller.estimated_bitrate().as_u64(),
424                101_000,
425                "Initial estimate should increase by the minimum(1Kbit/s)"
426            );
427 
428            rate_controller.update(Signal::Normal, 95_000.into(), None, now + duration_ms(500));
429            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 104_963);
430 
431            rate_controller.update(Signal::Normal, 97_000.into(), None, now + duration_ms(1000));
432            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 109_081);
433        }
434 
435        #[test]
436        fn test_normal_to_under_use_yields_hold() {
437            let now = Instant::now();
438            let mut rate_controller = make_control(100_000);
439            // Seed last estimate value
440            rate_controller.update(Signal::Normal, 85_000.into(), None, now);
441            assert_eq!(
442                rate_controller.estimated_bitrate().as_u64(),
443                101_000,
444                "Initial estimate should increase by the minimum(1Kbit/s)"
445            );
446 
447            // Should remain in increase and increase estimate
448            rate_controller.update(Signal::Normal, 95_000.into(), None, now + duration_ms(500));
449            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 104_963);
450 
451            // Should transition to hold
452            rate_controller.update(
453                Signal::Underuse,
454                97_000.into(),
455                None,
456                now + duration_ms(1000),
457            );
458            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 104_963);
459 
460            // Should remain in hold and not modify estimates
461            rate_controller.update(
462                Signal::Underuse,
463                97_000.into(),
464                None,
465                now + duration_ms(2000),
466            );
467            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 104_963);
468        }
469 
470        #[test]
471        fn test_immediate_overuse() {
472            let now = Instant::now();
473            let mut rate_controller = make_control(100_000);
474            // Seed last estimate value
475            rate_controller.update(Signal::Normal, 85_000.into(), None, now);
476 
477            rate_controller.update(Signal::Overuse, 90_000.into(), None, now + duration_ms(500));
478            assert_eq!(
479                rate_controller.estimated_bitrate().as_u64(),
480                76_500,
481                "When overuse is detected we should reduce the estimate to \
482                85% of the obeserved rate immediately"
483            );
484        }
485 
486        #[test]
487        fn test_immediate_overuse_then_stable() {
488            let now = Instant::now();
489            let mut rate_controller = make_control(100_000);
490            // Seed last estimate value
491            rate_controller.update(Signal::Normal, 85_000.into(), Some(duration_ms(80)), now);
492 
493            rate_controller.update(Signal::Overuse, 90_000.into(), None, now + duration_ms(500));
494            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 76_500);
495 
496            rate_controller.update(
497                Signal::Overuse,
498                75_000.into(),
499                None,
500                now + duration_ms(1000),
501            );
502            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 63_750);
503 
504            rate_controller.update(Signal::Normal, 60_000.into(), None, now + duration_ms(1500));
505            // NB: This matches libWebRTC but diverges from the spec
506            assert_eq!(
507                rate_controller.estimated_bitrate().as_u64(),
508                66_251,
509                "After adjusting on overuse we immediately return to increase on the next normal signal"
510            );
511 
512            rate_controller.update(Signal::Normal, 60_000.into(), None, now + duration_ms(2500));
513            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 71_552,);
514 
515            // NB: Additive increase because we are nearing convergence
516            rate_controller.update(Signal::Normal, 70_000.into(), None, now + duration_ms(3500));
517            assert_eq!(rate_controller.estimated_bitrate().as_u64(), 72552);
518        }
519    }
520 
521    fn duration_ms(ms: u64) -> Duration {
522        Duration::from_millis(ms)
523    }
524}