Merge pull request #5 from zerotier/dev

Improved sync API

Cargo.lock

@@ -234,7 +234,7 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
 [[package]]
 name = "seq-ex"
-version = "1.0.0"
+version = "1.0.1"
 dependencies = [
  "rand_core",
  "serde",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "seq-ex"
-version = "1.0.0"
+version = "1.0.1"
 authors = ["Monica Moniot"]
 edition = "2021"
 

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 {

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);

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);

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<TryRecvError> 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<TryError> for TrySendError {
+    fn from(value: TryError) -> Self {
+        match value {
+            TryError::WaitingForRecv => TrySendError::WaitingForRecv,
+            TryError::WaitingForReply => TrySendError::WaitingForReply,
+        }
+    }
+}
+
+impl From<TryRawRecvError> 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,
+        }
+    }
+}

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;

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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         &mut self,
         current_time: i64,
         seq_cst: bool,
-        packet_data: F,
+        f: F,
     ) -> Result<Packet<&SendData>, (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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
             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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     }
 
     /// If this returns `Ok` then `try_send` might succeed on next call.
-    pub fn receive_raw_and_direct<P: Into<RecvData>>(&mut self, packet: Packet<P>) -> Result<(RecvOkRaw<SendData, P>, bool), TryRecvError> {
+    pub fn try_receive_raw_and_direct<P: Into<RecvData>>(&mut self, packet: Packet<P>) -> Result<(RecvOkRaw<SendData, P>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
                 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
             // 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
                 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)
             };
         }
 

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>, 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>, SendData, RecvData, const CAP: usize> Rep
     pub fn reply_stay_locked(self, seq_cst: bool, packet_data: SendData) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
         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<SeqCstGuard<'a, SendData, RecvData, CAP>> {
+    pub fn reply_with_stay_locked(self, seq_cst: bool, f: impl FnOnce(SeqNo, SeqNo) -> SendData) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
         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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         &mut self,
         mut tl: impl TransportLayer<SendData>,
         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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         }
     }
     /// If this returns `Ok` then `try_send` might succeed on next call.
-    pub fn receive_raw<P: Into<RecvData>>(
+    pub fn try_receive_raw<P: Into<RecvData>>(
         &mut self,
         mut tl: impl TransportLayer<SendData>,
         packet: Packet<P>,
-    ) -> Result<(RecvOkRaw<SendData, P>, bool), RecvError> {
-        match self.receive_raw_and_direct(packet) {
+    ) -> Result<(RecvOkRaw<SendData, P>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         &mut self,
         tl: TL,
         packet: Packet<RecvData>,
-    ) -> 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<TL: TransportLayer<SendData>>(&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))

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<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
     inner: Mutex<SeqExInner<SendData, RecvData, CAP>>,
     wait_on_recv: Condvar,
-    wait_on_reply_sender: Condvar,
-    wait_on_reply_receiver: Condvar,
+    wait_on_reply: Condvar,
 }
 struct SeqExInner<SendData, RecvData, const CAP: usize> {
     seq: crate::no_std::SeqEx<SendData, RecvData, CAP>,
     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>, 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<i64> {
+    pub fn reply_with(self, seq_cst: bool, f: impl FnOnce(SeqNo, SeqNo) -> SendData) -> Option<i64> {
         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>, 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<SeqCstGuard<'a, SendData, RecvData, CAP>> {
         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>, 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<SeqCstGuard<'a, SendData, RecvData, CAP>>, Option<i64>) {
         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>, SendData, RecvData, const CAP: usize> Rep
 impl<'a, TL: TransportLayer<SendData>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
                 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<SendData, RecvData, CAP>>) {
-        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<SendData, RecvData, CAP>>) {
+    fn notify_recv(&self, inner: MutexGuard<'_, SeqExInner<SendData, RecvData, CAP>>) {
         if inner.recv_waiters > 0 {
-            inner.recv_waiters -= 1;
             drop(inner);
             self.wait_on_recv.notify_one();
         }
@@ -384,12 +387,15 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         packet: Packet<RecvData>,
     ) -> 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// 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<TL: TransportLayer<SendData>>(&self, tl: TL, packet: Packet<RecvData>) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> {
+    pub fn receive<TL: TransportLayer<SendData>>(
+        &self,
+        tl: TL,
+        packet: Packet<RecvData>,
+    ) -> 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
                 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// automatically calls `SeqEx::pump` as many times as needed to empty the receive window.
     pub fn receive_all<TL: TransportLayer<SendData>>(&self, tl: TL, packet: Packet<RecvData>) -> 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         &self,
         tl: TL,
         seq_cst: bool,
-        packet_data: F,
-    ) -> Result<Option<i64>, (TryError, F)> {
+        f: F,
+    ) -> Result<Option<i64>, (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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     ///
     /// 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<TL: TransportLayer<SendData>>(&self, tl: TL, seq_cst: bool, packet_data: SendData) -> Result<Option<i64>, (TryError, SendData)> {
+    pub fn try_send<TL: TransportLayer<SendData>>(
+        &self,
+        tl: TL,
+        seq_cst: bool,
+        packet_data: SendData,
+    ) -> Result<Option<i64>, (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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// 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<TL: TransportLayer<SendData>>(&self, tl: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) -> Option<i64> {
+    pub fn send_with<TL: TransportLayer<SendData>>(
+        &self,
+        tl: TL,
+        seq_cst: bool,
+        mut f: impl FnOnce(SeqNo) -> SendData,
+    ) -> Result<Option<i64>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     ///
     /// 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<TL: TransportLayer<SendData>>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) -> Option<i64> {
+    pub fn send<TL: TransportLayer<SendData>>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) -> Result<Option<i64>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
                 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// 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<SendData>) -> i64 {
+    pub fn service_scheduled(&self, mut tl: impl TransportLayer<SendData>) -> Result<i64, ClosedError> {
         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<SendData, RecvData, const CAP: usize> Default for SeqEx<SendData, RecvData, CAP> {

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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         packet: Packet<IntoOneshot<'_, RecvData>>,
     ) -> Result<(ReplyGuard<'_, TL, SendData, RecvData, CAP>, RecvData), Option<AsyncRecvError>> {
         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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
                 }
                 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),
         };
     }