1use std::time::Duration;
28use std::time::Instant;
29
30const WINDOW_INCREASE_FACTOR: u64 = 2;
33
34const WINDOW_TRIGGER_FACTOR: u32 = 2;
37
38#[derive(Default, Debug)]
39pub struct FlowControl {
40 consumed: u64,
42
43 max_data: u64,
45
46 window: u64,
49
50 max_window: u64,
52
53 last_update: Option<Instant>,
55}
56
57impl FlowControl {
58 pub fn new(max_data: u64, window: u64, max_window: u64) -> Self {
59 Self {
60 max_data,
61
62 window: std::cmp::min(window, max_window),
63
64 max_window,
65
66 ..Default::default()
67 }
68 }
69
70 pub fn window(&self) -> u64 {
72 self.window
73 }
74
75 pub fn max_data(&self) -> u64 {
77 self.max_data
78 }
79
80 #[cfg(test)]
82 pub fn consumed(&self) -> u64 {
83 self.consumed
84 }
85
86 pub fn add_consumed(&mut self, consumed: u64) {
88 self.consumed += consumed;
89 }
90
91 pub fn should_update_max_data(&self) -> bool {
96 let available_window = self.max_data - self.consumed;
97
98 available_window < (self.window / 2)
99 }
100
101 pub fn max_data_next(&self) -> u64 {
103 self.consumed + self.window
104 }
105
106 pub fn update_max_data(&mut self, now: Instant) {
108 self.max_data = self.max_data_next();
109 self.last_update = Some(now);
110 }
111
112 pub fn autotune_window(&mut self, now: Instant, rtt: Duration) {
116 if let Some(last_update) = self.last_update {
117 if now - last_update < rtt * WINDOW_TRIGGER_FACTOR {
118 self.set_window(self.window * WINDOW_INCREASE_FACTOR);
119 }
120 }
121 }
122
123 fn set_window(&mut self, window: u64) {
124 self.window = std::cmp::min(window, self.max_window);
125 }
126
127 pub fn ensure_window_lower_bound(&mut self, min_window: u64) {
130 if min_window > self.window {
131 self.set_window(min_window);
133 }
134 }
135}
136
137#[cfg(test)]
138mod tests {
139 use super::*;
140
141 #[test]
142 fn max_window_in_new() {
143 let fc = FlowControl::new(100, 100, 50);
144 assert_eq!(fc.max_data(), 100);
145 assert_eq!(fc.window, 50);
146 }
147
148 #[test]
149 fn max_data() {
150 let fc = FlowControl::new(100, 20, 100);
151
152 assert_eq!(fc.max_data(), 100);
153 }
154
155 #[test]
156 fn should_update_max_data() {
157 let mut fc = FlowControl::new(100, 20, 100);
158
159 fc.add_consumed(85);
160 assert!(!fc.should_update_max_data());
161
162 fc.add_consumed(10);
163 assert!(fc.should_update_max_data());
164 }
165
166 #[test]
167 fn max_data_next() {
168 let mut fc = FlowControl::new(100, 20, 100);
169
170 let consumed = 95;
171
172 fc.add_consumed(consumed);
173 assert!(fc.should_update_max_data());
174 assert_eq!(fc.max_data_next(), consumed + 20);
175 }
176
177 #[test]
178 fn update_max_data() {
179 let mut fc = FlowControl::new(100, 20, 100);
180
181 let consumed = 95;
182
183 fc.add_consumed(consumed);
184 assert!(fc.should_update_max_data());
185
186 let max_data_next = fc.max_data_next();
187 assert_eq!(fc.max_data_next(), consumed + 20);
188
189 fc.update_max_data(Instant::now());
190 assert_eq!(fc.max_data(), max_data_next);
191 }
192
193 #[test]
194 fn autotune_window() {
195 let w = 20;
196 let mut fc = FlowControl::new(100, w, 100);
197
198 let consumed = 95;
199
200 fc.add_consumed(consumed);
201 assert!(fc.should_update_max_data());
202
203 let max_data_next = fc.max_data_next();
204 assert_eq!(max_data_next, consumed + w);
205
206 fc.update_max_data(Instant::now());
207 assert_eq!(fc.max_data(), max_data_next);
208
209 fc.autotune_window(Instant::now(), Duration::from_millis(100));
211
212 let w = w * 2;
213 let consumed_inc = 15;
214
215 fc.add_consumed(consumed_inc);
216 assert!(fc.should_update_max_data());
217
218 let max_data_next = fc.max_data_next();
219 assert_eq!(max_data_next, consumed + consumed_inc + w);
220 }
221
222 #[test]
223 fn ensure_window_lower_bound() {
224 let w = 20;
225 let mut fc = FlowControl::new(100, w, 100);
226
227 fc.ensure_window_lower_bound(w);
229 assert_eq!(fc.window(), 20);
230
231 fc.ensure_window_lower_bound(w * 2);
233 assert_eq!(fc.window(), 40);
234
235 fc.ensure_window_lower_bound(101);
237 assert_eq!(fc.window(), 100);
238 }
239}