diff --git a/Cargo.lock b/Cargo.lock index 8788b3f..4623dca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -234,7 +234,7 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" [[package]] name = "seq-ex" -version = "1.0.0" +version = "1.0.1" dependencies = [ "rand_core", "serde", diff --git a/Cargo.toml b/Cargo.toml index f6f7f85..9d49e40 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "seq-ex" -version = "1.0.0" +version = "1.0.1" authors = ["Monica Moniot"] edition = "2021" diff --git a/examples/calculator.rs b/examples/calculator.rs index 4fd9a40..6ff9cf7 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -47,15 +47,15 @@ fn main() { let mut value = 0.0; let mut remote_value = value; - seq1.send(&transport1, true, Payload::Add(1.0)); + seq1.send(&transport1, true, Payload::Add(1.0)).unwrap(); value += 1.0; - seq1.send(&transport1, true, Payload::Sub(2.0)); + seq1.send(&transport1, true, Payload::Sub(2.0)).unwrap(); value -= 2.0; - seq1.send(&transport1, true, Payload::Mul(3.0)); + seq1.send(&transport1, true, Payload::Mul(3.0)).unwrap(); value *= 3.0; - seq1.send(&transport1, true, Payload::Div(4.0)); + seq1.send(&transport1, true, Payload::Div(4.0)).unwrap(); value /= 4.0; - seq1.send(&transport1, true, Payload::Mod(5.0)); + seq1.send(&transport1, true, Payload::Mod(5.0)).unwrap(); value %= 5.0; for _ in 0..16 { diff --git a/examples/file_download.rs b/examples/file_download.rs index 931570a..8ee90a2 100644 --- a/examples/file_download.rs +++ b/examples/file_download.rs @@ -70,7 +70,7 @@ fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload>) { let mut i = 0; while i < file.len() { let j = file.len().min(i + FILE_CHUNK_SIZE); - seqex.send( + let _ = seqex.send( &transport, true, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() }, @@ -142,9 +142,18 @@ fn main() { }; let tl = &peer1.transport; - peer1.seqex.send(tl, false, Payload::RequestFile { filename: "File1".to_string() }); - peer1.seqex.send(tl, false, Payload::RequestFile { filename: "File3".to_string() }); - peer1.seqex.send(tl, false, Payload::RequestFile { filename: "File2".to_string() }); + peer1 + .seqex + .send(tl, false, Payload::RequestFile { filename: "File1".to_string() }) + .unwrap(); + peer1 + .seqex + .send(tl, false, Payload::RequestFile { filename: "File3".to_string() }) + .unwrap(); + peer1 + .seqex + .send(tl, false, Payload::RequestFile { filename: "File2".to_string() }) + .unwrap(); for _ in 0..400 { receive(&peer1); diff --git a/examples/hello_world.rs b/examples/hello_world.rs index d084466..6106335 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -52,7 +52,7 @@ fn main() { let seq2 = SeqEx::default(); // We begin a "Hello World" exchange right here. - seq1.send(&transport1, false, Payload::Hello); + seq1.send(&transport1, false, Payload::Hello).unwrap(); receive(&recv2, &seq2, &transport2); receive(&recv1, &seq1, &transport1); diff --git a/src/error.rs b/src/error.rs index f7fa8fe..79b388a 100644 --- a/src/error.rs +++ b/src/error.rs @@ -4,7 +4,7 @@ use crate::SeqNo; /// /// Some of these errors specify that SEQEX is waiting on some event to occur before it can proceed. #[derive(Debug, Clone, PartialEq, Eq)] -pub enum TryRecvError { +pub enum TryRawRecvError { /// This packet had to be dropped because it arrived far enough out-of-order that it was outside /// the receive window. /// This packet will eventually be resent, so no data will be lost. @@ -36,7 +36,7 @@ pub enum TryRecvError { /// /// Some of these errors specify that SEQEX is waiting on some event to occur before it can proceed. #[derive(Debug, Clone, PartialEq, Eq)] -pub enum RecvError { +pub enum TryRecvError { /// This packet had to be dropped because it arrived far enough out-of-order that it was outside /// the receive window. /// This packet will eventually be resent, so no data will be lost. @@ -57,6 +57,37 @@ pub enum RecvError { WaitingForReply, } +/// These are the error types that can be returned by a blocking SEQEX receive function. +/// +/// Some of these errors specify that SEQEX is waiting on some event to occur before it can proceed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RecvError { + /// This packet had to be dropped because it arrived far enough out-of-order that it was outside + /// the receive window. + /// This packet will eventually be resent, so no data will be lost. + DroppedTooEarly, + /// This packet was a duplicate of a previously received packet. It must have been resent before + /// the remote peer received the ack for the packet. + /// Since this packet is a duplicate, no data is lost by dropping it. + DroppedDuplicate, + /// In order to preserve losslessness or in-order transport, the received packet cannot be + /// process until some other packet is received. The packet was saved to the receive window. + WaitingForRecv, + /// Either the receive window is full, or the received packet is SeqCst and cannot enter the + /// critical section where it is processed. In either case there currently exists some reply + /// guard that must be dropped or consumed before this packet can be processed. + /// + /// If the receive window was full, the packet was dropped. + /// Otherwise if the packet is SeqCst, then the packet was saved to the receive window. + WaitingForReply, + /// This instance of `SeqEx` has been explicitly closed. + /// It can no longer send or receive data. + /// + /// This error can only occur after `SeqEx::close` has been called. + /// An instance of `SeqEx` will never close by itself, only the caller can close it. + Closed, +} + /// A generic error that can be returned by a `try_send` or `try_pump` function. /// They specify what event must occur before a future call to `try_send` or `try_pump` can succeed. #[derive(Debug, Clone, PartialEq, Eq)] @@ -70,16 +101,56 @@ pub enum TryError { /// Until this occurs the packet cannot be sent or processed. WaitingForReply, } +/// A generic error that can be returned by a `try_send` or `try_pump` function. +/// They specify what event must occur before a future call to `try_send` or `try_pump` can succeed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TrySendError { + /// The packet could not be sent or processed at this time. + /// Some other packet must be received from the remote peer first. + WaitingForRecv, + /// Some currently issued reply number must be returned to SEQEX to send either an Ack or a + /// Reply. If using reply guards, then some currently existing reply guard must be dropped or + /// consumed. + /// Until this occurs the packet cannot be sent or processed. + WaitingForReply, + /// This instance of `SeqEx` has been explicitly closed. + /// It can no longer send or receive data. + /// + /// This error can only occur after `SeqEx::close` has been called. + /// An instance of `SeqEx` will never close by itself, only the caller can close it. + Closed, +} +/// This instance of `SeqEx` has been explicitly closed. +/// It can no longer send or receive data. +/// +/// This error can only occur after `SeqEx::close` has been called. +/// An instance of `SeqEx` will never close by itself, only the caller can close it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClosedError; + +#[cfg(feature = "std")] +impl std::fmt::Display for TryRawRecvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + TryRawRecvError::DroppedTooEarly => write!(f, "packet arrived too early"), + TryRawRecvError::DroppedDuplicate => write!(f, "packet was a duplicate"), + TryRawRecvError::DroppedDuplicateResendAck(_) => write!(f, "packet was a duplicate, resending ack"), + TryRawRecvError::WaitingForRecv => write!(f, "can't process until another packet is received"), + TryRawRecvError::WaitingForReply => write!(f, "can't process until a reply is finished"), + } + } +} +#[cfg(feature = "std")] +impl std::error::Error for TryRawRecvError {} #[cfg(feature = "std")] impl std::fmt::Display for TryRecvError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - TryRecvError::DroppedTooEarly => write!(f, "packet arrived too early"), - TryRecvError::DroppedDuplicate => write!(f, "packet was a duplicate"), - TryRecvError::DroppedDuplicateResendAck(_) => write!(f, "packet was a duplicate, resending ack"), - TryRecvError::WaitingForRecv => write!(f, "can't process until another packet is received"), - TryRecvError::WaitingForReply => write!(f, "can't process until a reply is finished"), + Self::DroppedTooEarly => write!(f, "packet arrived too early"), + Self::DroppedDuplicate => write!(f, "packet was a duplicate"), + Self::WaitingForRecv => write!(f, "can't process packet until another packet is received"), + Self::WaitingForReply => write!(f, "can't process packet until a reply is finished"), } } } @@ -90,10 +161,11 @@ impl std::error::Error for TryRecvError {} impl std::fmt::Display for RecvError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - RecvError::DroppedTooEarly => write!(f, "packet arrived too early"), - RecvError::DroppedDuplicate => write!(f, "packet was a duplicate"), - RecvError::WaitingForRecv => write!(f, "can't process until another packet is received"), - RecvError::WaitingForReply => write!(f, "can't process until a reply is finished"), + Self::DroppedTooEarly => write!(f, "packet arrived too early"), + Self::DroppedDuplicate => write!(f, "packet was a duplicate"), + Self::WaitingForRecv => write!(f, "can't process packet until another packet is received"), + Self::WaitingForReply => write!(f, "can't process packet until a reply is finished"), + Self::Closed => write!(f, "can't receive packet because the session was explicitly closed"), } } } @@ -111,3 +183,57 @@ impl std::fmt::Display for TryError { } #[cfg(feature = "std")] impl std::error::Error for TryError {} + +#[cfg(feature = "std")] +impl std::fmt::Display for TrySendError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + TrySendError::WaitingForRecv => write!(f, "can't send until another packet is received"), + TrySendError::WaitingForReply => write!(f, "can't send until a reply is finished"), + TrySendError::Closed => write!(f, "can't send because the session was explicitly closed"), + } + } +} +#[cfg(feature = "std")] +impl std::error::Error for TrySendError {} + +#[cfg(feature = "std")] +impl std::fmt::Display for ClosedError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "can't send because the session was explicitly closed") + } +} +#[cfg(feature = "std")] +impl std::error::Error for ClosedError {} + +impl From for RecvError { + fn from(value: TryRecvError) -> Self { + match value { + TryRecvError::DroppedTooEarly => RecvError::DroppedTooEarly, + TryRecvError::DroppedDuplicate => RecvError::DroppedDuplicate, + TryRecvError::WaitingForRecv => RecvError::WaitingForRecv, + TryRecvError::WaitingForReply => RecvError::WaitingForReply, + } + } +} + +impl From for TrySendError { + fn from(value: TryError) -> Self { + match value { + TryError::WaitingForRecv => TrySendError::WaitingForRecv, + TryError::WaitingForReply => TrySendError::WaitingForReply, + } + } +} + +impl From for TryRecvError { + fn from(value: TryRawRecvError) -> Self { + match value { + TryRawRecvError::DroppedTooEarly => TryRecvError::DroppedTooEarly, + TryRawRecvError::DroppedDuplicate => TryRecvError::DroppedDuplicate, + TryRawRecvError::DroppedDuplicateResendAck(_) => TryRecvError::DroppedDuplicate, + TryRawRecvError::WaitingForRecv => TryRecvError::WaitingForRecv, + TryRawRecvError::WaitingForReply => TryRecvError::WaitingForReply, + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 3fc97d2..f93c580 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -58,5 +58,9 @@ mod single_thread; pub mod sync; /// This module contains the API for using SEQEX safely with tokio for async-await style code. +/// +/// There is a known issue with this API that it currently exposes no direct way to close a session +/// and cause all paused awaits to return. The user can currently implement this with async timeouts +/// or an async "session closed" event. #[cfg(feature = "tokio")] pub mod tokio; diff --git a/src/no_std.rs b/src/no_std.rs index dfb9137..f947a4b 100644 --- a/src/no_std.rs +++ b/src/no_std.rs @@ -1,7 +1,7 @@ pub use crate::single_thread::*; use crate::{ - error::{TryError, TryRecvError}, + error::{TryError, TryRawRecvError}, transport_layer::SeqNo, Packet, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; @@ -211,10 +211,10 @@ impl SeqEx { &mut self, current_time: i64, seq_cst: bool, - packet_data: F, + f: F, ) -> Result, (TryError, F)> { if let Err(e) = self.is_full_inner(None) { - return Err((e, packet_data)); + return Err((e, f)); } let seq_no = self.next_send_seq_no; @@ -231,7 +231,7 @@ impl SeqEx { reply_no: None, seq_cst, next_resend_time, - data: packet_data(seq_no), + data: f(seq_no), }); Ok(entry.to_packet()) @@ -254,7 +254,7 @@ impl SeqEx { } /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn receive_raw_and_direct>(&mut self, packet: Packet

) -> Result<(RecvOkRaw, bool), TryRecvError> { + pub fn try_receive_raw_and_direct>(&mut self, packet: Packet

) -> Result<(RecvOkRaw, bool), TryRawRecvError> { let seq_cst = packet.is_seq_cst(); let (seq_no, reply_no, recv_data) = match packet { Payload(seq_no, recv_data) | SeqCstPayload(seq_no, recv_data) => (seq_no, None, recv_data), @@ -263,7 +263,7 @@ impl SeqEx { return self .take_send(reply_no) .map(|send_data| (RecvOkRaw::Ack { send_data }, false)) - .ok_or(TryRecvError::DroppedDuplicate) + .ok_or(TryRawRecvError::DroppedDuplicate) } }; // We only want to accept packets with sequence numbers in the range: @@ -286,17 +286,17 @@ impl SeqEx { // resending the packet. for entry in self.send_window.iter().flatten() { if entry.reply_no == Some(seq_no) { - return Err(TryRecvError::DroppedDuplicate); + return Err(TryRawRecvError::DroppedDuplicate); } } for i in 0..self.concurrent_replies_total { if self.concurrent_replies[i] == seq_no { - return Err(TryRecvError::DroppedDuplicate); + return Err(TryRawRecvError::DroppedDuplicate); } } - return Err(TryRecvError::DroppedDuplicateResendAck(seq_no)); + return Err(TryRawRecvError::DroppedDuplicateResendAck(seq_no)); } else if is_above_range { - return Err(TryRecvError::DroppedTooEarly); + return Err(TryRawRecvError::DroppedTooEarly); } // Check whether or not we've already received this packet @@ -322,9 +322,9 @@ impl SeqEx { self.recv_window[i] = RecvEntry::Occupied { seq_no, reply_no, seq_cst, data: recv_data.into() } } return if wait_recv { - Err(TryRecvError::WaitingForRecv) + Err(TryRawRecvError::WaitingForRecv) } else { - Err(TryRecvError::WaitingForReply) + Err(TryRawRecvError::WaitingForReply) }; } diff --git a/src/single_thread.rs b/src/single_thread.rs index 0ba16bc..ea9a752 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,4 +1,4 @@ -use crate::error::{RecvError, TryError, TryRecvError}; +use crate::error::{TryError, TryRawRecvError, TryRecvError}; use crate::no_std::{RecvOkRaw, SeqEx}; use crate::{Packet, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; @@ -43,10 +43,10 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep pub fn reply(self, seq_cst: bool, packet_data: SendData) { self.reply_with(seq_cst, |_, _| packet_data) } - pub fn reply_with(self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_with(self, seq_cst: bool, f: impl FnOnce(SeqNo, SeqNo) -> SendData) { let seq_no = self.seq.seq_no(); self.seq - .reply_raw(self.tl, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); + .reply_raw(self.tl, self.reply_no, self.is_holding_lock, seq_cst, f(seq_no, self.reply_no)); core::mem::forget(self); } @@ -76,14 +76,9 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep pub fn reply_stay_locked(self, seq_cst: bool, packet_data: SendData) -> Option> { self.reply_with_stay_locked(seq_cst, |_, _| packet_data) } - pub fn reply_with_stay_locked( - self, - seq_cst: bool, - packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData, - ) -> Option> { + pub fn reply_with_stay_locked(self, seq_cst: bool, f: impl FnOnce(SeqNo, SeqNo) -> SendData) -> Option> { let seq_no = self.seq.seq_no(); - self.seq - .reply_raw(self.tl, self.reply_no, false, seq_cst, packet_data(seq_no, self.reply_no)); + self.seq.reply_raw(self.tl, self.reply_no, false, seq_cst, f(seq_no, self.reply_no)); self.consume_lock() } @@ -231,9 +226,9 @@ impl SeqEx { &mut self, mut tl: impl TransportLayer, seq_cst: bool, - packet_data: F, + f: F, ) -> Result<(), (TryError, F)> { - match self.try_send_direct_with(tl.time(), seq_cst, packet_data) { + match self.try_send_direct_with(tl.time(), seq_cst, f) { Ok(p) => { tl.send(p); Ok(()) @@ -242,21 +237,18 @@ impl SeqEx { } } /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn receive_raw>( + pub fn try_receive_raw>( &mut self, mut tl: impl TransportLayer, packet: Packet

, - ) -> Result<(RecvOkRaw, bool), RecvError> { - match self.receive_raw_and_direct(packet) { + ) -> Result<(RecvOkRaw, bool), TryRecvError> { + match self.try_receive_raw_and_direct(packet) { Ok(a) => Ok(a), - Err(TryRecvError::DroppedDuplicateResendAck(reply_no)) => { + Err(TryRawRecvError::DroppedDuplicateResendAck(reply_no)) => { tl.send(Packet::Ack(reply_no)); - Err(RecvError::DroppedDuplicate) + Err(TryRecvError::DroppedDuplicate) } - Err(TryRecvError::DroppedTooEarly) => Err(RecvError::DroppedTooEarly), - Err(TryRecvError::DroppedDuplicate) => Err(RecvError::DroppedDuplicate), - Err(TryRecvError::WaitingForRecv) => Err(RecvError::WaitingForRecv), - Err(TryRecvError::WaitingForReply) => Err(RecvError::WaitingForReply), + Err(e) => Err(e.into()), } } /// Can decrease `next_service_timestamp`. @@ -293,8 +285,9 @@ impl SeqEx { &mut self, tl: TL, packet: Packet, - ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), RecvError> { - self.receive_raw(tl, packet).map(|(r, do_pump)| (RecvOk::from_raw(self, tl, r), do_pump)) + ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryRecvError> { + self.try_receive_raw(tl, packet) + .map(|(r, do_pump)| (RecvOk::from_raw(self, tl, r), do_pump)) } pub fn try_pump>(&mut self, tl: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> { self.try_pump_raw().map(|(r, do_pump)| (RecvOk::from_raw(self, tl, r), do_pump)) diff --git a/src/sync.rs b/src/sync.rs index 62ad3ed..d7fa42b 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -7,7 +7,7 @@ use std::{ }; use crate::{ - error::{RecvError, TryError}, + error::{ClosedError, RecvError, TryError, TrySendError}, no_std::RecvOkRaw, Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; @@ -32,14 +32,14 @@ use crate::{ pub struct SeqEx { inner: Mutex>, wait_on_recv: Condvar, - wait_on_reply_sender: Condvar, - wait_on_reply_receiver: Condvar, + wait_on_reply: Condvar, } struct SeqExInner { seq: crate::no_std::SeqEx, recv_waiters: usize, reply_sender_waiters: bool, reply_receiver_waiters: bool, + closed: bool, } /// A guard type that allows Rust's borrow-checker to guarantee that the invariants of the SEQEX /// protocol are maintained. @@ -97,13 +97,16 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// /// Returns the timestamp of when `service_scheduled` should be called next, only if it has decrease. /// This can be safely ignored if `service_scheduled` is not being used. - pub fn reply_with(self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) -> Option { + pub fn reply_with(self, seq_cst: bool, f: impl FnOnce(SeqNo, SeqNo) -> SendData) -> Option { let mut inner = self.seq.inner.lock().unwrap(); + if inner.closed { + return None; + } 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)); + .reply_raw(self.tl, self.reply_no, self.is_holding_lock, seq_cst, f(seq_no, self.reply_no)); let nst = inner.seq.next_service_timestamp; self.seq.notify_reply(inner); core::mem::forget(self); @@ -151,7 +154,9 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// Casual users are recommended to simply `drop` this reply guard instead of calling this function. pub fn ack(self) -> Option> { let mut inner = self.seq.inner.lock().unwrap(); - inner.seq.ack_raw(self.tl, self.reply_no, false); + if !inner.closed { + inner.seq.ack_raw(self.tl, self.reply_no, false); + } self.consume_lock(inner) } /// If this guard is holding the SeqCst lock, and preventing other SeqCst packets from @@ -188,14 +193,15 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep pub fn reply_with_stay_locked( self, seq_cst: bool, - packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData, + f: impl FnOnce(SeqNo, SeqNo) -> SendData, ) -> (Option>, Option) { let mut inner = self.seq.inner.lock().unwrap(); + if inner.closed { + return (self.consume_lock(inner), None); + } let pre_nst = inner.seq.next_service_timestamp; let seq_no = inner.seq.seq_no(); - inner - .seq - .reply_raw(self.tl, self.reply_no, false, seq_cst, packet_data(seq_no, self.reply_no)); + inner.seq.reply_raw(self.tl, self.reply_no, false, seq_cst, f(seq_no, self.reply_no)); let nst = inner.seq.next_service_timestamp; (self.consume_lock(inner), (pre_nst > nst).then_some(nst)) @@ -260,7 +266,9 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { let mut inner = self.seq.inner.lock().unwrap(); - inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock); + if !inner.closed { + inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock); + } self.seq.notify_reply(inner); } } @@ -346,26 +354,21 @@ impl SeqEx { recv_waiters: 0, reply_sender_waiters: false, reply_receiver_waiters: false, + closed: false, }), wait_on_recv: Condvar::default(), - wait_on_reply_receiver: Condvar::default(), - wait_on_reply_sender: Condvar::default(), + wait_on_reply: Condvar::default(), } } fn notify_reply(&self, mut inner: MutexGuard<'_, SeqExInner>) { - if inner.reply_receiver_waiters { - inner.reply_receiver_waiters = false; - drop(inner); - self.wait_on_reply_receiver.notify_all(); - } else if inner.reply_sender_waiters { + if inner.reply_sender_waiters || inner.reply_receiver_waiters { inner.reply_sender_waiters = false; drop(inner); - self.wait_on_reply_sender.notify_all(); + self.wait_on_reply.notify_all(); } } - fn notify_recv(&self, mut inner: MutexGuard<'_, SeqExInner>) { + fn notify_recv(&self, inner: MutexGuard<'_, SeqExInner>) { if inner.recv_waiters > 0 { - inner.recv_waiters -= 1; drop(inner); self.wait_on_recv.notify_one(); } @@ -384,12 +387,15 @@ impl SeqEx { packet: Packet, ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), RecvError> { let mut inner = self.inner.lock().unwrap(); - match inner.seq.receive_raw(tl, packet) { + if inner.closed { + return Err(RecvError::Closed); + } + match inner.seq.try_receive_raw(tl, packet) { Ok((r, do_pump)) => { self.notify_recv(inner); Ok((RecvOk::from_raw(self, tl, r), do_pump)) } - Err(e) => Err(e), + Err(e) => Err(e.into()), } } /// A SEQEX packet was just received and deserialized from the transport layer. @@ -407,12 +413,16 @@ impl SeqEx { /// the caller can now process, and the second value is a boolean specifying if there is yet /// more received data to be processed. This boolean is true if `SeqEx::pump` should be called, /// because there are more packets ready to be processed in the receive window. - pub fn receive>(&self, tl: TL, packet: Packet) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { + pub fn receive>( + &self, + tl: TL, + packet: Packet, + ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), RecvError> { let result = self.try_receive(tl, packet); if let Err(RecvError::WaitingForReply) = result { - self.pump(tl) + self.pump(tl).ok_or(RecvError::WaitingForReply) } else { - result.ok() + result } } /// Non-blocking variant of `SeqEx::pump`. @@ -454,7 +464,8 @@ impl SeqEx { Err(TryError::WaitingForRecv) => return None, Err(TryError::WaitingForReply) => { inner.reply_receiver_waiters = true; - inner = self.wait_on_reply_receiver.wait(inner).unwrap(); + inner = self.wait_on_reply.wait(inner).unwrap(); + inner.reply_receiver_waiters = false; } } } @@ -463,7 +474,7 @@ impl SeqEx { /// 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 { + if let Ok((first, do_pump)) = ret { RecvIter { seq: do_pump.then_some(self), tl, @@ -507,14 +518,21 @@ impl SeqEx { &self, tl: TL, seq_cst: bool, - packet_data: F, - ) -> Result, (TryError, F)> { + f: F, + ) -> Result, (TrySendError, F)> { let mut inner = self.inner.lock().unwrap(); + if inner.closed { + return Err((TrySendError::Closed, f)); + } 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) - }) + inner + .seq + .try_send_with(tl, seq_cst, f) + .map(|()| { + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) + }) + .map_err(|(e, b)| (e.into(), b)) } /// Non-blocking variant of `SeqEx::send`. /// See the documentation for `SeqEx::send` for more information. @@ -523,13 +541,25 @@ impl SeqEx { /// /// A return value of `Ok` contains the timestamp of when `service_scheduled` should be called next, /// only if it has decrease. This can be safely ignored if `service_scheduled` is not being used. - pub fn try_send>(&self, tl: TL, seq_cst: bool, packet_data: SendData) -> Result, (TryError, SendData)> { + pub fn try_send>( + &self, + tl: TL, + seq_cst: bool, + packet_data: SendData, + ) -> Result, (TrySendError, SendData)> { let mut inner = self.inner.lock().unwrap(); + if inner.closed { + return Err((TrySendError::Closed, 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) - }) + inner + .seq + .try_send(tl, seq_cst, packet_data) + .map(|()| { + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) + }) + .map_err(|(e, b)| (e.into(), b)) } /// Variant of `SeqEx::send` that allows for a provided function to write data to the send @@ -539,31 +569,47 @@ impl SeqEx { /// 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>(&self, tl: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) -> Option { + pub fn send_with>( + &self, + tl: TL, + seq_cst: bool, + mut f: impl FnOnce(SeqNo) -> SendData, + ) -> Result, ClosedError> { let mut inner = self.inner.lock().unwrap(); + if inner.closed { + return Err(ClosedError); + } 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; + while let Err((e, p)) = inner.seq.try_send_with(tl, seq_cst, f) { + f = p; match e { TryError::WaitingForRecv => { inner.recv_waiters += 1; inner = self.wait_on_recv.wait(inner).unwrap(); + inner.recv_waiters -= 1; pre_nst = inner.seq.next_service_timestamp; } TryError::WaitingForReply => { inner.reply_sender_waiters = true; - inner = self.wait_on_reply_sender.wait(inner).unwrap(); + inner = self.wait_on_reply.wait(inner).unwrap(); pre_nst = inner.seq.next_service_timestamp; } } + if inner.closed { + return Err(ClosedError); + } } let nst = inner.seq.next_service_timestamp; - (pre_nst > nst).then_some(nst) + Ok((pre_nst > nst).then_some(nst)) } /// 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. + /// The amount of time this blocks for can depend on how long it takes the remote peer to respond. + /// + /// This function can only return `Err(ClosedError)` if the function `SeqEx::close` has + /// previously been called on this instance of `SeqEx`. /// /// By default this function guarantees lossless transport. When `seq_cst` is set to true, this /// function also guarantees in-order transport. @@ -576,8 +622,11 @@ impl SeqEx { /// /// Returns the timestamp of when `service_scheduled` should be called next, only if it has decrease. /// This can be safely ignored if `service_scheduled` is not being used. - pub fn send>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) -> Option { + pub fn send>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) -> Result, ClosedError> { let mut inner = self.inner.lock().unwrap(); + if inner.closed { + return Err(ClosedError); + } 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; @@ -585,17 +634,21 @@ impl SeqEx { TryError::WaitingForRecv => { inner.recv_waiters += 1; inner = self.wait_on_recv.wait(inner).unwrap(); + inner.recv_waiters -= 1; pre_nst = inner.seq.next_service_timestamp; } TryError::WaitingForReply => { inner.reply_sender_waiters = true; - inner = self.wait_on_reply_sender.wait(inner).unwrap(); + inner = self.wait_on_reply.wait(inner).unwrap(); pre_nst = inner.seq.next_service_timestamp; } } + if inner.closed { + return Err(ClosedError); + } } let nst = inner.seq.next_service_timestamp; - (pre_nst > nst).then_some(nst) + Ok((pre_nst > nst).then_some(nst)) } /// Function which handles resending unacknowledged packets. /// It returns the duration of time in milliseconds that should be waited @@ -621,14 +674,56 @@ impl SeqEx { /// 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_scheduled(&self, mut tl: impl TransportLayer) -> i64 { + pub fn service_scheduled(&self, mut tl: impl TransportLayer) -> Result { let mut inner = self.inner.lock().unwrap(); + if inner.closed { + return Err(ClosedError); + } 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 + Ok(inner.seq.next_service_timestamp) + } + /// When this function is called, sending and receiving packets is immediately disabled for this + /// instance of `SeqEx`. Nothing else will be disabled, it will still be possible to empty any + /// packets left in the receive window with `pump`. + /// + /// Normally if the send window is full, threads that call into `send` or `send_with` will block + /// until the remote peer acknowledges some of the packets in the send window. + /// If the remote peer never sends an acknowledgement, then these threads will be + /// permanently blocked, unless this function is called. + /// + /// This function causes all threads currently waiting for the remote peer to send an ack to + /// unblock and return an error. + /// Future calls to `send` or `send_with` will return an error instead of blocking. + /// All future packets received from the remote peer will be dropped and an error will be returned. + /// + /// Calling this function can cause data loss due to packets being dropped from the send window, + /// but it will maintain state synchronization with the remote peer up to the moment it was + /// called. + /// + /// If you are using functions `send` or `send_with`, + /// and are communicating with an unreliable peer, then threads permanently blocking on send + /// is a possibility. To avoid this you must implement some system that will detect if the + /// remote peer has disconnected and then call this function when. + pub fn close(&self) { + // It is not practical to implement this function as a called-on-drop RAII object. + // Since SEQEX does not specify a protocol for closing a session, we have to allow the user + // to call this function at any time for any reason. + // RAII is good for when you need to guarantee some object is alive when a thread has access + // to it, but this is the opposite of the semantics we want for closing a session. + // When the user decides to close a session, this instance of `SeqEx` needs to immediately + // close even while other threads have access to it. + let mut inner = self.inner.lock().unwrap(); + inner.closed = true; + self.notify_recv(inner); + } + /// Return whether or not `close` has previously been called on this instance of `SeqEx`. + pub fn is_closed(&self) -> bool { + let inner = self.inner.lock().unwrap(); + inner.closed } } impl Default for SeqEx { diff --git a/src/tokio.rs b/src/tokio.rs index 74e2f21..dabdcec 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -5,7 +5,7 @@ use tokio::{ }; use crate::{ - error::{RecvError, TryError}, + error::{TryError, TryRecvError}, no_std::RecvOkRaw, Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; @@ -295,7 +295,7 @@ impl SeqEx { packet: Packet>, ) -> Result<(ReplyGuard<'_, TL, SendData, RecvData, CAP>, RecvData), Option> { let mut inner = self.inner.lock().unwrap(); - return match inner.seq.receive_raw(tl, packet) { + return match inner.seq.try_receive_raw(tl, packet) { Ok((recv_data, do_pump)) => { // pump first, handle return value second. let mut total_recv = 1; @@ -350,9 +350,9 @@ impl SeqEx { } ret } - Err(RecvError::DroppedTooEarly) => Err(Some(AsyncRecvError::DroppedTooEarly)), - Err(RecvError::DroppedDuplicate) => Err(Some(AsyncRecvError::DroppedDuplicate)), - Err(RecvError::WaitingForRecv) | Err(RecvError::WaitingForReply) => Err(None), + Err(TryRecvError::DroppedTooEarly) => Err(Some(AsyncRecvError::DroppedTooEarly)), + Err(TryRecvError::DroppedDuplicate) => Err(Some(AsyncRecvError::DroppedDuplicate)), + Err(TryRecvError::WaitingForRecv) | Err(TryRecvError::WaitingForReply) => Err(None), }; }