quinn_proto/connection/
datagrams.rs

1use std::collections::VecDeque;
2
3use bytes::Bytes;
4use thiserror::Error;
5use tracing::{debug, trace};
6
7use super::{Connection, Event};
8use crate::{
9    TransportError,
10    frame::{Datagram, FrameStruct},
11};
12
13/// API to control datagram traffic
14pub struct Datagrams<'a> {
15    pub(super) conn: &'a mut Connection,
16}
17
18impl Datagrams<'_> {
19    /// Queue an unreliable, unordered datagram for immediate transmission
20    ///
21    /// If `drop` is true, previously queued datagrams which are still unsent may be discarded to
22    /// make space for this datagram, in order of oldest to newest. If `drop` is false, and there
23    /// isn't enough space due to previously queued datagrams, this function will return
24    /// `SendDatagramError::Blocked`. `Event::DatagramsUnblocked` will be emitted once datagrams
25    /// have been sent.
26    ///
27    /// Returns `Err` iff a `len`-byte datagram cannot currently be sent.
28    pub fn send(&mut self, data: Bytes, drop: bool) -> Result<(), SendDatagramError> {
29        if self.conn.config.datagram_receive_buffer_size.is_none() {
30            return Err(SendDatagramError::Disabled);
31        }
32        let max = self
33            .max_size()
34            .ok_or(SendDatagramError::UnsupportedByPeer)?;
35        let send_buffer_size = self.conn.config.datagram_send_buffer_size;
36        let Some(max_buffer_payload) = send_buffer_size.checked_sub(size_of::<Datagram>()) else {
37            return Err(SendDatagramError::TooLarge);
38        };
39        if data.len() > Ord::min(max, max_buffer_payload) {
40            return Err(SendDatagramError::TooLarge);
41        }
42        if drop {
43            self.conn
44                .datagrams
45                .make_space_for(data.len(), send_buffer_size);
46        } else if !self
47            .conn
48            .datagrams
49            .has_send_buffer_space(data.len(), send_buffer_size)
50        {
51            self.conn.datagrams.send_blocked = true;
52            return Err(SendDatagramError::Blocked(data));
53        }
54        self.conn.datagrams.outgoing.push_back(Datagram { data });
55        Ok(())
56    }
57
58    /// Discard queued datagrams that no longer fit the active path, waking blocked senders.
59    pub(super) fn drop_oversized(&mut self) {
60        let Some(max_datagram_size) = self.max_size() else {
61            return;
62        };
63        if !self.conn.datagrams.drop_oversized(max_datagram_size)
64            || !self.conn.datagrams.send_blocked
65        {
66            return;
67        }
68
69        self.conn.datagrams.send_blocked = false;
70        self.conn.events.push_back(Event::DatagramsUnblocked);
71    }
72
73    /// Compute the maximum size of datagrams that may passed to `send_datagram`
74    ///
75    /// Returns `None` if datagrams are unsupported by the peer or disabled locally.
76    ///
77    /// This may change over the lifetime of a connection according to variation in the path MTU
78    /// estimate. The peer can also enforce an arbitrarily small fixed limit, but if the peer's
79    /// limit is large this is guaranteed to be a little over a kilobyte at minimum.
80    ///
81    /// Not necessarily the maximum size of received datagrams.
82    pub fn max_size(&self) -> Option<usize> {
83        // We use the conservative overhead bound for any packet number, reducing the budget by at
84        // most 3 bytes, so that PN size fluctuations don't cause users sending maximum-size
85        // datagrams to suffer avoidable packet loss.
86        let max_size = self.conn.path.current_mtu() as usize
87            - self.conn.predict_1rtt_overhead(None)
88            - Datagram::SIZE_BOUND;
89        let limit = self
90            .conn
91            .peer_params
92            .max_datagram_frame_size?
93            .into_inner()
94            .saturating_sub(Datagram::SIZE_BOUND as u64);
95        Some(limit.min(max_size as u64) as usize)
96    }
97
98    /// Receive an unreliable, unordered datagram
99    pub fn recv(&mut self) -> Option<Bytes> {
100        self.conn.datagrams.recv()
101    }
102
103    /// Bytes available in the outgoing datagram buffer
104    ///
105    /// When greater than zero, [`send`](Self::send)ing a datagram of at most this size is
106    /// guaranteed not to cause older datagrams to be dropped.
107    pub fn send_buffer_space(&self) -> usize {
108        self.conn
109            .config
110            .datagram_send_buffer_size
111            .saturating_sub(self.conn.datagrams.outgoing.memory_used())
112            .saturating_sub(size_of::<Datagram>())
113    }
114}
115
116#[derive(Default)]
117pub(super) struct DatagramState {
118    pub(super) incoming: DatagramBuffer,
119    pub(super) outgoing: DatagramBuffer,
120    pub(super) send_blocked: bool,
121}
122
123impl DatagramState {
124    pub(super) fn received(
125        &mut self,
126        datagram: Datagram,
127        window: &Option<usize>,
128    ) -> Result<bool, TransportError> {
129        let window = match window {
130            None => {
131                return Err(TransportError::PROTOCOL_VIOLATION(
132                    "unexpected DATAGRAM frame",
133                ));
134            }
135            Some(x) => *x,
136        };
137
138        let size_with_overhead = datagram.data.len() + size_of::<Datagram>();
139
140        if size_with_overhead > window {
141            return Err(TransportError::PROTOCOL_VIOLATION("oversized datagram"));
142        }
143
144        let was_empty = self.incoming.is_empty();
145        while self.incoming.memory_used() + size_with_overhead > window {
146            debug!("dropping stale datagram");
147            self.recv();
148        }
149
150        self.incoming.push_back(datagram);
151        Ok(was_empty)
152    }
153
154    fn make_space_for(&mut self, datagram_len: usize, send_buffer_size: usize) {
155        while !self.has_send_buffer_space(datagram_len, send_buffer_size) {
156            let Some(prev) = self.outgoing.pop_front() else {
157                break;
158            };
159            trace!(len = prev.data.len(), "dropping outgoing datagram");
160        }
161    }
162
163    fn has_send_buffer_space(&self, datagram_len: usize, send_buffer_size: usize) -> bool {
164        let Some(total) = self
165            .outgoing
166            .memory_used()
167            .checked_add(datagram_len)
168            .and_then(|total| total.checked_add(size_of::<Datagram>()))
169        else {
170            return false;
171        };
172
173        total <= send_buffer_size
174    }
175
176    /// Discard outgoing datagrams with a payload larger than `max_payload` bytes
177    ///
178    /// Returns whether any datagrams were dropped.
179    ///
180    /// Used to ensure that reductions in MTU don't get us stuck in a state where we have a datagram
181    /// queued but can't send it.
182    pub(super) fn drop_oversized(&mut self, max_payload: usize) -> bool {
183        let mut dropped_any = false;
184        self.outgoing.queue.retain(|datagram| {
185            let result = datagram.data.len() <= max_payload;
186            if !result {
187                trace!(
188                    "dropping {} byte datagram violating {} byte limit",
189                    datagram.data.len(),
190                    max_payload
191                );
192                self.outgoing.payload_bytes -= datagram.data.len();
193                dropped_any = true;
194            }
195            result
196        });
197        dropped_any
198    }
199
200    /// Attempt to write a datagram frame into `buf`, consuming it from `self.outgoing`
201    ///
202    /// Returns whether a frame was written. At most `max_size` bytes will be written, including
203    /// framing.
204    pub(super) fn write(&mut self, buf: &mut Vec<u8>, max_size: usize) -> bool {
205        let datagram = match self.outgoing.pop_front() {
206            Some(x) => x,
207            None => return false,
208        };
209
210        if buf.len() + datagram.size(true) > max_size {
211            // Future work: we could be more clever about cramming small datagrams into
212            // mostly-full packets when a larger one is queued first
213            self.outgoing.push_front(datagram);
214            return false;
215        }
216
217        trace!(len = datagram.data.len(), "DATAGRAM");
218        datagram.encode(true, buf);
219        true
220    }
221
222    pub(super) fn recv(&mut self) -> Option<Bytes> {
223        let x = self.incoming.pop_front()?.data;
224        Some(x)
225    }
226}
227
228#[derive(Default)]
229pub(super) struct DatagramBuffer {
230    queue: VecDeque<Datagram>,
231    payload_bytes: usize,
232}
233
234impl DatagramBuffer {
235    fn push_back(&mut self, datagram: Datagram) {
236        self.payload_bytes += datagram.data.len();
237        self.queue.push_back(datagram);
238    }
239
240    fn pop_front(&mut self) -> Option<Datagram> {
241        let datagram = self.queue.pop_front()?;
242        self.payload_bytes -= datagram.data.len();
243        Some(datagram)
244    }
245
246    fn push_front(&mut self, datagram: Datagram) {
247        self.payload_bytes += datagram.data.len();
248        self.queue.push_front(datagram);
249    }
250
251    fn memory_used(&self) -> usize {
252        self.payload_bytes
253            .saturating_add(self.queue.len() * size_of::<Datagram>())
254    }
255
256    pub(super) fn can_send_1rtt(&self, max_size: usize) -> bool {
257        self.queue.front().is_some_and(|x| x.size(true) <= max_size)
258    }
259
260    pub(super) fn is_empty(&self) -> bool {
261        self.queue.is_empty()
262    }
263}
264
265#[cfg(test)]
266mod tests {
267    use super::*;
268
269    #[test]
270    fn make_space_for_accounts_for_new_datagram() {
271        let mut state = DatagramState::default();
272        state.outgoing.push_back(Datagram {
273            data: Bytes::from_static(&[0; 7]),
274        });
275        state.outgoing.push_back(Datagram {
276            data: Bytes::from_static(&[0; 2]),
277        });
278        state.make_space_for(4, 10 + 2 * size_of::<Datagram>());
279
280        assert_eq!(state.outgoing.queue.len(), 1);
281        assert_eq!(state.outgoing.queue[0].data.len(), 2);
282        assert_eq!(state.outgoing.payload_bytes, 2);
283    }
284
285    #[test]
286    fn make_space_for_handles_overflowing_capacity_check() {
287        let mut state = DatagramState::default();
288        state.outgoing.queue.push_back(Datagram {
289            data: Bytes::from_static(&[0]),
290        });
291        state.outgoing.payload_bytes = usize::MAX - 1;
292
293        state.make_space_for(2, usize::MAX);
294
295        assert!(state.outgoing.is_empty());
296        assert_eq!(state.outgoing.payload_bytes, usize::MAX - 2);
297    }
298
299    #[test]
300    fn make_space_for_accounts_for_empty_datagram_metadata() {
301        let mut state = DatagramState::default();
302        for _ in 0..2 {
303            state.outgoing.push_back(Datagram { data: Bytes::new() });
304        }
305        let window = 2 * size_of::<Datagram>();
306
307        assert!(!state.has_send_buffer_space(0, window));
308        state.make_space_for(0, window);
309        assert_eq!(state.outgoing.queue.len(), 1);
310        assert!(state.has_send_buffer_space(0, window));
311        state.outgoing.push_back(Datagram { data: Bytes::new() });
312        assert_eq!(state.outgoing.memory_used(), window);
313    }
314
315    #[test]
316    fn send_buffer_space_handles_metadata_overflow() {
317        assert!(!DatagramState::default().has_send_buffer_space(usize::MAX, usize::MAX));
318    }
319
320    #[test]
321    fn drop_oversized_keeps_datagrams_at_limit() {
322        let mut state = DatagramState::default();
323        state.outgoing.push_back(Datagram {
324            data: Bytes::from_static(&[0; 10]),
325        });
326        state.outgoing.push_back(Datagram {
327            data: Bytes::from_static(&[0; 11]),
328        });
329
330        assert!(state.drop_oversized(10));
331
332        assert_eq!(state.outgoing.queue.len(), 1);
333        assert_eq!(state.outgoing.queue[0].data.len(), 10);
334        assert_eq!(state.outgoing.payload_bytes, 10);
335    }
336}
337
338/// Errors that can arise when sending a datagram
339#[derive(Debug, Error, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
340pub enum SendDatagramError {
341    /// The peer does not support receiving datagram frames
342    #[error("datagrams not supported by peer")]
343    UnsupportedByPeer,
344    /// Datagram support is disabled locally
345    #[error("datagram support disabled")]
346    Disabled,
347    /// The datagram is larger than the connection can currently accommodate
348    ///
349    /// Indicates that the path MTU minus overhead or the limit advertised by the peer has been
350    /// exceeded.
351    #[error("datagram too large")]
352    TooLarge,
353    /// Send would block
354    #[error("datagram send blocked")]
355    Blocked(Bytes),
356}
quinn_proto/connection/datagrams.rs

quinn_proto/connection/
datagrams.rs
