diff --git a/src/lib.rs b/src/lib.rs index f3c6358..c434abc 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -35,10 +35,19 @@ //! #![no_std] #![forbid(unsafe_code)] -//#![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)] +//#![warn(missing_docs, rust_2018_idioms)] +/// A 32-bit sequence number. Packets transported with SEP are expected to contain at least one +/// sequence number, and sometimes two. +/// All packets will either have a seq_no, a reply_no, or both. pub type SeqNo = u32; +/// A trait for giving an instance of SeqEx access to the transport layer. +/// +/// The implementor is free to choose how to define the generic types based on how they want to +/// manage memory. +/// It is possible through these generics to implement SeqEx to be no-alloc and zero-copy, but otherwise +/// a lot of them are most easily implemented as tuples of custom enums and Vec. pub trait TransportLayer: Sized { type RecvData; type RecvDataRef<'a>; @@ -47,14 +56,23 @@ pub trait TransportLayer: Sized { type SendData; fn send(&self, data: &Self::SendData); - fn send_ack(&self, reply_num: SeqNo); - fn send_empty_reply(&self, reply_num: SeqNo); + fn send_ack(&self, reply_no: SeqNo); + fn send_empty_reply(&self, reply_no: SeqNo); fn deserialize<'a>(data: &'a Self::RecvData) -> Self::RecvDataRef<'a>; fn process(&self, reply_cx: ReplyGuard<'_, Self>, recv_packet: Self::RecvDataRef<'_>, send_data: Option) -> Self::RecvReturn; } - +/// A trait for abstracting the process of receiving a packet, it allows SeqEx to either immediately +/// process a reference to the packet, or take ownership of the packet so it can be processed later. +/// +/// SeqEx has to take ownership of packets when they are received out-of-order, they are held in a +/// buffer until the time comes that they can be processed in order. +/// +/// It is possible through this trait to avoid a copy, allocation or other expensive ownership +/// operation whenever a packet is received in order and can immediately be processed. pub trait IntoRecvData: Into { + /// Return some form of reference to the data that `process` expects to receive. + /// This function can do anything from complex deserialization to a basic dereference. fn as_ref(&self) -> TL::RecvDataRef<'_>; } impl IntoRecvData for TL::RecvData { @@ -62,31 +80,41 @@ impl IntoRecvData for TL::RecvData { TL::deserialize(self) } } - +/// a pub struct SeqEx { - pub retry_interval: i64, - next_send_seq_num: SeqNo, - pre_recv_seq_num: SeqNo, + /// The interval at which packets will be resent if they have not yet been acknowledged by the + /// remote peer. + /// It can be statically or dynamically set, it is up to the user to decide. + pub resend_interval: i64, + next_send_seq_no: SeqNo, + pre_recv_seq_no: SeqNo, send_window: [Option>; SLEN], recv_window: [Option>; RLEN], } struct RecvEntry { - seq_num: SeqNo, - reply_num: Option, + seq_no: SeqNo, + reply_no: Option, data: TL::RecvData, } struct SendEntry { - seq_num: SeqNo, - reply_num: Option, + seq_no: SeqNo, + reply_no: Option, next_resent_time: i64, data: TL::SendData, } +/// The error type for when a packet has been received, but for whatever reason could not be +/// immediately processed. +#[derive(Debug, Clone)] pub enum Error { + /// The packet is out-of-sequence. It was either received too soon or too late and so it would be + /// invalid to process it right now. No action needs to be taken by the caller. OutOfSequence, - QueueIsFull, + /// The Send Window is currently full. The received packet cannot be processed right now because + /// it could cause the send window to overflow. No action needs to be taken by the caller. + WindowIsFull, } /// Whenever a packet is received, it must be replied to. /// This Guard object guarantees that this is the case. @@ -94,53 +122,89 @@ pub enum Error { pub struct ReplyGuard<'a, TL: TransportLayer> { app: Option<&'a TL>, seq_queue: &'a mut SeqEx, - reply_num: SeqNo, + reply_no: SeqNo, } +/// An iterator over all packets in the send window. It will iterate over all packets currently +/// being sent to the remote peer. +/// These packets are awaiting a reply from the remote peer. pub struct Iter<'a, TL: TransportLayer>(core::slice::Iter<'a, Option>>); +/// A mutable iterator over all packets in the send window. +/// +/// The user is able to mutate the contents of the packet being sent to the remote peer, as well as +/// any local metadata associated with the packet. +/// +/// Take note that if the packet itself is modified, SeqEx provides no guarantees about which +/// version of the packet will have been received by the remote peer. The local peer cannot be sure +/// if the remote peer will see the modified packet. For this reason it is not recommended to modify +/// the packet. pub struct IterMut<'a, TL: TransportLayer>(core::slice::IterMut<'a, Option>>); impl SeqEx { - pub fn new(retry_interval: i64, initial_seq_num: SeqNo) -> Self { + /// Creates a new instance of `SeqEx` for a new remote peer. + /// An instance of `SeqEx` expects to communicate with only exactly one other remote instance + /// of `SeqEx`. + /// + /// `retry_interval` is the initial value of the `retry_interval` field of `SeqEx`. It defines + /// how long `SeqEx` will wait until resending unacknowledged packets. It can be changed later. + /// + /// `initial_seq_no` is the first sequence number that this instance of `SeqEx` will use. It must be + /// exactly the same as the `initial_seq_no` of the remote instance of `SeqEx`. It can just be 1. + pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self { Self { - retry_interval, - next_send_seq_num: initial_seq_num, - pre_recv_seq_num: initial_seq_num.wrapping_sub(1), + resend_interval: retry_interval, + next_send_seq_no: initial_seq_no, + pre_recv_seq_no: initial_seq_no.wrapping_sub(1), recv_window: core::array::from_fn(|_| None), send_window: core::array::from_fn(|_| None), } } + /// Returns whether or not the send window is full. + /// If the send window is full calls to `SeqEx::send` will always fail. pub fn is_full(&self) -> bool { // We claim that the window is full one entry before it is actually full for the sake of // making it always possible for both peers to process at least one reply at all times. - let next_i = self.next_send_seq_num as usize; + let next_i = self.next_send_seq_no as usize; self.send_window[next_i % self.send_window.len()].is_some() || self.send_window[(next_i + 1) % self.send_window.len()].is_some() } - - pub fn seq_num(&self) -> SeqNo { - self.next_send_seq_num + /// Returns the next sequence number to be attached to the next sent packet. + /// This should be called before `SeqEx::send`, and the return value should be + /// included in some way with the `packet_data` parameter passed to `SeqEx::send`. + /// + /// When `packet_data` is sent to the remote peer, the receiver should be able to quickly read + /// the sequence number off of it. + pub fn seq_no(&self) -> SeqNo { + self.next_send_seq_no } + /// Sends the given packet to the remote peer and adds it to the send window. + /// /// If the return value is `false` the queue is full and the packet will not be sent. /// The caller must either cancel sending, abort the connection, or wait until a call to /// `receive` or `receive_empty_reply` returns `Some` and try again. /// /// If true is returned then the packet was successfully sent. /// - /// `packet_data` must contain the latest sequence number returned by `seq_num()` - /// There should always be a call to `SeqQueue::seq_num()` preceding every call to `send_seq`. + /// `packet_data` should contain both the packet to be sent as well as any local metadata the + /// caller wants to store with the packet. This metadata allows the exchange to be stateful. + /// `packet_data` must contain the latest sequence number returned by `seq_no()` + /// There should always be a call to `SeqEx::seq_no` preceding every call to `send`. + /// + /// `current_time` should be a timestamp of the current time, using whatever units of time the + /// user would like. However this choice of units must be consistent with the units of the + /// `retry_interval`. `current_time` does not have to be monotonically increasing. #[must_use = "The queue might be full causing the packet to not be sent"] - pub fn send_seq(&mut self, app: TL, packet_data: TL::SendData, current_time: i64) -> bool { + pub fn send(&mut self, app: TL, packet_data: TL::SendData, current_time: i64) -> bool { if self.is_full() { return false; } - let seq_num = self.next_send_seq_num; - self.next_send_seq_num = self.next_send_seq_num.wrapping_add(1); + let seq_no = self.next_send_seq_no; + self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1); - let next_resent_time = current_time + self.retry_interval; - let entry = self.send_window[seq_num as usize % self.send_window.len()].insert(SendEntry { - seq_num, - reply_num: None, + let next_resent_time = current_time + self.resend_interval; + let entry = self.send_window[seq_no as usize % self.send_window.len()].insert(SendEntry { + seq_no, + reply_no: None, next_resent_time, data: packet_data, }); @@ -152,26 +216,26 @@ impl SeqEx { pub fn receive( &mut self, app: TL, - seq_num: SeqNo, - reply_num: Option, + seq_no: SeqNo, + reply_no: Option, packet: impl IntoRecvData, ) -> Result { - // We only want to accept packets with seq_nums in the range: - // `self.pre_recv_seq_num < seq_num <= self.pre_recv_seq_num + self.recv_window.len()`. - // To check that range we compute `seq_num - (self.pre_recv_seq_num + 1)` and check + // We only want to accept packets with seq_nos in the range: + // `self.pre_recv_seq_no < seq_no <= self.pre_recv_seq_no + self.recv_window.len()`. + // To check that range we compute `seq_no - (self.pre_recv_seq_no + 1)` and check // if the number wrapped below 0, or if it is above `self.recv_window.len()`. - let normalized_seq_num = seq_num.wrapping_sub(self.pre_recv_seq_num).wrapping_sub(1); - let is_below_range = normalized_seq_num > SeqNo::MAX / 2; - let is_above_range = !is_below_range && normalized_seq_num >= self.recv_window.len() as u32; - let is_next = normalized_seq_num == 0; + let normalized_seq_no = seq_no.wrapping_sub(self.pre_recv_seq_no).wrapping_sub(1); + let is_below_range = normalized_seq_no > SeqNo::MAX / 2; + let is_above_range = !is_below_range && normalized_seq_no >= self.recv_window.len() as u32; + let is_next = normalized_seq_no == 0; if is_below_range { // Check whether or not we are already replying to this packet. for entry in self.send_window.iter() { - if entry.as_ref().map_or(false, |e| e.reply_num == Some(seq_num)) { + if entry.as_ref().map_or(false, |e| e.reply_no == Some(seq_no)) { return Err(Error::OutOfSequence); } } - app.send_empty_reply(seq_num); + app.send_empty_reply(seq_no); return Err(Error::OutOfSequence); } else if is_above_range { return Err(Error::OutOfSequence); @@ -179,65 +243,83 @@ impl SeqEx { // If the send window is full we cannot safely process received packets, // because there would be no way to reply. // We can only process this packet if processing it would make space in the send window. - let next_i = self.next_send_seq_num as usize % self.send_window.len(); - let would_be_full = self.send_window[next_i].as_ref().map_or(false, |e| Some(e.seq_num) != reply_num); - let i = seq_num as usize % self.recv_window.len(); + let next_i = self.next_send_seq_no as usize % self.send_window.len(); + let would_be_full = self.send_window[next_i].as_ref().map_or(false, |e| Some(e.seq_no) != reply_no); + let i = seq_no as usize % self.recv_window.len(); if let Some(pre) = self.recv_window[i].as_mut() { - if seq_num == pre.seq_num { + if seq_no == pre.seq_no { if is_next && !would_be_full { self.recv_window[i] = None; } else { - app.send_ack(seq_num); + app.send_ack(seq_no); return if would_be_full { - Err(Error::QueueIsFull) + Err(Error::WindowIsFull) } else { Err(Error::OutOfSequence) }; } } else { + // NOTE: I believe this return is currently unreachable. return Err(Error::OutOfSequence); } } if is_next && !would_be_full { - self.pre_recv_seq_num = seq_num; - let data = reply_num.and_then(|r| self.take_send(r)); - Ok(app.process(ReplyGuard { app: Some(&app), seq_queue: self, reply_num: seq_num }, packet.as_ref(), data)) + self.pre_recv_seq_no = seq_no; + let data = reply_no.and_then(|r| self.take_send(r)); + Ok(app.process(ReplyGuard { app: Some(&app), seq_queue: self, reply_no: seq_no }, packet.as_ref(), data)) } else { - self.recv_window[i] = Some(RecvEntry { seq_num, reply_num, data: packet.into() }); - if let Some(reply_num) = reply_num { - self.receive_ack(reply_num); + self.recv_window[i] = Some(RecvEntry { seq_no, reply_no, data: packet.into() }); + if let Some(reply_no) = reply_no { + self.receive_ack(reply_no); } - app.send_ack(seq_num); + app.send_ack(seq_no); if would_be_full { - Err(Error::QueueIsFull) + Err(Error::WindowIsFull) } else { Err(Error::OutOfSequence) } } } + pub fn receive_ack(&mut self, reply_no: SeqNo) { + let i = reply_no as usize % self.send_window.len(); + if let Some(entry) = self.send_window[i].as_mut() { + if entry.seq_no == reply_no { + entry.next_resent_time = i64::MAX; + } + } + } + pub fn receive_empty_reply(&mut self, reply_no: SeqNo) -> Option { + let i = reply_no as usize % self.send_window.len(); + if self.send_window[i].as_ref().map_or(false, |e| e.seq_no == reply_no) { + let entry = self.send_window[i].take().unwrap(); + Some(entry.data) + } else { + None + } + } - fn take_send(&mut self, reply_num: SeqNo) -> Option { - let i = reply_num as usize % self.send_window.len(); - if self.send_window[i].as_ref().map_or(false, |e| e.seq_num == reply_num) { + fn take_send(&mut self, reply_no: SeqNo) -> Option { + let i = reply_no as usize % self.send_window.len(); + if self.send_window[i].as_ref().map_or(false, |e| e.seq_no == reply_no) { self.send_window[i].take().map(|e| e.data) } else { None } } pub fn pump(&mut self, app: TL) -> Result { - let next_seq_num = self.pre_recv_seq_num.wrapping_add(1); - let i = next_seq_num as usize % self.recv_window.len(); + let next_seq_no = self.pre_recv_seq_no.wrapping_add(1); + let i = next_seq_no as usize % self.recv_window.len(); - if self.recv_window[i].as_ref().map_or(false, |pre| pre.seq_num == next_seq_num) { - let next_i = self.next_send_seq_num as usize % self.send_window.len(); + if self.recv_window[i].as_ref().map_or(false, |pre| pre.seq_no == next_seq_no) { + let next_i = self.next_send_seq_no as usize % self.send_window.len(); if self.send_window[next_i].is_some() { - return Err(Error::QueueIsFull); + return Err(Error::WindowIsFull); } let entry = self.recv_window[i].take().unwrap(); - self.pre_recv_seq_num = next_seq_num; - let data = entry.reply_num.and_then(|r| self.take_send(r)); + self.pre_recv_seq_no = next_seq_no; + let data = entry.reply_no.and_then(|r| self.take_send(r)); Ok(app.process( - ReplyGuard { app: Some(&app), seq_queue: self, reply_num: entry.seq_num }, + ReplyGuard { app: Some(&app), seq_queue: self, reply_no: entry.seq_no }, TL::deserialize(&entry.data), data, )) @@ -246,26 +328,8 @@ impl SeqEx { } } - pub fn receive_ack(&mut self, reply_num: SeqNo) { - let i = reply_num as usize % self.send_window.len(); - if let Some(entry) = self.send_window[i].as_mut() { - if entry.seq_num == reply_num { - entry.next_resent_time = i64::MAX; - } - } - } - pub fn receive_empty_reply(&mut self, reply_num: SeqNo) -> Option { - let i = reply_num as usize % self.send_window.len(); - if self.send_window[i].as_ref().map_or(false, |e| e.seq_num == reply_num) { - let entry = self.send_window[i].take().unwrap(); - Some(entry.data) - } else { - None - } - } - pub fn service(&mut self, app: TL, current_time: i64) -> i64 { - let next_interval = current_time + self.retry_interval; + let next_interval = current_time + self.resend_interval; let mut next_activity = next_interval; for item in self.send_window.iter_mut() { if let Some(entry) = item { @@ -305,23 +369,23 @@ impl<'a, TL: TransportLayer> IntoIterator for &'a mut SeqEx { } impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { - pub fn seq_num(&self) -> SeqNo { - self.seq_queue.next_send_seq_num + pub fn seq_no(&self) -> SeqNo { + self.seq_queue.next_send_seq_no } - pub fn reply_num(&self) -> SeqNo { - self.reply_num + pub fn reply_no(&self) -> SeqNo { + self.reply_no } pub fn reply(mut self, packet_data: TL::SendData, current_time: i64) { if let Some(app) = self.app { let seq_queue = &mut self.seq_queue; - let seq_num = seq_queue.next_send_seq_num; - seq_queue.next_send_seq_num = seq_queue.next_send_seq_num.wrapping_add(1); + let seq_no = seq_queue.next_send_seq_no; + seq_queue.next_send_seq_no = seq_queue.next_send_seq_no.wrapping_add(1); - let i = seq_num as usize % seq_queue.send_window.len(); - let next_resent_time = current_time + seq_queue.retry_interval; + let i = seq_no as usize % seq_queue.send_window.len(); + let next_resent_time = current_time + seq_queue.resend_interval; let entry = seq_queue.send_window[i].insert(SendEntry { - seq_num, - reply_num: Some(self.reply_num), + seq_no, + reply_no: Some(self.reply_no), next_resent_time, data: packet_data, }); @@ -334,7 +398,7 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { fn drop(&mut self) { if let Some(app) = self.app { - app.send_empty_reply(self.reply_num); + app.send_empty_reply(self.reply_no); } } }