diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 1992e78..8ff0763 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -65,11 +65,13 @@ pub struct SeqEx { /// when `reply_raw` or `ack_raw` are called with it, they are guaranteed not to fail. /// To accomplish this we must track all issued reply numbers. concurrent_replies_total: usize, + is_locked: bool, } struct RecvEntry { seq_no: SeqNo, reply_no: Option, + locked: bool, data: RecvData, } @@ -80,51 +82,99 @@ struct SendEntry { data: SendData, } -/// The error type for when a packet has been received, but for whatever reason could not be -/// immediately processed. -#[derive(Debug, Clone)] -pub enum DirectError { +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DirectError { /// The packet is out-of-sequence. It was either received too soon or too late and so it would be /// invalid to process it right now. No action needs to be taken by the caller. OutOfSequence, /// The Send Window is currently full. The received packet cannot be processed right now because /// it could cause the send window to overflow. No action needs to be taken by the caller. - WindowIsFull, + WindowIsFull(Packet), + WindowIsLocked(Packet), ResendAck(SeqNo), } -/// The error type for when a packet has been received, but for whatever reason could not be -/// immediately processed. -#[derive(Debug, Clone)] -pub enum Error { +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PumpError { /// The packet is out-of-sequence. It was either received too soon or too late and so it would be /// invalid to process it right now. No action needs to be taken by the caller. OutOfSequence, /// The Send Window is currently full. The received packet cannot be processed right now because /// it could cause the send window to overflow. No action needs to be taken by the caller. WindowIsFull, + WindowIsLocked, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] pub enum Packet { Payload(SeqNo, RecvData), + LockPayload(SeqNo, RecvData), Reply(SeqNo, SeqNo, RecvData), + LockReply(SeqNo, SeqNo, RecvData), Ack(SeqNo), } +use Packet::*; impl Packet { + pub fn new_with_data(seq_no: SeqNo, reply_no: Option, is_locking: bool, data: RecvData) -> Self { + Self::new(Some(seq_no), reply_no, is_locking, Some(data)).unwrap() + } + pub fn new(seq_no: Option, reply_no: Option, is_locking: bool, data: Option) -> Option { + match (seq_no, reply_no, is_locking, data) { + (Some(s), None, false, Some(d)) => Some(Payload(s, d)), + (Some(s), None, true, Some(d)) => Some(LockPayload(s, d)), + (Some(s), Some(r), false, Some(d)) => Some(Reply(s, r, d)), + (Some(s), Some(r), true, Some(d)) => Some(LockReply(s, r, d)), + (None, Some(r), false, None) => Some(Ack(r)), + _ => None, + } + } pub fn as_ref(&self) -> Packet<&RecvData> { match self { - Packet::Payload(seq_no, data) => Packet::Payload(*seq_no, data), - Packet::Reply(seq_no, reply_no, data) => Packet::Reply(*seq_no, *reply_no, data), - Packet::Ack(reply_no) => Packet::Ack(*reply_no), + Payload(seq_no, data) => Payload(*seq_no, data), + LockPayload(seq_no, data) => LockPayload(*seq_no, data), + Reply(seq_no, reply_no, data) => Reply(*seq_no, *reply_no, data), + LockReply(seq_no, reply_no, data) => LockReply(*seq_no, *reply_no, data), + Ack(reply_no) => Ack(*reply_no), } } pub fn map(self, f: impl FnOnce(RecvData) -> SendData) -> Packet { match self { - Packet::Payload(seq_no, data) => Packet::Payload(seq_no, f(data)), - Packet::Reply(seq_no, reply_no, data) => Packet::Reply(seq_no, reply_no, f(data)), - Packet::Ack(reply_no) => Packet::Ack(reply_no), + Payload(seq_no, data) => Payload(seq_no, f(data)), + LockPayload(seq_no, data) => LockPayload(seq_no, f(data)), + Reply(seq_no, reply_no, data) => Reply(seq_no, reply_no, f(data)), + LockReply(seq_no, reply_no, data) => LockReply(seq_no, reply_no, f(data)), + Ack(reply_no) => Ack(reply_no), + } + } + pub fn payload(self) -> Option { + match self { + Payload(_, data) | LockPayload(_, data) | Reply(_, _, data) | LockReply(_, _, data) => Some(data), + Ack(_) => None, + } + } + pub fn is_locking(&self) -> bool { + matches!(self, LockPayload(..) | LockReply(..)) + } + pub fn set_locking(&mut self, locking: bool) { + let mut tmp = Ack(0); + core::mem::swap(&mut tmp, self); + match tmp { + Payload(seq_no, data) | LockPayload(seq_no, data) => { + *self = if locking { + LockPayload(seq_no, data) + } else { + Payload(seq_no, data) + } + } + Reply(seq_no, reply_no, data) | LockReply(seq_no, reply_no, data) => { + *self = if locking { + LockReply(seq_no, reply_no, data) + } else { + Reply(seq_no, reply_no, data) + } + } + Ack(reply_no) => *self = Ack(reply_no), } } } @@ -137,10 +187,12 @@ impl Packet<&RecvData> { pub enum RecvOkRaw { Payload { reply_no: SeqNo, + locked: bool, recv_data: RecvData, }, Reply { reply_no: SeqNo, + locked: bool, recv_data: RecvData, send_data: SendData, }, @@ -190,6 +242,7 @@ impl SeqEx { send_window: core::array::from_fn(|_| None), concurrent_replies: core::array::from_fn(|_| 0), concurrent_replies_total: 0, + is_locked: false, } } fn send_window_slot_mut(&mut self, seq_no: SeqNo) -> &mut Option> { @@ -264,10 +317,9 @@ impl SeqEx { /// `current_time` should be a timestamp of the current time, using whatever units of time the /// user would like. However this choice of units must be consistent with the units of the /// `retry_interval`. `current_time` does not have to be monotonically increasing. - #[must_use = "The queue might be full causing the packet to not be sent"] - pub fn try_send_direct(&mut self, packet_data: SendData, current_time: i64) -> Result, SendData> { + fn try_send_direct_inner(&mut self, current_time: i64) -> Option<(&mut Option>, SeqNo, i64)> { if self.is_full() { - return Err(packet_data); + return None; } let seq_no = self.next_send_seq_no; self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1); @@ -278,16 +330,36 @@ impl SeqEx { } let slot = self.send_window_slot_mut(seq_no); debug_assert!(slot.is_none()); - let entry = slot.insert(SendEntry { seq_no, reply_no: None, next_resend_time, data: packet_data }); - - Ok(Packet::Payload(entry.seq_no, &entry.data)) + Some((slot, seq_no, next_resend_time)) + } + pub fn try_send_direct(&mut self, packet_data: SendData, current_time: i64) -> Result, SendData> { + if let Some((slot, seq_no, next_resend_time)) = self.try_send_direct_inner(current_time) { + let entry = slot.insert(SendEntry { seq_no, reply_no: None, next_resend_time, data: packet_data }); + Ok(Packet::Payload(entry.seq_no, &entry.data)) + } else { + Err(packet_data) + } + } + pub fn try_send_direct_with(&mut self, packet_data: impl FnOnce(SeqNo) -> SendData, current_time: i64) -> Result, ()> { + if let Some((slot, seq_no, next_resend_time)) = self.try_send_direct_inner(current_time) { + let entry = slot.insert(SendEntry { + seq_no, + reply_no: None, + next_resend_time, + data: packet_data(seq_no), + }); + Ok(Packet::Payload(entry.seq_no, &entry.data)) + } else { + Err(()) + } } /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn receive_raw_and_direct>(&mut self, packet: Packet

) -> Result, DirectError> { + pub fn receive_raw_and_direct>(&mut self, packet: Packet

) -> Result, DirectError

> { + let locked = packet.is_locking(); let (seq_no, reply_no, recv_data) = match packet { - Packet::Payload(seq_no, recv_data) => (seq_no, None, recv_data), - Packet::Reply(seq_no, reply_no, recv_data) => (seq_no, Some(reply_no), recv_data), + Packet::Payload(seq_no, recv_data) | Packet::LockPayload(seq_no, recv_data) => (seq_no, None, recv_data), + Packet::Reply(seq_no, reply_no, recv_data) | Packet::LockReply(seq_no, reply_no, recv_data) => (seq_no, Some(reply_no), recv_data), Packet::Ack(reply_no) => { return self .take_send(reply_no) @@ -332,64 +404,87 @@ impl SeqEx { // We can only process this packet if processing it would make space in the send window. let is_full = self.is_full_inner(false); + // Check whether or not we've already received this packet let i = seq_no as usize % self.recv_window.len(); - if let Some(pre) = self.recv_window[i].as_mut() { + let is_in_window = if let Some(pre) = self.recv_window[i].as_mut() { if seq_no == pre.seq_no { - if is_next && !is_full { - self.recv_window[i] = None; - } else { - return if is_full { - Err(DirectError::WindowIsFull) - } else { - Err(DirectError::OutOfSequence) - }; - } + true } else { // This is currently unreachable due to the range check. // `self.pre_recv_seq_no < seq_no <= self.pre_recv_seq_no + self.recv_window.len()`. return Err(DirectError::OutOfSequence); } - } - if is_next && !is_full { + } else { + false + }; + + if !is_next { + if !is_in_window { + self.recv_window[i] = Some(RecvEntry { seq_no, reply_no, locked, data: recv_data.into() }) + } + Err(DirectError::OutOfSequence) + } else if is_full { + Err(DirectError::WindowIsFull(Packet::new_with_data(seq_no, reply_no, locked, recv_data))) + } else if locked && self.is_locked { + Err(DirectError::WindowIsLocked(Packet::new_with_data(seq_no, reply_no, locked, recv_data))) + } else { + if is_in_window { + self.recv_window[i] = None; + } self.pre_recv_seq_no = seq_no; self.concurrent_replies[self.concurrent_replies_total] = seq_no; self.concurrent_replies_total += 1; - return 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 } - } else { - RecvOkRaw::Payload { reply_no: seq_no, recv_data } - }); - } else { - self.recv_window[i] = Some(RecvEntry { seq_no, reply_no, data: recv_data.into() }); - if is_full { - Err(DirectError::WindowIsFull) - } else { - Err(DirectError::OutOfSequence) + if locked { + 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, locked } + } else { + RecvOkRaw::Payload { reply_no: seq_no, recv_data, locked } + }) } } /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn pump_raw(&mut self) -> Result, Error> { + pub fn pump_raw(&mut self) -> Result, PumpError> { let next_seq_no = self.pre_recv_seq_no.wrapping_add(1); let i = next_seq_no as usize % self.recv_window.len(); - if self.recv_window[i].as_ref().map_or(false, |pre| pre.seq_no == next_seq_no) { - if self.is_full_inner(false) { - return Err(Error::WindowIsFull); - } + if let Some(entry) = &self.recv_window[i].as_ref() { + if entry.seq_no == next_seq_no { + if self.is_full_inner(false) { + return Err(PumpError::WindowIsFull); + } - let entry = self.recv_window[i].take().unwrap(); - self.pre_recv_seq_no = next_seq_no; - self.concurrent_replies[self.concurrent_replies_total] = entry.seq_no; - self.concurrent_replies_total += 1; - return Ok(if let Some(send_data) = entry.reply_no.and_then(|r| self.take_send(r)) { - RecvOkRaw::Reply { reply_no: entry.seq_no, recv_data: entry.data, send_data } - } else { - RecvOkRaw::Payload { reply_no: entry.seq_no, recv_data: entry.data } - }); - } else { - Err(Error::OutOfSequence) + if !entry.locked || !self.is_locked { + let entry = self.recv_window[i].take().unwrap(); + self.pre_recv_seq_no = next_seq_no; + self.concurrent_replies[self.concurrent_replies_total] = entry.seq_no; + self.concurrent_replies_total += 1; + if entry.locked { + debug_assert!(!self.is_locked); + self.is_locked = true; + } + return Ok(if let Some(send_data) = entry.reply_no.and_then(|r| self.take_send(r)) { + RecvOkRaw::Reply { + reply_no: entry.seq_no, + locked: entry.locked, + recv_data: entry.data, + send_data, + } + } else { + RecvOkRaw::Payload { + reply_no: entry.seq_no, + locked: entry.locked, + recv_data: entry.data, + } + }); + } else { + return Err(PumpError::WindowIsLocked); + } + } } + Err(PumpError::OutOfSequence) } /// 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. @@ -400,7 +495,7 @@ impl SeqEx { /// and since each fragment will be received in order it will be trivial for them to reconstruct /// the original file. #[must_use] - pub fn reply_raw_and_direct(&mut self, reply_no: SeqNo, packet_data: SendData, current_time: i64) -> Option> { + pub fn reply_raw_and_direct(&mut self, reply_no: SeqNo, unlock: bool, packet_data: SendData, current_time: i64) -> Option> { if self.remove_reservation(reply_no) { let seq_no = self.next_send_seq_no; self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1); @@ -409,6 +504,10 @@ impl SeqEx { if self.next_service_timestamp > next_resend_time { self.next_service_timestamp = next_resend_time; } + if unlock { + debug_assert!(self.is_locked, "The window must be locked to attempt to unlock: double unlock detected."); + self.is_locked = false; + } let slot = self.send_window_slot_mut(seq_no); debug_assert!(slot.is_none()); let entry = slot.insert(SendEntry { @@ -423,8 +522,13 @@ impl SeqEx { None } } - pub fn ack_raw_and_direct(&mut self, reply_no: SeqNo) -> Option> { + #[must_use] + pub fn ack_raw_and_direct(&mut self, reply_no: SeqNo, unlock: bool) -> Option> { if self.remove_reservation(reply_no) { + if unlock { + debug_assert!(self.is_locked, "The window must be locked to attempt to unlock: double unlock detected."); + self.is_locked = false; + } Some(Packet::Ack(reply_no)) } else { None diff --git a/src/single_thread.rs b/src/single_thread.rs index d9924ac..f4c5eb4 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,9 +1,10 @@ -use crate::{DirectError, Error, Packet, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; +use crate::{DirectError, Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { seq: &'a mut SeqEx, app: Option, reply_no: SeqNo, + locked: bool, } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { /// If you need to reply more than once, say to fragment a large file, then include in your @@ -11,17 +12,22 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// 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. - pub fn reply(mut self, packet_data: SendData) { + pub fn reply(self, packet_data: SendData) { + self.reply_with(|_, _| packet_data) + } + pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { let mut app = None; core::mem::swap(&mut app, &mut self.app); - self.seq.reply_raw(app.unwrap(), self.reply_no, packet_data); + let seq_no = self.seq.seq_no(); + self.seq + .reply_raw(app.unwrap(), self.reply_no, self.locked, packet_data(seq_no, self.reply_no)); core::mem::forget(self); } } impl<'a, TL: TransportLayer, 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) { + if let Some(p) = self.seq.ack_raw_and_direct(self.reply_no, self.locked) { app.send(p) } } @@ -33,7 +39,17 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std } } -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Error { + /// The packet is out-of-sequence. It was either received too soon or too late and so it would be + /// invalid to process it right now. No action needs to be taken by the caller. + OutOfSequence, + /// The Send Window is currently full. The received packet cannot be processed right now because + /// it could cause the send window to overflow. No action needs to be taken by the caller. + WindowIsFull(Packet), + WindowIsLocked(Packet), +} + pub enum RecvOk<'a, TL: TransportLayer, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { Payload { reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, @@ -48,46 +64,73 @@ pub enum RecvOk<'a, TL: TransportLayer, P, SendData, RecvData, const C send_data: SendData, }, } -impl<'a, TL: TransportLayer, P, SendData, RecvData, const CAP: usize> RecvOk<'a, TL, P, SendData, RecvData, CAP> { - pub fn from_raw(seq: &'a mut SeqEx, app: TL, value: RecvOkRaw) -> Self { - match value { - RecvOkRaw::Payload { reply_no, recv_data } => RecvOk::Payload { - reply_guard: ReplyGuard { seq, app: Some(app), reply_no }, - recv_data, - }, - RecvOkRaw::Reply { reply_no, recv_data, send_data } => RecvOk::Reply { - reply_guard: ReplyGuard { seq, app: Some(app), reply_no }, - recv_data, - send_data, - }, - RecvOkRaw::Ack { send_data } => RecvOk::Ack { send_data }, +macro_rules! impl_recvok { + ($recv:tt, $seq_ex:ty) => { + #[cfg(feature = "std")] + impl<'a, TL: TransportLayer, P: std::fmt::Debug, SendData: std::fmt::Debug, RecvData, const CAP: usize> std::fmt::Debug + for $recv<'a, TL, P, SendData, RecvData, CAP> + { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Payload { reply_guard, recv_data } => f + .debug_struct("Payload") + .field("reply_guard", reply_guard) + .field("recv_data", recv_data) + .finish(), + Self::Reply { reply_guard, recv_data, send_data } => f + .debug_struct("Reply") + .field("reply_guard", reply_guard) + .field("recv_data", recv_data) + .field("send_data", send_data) + .finish(), + Self::Ack { send_data } => f.debug_struct("Ack").field("send_data", send_data).finish(), + } + } } - } - pub fn consume(self) -> (Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, Option) { - match self { - RecvOk::Payload { reply_guard, recv_data } => (Some((reply_guard, recv_data)), None), - RecvOk::Reply { reply_guard, recv_data, send_data } => (Some((reply_guard, recv_data)), Some(send_data)), - RecvOk::Ack { send_data } => (None, Some(send_data)), + impl<'a, TL: TransportLayer, P, SendData, RecvData, const CAP: usize> $recv<'a, TL, P, SendData, RecvData, CAP> { + pub fn from_raw(seq: $seq_ex, app: TL, value: RecvOkRaw) -> Self { + match value { + RecvOkRaw::Payload { reply_no, locked, recv_data } => Self::Payload { + reply_guard: ReplyGuard { seq, app: Some(app), reply_no, locked }, + recv_data, + }, + RecvOkRaw::Reply { reply_no, locked, recv_data, send_data } => Self::Reply { + reply_guard: ReplyGuard { seq, app: Some(app), reply_no, locked }, + recv_data, + send_data, + }, + RecvOkRaw::Ack { send_data } => Self::Ack { send_data }, + } + } + pub fn consume(self) -> (Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, Option) { + 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) -> Option { + 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 }), + (None, Some(send_data)) => Some(Self::Ack { send_data }), + (None, None) => None, + } + } } - } - pub fn new(recv_data: Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, send_data: Option) -> Option { - match (recv_data, send_data) { - (Some((reply_guard, recv_data)), None) => Some(RecvOk::Payload { reply_guard, recv_data }), - (Some((reply_guard, recv_data)), Some(send_data)) => Some(RecvOk::Reply { reply_guard, recv_data, send_data }), - (None, Some(send_data)) => Some(RecvOk::Ack { send_data }), - (None, None) => None, + impl<'a, TL: TransportLayer, P: Into, 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<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize> RecvOk<'a, TL, P, SendData, RecvData, CAP> { - pub fn into(self) -> RecvOk<'a, TL, RecvData, SendData, RecvData, CAP> { - match self { - RecvOk::Payload { reply_guard, recv_data } => RecvOk::Payload { reply_guard, recv_data: recv_data.into() }, - RecvOk::Reply { reply_guard, recv_data, send_data } => RecvOk::Reply { reply_guard, recv_data: recv_data.into(), send_data }, - RecvOk::Ack { send_data } => RecvOk::Ack { send_data }, - } - } + }; } +impl_recvok!(RecvOk, &'a mut SeqEx); +pub(crate) use impl_recvok; impl SeqEx { pub fn try_send(&mut self, mut app: impl TransportLayer, packet_data: SendData) -> Result<(), SendData> { @@ -99,12 +142,21 @@ impl SeqEx { Err(e) => Err(e), } } + pub fn try_send_with(&mut self, mut app: impl TransportLayer, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), ()> { + match self.try_send_direct_with(packet_data, app.time()) { + Ok(p) => { + app.send(p); + Ok(()) + } + Err(e) => Err(e), + } + } /// If this returns `Ok` then `try_send` might succeed on next call. pub fn receive_raw>( &mut self, mut app: impl TransportLayer, packet: Packet

, - ) -> Result, Error> { + ) -> Result, Error

> { match self.receive_raw_and_direct(packet) { Ok(a) => Ok(a), Err(DirectError::ResendAck(reply_no)) => { @@ -112,16 +164,17 @@ impl SeqEx { Err(Error::OutOfSequence) } Err(DirectError::OutOfSequence) => Err(Error::OutOfSequence), - Err(DirectError::WindowIsFull) => Err(Error::WindowIsFull), + Err(DirectError::WindowIsFull(p)) => Err(Error::WindowIsFull(p)), + Err(DirectError::WindowIsLocked(p)) => Err(Error::WindowIsLocked(p)), } } - pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, packet_data: SendData) { - if let Some(p) = self.reply_raw_and_direct(reply_no, packet_data, app.time()) { + pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, unlock: bool, packet_data: SendData) { + if let Some(p) = self.reply_raw_and_direct(reply_no, unlock, packet_data, app.time()) { app.send(p) } } - pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo) { - if let Some(p) = self.ack_raw_and_direct(reply_no) { + pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, unlock: bool) { + if let Some(p) = self.ack_raw_and_direct(reply_no, unlock) { app.send(p) } } @@ -137,10 +190,10 @@ impl SeqEx { &mut self, app: TL, packet: Packet

, - ) -> Result, Error> { + ) -> Result, Error

> { self.receive_raw(app.clone(), packet).map(|r| RecvOk::from_raw(self, app, r)) } - pub fn pump>(&mut self, app: TL) -> Result, Error> { + pub fn pump>(&mut self, app: TL) -> Result, PumpError> { self.pump_raw().map(|r| RecvOk::from_raw(self, app, r)) } } diff --git a/src/sync.rs b/src/sync.rs index 69ee58e..6f62973 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -7,17 +7,19 @@ use std::{ time::Instant, }; -use crate::{Error, Packet, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; +use crate::{Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; pub struct SeqExSync { seq_ex: Mutex<(SeqEx, usize)>, send_block: Condvar, + lock: Condvar, } pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { seq: &'a SeqExSync, - app: TL, + app: Option, reply_no: SeqNo, + locked: bool, } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { /// If you need to reply more than once, say to fragment a large file, then include in your @@ -26,21 +28,25 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// and since each fragment will be received in order it will be trivial for them to reconstruct /// the original file. pub fn reply(self, packet_data: SendData) { - let mut seq = self.seq.lock(); - seq.reply_raw(self.app.clone(), self.reply_no, packet_data); - core::mem::forget(self); + self.reply_with(|_, _| packet_data) } - pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + let mut app = None; + core::mem::swap(&mut app, &mut self.app); let mut seq = self.seq.lock(); let seq_no = seq.seq_no(); - seq.reply_raw(self.app.clone(), self.reply_no, packet_data(seq_no, self.reply_no)); + seq.reply_raw(app.unwrap(), self.reply_no, self.locked, packet_data(seq_no, self.reply_no)); core::mem::forget(self); } } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { - let mut seq = self.seq.lock(); - seq.ack_raw(self.app.clone(), self.reply_no); + if let Some(app) = self.app.as_mut() { + let mut seq = self.seq.lock(); + if let Some(p) = seq.ack_raw_and_direct(self.reply_no, self.locked) { + app.send(p) + } + } } } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> { @@ -49,6 +55,16 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std } } +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Error { + /// The packet is out-of-sequence. It was either received too soon or too late and so it would be + /// invalid to process it right now. No action needs to be taken by the caller. + OutOfSequence, + /// The Send Window is currently full. The received packet cannot be processed right now because + /// it could cause the send window to overflow. No action needs to be taken by the caller. + WindowIsFull, +} + pub enum RecvOk<'a, TL: TransportLayer, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { Payload { reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, @@ -63,66 +79,10 @@ pub enum RecvOk<'a, TL: TransportLayer, P, SendData, RecvData, const C send_data: SendData, }, } -impl<'a, TL: TransportLayer, P: std::fmt::Debug, SendData: std::fmt::Debug, RecvData, const CAP: usize> std::fmt::Debug - for RecvOk<'a, TL, P, SendData, RecvData, CAP> -{ - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Payload { reply_guard, recv_data } => f - .debug_struct("Payload") - .field("reply_guard", reply_guard) - .field("recv_data", recv_data) - .finish(), - Self::Reply { reply_guard, recv_data, send_data } => f - .debug_struct("Reply") - .field("reply_guard", reply_guard) - .field("recv_data", recv_data) - .field("send_data", send_data) - .finish(), - Self::Ack { send_data } => f.debug_struct("Ack").field("send_data", send_data).finish(), - } - } -} -impl<'a, TL: TransportLayer, P, SendData, RecvData, const CAP: usize> RecvOk<'a, TL, P, SendData, RecvData, CAP> { - pub fn from_raw(seq: &'a SeqExSync, app: TL, value: RecvOkRaw) -> Self { - match value { - RecvOkRaw::Payload { reply_no, recv_data } => RecvOk::Payload { reply_guard: ReplyGuard { seq, app, reply_no }, recv_data }, - RecvOkRaw::Reply { reply_no, recv_data, send_data } => RecvOk::Reply { - reply_guard: ReplyGuard { seq, app, reply_no }, - recv_data, - send_data, - }, - RecvOkRaw::Ack { send_data } => RecvOk::Ack { send_data }, - } - } - pub fn consume(self) -> (Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, Option) { - match self { - RecvOk::Payload { reply_guard, recv_data } => (Some((reply_guard, recv_data)), None), - RecvOk::Reply { reply_guard, recv_data, send_data } => (Some((reply_guard, recv_data)), Some(send_data)), - RecvOk::Ack { send_data } => (None, Some(send_data)), - } - } - pub fn new(recv_data: Option<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, P)>, send_data: Option) -> Option { - match (recv_data, send_data) { - (Some((reply_guard, recv_data)), None) => Some(RecvOk::Payload { reply_guard, recv_data }), - (Some((reply_guard, recv_data)), Some(send_data)) => Some(RecvOk::Reply { reply_guard, recv_data, send_data }), - (None, Some(send_data)) => Some(RecvOk::Ack { send_data }), - (None, None) => None, - } - } -} -impl<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize> RecvOk<'a, TL, P, SendData, RecvData, CAP> { - pub fn into(self) -> RecvOk<'a, TL, RecvData, SendData, RecvData, CAP> { - match self { - RecvOk::Payload { reply_guard, recv_data } => RecvOk::Payload { reply_guard, recv_data: recv_data.into() }, - RecvOk::Reply { reply_guard, recv_data, send_data } => RecvOk::Reply { reply_guard, recv_data: recv_data.into(), send_data }, - RecvOk::Ack { send_data } => RecvOk::Ack { send_data }, - } - } -} +crate::impl_recvok!(RecvOk, &'a SeqExSync); pub struct ReplyIter<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { - origin: Option<&'a SeqExSync>, + seq: Option<&'a SeqExSync>, app: TL, first: Option>, } @@ -146,28 +106,50 @@ impl SeqExSync { Self { seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)), send_block: Condvar::default(), + lock: Condvar::default(), } } pub fn receive, P: Into>( &self, app: TL, - packet: Packet

, + mut packet: Packet

, ) -> Result, Error> { let mut seq = self.seq_ex.lock().unwrap(); - let ret = seq.0.receive_raw(app.clone(), packet); - if seq.1 > 0 && ret.is_ok() { - self.send_block.notify_one(); + loop { + match seq.0.receive_raw(app.clone(), packet) { + Ok(r) => { + if seq.1 > 0 { + self.send_block.notify_one(); + } + return Ok(RecvOk::from_raw(self, app, r)); + } + Err(crate::Error::OutOfSequence) => return Err(Error::OutOfSequence), + Err(crate::Error::WindowIsFull(_)) => return Err(Error::WindowIsFull), + Err(crate::Error::WindowIsLocked(p)) => { + seq = self.lock.wait(seq).unwrap(); + packet = p; + } + } } - ret.map(|r| RecvOk::from_raw(self, app, r)) } pub fn pump>(&self, app: TL) -> Result, Error> { let mut seq = self.seq_ex.lock().unwrap(); - let ret = seq.0.pump_raw(); - if seq.1 > 0 && ret.is_ok() { - self.send_block.notify_one(); + loop { + match seq.0.pump_raw() { + Ok(r) => { + if seq.1 > 0 { + self.send_block.notify_one(); + } + return Ok(RecvOk::from_raw(self, app, r)); + } + Err(PumpError::OutOfSequence) => return Err(Error::OutOfSequence), + Err(PumpError::WindowIsFull) => return Err(Error::WindowIsFull), + Err(PumpError::WindowIsLocked) => { + seq = self.lock.wait(seq).unwrap(); + } + } } - ret.map(|r| RecvOk::from_raw(self, app, r)) } pub fn receive_all, P: Into>( &self, @@ -175,21 +157,24 @@ impl SeqExSync { packet: Packet

, ) -> ReplyIter<'_, TL, P, SendData, RecvData, CAP> { if let Ok(r) = self.receive(app.clone(), packet) { - ReplyIter { origin: Some(self), app, first: Some(r) } + ReplyIter { seq: Some(self), app, first: Some(r) } } else { - ReplyIter { origin: None, app, first: None } + ReplyIter { seq: None, app, first: None } } } pub fn try_send>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> { - let mut seq = self.lock(); - seq.try_send(app, packet_data) + self.try_send_with(app, |_| packet_data) } - fn send_inner>( - &self, - mut seq: MutexGuard<'_, (SeqEx, usize)>, - app: TL, - mut packet_data: SendData, - ) { + pub fn send_with>(&self, app: TL, mut packet_data: impl FnMut(SeqNo) -> SendData) { + let mut seq = self.seq_ex.lock().unwrap(); + while let Err(()) = seq.0.try_send_with(app.clone(), &mut packet_data) { + seq.1 += 1; + seq = self.send_block.wait(seq).unwrap(); + seq.1 -= 1; + } + } + pub fn send>(&self, app: TL, mut packet_data: SendData) { + let mut seq = self.seq_ex.lock().unwrap(); while let Err(p) = seq.0.try_send(app.clone(), packet_data) { packet_data = p; seq.1 += 1; @@ -197,19 +182,11 @@ impl SeqExSync { seq.1 -= 1; } } - pub fn send>(&self, app: TL, packet_data: SendData) { - self.send_inner(self.seq_ex.lock().unwrap(), app, packet_data) - } pub fn try_send_with>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> { let mut seq = self.lock(); let seq_no = seq.seq_no(); seq.try_send(app, packet_data(seq_no)) } - pub fn send_with>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) { - let seq = self.seq_ex.lock().unwrap(); - let seq_no = seq.0.seq_no(); - self.send_inner(seq, app, packet_data(seq_no)) - } pub fn service>(&self, app: TL) -> i64 { self.lock().service(app) } @@ -236,7 +213,7 @@ impl<'a, TL: TransportLayer, P: Into, SendData, RecvData, co fn next(&mut self) -> Option { if let Some(g) = self.first.take() { Some(g.into()) - } else if let Some(origin) = self.origin { + } else if let Some(origin) = self.seq { origin.pump(self.app.clone()).ok() } else { None