1use super::{
8 CongestionControlAlgorithm, CongestionSignals, CongestionWindow, State,
9 params::{Algorithm, VegasParams},
10 rtt::{ClockStall, RoundtripTimeEstimator},
11};
12use crate::Result;
13
14use tor_error::{error_report, internal};
15
16#[derive(Clone, Debug, Default)]
20pub(crate) struct BdpEstimator {
21 bdp: u32,
23}
24
25impl BdpEstimator {
26 fn get(&self) -> u32 {
28 self.bdp
29 }
30
31 fn update(
36 &mut self,
37 cwnd: &CongestionWindow,
38 rtt: &RoundtripTimeEstimator,
39 signals: &CongestionSignals,
40 ) {
41 if rtt.clock_stalled() {
43 self.bdp = if signals.channel_blocked {
44 cwnd.get()
47 .saturating_sub(signals.channel_outbound_size)
48 .max(cwnd.min())
49 } else {
50 cwnd.get()
51 };
52 } else {
53 let min_rtt_usec = rtt
59 .min_rtt()
60 .and_then(|x| u32::try_from(x.as_micros()).ok())
61 .unwrap_or(u32::MAX);
62 let ewma_rtt_usec = rtt
63 .ewma_rtt()
64 .and_then(|x| u32::try_from(x.as_micros()).ok())
65 .unwrap_or(u32::MAX);
66 self.bdp = cwnd
67 .get()
68 .saturating_mul(min_rtt_usec)
69 .saturating_div(ewma_rtt_usec.max(1));
71 }
72 }
73}
74
75#[derive(Clone, Debug)]
85pub(crate) struct Vegas {
86 params: VegasParams,
88 bdp: BdpEstimator,
91 cwnd: CongestionWindow,
94 num_cell_until_sendme: u32,
96 num_sendme_until_cwnd_update: u32,
99 num_sendme_per_cwnd: u32,
102 num_inflight: u32,
105 is_blocked_on_chan: bool,
109}
110
111impl Vegas {
112 pub(crate) fn new(params: VegasParams, state: &State, cwnd: CongestionWindow) -> Self {
114 Self {
115 params,
116 bdp: BdpEstimator::default(),
117 num_cell_until_sendme: cwnd.sendme_inc(),
118 num_inflight: 0,
119 num_sendme_per_cwnd: 0,
120 num_sendme_until_cwnd_update: cwnd.update_rate(state),
121 cwnd,
122 is_blocked_on_chan: false,
123 }
124 }
125}
126
127impl CongestionControlAlgorithm for Vegas {
128 fn uses_stream_sendme(&self) -> bool {
129 false
131 }
132
133 fn uses_xon_xoff(&self) -> bool {
134 true
135 }
136
137 fn is_next_cell_sendme(&self) -> bool {
138 self.num_inflight.is_multiple_of(self.cwnd.sendme_inc())
141 }
142
143 fn can_send(&self) -> bool {
144 self.num_inflight < self.cwnd.get()
145 }
146
147 fn cwnd(&self) -> Option<CongestionWindow> {
148 Some(self.cwnd)
149 }
150
151 fn sendme_received(
162 &mut self,
163 state: &mut State,
164 rtt: &mut RoundtripTimeEstimator,
165 signals: CongestionSignals,
166 clock_stall: ClockStall,
167 ) -> Result<()> {
168 self.num_sendme_until_cwnd_update = self.num_sendme_until_cwnd_update.saturating_sub(1);
170 self.num_sendme_per_cwnd = self.num_sendme_per_cwnd.saturating_sub(1);
172
173 if clock_stall == ClockStall::Detected {
176 self.num_inflight = self.num_inflight.saturating_sub(self.cwnd.sendme_inc());
178 return Ok(());
179 }
180
181 self.bdp.update(&self.cwnd, rtt, &signals);
188
189 if rtt.is_ready() {
192 if signals.channel_blocked {
193 if !self.is_blocked_on_chan {
196 self.num_sendme_until_cwnd_update = 0;
197 }
198 } else {
199 if self.is_blocked_on_chan {
202 self.num_sendme_until_cwnd_update = 0;
203 }
204 }
205 }
206 self.is_blocked_on_chan = signals.channel_blocked;
207
208 if !rtt.is_ready() && !self.is_blocked_on_chan {
210 debug_assert!(self.num_inflight >= self.cwnd.sendme_inc());
214 self.num_inflight = self.num_inflight.saturating_sub(self.cwnd.sendme_inc());
215 return Ok(());
216 }
217
218 let queue_use = self.cwnd.get().saturating_sub(self.bdp.get());
221
222 self.cwnd.eval_fullness(
224 self.num_inflight,
225 self.params.cwnd_full_gap(),
226 self.params.cwnd_full_min_pct().as_percent(),
227 );
228
229 if state.in_slow_start() {
231 if queue_use < self.params.cell_in_queue_params().gamma() && !self.is_blocked_on_chan {
232 if self.cwnd.is_full() {
234 let inc = self
236 .cwnd
237 .rfc3742_ss_inc(self.params.cell_in_queue_params().ss_cwnd_cap());
238
239 if (inc * self.cwnd.sendme_per_cwnd())
242 <= (self.cwnd.increment() * self.cwnd.increment_rate())
243 {
244 *state = State::Steady;
245 }
246 }
247 } else {
248 self.cwnd
250 .set(self.bdp.get() + self.params.cell_in_queue_params().gamma());
251 *state = State::Steady;
253 }
254
255 if self.cwnd.get() >= self.params.ss_cwnd_max() {
257 self.cwnd.set(self.params.ss_cwnd_max());
258 *state = State::Steady;
259 }
260 } else if self.num_sendme_until_cwnd_update == 0 {
261 if queue_use > self.params.cell_in_queue_params().delta() {
263 self.cwnd.set(
265 self.bdp.get() + self.params.cell_in_queue_params().delta()
266 - self.cwnd.increment(),
267 );
268 } else if queue_use > self.params.cell_in_queue_params().beta()
269 || self.is_blocked_on_chan
270 {
271 self.cwnd.dec();
273 } else if self.cwnd.is_full() && queue_use < self.params.cell_in_queue_params().alpha()
274 {
275 self.cwnd.inc();
277 }
278 }
279
280 if self.num_sendme_until_cwnd_update == 0 {
282 self.num_sendme_until_cwnd_update = self.cwnd.update_rate(state);
283 }
284 if self.num_sendme_per_cwnd == 0 {
285 self.num_sendme_per_cwnd = self.cwnd.sendme_per_cwnd();
286 }
287
288 if self.params.cwnd_full_per_cwnd() != 0 {
290 if self.num_sendme_per_cwnd == self.cwnd.sendme_per_cwnd() {
291 self.cwnd.reset_full();
292 }
293 } else if self.num_sendme_until_cwnd_update == self.cwnd.update_rate(state) {
294 self.cwnd.reset_full();
295 }
296
297 self.num_inflight = self.num_inflight.saturating_sub(self.cwnd.sendme_inc());
299 Ok(())
300 }
301
302 fn sendme_sent(&mut self) -> Result<()> {
303 self.num_cell_until_sendme = self.cwnd.sendme_inc();
305 Ok(())
306 }
307
308 fn data_received(&mut self) -> Result<bool> {
309 if self.num_cell_until_sendme == 0 {
310 error_report!(internal!("Congestion control unexptected data cell"), "");
314 return Ok(false);
315 }
316
317 self.num_cell_until_sendme = self.num_cell_until_sendme.saturating_sub(1);
319
320 Ok(self.num_cell_until_sendme == 0)
323 }
324
325 fn data_sent(&mut self) -> Result<()> {
326 self.num_inflight = self.num_inflight.saturating_add(1);
328 Ok(())
329 }
330
331 #[cfg(feature = "conflux")]
332 fn inflight(&self) -> Option<u32> {
333 Some(self.num_inflight)
334 }
335
336 #[cfg(test)]
337 fn send_window(&self) -> u32 {
338 self.cwnd.get()
339 }
340
341 fn algorithm(&self) -> Algorithm {
342 Algorithm::Vegas(self.params)
343 }
344}
345
346#[cfg(test)]
347pub(crate) mod test {
348 #![allow(clippy::bool_assert_comparison)]
350 #![allow(clippy::clone_on_copy)]
351 #![allow(clippy::dbg_macro)]
352 #![allow(clippy::mixed_attributes_style)]
353 #![allow(clippy::print_stderr)]
354 #![allow(clippy::print_stdout)]
355 #![allow(clippy::single_char_pattern)]
356 #![allow(clippy::unwrap_used)]
357 #![allow(clippy::unchecked_time_subtraction)]
358 #![allow(clippy::useless_vec)]
359 #![allow(clippy::needless_pass_by_value)]
360 #![allow(clippy::string_slice)] use std::collections::VecDeque;
364 use tor_units::Percentage;
365 use web_time_compat::{Duration, Instant, InstantExt};
366
367 use super::*;
368 use crate::congestion::{
369 params::VegasParamsBuilder,
370 test_utils::{new_cwnd, new_rtt_estimator},
371 };
372
373 impl Vegas {
374 pub(crate) fn set_inflight(&mut self, v: u32) {
376 self.num_inflight = v;
377 }
378 fn is_blocked_on_chan(&self) -> bool {
380 self.is_blocked_on_chan
381 }
382 fn set_is_blocked_on_chan(&mut self, v: bool) {
384 self.is_blocked_on_chan = v;
385 }
386 }
387
388 #[derive(Debug)]
391 struct TestVectorParams {
392 sent_usec_in: u64,
394 got_sendme_usec_in: u64,
395 or_conn_blocked_in: bool,
396 inflight_in: u32,
397 ewma_rtt_out: Duration,
399 min_rtt_out: Duration,
400 cwnd_out: u32,
401 in_slow_start_out: bool,
402 cwnd_full_out: bool,
403 blocked_chan_out: bool,
404 }
405
406 impl From<[u32; 10]> for TestVectorParams {
407 fn from(arr: [u32; 10]) -> Self {
408 Self {
409 sent_usec_in: u64::from(arr[0]),
410 got_sendme_usec_in: u64::from(arr[1]),
411 or_conn_blocked_in: arr[2] == 1,
412 inflight_in: arr[3],
413 ewma_rtt_out: Duration::from_micros(arr[4].into()),
414 min_rtt_out: Duration::from_micros(arr[5].into()),
415 cwnd_out: arr[6],
416 in_slow_start_out: arr[7] == 1,
417 cwnd_full_out: arr[8] == 1,
418 blocked_chan_out: arr[9] == 1,
419 }
420 }
421 }
422
423 struct VegasTest {
424 params: VecDeque<TestVectorParams>,
425 rtt: RoundtripTimeEstimator,
426 state: State,
427 vegas: Vegas,
428 }
429
430 impl VegasTest {
431 fn new(vec: Vec<[u32; 10]>) -> Self {
432 let mut params = VecDeque::new();
433 for values in vec {
434 params.push_back(values.into());
435 }
436 let state = State::default();
437 Self {
438 params,
439 rtt: new_rtt_estimator(),
440 vegas: Vegas::new(build_vegas_params(), &state, new_cwnd()),
441 state,
442 }
443 }
444
445 fn run_once(&mut self, p: &TestVectorParams) {
446 eprintln!("Testing vector: {:?}", p);
447 self.vegas.set_inflight(p.inflight_in);
449 self.vegas.set_is_blocked_on_chan(p.or_conn_blocked_in);
450
451 let now = Instant::get();
452 self.rtt
453 .expect_sendme(now + Duration::from_micros(p.sent_usec_in));
454 let ret = self.rtt.update(
455 now + Duration::from_micros(p.got_sendme_usec_in),
456 &self.state,
457 &self.vegas.cwnd().expect("No CWND"),
458 );
459 assert!(ret.is_ok());
460
461 let signals = CongestionSignals::new(p.or_conn_blocked_in, 0);
462 let clock_stall = ClockStall::NotDetected;
463 let ret =
464 self.vegas
465 .sendme_received(&mut self.state, &mut self.rtt, signals, clock_stall);
466 assert!(ret.is_ok());
467
468 assert_eq!(self.rtt.ewma_rtt().unwrap(), p.ewma_rtt_out);
469 assert_eq!(self.rtt.min_rtt().unwrap(), p.min_rtt_out);
470 assert_eq!(self.vegas.cwnd().expect("No CWND").get(), p.cwnd_out);
471 assert_eq!(
472 self.vegas.cwnd().expect("No CWND").is_full(),
473 p.cwnd_full_out
474 );
475 assert_eq!(self.state.in_slow_start(), p.in_slow_start_out);
476 assert_eq!(self.vegas.is_blocked_on_chan(), p.blocked_chan_out);
477 }
478
479 fn run(&mut self) {
480 while let Some(param) = self.params.pop_front() {
481 self.run_once(¶m);
482 }
483 }
484 }
485
486 pub(crate) fn build_vegas_params() -> VegasParams {
487 const OUTBUF_CELLS: u32 = 62;
488 VegasParamsBuilder::default()
489 .cell_in_queue_params(
490 (
491 3 * OUTBUF_CELLS, 4 * OUTBUF_CELLS, 5 * OUTBUF_CELLS, 3 * OUTBUF_CELLS, 600, )
497 .into(),
498 )
499 .ss_cwnd_max(5_000)
500 .cwnd_full_gap(4)
501 .cwnd_full_min_pct(Percentage::new(25))
502 .cwnd_full_per_cwnd(1)
503 .build()
504 .expect("Unable to build Vegas parameters")
505 }
506
507 #[test]
508 fn test_vectors() {
509 let vec1 = vec![
510 [100000, 200000, 0, 124, 100000, 100000, 155, 1, 0, 0],
511 [200000, 300000, 0, 155, 100000, 100000, 186, 1, 1, 0],
512 [350000, 500000, 0, 186, 133333, 100000, 217, 1, 1, 0],
513 [500000, 550000, 0, 217, 77777, 77777, 248, 1, 1, 0],
514 [600000, 700000, 0, 248, 92592, 77777, 279, 1, 1, 0],
515 [700000, 750000, 0, 279, 64197, 64197, 310, 1, 0, 0], [750000, 875000, 0, 310, 104732, 64197, 341, 1, 1, 0],
517 [875000, 900000, 0, 341, 51577, 51577, 372, 1, 1, 0],
518 [900000, 950000, 0, 279, 50525, 50525, 403, 1, 1, 0],
519 [950000, 1000000, 0, 279, 50175, 50175, 434, 1, 1, 0],
520 [1000000, 1050000, 0, 279, 50058, 50058, 465, 1, 1, 0],
521 [1050000, 1100000, 0, 279, 50019, 50019, 496, 1, 1, 0],
522 [1100000, 1150000, 0, 279, 50006, 50006, 527, 1, 1, 0],
523 [1150000, 1200000, 0, 279, 50002, 50002, 558, 1, 1, 0],
524 [1200000, 1250000, 0, 550, 50000, 50000, 589, 1, 1, 0],
525 [1250000, 1300000, 0, 550, 50000, 50000, 620, 1, 0, 0], [1300000, 1350000, 0, 550, 50000, 50000, 635, 1, 1, 0],
527 [1350000, 1400000, 0, 550, 50000, 50000, 650, 1, 1, 0],
528 [1400000, 1450000, 0, 150, 50000, 50000, 650, 1, 0, 0], [1450000, 1500000, 0, 150, 50000, 50000, 650, 1, 0, 0], [1500000, 1550000, 0, 550, 50000, 50000, 664, 1, 1, 0], [1500000, 1600000, 0, 550, 83333, 50000, 584, 0, 1, 0], [1600000, 1650000, 0, 550, 61111, 50000, 585, 0, 1, 0], [1650000, 1700000, 0, 550, 53703, 50000, 586, 0, 1, 0],
534 [1700000, 1750000, 0, 100, 51234, 50000, 586, 0, 0, 0], [1750000, 1900000, 0, 100, 117078, 50000, 559, 0, 0, 0], [1900000, 2000000, 0, 100, 105692, 50000, 558, 0, 0, 0], [2000000, 2075000, 0, 500, 85230, 50000, 558, 0, 1, 0], [2075000, 2125000, 1, 500, 61743, 50000, 557, 0, 1, 1], [2125000, 2150000, 0, 500, 37247, 37247, 558, 0, 1, 0], [2150000, 2350000, 0, 500, 145749, 37247, 451, 0, 1, 0], ];
542 VegasTest::new(vec1).run();
543
544 let vec2 = vec![
545 [100000, 200000, 0, 124, 100000, 100000, 155, 1, 0, 0],
546 [200000, 300000, 0, 155, 100000, 100000, 186, 1, 1, 0],
547 [350000, 500000, 0, 186, 133333, 100000, 217, 1, 1, 0],
548 [500000, 550000, 1, 217, 77777, 77777, 403, 0, 1, 1], [600000, 700000, 0, 248, 92592, 77777, 404, 0, 1, 0], [700000, 750000, 1, 404, 64197, 64197, 403, 0, 0, 1], [750000, 875000, 0, 403, 104732, 64197, 404, 0, 1, 0],
552 ];
553 VegasTest::new(vec2).run();
554
555 let vec3 = vec![
556 [18258527, 19002938, 0, 83, 744411, 744411, 155, 1, 0, 0],
557 [18258580, 19254257, 0, 52, 911921, 744411, 186, 1, 1, 0],
558 [20003224, 20645298, 0, 164, 732023, 732023, 217, 1, 1, 0],
559 [20003367, 21021444, 0, 133, 922725, 732023, 248, 1, 1, 0],
560 [20003845, 21265508, 0, 102, 1148683, 732023, 279, 1, 1, 0],
561 [20003975, 21429157, 0, 71, 1333015, 732023, 310, 1, 0, 0],
562 [20004309, 21707677, 0, 40, 1579917, 732023, 310, 1, 0, 0],
563 ];
564 VegasTest::new(vec3).run();
565
566 let vec4 = vec![
567 [358297091, 358854163, 0, 83, 557072, 557072, 155, 1, 0, 0],
568 [358297649, 359123845, 0, 52, 736488, 557072, 186, 1, 1, 0],
569 [359492879, 359995330, 0, 186, 580463, 557072, 217, 1, 1, 0],
570 [359493043, 360489243, 0, 217, 857621, 557072, 248, 1, 1, 0],
571 [359493232, 360489673, 0, 248, 950167, 557072, 279, 1, 1, 0],
572 [359493795, 360489971, 0, 279, 980839, 557072, 310, 1, 0, 0],
573 [359493918, 360490248, 0, 310, 991166, 557072, 341, 1, 1, 0],
574 [359494029, 360716465, 0, 341, 1145346, 557072, 372, 1, 1, 0],
575 [359996888, 360948867, 0, 372, 1016434, 557072, 403, 1, 1, 0],
576 [359996979, 360949330, 0, 403, 973712, 557072, 434, 1, 1, 0],
577 [360489528, 361113615, 0, 434, 740628, 557072, 465, 1, 1, 0],
578 [360489656, 361281604, 0, 465, 774841, 557072, 496, 1, 1, 0],
579 [360489837, 361500461, 0, 496, 932029, 557072, 482, 0, 1, 0],
580 [360489963, 361500631, 0, 482, 984455, 557072, 482, 0, 1, 0],
581 [360490117, 361842481, 0, 482, 1229727, 557072, 481, 0, 1, 0],
582 ];
583 VegasTest::new(vec4).run();
584 }
585}