reactor_webrtc_dc_chunking/
send.rs

1//! The per-channel send queue.
2//!
3//! `push` accepts a whole message, however large, up to the configured
4//! limits; `next_frame` hands out its frames one at a time, but only while the
5//! native channel's buffer is below the high-water mark. The platform calls
6//! `next_frame` after every `push` and from the native low-threshold callback,
7//! so the native buffer never approaches libwebrtc's 16 MiB limit and the
8//! SCTP queue is refilled before it drains (an empty queue would collapse the
9//! flight and restart the ramp in the middle of a message).
10
11use std::collections::VecDeque;
12use std::fmt;
13
14use crate::frame::Header;
15
16/// How a queue splits and paces messages.
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub struct SendConfig {
19    /// Frame size on the wire, header included.
20    pub chunk_size: usize,
21    /// Hand out no frame while the native buffer holds this many bytes or more.
22    pub high_water: u64,
23    /// The native low-threshold the platform registers; the pump resumes there.
24    pub low_water: u64,
25    /// Bytes the queue may hold that the native channel has not taken yet.
26    pub queue_limit: u64,
27    /// Largest message `push` accepts: the smaller of the local limit and the
28    /// peer's advertised one (see [`crate::sdp::effective_max_message_size`]).
29    pub max_message_size: u64,
30}
31
32impl Default for SendConfig {
33    fn default() -> Self {
34        Self {
35            chunk_size: crate::DEFAULT_CHUNK_SIZE,
36            high_water: crate::DEFAULT_HIGH_WATER,
37            low_water: crate::DEFAULT_LOW_WATER,
38            queue_limit: crate::DEFAULT_SEND_BUFFER_LIMIT,
39            max_message_size: crate::DEFAULT_MAX_MESSAGE_SIZE,
40        }
41    }
42}
43
44impl SendConfig {
45    /// Check the values against each other and against libwebrtc's limits.
46    pub fn validate(&self) -> Result<(), ConfigError> {
47        // A frame must carry at least one payload byte and fit the 256 KiB a
48        // data-channel message may be on any peer.
49        if self.chunk_size < 2 || self.chunk_size as u64 > crate::LEGACY_MAX_MESSAGE_SIZE {
50            return Err(ConfigError::ChunkSize(self.chunk_size));
51        }
52        if self.low_water >= self.high_water {
53            return Err(ConfigError::Watermarks {
54                low: self.low_water,
55                high: self.high_water,
56            });
57        }
58        // high_water plus one frame must stay under the native limit. Written
59        // as a subtraction (chunk_size is at most 256 KiB here) so an extreme
60        // high_water cannot overflow.
61        if self.high_water > crate::NATIVE_SEND_BUFFER_LIMIT - self.chunk_size as u64 {
62            return Err(ConfigError::HighWater(self.high_water));
63        }
64        if self.queue_limit < self.max_message_size {
65            return Err(ConfigError::QueueBelowMessage {
66                queue: self.queue_limit,
67                message: self.max_message_size,
68            });
69        }
70        Ok(())
71    }
72}
73
74/// A [`SendConfig`] whose values cannot work together.
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub enum ConfigError {
77    ChunkSize(usize),
78    Watermarks { low: u64, high: u64 },
79    HighWater(u64),
80    QueueBelowMessage { queue: u64, message: u64 },
81}
82
83impl fmt::Display for ConfigError {
84    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
85        match self {
86            Self::ChunkSize(n) => write!(f, "chunk_size {n} must be between 2 bytes and 256 KiB"),
87            Self::Watermarks { low, high } => {
88                write!(f, "low_water {low} must be below high_water {high}")
89            }
90            Self::HighWater(n) => write!(
91                f,
92                "high_water {n} plus one frame must stay under libwebrtc's 16 MiB send buffer"
93            ),
94            Self::QueueBelowMessage { queue, message } => {
95                write!(
96                    f,
97                    "queue_limit {queue} must be at least max_message_size {message}"
98                )
99            }
100        }
101    }
102}
103
104impl std::error::Error for ConfigError {}
105
106/// Why `push` refused a message. Nothing of it was queued.
107#[derive(Debug, Clone, PartialEq, Eq)]
108pub enum SendError {
109    /// The message is larger than the effective max message size.
110    TooLarge { size: u64, max: u64 },
111    /// Queueing it would pass the queue limit.
112    QueueFull { queued: u64, size: u64, limit: u64 },
113}
114
115impl fmt::Display for SendError {
116    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
117        match self {
118            Self::TooLarge { size, max } => write!(
119                f,
120                "message of {size} bytes is larger than the {max}-byte limit"
121            ),
122            Self::QueueFull {
123                queued,
124                size,
125                limit,
126            } => {
127                write!(
128                    f,
129                    "queue holds {queued} bytes; adding {size} would pass its {limit}-byte limit"
130                )
131            }
132        }
133    }
134}
135
136impl std::error::Error for SendError {}
137
138struct Pending {
139    data: Vec<u8>,
140    text: bool,
141    /// Payload bytes already handed out as frames.
142    offset: usize,
143}
144
145/// Messages waiting to leave one channel, in send order.
146pub struct SendQueue {
147    config: SendConfig,
148    pending: VecDeque<Pending>,
149    queued: u64,
150}
151
152impl SendQueue {
153    /// A queue for one channel.
154    ///
155    /// # Panics
156    ///
157    /// When `config` fails [`SendConfig::validate`]: a frame size below 2
158    /// bytes, for one, would emit header-only frames forever.
159    pub fn new(config: SendConfig) -> Self {
160        if let Err(e) = config.validate() {
161            panic!("invalid SendConfig: {e}");
162        }
163        Self {
164            config,
165            pending: VecDeque::new(),
166            queued: 0,
167        }
168    }
169
170    pub fn config(&self) -> &SendConfig {
171        &self.config
172    }
173
174    /// Queue a whole message. `binary` is the type the receiver should see.
175    pub fn push(&mut self, data: Vec<u8>, binary: bool) -> Result<(), SendError> {
176        let size = data.len() as u64;
177        if size > self.config.max_message_size {
178            return Err(SendError::TooLarge {
179                size,
180                max: self.config.max_message_size,
181            });
182        }
183        if self.queued + size > self.config.queue_limit {
184            return Err(SendError::QueueFull {
185                queued: self.queued,
186                size,
187                limit: self.config.queue_limit,
188            });
189        }
190        self.queued += size;
191        self.pending.push_back(Pending {
192            data,
193            text: !binary,
194            offset: 0,
195        });
196        Ok(())
197    }
198
199    /// The next frame to hand to the native channel, or `None` when the queue
200    /// is empty or `native_buffered` is at the high-water mark. Frames always
201    /// go out as binary messages.
202    pub fn next_frame(&mut self, native_buffered: u64) -> Option<Vec<u8>> {
203        if native_buffered >= self.config.high_water {
204            return None;
205        }
206        let head = self.pending.front_mut()?;
207        let room = self.config.chunk_size - 1;
208        let take = room.min(head.data.len() - head.offset);
209        let more = head.offset + take < head.data.len();
210        let mut frame = Vec::with_capacity(take + 1);
211        frame.push(
212            Header {
213                more,
214                text: head.text,
215            }
216            .encode(),
217        );
218        frame.extend_from_slice(&head.data[head.offset..head.offset + take]);
219        head.offset += take;
220        self.queued -= take as u64;
221        if !more {
222            self.pending.pop_front();
223        }
224        Some(frame)
225    }
226
227    /// Payload bytes not yet handed to the native channel.
228    pub fn queued(&self) -> u64 {
229        self.queued
230    }
231
232    pub fn is_empty(&self) -> bool {
233        self.pending.is_empty()
234    }
235
236    /// Whether a message has started leaving but its last frame has not, so a
237    /// failure now would leave the peer with a partial message.
238    pub fn mid_message(&self) -> bool {
239        self.pending.front().is_some_and(|p| p.offset > 0)
240    }
241
242    /// Drop everything queued (the channel closed).
243    pub fn clear(&mut self) {
244        self.pending.clear();
245        self.queued = 0;
246    }
247}
248
249#[cfg(test)]
250mod tests {
251    use super::*;
252
253    fn small() -> SendConfig {
254        SendConfig {
255            chunk_size: 5,
256            high_water: 100,
257            low_water: 50,
258            queue_limit: 64,
259            max_message_size: 32,
260        }
261    }
262
263    fn drain(q: &mut SendQueue) -> Vec<Vec<u8>> {
264        std::iter::from_fn(|| q.next_frame(0)).collect()
265    }
266
267    #[test]
268    fn default_config_is_valid() {
269        assert_eq!(SendConfig::default().validate(), Ok(()));
270    }
271
272    #[test]
273    #[should_panic(expected = "invalid SendConfig: chunk_size 1")]
274    fn a_queue_refuses_a_config_that_cannot_drain() {
275        SendQueue::new(SendConfig {
276            chunk_size: 1,
277            ..SendConfig::default()
278        });
279    }
280
281    #[test]
282    fn an_extreme_high_water_is_rejected_without_overflow() {
283        let c = SendConfig {
284            high_water: u64::MAX,
285            ..SendConfig::default()
286        };
287        assert_eq!(c.validate(), Err(ConfigError::HighWater(u64::MAX)));
288    }
289
290    #[test]
291    fn rejects_inconsistent_configs() {
292        let ok = SendConfig::default();
293        assert!(SendConfig {
294            chunk_size: 1,
295            ..ok
296        }
297        .validate()
298        .is_err());
299        assert!(SendConfig {
300            chunk_size: 300 * 1024,
301            ..ok
302        }
303        .validate()
304        .is_err());
305        assert!(SendConfig {
306            low_water: ok.high_water,
307            ..ok
308        }
309        .validate()
310        .is_err());
311        assert!(SendConfig {
312            high_water: crate::NATIVE_SEND_BUFFER_LIMIT,
313            ..ok
314        }
315        .validate()
316        .is_err());
317        assert!(SendConfig {
318            queue_limit: ok.max_message_size - 1,
319            ..ok
320        }
321        .validate()
322        .is_err());
323    }
324
325    #[test]
326    fn splits_into_frames_of_chunk_size_with_header() {
327        let mut q = SendQueue::new(small());
328        q.push(b"abcdefghij".to_vec(), true).unwrap();
329        // chunk_size 5 = 1 header byte + 4 payload bytes.
330        assert_eq!(
331            drain(&mut q),
332            vec![
333                b"\x01abcd".to_vec(),
334                b"\x01efgh".to_vec(),
335                b"\x00ij".to_vec()
336            ]
337        );
338        assert!(q.is_empty());
339        assert_eq!(q.queued(), 0);
340    }
341
342    #[test]
343    fn exact_multiple_ends_with_a_full_last_frame() {
344        let mut q = SendQueue::new(small());
345        q.push(b"abcdefgh".to_vec(), true).unwrap();
346        assert_eq!(
347            drain(&mut q),
348            vec![b"\x01abcd".to_vec(), b"\x00efgh".to_vec()]
349        );
350    }
351
352    #[test]
353    fn empty_message_is_one_header_only_frame() {
354        let mut q = SendQueue::new(small());
355        q.push(Vec::new(), false).unwrap();
356        assert_eq!(drain(&mut q), vec![vec![0x02]]);
357    }
358
359    #[test]
360    fn text_flag_is_on_every_frame() {
361        let mut q = SendQueue::new(small());
362        q.push(b"hello".to_vec(), false).unwrap();
363        assert_eq!(drain(&mut q), vec![b"\x03hell".to_vec(), b"\x02o".to_vec()]);
364    }
365
366    #[test]
367    fn messages_leave_in_order_and_never_interleave() {
368        let mut q = SendQueue::new(small());
369        q.push(b"aaaaaa".to_vec(), true).unwrap();
370        q.push(b"bb".to_vec(), true).unwrap();
371        assert_eq!(
372            drain(&mut q),
373            vec![b"\x01aaaa".to_vec(), b"\x00aa".to_vec(), b"\x00bb".to_vec()]
374        );
375    }
376
377    #[test]
378    fn holds_frames_at_the_high_water_mark() {
379        let mut q = SendQueue::new(small());
380        q.push(b"abcdefgh".to_vec(), true).unwrap();
381        assert_eq!(q.next_frame(100), None);
382        assert_eq!(q.next_frame(99), Some(b"\x01abcd".to_vec()));
383        assert!(q.mid_message());
384        assert_eq!(q.queued(), 4);
385    }
386
387    #[test]
388    fn rejects_oversized_messages_and_a_full_queue() {
389        let mut q = SendQueue::new(small());
390        assert_eq!(
391            q.push(vec![0; 33], true),
392            Err(SendError::TooLarge { size: 33, max: 32 })
393        );
394        q.push(vec![0; 32], true).unwrap();
395        q.push(vec![0; 32], true).unwrap();
396        assert_eq!(
397            q.push(vec![0; 1], true),
398            Err(SendError::QueueFull {
399                queued: 64,
400                size: 1,
401                limit: 64
402            })
403        );
404        assert_eq!(q.queued(), 64);
405    }
406
407    #[test]
408    fn clear_drops_everything() {
409        let mut q = SendQueue::new(small());
410        q.push(vec![1; 10], true).unwrap();
411        q.next_frame(0);
412        q.clear();
413        assert!(q.is_empty());
414        assert!(!q.mid_message());
415        assert_eq!(q.queued(), 0);
416    }
417}
reactor_webrtc_dc_chunking/send.rs

reactor_webrtc_dc_chunking/
send.rs
