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
use std::collections::VecDeque;
use bytes::Bytes;
use thiserror::Error;
use tracing::{debug, trace};
use super::Connection;
use crate::{
frame::{Datagram, FrameStruct},
TransportError,
};
/// API to control datagram traffic
pub struct Datagrams<'a> {
pub(super) conn: &'a mut Connection,
}
impl<'a> Datagrams<'a> {
/// 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 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.
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.path.current_mtu() as usize
- self.conn.predict_1rtt_overhead(None)
- 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
///
/// 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) {
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
);
}
result
});
}
/// 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(&mut self, buf: &mut Vec<u8>, max_size: usize) -> bool {
let datagram = match self.outgoing.pop_front() {
Some(x) => x,
None => return false,
};
if buf.len() + datagram.size(true) > max_size {
// 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;
}
trace!(len = datagram.data.len(), "DATAGRAM");
self.outgoing_total -= datagram.data.len();
datagram.encode(true, buf);
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),
}