Skip to content
rachel/esp32-radio
File

Blob: firmware/vendor/str0m/src/bwe/mod.rs

rust487 lines
1//! Google Congestion Control (GoogCC) Bandwidth Estimation based on TWCC feedback.
2//!
3//! This implementation is ported from libWebRTC's GoogCC and goes beyond the simplified
4//! IETF draft (<https://datatracker.ietf.org/doc/html/draft-ietf-rmcat-gcc-02>) to include
5//! WebRTC's production features:
6//!
7//! - Delay-based control (trendline estimator with AIMD rate control)
8//! - Loss-based control (with inherent loss rate estimation)
9//! - Probe controller with state machine and multi-stage probing strategy
10//! - ALR (Application Limited Region) detection and periodic probing
11//! - Link capacity estimation from ALR probes
12//!
13//! The probe controller in particular closely matches WebRTC's `ProbeController` behavior
14//! and default constants, enabling compatible bandwidth discovery with WebRTC endpoints.
15 
16use std::cmp::Ordering;
17use std::fmt;
18use std::time::{Duration, Instant};
19 
20use crate::Reason;
21use crate::rtp_::{Bitrate, DataSize, TwccClusterId, TwccSendRecord, TwccSeq};
22use crate::util::Soonest;
23 
24mod acked_bitrate_estimator;
25mod alr_detector;
26pub(crate) mod api;
27mod delay;
28mod link_capacity_estimator;
29mod loss_controller;
30mod macros;
31mod probe;
32mod smoother;
33mod time;
34 
35use acked_bitrate_estimator::AckedBitrateEstimator;
36use alr_detector::AlrDetector;
37use delay::DelayController;
38use link_capacity_estimator::LinkCapacityEstimator;
39use loss_controller::{LossController, LossControllerState};
40use macros::log_loss;
41use smoother::EstimateSmoother;
42 
43pub(crate) use macros::{log_pacer_media_debt, log_pacer_padding_debt};
44pub(crate) use probe::{BandwidthLimitedCause, ProbeEstimator};
45pub(crate) use probe::{ProbeClusterState, ProbeControl};
46 
47#[cfg(feature = "_internal_test_exports")]
48pub use probe::ProbeClusterConfig;
49#[cfg(not(feature = "_internal_test_exports"))]
50pub(crate) use probe::ProbeClusterConfig;
51 
52const INITIAL_BITRATE_WINDOW: Duration = Duration::from_millis(500);
53const BITRATE_WINDOW: Duration = Duration::from_millis(150);
54const STARTUP_PHASE: Duration = Duration::from_secs(2);
55 
56pub struct Bwe {
57    bwe: SendSideBandwidthEstimator,
58    desired_bitrate: Bitrate,
59    smoother: EstimateSmoother,
60}
61 
62impl Bwe {
63    pub fn new(initial: Bitrate) -> Self {
64        let send_side_bwe = SendSideBandwidthEstimator::new(initial);
65        Bwe {
66            bwe: send_side_bwe,
67            desired_bitrate: Bitrate::ZERO,
68            smoother: EstimateSmoother::new(),
69        }
70    }
71 
72    pub fn handle_timeout(&mut self, now: Instant, do_probe: bool) -> Option<ProbeClusterConfig> {
73        let result = self.bwe.handle_timeout(self.desired_bitrate, do_probe, now);
74        if let Some(estimate) = self.bwe.last_estimate() {
75            self.smoother.record(now, estimate);
76        }
77        result
78    }
79 
80    pub fn start_probe(&mut self, config: ProbeClusterConfig, now: Instant) -> bool {
81        self.bwe.start_probe(config, now)
82    }
83 
84    pub fn end_probe(&mut self, now: Instant, cluster_id: TwccClusterId) {
85        self.bwe.end_probe(now, cluster_id);
86    }
87 
88    pub fn reset(&mut self, init_bitrate: Bitrate) {
89        self.bwe.reset(init_bitrate);
90    }
91 
92    pub fn update<'t>(
93        &mut self,
94        records: impl Iterator<Item = &'t crate::rtp_::TwccSendRecord>,
95        now: Instant,
96    ) {
97        self.bwe.update(records, now);
98    }
99 
100    pub fn poll_estimate(&mut self) -> Option<Bitrate> {
101        self.smoother.poll()
102    }
103 
104    pub fn poll_timeout(&self) -> (Option<Instant>, Reason) {
105        self.bwe.poll_timeout()
106    }
107 
108    pub fn last_estimate(&self) -> Option<Bitrate> {
109        self.bwe.last_estimate()
110    }
111 
112    pub fn on_media_sent(&mut self, payload_size: DataSize, is_padding: bool, now: Instant) {
113        if !is_padding {
114            // Update ALR detector with media bytes sent
115            self.bwe.on_media_sent(payload_size, now);
116        }
117    }
118 
119    pub fn is_overusing(&self) -> bool {
120        self.bwe.is_overusing()
121    }
122 
123    pub fn set_desired_bitrate(&mut self, v: Bitrate) {
124        self.desired_bitrate = v;
125    }
126}
127 
128struct SendSideBandwidthEstimator {
129    delay_controller: DelayController,
130    loss_controller: LossController,
131    acked_bitrate_estimator: AckedBitrateEstimator,
132    probe_control: ProbeControl,
133    probe_estimator: ProbeEstimator,
134    started_at: Option<Instant>,
135    alr_detector: AlrDetector,
136    link_capacity_estimator: LinkCapacityEstimator,
137    last_updated_estimate: Option<Bitrate>,
138}
139 
140impl SendSideBandwidthEstimator {
141    pub fn new(initial_bitrate: Bitrate) -> Self {
142        let mut alr_detector = AlrDetector::new();
143        alr_detector.set_estimated_bitrate(initial_bitrate);
144 
145        let mut loss_controller = LossController::new();
146        loss_controller.set_bandwidth_estimate(initial_bitrate);
147 
148        Self {
149            delay_controller: DelayController::new(initial_bitrate),
150            loss_controller,
151            acked_bitrate_estimator: AckedBitrateEstimator::new(
152                INITIAL_BITRATE_WINDOW,
153                BITRATE_WINDOW,
154            ),
155            probe_control: ProbeControl::new(),
156            probe_estimator: ProbeEstimator::new(),
157            started_at: None,
158            alr_detector,
159            link_capacity_estimator: LinkCapacityEstimator::new(),
160            last_updated_estimate: None,
161        }
162    }
163 
164    /// Whether the delay-based detector currently signals overuse.
165    ///
166    /// This is useful for gating behaviors (like padding/probing) that would otherwise
167    /// re-excite the system while we're already congested.
168    pub fn is_overusing(&self) -> bool {
169        self.delay_controller.is_overusing()
170    }
171 
172    /// Update ALR detector with actual bytes sent.
173    ///
174    /// Should be called for media packets (not padding/probes).
175    /// This is typically called from the session's packet sending logic.
176    pub fn on_media_sent(&mut self, bytes: DataSize, now: Instant) {
177        self.alr_detector.on_bytes_sent(bytes, now);
178    }
179 
180    /// Record a packet from a TWCC report.
181    pub fn update<'t>(&mut self, records: impl Iterator<Item = &'t TwccSendRecord>, now: Instant) {
182        let _ = self.started_at.get_or_insert(now);
183 
184        let send_records: Vec<_> = records.collect();
185 
186        // Feed records to probe estimator for analysis and process any new probe results
187        let mut latest_probe_result = None;
188        for (config, bitrate) in self.probe_estimator.update(send_records.iter().copied()) {
189            latest_probe_result = Some(bitrate);
190 
191            // Update link capacity estimator for every successful ALR probe, not just the latest.
192            // The estimator internally takes the max of all probe results, building up knowledge
193            // of proven link capacity. This differs from the delay controller, which only receives
194            // the latest probe result (matching WebRTC's FetchAndResetLastEstimatedBitrate behavior).
195            if config.is_alr_probe() {
196                self.link_capacity_estimator.update_from_probe(bitrate, now);
197            }
198        }
199 
200        let mut acked_packets = vec![];
201 
202        let mut max_rtt = None;
203        let mut count = 0;
204        let mut lost = 0;
205        for record in send_records.iter() {
206            count += 1;
207            let Ok(acked_packet) = (*record).try_into() else {
208                lost += 1;
209                continue;
210            };
211            acked_packets.push(acked_packet);
212            max_rtt = max_rtt.max(record.rtt());
213        }
214        acked_packets.sort_by(AckedPacket::order_by_receive_time);
215 
216        for acked_packet in acked_packets.iter() {
217            self.acked_bitrate_estimator
218                .update(acked_packet.remote_recv_time, acked_packet.size);
219        }
220 
221        let acked_bitrate = self.acked_bitrate_estimator.current_estimate();
222 
223        // Use the latest probe result from this update, if any
224        let probe_result = latest_probe_result;
225 
226        let is_probe_result = probe_result.is_some();
227 
228        // Update delay controller with the latest probe result
229        let maybe_estimate =
230            self.delay_controller
231                .update(&acked_packets, acked_bitrate, probe_result, now);
232 
233        let Some(delay_estimate) = maybe_estimate else {
234            return;
235        };
236 
237        let loss = if count == 0 {
238            0.0
239        } else {
240            lost as f64 / count as f64
241        };
242        log_loss!(loss);
243 
244        // During startup with no loss, use delay-based estimate directly
245        if in_startup_phase(self.started_at, now) && loss <= 0.001 {
246            self.loss_controller.set_bandwidth_estimate(delay_estimate);
247            return;
248        }
249 
250        // When probe succeeds, set bandwidth directly
251        if is_probe_result {
252            self.loss_controller.set_bandwidth_estimate(delay_estimate);
253        }
254 
255        if let Some(acked_bitrate) = acked_bitrate {
256            self.loss_controller.set_acknowledged_bitrate(acked_bitrate);
257        }
258 
259        // This corresponds to UpdateLossBasedEstimator + UpdateEstimate
260        self.loss_controller
261            .update_bandwidth_estimate(&send_records, delay_estimate);
262 
263        // Loss-based result is capped by delay_based_limit
264        let loss_result = self.loss_controller.loss_based_result();
265        if let Some(loss_estimate) = loss_result.bandwidth_estimate {
266            if loss_estimate > delay_estimate {
267                // Loss controller produced higher estimate than delay controller
268                // Cap it at delay estimate (delay controller is the upper limit)
269                self.loss_controller.set_bandwidth_estimate(delay_estimate);
270            }
271        }
272 
273        // Feed the (possibly combined) estimate into subcomponents wanting it.
274        self.propagate_estimate();
275    }
276 
277    pub fn poll_timeout(&self) -> (Option<Instant>, Reason) {
278        let delay_timeout = Some(self.delay_controller.poll_timeout());
279        let probe_timeout = Some(self.probe_control.poll_timeout());
280        let probe_estimator_timeout = Some(self.probe_estimator.poll_timeout());
281        (delay_timeout, Reason::BweDelayControl)
282            .soonest((probe_timeout, Reason::BweProbeControl))
283            .soonest((probe_estimator_timeout, Reason::BweProbeEstimator))
284    }
285 
286    /// Handle periodic timeout for BWE components.
287    pub fn handle_timeout(
288        &mut self,
289        desired_bitrate: Bitrate,
290        do_probe: bool,
291        now: Instant,
292    ) -> Option<ProbeClusterConfig> {
293        self.delay_controller
294            .handle_timeout(self.acked_bitrate_estimator.current_estimate(), now);
295 
296        // Update probe control with desired bitrate.
297        self.probe_control.set_desired_bitrate(desired_bitrate);
298 
299        // Get ALR state and forward to both probe control and loss controller
300        let alr_start_time = self.alr_detector.alr_start_time();
301        if let Some(t) = alr_start_time {
302            self.probe_control.set_alr_start_time(t);
303        } else {
304            self.probe_control.set_alr_stop_time(now);
305        }
306 
307        self.loss_controller.set_alr_start_time(alr_start_time);
308 
309        // Get link capacity estimate and forward to loss controller.
310        let link_capacity = self.link_capacity_estimator.capacity_estimate(now);
311        self.loss_controller
312            .set_link_capacity_estimate(link_capacity);
313 
314        // Clean up expired probe cluster state
315        self.probe_estimator.handle_timeout(now);
316 
317        // Feed the current estimate into subcontrollers, if it changed.
318        self.propagate_estimate();
319 
320        // If we can't probe, clear any pending/active probes
321        if !do_probe {
322            self.probe_estimator.clear_probes();
323        }
324 
325        self.probe_control.enable(do_probe);
326 
327        // Timer-driven probe logic (WebRTC `Process()` equivalent).
328        self.probe_control.handle_timeout(now)
329    }
330 
331    fn propagate_estimate(&mut self) {
332        // Do we have a value?
333        let Some(estimate) = self.last_estimate() else {
334            return;
335        };
336        // Did it change?
337        if self.last_updated_estimate == Some(estimate) {
338            return;
339        }
340 
341        let cause = self.bandwidth_limited_cause();
342 
343        self.probe_control.set_estimated_bitrate(estimate, cause);
344        self.alr_detector.set_estimated_bitrate(estimate);
345 
346        // Don't update until this changes.
347        self.last_updated_estimate = Some(estimate);
348    }
349 
350    fn bandwidth_limited_cause(&self) -> BandwidthLimitedCause {
351        if self.delay_controller.is_overusing() {
352            return BandwidthLimitedCause::DelayBasedLimitedDelayIncreased;
353        }
354 
355        match self.loss_controller.loss_based_result().state {
356            LossControllerState::DelayBased => BandwidthLimitedCause::DelayBasedLimited,
357            LossControllerState::Increasing => BandwidthLimitedCause::LossLimitedBweIncreasing,
358            LossControllerState::Decreasing => BandwidthLimitedCause::LossLimitedBwe,
359        }
360    }
361 
362    /// Get the latest estimate.
363    pub fn last_estimate(&self) -> Option<Bitrate> {
364        let delay_estimate = self.delay_controller.last_estimate();
365 
366        let loss_result = self.loss_controller.loss_based_result();
367 
368        // Only apply loss-based limiting when actively in a loss-limiting state
369        match loss_result.state {
370            LossControllerState::DelayBased => {
371                // Loss controller defers to delay-based estimate
372                delay_estimate
373            }
374            LossControllerState::Decreasing | LossControllerState::Increasing => {
375                // Loss controller is actively limiting or recovering
376                match (delay_estimate, loss_result.bandwidth_estimate) {
377                    (Some(de), Some(le)) => Some(de.min(le)),
378                    (None, le @ Some(_)) => le,
379                    (de @ Some(_), None) => de,
380                    (None, None) => None,
381                }
382            }
383        }
384    }
385 
386    /// Start analyzing a probe cluster.
387    ///
388    /// This should be called when the pacer starts sending a probe cluster,
389    /// to tell the estimator which cluster to watch for in TWCC feedback.
390    /// Returns `true` if the probe was started, `false` if rejected.
391    pub fn start_probe(&mut self, config: ProbeClusterConfig, now: Instant) -> bool {
392        self.probe_estimator.probe_start(config, now)
393    }
394 
395    /// End a probe cluster and mark it for cleanup.
396    ///
397    /// This should be called when the pacer finishes sending a probe cluster.
398    /// The estimator will continue collecting feedback for a cluster history period
399    /// to allow late-arriving TWCC reports to refine the estimate.
400    pub fn end_probe(&mut self, now: Instant, cluster_id: TwccClusterId) {
401        self.probe_estimator.end_probe(now, cluster_id);
402    }
403 
404    pub fn reset(&mut self, init_bitrate: Bitrate) {
405        *self = Self::new(init_bitrate);
406    }
407}
408 
409/// A RTP packet that has been sent and acknowledged by the receiver in a TWCC report.
410#[derive(Debug, Copy, Clone)]
411pub struct AckedPacket {
412    /// The TWCC sequence number
413    seq_no: TwccSeq,
414    /// The size of the packets in bytes.
415    size: DataSize,
416    /// When we sent the packet
417    local_send_time: Instant,
418    /// When the packet was received at the remote, note this Instant is only usable with other
419    /// instants of the same type i.e. those that represent a TWCC reported receive time for this
420    /// session.
421    remote_recv_time: Instant,
422    /// The local time when received confirmation that the other side received the seq i.e. when we
423    /// received the TWCC report for this packet.
424    local_recv_time: Instant,
425}
426 
427impl AckedPacket {
428    fn rtt(&self) -> Duration {
429        self.local_recv_time - self.local_send_time
430    }
431 
432    fn order_by_receive_time(lhs: &Self, rhs: &Self) -> Ordering {
433        if lhs.remote_recv_time != rhs.remote_recv_time {
434            lhs.remote_recv_time.cmp(&rhs.remote_recv_time)
435        } else if lhs.local_send_time != rhs.local_send_time {
436            lhs.local_send_time.cmp(&rhs.local_send_time)
437        } else {
438            lhs.seq_no.cmp(&rhs.seq_no)
439        }
440    }
441}
442 
443// NB: Extracted for lifetime reasons
444fn in_startup_phase(started_at: Option<Instant>, now: Instant) -> bool {
445    started_at
446        .map(|s| now.duration_since(s) <= STARTUP_PHASE)
447        .unwrap_or(false)
448}
449 
450impl TryFrom<&TwccSendRecord> for AckedPacket {
451    type Error = ();
452 
453    fn try_from(value: &TwccSendRecord) -> Result<Self, Self::Error> {
454        let Some(remote_recv_time) = value.remote_recv_time() else {
455            return Err(());
456        };
457        let Some(local_recv_time) = value.local_recv_time() else {
458            return Err(());
459        };
460 
461        Ok(Self {
462            seq_no: value.seq(),
463            size: value.size().into(),
464            local_send_time: value.local_send_time(),
465            remote_recv_time,
466            local_recv_time,
467        })
468    }
469}
470 
471#[derive(Debug, Clone, Copy, PartialEq, Eq)]
472enum BandwidthUsage {
473    Overuse,
474    Normal,
475    Underuse,
476}
477 
478impl fmt::Display for BandwidthUsage {
479    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
480        match self {
481            BandwidthUsage::Overuse => write!(f, "overuse"),
482            BandwidthUsage::Normal => write!(f, "normal"),
483            BandwidthUsage::Underuse => write!(f, "underuse"),
484        }
485    }
486}