diff --git a/src/result.rs b/src/error.rs similarity index 100% rename from src/result.rs rename to src/error.rs diff --git a/src/lib.rs b/src/lib.rs index 9184a5d..a800b13 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -38,7 +38,8 @@ mod transport_layer; pub use transport_layer::*; -pub mod result; +/// Module which contains the various error types that can be returned by SEQEX. +pub mod error; /// This module contains the API for using SEQEX in a no-std environment. /// This API is low level and is the backbone of the `sync` and `tokio` implementations of SEQEX. diff --git a/src/no_std.rs b/src/no_std.rs index 941686a..f6e3e7a 100644 --- a/src/no_std.rs +++ b/src/no_std.rs @@ -1,7 +1,7 @@ pub use crate::single_thread::*; use crate::{ - result::{TryError, TryRecvError}, + error::{TryError, TryRecvError}, transport_layer::SeqNo, Packet, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; @@ -197,14 +197,14 @@ impl SeqEx { /// 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. /// - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. pub fn try_send_direct(&mut self, current_time: i64, seq_cst: bool, packet_data: SendData) -> Result, (TryError, SendData)> { let mut tmp = Some(packet_data); self.try_send_direct_with(current_time, seq_cst, |_| tmp.take().unwrap()) .map_err(|e| e.0) .map_err(|e| (e, tmp.unwrap())) } - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. pub fn try_send_direct_with SendData>( &mut self, current_time: i64, @@ -392,7 +392,7 @@ impl SeqEx { /// and since each fragment will be received in order it will be trivial for them to reconstruct /// the original file. /// - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. /// If `unlock` is true and the return value is `Some` pump may return new values. #[must_use] pub fn reply_raw_and_direct( @@ -444,7 +444,8 @@ impl SeqEx { } } - /// Can mutate `next_service_timestamp`. + /// Can increase `next_service_timestamp`. + #[inline] pub fn service_direct(&mut self, current_time: i64, iter: &mut Option) -> Option> { if self.next_service_timestamp <= current_time { let iter = iter.get_or_insert(ServiceIter { diff --git a/src/single_thread.rs b/src/single_thread.rs index d3f9bdc..28985d2 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,5 +1,5 @@ use crate::no_std::{RecvOkRaw, SeqEx}; -use crate::result::{RecvError, TryError, TryRecvError}; +use crate::error::{RecvError, TryError, TryRecvError}; use crate::{Packet, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { @@ -150,7 +150,7 @@ impl_recvok!(RecvOk, &'a mut SeqEx); pub(crate) use impl_recvok; impl SeqEx { - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. pub fn try_send(&mut self, mut tl: impl TransportLayer, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> { match self.try_send_direct(tl.time(), seq_cst, packet_data) { Ok(p) => { @@ -160,7 +160,7 @@ impl SeqEx { Err(e) => Err(e), } } - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. pub fn try_send_with SendData>( &mut self, mut tl: impl TransportLayer, @@ -193,7 +193,7 @@ impl SeqEx { Err(TryRecvError::WaitingForReply) => Err(RecvError::WaitingForReply), } } - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. /// If `unlock` is true and the return value is true pump may return new values. /// /// Only returns false if the reply number was incorrect or used twice. @@ -214,7 +214,7 @@ impl SeqEx { false } } - /// Can mutate `next_service_timestamp`. + /// Can increase `next_service_timestamp`. pub fn service(&mut self, mut tl: impl TransportLayer) -> i64 { let current_time = tl.time(); let mut iter = None; diff --git a/src/sync.rs b/src/sync.rs index 4b1cd75..157875e 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -8,7 +8,7 @@ use std::{ use crate::{ no_std::RecvOkRaw, - result::{RecvError, TryError}, + error::{RecvError, TryError}, Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; @@ -71,7 +71,7 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// Returns a mutable reference to the `TransportLayer` instance this guard was created with. /// /// Keep in mind that this cannot be used to change how replies and acks are resent, since - /// resends are handled with a separate instance of `TransportLayer`. + /// resends are handled with a separate instance of `TransportLayer` passed to `SeqEx::service`. pub fn get_tl_mut(&mut self) -> &mut TL { &mut self.tl } @@ -103,31 +103,42 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// This function receives as its first argument the packet sequence number, and as its second /// number the packet reply number. All calls to `TransportLayer::send` involving this packet /// will receive the exact same sequence and reply number. + /// + /// Returns the timestamp of when `service_ts` should be called next, only if it has decrease. + /// This can be safely ignored if `service_ts` is not being used. + /// /// # Panic /// This function will panic if `ack` has been called previously. - pub fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) -> Option { assert!(!self.has_replied, "Cannot reply after an ack has been sent"); self.has_replied = true; let mut inner = self.seq.inner.lock().unwrap(); + let pre_nst = inner.seq.next_service_timestamp; let seq_no = inner.seq.seq_no(); inner .seq .reply_raw(self.tl, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); + let nst = inner.seq.next_service_timestamp; self.seq.notify_reply(inner); core::mem::forget(self); + (pre_nst > nst).then_some(nst) } /// Consume this reply guard to add `packet_data` to the send window and immediately send it /// as a reply to the remote peer. Similar to `SeqEx::send`, except the remote peer will be /// explicitly informed that this packet is indeed a reply to a packet they sent. /// + /// Returns the timestamp of when `service_ts` should be called next, only if it has decrease. + /// This can be safely ignored if `service_ts` is not being used. + /// /// If you need to reply more than once, say to fragment a large file, then include in your /// first reply some identifier, and then `send` all fragments with the same included identifier. /// The identifier will tell the remote peer which packets contain fragments of the file, /// and since each fragment will be received in order it will be trivial for them to reconstruct /// the original file. + /// /// # Panic /// This function will panic if `ack` has been called previously. - pub fn reply(self, seq_cst: bool, packet_data: SendData) { + pub fn reply(self, seq_cst: bool, packet_data: SendData) -> Option { self.reply_with(seq_cst, |_, _| packet_data) } /// Break down a `ReplyGuard` into its primitive components, without causing it to send an ack @@ -185,22 +196,47 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std } } +/// When a packet received through `SeqEx` is ready to be processed, +/// it is returned as an instance of this enum. +/// +/// This enum specifies what kind of packet was received, and any additional data associated with the packet. pub enum RecvOk<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { + /// The received packet is a payload. Payload { + /// This type of packet can be optionally replied to, otherwise we need to send an ack. + /// This guard instance guarantees exactly one of those happens. + /// See the documentation of `ReplyGuard` for more information. reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, + /// The data associated with the packet we just received from the remote peer. + /// It was moved from the receive window to this enum. recv_data: RecvData, }, + /// The received packet is a reply to some packet we sent previously. Reply { + /// This type of packet can be optionally replied to, otherwise we need to send an ack. + /// This guard instance guarantees exactly one of those happens. + /// See the documentation of `ReplyGuard` for more information. reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, + /// The data associated with the packet we just received from the remote peer. + /// It was moved from the receive window to this enum. recv_data: RecvData, + /// The data associated with a packet we sent to the remote peer. + /// The received packet is a reply to that packet. + /// As a result, this data was moved from the `SeqEx` send window to this enum. send_data: SendData, }, + /// The received packet is an acknowledgment of some packet we sent previously. Ack { + /// The data associated with a packet we sent to the remote peer. + /// The received packet was an ack of this packet. + /// As a result, this data was moved from the `SeqEx` send window to this enum. send_data: SendData, }, } crate::no_std::impl_recvok!(RecvOk, &'a SeqEx); +/// An iterator which will automatically call either `SeqEx::pump` or `SeqEx::try_pump` to go +/// through all packets that can currently be processed. pub struct RecvIter<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { seq: Option<&'a SeqEx>, tl: TL, @@ -249,7 +285,14 @@ impl SeqEx { self.wait_on_recv.notify_one(); } } - + /// A SEQEX packet was just received and deserialized from the transport layer. + /// Passing it to this function officially writes it to the receive window so that it can be + /// eventually processed with the SEQEX transport guarantees. + /// + /// This function can only block to lock an internal mutex. + /// + /// If this function returns `Ok`, the contained boolean is true if `SeqEx::pump` should also + /// be called, because there are packets ready to be processed in the receive window. pub fn try_receive>( &self, tl: TL, @@ -265,7 +308,13 @@ impl SeqEx { } } /// A SEQEX packet was just received and deserialized from the transport layer. + /// Passing it to this function officially writes it to the receive window so that it can be + /// eventually processed with the SEQEX transport guarantees. /// + /// This function may block to preserve lossless or in-order transport. + /// + /// If this function returns `Some`, the contained boolean is true if `SeqEx::pump` should also + /// be called, because there are packets ready to be processed in the receive window. pub fn receive>(&self, tl: TL, packet: Packet) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { let result = self.try_receive(tl, packet); if let Err(RecvError::WaitingForReply) = result { @@ -274,6 +323,10 @@ impl SeqEx { result.ok() } } + /// Non-blocking variant of `SeqEx::pump`. + /// See the documentation for `SeqEx::pump` for more information. + /// + /// NOTE: This function can block for a short period of time to lock an internal mutex. pub fn try_pump>(&self, tl: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> { let mut inner = self.inner.lock().unwrap(); match inner.seq.try_pump_raw() { @@ -284,6 +337,16 @@ impl SeqEx { Err(e) => Err(e), } } + /// Check if there are any packets in the receive window that can be processed. + /// + /// This function may block to preserve lossless or in-order transport. + /// It will never block if the receive window is empty. + /// + /// If this function returns `Some`, the contained boolean is true if `SeqEx::pump` should be + /// called another time, because there are still packets ready to be processed in the receive window. + /// + /// It is recommended to assign the job of pumping `SeqEx` to a different thread so packets can + /// be processed in parallel. pub fn pump>(&self, tl: TL) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { let mut inner = self.inner.lock().unwrap(); // Enforce that only one thread may wait to pump at a time. @@ -304,7 +367,8 @@ impl SeqEx { } } } - + /// Similar to `SeqEx::receive`, except this function will return an iterator which + /// automatically calls `SeqEx::pump` as many times as needed to empty the receive window. pub fn receive_all>(&self, tl: TL, packet: Packet) -> RecvIter<'_, TL, SendData, RecvData, CAP> { let ret = self.receive(tl, packet); if let Some((first, do_pump)) = ret { @@ -318,6 +382,9 @@ impl SeqEx { RecvIter { seq: None, tl, first: None, blocking: true } } } + /// Similar to `SeqEx::try_receive`, except this function will return an iterator which + /// automatically calls `SeqEx::try_pump` as many times as it can before either the receive + /// window is empty, or one of the calls would block. pub fn try_receive_all>(&self, tl: TL, packet: Packet) -> RecvIter<'_, TL, SendData, RecvData, CAP> { let ret = self.try_receive(tl, packet); if let Ok((first, do_pump)) = ret { @@ -332,56 +399,145 @@ impl SeqEx { } } - pub fn try_send_with, F: FnOnce(SeqNo) -> SendData>( + /// Non-blocking variant of `SeqEx::send` that allows for a provided function to write data to + /// the send window. + /// See the documentation for `SeqEx::send` for more information. + /// + /// This function receives the packet sequence number the packet will be assigned. + /// All calls to `TransportLayer::send` involving this packet will receive this exact same + /// sequence number. + /// + /// NOTE: This function can block for a short period of time to lock an internal mutex. + /// + /// A return value of `Ok` contains the timestamp of when `service_ts` should be called next, + /// only if it has decrease. This can be safely ignored if `service_ts` is not being used. + pub fn try_send_with_ts, F: FnOnce(SeqNo) -> SendData>( &self, tl: TL, seq_cst: bool, packet_data: F, - ) -> Result<(), (TryError, F)> { + ) -> Result, (TryError, F)> { let mut inner = self.inner.lock().unwrap(); - inner.seq.try_send_with(tl, seq_cst, packet_data) + let pre_nst = inner.seq.next_service_timestamp; + inner.seq.try_send_with(tl, seq_cst, packet_data).map(|()| { + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) + }) } - pub fn try_send>(&self, tl: TL, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> { + /// Non-blocking variant of `SeqEx::send`. + /// See the documentation for `SeqEx::send` for more information. + /// + /// NOTE: This function can block for a short period of time to lock an internal mutex. + /// + /// A return value of `Ok` contains the timestamp of when `service_ts` should be called next, + /// only if it has decrease. This can be safely ignored if `service_ts` is not being used. + pub fn try_send_ts>(&self, tl: TL, seq_cst: bool, packet_data: SendData) -> Result, (TryError, SendData)> { let mut inner = self.inner.lock().unwrap(); - inner.seq.try_send(tl, seq_cst, packet_data) + let pre_nst = inner.seq.next_service_timestamp; + inner.seq.try_send(tl, seq_cst, packet_data).map(|()| { + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) + }) } - pub fn send_with>(&self, tl: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) { + /// Variant of `SeqEx::send` that allows for a provided function to write data to the send + /// window. + /// See the documentation for `SeqEx::send` for more information. + /// + /// This function receives the packet sequence number the packet will be assigned. + /// All calls to `TransportLayer::send` involving this packet will receive this exact same + /// sequence number. + pub fn send_with_ts>(&self, tl: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) -> Option { let mut inner = self.inner.lock().unwrap(); + let mut pre_nst = inner.seq.next_service_timestamp; while let Err((e, p)) = inner.seq.try_send_with(tl, seq_cst, packet_data) { packet_data = p; match e { TryError::WaitingForRecv => { inner.recv_waiters += 1; inner = self.wait_on_recv.wait(inner).unwrap(); + pre_nst = inner.seq.next_service_timestamp; } TryError::WaitingForReply => { inner.reply_sender_waiters = true; inner = self.wait_on_reply_sender.wait(inner).unwrap(); + pre_nst = inner.seq.next_service_timestamp; } } } + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) } - pub fn send>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) { + /// Add the given `packet_data` to the send window, to be sent immediately, and to be resent + /// if the payload is not successfully received by the remote peer. + /// + /// If the either send or receive window is full, this function will block until they are not. + /// + /// By default this function guarantees lossless transport. When `seq_cst` is set to true, this + /// function also guarantees in-order transport. + /// + /// Lossless transport guarantees that all payloads will be received by the remote peer exactly + /// once, but not necessarily in the same order they were sent. + /// + /// In-order transport guarantees that all SeqCst payloads are received in the same order that + /// they were sent. + /// + /// Returns the timestamp of when `service_ts` should be called next, only if it has decrease. + /// This can be safely ignored if `service_ts` is not being used. + pub fn send>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) -> Option { let mut inner = self.inner.lock().unwrap(); + let mut pre_nst = inner.seq.next_service_timestamp; while let Err((e, p)) = inner.seq.try_send(tl, seq_cst, packet_data) { packet_data = p; match e { TryError::WaitingForRecv => { inner.recv_waiters += 1; inner = self.wait_on_recv.wait(inner).unwrap(); + pre_nst = inner.seq.next_service_timestamp; } TryError::WaitingForReply => { inner.reply_sender_waiters = true; inner = self.wait_on_reply_sender.wait(inner).unwrap(); + pre_nst = inner.seq.next_service_timestamp; } } } + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) } - + /// Function which handles resending unacknowledged packets. + /// It returns the duration of time in milliseconds that should be waited + /// until calling this function again. + /// + /// This function should be called repeatedly in a loop, with the loop + /// sleeping the amount of time specified before calling again. pub fn service>(&self, tl: TL) -> i64 { self.inner.lock().unwrap().seq.service(tl) } + /// A variant of `SeqEx::service` for advanced users. + /// + /// Instead of returning a duration of time, this function returns the exact timestamp at which + /// this function should be called again, or i64::MAX if the send window is empty and nothing + /// currently needs to be resent. + /// + /// This can be used in combination with the return values of `SeqEx::send` and + /// `ReplyGuard::reply` to precisely schedule updates to `SeqEx`, allowing + /// updated to occur much less often. + /// + /// However this function can be very difficult to use correctly as it both requires a means + /// of dynamically scheduling calls, as well as discipline in correctly applying updates to + /// that schedule whenever `SeqEx::send`, `ReplyGuard::reply` or any of their variants are called. + /// + /// It is recommended to just use `SeqEx::service`. + pub fn service_ts(&mut self, mut tl: impl TransportLayer) -> i64 { + let mut inner = self.inner.lock().unwrap(); + let current_time = tl.time(); + let mut iter = None; + while let Some(p) = inner.seq.service_direct(current_time, &mut iter) { + tl.send(p) + } + inner.seq.next_service_timestamp + } } impl Default for SeqEx { fn default() -> Self { @@ -414,19 +570,28 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Ite } } +/// An implementation of `TransportLayer` using std::sync::mpsc channels. +/// If you prefer channels instead of callbacks then you can use this. +/// +/// If you choose to use this with `SeqEx`, one instance of `MpscTransport` should be created, +/// and only that instance and clones of that instance should be used with `SeqEx` #[derive(Clone, Debug)] pub struct MpscTransport { + /// The sender channel of this instance. pub channel: Sender>, + /// A std::time::Instant for providing answers to `time()` callbacks. pub time: Instant, } impl MpscTransport { + /// Create a new instance of `MpscTransport`. pub fn new() -> (Self, Receiver>) { let (send, recv) = channel(); (Self { channel: send, time: std::time::Instant::now() }, recv) } - pub fn from_sender(send: Sender>) -> Self { - Self { channel: send, time: std::time::Instant::now() } + /// Create a new instance of `MpscTransport` using the provided `sender`. + pub fn from_sender(sender: Sender>) -> Self { + Self { channel: sender, time: std::time::Instant::now() } } } impl TransportLayer for &MpscTransport {