1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
use std::collections::VecDeque;
use bytes::Bytes;
use thiserror::Error;
use tracing::{debug, trace};
use super::Connection;
use crate::{
FrameStats, TransportError,
connection::PacketBuilder,
frame::{Datagram, FrameStruct},
};
/// API to control datagram traffic
pub struct Datagrams<'a> {
pub(super) conn: &'a mut Connection,
}
impl Datagrams<'_> {
/// Queue an unreliable, unordered datagram for immediate transmission
///
/// If `drop` is true, previously queued datagrams which are still unsent may be discarded to
/// make space for this datagram, in order of oldest to newest. If `drop` is false, and there
/// isn't enough space due to previously queued datagrams, this function will return
/// `SendDatagramError::Blocked`. `Event::DatagramsUnblocked` will be emitted once datagrams
/// have been sent.
///
/// Returns `Err` iff a `len`-byte datagram cannot currently be sent.
pub fn send(&mut self, data: Bytes, drop: bool) -> Result<(), SendDatagramError> {
if self.conn.config.datagram_receive_buffer_size.is_none() {
return Err(SendDatagramError::Disabled);
}
let max = self
.max_size()
.ok_or(SendDatagramError::UnsupportedByPeer)?;
if data.len() > max {
return Err(SendDatagramError::TooLarge);
}
if drop {
while self.conn.datagrams.outgoing_total > self.conn.config.datagram_send_buffer_size {
let prev = self
.conn
.datagrams
.outgoing
.pop_front()
.expect("datagrams.outgoing_total desynchronized");
trace!(len = prev.data.len(), "dropping outgoing datagram");
self.conn.datagrams.outgoing_total -= prev.data.len();
}
} else if self.conn.datagrams.outgoing_total + data.len()
> self.conn.config.datagram_send_buffer_size
{
self.conn.datagrams.send_blocked = true;
return Err(SendDatagramError::Blocked(data));
}
self.conn.datagrams.outgoing_total += data.len();
self.conn.datagrams.outgoing.push_back(Datagram { data });
Ok(())
}
/// Compute the maximum size of datagrams that may be passed to `send_datagram`
///
/// Returns `None` if datagrams are unsupported by the peer or disabled locally.
///
/// This may change over the lifetime of a connection according to variation in the path MTU
/// estimate. The peer can also enforce an arbitrarily small fixed limit, but if the peer's
/// limit is large this is guaranteed to be a little over a kilobyte at minimum.
///
/// Not necessarily the maximum size of received datagrams.
///
/// When multipath is enabled, this is calculated using the smallest MTU across all
/// available paths.
pub fn max_size(&self) -> Option<usize> {
// We use the conservative overhead bound for any packet number, reducing the budget by at
// most 3 bytes, so that PN size fluctuations don't cause users sending maximum-size
// datagrams to suffer avoidable packet loss.
let max_size = self.conn.current_mtu() as usize
- self.conn.predict_1rtt_overhead_no_pn()
- Datagram::SIZE_BOUND;
let limit = self
.conn
.peer_params
.max_datagram_frame_size?
.into_inner()
.saturating_sub(Datagram::SIZE_BOUND as u64);
Some(limit.min(max_size as u64) as usize)
}
/// Receive an unreliable, unordered datagram
pub fn recv(&mut self) -> Option<Bytes> {
self.conn.datagrams.recv()
}
/// Bytes available in the outgoing datagram buffer
///
/// When greater than zero, [`send`](Self::send)ing a datagram of at most this size is
/// guaranteed not to cause older datagrams to be dropped.
pub fn send_buffer_space(&self) -> usize {
self.conn
.config
.datagram_send_buffer_size
.saturating_sub(self.conn.datagrams.outgoing_total)
}
}
#[derive(Default)]
pub(super) struct DatagramState {
/// Number of bytes of datagrams that have been received by the local transport but not
/// delivered to the application
pub(super) recv_buffered: usize,
pub(super) incoming: VecDeque<Datagram>,
pub(super) outgoing: VecDeque<Datagram>,
pub(super) outgoing_total: usize,
pub(super) send_blocked: bool,
}
impl DatagramState {
pub(super) fn received(
&mut self,
datagram: Datagram,
window: &Option<usize>,
) -> Result<bool, TransportError> {
let window = match window {
None => {
return Err(TransportError::PROTOCOL_VIOLATION(
"unexpected DATAGRAM frame",
));
}
Some(x) => *x,
};
if datagram.data.len() > window {
return Err(TransportError::PROTOCOL_VIOLATION("oversized datagram"));
}
let was_empty = self.recv_buffered == 0;
while datagram.data.len() + self.recv_buffered > window {
debug!("dropping stale datagram");
self.recv();
}
self.recv_buffered += datagram.data.len();
self.incoming.push_back(datagram);
Ok(was_empty)
}
/// Discard outgoing datagrams with a payload larger than `max_payload` bytes
///
/// Returns whether any datagrams were dropped.
///
/// Used to ensure that reductions in MTU don't get us stuck in a state where we have a datagram
/// queued but can't send it.
pub(super) fn drop_oversized(&mut self, max_payload: usize) -> bool {
let mut dropped_any = false;
self.outgoing.retain(|datagram| {
let result = datagram.data.len() < max_payload;
if !result {
trace!(
"dropping {} byte datagram violating {} byte limit",
datagram.data.len(),
max_payload
);
self.outgoing_total -= datagram.data.len();
dropped_any = true;
}
result
});
dropped_any
}
/// Attempt to write a datagram frame into `buf`, consuming it from `self.outgoing`
///
/// Returns whether a frame was written. At most `max_size` bytes will be written, including
/// framing.
pub(super) fn write<'a, 'b>(
&mut self,
buf: &mut PacketBuilder<'a, 'b>,
stat: &mut FrameStats,
) -> bool {
let Some(datagram) = self.outgoing.pop_front() else {
return false;
};
if buf.frame_space_remaining() < datagram.size(true) {
// Future work: we could be more clever about cramming small datagrams into
// mostly-full packets when a larger one is queued first
self.outgoing.push_front(datagram);
return false;
}
self.outgoing_total -= datagram.data.len();
buf.write_frame(datagram, stat);
true
}
pub(super) fn recv(&mut self) -> Option<Bytes> {
let x = self.incoming.pop_front()?.data;
self.recv_buffered -= x.len();
Some(x)
}
}
/// Errors that can arise when sending a datagram
#[derive(Debug, Error, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub enum SendDatagramError {
/// The peer does not support receiving datagram frames
#[error("datagrams not supported by peer")]
UnsupportedByPeer,
/// Datagram support is disabled locally
#[error("datagram support disabled")]
Disabled,
/// The datagram is larger than the connection can currently accommodate
///
/// Indicates that the path MTU minus overhead or the limit advertised by the peer has been
/// exceeded.
#[error("datagram too large")]
TooLarge,
/// Send would block
#[error("datagram send blocked")]
Blocked(Bytes),
}