Skip to content
rachel/esp32-radio
File

Blob: firmware/vendor/str0m/tests/bwe/common.rs

rust326 lines
1#![allow(unused)]
2 
3//! Common utilities for Bandwidth Estimation (BWE) integration tests.
4 
5use std::sync::Arc;
6use std::time::{Duration, Instant};
7 
8use netem::NetemConfig;
9use str0m::_internal_test_exports::ProbeClusterConfig;
10use str0m::bwe::{Bitrate, BweKind};
11use str0m::format::Codec;
12use str0m::media::MediaKind;
13use str0m::media::Pt;
14use str0m::rtp::{RtpWrite, Ssrc};
15use str0m::{Event, Rtc, RtcError};
16use tracing::info;
17 
18#[path = "../common.rs"]
19mod test_common;
20pub use test_common::*;
21 
22/// Helper to create two connected peers with BWE enabled on the sender.
23pub fn connect_with_bwe(initial_bitrate: Bitrate, desired_bitrate: Bitrate) -> (TestRtc, TestRtc) {
24    let start = Instant::now();
25 
26    // Only sender (L) needs BWE enabled
27    let rtc1 = Rtc::builder()
28        .set_rtp_mode(true)
29        .enable_bwe(Some(initial_bitrate))
30        .build(start);
31 
32    let rtc2 = Rtc::builder().set_rtp_mode(true).build(start);
33 
34    let (mut l, mut r) = connect_l_r_with_rtc(rtc1, rtc2);
35 
36    l.bwe().set_desired_bitrate(desired_bitrate);
37 
38    // The resolution must be smaller than the fastest send rate we want to test.
39    l.set_forced_time_advance(Duration::from_micros(100));
40    r.set_forced_time_advance(Duration::from_micros(100));
41 
42    // Normalize time after DTLS connection to make tests deterministic across backends
43    // This ensures all crypto backends start from the same simulated time
44    let normalized_time = l.start + Duration::from_millis(100);
45 
46    l.last = normalized_time;
47    r.last = normalized_time;
48 
49    (l, r)
50}
51 
52/// Extract the last BWE estimate from events.
53pub fn get_last_bwe_estimate(rtc: &TestRtc) -> Option<Bitrate> {
54    rtc.events
55        .iter()
56        .filter_map(|(_, e)| {
57            if let Event::EgressBitrateEstimate(BweKind::Twcc(bitrate)) = e {
58                Some(bitrate)
59            } else {
60                None
61            }
62        })
63        .last()
64        .copied()
65}
66 
67/// BweTestContext holds state for running BWE tests.
68pub struct BweTestContext {
69    ssrc: Ssrc,
70    pt: Pt,
71    seq_no: u64,
72    /// Current media send rate (excluding padding/probes)
73    media_send_rate: Bitrate,
74    /// Accumulated byte budget for sending (smooths out timing variations)
75    byte_budget: f64,
76}
77 
78#[derive(Clone)]
79pub enum Step {
80    /// Network conditions
81    Conditions {
82        description: &'static str,
83        config: NetemConfig,
84    },
85    /// Send media
86    Media {
87        description: &'static str,
88        desired_bitrate: Bitrate,
89        media_send_rate: Bitrate,
90    },
91    /// Run simulation for duration
92    Run {
93        description: &'static str,
94        duration: Duration,
95    },
96    /// Check the latest BWE estimate
97    Check {
98        description: &'static str,
99        at_least: Bitrate,
100    },
101    CheckProbe {
102        description: &'static str,
103        check: Arc<dyn Fn(usize, &ProbeClusterConfig) -> bool>,
104    },
105    /// Assert no probes fired since last event_offset update
106    AssertNoProbes { description: &'static str },
107}
108 
109impl BweTestContext {
110    pub fn new(l: &mut TestRtc, r: &mut TestRtc) -> Self {
111        let mid = "vid".into();
112        let ssrc_tx: Ssrc = 42.into();
113        let ssrc_rtx: Ssrc = 44.into();
114 
115        l.direct_api().declare_media(mid, MediaKind::Video);
116        l.direct_api()
117            .declare_stream_tx(ssrc_tx, Some(ssrc_rtx), mid, None);
118 
119        r.direct_api().declare_media(mid, MediaKind::Video);
120        r.direct_api()
121            .expect_stream_rx(ssrc_tx, Some(ssrc_rtx), mid, None);
122 
123        // Enable TWCC feedback on the receiver so it sends feedback to the sender
124        r.direct_api().enable_twcc_feedback();
125 
126        // Sync time
127        let max = l.last.max(r.last);
128        l.last = max;
129        r.last = max;
130 
131        let params = l.params_vp8();
132        assert_eq!(params.spec().codec, Codec::Vp8);
133        let pt = params.pt();
134 
135        let ssrc = l.direct_api().stream_tx_by_mid(mid, None).unwrap().ssrc();
136 
137        Self {
138            ssrc,
139            pt,
140            seq_no: 47_000,
141            media_send_rate: Bitrate::mbps(2), // Default to 2 Mbps
142            byte_budget: 0.0,
143        }
144    }
145 
146    pub fn run_plan(
147        &mut self,
148        l: &mut TestRtc,
149        r: &mut TestRtc,
150        plan: &[Step],
151    ) -> Result<(), RtcError> {
152        let total = plan.len();
153 
154        let mut event_offset = 0;
155 
156        for (no, step) in plan.iter().enumerate() {
157            match step {
158                Step::Conditions {
159                    description,
160                    config,
161                } => {
162                    info!("{}/{}: Set conditions: {}", no + 1, total, description);
163                    l.set_netem(*config);
164                    r.set_netem(*config);
165                    event_offset = l.events.len();
166                }
167                Step::Media {
168                    description,
169                    desired_bitrate,
170                    media_send_rate,
171                } => {
172                    info!("{}/{}: Media rates: {}", no + 1, total, description);
173                    l.bwe().set_desired_bitrate(*desired_bitrate);
174                    self.set_media_send_rate(*media_send_rate);
175                    event_offset = l.events.len();
176                }
177                Step::Run {
178                    description,
179                    duration,
180                } => {
181                    info!("{}/{}: Run: {}", no + 1, total, description);
182                    self.run_for_duration(l, r, *duration)?;
183                    // Don't update event_offset here - let CheckProbe see events from this Run
184                }
185                Step::Check {
186                    description,
187                    at_least,
188                } => {
189                    let estimate = get_last_bwe_estimate(l).expect("a BWE estimate");
190                    info!(
191                        "{}/{}: Check estimate: {} ({} >= {})",
192                        no + 1,
193                        total,
194                        description,
195                        estimate,
196                        *at_least
197                    );
198 
199                    let is_ok = estimate >= *at_least;
200 
201                    assert!(
202                        is_ok,
203                        "Step {} estimate {} should be at least {}",
204                        no + 1,
205                        estimate,
206                        *at_least
207                    );
208                    // Check does not reset event offset.
209                }
210                Step::CheckProbe { description, check } => {
211                    info!("{}/{}: Check probe: {}", no + 1, total, description);
212 
213                    // All probes since previous step started.
214                    let probes = l.events[event_offset..]
215                        .iter()
216                        .filter_map(|e| match e {
217                            (_, Event::Probe(probe)) => Some(probe),
218                            _ => None,
219                        })
220                        .enumerate();
221 
222                    let mut any_ok = false;
223 
224                    for (index, probe) in probes {
225                        let is_ok = (check)(index, probe);
226                        if is_ok {
227                            any_ok = true;
228                            break;
229                        }
230                    }
231 
232                    assert!(any_ok, "No probe check passed");
233 
234                    // Check does not reset event offset.
235                }
236                Step::AssertNoProbes { description } => {
237                    info!("{}/{}: Assert no probes: {}", no + 1, total, description);
238 
239                    // Count probes since previous step started.
240                    let probe_count = l.events[event_offset..]
241                        .iter()
242                        .filter(|e| matches!(e, (_, Event::Probe(_))))
243                        .count();
244 
245                    assert_eq!(
246                        probe_count, 0,
247                        "Expected no probes, but {} probe(s) fired",
248                        probe_count
249                    );
250 
251                    // AssertNoProbes does not reset event offset.
252                }
253            }
254        }
255 
256        Ok(())
257    }
258 
259    pub fn set_media_send_rate(&mut self, media_send_rate: Bitrate) {
260        self.media_send_rate = media_send_rate;
261    }
262 
263    /// Run traffic for specified duration
264    pub fn run_for_duration(
265        &mut self,
266        l: &mut TestRtc,
267        r: &mut TestRtc,
268        duration: Duration,
269    ) -> Result<Option<Bitrate>, RtcError> {
270        let start_duration = l.duration();
271        let end_time = start_duration + duration;
272 
273        let mut last_send_time = l.duration();
274 
275        while l.duration() < end_time {
276            let current_time = l.duration();
277 
278            // Calculate elapsed time since last send
279            let elapsed = current_time.saturating_sub(last_send_time);
280            last_send_time = current_time;
281 
282            // Accumulate byte budget based on elapsed time and target bitrate
283            // This smooths out timing variations by allowing budget to carry over
284            let elapsed_secs = elapsed.as_secs_f64();
285            let bytes_earned = (self.media_send_rate.as_u64() as f64 / 8.0) * elapsed_secs;
286            self.byte_budget += bytes_earned;
287 
288            // Use packet size of ~1150 bytes (allows RTX probe reuse)
289            let packet_size = 1150;
290 
291            let mut did_progress = false;
292 
293            // Send packets while we have budget
294            // IMPORTANT: Must call progress() after EACH write_rtp to consume outputs
295            while self.byte_budget >= packet_size as f64 {
296                // Calculate wallclock for THIS packet at current simulated time
297                let wallclock = l.start + l.duration();
298                let time = (self.seq_no * 1000 + 47_000_000) as u32;
299 
300                let mut direct = l.direct_api();
301                let stream = direct.stream_tx(&self.ssrc).unwrap();
302 
303                // Send a video packet
304                let payload = vec![0u8; packet_size];
305                stream.write_rtp(
306                    RtpWrite::new(self.pt, self.seq_no.into(), time, wallclock, payload)
307                        .nackable(true),
308                );
309 
310                self.seq_no += 1;
311                self.byte_budget -= packet_size as f64;
312 
313                // Must progress after each write_rtp to consume outputs
314                progress(l, r)?;
315                did_progress = true;
316            }
317 
318            if !did_progress {
319                progress(l, r)?;
320            }
321        }
322 
323        Ok(get_last_bwe_estimate(l))
324    }
325}