Merge branch 'dev'

README.md

@@ -2,24 +2,21 @@
 
 The reference implementation of the **Sequential Exchange Protocol**, or SEP.
 
-SEP is a peer-to-peer transport protocol that guarantees packets of data will always be received
-in the same order they were sent. In addition, it also guarantees the sequential consistency of
-stateful exchanges between the two communicating peers.
+SEP is a lightweight, peer-to-peer transport protocol that guarantees packets of data will be losslessly received by the remote peer, and can optionally guaranteed that specified packets arrive in the order that they were sent. In addition, SEP facilitates stateful exchanges between two peers, giving each peer the opportunity to "reply" to any packet sent by the remote peer. This makes SEP particularly well-suited for writing async-await code, because unlike TCP, SEP will handle multiplexing each reply to the correct awaiter. Even without async-await, a simple `match` statement is sufficient to correctly multiplex packets to their handling code.
 
 A "stateful exchange" is defined here as a sequence of packets, where the first packet
 initiates the exchange, and all subsequent packets are replies to the previous packet in the
-exchange.
-
-SEP guarantees both peers will agree upon which packets are members of which exchanges,
-and it guarantees each packet is received by each peer in sequential order.
+exchange. Every exchange can be thought of as a linked list, where the head node is a packet containing a normal payload, and all subsequent nodes are replies to previous nodes. The final node is always a simple acknowledgement packet, or Ack, that signals a given exchange is over. Both peers are guaranteed to agree upon the "topology" of these links. Links will never get crossed, replies will always be received and understood, a peer will never deadlock awaiting a reply, and in general it is much easier to write bug-free networking code.
 
 SEP is a tiny, dead simple protocol, the core of which we have implemented in less than 1000 lines of code.
 
+SEP is transport agnostic, and will not take over an entire UDP socket. As such it is relatively easier to run SEP in parrallel with raw UDP, or even with itself. Multiple instances of SEP can be opened between two peers and communication over each can occur in parrallel, making it very easy to reduce or even eliminate front-of-line latency in performance critical applications.
+
 ## Why not TCP?
 
 TCP only guarantees packets will be received in the same order they were sent.
 It has no inherent concept of "replying to a packet" and as such it cannot guarantee both sides
-of a conversation have the same view of any stateful exchanges that take place.
+of a conversation have the same view of any stateful exchanges that take place. This must be implemented manually by the user of TCP.
 
 TCP is also much higher overhead. It requires a 1.5 RTT handshake to begin any connection,
 it has a larger amount of metadata that must be transported with packets, and it has quite a few

examples/file_download.rs

@@ -52,7 +52,7 @@ fn drop_packet() -> bool {
     OsRng.next_u32() >= (u32::MAX / 4 * 3)
 }
 
-fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload, Payload>) {
+fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload>) {
     use Payload::*;
     match recv_data.consume() {
         (Some((guard, RequestFile { filename })), None) => {

examples/hello_world_tokio.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use tokio::{sync::mpsc, task};
 
 use seq_ex::{
-    tokio::{AsyncRecvError, MpscTransport, ReplyGuard, SeqExTokio},
+    tokio::{MpscTransport, ReplyGuard, SeqExTokio},
     Packet,
 };
 
@@ -17,7 +17,7 @@ enum Payload {
 }
 use Payload::*;
 
-async fn receive(reply_guard: ReplyGuard<'_, &MpscTransport<Payload>, Payload>, payload: Payload) -> Option<()> {
+async fn receive(reply_guard: ReplyGuard<'_, &MpscTransport<Payload>, Payload, Payload>, payload: Payload) -> Option<()> {
     match payload {
         Hello => {
             print!("Hello");
@@ -42,7 +42,7 @@ async fn receive(reply_guard: ReplyGuard<'_, &MpscTransport<Payload>, Payload>,
     Some(())
 }
 
-async fn say_hello(seq: &SeqExTokio<Payload>, transport: &MpscTransport<Payload>) -> Option<()> {
+async fn say_hello(seq: &SeqExTokio<Payload, Payload>, transport: &MpscTransport<Payload>) -> Option<()> {
     let (reply_guard, payload) = seq.send(transport, false, Hello).await.ok()?;
     if payload != Space {
         return None;
@@ -59,20 +59,8 @@ async fn say_hello(seq: &SeqExTokio<Payload>, transport: &MpscTransport<Payload>
     Some(())
 }
 
-async fn pump_all(peer: Arc<SeqExTokio<Payload>>, transport: MpscTransport<Payload>) {
-    if let Some((g, payload)) = peer.pump(&transport).await {
-        spawn_pump(peer.clone(), transport.clone());
-        receive(g, payload).await;
-    }
-}
-fn spawn_pump(peer: Arc<SeqExTokio<Payload>>, transport: MpscTransport<Payload>) {
-    task::spawn(async move {
-        pump_all(peer, transport).await;
-    });
-}
-
-fn peer_main(transport: MpscTransport<Payload>, mut recv: mpsc::Receiver<Packet<Payload>>) -> Arc<SeqExTokio<Payload>> {
-    let (seq, mut service) = SeqExTokio::<Payload>::new_default();
+fn peer_main(transport: MpscTransport<Payload>, mut recv: mpsc::Receiver<Packet<Payload>>) -> Arc<SeqExTokio<Payload, Payload>> {
+    let (seq, mut service) = SeqExTokio::new_default();
     let peer = Arc::new(seq);
     let peer_weak = Arc::downgrade(&peer);
     let tl = transport.clone();
@@ -87,11 +75,7 @@ fn peer_main(transport: MpscTransport<Payload>, mut recv: mpsc::Receiver<Packet<
             if let Some(peer) = peer_weak.upgrade() {
                 let tl = transport.clone();
                 task::spawn(async move {
-                    let result = peer.receive(&tl, packet);
-                    if matches!(&result, Ok(_) | Err(AsyncRecvError::AsyncReply)) {
-                        spawn_pump(peer.clone(), tl.clone());
-                    }
-                    if let Ok((g, payload)) = result {
+                    if let Ok((g, payload)) = peer.receive(&tl, packet).await {
                         receive(g, payload).await;
                     }
                 });
@@ -112,6 +96,7 @@ async fn main() {
 
     say_hello(&peer1, &transport1).await;
 }
+
 #[test]
 fn test() {
     main()

phi_accural.rs

@@ -0,0 +1,145 @@
+use std::sync::mpsc::Receiver;
+
+use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync};
+
+#[derive(Clone, Debug)]
+enum Packet {
+    Hello,
+    Space,
+    World,
+    Exclamation,
+}
+use Packet::*;
+
+fn process(guard: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, send_packet: Option<Packet>) {
+    match (recv_packet, send_packet) {
+        (Hello, None) => {
+            print!("Hello");
+            guard.reply(Space);
+        }
+        (Space, Some(Hello)) => {
+            print!(" ");
+            guard.reply(World);
+        }
+        (World, Some(Space)) => {
+            print!("World");
+            guard.reply(Exclamation);
+        }
+        (Exclamation, Some(World)) => {
+            print!("!");
+        }
+        (a, None) => {
+            print!("Unsolicited packet received: {:?}", a);
+        }
+        (a, Some(b)) => {
+            print!("Incorrect reply received: {:?}, was a reply to: {:?}", a, b);
+        }
+    }
+}
+
+fn receive<'a>(recv: &Receiver<PacketType<Packet>>, seq: &SeqExSync<&'a MpscTransport<Packet>>, transport: &'a MpscTransport<Packet>) {
+    let do_pump = match recv.recv().unwrap() {
+        PacketType::Ack { reply_no } => {
+            seq.receive_ack(reply_no);
+            return;
+        }
+        PacketType::EmptyReply { reply_no } => {
+            let result = seq.receive_empty_reply(reply_no);
+            if let Some(Exclamation) = &result {
+                // Our Hello World exchange ends right here.
+                print!("\n");
+            }
+            result.is_some()
+        }
+        PacketType::Payload { seq_no, reply_no, payload } => {
+            if let Ok(RecvSuccess { guard, packet, send_data }) = seq.receive(transport, seq_no, reply_no, payload) {
+                process(guard, packet, send_data);
+                true
+            } else {
+                false
+            }
+        }
+    };
+    if do_pump {
+        while let Ok(RecvSuccess { guard, packet, send_data }) = seq.pump(transport) {
+            process(guard, packet, send_data);
+        }
+    }
+}
+
+pub const WINDOW_SIZE: usize = 32;
+pub const DEFAULT_ALLOWED_MISSES: f64 = 1.0;
+pub const DEFAULT_ALLOWED_PROB: f64 = .01;
+
+/// This version of phi accural takes into account the possibility of packets being dropped uniformly at random from the network, and computes an approximation of that cdf. We do not attempt to dynamically compute the loss rate (since packet loss is not uniform or independent irl), but instead require the user preprogram an `allowed_misses` parameter.
+/// `allowed_misses` is an estimation of the number of phi accural packets that the user thinks could possibly be dropped in a row given that both the network and the remote peer are still alive.
+///
+/// The exact distribution we simulate is the probability that the peer is dead, given the amount of time since the last received phi accural packet, and given that `allowed_misses` number of phi accural packets have been or will be dropped from the network.
+#[derive(Clone)]
+pub struct PhiAccumulator {
+    pub allowed_misses: f64,
+    pub allowed_prob: f64,
+    head_idx: usize,
+    intervals: [f64; WINDOW_SIZE],
+    last_time: i64,
+    mean: f64,
+    std: f64,
+}
+
+pub fn normal_cdf_apprx(position: f64, mean: f64, std: f64) -> f64 {
+
+}
+
+impl PhiAccumulator {
+    pub fn new(allowed_prob_of_failure: f64, allowed_misses: f64, expected_first_interval: f64, current_time: i64) -> Self {
+        PhiAccumulator {
+            allowed_misses,
+            allowed_prob: allowed_prob_of_failure,
+            head_idx: 0,
+            intervals: std::array::from_fn(|_| expected_first_interval),
+            last_time: current_time,
+            mean: expected_first_interval,
+            std: 0.0,
+        }
+    }
+    /// Returns false if it is likely that the remote peer is dead or unreachable.
+    pub fn check(&self, current_time: i64) -> bool {
+        let prob = normal_cdf_apprx((current_time - self.last_time) as f32, (self.allowed_misses + 1.0)*self.mean, self.std);
+        prob > self.allowed_prob
+    }
+    /// Updates the internal state to acknowledge a just received phi accural packet.
+    pub fn just_received_phi_packet(&mut self, current_time: i64) {
+        let new_interval = (current_time - self.last_time) as f64;
+        self.last_time = current_time;
+        let idx = self.head_idx;
+        self.head_idx += 1;
+        // We compute a rolling mean, which is fast but suceptible to rounding errors. Hence f64.
+        self.mean += (new_interval - self.intervals[idx])/WINDOW_SIZE as f64;
+        self.intervals[idx] = new_interval;
+        let mut std = 0.0;
+        for x in self.intervals {
+            let diff = (x - self.mean);
+            std += diff*diff;
+        }
+        ///
+        self.std = std.sqrt()/WINDOW_SIZE as f64;
+    }
+}
+
+
+
+fn main() {
+    let (transport1, recv2) = MpscTransport::new();
+    let (transport2, recv1) = MpscTransport::new();
+    let seq1 = SeqExSync::default();
+    let seq2 = SeqExSync::default();
+
+    // We begin a "Hello World" exchange right here.
+    seq1.send(&transport1, Packet::Hello);
+
+    receive(&recv2, &seq2, &transport2);
+    receive(&recv1, &seq1, &transport1);
+    receive(&recv2, &seq2, &transport2);
+    receive(&recv1, &seq1, &transport1);
+    receive(&recv2, &seq2, &transport2);
+}

src/seq_queue.rs

@@ -40,7 +40,7 @@
 pub type SeqNo = u32;
 
 /// The resend interval for a default instance of SeqEx.
-pub const DEFAULT_RESEND_INTERVAL_MS: i64 = 200;
+pub const DEFAULT_RESEND_INTERVAL_MS: i64 = 250;
 /// The initial sequence number for a default instance of SeqEx.
 pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 0;
 
@@ -50,11 +50,11 @@ pub const DEFAULT_WINDOW_CAP: usize = 64;
 pub struct SeqEx<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
     /// The interval at which packets will be resent if they have not yet been acknowledged by the
     /// remote peer.
-    /// It can be statically or dynamically set, it is up to the user to decide.
+    /// It can be statically or dynamically set.
     pub resend_interval: i64,
     pub next_service_timestamp: i64,
     next_send_seq_no: SeqNo,
-    pre_recv_seq_no: SeqNo,
+    next_recv_seq_no: SeqNo,
     /// This could be made more efficient by changing to SoA format.
     send_window: [Option<SendEntry<SendData>>; CAP],
     recv_window: [RecvEntry<RecvData>; CAP],
@@ -91,9 +91,20 @@ struct SendEntry<SendData> {
     next_resend_time: i64,
     data: SendData,
 }
+impl<SendData> SendEntry<SendData> {
+    fn to_packet(&self) -> Packet<&SendData> {
+        let mut p = if let Some(reply_no) = self.reply_no {
+            Reply(self.seq_no, reply_no, &self.data)
+        } else {
+            Payload(self.seq_no, &self.data)
+        };
+        p.set_seq_cst(self.seq_cst);
+        p
+    }
+}
 
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub enum DirectRecvError {
+pub enum TryRecvError {
     DroppedTooEarly,
     DroppedDuplicate,
     DroppedDuplicateResendAck(SeqNo),
@@ -101,36 +112,36 @@ pub enum DirectRecvError {
     WaitingForReply,
 }
 #[cfg(feature = "std")]
-impl std::fmt::Display for DirectRecvError {
+impl std::fmt::Display for TryRecvError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            DirectRecvError::DroppedTooEarly => write!(f, "packet arrived too early"),
-            DirectRecvError::DroppedDuplicate => write!(f, "packet was a duplicate"),
-            DirectRecvError::DroppedDuplicateResendAck(_) => write!(f, "packet was a duplicate, resending ack"),
-            DirectRecvError::WaitingForRecv => write!(f, "can't process until another packet is received"),
-            DirectRecvError::WaitingForReply => write!(f, "can't process until a reply is finished"),
+            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"),
         }
     }
 }
 #[cfg(feature = "std")]
-impl std::error::Error for DirectRecvError {}
+impl std::error::Error for TryRecvError {}
 
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub enum PumpError {
+pub enum TryError {
     WaitingForRecv,
     WaitingForReply,
 }
 #[cfg(feature = "std")]
-impl std::fmt::Display for PumpError {
+impl std::fmt::Display for TryError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            PumpError::WaitingForRecv => write!(f, "can't process until another packet is received"),
-            PumpError::WaitingForReply => write!(f, "can't process until a reply is finished"),
+            TryError::WaitingForRecv => write!(f, "can't process until another packet is received"),
+            TryError::WaitingForReply => write!(f, "can't process until a reply is finished"),
         }
     }
 }
 #[cfg(feature = "std")]
-impl std::error::Error for PumpError {}
+impl std::error::Error for TryError {}
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
@@ -249,7 +260,7 @@ pub struct IterMut<'a, SendData>(core::slice::IterMut<'a, Option<SendEntry<SendD
 
 #[derive(Clone, Debug)]
 pub struct ServiceIter {
-    idx: usize,
+    seq_no: SeqNo,
     next_time: i64,
 }
 
@@ -264,11 +275,12 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// `initial_seq_no` is the first sequence number that this instance of `SeqEx` will use. It must be
     /// exactly the same as the `initial_seq_no` of the remote instance of `SeqEx`. It can just be 1.
     pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
+        debug_assert!(CAP > 1);
         Self {
             resend_interval: retry_interval,
             next_service_timestamp: i64::MAX,
             next_send_seq_no: initial_seq_no,
-            pre_recv_seq_no: initial_seq_no.wrapping_sub(1),
+            next_recv_seq_no: initial_seq_no,
             recv_window: core::array::from_fn(|_| RecvEntry::Empty),
             send_window: core::array::from_fn(|_| None),
             concurrent_replies: core::array::from_fn(|_| 0),
@@ -281,15 +293,22 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         &mut self.send_window[seq_no as usize % self.send_window.len()]
     }
     #[inline]
-    fn is_full_inner(&self, is_for_reply: bool, reply_no: Option<SeqNo>) -> bool {
+    fn is_full_inner(&self, reply_no: Option<SeqNo>) -> Result<(), TryError> {
+        if self.concurrent_replies_total >= self.concurrent_replies.len() - 2 {
+            return Err(TryError::WaitingForReply);
+        }
         let reply_idx = reply_no.map_or(self.send_window.len(), |r| r as usize % self.send_window.len());
-        for i in 0..self.concurrent_replies_total as u32 + 1 + is_for_reply as u32 {
+        for i in 0..1 + self.concurrent_replies_total as u32 {
             let idx = self.next_send_seq_no.wrapping_add(i) as usize % self.send_window.len();
             if self.send_window[idx].is_some() && reply_idx != idx {
-                return true;
+                return if i == 0 {
+                    Err(TryError::WaitingForRecv)
+                } else {
+                    Err(TryError::WaitingForReply)
+                };
             }
         }
-        false
+        Ok(())
     }
 
     #[inline]
@@ -312,14 +331,6 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         }
         false
     }
-
-    /// Returns whether or not the send window is full.
-    /// If the send window is full calls to `SeqEx::send` will always fail.
-    pub fn is_full(&self) -> bool {
-        // We claim that the window is full one entry before it is actually full for the sake of
-        // making it always possible for both peers to process at least one reply at all times.
-        self.is_full_inner(true, None)
-    }
     /// Returns the next sequence number to be attached to the next sent packet.
     /// This should be called before `SeqEx::send`, and the return value should be
     /// included in some way with the `packet_data` parameter passed to `SeqEx::send`.
@@ -348,11 +359,11 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// `retry_interval`. `current_time` does not have to be monotonically increasing.
     ///
     /// Can mutate `next_service_timestamp`.
-    pub fn try_send_direct(&mut self, current_time: i64, seq_cst: bool, packet_data: SendData) -> Result<Packet<&SendData>, SendData> {
+    pub fn try_send_direct(&mut self, current_time: i64, seq_cst: bool, packet_data: SendData) -> Result<Packet<&SendData>, (TryError, SendData)> {
         let mut tmp = Some(packet_data);
         self.try_send_direct_with(current_time, seq_cst, |_| tmp.take().unwrap())
-            .map_err(|_| ())
-            .map_err(|_| tmp.unwrap())
+            .map_err(|e| e.0)
+            .map_err(|e| (e, tmp.unwrap()))
     }
     /// Can mutate `next_service_timestamp`.
     pub fn try_send_direct_with<F: FnOnce(SeqNo) -> SendData>(
@@ -360,10 +371,11 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         current_time: i64,
         seq_cst: bool,
         packet_data: F,
-    ) -> Result<Packet<&SendData>, F> {
-        if self.is_full() {
-            return Err(packet_data);
+    ) -> Result<Packet<&SendData>, (TryError, F)> {
+        if let Err(e) = self.is_full_inner(None) {
+            return Err((e, packet_data));
         }
+
         let seq_no = self.next_send_seq_no;
         self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1);
 
@@ -381,13 +393,27 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
             data: packet_data(seq_no),
         });
 
-        let mut p = Payload(entry.seq_no, &entry.data);
-        p.set_seq_cst(seq_cst);
-        Ok(p)
+        Ok(entry.to_packet())
+    }
+
+    fn fast_forward(&mut self) -> bool {
+        loop {
+            let next_seq_no = self.next_recv_seq_no;
+            let i = next_seq_no as usize % self.recv_window.len();
+            match &self.recv_window[i] {
+                RecvEntry::Unlocked { seq_no } => {
+                    debug_assert_eq!(*seq_no, next_seq_no);
+                    self.recv_window[i] = RecvEntry::Empty;
+                    self.next_recv_seq_no = next_seq_no.wrapping_add(1);
+                }
+                RecvEntry::Occupied { .. } => return true,
+                RecvEntry::Empty => return false,
+            }
+        }
     }
 
     /// 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>, DirectRecvError> {
+    pub fn receive_raw_and_direct<P: Into<RecvData>>(&mut self, packet: Packet<P>) -> Result<(RecvOkRaw<SendData, P>, bool), TryRecvError> {
         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),
@@ -395,15 +421,15 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
             Ack(reply_no) => {
                 return self
                     .take_send(reply_no)
-                    .map(|send_data| RecvOkRaw::Ack { send_data })
-                    .ok_or(DirectRecvError::DroppedDuplicate)
+                    .map(|send_data| (RecvOkRaw::Ack { send_data }, false))
+                    .ok_or(TryRecvError::DroppedDuplicate)
             }
         };
         // We only want to accept packets with sequence numbers in the range:
-        // `self.pre_recv_seq_no < seq_no <= self.pre_recv_seq_no + self.recv_window.len()`.
-        // To check that range we compute `seq_no - (self.pre_recv_seq_no + 1)` and check
+        // `self.next_recv_seq_no <= seq_no < self.next_recv_seq_no + self.recv_window.len()`.
+        // To check that range we compute `seq_no - self.next_recv_seq_no` and check
         // if the number wrapped below 0, or if it is above `self.recv_window.len()`.
-        let normalized_seq_no = seq_no.wrapping_sub(self.pre_recv_seq_no).wrapping_sub(1);
+        let normalized_seq_no = seq_no.wrapping_sub(self.next_recv_seq_no);
         let is_below_range = normalized_seq_no > SeqNo::MAX / 2;
         let is_above_range = !is_below_range && normalized_seq_no >= self.recv_window.len() as u32;
         let is_next = normalized_seq_no == 0;
@@ -419,17 +445,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(DirectRecvError::DroppedDuplicate);
+                    return Err(TryRecvError::DroppedDuplicate);
                 }
             }
             for i in 0..self.concurrent_replies_total {
                 if self.concurrent_replies[i] == seq_no {
-                    return Err(DirectRecvError::DroppedDuplicate);
+                    return Err(TryRecvError::DroppedDuplicate);
                 }
             }
-            return Err(DirectRecvError::DroppedDuplicateResendAck(seq_no));
+            return Err(TryRecvError::DroppedDuplicateResendAck(seq_no));
         } else if is_above_range {
-            return Err(DirectRecvError::DroppedTooEarly);
+            return Err(TryRecvError::DroppedTooEarly);
         }
 
         // Check whether or not we've already received this packet
@@ -445,84 +471,78 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         // If the send window is full we cannot safely process received packets,
         // because there would be no way to reply.
         // We can only process this packet if processing it would make space in the send window.
-        let wait_for_recv = self.is_full_inner(false, reply_no) || (seq_cst && !is_next);
-        let wait_for_reply = self.concurrent_replies_total >= self.concurrent_replies.len() || (seq_cst && self.is_locked);
-        if wait_for_recv || wait_for_reply {
+        let (wait_recv, wait_reply) = match self.is_full_inner(reply_no) {
+            Ok(()) => (seq_cst && !is_next, seq_cst && self.is_locked),
+            Err(TryError::WaitingForRecv) => (true, false),
+            Err(TryError::WaitingForReply) => (false, true),
+        };
+        if wait_recv || wait_reply {
             if !is_duplicate {
                 self.recv_window[i] = RecvEntry::Occupied { seq_no, reply_no, seq_cst, data: recv_data.into() }
             }
-            return if wait_for_recv {
-                Err(DirectRecvError::WaitingForRecv)
+            return if wait_recv {
+                Err(TryRecvError::WaitingForRecv)
             } else {
-                Err(DirectRecvError::WaitingForReply)
+                Err(TryRecvError::WaitingForReply)
             };
         }
 
-        if is_next {
+        let do_pump = if is_next {
             self.recv_window[i] = RecvEntry::Empty;
-            self.pre_recv_seq_no = seq_no;
+            self.next_recv_seq_no = seq_no.wrapping_add(1);
+            self.fast_forward()
         } else {
             debug_assert!(!seq_cst);
-            self.recv_window[i] = RecvEntry::Unlocked { seq_no }
-        }
+            self.recv_window[i] = RecvEntry::Unlocked { seq_no };
+            false
+        };
         self.concurrent_replies[self.concurrent_replies_total] = seq_no;
         self.concurrent_replies_total += 1;
         if seq_cst {
             debug_assert!(!self.is_locked);
             self.is_locked = true;
         }
-        Ok(if let Some(send_data) = reply_no.and_then(|r| self.take_send(r)) {
-            RecvOkRaw::Reply { reply_no: seq_no, recv_data, send_data, seq_cst }
+        if let Some(send_data) = reply_no.and_then(|r| self.take_send(r)) {
+            Ok((RecvOkRaw::Reply { reply_no: seq_no, recv_data, send_data, seq_cst }, do_pump))
         } else {
-            RecvOkRaw::Payload { reply_no: seq_no, recv_data, seq_cst }
-        })
+            Ok((RecvOkRaw::Payload { reply_no: seq_no, recv_data, seq_cst }, do_pump))
+        }
     }
     /// If this returns `Ok` then `try_send` might succeed on next call.
-    pub fn try_pump_raw(&mut self) -> Result<RecvOkRaw<SendData, RecvData>, PumpError> {
-        let mut next_seq_no;
-        let mut i;
-        loop {
-            next_seq_no = self.pre_recv_seq_no.wrapping_add(1);
-            i = next_seq_no as usize % self.recv_window.len();
-            if let RecvEntry::Unlocked { seq_no } = &self.recv_window[i] {
-                debug_assert_eq!(*seq_no, next_seq_no);
-                self.recv_window[i] = RecvEntry::Empty;
-                self.pre_recv_seq_no = next_seq_no;
-            } else {
-                break;
-            }
-        }
-
+    pub fn try_pump_raw(&mut self) -> Result<(RecvOkRaw<SendData, RecvData>, bool), TryError> {
+        let next_seq_no = self.next_recv_seq_no;
+        let i = next_seq_no as usize % self.recv_window.len();
         if let RecvEntry::Occupied { seq_no, reply_no, seq_cst, .. } = &self.recv_window[i] {
             debug_assert_eq!(*seq_no, next_seq_no);
-            if self.is_full_inner(false, *reply_no) {
-                return Err(PumpError::WaitingForRecv);
-            }
+            // We cannot safely reserve a reply no if the window is full.
+            self.is_full_inner(*reply_no)?;
 
-            if (!*seq_cst || !self.is_locked) && self.concurrent_replies_total < self.concurrent_replies.len() {
+            if !*seq_cst || !self.is_locked {
                 let mut entry = RecvEntry::Empty;
                 core::mem::swap(&mut entry, &mut self.recv_window[i]);
                 if let RecvEntry::Occupied { seq_no, reply_no, seq_cst, data } = entry {
-                    self.pre_recv_seq_no = next_seq_no;
+                    self.next_recv_seq_no = next_seq_no.wrapping_add(1);
+                    let do_pump = self.fast_forward();
                     self.concurrent_replies[self.concurrent_replies_total] = seq_no;
                     self.concurrent_replies_total += 1;
                     if seq_cst {
                         debug_assert!(!self.is_locked);
                         self.is_locked = true;
                     }
-                    return Ok(if let Some(send_data) = reply_no.and_then(|r| self.take_send(r)) {
+                    let ret = if let Some(send_data) = reply_no.and_then(|r| self.take_send(r)) {
                         RecvOkRaw::Reply { reply_no: seq_no, seq_cst, recv_data: data, send_data }
                     } else {
                         RecvOkRaw::Payload { reply_no: seq_no, seq_cst, recv_data: data }
-                    });
+                    };
+                    return Ok((ret, do_pump));
                 } else {
                     unreachable!();
                 }
             } else {
-                return Err(PumpError::WaitingForReply);
+                return Err(TryError::WaitingForReply);
             }
         }
-        Err(PumpError::WaitingForRecv)
+        Err(TryError::WaitingForRecv)
     }
     /// This function must be passed a reply number given by `receive_raw` or `pump_raw`, otherwise
     /// it will do nothing. This reply number can only be used to reply once.
@@ -566,9 +586,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
                 data: packet_data,
             });
 
-            let mut p = Reply(entry.seq_no, reply_no, &entry.data);
-            p.set_seq_cst(seq_cst);
-            Some(p)
+            Some(entry.to_packet())
         } else {
             None
         }
@@ -590,22 +608,19 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// Can mutate `next_service_timestamp`.
     pub fn service_direct(&mut self, current_time: i64, iter: &mut Option<ServiceIter>) -> Option<Packet<&SendData>> {
         if self.next_service_timestamp <= current_time {
-            let iter = iter.get_or_insert(ServiceIter { idx: 0, next_time: i64::MAX });
-            while let Some(entry) = self.send_window.get(iter.idx) {
-                iter.idx += 1;
-                if let Some(entry) = entry {
+            let iter = iter.get_or_insert(ServiceIter {
+                seq_no: self.next_send_seq_no.wrapping_sub(self.send_window.len() as u32),
+                next_time: i64::MAX,
+            });
+            while iter.seq_no != self.next_send_seq_no {
+                let idx = iter.seq_no as usize % self.send_window.len();
+                iter.seq_no = iter.seq_no.wrapping_add(1);
+                if let Some(entry) = &mut self.send_window[idx] {
                     if entry.next_resend_time <= current_time {
-                        let entry = self.send_window[iter.idx - 1].as_mut().unwrap();
                         entry.next_resend_time = current_time + self.resend_interval;
                         iter.next_time = iter.next_time.min(entry.next_resend_time);
 
-                        let mut p = if let Some(reply_no) = entry.reply_no {
-                            Reply(entry.seq_no, reply_no, &entry.data)
-                        } else {
-                            Payload(entry.seq_no, &entry.data)
-                        };
-                        p.set_seq_cst(entry.seq_cst);
-                        return Some(p);
+                        return Some(self.send_window[idx].as_ref().unwrap().to_packet());
                     } else {
                         iter.next_time = iter.next_time.min(entry.next_resend_time);
                     }

src/single_thread.rs

@@ -1,4 +1,4 @@
-use crate::{DirectRecvError, Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
+use crate::{Packet, RecvOkRaw, SeqEx, SeqNo, TransportLayer, TryError, TryRecvError, DEFAULT_WINDOW_CAP};
 
 pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
     seq: &'a mut SeqEx<SendData, RecvData, CAP>,
@@ -7,28 +7,35 @@ pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, cons
     is_holding_lock: bool,
 }
 impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
+    pub fn ack(&mut self) {
+        if let Some(app) = self.app.take() {
+            self.seq.ack_raw(app, self.reply_no, self.is_holding_lock);
+        }
+    }
     /// If you need to reply more than once, say to fragment a large file, then include in your
     /// first reply some identifier, and then `send` all fragments with the same included identifier.
     /// The identifier will tell the remote peer which packets contain fragments of the file,
     /// and since each fragment will be received in order it will be trivial for them to reconstruct
     /// the original file.
+    /// # Panic
+    /// This function will panic if `ack` has been called.
     pub fn reply(self, seq_cst: bool, packet_data: SendData) {
         self.reply_with(seq_cst, |_, _| packet_data)
     }
+    /// # Panic
+    /// This function will panic if `ack` has been called.
     fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
-        let app = self.app.take();
+        let app = self.app.take().expect("Cannot reply after an ack has been sent");
         let seq_no = self.seq.seq_no();
-        self.seq.reply_raw(
-            app.unwrap(),
-            self.reply_no,
-            self.is_holding_lock,
-            seq_cst,
-            packet_data(seq_no, self.reply_no),
-        );
+        self.seq
+            .reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no));
+        core::mem::forget(self);
     }
 
-    pub fn to_components(mut self) -> (TL, SeqNo, bool) {
-        (self.app.take().unwrap(), self.reply_no, self.is_holding_lock)
+    pub fn to_components(self) -> (SeqNo, bool) {
+        let ret = (self.reply_no, self.is_holding_lock);
+        core::mem::forget(self);
+        ret
     }
     pub unsafe fn from_components(seq: &'a mut SeqEx<SendData, RecvData, CAP>, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self {
         ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock }
@@ -36,11 +43,7 @@ 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) {
-        if let Some(app) = &mut self.app {
-            if let Some(p) = self.seq.ack_raw_and_direct(self.reply_no, self.is_holding_lock) {
-                app.send(p)
-            }
-        }
+        self.ack();
     }
 }
 impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
@@ -73,14 +76,14 @@ impl std::fmt::Display for RecvError {
 #[cfg(feature = "std")]
 impl std::error::Error for RecvError {}
 
-pub enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
+pub enum RecvOk<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
     Payload {
         reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
-        recv_data: P,
+        recv_data: RecvData,
     },
     Reply {
         reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
-        recv_data: P,
+        recv_data: RecvData,
         send_data: SendData,
     },
     Ack {
@@ -90,8 +93,8 @@ pub enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const C
 macro_rules! impl_recvok {
     ($recv:tt, $seq_ex:ty) => {
         #[cfg(feature = "std")]
-        impl<'a, TL: TransportLayer<SendData>, P: std::fmt::Debug, SendData: std::fmt::Debug, RecvData, const CAP: usize> std::fmt::Debug
-            for $recv<'a, TL, P, SendData, RecvData, CAP>
+        impl<'a, TL: TransportLayer<SendData>, SendData: std::fmt::Debug, RecvData: std::fmt::Debug, const CAP: usize> std::fmt::Debug
+            for $recv<'a, TL, SendData, RecvData, CAP>
         {
             fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                 match self {
@@ -110,8 +113,8 @@ macro_rules! impl_recvok {
                 }
             }
         }
-        impl<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize> $recv<'a, TL, P, SendData, RecvData, CAP> {
-            fn from_raw(seq: $seq_ex, app: TL, value: RecvOkRaw<SendData, P>) -> Self {
+        impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> $recv<'a, TL, SendData, RecvData, CAP> {
+            fn from_raw(seq: $seq_ex, app: TL, value: RecvOkRaw<SendData, RecvData>) -> Self {
                 match value {
                     RecvOkRaw::Payload { reply_no, seq_cst, recv_data } => Self::Payload {
                         reply_guard: ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock: seq_cst },
@@ -125,14 +128,14 @@ macro_rules! impl_recvok {
                     RecvOkRaw::Ack { send_data } => Self::Ack { send_data },
                 }
             }
-            pub fn consume(self) -> (Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, Option<SendData>) {
+            pub fn consume(self) -> (Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, RecvData)>, Option<SendData>) {
                 match self {
                     Self::Payload { reply_guard, recv_data } => (Some((reply_guard, recv_data)), None),
                     Self::Reply { reply_guard, recv_data, send_data } => (Some((reply_guard, recv_data)), Some(send_data)),
                     Self::Ack { send_data } => (None, Some(send_data)),
                 }
             }
-            pub fn new(recv_data: Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, send_data: Option<SendData>) -> Option<Self> {
+            pub fn new(recv_data: Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, RecvData)>, send_data: Option<SendData>) -> Option<Self> {
                 match (recv_data, send_data) {
                     (Some((reply_guard, recv_data)), None) => Some(Self::Payload { reply_guard, recv_data }),
                     (Some((reply_guard, recv_data)), Some(send_data)) => Some(Self::Reply { reply_guard, recv_data, send_data }),
@@ -141,15 +144,6 @@ macro_rules! impl_recvok {
                 }
             }
         }
-        impl<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize> $recv<'a, TL, P, SendData, RecvData, CAP> {
-            pub fn into(self) -> $recv<'a, TL, RecvData, SendData, RecvData, CAP> {
-                match self {
-                    Self::Payload { reply_guard, recv_data } => $recv::Payload { reply_guard, recv_data: recv_data.into() },
-                    Self::Reply { reply_guard, recv_data, send_data } => $recv::Reply { reply_guard, recv_data: recv_data.into(), send_data },
-                    Self::Ack { send_data } => $recv::Ack { send_data },
-                }
-            }
-        }
     };
 }
 impl_recvok!(RecvOk, &'a mut SeqEx<SendData, RecvData, CAP>);
@@ -157,7 +151,7 @@ pub(crate) use impl_recvok;
 
 impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
     /// Can mutate `next_service_timestamp`.
-    pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, seq_cst: bool, packet_data: SendData) -> Result<(), SendData> {
+    pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> {
         match self.try_send_direct(app.time(), seq_cst, packet_data) {
             Ok(p) => {
                 app.send(p);
@@ -172,7 +166,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         mut app: impl TransportLayer<SendData>,
         seq_cst: bool,
         packet_data: F,
-    ) -> Result<(), F> {
+    ) -> Result<(), (TryError, F)> {
         match self.try_send_direct_with(app.time(), seq_cst, packet_data) {
             Ok(p) => {
                 app.send(p);
@@ -186,17 +180,17 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         &mut self,
         mut app: impl TransportLayer<SendData>,
         packet: Packet<P>,
-    ) -> Result<RecvOkRaw<SendData, P>, RecvError> {
+    ) -> Result<(RecvOkRaw<SendData, P>, bool), RecvError> {
         match self.receive_raw_and_direct(packet) {
             Ok(a) => Ok(a),
-            Err(DirectRecvError::DroppedDuplicateResendAck(reply_no)) => {
+            Err(TryRecvError::DroppedDuplicateResendAck(reply_no)) => {
                 app.send(Packet::Ack(reply_no));
                 Err(RecvError::DroppedDuplicate)
             }
-            Err(DirectRecvError::DroppedTooEarly) => Err(RecvError::DroppedTooEarly),
-            Err(DirectRecvError::DroppedDuplicate) => Err(RecvError::DroppedDuplicate),
-            Err(DirectRecvError::WaitingForRecv) => Err(RecvError::WaitingForRecv),
-            Err(DirectRecvError::WaitingForReply) => Err(RecvError::WaitingForReply),
+            Err(TryRecvError::DroppedTooEarly) => Err(RecvError::DroppedTooEarly),
+            Err(TryRecvError::DroppedDuplicate) => Err(RecvError::DroppedDuplicate),
+            Err(TryRecvError::WaitingForRecv) => Err(RecvError::WaitingForRecv),
+            Err(TryRecvError::WaitingForReply) => Err(RecvError::WaitingForReply),
         }
     }
     /// Can mutate `next_service_timestamp`.
@@ -229,14 +223,15 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
         }
         self.resend_interval.min(self.next_service_timestamp - current_time)
     }
-    pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
+    pub fn receive<TL: TransportLayer<SendData>>(
         &mut self,
         app: TL,
-        packet: Packet<P>,
-    ) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, RecvError> {
-        self.receive_raw(app.clone(), packet).map(|r| RecvOk::from_raw(self, app, r))
+        packet: Packet<RecvData>,
+    ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), RecvError> {
+        self.receive_raw(app.clone(), packet)
+            .map(|(r, do_pump)| (RecvOk::from_raw(self, app, r), do_pump))
     }
-    pub fn try_pump<TL: TransportLayer<SendData>>(&mut self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, PumpError> {
-        self.try_pump_raw().map(|r| RecvOk::from_raw(self, app, r))
+    pub fn try_pump<TL: TransportLayer<SendData>>(&mut self, app: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> {
+        self.try_pump_raw().map(|(r, do_pump)| (RecvOk::from_raw(self, app, r), do_pump))
     }
 }

src/sync.rs

@@ -1,19 +1,26 @@
 use std::{
     sync::{
         mpsc::{channel, Receiver, Sender},
-        Condvar, Mutex,
+        Condvar, Mutex, MutexGuard,
     },
     time::Instant,
 };
 
 use crate::{
-    Packet, PumpError, RecvError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP,
+    Packet, RecvError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, TryError, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP,
 };
 
 pub struct SeqExSync<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
-    seq_ex: Mutex<(SeqEx<SendData, RecvData, CAP>, usize, bool)>,
-    send_block: Condvar,
-    reply_block: Condvar,
+    inner: Mutex<SeqExInner<SendData, RecvData, CAP>>,
+    wait_on_recv: Condvar,
+    wait_on_reply_sender: Condvar,
+    wait_on_reply_receiver: Condvar,
+}
+struct SeqExInner<SendData, RecvData, const CAP: usize> {
+    seq: SeqEx<SendData, RecvData, CAP>,
+    recv_waiters: usize,
+    reply_sender_waiters: bool,
+    reply_receiver_waiters: bool,
 }
 
 pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
@@ -23,29 +30,39 @@ pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, cons
     is_holding_lock: bool,
 }
 impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
-    fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
-        let app = self.app.take().unwrap();
-        let mut seq = self.seq.seq_ex.lock().unwrap();
-        let seq_no = seq.0.seq_no();
-        seq.0
-            .reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no));
-        if seq.2 {
-            seq.2 = false;
-            drop(seq);
-            self.seq.reply_block.notify_all();
+    pub fn ack(&mut self) {
+        if let Some(app) = self.app.take() {
+            let mut inner = self.seq.inner.lock().unwrap();
+            inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock);
         }
     }
+    /// # Panic
+    /// This function will panic if `ack` has been called.
+    pub fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
+        let app = self.app.take().expect("Cannot reply after an ack has been sent");
+        let mut inner = self.seq.inner.lock().unwrap();
+        let seq_no = inner.seq.seq_no();
+        inner
+            .seq
+            .reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no));
+        self.seq.notify_reply(inner);
+        core::mem::forget(self);
+    }
     /// If you need to reply more than once, say to fragment a large file, then include in your
     /// first reply some identifier, and then `send` all fragments with the same included identifier.
     /// The identifier will tell the remote peer which packets contain fragments of the file,
     /// and since each fragment will be received in order it will be trivial for them to reconstruct
     /// the original file.
+    /// # Panic
+    /// This function will panic if `ack` has been called.
     pub fn reply(self, seq_cst: bool, packet_data: SendData) {
         self.reply_with(seq_cst, |_, _| packet_data)
     }
 
-    pub fn to_components(mut self) -> (TL, SeqNo, bool) {
-        (self.app.take().unwrap(), self.reply_no, self.is_holding_lock)
+    pub fn to_components(self) -> (SeqNo, bool) {
+        let ret = (self.reply_no, self.is_holding_lock);
+        core::mem::forget(self);
+        ret
     }
     pub unsafe fn from_components(seq: &'a SeqExSync<SendData, RecvData, CAP>, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self {
         ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock }
@@ -53,15 +70,11 @@ 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();
         if let Some(app) = self.app.take() {
-            let mut seq = self.seq.seq_ex.lock().unwrap();
-            seq.0.ack_raw(app, self.reply_no, self.is_holding_lock);
-            if seq.2 {
-                seq.2 = false;
-                drop(seq);
-                self.seq.reply_block.notify_all();
-            }
+            inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock);
         }
+        self.seq.notify_reply(inner);
     }
 }
 impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
@@ -73,14 +86,14 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> std
     }
 }
 
-pub enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
+pub enum RecvOk<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
     Payload {
         reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
-        recv_data: P,
+        recv_data: RecvData,
     },
     Reply {
         reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
-        recv_data: P,
+        recv_data: RecvData,
         send_data: SendData,
     },
     Ack {
@@ -89,138 +102,179 @@ pub enum RecvOk<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const C
 }
 crate::impl_recvok!(RecvOk, &'a SeqExSync<SendData, RecvData, CAP>);
 
-pub struct RecvIter<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
+pub struct RecvIter<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
     seq: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
     app: TL,
-    first: Option<RecvOk<'a, TL, P, SendData, RecvData, CAP>>,
+    first: Option<RecvOk<'a, TL, SendData, RecvData, CAP>>,
     blocking: bool,
 }
 
 impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
     pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
         Self {
-            seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0, false)),
-            send_block: Condvar::default(),
-            reply_block: Condvar::default(),
+            inner: Mutex::new(SeqExInner {
+                seq: SeqEx::new(retry_interval, initial_seq_no),
+                recv_waiters: 0,
+                reply_sender_waiters: false,
+                reply_receiver_waiters: false,
+            }),
+            wait_on_recv: Condvar::default(),
+            wait_on_reply_receiver: Condvar::default(),
+            wait_on_reply_sender: 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 {
+            inner.reply_sender_waiters = false;
+            drop(inner);
+            self.wait_on_reply_sender.notify_all();
+        }
+    }
+    fn notify_recv(&self, mut inner: MutexGuard<'_, SeqExInner<SendData, RecvData, CAP>>) {
+        if inner.recv_waiters > 0 {
+            inner.recv_waiters -= 1;
+            drop(inner);
+            self.wait_on_recv.notify_one();
         }
     }
 
-    pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
+    pub fn try_receive<TL: TransportLayer<SendData>>(
         &self,
         app: TL,
-        packet: Packet<P>,
-    ) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, RecvError> {
-        let mut seq = self.seq_ex.lock().unwrap();
-        match seq.0.receive_raw(app.clone(), packet) {
-            Ok(r) => {
-                if seq.1 > 0 {
-                    seq.1 -= 1;
-                    drop(seq);
-                    self.send_block.notify_one();
-                }
-                Ok(RecvOk::from_raw(self, app, r))
+        packet: Packet<RecvData>,
+    ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), RecvError> {
+        let mut inner = self.inner.lock().unwrap();
+        match inner.seq.receive_raw(app.clone(), packet) {
+            Ok((r, do_pump)) => {
+                self.notify_recv(inner);
+                Ok((RecvOk::from_raw(self, app, r), do_pump))
             }
             Err(e) => Err(e),
         }
     }
-    pub fn try_pump<TL: TransportLayer<SendData>>(&self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, PumpError> {
-        let mut seq = self.seq_ex.lock().unwrap();
-        // Enforce that only one thread may pump at a time.
-        if seq.2 {
-            return Err(PumpError::WaitingForReply);
+    pub fn receive<TL: TransportLayer<SendData>>(
+        &self,
+        app: TL,
+        packet: Packet<RecvData>,
+    ) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> {
+        let result = self.try_receive(app.clone(), packet);
+        if let Err(RecvError::WaitingForReply) = result {
+            self.pump(app)
+        } else {
+            result.ok()
         }
-        match seq.0.try_pump_raw() {
-            Ok(r) => {
-                if seq.1 > 0 {
-                    seq.1 -= 1;
-                    drop(seq);
-                    self.send_block.notify_one();
-                }
-                Ok(RecvOk::from_raw(self, app, r))
+    }
+    pub fn try_pump<TL: TransportLayer<SendData>>(&self, app: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> {
+        let mut inner = self.inner.lock().unwrap();
+        match inner.seq.try_pump_raw() {
+            Ok((r, do_pump)) => {
+                self.notify_recv(inner);
+                Ok((RecvOk::from_raw(self, app, r), do_pump))
             }
             Err(e) => Err(e),
         }
     }
-    pub fn pump<TL: TransportLayer<SendData>>(&self, app: TL) -> Option<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>> {
-        let mut seq = self.seq_ex.lock().unwrap();
-        // Enforce that only one thread may pump at a time.
-        if seq.2 {
+    pub fn pump<TL: TransportLayer<SendData>>(&self, app: 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 {
-            match seq.0.try_pump_raw() {
-                Ok(r) => {
-                    if seq.1 > 0 {
-                        seq.1 -= 1;
-                        drop(seq);
-                        self.send_block.notify_one();
-                    }
-                    return Some(RecvOk::from_raw(self, app, r));
+            match inner.seq.try_pump_raw() {
+                Ok((r, do_pump)) => {
+                    self.notify_recv(inner);
+                    return Some((RecvOk::from_raw(self, app, r), do_pump));
                 }
-                Err(PumpError::WaitingForRecv) => return None,
-                Err(PumpError::WaitingForReply) => {
-                    seq.2 = true;
-                    seq = self.reply_block.wait(seq).unwrap();
+                Err(TryError::WaitingForRecv) => return None,
+                Err(TryError::WaitingForReply) => {
+                    inner.reply_receiver_waiters = true;
+                    inner = self.wait_on_reply_receiver.wait(inner).unwrap();
                 }
             }
         }
     }
 
-    fn receive_all_inner<TL: TransportLayer<SendData>, P: Into<RecvData>>(
-        &self,
-        app: TL,
-        blocking: bool,
-        packet: Packet<P>,
-    ) -> RecvIter<'_, TL, P, SendData, RecvData, CAP> {
-        match self.receive(app.clone(), packet) {
-            Ok(r) => RecvIter { seq: Some(self), app, first: Some(r), blocking },
-            Err(RecvError::WaitingForReply) if blocking => RecvIter { seq: Some(self), app, first: None, blocking },
-            Err(_) => RecvIter { seq: None, app, first: None, blocking },
+    pub fn receive_all<TL: TransportLayer<SendData>>(&self, app: TL, packet: Packet<RecvData>) -> RecvIter<'_, TL, SendData, RecvData, CAP> {
+        let ret = self.receive(app.clone(), packet);
+        if let Some((first, do_pump)) = ret {
+            RecvIter {
+                seq: do_pump.then_some(self),
+                app,
+                first: Some(first),
+                blocking: true,
+            }
+        } else {
+            RecvIter { seq: None, app, first: None, blocking: true }
         }
     }
-    pub fn receive_all<TL: TransportLayer<SendData>, P: Into<RecvData>>(
-        &self,
-        app: TL,
-        packet: Packet<P>,
-    ) -> RecvIter<'_, TL, P, SendData, RecvData, CAP> {
-        self.receive_all_inner(app, true, packet)
-    }
-    pub fn try_receive_all<TL: TransportLayer<SendData>, P: Into<RecvData>>(
-        &self,
-        app: TL,
-        packet: Packet<P>,
-    ) -> RecvIter<'_, TL, P, SendData, RecvData, CAP> {
-        self.receive_all_inner(app, false, packet)
+    pub fn try_receive_all<TL: TransportLayer<SendData>>(&self, app: TL, packet: Packet<RecvData>) -> RecvIter<'_, TL, SendData, RecvData, CAP> {
+        let ret = self.try_receive(app.clone(), packet);
+        if let Ok((first, do_pump)) = ret {
+            RecvIter {
+                seq: do_pump.then_some(self),
+                app,
+                first: Some(first),
+                blocking: false,
+            }
+        } else {
+            RecvIter { seq: None, app, first: None, blocking: false }
+        }
     }
 
-    pub fn try_send_with<TL: TransportLayer<SendData>, F: FnOnce(SeqNo) -> SendData>(&self, app: TL, seq_cst: bool, packet_data: F) -> Result<(), F> {
-        let mut seq = self.seq_ex.lock().unwrap();
-        seq.0.try_send_with(app, seq_cst, packet_data)
+    pub fn try_send_with<TL: TransportLayer<SendData>, F: FnOnce(SeqNo) -> SendData>(
+        &self,
+        app: TL,
+        seq_cst: bool,
+        packet_data: F,
+    ) -> Result<(), (TryError, F)> {
+        let mut inner = self.inner.lock().unwrap();
+        inner.seq.try_send_with(app, seq_cst, packet_data)
     }
-    pub fn try_send<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, packet_data: SendData) -> Result<(), SendData> {
-        let mut seq = self.seq_ex.lock().unwrap();
-        seq.0.try_send(app, seq_cst, packet_data)
+    pub fn try_send<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> {
+        let mut inner = self.inner.lock().unwrap();
+        inner.seq.try_send(app, seq_cst, packet_data)
     }
 
     pub fn send_with<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) {
-        let mut seq = self.seq_ex.lock().unwrap();
-        while let Err(p) = seq.0.try_send_with(app.clone(), seq_cst, packet_data) {
+        let mut inner = self.inner.lock().unwrap();
+        while let Err((e, p)) = inner.seq.try_send_with(app.clone(), seq_cst, packet_data) {
             packet_data = p;
-            seq.1 += 1;
-            seq = self.send_block.wait(seq).unwrap();
+            match e {
+                TryError::WaitingForRecv => {
+                    inner.recv_waiters += 1;
+                    inner = self.wait_on_recv.wait(inner).unwrap();
+                }
+                TryError::WaitingForReply => {
+                    inner.reply_sender_waiters = true;
+                    inner = self.wait_on_reply_sender.wait(inner).unwrap();
+                }
+            }
         }
     }
     pub fn send<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, mut packet_data: SendData) {
-        let mut seq = self.seq_ex.lock().unwrap();
-        while let Err(p) = seq.0.try_send(app.clone(), seq_cst, packet_data) {
+        let mut inner = self.inner.lock().unwrap();
+        while let Err((e, p)) = inner.seq.try_send(app.clone(), seq_cst, packet_data) {
             packet_data = p;
-            seq.1 += 1;
-            seq = self.send_block.wait(seq).unwrap();
+            match e {
+                TryError::WaitingForRecv => {
+                    inner.recv_waiters += 1;
+                    inner = self.wait_on_recv.wait(inner).unwrap();
+                }
+                TryError::WaitingForReply => {
+                    inner.reply_sender_waiters = true;
+                    inner = self.wait_on_reply_sender.wait(inner).unwrap();
+                }
+            }
         }
     }
 
     pub fn service<TL: TransportLayer<SendData>>(&self, app: TL) -> i64 {
-        self.seq_ex.lock().unwrap().0.service(app)
+        self.inner.lock().unwrap().seq.service(app)
     }
 }
 impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvData, CAP> {
@@ -229,23 +283,24 @@ impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvD
     }
 }
 
-impl<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize> RecvIter<'a, TL, P, SendData, RecvData, CAP> {
-    pub fn take_first(&mut self) -> Option<RecvOk<'a, TL, P, SendData, RecvData, CAP>> {
-        self.first.take()
-    }
-}
-impl<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize> Iterator
-    for RecvIter<'a, TL, P, SendData, RecvData, CAP>
-{
-    type Item = RecvOk<'a, TL, RecvData, SendData, RecvData, CAP>;
+impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Iterator for RecvIter<'a, TL, SendData, RecvData, CAP> {
+    type Item = RecvOk<'a, TL, SendData, RecvData, CAP>;
     fn next(&mut self) -> Option<Self::Item> {
-        if let Some(g) = self.first.take() {
-            Some(g.into())
+        if let Some(item) = self.first.take() {
+            Some(item)
         } else if let Some(origin) = self.seq {
-            if self.blocking {
+            let ret = if self.blocking {
                 origin.pump(self.app.clone())
             } else {
                 origin.try_pump(self.app.clone()).ok()
+            };
+            if let Some((item, do_pump)) = ret {
+                if !do_pump {
+                    self.seq = None;
+                }
+                Some(item)
+            } else {
+                None
             }
         } else {
             None