From 2740854710ffa2ec0250a259025034ecae1665b4 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 26 Oct 2023 10:33:54 -0400 Subject: [PATCH] implemented new API --- examples/calculator.rs | 10 +- examples/file_download.rs | 17 +++- examples/hello_world.rs | 2 +- src/error.rs | 138 ++++++++++++++++++++++++--- src/no_std.rs | 24 ++--- src/single_thread.rs | 39 ++++---- src/sync.rs | 190 ++++++++++++++++++++++---------------- 7 files changed, 283 insertions(+), 137 deletions(-) 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..e159d1c 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,32 @@ 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, + 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 +96,51 @@ 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, + 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 +151,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 +173,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/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 c7e3e0c..a712be3 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -3,11 +3,11 @@ use std::{ mpsc::{channel, Receiver, Sender}, Condvar, Mutex, MutexGuard, }, - time::{Instant, Duration}, + time::Instant, }; 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, }; @@ -39,7 +39,7 @@ struct SeqExInner { recv_waiters: usize, reply_sender_waiters: bool, reply_receiver_waiters: bool, - is_dead: 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,7 +354,7 @@ impl SeqEx { recv_waiters: 0, reply_sender_waiters: false, reply_receiver_waiters: false, - is_dead: false, + closed: false, }), wait_on_recv: Condvar::default(), wait_on_reply: Condvar::default(), @@ -359,7 +367,7 @@ impl SeqEx { 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 { drop(inner); self.wait_on_recv.notify_one(); @@ -379,15 +387,15 @@ impl SeqEx { packet: Packet, ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), RecvError> { let mut inner = self.inner.lock().unwrap(); - if inner.is_dead { - return Err(RecvError::WaitingForRecv); + if inner.closed { + return Err(RecvError::Closed); } - match inner.seq.receive_raw(tl, packet) { + 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. @@ -405,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`. @@ -439,11 +451,11 @@ impl SeqEx { /// 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. + if inner.reply_receiver_waiters { + return None; + } loop { - // Enforce that only one thread may wait to pump at a time. - if inner.reply_receiver_waiters { - return None; - } match inner.seq.try_pump_raw() { Ok((r, do_pump)) => { self.notify_recv(inner); @@ -462,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, @@ -506,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. @@ -522,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 @@ -538,16 +569,21 @@ 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 => { - if inner.is_dead { - return None; - } inner.recv_waiters += 1; inner = self.wait_on_recv.wait(inner).unwrap(); inner.recv_waiters -= 1; @@ -559,9 +595,12 @@ impl SeqEx { 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. @@ -569,6 +608,9 @@ impl SeqEx { /// 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` is called + /// while this thread is blocked on this + /// /// By default this function guarantees lossless transport. When `seq_cst` is set to true, this /// function also guarantees in-order transport. /// @@ -580,16 +622,16 @@ 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; match e { TryError::WaitingForRecv => { - if inner.is_dead { - return None; - } inner.recv_waiters += 1; inner = self.wait_on_recv.wait(inner).unwrap(); inner.recv_waiters -= 1; @@ -601,32 +643,8 @@ impl SeqEx { pre_nst = inner.seq.next_service_timestamp; } } - } - let nst = inner.seq.next_service_timestamp; - (pre_nst > nst).then_some(nst) - } - pub fn send_with_timeout, F: FnOnce(SeqNo) -> SendData>(&self, tl: TL, seq_cst: bool, dur: Duration, mut packet_data: F) -> Result, F> { - 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 => { - if inner.is_dead { - return Err(packet_data); - } - inner.recv_waiters += 1; - let result = self.wait_on_recv.wait_timeout(inner, dur).unwrap(); - inner.recv_waiters -= 1; - result.1.timed_out() - inner = .0; - pre_nst = inner.seq.next_service_timestamp; - } - TryError::WaitingForReply => { - inner.reply_sender_waiters = true; - 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; @@ -656,14 +674,17 @@ 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) } /// 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. @@ -671,7 +692,7 @@ impl SeqEx { /// permanently blocked, unless this function is called. /// /// This function cause all threads currently waiting for the remote peer to send us acks to - /// spuriously unblock and return. Future calls to `send` or `send_with` will spuriously return + /// unblock and return. Future calls to `send` or `send_with` will return an error /// instead of waiting for the remote peer to send acks. /// Any future packets received from the remote peer will be dropped. /// @@ -685,15 +706,22 @@ impl SeqEx { /// If you are using functions `send` or `send_with`, and are communicating with an unreliable /// peer, then threads permanently blocking is a possibility. To avoid this you must implement /// some system that will detect if the remote peer has disconnected and call this function. - pub fn assume_dead_peer(&self) { + 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.is_dead = true; + inner.closed = true; self.notify_recv(inner); } - /// Return whether or not `assume_dead_peer` has previously been called on this instance of `SeqEx`. - pub fn is_dead(&self) -> bool { + /// 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.is_dead + inner.closed } } impl Default for SeqEx {