zerotier/sequential-exchange
0
0
Fork 0
mirror of https://github.com/zerotier/sequential-exchange.git synced 2026-05-22 16:28:28 -07:00
Code Issues Packages Projects Releases Wiki Activity

added mutual exclusion packets

Browse Source
This commit is contained in:
Monica Moniot
2023-08-18 11:27:10 -04:00
parent fd85f8e697
commit 55b19e3af8
3 changed files with 343 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/seq_queue.rs
+169 -65
View File
@@ -65,11 +65,13 @@ pub struct SeqEx<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
/// 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<RecvData> {
seq_no: SeqNo,
reply_no: Option<SeqNo>,
locked: bool,
data: RecvData,
}
@@ -80,51 +82,99 @@ struct SendEntry<SendData> {
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<RecvData> {
/// 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<RecvData>),
WindowIsLocked(Packet<RecvData>),
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<RecvData> {
Payload(SeqNo, RecvData),
LockPayload(SeqNo, RecvData),
Reply(SeqNo, SeqNo, RecvData),
LockReply(SeqNo, SeqNo, RecvData),
Ack(SeqNo),
}
use Packet::*;
impl<RecvData> Packet<RecvData> {
pub fn new_with_data(seq_no: SeqNo, reply_no: Option<SeqNo>, is_locking: bool, data: RecvData) -> Self {
Self::new(Some(seq_no), reply_no, is_locking, Some(data)).unwrap()
}
pub fn new(seq_no: Option<SeqNo>, reply_no: Option<SeqNo>, is_locking: bool, data: Option<RecvData>) -> Option<Self> {
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<SendData>(self, f: impl FnOnce(RecvData) -> SendData) -> Packet<SendData> {
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<RecvData> {
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<RecvData: Clone> Packet<&RecvData> {
pub enum RecvOkRaw<SendData, RecvData> {
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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<SendEntry<SendData>> {
@@ -264,10 +317,9 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// `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<Packet<&SendData>, SendData> {
fn try_send_direct_inner(&mut self, current_time: i64) -> Option<(&mut Option<SendEntry<SendData>>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
}
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<Packet<&SendData>, 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<Packet<&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(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<P: Into<RecvData>>(&mut self, packet: Packet<P>) -> Result<RecvOkRaw<SendData, P>, DirectError> {
pub fn receive_raw_and_direct<P: Into<RecvData>>(&mut self, packet: Packet<P>) -> Result<RecvOkRaw<SendData, P>, DirectError<P>> {
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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
// 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<RecvOkRaw<SendData, RecvData>, Error> {
pub fn pump_raw(&mut self) -> Result<RecvOkRaw<SendData, RecvData>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// 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<Packet<&SendData>> {
pub fn reply_raw_and_direct(&mut self, reply_no: SeqNo, unlock: bool, packet_data: SendData, current_time: i64) -> Option<Packet<&SendData>> {
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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
None
}
}
pub fn ack_raw_and_direct(&mut self, reply_no: SeqNo) -> Option<Packet<&SendData>> {
#[must_use]
pub fn ack_raw_and_direct(&mut self, reply_no: SeqNo, unlock: bool) -> Option<Packet<&SendData>> {
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
src/single_thread.rs
+102 -49
View File
@@ -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>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a mut SeqEx<SendData, RecvData, CAP>,
app: Option<TL>,
reply_no: SeqNo,
locked: bool,
}
impl<'a, TL: TransportLayer<SendData>, 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>, 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>, 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>, SendData, RecvData, const CAP: usize> std
}
}
#[derive(Debug)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Error<RecvData> {
/// 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<RecvData>),
WindowIsLocked(Packet<RecvData>),
}
pub enum RecvOk<'a, TL: TransportLayer<SendData>, 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<SendData>, P, SendData, RecvData, const C
send_data: SendData,
},
}
impl<'a, TL: TransportLayer<SendData>, P, SendData, RecvData, const CAP: usize> RecvOk<'a, TL, P, SendData, RecvData, CAP> {
pub fn from_raw(seq: &'a mut SeqEx<SendData, RecvData, CAP>, app: TL, value: RecvOkRaw<SendData, P>) -> 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<SendData>, 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<SendData>) {
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<SendData>, P, SendData, RecvData, const CAP: usize> $recv<'a, TL, P, SendData, RecvData, CAP> {
pub fn from_raw(seq: $seq_ex, app: TL, value: RecvOkRaw<SendData, P>) -> 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<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> {
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<SendData>) -> Option<Self> {
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<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<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, 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<SendData, RecvData, CAP>);
pub(crate) use impl_recvok;
impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, packet_data: SendData) -> Result<(), SendData> {
@@ -99,12 +142,21 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
Err(e) => Err(e),
}
}
pub fn try_send_with(&mut self, mut app: impl TransportLayer<SendData>, 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<P: Into<RecvData>>(
&mut self,
mut app: impl TransportLayer<SendData>,
packet: Packet<P>,
) -> Result<crate::seq_queue::RecvOkRaw<SendData, P>, Error> {
) -> Result<crate::seq_queue::RecvOkRaw<SendData, P>, Error<P>> {
match self.receive_raw_and_direct(packet) {
Ok(a) => Ok(a),
Err(DirectError::ResendAck(reply_no)) => {
@@ -112,16 +164,17 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<SendData>, 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<SendData>, 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<SendData>, reply_no: SeqNo) {
if let Some(p) = self.ack_raw_and_direct(reply_no) {
pub fn ack_raw(&mut self, mut app: impl TransportLayer<SendData>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
&mut self,
app: TL,
packet: Packet<P>,
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, Error> {
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, Error<P>> {
self.receive_raw(app.clone(), packet).map(|r| RecvOk::from_raw(self, app, r))
}
pub fn pump<TL: TransportLayer<SendData>>(&mut self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
pub fn pump<TL: TransportLayer<SendData>>(&mut self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, PumpError> {
self.pump_raw().map(|r| RecvOk::from_raw(self, app, r))
}
}
src/sync.rs
+72 -95
View File
@@ -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<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq_ex: Mutex<(SeqEx<SendData, RecvData, CAP>, usize)>,
send_block: Condvar,
lock: Condvar,
}
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a SeqExSync<SendData, RecvData, CAP>,
app: TL,
app: Option<TL>,
reply_no: SeqNo,
locked: bool,
}
impl<'a, TL: TransportLayer<SendData>, 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>, 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>, 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>, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
@@ -49,6 +55,16 @@ impl<'a, TL: TransportLayer<SendData>, 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<SendData>, 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<SendData>, P, SendData, RecvData, const C
send_data: SendData,
},
}
impl<'a, TL: TransportLayer<SendData>, 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<SendData>, P, SendData, RecvData, const CAP: usize> RecvOk<'a, TL, P, SendData, RecvData, CAP> {
pub fn from_raw(seq: &'a SeqExSync<SendData, RecvData, CAP>, app: TL, value: RecvOkRaw<SendData, P>) -> 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<SendData>) {
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<SendData>) -> Option<Self> {
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<SendData>, P: Into<RecvData>, 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<SendData, RecvData, CAP>);
pub struct ReplyIter<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
seq: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
app: TL,
first: Option<RecvOk<'a, TL, P, SendData, RecvData, CAP>>,
}
@@ -146,28 +106,50 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
Self {
seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)),
send_block: Condvar::default(),
lock: Condvar::default(),
}
}
pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&self,
app: TL,
packet: Packet<P>,
mut packet: Packet<P>,
) -> Result<RecvOk<'_, TL, P, SendData, RecvData, CAP>, 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<TL: TransportLayer<SendData>>(&self, app: TL) -> Result<RecvOk<'_, TL, RecvData, SendData, RecvData, CAP>, 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<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&self,
@@ -175,21 +157,24 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
packet: Packet<P>,
) -> 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<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(
&self,
mut seq: MutexGuard<'_, (SeqEx<SendData, RecvData, CAP>, usize)>,
app: TL,
mut packet_data: SendData,
) {
pub fn send_with<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&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<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
seq.1 -= 1;
}
}
pub fn send<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: SendData) {
self.send_inner(self.seq_ex.lock().unwrap(), app, packet_data)
}
pub fn try_send_with<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&self, app: TL) -> i64 {
self.lock().service(app)
}
@@ -236,7 +213,7 @@ impl<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, co
fn next(&mut self) -> Option<Self::Item> {
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