1use crate::interfaces::{AirtimeDutyCycle, AirtimeUtilization};
2use heapless::Deque;
3use heapless::Vec as HeaplessVec;
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub enum DutyPush {
7 Queued,
8 Refused,
9}
10
11pub trait DutyQueue: Default {
12 fn push_back(&mut self, bytes: &[u8], airtime_us: u64) -> DutyPush;
13 fn pop_front_with<R>(&mut self, f: impl FnOnce(&[u8]) -> R) -> Option<(u64, R)>;
14 fn is_empty(&self) -> bool;
15 fn is_full(&self) -> bool;
16}
17
18struct Queued<F> {
19 airtime_us: u64,
20 frame: F,
21}
22
23#[derive(Default)]
24pub struct FixedDutyQueue<const FRAMES: usize, const MTU: usize> {
25 entries: Deque<Queued<HeaplessVec<u8, MTU>>, FRAMES>,
26}
27
28impl<const FRAMES: usize, const MTU: usize> DutyQueue for FixedDutyQueue<FRAMES, MTU> {
29 fn push_back(&mut self, bytes: &[u8], airtime_us: u64) -> DutyPush {
30 let mut frame = HeaplessVec::new();
31 if frame.extend_from_slice(bytes).is_err() {
32 return DutyPush::Refused;
33 }
34 match self.entries.push_back(Queued { airtime_us, frame }) {
35 Ok(()) => DutyPush::Queued,
36 Err(_) => DutyPush::Refused,
37 }
38 }
39
40 fn pop_front_with<R>(&mut self, f: impl FnOnce(&[u8]) -> R) -> Option<(u64, R)> {
41 let entry = self.entries.pop_front()?;
42 Some((entry.airtime_us, f(entry.frame.as_slice())))
43 }
44
45 fn is_empty(&self) -> bool {
46 self.entries.is_empty()
47 }
48
49 fn is_full(&self) -> bool {
50 self.entries.is_full()
51 }
52}
53
54#[cfg(feature = "alloc")]
55pub use heap::HeapDutyQueue;
56
57#[cfg(feature = "alloc")]
58mod heap {
59 use super::{DutyPush, DutyQueue, Queued};
60 use alloc::collections::VecDeque;
61 use alloc::vec::Vec;
62
63 #[derive(Default)]
64 pub struct HeapDutyQueue {
65 entries: VecDeque<Queued<Vec<u8>>>,
66 }
67
68 impl DutyQueue for HeapDutyQueue {
69 fn push_back(&mut self, bytes: &[u8], airtime_us: u64) -> DutyPush {
70 self.entries.push_back(Queued {
71 airtime_us,
72 frame: bytes.to_vec(),
73 });
74 DutyPush::Queued
75 }
76
77 fn pop_front_with<R>(&mut self, f: impl FnOnce(&[u8]) -> R) -> Option<(u64, R)> {
78 let entry = self.entries.pop_front()?;
79 Some((entry.airtime_us, f(&entry.frame)))
80 }
81
82 fn is_empty(&self) -> bool {
83 self.entries.is_empty()
84 }
85
86 fn is_full(&self) -> bool {
87 false
88 }
89 }
90}
91
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub enum DutyVerdict {
94 Transmit,
95 Held,
96}
97
98pub struct DutyGate<Q: DutyQueue> {
99 queue: Q,
100 queued_airtime_us: u64,
101 dropped: u64,
102}
103
104impl<Q: DutyQueue> Default for DutyGate<Q> {
105 fn default() -> Self {
106 Self::new()
107 }
108}
109
110impl<Q: DutyQueue> DutyGate<Q> {
111 #[must_use]
112 pub fn new() -> Self {
113 Self {
114 queue: Q::default(),
115 queued_airtime_us: 0,
116 dropped: 0,
117 }
118 }
119
120 pub fn offer(
121 &mut self,
122 wire: &[u8],
123 airtime_us: u64,
124 projected_utilization: AirtimeUtilization,
125 duty: &AirtimeDutyCycle,
126 ) -> DutyVerdict {
127 if self.queue.is_empty() && duty.permits(projected_utilization) {
128 return DutyVerdict::Transmit;
129 }
130 self.enqueue(wire, airtime_us, duty);
131 DutyVerdict::Held
132 }
133
134 pub fn release_ready(
135 &mut self,
136 projected_utilization: AirtimeUtilization,
137 duty: &AirtimeDutyCycle,
138 send: impl FnOnce(&[u8]),
139 ) -> bool {
140 if !duty.permits(projected_utilization) {
141 return false;
142 }
143 let Some((airtime_us, ())) = self.queue.pop_front_with(send) else {
144 return false;
145 };
146 self.queued_airtime_us = self.queued_airtime_us.saturating_sub(airtime_us);
147 true
148 }
149
150 #[must_use]
151 pub fn is_empty(&self) -> bool {
152 self.queue.is_empty()
153 }
154
155 #[must_use]
156 pub fn dropped_count(&self) -> u64 {
157 self.dropped
158 }
159
160 fn enqueue(&mut self, wire: &[u8], airtime_us: u64, duty: &AirtimeDutyCycle) {
161 let budget_us = u64::from(duty.max_queued_airtime_ms).saturating_mul(1_000);
162 if airtime_us > budget_us {
163 self.dropped += 1;
164 return;
165 }
166 while self.queued_airtime_us.saturating_add(airtime_us) > budget_us || self.queue.is_full()
167 {
168 let Some((evicted_airtime_us, ())) = self.queue.pop_front_with(|_| ()) else {
169 break;
170 };
171 self.queued_airtime_us = self.queued_airtime_us.saturating_sub(evicted_airtime_us);
172 self.dropped += 1;
173 }
174 match self.queue.push_back(wire, airtime_us) {
175 DutyPush::Queued => {
176 self.queued_airtime_us = self.queued_airtime_us.saturating_add(airtime_us);
177 }
178 DutyPush::Refused => self.dropped += 1,
179 }
180 }
181}
182
183#[cfg(test)]
184mod tests {
185 use super::*;
186
187 const DUTY: AirtimeDutyCycle = AirtimeDutyCycle {
188 limit_short_per_mille: Some(100),
189 limit_long_per_mille: None,
190 max_queued_airtime_ms: 2_000,
191 };
192
193 fn idle() -> AirtimeUtilization {
194 AirtimeUtilization {
195 short_per_mille: 0,
196 long_per_mille: 0,
197 }
198 }
199
200 fn saturated() -> AirtimeUtilization {
201 AirtimeUtilization {
202 short_per_mille: 101,
203 long_per_mille: 0,
204 }
205 }
206
207 fn permitted() -> AirtimeUtilization {
208 AirtimeUtilization {
209 short_per_mille: 99,
210 long_per_mille: 0,
211 }
212 }
213
214 fn transmits_under_limit_holds_over_it<Q: DutyQueue>() {
215 let mut gate: DutyGate<Q> = DutyGate::new();
216 assert_eq!(
217 gate.offer(&[1, 2, 3], 50_000, idle(), &DUTY),
218 DutyVerdict::Transmit
219 );
220 assert!(gate.is_empty());
221
222 assert_eq!(
223 gate.offer(&[7; 10], 50_000, saturated(), &DUTY),
224 DutyVerdict::Held
225 );
226
227 let mut sent = std::vec::Vec::new();
228 assert!(!gate.release_ready(saturated(), &DUTY, |bytes| sent.push(bytes.to_vec())));
229 assert!(sent.is_empty(), "still over the limit");
230
231 assert!(gate.release_ready(permitted(), &DUTY, |bytes| sent.push(bytes.to_vec())));
232 assert_eq!(sent, std::vec![std::vec![7u8; 10]]);
233 assert!(gate.is_empty());
234 assert_eq!(gate.dropped_count(), 0);
235 }
236
237 fn budgets_the_queue_in_airtime<Q: DutyQueue>() {
238 let mut gate: DutyGate<Q> = DutyGate::new();
239 gate.offer(&[1], 900_000, saturated(), &DUTY);
240 gate.offer(&[2], 900_000, saturated(), &DUTY);
241 assert_eq!(gate.dropped_count(), 0, "1.8s of the 2s budget queued");
242
243 gate.offer(&[3], 900_000, saturated(), &DUTY);
244 assert_eq!(
245 gate.dropped_count(),
246 1,
247 "the oldest fell out to fit the newcomer"
248 );
249
250 let mut sent = std::vec::Vec::new();
251 while gate.release_ready(permitted(), &DUTY, |bytes| sent.push(bytes.to_vec())) {}
252 assert_eq!(
253 sent,
254 std::vec![std::vec![2u8], std::vec![3u8]],
255 "FIFO order among the survivors",
256 );
257
258 gate.offer(&[9; 5], 2_500_000, saturated(), &DUTY);
259 assert!(
260 gate.is_empty(),
261 "a frame bigger than the whole budget drops"
262 );
263 assert_eq!(gate.dropped_count(), 2);
264 }
265
266 const TEST_MTU: usize = 500;
267
268 #[test]
269 fn the_fixed_queue_transmits_under_limit_holds_over_it() {
270 transmits_under_limit_holds_over_it::<FixedDutyQueue<8, TEST_MTU>>();
271 }
272
273 #[test]
274 fn the_fixed_queue_budgets_in_airtime() {
275 budgets_the_queue_in_airtime::<FixedDutyQueue<8, TEST_MTU>>();
276 }
277
278 #[test]
279 fn the_fixed_frame_capacity_is_only_the_allocation_ceiling() {
280 let mut gate: DutyGate<FixedDutyQueue<2, TEST_MTU>> = DutyGate::new();
281 gate.offer(&[1], 100_000, saturated(), &DUTY);
282 gate.offer(&[2], 100_000, saturated(), &DUTY);
283 gate.offer(&[3], 100_000, saturated(), &DUTY);
284 assert_eq!(
285 gate.dropped_count(),
286 1,
287 "well under the airtime budget, the ring itself still bounds memory",
288 );
289 }
290
291 #[test]
292 fn an_oversized_frame_is_refused_without_phantom_airtime() {
293 let mut gate: DutyGate<FixedDutyQueue<4, 8>> = DutyGate::new();
294 assert_eq!(
295 gate.offer(&[9; 9], 100_000, saturated(), &DUTY),
296 DutyVerdict::Held
297 );
298 assert!(gate.is_empty(), "nine bytes cannot be held in 8-byte slots");
299 assert_eq!(gate.dropped_count(), 1);
300
301 gate.offer(&[1; 8], 900_000, saturated(), &DUTY);
302 gate.offer(&[2; 8], 900_000, saturated(), &DUTY);
303 assert_eq!(
304 gate.dropped_count(),
305 1,
306 "the full 2s budget still fits 1.8s of real frames: the refusal left no ghost airtime",
307 );
308 }
309
310 #[test]
311 fn a_candidate_that_would_cross_the_limit_is_held() {
312 let mut gate: DutyGate<FixedDutyQueue<2, TEST_MTU>> = DutyGate::new();
313 assert_eq!(
314 gate.offer(
315 &[1, 2, 3],
316 300_000,
317 AirtimeUtilization {
318 short_per_mille: 101,
319 long_per_mille: 0,
320 },
321 &DUTY,
322 ),
323 DutyVerdict::Held
324 );
325 assert!(!gate.is_empty());
326 }
327
328 #[cfg(feature = "alloc")]
329 #[test]
330 fn the_heap_queue_transmits_under_limit_holds_over_it() {
331 transmits_under_limit_holds_over_it::<HeapDutyQueue>();
332 }
333
334 #[cfg(feature = "alloc")]
335 #[test]
336 fn the_heap_queue_budgets_in_airtime() {
337 budgets_the_queue_in_airtime::<HeapDutyQueue>();
338 }
339}