diff --git a/src/connection/connection.rs b/src/connection/connection.rs index 1b4190bc..cf62ae7c 100644 --- a/src/connection/connection.rs +++ b/src/connection/connection.rs @@ -87,6 +87,12 @@ pub struct Connection { /// Whether this is a server connection. is_server: bool, + /// RFC 9221 DATAGRAM frames received, awaiting the application (FIFO). + dgram_recv: VecDeque>, + /// DATAGRAM frames queued by the application to send (FIFO). Unreliable: + /// dropped, never retransmitted, if lost or if the queue overflows. + dgram_send: VecDeque>, + /// Connection Identifiers. cids: cid::ConnectionIdMgr, @@ -251,6 +257,8 @@ impl Connection { let mut conn = Connection { version: crate::QUIC_VERSION_V1, is_server, + dgram_recv: VecDeque::new(), + dgram_send: VecDeque::new(), cids, spaces: space::PacketNumSpaceMap::new(), paths, @@ -719,6 +727,17 @@ impl Connection { Frame::Ping { .. } => (), // just ignore + Frame::Datagram { data, .. } => { + // RFC 9221: deliver to the application recv queue. Bounded so + // a flood can't grow memory unboundedly (unreliable: drop + // oldest, mirroring a full NIC rx ring). + const MAX_DGRAM_RECV: usize = 1024; + if self.dgram_recv.len() >= MAX_DGRAM_RECV { + self.dgram_recv.pop_front(); + } + self.dgram_recv.push_back(data.to_vec()); + } + Frame::Ack { ack_delay, ack_ranges, @@ -2009,6 +2028,10 @@ impl Connection { // Write STREAM frames self.try_write_stream_frames(out, st, pkt_type, path_id)?; + // Write DATAGRAM frames (RFC 9221) — after streams so bulk stream + // data doesn't starve, but they still ride each packet's spare room. + self.try_write_datagram_frames(out, st, pkt_type)?; + // Write a NEW_TOKEN frame self.try_write_new_token_frame(out, st, pkt_type, path_id)?; @@ -2068,6 +2091,54 @@ impl Connection { Ok(()) } + /// Write DATAGRAM frames (RFC 9221) from the send queue into the packet. + /// Unreliable: each is emitted at most once (no retransmit on loss), and + /// one too large to fit any packet is dropped rather than blocking the + /// queue. Only in 1-RTT packets (application data). + fn try_write_datagram_frames( + &mut self, + out: &mut [u8], + st: &mut FrameWriteStatus, + pkt_type: PacketType, + ) -> Result<()> { + if pkt_type != PacketType::OneRTT { + return Ok(()); + } + // Only emit if the peer advertised DATAGRAM support (RFC 9221); + // otherwise sending one is a protocol violation. + if self.peer_transport_params.max_datagram_frame_size == 0 { + self.dgram_send.clear(); + return Ok(()); + } + while let Some(front) = self.dgram_send.front() { + // length-prefixed form (0x31): 1-byte type + varint len + data, + // so other frames may coexist in the same packet. + let wire = 1 + codec::encode_varint_len(front.len() as u64) + front.len(); + let remaining = out.len().saturating_sub(st.written); + if wire > remaining { + // "Too big for any packet" is judged against the negotiated + // max, not this packet's leftover room — otherwise a datagram + // that doesn't fit the current (ACK-filled) packet but WOULD + // fit a fresh one gets stuck in the queue forever. + let peer_max = self.peer_transport_params.max_datagram_frame_size as usize; + if wire > peer_max { + self.dgram_send.pop_front(); // exceeds peer's max — drop + continue; + } + break; // try the next (emptier) packet + } + let data = self.dgram_send.pop_front().unwrap(); + let frame = Frame::Datagram { + length: Some(data.len()), + data: Bytes::from(data), + }; + Connection::write_frame_to_packet(frame, out, st)?; + st.ack_eliciting = true; + st.in_flight = true; + } + Ok(()) + } + /// Write PATH_RESPONSE/PATH_CHALLENGE frames if needed. fn try_write_path_validation_frames( &mut self, @@ -3111,6 +3182,7 @@ impl Connection { || path.need_send_ping || self.cids.need_send_cid_control_frames() || self.streams.need_send_stream_frames() + || !self.dgram_send.is_empty() || self.spaces.need_send_buffered_frames()) { if !self.is_server && self.tls_session.is_in_early_data() { @@ -3129,6 +3201,10 @@ impl Connection { || self.local_error.as_ref().is_some_and(|e| e.is_app) || self.cids.need_send_cid_control_frames() || self.streams.need_send_stream_frames() + // DATAGRAM frames may be sent on any path; without this a + // datagram-only workload never consults the multipath scheduler + // and everything defaults to the primary path. + || !self.dgram_send.is_empty() } /// Find space id for the specified packet type and path id. @@ -3396,6 +3472,48 @@ impl Connection { self.flags.contains(HandshakeCompleted) } + /// Queue an unreliable DATAGRAM (RFC 9221) to send. It is emitted in the + /// next outgoing packet(s) and never retransmitted if lost. Bounded send + /// queue: returns Error::Done when full (caller drops, like a NIC tx ring). + pub fn datagram_send(&mut self, data: &[u8]) -> Result<()> { + const MAX_DGRAM_SEND: usize = 1024; + if self.dgram_send.len() >= MAX_DGRAM_SEND { + return Err(Error::Done); + } + self.dgram_send.push_back(data.to_vec()); + // Put the connection on the endpoint's send/tick queues so the queued + // datagram is flushed on the next poll (mirrors stream_write). + self.mark_tickable(true); + self.mark_sendable(true); + Ok(()) + } + + /// Read the next received DATAGRAM into `out`. Returns the byte length, or + /// Error::Done if none are queued. + pub fn datagram_recv(&mut self, out: &mut [u8]) -> Result { + let d = self.dgram_recv.pop_front().ok_or(Error::Done)?; + let n = d.len().min(out.len()); + out[..n].copy_from_slice(&d[..n]); + Ok(n) + } + + /// Number of DATAGRAMs waiting to be read. + pub fn datagram_recv_queue_len(&self) -> usize { + self.dgram_recv.len() + } + + /// Largest DATAGRAM payload the peer will accept, or None if the peer did + /// not advertise DATAGRAM support. The caller sizes its frames below this. + pub fn datagram_max_send_size(&self) -> Option { + let m = self.peer_transport_params.max_datagram_frame_size; + if m == 0 { + return None; + } + // m bounds the whole DATAGRAM frame; subtract worst-case header + // (1-byte type + up to 8-byte length varint). + Some((m as usize).saturating_sub(9)) + } + /// Check whether the connection handshake is confirmed. pub fn is_confirmed(&self) -> bool { self.flags.contains(HandshakeConfirmed) @@ -4963,6 +5081,54 @@ pub(crate) mod tests { } } + #[test] + fn datagram_end_to_end() -> Result<()> { + // Enable RFC 9221 DATAGRAM support on both ends. + let mut client_config = TestPair::new_test_config(false)?; + client_config.set_max_datagram_frame_size(1500); + let mut server_config = TestPair::new_test_config(true)?; + server_config.set_max_datagram_frame_size(1500); + let mut pair = TestPair::new(&mut client_config, &mut server_config)?; + pair.handshake()?; + + // Peer support must have been negotiated via transport params. + assert!(pair.client.datagram_max_send_size().is_some()); + assert!(pair.server.datagram_max_send_size().is_some()); + + // client -> server unreliable datagram (the tunnel's uplink) + let up = b"tunneled IP packet over a TQUIC DATAGRAM"; + pair.client.datagram_send(up)?; + pair.move_forward()?; + let mut out = [0u8; 2048]; + let n = pair.server.datagram_recv(&mut out)?; + assert_eq!(&out[..n], up); + assert_eq!(pair.server.datagram_recv(&mut out), Err(Error::Done)); // no more + + // server -> client (the downlink) + let down = b"reply datagram"; + pair.server.datagram_send(down)?; + pair.move_forward()?; + let n = pair.client.datagram_recv(&mut out)?; + assert_eq!(&out[..n], down); + Ok(()) + } + + #[test] + fn datagram_peer_no_support_is_dropped() -> Result<()> { + // Client enables datagrams; server does NOT advertise support. + let mut client_config = TestPair::new_test_config(false)?; + client_config.set_max_datagram_frame_size(1500); + let mut server_config = TestPair::new_test_config(true)?; + let mut pair = TestPair::new(&mut client_config, &mut server_config)?; + pair.handshake()?; + // Server didn't advertise → client must not attempt to send datagrams. + assert!(pair.client.datagram_max_send_size().is_none()); + pair.client.datagram_send(b"should be dropped, not a protocol error")?; + pair.move_forward()?; // must not error out the connection + assert!(!pair.client.is_closed() && !pair.server.is_closed()); + Ok(()) + } + #[test] fn version_negotiation_with_unknown_version() -> Result<()> { let mut test_pair = TestPair::new_with_test_config()?; diff --git a/src/frame.rs b/src/frame.rs index a5ec9657..54c7900e 100644 --- a/src/frame.rs +++ b/src/frame.rs @@ -178,6 +178,15 @@ pub enum Frame { seq_num: u64, status: u64, }, + + /// DATAGRAM frame (type=0x30..0x31) carries application data with no + /// reliability or ordering guarantees (RFC 9221). `length` is Some when + /// the LEN bit (0x01) is set (an explicit length prefix, so more frames + /// may follow); None means the data runs to the end of the packet. + Datagram { + length: Option, + data: Bytes, + }, } impl Frame { @@ -359,6 +368,24 @@ impl Frame { status: b.read_varint()?, }, + // DATAGRAM frame (RFC 9221). 0x30 = no length (data to end of + // packet); 0x31 = explicit varint length prefix. + 0x30 | 0x31 => { + let length = if frame_type & 0x01 != 0 { + Some(b.read_varint()? as usize) + } else { + None + }; + let len = length.unwrap_or_else(|| b.len()); + if len > b.len() { + return Err(Error::BufferTooShort); + } + let start = buf.len() - b.len(); + let data = buf.slice(start..(start + len)); + b.skip(len)?; + Frame::Datagram { length, data } + } + _ => return Err(Error::FrameEncodingError), }; @@ -609,6 +636,19 @@ impl Frame { b.write_varint(*seq_num)?; b.write_varint(*status)?; } + + Frame::Datagram { length, data } => { + match length { + Some(len) => { + b.write_varint(0x31)?; + b.write_varint(*len as u64)?; + } + None => { + b.write_varint(0x30)?; + } + } + b.write(data.as_ref())?; + } } Ok(len - b.len()) @@ -767,6 +807,14 @@ impl Frame { + codec::encode_varint_len(*seq_num) + codec::encode_varint_len(*status) } + + Frame::Datagram { length, data } => { + // 1-byte frame type + optional length varint + data + 1 + match length { + Some(len) => codec::encode_varint_len(*len as u64), + None => 0, + } + data.len() + } } } @@ -933,6 +981,12 @@ impl Frame { frame_type_value: None, raw: None, }, + + Frame::Datagram { .. } => QuicFrame::Unknown { + raw_frame_type: 0x30, + frame_type_value: None, + raw: None, + }, } } @@ -1112,6 +1166,10 @@ impl std::fmt::Debug for Frame { "PATH_STATUS dcid_seq_num={dcid_seq_num:x} seq_num={seq_num:x} status={status:x}", )?; } + + Frame::Datagram { length, data } => { + write!(f, "DATAGRAM length={length:?} len={}", data.len())?; + } } Ok(()) @@ -1845,6 +1903,45 @@ mod tests { Ok(()) } + #[test] + fn datagram() -> Result<()> { + // with explicit length (type 0x31) — round-trips and is ack-eliciting + let payload = Bytes::from_static(b"hello bonded world"); + let frame = Frame::Datagram { + length: Some(payload.len()), + data: payload.clone(), + }; + assert!(frame.ack_eliciting()); + assert!(!frame.probing()); + let mut buf = [0; 128]; + let len = frame.to_bytes(&mut buf[..])?; + assert_eq!(len, frame.wire_len()); + let mut b = Bytes::copy_from_slice(&buf[..len]); + assert_eq!((frame, len), Frame::from_bytes(&mut b, PacketType::OneRTT)?); + + // without length (type 0x30) — data runs to end of buffer + let frame2 = Frame::Datagram { + length: None, + data: payload.clone(), + }; + let len2 = frame2.to_bytes(&mut buf[..])?; + let mut b2 = Bytes::copy_from_slice(&buf[..len2]); + let (decoded, n) = Frame::from_bytes(&mut b2, PacketType::OneRTT)?; + assert_eq!(n, len2); + match decoded { + Frame::Datagram { length, data } => { + assert_eq!(length, None); + assert_eq!(data, payload); + } + _ => panic!("expected Datagram"), + } + + // DATAGRAM is forbidden in Initial/Handshake packets + let mut b3 = Bytes::copy_from_slice(&buf[..len]); + assert!(Frame::from_bytes(&mut b3, PacketType::Initial).is_err()); + Ok(()) + } + #[test] fn special_frames() -> Result<()> { assert_eq!( diff --git a/src/lib.rs b/src/lib.rs index d1d259dc..cff5029c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -449,6 +449,13 @@ impl Config { self.local_transport_params.initial_max_data = cmp::min(v, self.max_connection_window); } + /// Set `max_datagram_frame_size` (RFC 9221). Non-zero enables receiving + /// unreliable DATAGRAM frames and advertises support to the peer; the peer + /// may then send us datagrams up to this size. 0 (default) disables it. + pub fn set_max_datagram_frame_size(&mut self, v: u64) { + self.local_transport_params.max_datagram_frame_size = cmp::min(v, VINT_MAX); + } + /// Set the `initial_max_stream_data_bidi_local` transport parameter. /// The value is capped by the setting `max_stream_window`. /// The default value is `5242880`. diff --git a/src/trans_param.rs b/src/trans_param.rs index b2db8262..8a473055 100644 --- a/src/trans_param.rs +++ b/src/trans_param.rs @@ -121,6 +121,11 @@ pub struct TransportParams { /// completely trust the path between themselves. /// See draft-banks-quic-disable-encryption-00. pub disable_encryption: bool, + + /// max_datagram_frame_size (RFC 9221, id 0x20): the largest DATAGRAM frame + /// this endpoint will RECEIVE. 0 (absent) means DATAGRAM frames are not + /// supported by this endpoint. + pub max_datagram_frame_size: u64, } impl TransportParams { @@ -174,6 +179,11 @@ impl TransportParams { } } + // max_datagram_frame_size (RFC 9221) + 0x0020 => { + tp.max_datagram_frame_size = val.read_varint()?; + } + 0x0004 => { tp.initial_max_data = val.read_varint()?; } @@ -310,6 +320,14 @@ impl TransportParams { buf.write_varint(tp.max_udp_payload_size)?; } + // max_datagram_frame_size (RFC 9221, id 0x20) — advertise only when + // this endpoint accepts DATAGRAM frames. + if tp.max_datagram_frame_size != 0 { + buf.write_varint(0x0020)?; + buf.write_varint(codec::encode_varint_len(tp.max_datagram_frame_size) as u64)?; + buf.write_varint(tp.max_datagram_frame_size)?; + } + if tp.initial_max_data != 0 { buf.write_varint(0x0004)?; buf.write_varint(codec::encode_varint_len(tp.initial_max_data) as u64)?; @@ -483,6 +501,7 @@ impl Default for TransportParams { enable_multipath: false, disable_encryption: false, + max_datagram_frame_size: 0, } } } @@ -583,6 +602,7 @@ mod tests { retry_source_connection_id: None, enable_multipath: true, disable_encryption: false, + max_datagram_frame_size: 65535, }; // encode on the client side @@ -627,6 +647,7 @@ mod tests { retry_source_connection_id: Some(ConnectionId::random()), enable_multipath: false, disable_encryption: true, + max_datagram_frame_size: 1200, }; // encode on the server side