From fdc4b6c63760ec11dc6b71e9ac7d05fbab2d6c85 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 17 Aug 2023 16:01:18 -0400 Subject: [PATCH 1/9] incomplete experimental change --- src/seq_queue.rs | 77 +++++++++++++++++++++++------------------- src/single_thread.rs | 14 +++++--- src/sync.rs | 50 +++++++++++++++++---------- src/transport_layer.rs | 2 +- 4 files changed, 85 insertions(+), 58 deletions(-) diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 01ca110..28309ed 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -34,8 +34,6 @@ //! ## Examples //! -use crate::TransportLayer; - /// A 32-bit sequence number. Packets transported with SEP are expected to contain at least one /// sequence number, and sometimes two. /// All packets will either have a seq_no, a reply_no, or both. @@ -92,6 +90,13 @@ pub enum Error { /// 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, + ResendAck(SeqNo), +} + +pub struct Payload<'a, SendData> { + pub seq_no: SeqNo, + pub reply_no: Option, + pub data: &'a SendData, } /// An iterator over all packets in the send window. It will iterate over all packets currently @@ -109,6 +114,11 @@ pub struct Iter<'a, SendData>(core::slice::Iter<'a, Option>> /// the packet. pub struct IterMut<'a, SendData>(core::slice::IterMut<'a, Option>>); +pub struct ServiceIter { + idx: usize, + next_time: i64, +} + impl SeqEx { /// Creates a new instance of `SeqEx` for a new remote peer. /// An instance of `SeqEx` expects to communicate with only exactly one other remote instance @@ -172,14 +182,13 @@ impl SeqEx { /// 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(&mut self, mut app: impl TransportLayer, packet_data: SendData) -> Result<(), SendData> { + pub fn try_send(&mut self, packet_data: SendData, current_time: i64) -> Result, SendData> { if self.is_full() { return Err(packet_data); } let seq_no = self.next_send_seq_no; self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1); - let current_time = app.time(); let next_resend_time = current_time + self.resend_interval; if self.next_service_timestamp > next_resend_time { self.next_service_timestamp = next_resend_time; @@ -188,14 +197,12 @@ impl SeqEx { debug_assert!(slot.is_none()); let entry = slot.insert(SendEntry { seq_no, reply_no: None, next_resend_time, data: packet_data }); - app.send(entry.seq_no, entry.reply_no, &entry.data); - Ok(()) + Ok(Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }) } /// If this returns `Ok` then `try_send` might succeed on next call. pub fn receive_raw>( &mut self, - mut app: impl TransportLayer, seq_no: SeqNo, reply_no: Option, packet: P, @@ -228,8 +235,7 @@ impl SeqEx { return Err(Error::OutOfSequence); } } - app.send_ack(seq_no); - return Err(Error::OutOfSequence); + return Err(Error::ResendAck(seq_no)); } else if is_above_range { return Err(Error::OutOfSequence); } @@ -326,12 +332,12 @@ impl SeqEx { /// 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_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, packet_data: SendData) { + #[must_use] + pub fn reply_raw(&mut self, reply_no: SeqNo, 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); - let current_time = app.time(); let next_resend_time = current_time + self.resend_interval; if self.next_service_timestamp > next_resend_time { self.next_service_timestamp = next_resend_time; @@ -345,14 +351,18 @@ impl SeqEx { data: packet_data, }); - app.send(entry.seq_no, entry.reply_no, &entry.data); + Some(Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }) + } else { + None } } - pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo) { + pub fn ack_raw(&mut self, reply_no: SeqNo) -> Option { if self.remove_reservation(reply_no) { // Acks are only sent once. There is code in `receive_raw` to handle resending // an ack in the event that the first one here was dropped by the network. - app.send_ack(reply_no); + Some(reply_no) + } else { + None } } fn remove_reservation(&mut self, reply_no: SeqNo) -> bool { @@ -367,24 +377,27 @@ impl SeqEx { false } - pub fn service(&mut self, mut app: impl TransportLayer) -> i64 { - let current_time = app.time(); - let real_interval = if self.next_service_timestamp <= current_time { - let next_resend_time = current_time + self.resend_interval; - let mut next_activity = i64::MAX; - for entry in self.send_window.iter_mut().flatten() { - if entry.next_resend_time <= current_time { - entry.next_resend_time = next_resend_time; - app.send(entry.seq_no, entry.reply_no, &entry.data); + pub fn service<'a>(&'a mut self, current_time: i64, iter: &mut Option) -> Option> { + 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 { + if entry.next_resend_time <= current_time { + entry.next_resend_time = current_time + self.resend_interval; + iter.next_time = iter.next_time.min(entry.next_resend_time); + return Some(Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }); + } else { + iter.next_time = iter.next_time.min(entry.next_resend_time); + } } - next_activity = next_activity.min(entry.next_resend_time); } - self.next_service_timestamp = next_activity; - next_activity - current_time - } else { - self.next_service_timestamp - current_time - }; - self.resend_interval.min(real_interval) + self.next_service_timestamp = iter.next_time; + } + None } pub fn iter(&self) -> Iter<'_, SendData> { @@ -428,10 +441,6 @@ macro_rules! iterator { } None } - - fn size_hint(&self) -> (usize, Option) { - (0, Some(self.0.len())) - } } impl<'a, SendData> DoubleEndedIterator for $iter<'a, SendData> { fn next_back(&mut self) -> Option { diff --git a/src/single_thread.rs b/src/single_thread.rs index 0faae3c..a0190cd 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,4 +1,4 @@ -use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; +use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP, Payload}; pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>( &'a mut SeqEx, @@ -11,14 +11,18 @@ 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(self, packet_data: SendData) { - self.0.reply_raw(self.1.clone(), self.2, packet_data); + pub fn reply(mut self, packet_data: SendData) { + if let Some(Payload { seq_no, reply_no, data }) = self.0.reply_raw(self.2, packet_data, self.1.time()) { + self.1.send(seq_no, reply_no, data) + } 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) { - self.0.ack_raw(self.1.clone(), self.2) + if let Some(reply_no) = self.0.ack_raw(self.2) { + self.1.send_ack(reply_no) + } } } @@ -36,7 +40,7 @@ impl SeqEx { reply_no: Option, packet: P, ) -> Result, Error> { - self.receive_raw(app.clone(), seq_no, reply_no, packet) + self.receive_raw(seq_no, reply_no, packet) .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } pub fn pump>(&mut self, app: TL) -> Result, Error> { diff --git a/src/sync.rs b/src/sync.rs index 33c23ef..040c6f0 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -7,7 +7,7 @@ use std::{ time::Instant, }; -use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; +use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, Payload}; pub struct SeqExSync { seq_ex: Mutex<(SeqEx, usize)>, @@ -25,22 +25,24 @@ 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(self, packet_data: SendData) { - let mut seq = self.origin.lock(); - seq.reply_raw(self.app.clone(), self.reply_no, packet_data); - core::mem::forget(self); + pub fn reply(mut self, packet_data: SendData) { + 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 seq = self.origin.lock(); let seq_no = seq.seq_no(); - seq.reply_raw(self.app.clone(), self.reply_no, packet_data(seq_no, self.reply_no)); + if let Some(Payload { seq_no, reply_no, data }) = seq.reply_raw(self.reply_no, packet_data(seq_no, self.reply_no), self.app.time()) { + self.app.send(seq_no, reply_no, data) + } 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.origin.lock(); - seq.ack_raw(self.app.clone(), self.reply_no); + if let Some(reply_no) = seq.ack_raw(self.reply_no) { + self.app.send_ack(reply_no) + } } } @@ -80,13 +82,16 @@ impl SeqExSync { pub fn receive, P: Into>( &self, - app: TL, + mut app: TL, seq_no: SeqNo, reply_no: Option, packet: P, ) -> Result, Error> { let mut seq = self.seq_ex.lock().unwrap(); - let ret = seq.0.receive_raw(app.clone(), seq_no, reply_no, packet); + let ret = seq.0.receive_raw(seq_no, reply_no, packet); + if let Err(Error::ResendAck(ack_no)) = ret { + app.send_ack(ack_no); + } if seq.1 > 0 && ret.is_ok() { self.send_block.notify_one(); } @@ -121,21 +126,30 @@ impl SeqExSync { ReplyIter { origin: None, app, first: None } } } - pub fn try_send>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> { + pub fn try_send>(&self, mut app: TL, packet_data: SendData) -> Result<(), SendData> { let mut seq = self.lock(); - seq.try_send(app, packet_data) + let ret = seq.try_send(packet_data, app.time()); + if let Ok(Payload { seq_no, reply_no, data }) = ret { + app.send(seq_no, reply_no, data) + } + ret.map(|_| ()) } fn send_inner>( &self, mut seq: MutexGuard<'_, (SeqEx, usize)>, - app: TL, + mut app: TL, mut packet_data: SendData, ) { - while let Err(p) = seq.0.try_send(app.clone(), packet_data) { - packet_data = p; - seq.1 += 1; - seq = self.send_block.wait(seq).unwrap(); - seq.1 -= 1; + loop { + let ret = seq.0.try_send(packet_data, app.time()); + if let Err(p) = ret { + packet_data = p; + seq.1 += 1; + seq = self.send_block.wait(seq).unwrap(); + seq.1 -= 1; + } else if let Ok(Payload { seq_no, reply_no, data }) = ret { + app.send(seq_no, reply_no, data) + } } } pub fn send>(&self, app: TL, packet_data: SendData) { diff --git a/src/transport_layer.rs b/src/transport_layer.rs index 53bc10d..d1a3ac4 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -9,6 +9,6 @@ use crate::SeqNo; pub trait TransportLayer: Clone { fn time(&mut self) -> i64; - fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &SendData); + fn send(&mut self, seq_no: SeqNo, reply_no: Option, data: &SendData); fn send_ack(&mut self, reply_no: SeqNo); } From c2d09e6398a6bf5137962a2a3e804c259f5f01f2 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 17 Aug 2023 16:19:54 -0400 Subject: [PATCH 2/9] incomplete updated --- src/seq_queue.rs | 14 ++++--------- src/single_thread.rs | 30 ++++++++++++++++++++------ src/sync.rs | 50 ++++++++++++++++---------------------------- 3 files changed, 46 insertions(+), 48 deletions(-) diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 28309ed..011b177 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -201,7 +201,7 @@ impl SeqEx { } /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn receive_raw>( + pub fn receive_direct>( &mut self, seq_no: SeqNo, reply_no: Option, @@ -333,7 +333,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(&mut self, reply_no: SeqNo, packet_data: SendData, current_time: i64) -> Option> { + pub fn reply_direct(&mut self, reply_no: SeqNo, 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); @@ -356,14 +356,8 @@ impl SeqEx { None } } - pub fn ack_raw(&mut self, reply_no: SeqNo) -> Option { - if self.remove_reservation(reply_no) { - // Acks are only sent once. There is code in `receive_raw` to handle resending - // an ack in the event that the first one here was dropped by the network. - Some(reply_no) - } else { - None - } + pub fn ack_direct(&mut self, reply_no: SeqNo) -> bool { + self.remove_reservation(reply_no) } fn remove_reservation(&mut self, reply_no: SeqNo) -> bool { for i in 0..self.concurrent_replies_total { diff --git a/src/single_thread.rs b/src/single_thread.rs index a0190cd..f7b5846 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -12,16 +12,14 @@ 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(mut self, packet_data: SendData) { - if let Some(Payload { seq_no, reply_no, data }) = self.0.reply_raw(self.2, packet_data, self.1.time()) { - self.1.send(seq_no, reply_no, data) - } + self.0.reply_raw(self.1, self.2, packet_data); 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(reply_no) = self.0.ack_raw(self.2) { - self.1.send_ack(reply_no) + if self.0.ack_direct(self.2) { + self.1.send_ack(self.2) } } } @@ -33,6 +31,26 @@ pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, Send } impl SeqEx { + /// If this returns `Ok` then `try_send` might succeed on next call. + pub fn receive_raw>( + &mut self, + mut app: impl TransportLayer, + seq_no: SeqNo, + reply_no: Option, + packet: P, + ) -> Result<(SeqNo, P, Option), Error> { + let ret = self.receive_direct(seq_no, reply_no, packet); + } + pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, packet_data: SendData) { + if let Some(Payload { seq_no, reply_no, data }) = self.reply_direct(reply_no, packet_data, app.time()) { + app.send(seq_no, reply_no, data) + } + } + pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo) { + if self.ack_direct(reply_no) { + app.send_ack(reply_no) + } + } pub fn receive, P: Into>( &mut self, app: TL, @@ -40,7 +58,7 @@ impl SeqEx { reply_no: Option, packet: P, ) -> Result, Error> { - self.receive_raw(seq_no, reply_no, packet) + self.receive_raw(app, seq_no, reply_no, packet) .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } pub fn pump>(&mut self, app: TL) -> Result, Error> { diff --git a/src/sync.rs b/src/sync.rs index 040c6f0..33c23ef 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -7,7 +7,7 @@ use std::{ time::Instant, }; -use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, Payload}; +use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; pub struct SeqExSync { seq_ex: Mutex<(SeqEx, usize)>, @@ -25,24 +25,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) { - self.reply_with(|_, _| packet_data) + pub fn reply(self, packet_data: SendData) { + let mut seq = self.origin.lock(); + seq.reply_raw(self.app.clone(), self.reply_no, packet_data); + core::mem::forget(self); } - pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { let mut seq = self.origin.lock(); let seq_no = seq.seq_no(); - if let Some(Payload { seq_no, reply_no, data }) = seq.reply_raw(self.reply_no, packet_data(seq_no, self.reply_no), self.app.time()) { - self.app.send(seq_no, reply_no, data) - } + seq.reply_raw(self.app.clone(), self.reply_no, 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.origin.lock(); - if let Some(reply_no) = seq.ack_raw(self.reply_no) { - self.app.send_ack(reply_no) - } + seq.ack_raw(self.app.clone(), self.reply_no); } } @@ -82,16 +80,13 @@ impl SeqExSync { pub fn receive, P: Into>( &self, - mut app: TL, + app: TL, seq_no: SeqNo, reply_no: Option, packet: P, ) -> Result, Error> { let mut seq = self.seq_ex.lock().unwrap(); - let ret = seq.0.receive_raw(seq_no, reply_no, packet); - if let Err(Error::ResendAck(ack_no)) = ret { - app.send_ack(ack_no); - } + let ret = seq.0.receive_raw(app.clone(), seq_no, reply_no, packet); if seq.1 > 0 && ret.is_ok() { self.send_block.notify_one(); } @@ -126,30 +121,21 @@ impl SeqExSync { ReplyIter { origin: None, app, first: None } } } - pub fn try_send>(&self, mut app: TL, packet_data: SendData) -> Result<(), SendData> { + pub fn try_send>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> { let mut seq = self.lock(); - let ret = seq.try_send(packet_data, app.time()); - if let Ok(Payload { seq_no, reply_no, data }) = ret { - app.send(seq_no, reply_no, data) - } - ret.map(|_| ()) + seq.try_send(app, packet_data) } fn send_inner>( &self, mut seq: MutexGuard<'_, (SeqEx, usize)>, - mut app: TL, + app: TL, mut packet_data: SendData, ) { - loop { - let ret = seq.0.try_send(packet_data, app.time()); - if let Err(p) = ret { - packet_data = p; - seq.1 += 1; - seq = self.send_block.wait(seq).unwrap(); - seq.1 -= 1; - } else if let Ok(Payload { seq_no, reply_no, data }) = ret { - app.send(seq_no, reply_no, data) - } + while let Err(p) = seq.0.try_send(app.clone(), packet_data) { + packet_data = p; + seq.1 += 1; + seq = self.send_block.wait(seq).unwrap(); + seq.1 -= 1; } } pub fn send>(&self, app: TL, packet_data: SendData) { From 0e47612158362dceccb5c727c3ff936dfceae41d Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 17 Aug 2023 21:58:03 -0400 Subject: [PATCH 3/9] saving progress --- src/seq_queue.rs | 132 ++++++++++++++++++++++++----------------- src/single_thread.rs | 7 ++- src/transport_layer.rs | 5 +- 3 files changed, 83 insertions(+), 61 deletions(-) diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 011b177..6e3ecc2 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -83,7 +83,7 @@ struct SendEntry { /// 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 { +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, @@ -93,10 +93,27 @@ pub enum Error { ResendAck(SeqNo), } -pub struct Payload<'a, SendData> { - pub seq_no: SeqNo, - pub reply_no: Option, - pub data: &'a 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 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 Packet<'a, SendData> { + Payload { + seq_no: SeqNo, + reply_no: Option, + data: &'a SendData, + }, + Ack { + reply_no: SeqNo, + } } /// An iterator over all packets in the send window. It will iterate over all packets currently @@ -114,6 +131,7 @@ pub struct Iter<'a, SendData>(core::slice::Iter<'a, Option>> /// the packet. pub struct IterMut<'a, SendData>(core::slice::IterMut<'a, Option>>); +#[derive(Clone, Debug)] pub struct ServiceIter { idx: usize, next_time: i64, @@ -147,6 +165,38 @@ impl SeqEx { fn send_window_slot(&self, seq_no: SeqNo) -> &Option> { &self.send_window[seq_no as usize % self.send_window.len()] } + fn is_full_inner(&self, reserve_one: bool) -> bool { + if self.concurrent_replies_total >= self.concurrent_replies.len() { + return true; + } + for i in 0..self.concurrent_replies_total as u32 + 1 + reserve_one as u32 { + let slot = self.send_window_slot(self.next_send_seq_no.wrapping_add(i)); + if slot.is_some() { + return true; + } + } + false + } + fn take_send(&mut self, reply_no: SeqNo) -> Option { + let slot = self.send_window_slot_mut(reply_no); + if slot.as_ref().map_or(false, |e| e.seq_no == reply_no) { + slot.take().map(|e| e.data) + } else { + None + } + } + + fn remove_reservation(&mut self, reply_no: SeqNo) -> bool { + for i in 0..self.concurrent_replies_total { + if self.concurrent_replies[i] == reply_no { + // swap remove + self.concurrent_replies_total -= 1; + self.concurrent_replies[i] = self.concurrent_replies[self.concurrent_replies_total]; + return true; + } + } + false + } /// Returns whether or not the send window is full. /// If the send window is full calls to `SeqEx::send` will always fail. @@ -182,7 +232,7 @@ impl SeqEx { /// 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(&mut self, packet_data: SendData, current_time: i64) -> Result, SendData> { + pub fn try_send_direct(&mut self, packet_data: SendData, current_time: i64) -> Result, SendData> { if self.is_full() { return Err(packet_data); } @@ -197,7 +247,7 @@ impl SeqEx { debug_assert!(slot.is_none()); let entry = slot.insert(SendEntry { seq_no, reply_no: None, next_resend_time, data: packet_data }); - Ok(Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }) + Ok(Packet::Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }) } /// If this returns `Ok` then `try_send` might succeed on next call. @@ -206,7 +256,7 @@ impl SeqEx { seq_no: SeqNo, reply_no: Option, packet: P, - ) -> Result<(SeqNo, P, Option), Error> { + ) -> Result<(SeqNo, P, Option), DirectError> { // 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 @@ -227,17 +277,17 @@ impl SeqEx { // resending the packet. for entry in self.send_window.iter().flatten() { if entry.reply_no == Some(seq_no) { - return Err(Error::OutOfSequence); + return Err(DirectError::OutOfSequence); } } for i in 0..self.concurrent_replies_total { if self.concurrent_replies[i] == seq_no { - return Err(Error::OutOfSequence); + return Err(DirectError::OutOfSequence); } } - return Err(Error::ResendAck(seq_no)); + return Err(DirectError::ResendAck(seq_no)); } else if is_above_range { - return Err(Error::OutOfSequence); + return Err(DirectError::OutOfSequence); } // If the send window is full we cannot safely process received packets, // because there would be no way to reply. @@ -251,15 +301,15 @@ impl SeqEx { self.recv_window[i] = None; } else { return if is_full { - Err(Error::WindowIsFull) + Err(DirectError::WindowIsFull) } else { - Err(Error::OutOfSequence) + Err(DirectError::OutOfSequence) }; } } 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(Error::OutOfSequence); + return Err(DirectError::OutOfSequence); } } if is_next && !is_full { @@ -271,9 +321,9 @@ impl SeqEx { } else { self.recv_window[i] = Some(RecvEntry { seq_no, reply_no, data: packet.into() }); if is_full { - Err(Error::WindowIsFull) + Err(DirectError::WindowIsFull) } else { - Err(Error::OutOfSequence) + Err(DirectError::OutOfSequence) } } } @@ -281,28 +331,6 @@ impl SeqEx { pub fn receive_ack(&mut self, reply_no: SeqNo) -> Result { self.take_send(reply_no).ok_or(Error::OutOfSequence) } - - fn is_full_inner(&self, reserve_one: bool) -> bool { - if self.concurrent_replies_total >= self.concurrent_replies.len() { - return true; - } - for i in 0..self.concurrent_replies_total as u32 + 1 + reserve_one as u32 { - let slot = self.send_window_slot(self.next_send_seq_no.wrapping_add(i)); - if slot.is_some() { - return true; - } - } - false - } - - fn take_send(&mut self, reply_no: SeqNo) -> Option { - let slot = self.send_window_slot_mut(reply_no); - if slot.as_ref().map_or(false, |e| e.seq_no == reply_no) { - slot.take().map(|e| e.data) - } else { - None - } - } /// If this returns `Ok` then `try_send` might succeed on next call. pub fn pump_raw(&mut self) -> Result<(SeqNo, RecvData, Option), Error> { let next_seq_no = self.pre_recv_seq_no.wrapping_add(1); @@ -323,7 +351,6 @@ impl SeqEx { Err(Error::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. /// @@ -333,7 +360,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_direct(&mut self, reply_no: SeqNo, packet_data: SendData, current_time: i64) -> Option> { + pub fn reply_direct(&mut self, reply_no: SeqNo, 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); @@ -351,27 +378,20 @@ impl SeqEx { data: packet_data, }); - Some(Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }) + Some(Packet::Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }) } else { None } } - pub fn ack_direct(&mut self, reply_no: SeqNo) -> bool { - self.remove_reservation(reply_no) - } - fn remove_reservation(&mut self, reply_no: SeqNo) -> bool { - for i in 0..self.concurrent_replies_total { - if self.concurrent_replies[i] == reply_no { - // swap remove - self.concurrent_replies_total -= 1; - self.concurrent_replies[i] = self.concurrent_replies[self.concurrent_replies_total]; - return true; - } + pub fn ack_direct(&mut self, reply_no: SeqNo) -> Option> { + if self.remove_reservation(reply_no) { + Some(Packet::Ack { reply_no }) + } else { + None } - false } - pub fn service<'a>(&'a mut self, current_time: i64, iter: &mut Option) -> Option> { + pub fn service<'a>(&'a mut self, current_time: i64, iter: &mut Option) -> Option> { if self.next_service_timestamp <= current_time { let iter = iter.get_or_insert(ServiceIter { idx: 0, @@ -383,7 +403,7 @@ impl SeqEx { if entry.next_resend_time <= current_time { entry.next_resend_time = current_time + self.resend_interval; iter.next_time = iter.next_time.min(entry.next_resend_time); - return Some(Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }); + return Some(Packet::Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }); } else { iter.next_time = iter.next_time.min(entry.next_resend_time); } diff --git a/src/single_thread.rs b/src/single_thread.rs index f7b5846..9514063 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -18,8 +18,8 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { - if self.0.ack_direct(self.2) { - self.1.send_ack(self.2) + if let Some(p) = self.0.ack_direct(self.2) { + self.1.send(p) } } } @@ -31,6 +31,9 @@ pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, Send } impl SeqEx { + pub fn try_send_raw(&mut self, packet_data: SendData, current_time: i64) -> Result, SendData> { + + } /// If this returns `Ok` then `try_send` might succeed on next call. pub fn receive_raw>( &mut self, diff --git a/src/transport_layer.rs b/src/transport_layer.rs index d1a3ac4..62c19dc 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -1,4 +1,4 @@ -use crate::SeqNo; +use crate::Packet; /// A trait for giving an instance of SeqEx access to the transport layer. /// @@ -9,6 +9,5 @@ use crate::SeqNo; pub trait TransportLayer: Clone { fn time(&mut self) -> i64; - fn send(&mut self, seq_no: SeqNo, reply_no: Option, data: &SendData); - fn send_ack(&mut self, reply_no: SeqNo); + fn send(&mut self, packet: Packet<'_, SendData>); } From 63ea9c6ed24d784c57874774ac9c92477a8278ae Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 17 Aug 2023 22:25:22 -0400 Subject: [PATCH 4/9] integrated changes --- examples/calculator.rs | 8 ++--- examples/file_download.rs | 21 +++++-------- examples/hello_world.rs | 8 ++--- src/seq_queue.rs | 28 +++++++++++------ src/single_thread.rs | 65 +++++++++++++++++++++++++++++---------- src/sync.rs | 25 +++++++++------ src/tokio.rs | 40 ++++++++++++++---------- 7 files changed, 121 insertions(+), 74 deletions(-) diff --git a/examples/calculator.rs b/examples/calculator.rs index 63ca26d..fd4df9a 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -1,6 +1,6 @@ use std::{sync::mpsc::Receiver, thread, time::Duration}; -use seq_ex::sync::{MpscGuard, MpscSeqEx, MpscTransport, PacketType, RecvSuccess}; +use seq_ex::sync::{MpscGuard, MpscSeqEx, MpscTransport, PacketOwned, RecvSuccess}; #[derive(Clone)] enum Packet { @@ -27,14 +27,14 @@ fn process(_: MpscGuard<'_, Packet>, recv_packet: Packet, _: Option, val } } -fn receive(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport, value: &mut f32) { +fn receive(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport, value: &mut f32) { while let Ok(packet) = recv.try_recv() { if !drop_packet() { match packet { - PacketType::Ack(reply_no) => { + PacketOwned::Ack(reply_no) => { let _ = seq.receive_ack(reply_no); } - PacketType::Payload(seq_no, reply_no, payload) => { + PacketOwned::Payload(seq_no, reply_no, payload) => { for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) { process(guard, packet, send_data, value); } diff --git a/examples/file_download.rs b/examples/file_download.rs index 8d2be99..3f0bd5c 100644 --- a/examples/file_download.rs +++ b/examples/file_download.rs @@ -11,8 +11,8 @@ use std::{ use rand_core::{OsRng, RngCore}; use seq_ex::{ - sync::{PacketType, RecvSuccess, ReplyGuard, SeqExSync}, - SeqNo, TransportLayer, + sync::{PacketOwned, RecvSuccess, ReplyGuard, SeqExSync}, + TransportLayer, }; use serde::{Deserialize, Serialize}; @@ -41,15 +41,8 @@ impl TransportLayer for &Transport { self.time.elapsed().as_millis() as i64 } - fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &Packet) { - let p = PacketType::Payload(seq_no, reply_no, payload.clone()); - if let Ok(p) = serde_json::to_vec(&p) { - let _ = self.sender.send(p); - } - } - - fn send_ack(&mut self, reply_no: SeqNo) { - let p = PacketType::::Ack(reply_no); + fn send(&mut self, packet: seq_ex::Packet<'_, Packet>) { + let p = PacketOwned::from(packet); if let Ok(p) = serde_json::to_vec(&p) { let _ = self.sender.send(p); } @@ -110,12 +103,12 @@ fn receive(peer: &Peer) { if drop_packet() { continue; } - let parsed_packet = serde_json::from_slice::>(&packet); + let parsed_packet = serde_json::from_slice::>(&packet); match parsed_packet { - Ok(PacketType::Ack(reply_no)) => { + Ok(PacketOwned::Ack(reply_no)) => { let _ = peer.seqex.receive_ack(reply_no); } - Ok(PacketType::Payload(seq_no, reply_no, payload)) => { + Ok(PacketOwned::Payload(seq_no, reply_no, payload)) => { for RecvSuccess { guard, packet, send_data } in peer.seqex.receive_all(&peer.transport, seq_no, reply_no, payload) { process(peer, guard, packet, send_data); } diff --git a/examples/hello_world.rs b/examples/hello_world.rs index 7202448..b2bc0d0 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -1,6 +1,6 @@ use std::sync::mpsc::Receiver; -use seq_ex::sync::{MpscGuard, MpscSeqEx, MpscTransport, PacketType, RecvSuccess}; +use seq_ex::sync::{MpscGuard, MpscSeqEx, MpscTransport, PacketOwned, RecvSuccess}; #[derive(Clone, Debug)] enum Packet { @@ -37,16 +37,16 @@ fn process(guard: MpscGuard<'_, Packet>, recv_packet: Packet, send_packet: Optio } } -fn receive(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport) { +fn receive(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport) { match recv.recv().unwrap() { - PacketType::Ack(reply_no) => { + PacketOwned::Ack(reply_no) => { let result = seq.receive_ack(reply_no); if let Ok(Exclamation) = result { // Our Hello World exchange ends right here. print!("\n"); } } - PacketType::Payload(seq_no, reply_no, payload) => { + PacketOwned::Payload(seq_no, reply_no, payload) => { for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) { process(guard, packet, send_data) } diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 6e3ecc2..4d5f58f 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -113,7 +113,7 @@ pub enum Packet<'a, SendData> { }, Ack { reply_no: SeqNo, - } + }, } /// An iterator over all packets in the send window. It will iterate over all packets currently @@ -247,7 +247,11 @@ impl SeqEx { debug_assert!(slot.is_none()); let entry = slot.insert(SendEntry { seq_no, reply_no: None, next_resend_time, data: packet_data }); - Ok(Packet::Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }) + Ok(Packet::Payload { + seq_no: entry.seq_no, + reply_no: entry.reply_no, + data: &entry.data, + }) } /// If this returns `Ok` then `try_send` might succeed on next call. @@ -378,7 +382,11 @@ impl SeqEx { data: packet_data, }); - Some(Packet::Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }) + Some(Packet::Payload { + seq_no: entry.seq_no, + reply_no: entry.reply_no, + data: &entry.data, + }) } else { None } @@ -391,19 +399,21 @@ impl SeqEx { } } - pub fn service<'a>(&'a mut self, current_time: i64, iter: &mut Option) -> Option> { + pub fn service_direct<'a>(&'a mut self, current_time: i64, iter: &mut Option) -> Option> { if self.next_service_timestamp <= current_time { - let iter = iter.get_or_insert(ServiceIter { - idx: 0, - next_time: i64::MAX, - }); + 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 { 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); - return Some(Packet::Payload { seq_no: entry.seq_no, reply_no: entry.reply_no, data: &entry.data }); + return Some(Packet::Payload { + seq_no: entry.seq_no, + reply_no: entry.reply_no, + data: &entry.data, + }); } else { iter.next_time = iter.next_time.min(entry.next_resend_time); } diff --git a/src/single_thread.rs b/src/single_thread.rs index 9514063..0681679 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,8 +1,8 @@ -use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP, Payload}; +use crate::{DirectError, Error, Packet, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>( &'a mut SeqEx, - TL, + Option, SeqNo, ); impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { @@ -12,14 +12,18 @@ 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(mut self, packet_data: SendData) { - self.0.reply_raw(self.1, self.2, packet_data); + let mut app = None; + core::mem::swap(&mut app, &mut self.1); + self.0.reply_raw(app.unwrap(), self.2, packet_data); 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(p) = self.0.ack_direct(self.2) { - self.1.send(p) + if let Some(app) = &mut self.1 { + if let Some(p) = self.0.ack_direct(self.2) { + app.send(p) + } } } } @@ -31,8 +35,14 @@ pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, Send } impl SeqEx { - pub fn try_send_raw(&mut self, packet_data: SendData, current_time: i64) -> Result, SendData> { - + pub fn try_send(&mut self, mut app: impl TransportLayer, packet_data: SendData) -> Result<(), SendData> { + match self.try_send_direct(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>( @@ -42,18 +52,34 @@ impl SeqEx { reply_no: Option, packet: P, ) -> Result<(SeqNo, P, Option), Error> { - let ret = self.receive_direct(seq_no, reply_no, packet); + match self.receive_direct(seq_no, reply_no, packet) { + Ok(a) => Ok(a), + Err(DirectError::ResendAck(reply_no)) => { + app.send(Packet::Ack { reply_no }); + Err(Error::OutOfSequence) + } + Err(DirectError::OutOfSequence) => Err(Error::OutOfSequence), + Err(DirectError::WindowIsFull) => Err(Error::WindowIsFull), + } } pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, packet_data: SendData) { - if let Some(Payload { seq_no, reply_no, data }) = self.reply_direct(reply_no, packet_data, app.time()) { - app.send(seq_no, reply_no, data) + if let Some(p) = self.reply_direct(reply_no, packet_data, app.time()) { + app.send(p) } } pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo) { - if self.ack_direct(reply_no) { - app.send_ack(reply_no) + if let Some(p) = self.ack_direct(reply_no) { + app.send(p) } } + pub fn service(&mut self, mut app: impl TransportLayer) -> i64 { + let current_time = app.time(); + let mut iter = None; + while let Some(p) = self.service_direct(current_time, &mut iter) { + app.send(p) + } + self.resend_interval.min(self.next_service_timestamp - current_time) + } pub fn receive, P: Into>( &mut self, app: TL, @@ -61,11 +87,18 @@ impl SeqEx { reply_no: Option, packet: P, ) -> Result, Error> { - self.receive_raw(app, seq_no, reply_no, packet) - .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) + self.receive_raw(app.clone(), seq_no, reply_no, packet) + .map(|(reply_no, packet, send_data)| RecvSuccess { + guard: ReplyGuard(self, Some(app), reply_no), + packet, + send_data, + }) } pub fn pump>(&mut self, app: TL) -> Result, Error> { - self.pump_raw() - .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) + self.pump_raw().map(|(reply_no, packet, send_data)| RecvSuccess { + guard: ReplyGuard(self, Some(app), reply_no), + packet, + send_data, + }) } } diff --git a/src/sync.rs b/src/sync.rs index 33c23ef..8830801 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -7,7 +7,7 @@ use std::{ time::Instant, }; -use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; +use crate::{Error, Packet, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; pub struct SeqExSync { seq_ex: Mutex<(SeqEx, usize)>, @@ -189,25 +189,33 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Ite #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] #[derive(Clone)] -pub enum PacketType { +pub enum PacketOwned { Payload(SeqNo, Option, Payload), Ack(SeqNo), } +impl<'a, Payload: Clone> From> for PacketOwned { + fn from(value: Packet<'a, Payload>) -> Self { + match value { + Packet::Payload { seq_no, reply_no, data } => PacketOwned::Payload(seq_no, reply_no, data.clone()), + Packet::Ack { reply_no } => PacketOwned::Ack(reply_no), + } + } +} #[derive(Clone)] pub struct MpscTransport { - pub channel: Sender>, + pub channel: Sender>, pub time: Instant, } pub type MpscGuard<'a, Packet> = ReplyGuard<'a, &'a MpscTransport, Packet, Packet>; pub type MpscSeqEx = SeqExSync; impl MpscTransport { - pub fn new() -> (Self, Receiver>) { + pub fn new() -> (Self, Receiver>) { let (send, recv) = channel(); (Self { channel: send, time: std::time::Instant::now() }, recv) } - pub fn from_sender(send: Sender>) -> Self { + pub fn from_sender(send: Sender>) -> Self { Self { channel: send, time: std::time::Instant::now() } } } @@ -216,10 +224,7 @@ impl TransportLayer for &MpscTransport { self.time.elapsed().as_millis() as i64 } - fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &Payload) { - let _ = self.channel.send(PacketType::Payload(seq_no, reply_no, payload.clone())); - } - fn send_ack(&mut self, reply_no: SeqNo) { - let _ = self.channel.send(PacketType::Ack(reply_no)); + fn send(&mut self, packet: Packet<'_, Payload>) { + let _ = self.channel.send(PacketOwned::from(packet)); } } diff --git a/src/tokio.rs b/src/tokio.rs index 814eac7..c2ec721 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -1,9 +1,8 @@ -use std:: - sync::{ - Mutex, MutexGuard, - } -; -use tokio::{task, sync::{Notify, oneshot, mpsc}, time}; +use std::sync::{Mutex, MutexGuard}; +use tokio::{ + sync::{mpsc, oneshot, Notify}, + task, time, +}; use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; @@ -82,11 +81,7 @@ pub struct TokioTransport { impl TokioTransport { pub fn new> + Send + 'static>(app: TL, seq: S) -> Self { let (update_queue, mut recv) = mpsc::channel(4); - let ret = TokioTransport { - time: time::Instant::now(), - update_queue, - app, - }; + let ret = TokioTransport { time: time::Instant::now(), update_queue, app }; let task_tl = ret.clone(); task::spawn(async move { let mut update_ts = i64::MAX; @@ -96,7 +91,7 @@ impl TokioTransport { let mut do_update = diff <= 0; if diff > 0 { let sleep = time::sleep(time::Duration::from_millis(diff as u64)); - tokio::select!{ + tokio::select! { Some(up) = recv.recv() => { update_ts = up; } @@ -131,7 +126,7 @@ impl SeqExTokio { &'a self, app: &'a TokioTransport, mut seq: MutexGuard<'_, (SeqEx, Packet, CAP>, usize)>, - result: Result<(SeqNo, Packet, Option>), Error> + result: Result<(SeqNo, Packet, Option>), Error>, ) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { if let Ok((reply_no, packet, send_data)) = result { if seq.1 > 0 { @@ -161,7 +156,10 @@ impl SeqExTokio { let result = seq.0.receive_raw(app, seq_no, reply_no, packet); self.process(app, seq, result) } - pub fn pump<'a, TL: TokioTransportLayer>(&'a self, app: &'a TokioTransport) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { + pub fn pump<'a, TL: TokioTransportLayer>( + &'a self, + app: &'a TokioTransport, + ) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { let mut seq = self.seq_ex.lock().unwrap(); let result = seq.0.pump_raw(); self.process(app, seq, result) @@ -197,7 +195,7 @@ impl SeqExTokio { mut seq: MutexGuard<'_, (SeqEx, Packet, CAP>, usize)>, app: &TokioTransport, mut tx: oneshot::Sender<(Packet, SeqNo)>, - mut packet: Packet + mut packet: Packet, ) { let mut pre_ts = seq.0.next_service_timestamp; while let Err(e) = seq.0.try_send(app, (tx, packet)) { @@ -214,7 +212,11 @@ impl SeqExTokio { } } /// If this future is dropped then the remote peer's reply to this packet will also be dropped. - pub async fn send<'a, TL: TokioTransportLayer>(&'a self, app: &'a TokioTransport, packet: Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { + pub async fn send<'a, TL: TokioTransportLayer>( + &'a self, + app: &'a TokioTransport, + packet: Packet, + ) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { self.send_with(app, |_| packet).await } //pub fn try_send_with>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> { @@ -222,7 +224,11 @@ impl SeqExTokio { // let seq_no = seq.seq_no(); // seq.try_send(app, packet_data(seq_no)) //} - pub async fn send_with<'a, TL: TokioTransportLayer>(&'a self, app: &'a TokioTransport, packet: impl FnOnce(SeqNo) -> Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { + pub async fn send_with<'a, TL: TokioTransportLayer>( + &'a self, + app: &'a TokioTransport, + packet: impl FnOnce(SeqNo) -> Packet, + ) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { let (tx, rx) = oneshot::channel(); let seq = self.seq_ex.lock().unwrap(); let seq_no = seq.0.seq_no(); From 7454868a1e156f78eafbca7e7d8a5b99e4447bc4 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 17 Aug 2023 22:28:31 -0400 Subject: [PATCH 5/9] renamed functions --- src/seq_queue.rs | 6 +++--- src/single_thread.rs | 8 ++++---- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 4d5f58f..689f994 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -255,7 +255,7 @@ impl SeqEx { } /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn receive_direct>( + pub fn receive_raw_and_direct>( &mut self, seq_no: SeqNo, reply_no: Option, @@ -364,7 +364,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_direct(&mut self, reply_no: SeqNo, packet_data: SendData, current_time: i64) -> Option> { + pub fn reply_raw_and_direct(&mut self, reply_no: SeqNo, 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); @@ -391,7 +391,7 @@ impl SeqEx { None } } - pub fn ack_direct(&mut self, reply_no: SeqNo) -> Option> { + pub fn ack_raw_and_direct(&mut self, reply_no: SeqNo) -> Option> { if self.remove_reservation(reply_no) { Some(Packet::Ack { reply_no }) } else { diff --git a/src/single_thread.rs b/src/single_thread.rs index 0681679..15ba365 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -21,7 +21,7 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { if let Some(app) = &mut self.1 { - if let Some(p) = self.0.ack_direct(self.2) { + if let Some(p) = self.0.ack_raw_and_direct(self.2) { app.send(p) } } @@ -52,7 +52,7 @@ impl SeqEx { reply_no: Option, packet: P, ) -> Result<(SeqNo, P, Option), Error> { - match self.receive_direct(seq_no, reply_no, packet) { + match self.receive_raw_and_direct(seq_no, reply_no, packet) { Ok(a) => Ok(a), Err(DirectError::ResendAck(reply_no)) => { app.send(Packet::Ack { reply_no }); @@ -63,12 +63,12 @@ impl SeqEx { } } pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, packet_data: SendData) { - if let Some(p) = self.reply_direct(reply_no, packet_data, app.time()) { + if let Some(p) = self.reply_raw_and_direct(reply_no, 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_direct(reply_no) { + if let Some(p) = self.ack_raw_and_direct(reply_no) { app.send(p) } } From fd85f8e697349a067846ae677e904e50be65d405 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Fri, 18 Aug 2023 01:10:47 -0400 Subject: [PATCH 6/9] massive improvement --- examples/calculator.rs | 51 +++++------ examples/file_download.rs | 57 +++++------- examples/hello_world.rs | 69 +++++++-------- src/seq_queue.rs | 112 +++++++++++++++--------- src/single_thread.rs | 114 ++++++++++++++++-------- src/sync.rs | 178 +++++++++++++++++++++++--------------- src/transport_layer.rs | 2 +- 7 files changed, 338 insertions(+), 245 deletions(-) diff --git a/examples/calculator.rs b/examples/calculator.rs index fd4df9a..91e7377 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -1,9 +1,12 @@ use std::{sync::mpsc::Receiver, thread, time::Duration}; -use seq_ex::sync::{MpscGuard, MpscSeqEx, MpscTransport, PacketOwned, RecvSuccess}; +use seq_ex::{ + sync::{MpscSeqEx, MpscTransport}, + Packet, +}; #[derive(Clone)] -enum Packet { +enum Payload { Add(f32), Sub(f32), Mul(f32), @@ -16,28 +19,20 @@ fn drop_packet() -> bool { rand_core::OsRng.next_u32() & 1 > 0 } -fn process(_: MpscGuard<'_, Packet>, recv_packet: Packet, _: Option, value: &mut f32) { - use Packet::*; - match recv_packet { - Add(n) => *value = *value + n, - Sub(n) => *value = *value - n, - Mul(n) => *value = *value * n, - Div(n) => *value = *value / n, - Mod(n) => *value = *value % n, - } -} - -fn receive(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport, value: &mut f32) { +fn receive(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport, value: &mut f32) { while let Ok(packet) = recv.try_recv() { - if !drop_packet() { - match packet { - PacketOwned::Ack(reply_no) => { - let _ = seq.receive_ack(reply_no); - } - PacketOwned::Payload(seq_no, reply_no, payload) => { - for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) { - process(guard, packet, send_data, value); - } + if drop_packet() { + continue; + } + for recv_data in seq.receive_all(transport, packet) { + use Payload::*; + if let Some((_, recv_packet)) = recv_data.consume().0 { + match recv_packet { + Add(n) => *value = *value + n, + Sub(n) => *value = *value - n, + Mul(n) => *value = *value * n, + Div(n) => *value = *value / n, + Mod(n) => *value = *value % n, } } } @@ -52,15 +47,15 @@ fn main() { let mut value = 0.0; let mut remote_value = value; - seq1.send(&transport1, Packet::Add(1.0)); + seq1.send(&transport1, Payload::Add(1.0)); value += 1.0; - seq1.send(&transport1, Packet::Sub(2.0)); + seq1.send(&transport1, Payload::Sub(2.0)); value -= 2.0; - seq1.send(&transport1, Packet::Mul(3.0)); + seq1.send(&transport1, Payload::Mul(3.0)); value *= 3.0; - seq1.send(&transport1, Packet::Div(4.0)); + seq1.send(&transport1, Payload::Div(4.0)); value /= 4.0; - seq1.send(&transport1, Packet::Mod(5.0)); + seq1.send(&transport1, Payload::Mod(5.0)); value %= 5.0; for _ in 0..16 { diff --git a/examples/file_download.rs b/examples/file_download.rs index 3f0bd5c..0319dfe 100644 --- a/examples/file_download.rs +++ b/examples/file_download.rs @@ -11,14 +11,14 @@ use std::{ use rand_core::{OsRng, RngCore}; use seq_ex::{ - sync::{PacketOwned, RecvSuccess, ReplyGuard, SeqExSync}, - TransportLayer, + sync::{RecvOk, SeqExSync}, + Packet, TransportLayer, }; use serde::{Deserialize, Serialize}; const FILE_CHUNK_SIZE: usize = 1000; #[derive(Clone, Debug, Serialize, Deserialize)] -enum Packet { +enum Payload { RequestFile { filename: String }, ConfirmRequestFile { filesize: u64 }, FileDownload { filename: String, file_chunk: Vec }, @@ -32,18 +32,17 @@ struct Transport { struct Peer { filesystem: Arc>>>, transport: Transport, - seqex: Arc>, + seqex: Arc>, receiver: Receiver>, } -impl TransportLayer for &Transport { +impl TransportLayer for &Transport { fn time(&mut self) -> i64 { self.time.elapsed().as_millis() as i64 } - fn send(&mut self, packet: seq_ex::Packet<'_, Packet>) { - let p = PacketOwned::from(packet); - if let Ok(p) = serde_json::to_vec(&p) { + fn send(&mut self, packet: Packet<&Payload>) { + if let Ok(p) = serde_json::to_vec(&packet) { let _ = self.sender.send(p); } } @@ -53,14 +52,15 @@ fn drop_packet() -> bool { OsRng.next_u32() >= (u32::MAX / 4 * 3) } -fn process(peer: &Peer, guard: ReplyGuard<'_, &Transport, Packet, Packet>, recv_packet: Packet, sent_packet: Option) { - match (recv_packet, sent_packet) { - (Packet::RequestFile { filename }, None) => { +fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload, Payload>) { + use Payload::*; + match recv_data.consume() { + (Some((guard, RequestFile { filename })), None) => { let filesystem = peer.filesystem.clone(); let transport = peer.transport.clone(); let seqex = peer.seqex.clone(); if let Some(file) = filesystem.read().unwrap().get(&filename) { - guard.reply(Packet::ConfirmRequestFile { filesize: file.len() as u64 }); + guard.reply(ConfirmRequestFile { filesize: file.len() as u64 }); } thread::spawn(move || { let filesystem = filesystem.read().unwrap(); @@ -70,21 +70,18 @@ fn process(peer: &Peer, guard: ReplyGuard<'_, &Transport, Packet, Packet>, recv_ let mut i = 0; while i < file.len() { let j = file.len().min(i + FILE_CHUNK_SIZE); - seqex.send( - &transport, - Packet::FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() }, - ); + seqex.send(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() }); i = j; } } }); } - (Packet::ConfirmRequestFile { filesize }, Some(Packet::RequestFile { filename })) => { + (Some((_, ConfirmRequestFile { filesize })), Some(RequestFile { filename })) => { let mut filesystem = peer.filesystem.write().unwrap(); let file = Vec::with_capacity(filesize as usize); filesystem.insert(filename, file); } - (Packet::FileDownload { filename, file_chunk }, None) => { + (Some((_, FileDownload { filename, file_chunk })), None) => { let mut filesystem = peer.filesystem.write().unwrap(); if let Some(file) = filesystem.get_mut(&filename) { if file.len() + file_chunk.len() <= file.capacity() { @@ -92,9 +89,10 @@ fn process(peer: &Peer, guard: ReplyGuard<'_, &Transport, Packet, Packet>, recv_ } } } - _ => { - assert!(false); + (Some(a), b) => { + print!("Unsolicited packet received: {:?}", RecvOk::new(Some(a), b)); } + _ => {} } } @@ -103,17 +101,10 @@ fn receive(peer: &Peer) { if drop_packet() { continue; } - let parsed_packet = serde_json::from_slice::>(&packet); - match parsed_packet { - Ok(PacketOwned::Ack(reply_no)) => { - let _ = peer.seqex.receive_ack(reply_no); + if let Ok(parsed_packet) = serde_json::from_slice::>(&packet) { + for recv_data in peer.seqex.receive_all(&peer.transport, parsed_packet) { + process(peer, recv_data); } - Ok(PacketOwned::Payload(seq_no, reply_no, payload)) => { - for RecvSuccess { guard, packet, send_data } in peer.seqex.receive_all(&peer.transport, seq_no, reply_no, payload) { - process(peer, guard, packet, send_data); - } - } - _ => {} } } } @@ -146,9 +137,9 @@ fn main() { receiver: recv2, }; - peer1.seqex.send(&peer1.transport, Packet::RequestFile { filename: "File1".to_string() }); - peer1.seqex.send(&peer1.transport, Packet::RequestFile { filename: "File3".to_string() }); - peer1.seqex.send(&peer1.transport, Packet::RequestFile { filename: "File2".to_string() }); + peer1.seqex.send(&peer1.transport, Payload::RequestFile { filename: "File1".to_string() }); + peer1.seqex.send(&peer1.transport, Payload::RequestFile { filename: "File3".to_string() }); + peer1.seqex.send(&peer1.transport, Payload::RequestFile { filename: "File2".to_string() }); for _ in 0..300 { receive(&peer1); diff --git a/examples/hello_world.rs b/examples/hello_world.rs index b2bc0d0..3e284ca 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -1,55 +1,46 @@ use std::sync::mpsc::Receiver; -use seq_ex::sync::{MpscGuard, MpscSeqEx, MpscTransport, PacketOwned, RecvSuccess}; +use seq_ex::{ + sync::{MpscSeqEx, MpscTransport, RecvOk}, + Packet, +}; #[derive(Clone, Debug)] -enum Packet { +enum Payload { Hello, Space, World, Exclamation, } -use Packet::*; +use Payload::*; -fn process(guard: MpscGuard<'_, Packet>, recv_packet: Packet, send_packet: Option) { - 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(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport) { - match recv.recv().unwrap() { - PacketOwned::Ack(reply_no) => { - let result = seq.receive_ack(reply_no); - if let Ok(Exclamation) = result { +fn receive(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport) { + let packet = recv.recv().unwrap(); + for recv_data in seq.receive_all(transport, packet) { + match recv_data.consume() { + (Some((guard, Hello)), None) => { + print!("Hello"); + guard.reply(Space); + } + (Some((guard, Space)), Some(Hello)) => { + print!(" "); + guard.reply(World); + } + (Some((guard, World)), Some(Space)) => { + print!("World"); + guard.reply(Exclamation); + } + (Some((_, Exclamation)), Some(World)) => { + print!("!"); + } + (None, Some(Exclamation)) => { // Our Hello World exchange ends right here. print!("\n"); } - } - PacketOwned::Payload(seq_no, reply_no, payload) => { - for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) { - process(guard, packet, send_data) + (Some(a), b) => { + print!("Unsolicited packet received: {:?}", RecvOk::new(Some(a), b)); } + _ => {} } } } @@ -61,7 +52,7 @@ fn main() { let seq2 = MpscSeqEx::default(); // We begin a "Hello World" exchange right here. - seq1.send(&transport1, Packet::Hello); + seq1.send(&transport1, Payload::Hello); receive(&recv2, &seq2, &transport2); receive(&recv1, &seq1, &transport1); diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 689f994..1992e78 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -105,14 +105,47 @@ pub enum Error { WindowIsFull, } -pub enum Packet<'a, SendData> { +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] +pub enum Packet { + Payload(SeqNo, RecvData), + Reply(SeqNo, SeqNo, RecvData), + Ack(SeqNo), +} +impl Packet { + 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), + } + } + 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), + } + } +} +impl Packet<&RecvData> { + pub fn cloned(&self) -> Packet { + self.map(|d| d.clone()) + } +} + +pub enum RecvOkRaw { Payload { - seq_no: SeqNo, - reply_no: Option, - data: &'a SendData, + reply_no: SeqNo, + recv_data: RecvData, + }, + Reply { + reply_no: SeqNo, + recv_data: RecvData, + send_data: SendData, }, Ack { - reply_no: SeqNo, + send_data: SendData, }, } @@ -232,7 +265,7 @@ impl SeqEx { /// 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> { + pub fn try_send_direct(&mut self, packet_data: SendData, current_time: i64) -> Result, SendData> { if self.is_full() { return Err(packet_data); } @@ -247,20 +280,21 @@ impl SeqEx { debug_assert!(slot.is_none()); let entry = slot.insert(SendEntry { seq_no, reply_no: None, next_resend_time, data: packet_data }); - Ok(Packet::Payload { - seq_no: entry.seq_no, - reply_no: entry.reply_no, - data: &entry.data, - }) + Ok(Packet::Payload(entry.seq_no, &entry.data)) } /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn receive_raw_and_direct>( - &mut self, - seq_no: SeqNo, - reply_no: Option, - packet: P, - ) -> Result<(SeqNo, P, Option), DirectError> { + pub fn receive_raw_and_direct>(&mut self, packet: Packet

) -> Result, DirectError> { + 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::Ack(reply_no) => { + return self + .take_send(reply_no) + .map(|send_data| RecvOkRaw::Ack { send_data }) + .ok_or(DirectError::OutOfSequence) + } + }; // 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 @@ -320,10 +354,13 @@ impl SeqEx { self.pre_recv_seq_no = seq_no; self.concurrent_replies[self.concurrent_replies_total] = seq_no; self.concurrent_replies_total += 1; - let data = reply_no.and_then(|r| self.take_send(r)); - Ok((seq_no, packet, data)) + 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: packet.into() }); + self.recv_window[i] = Some(RecvEntry { seq_no, reply_no, data: recv_data.into() }); if is_full { Err(DirectError::WindowIsFull) } else { @@ -332,11 +369,7 @@ impl SeqEx { } } /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn receive_ack(&mut self, reply_no: SeqNo) -> Result { - self.take_send(reply_no).ok_or(Error::OutOfSequence) - } - /// If this returns `Ok` then `try_send` might succeed on next call. - pub fn pump_raw(&mut self) -> Result<(SeqNo, RecvData, Option), Error> { + pub fn pump_raw(&mut self) -> Result, Error> { let next_seq_no = self.pre_recv_seq_no.wrapping_add(1); let i = next_seq_no as usize % self.recv_window.len(); @@ -349,8 +382,11 @@ impl SeqEx { self.pre_recv_seq_no = next_seq_no; self.concurrent_replies[self.concurrent_replies_total] = entry.seq_no; self.concurrent_replies_total += 1; - let data = entry.reply_no.and_then(|r| self.take_send(r)); - Ok((entry.seq_no, entry.data, data)) + 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) } @@ -364,7 +400,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, 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); @@ -382,24 +418,20 @@ impl SeqEx { data: packet_data, }); - Some(Packet::Payload { - seq_no: entry.seq_no, - reply_no: entry.reply_no, - data: &entry.data, - }) + Some(Packet::Reply(entry.seq_no, reply_no, &entry.data)) } else { None } } - pub fn ack_raw_and_direct(&mut self, reply_no: SeqNo) -> Option> { + pub fn ack_raw_and_direct(&mut self, reply_no: SeqNo) -> Option> { if self.remove_reservation(reply_no) { - Some(Packet::Ack { reply_no }) + Some(Packet::Ack(reply_no)) } else { None } } - pub fn service_direct<'a>(&'a mut self, current_time: i64, iter: &mut Option) -> Option> { + pub fn service_direct<'a>(&'a mut self, current_time: i64, iter: &mut Option) -> Option> { 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) { @@ -409,10 +441,10 @@ impl SeqEx { 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); - return Some(Packet::Payload { - seq_no: entry.seq_no, - reply_no: entry.reply_no, - data: &entry.data, + return Some(if let Some(reply_no) = entry.reply_no { + Packet::Reply(entry.seq_no, reply_no, &entry.data) + } else { + Packet::Payload(entry.seq_no, &entry.data) }); } else { iter.next_time = iter.next_time.min(entry.next_resend_time); diff --git a/src/single_thread.rs b/src/single_thread.rs index 15ba365..d9924ac 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,10 +1,10 @@ -use crate::{DirectError, Error, Packet, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; +use crate::{DirectError, Error, Packet, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; -pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>( - &'a mut SeqEx, - Option, - SeqNo, -); +pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { + seq: &'a mut SeqEx, + app: Option, + reply_no: SeqNo, +} 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 /// first reply some identifier, and then `send` all fragments with the same included identifier. @@ -13,25 +13,80 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// the original file. pub fn reply(mut self, packet_data: SendData) { let mut app = None; - core::mem::swap(&mut app, &mut self.1); - self.0.reply_raw(app.unwrap(), self.2, packet_data); + core::mem::swap(&mut app, &mut self.app); + self.seq.reply_raw(app.unwrap(), self.reply_no, packet_data); 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.1 { - if let Some(p) = self.0.ack_raw_and_direct(self.2) { + if let Some(app) = &mut self.app { + if let Some(p) = self.seq.ack_raw_and_direct(self.reply_no) { app.send(p) } } } } +impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ReplyGuard").field("reply_no", &self.reply_no).finish() + } +} -pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { - pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, - pub packet: P, - pub send_data: Option, +#[derive(Debug)] +pub enum RecvOk<'a, TL: TransportLayer, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { + Payload { + reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, + recv_data: P, + }, + Reply { + reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, + recv_data: P, + send_data: SendData, + }, + Ack { + 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 }, + } + } + 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 }, + } + } } impl SeqEx { @@ -48,14 +103,12 @@ impl SeqEx { pub fn receive_raw>( &mut self, mut app: impl TransportLayer, - seq_no: SeqNo, - reply_no: Option, - packet: P, - ) -> Result<(SeqNo, P, Option), Error> { - match self.receive_raw_and_direct(seq_no, reply_no, packet) { + packet: Packet

, + ) -> Result, Error> { + match self.receive_raw_and_direct(packet) { Ok(a) => Ok(a), Err(DirectError::ResendAck(reply_no)) => { - app.send(Packet::Ack { reply_no }); + app.send(Packet::Ack(reply_no)); Err(Error::OutOfSequence) } Err(DirectError::OutOfSequence) => Err(Error::OutOfSequence), @@ -83,22 +136,11 @@ impl SeqEx { pub fn receive, P: Into>( &mut self, app: TL, - seq_no: SeqNo, - reply_no: Option, - packet: P, - ) -> Result, Error> { - self.receive_raw(app.clone(), seq_no, reply_no, packet) - .map(|(reply_no, packet, send_data)| RecvSuccess { - guard: ReplyGuard(self, Some(app), reply_no), - packet, - send_data, - }) + packet: Packet

, + ) -> 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> { - self.pump_raw().map(|(reply_no, packet, send_data)| RecvSuccess { - guard: ReplyGuard(self, Some(app), reply_no), - packet, - send_data, - }) + pub fn pump>(&mut self, app: TL) -> Result, Error> { + self.pump_raw().map(|r| RecvOk::from_raw(self, app, r)) } } diff --git a/src/sync.rs b/src/sync.rs index 8830801..69ee58e 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -7,7 +7,7 @@ use std::{ time::Instant, }; -use crate::{Error, Packet, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; +use crate::{Error, Packet, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; pub struct SeqExSync { seq_ex: Mutex<(SeqEx, usize)>, @@ -15,7 +15,7 @@ pub struct SeqExSync } pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { - origin: &'a SeqExSync, + seq: &'a SeqExSync, app: TL, reply_no: SeqNo, } @@ -26,12 +26,12 @@ 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.origin.lock(); + let mut seq = self.seq.lock(); seq.reply_raw(self.app.clone(), self.reply_no, packet_data); core::mem::forget(self); } pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { - let mut seq = self.origin.lock(); + 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)); core::mem::forget(self); @@ -39,21 +39,92 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { - let mut seq = self.origin.lock(); + let mut seq = self.seq.lock(); seq.ack_raw(self.app.clone(), self.reply_no); } } - -pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { - pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, - pub packet: P, - pub send_data: Option, +impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ReplyGuard").field("reply_no", &self.reply_no).finish() + } } -pub struct ReplyIter<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { +pub enum RecvOk<'a, TL: TransportLayer, P, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { + Payload { + reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, + recv_data: P, + }, + Reply { + reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, + recv_data: P, + send_data: SendData, + }, + Ack { + 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 }, + } + } +} + +pub struct ReplyIter<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { origin: Option<&'a SeqExSync>, app: TL, - first: Option>, + first: Option>, } pub struct SeqExGuard<'a, SendData, RecvData, const CAP: usize>(MutexGuard<'a, (SeqEx, usize)>); @@ -81,42 +152,30 @@ impl SeqExSync { pub fn receive, P: Into>( &self, app: TL, - seq_no: SeqNo, - reply_no: Option, - packet: P, - ) -> Result, Error> { + packet: Packet

, + ) -> Result, Error> { let mut seq = self.seq_ex.lock().unwrap(); - let ret = seq.0.receive_raw(app.clone(), seq_no, reply_no, packet); + let ret = seq.0.receive_raw(app.clone(), packet); if seq.1 > 0 && ret.is_ok() { self.send_block.notify_one(); } - ret.map(|(reply_no, packet, send_data)| RecvSuccess { - guard: ReplyGuard { origin: self, app, reply_no }, - packet, - send_data, - }) + ret.map(|r| RecvOk::from_raw(self, app, r)) } - pub fn pump>(&self, app: TL) -> Result, Error> { + 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(); } - ret.map(|(reply_no, packet, send_data)| RecvSuccess { - guard: ReplyGuard { origin: self, app, reply_no }, - packet, - send_data, - }) + ret.map(|r| RecvOk::from_raw(self, app, r)) } - pub fn receive_all>( + pub fn receive_all, P: Into>( &self, app: TL, - seq_no: SeqNo, - reply_no: Option, - packet: RecvData, - ) -> ReplyIter<'_, TL, SendData, RecvData, CAP> { - if let Ok(g) = self.receive(app.clone(), seq_no, reply_no, packet) { - ReplyIter { origin: Some(self), app, first: Some(g) } + 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) } } else { ReplyIter { origin: None, app, first: None } } @@ -151,15 +210,6 @@ impl SeqExSync { let seq_no = seq.0.seq_no(); self.send_inner(seq, app, packet_data(seq_no)) } - - pub fn receive_ack(&self, reply_no: SeqNo) -> Result { - let mut seq = self.seq_ex.lock().unwrap(); - let ret = seq.0.receive_ack(reply_no); - if seq.1 > 0 && ret.is_ok() { - self.send_block.notify_one(); - } - ret - } pub fn service>(&self, app: TL) -> i64 { self.lock().service(app) } @@ -174,11 +224,18 @@ impl Default for SeqExSync, SendData, RecvData, const CAP: usize> Iterator for ReplyIter<'a, TL, SendData, RecvData, CAP> { - type Item = RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>; +impl<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize> ReplyIter<'a, TL, P, SendData, RecvData, CAP> { + pub fn take_first(&mut self) -> Option> { + self.first.take() + } +} +impl<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize> Iterator + for ReplyIter<'a, TL, P, SendData, RecvData, CAP> +{ + type Item = RecvOk<'a, TL, RecvData, SendData, RecvData, CAP>; fn next(&mut self) -> Option { if let Some(g) = self.first.take() { - Some(g) + Some(g.into()) } else if let Some(origin) = self.origin { origin.pump(self.app.clone()).ok() } else { @@ -187,35 +244,20 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Ite } } -#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] -#[derive(Clone)] -pub enum PacketOwned { - Payload(SeqNo, Option, Payload), - Ack(SeqNo), -} -impl<'a, Payload: Clone> From> for PacketOwned { - fn from(value: Packet<'a, Payload>) -> Self { - match value { - Packet::Payload { seq_no, reply_no, data } => PacketOwned::Payload(seq_no, reply_no, data.clone()), - Packet::Ack { reply_no } => PacketOwned::Ack(reply_no), - } - } -} - -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct MpscTransport { - pub channel: Sender>, + pub channel: Sender>, pub time: Instant, } pub type MpscGuard<'a, Packet> = ReplyGuard<'a, &'a MpscTransport, Packet, Packet>; pub type MpscSeqEx = SeqExSync; impl MpscTransport { - pub fn new() -> (Self, Receiver>) { + pub fn new() -> (Self, Receiver>) { let (send, recv) = channel(); (Self { channel: send, time: std::time::Instant::now() }, recv) } - pub fn from_sender(send: Sender>) -> Self { + pub fn from_sender(send: Sender>) -> Self { Self { channel: send, time: std::time::Instant::now() } } } @@ -224,7 +266,7 @@ impl TransportLayer for &MpscTransport { self.time.elapsed().as_millis() as i64 } - fn send(&mut self, packet: Packet<'_, Payload>) { - let _ = self.channel.send(PacketOwned::from(packet)); + fn send(&mut self, packet: Packet<&Payload>) { + let _ = self.channel.send(packet.cloned()); } } diff --git a/src/transport_layer.rs b/src/transport_layer.rs index 62c19dc..ef67169 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -9,5 +9,5 @@ use crate::Packet; pub trait TransportLayer: Clone { fn time(&mut self) -> i64; - fn send(&mut self, packet: Packet<'_, SendData>); + fn send(&mut self, packet: Packet<&SendData>); } From 55b19e3af8c95c55d152df52c2ca025eff2096f4 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Fri, 18 Aug 2023 11:27:10 -0400 Subject: [PATCH 7/9] added mutual exclusion packets --- src/seq_queue.rs | 234 +++++++++++++++++++++++++++++++------------ src/single_thread.rs | 151 +++++++++++++++++++--------- src/sync.rs | 167 +++++++++++++----------------- 3 files changed, 343 insertions(+), 209 deletions(-) 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 From bdd85155a35a103e4709edfad563f7286fb93e62 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Fri, 18 Aug 2023 12:00:25 -0400 Subject: [PATCH 8/9] finalized new api --- examples/file_download.rs | 2 +- src/seq_queue.rs | 2 +- src/single_thread.rs | 34 +++++++++++++++-------- src/sync.rs | 57 ++++++++++++++++++++++++++++----------- 4 files changed, 66 insertions(+), 29 deletions(-) diff --git a/examples/file_download.rs b/examples/file_download.rs index 0319dfe..5356b69 100644 --- a/examples/file_download.rs +++ b/examples/file_download.rs @@ -70,7 +70,7 @@ fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload, Payl let mut i = 0; while i < file.len() { let j = file.len().min(i + FILE_CHUNK_SIZE); - seqex.send(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() }); + seqex.send_locked(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() }); i = j; } } diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 8ff0763..9f4f081 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -88,7 +88,7 @@ pub enum DirectError { /// 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. + /// it could cause the send window to overflow. WindowIsFull(Packet), WindowIsLocked(Packet), ResendAck(SeqNo), diff --git a/src/single_thread.rs b/src/single_thread.rs index f4c5eb4..7e88d57 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -12,15 +12,24 @@ 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(self, packet_data: SendData) { - self.reply_with(|_, _| packet_data) + pub fn reply(self, packet_data: SendData) { + self.reply_inner(false, |_, _| packet_data) } - pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_locked(self, packet_data: SendData) { + self.reply_inner(true, |_, _| packet_data) + } + pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + self.reply_inner(false, packet_data) + } + pub fn reply_locked_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + self.reply_inner(true, packet_data) + } + fn reply_inner(mut self, locked: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { let mut app = None; core::mem::swap(&mut app, &mut self.app); 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)); + .reply_raw(app.unwrap(), self.reply_no, self.locked, locked, packet_data(seq_no, self.reply_no)); core::mem::forget(self); } } @@ -45,7 +54,7 @@ pub enum Error { /// 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. + /// it could cause the send window to overflow. WindowIsFull(Packet), WindowIsLocked(Packet), } @@ -133,18 +142,20 @@ 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> { + pub fn try_send(&mut self, mut app: impl TransportLayer, locked: bool, packet_data: SendData) -> Result<(), SendData> { match self.try_send_direct(packet_data, app.time()) { - Ok(p) => { + Ok(mut p) => { + p.set_locking(locked); app.send(p); Ok(()) } Err(e) => Err(e), } } - pub fn try_send_with(&mut self, mut app: impl TransportLayer, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), ()> { + pub fn try_send_with(&mut self, mut app: impl TransportLayer, locked: bool, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), ()> { match self.try_send_direct_with(packet_data, app.time()) { - Ok(p) => { + Ok(mut p) => { + p.set_locking(locked); app.send(p); Ok(()) } @@ -168,8 +179,9 @@ impl SeqEx { Err(DirectError::WindowIsLocked(p)) => Err(Error::WindowIsLocked(p)), } } - 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()) { + pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, unlock: bool, locked_packet: bool, packet_data: SendData) { + if let Some(mut p) = self.reply_raw_and_direct(reply_no, unlock, packet_data, app.time()) { + p.set_locking(locked_packet); app.send(p) } } diff --git a/src/sync.rs b/src/sync.rs index 6f62973..ebb2fd6 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -12,7 +12,7 @@ use crate::{Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_ pub struct SeqExSync { seq_ex: Mutex<(SeqEx, usize)>, send_block: Condvar, - lock: Condvar, + recv_lock: Condvar, } pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { @@ -27,15 +27,28 @@ 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(self, packet_data: SendData) { - self.reply_with(|_, _| packet_data) + pub fn reply(self, packet_data: SendData) { + self.reply_inner(false, |_, _| packet_data) } - pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_locked(self, packet_data: SendData) { + self.reply_inner(true, |_, _| packet_data) + } + pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + self.reply_inner(false, packet_data) + } + pub fn reply_locked_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + self.reply_inner(true, packet_data) + } + fn reply_inner(mut self, locked: bool, 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(app.unwrap(), self.reply_no, self.locked, packet_data(seq_no, self.reply_no)); + seq.reply_raw(app.unwrap(), self.reply_no, self.locked, locked, packet_data(seq_no, self.reply_no)); + drop(seq); + if self.locked { + self.seq.recv_lock.notify_all(); + } core::mem::forget(self); } } @@ -106,7 +119,7 @@ impl SeqExSync { Self { seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)), send_block: Condvar::default(), - lock: Condvar::default(), + recv_lock: Condvar::default(), } } @@ -127,7 +140,7 @@ impl SeqExSync { 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(); + seq = self.recv_lock.wait(seq).unwrap(); packet = p; } } @@ -146,7 +159,7 @@ impl SeqExSync { Err(PumpError::OutOfSequence) => return Err(Error::OutOfSequence), Err(PumpError::WindowIsFull) => return Err(Error::WindowIsFull), Err(PumpError::WindowIsLocked) => { - seq = self.lock.wait(seq).unwrap(); + seq = self.recv_lock.wait(seq).unwrap(); } } } @@ -162,30 +175,42 @@ impl SeqExSync { ReplyIter { seq: None, app, first: None } } } - pub fn try_send>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> { - self.try_send_with(app, |_| packet_data) + pub fn try_send>(&self, app: TL, locked: bool, packet_data: SendData) -> Result<(), SendData> { + self.try_send_with(app, locked, |_| packet_data) } - pub fn send_with>(&self, app: TL, mut packet_data: impl FnMut(SeqNo) -> SendData) { + fn send_with_inner>(&self, app: TL, locked: bool, 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) { + while let Err(()) = seq.0.try_send_with(app.clone(), locked, &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) { + fn send_inner>(&self, app: TL, locked: bool, mut packet_data: SendData) { let mut seq = self.seq_ex.lock().unwrap(); - while let Err(p) = seq.0.try_send(app.clone(), packet_data) { + while let Err(p) = seq.0.try_send(app.clone(), locked, packet_data) { packet_data = p; seq.1 += 1; seq = self.send_block.wait(seq).unwrap(); seq.1 -= 1; } } - pub fn try_send_with>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> { + pub fn send(&self, app: impl TransportLayer, packet_data: SendData) { + self.send_inner(app, false, packet_data) + } + pub fn send_locked(&self, app: impl TransportLayer, packet_data: SendData) { + self.send_inner(app, true, packet_data) + } + pub fn send_with(&self, app: impl TransportLayer, packet_data: impl FnMut(SeqNo) -> SendData) { + self.send_with_inner(app, false, packet_data) + } + pub fn send_locked_with(&self, app: impl TransportLayer, packet_data: impl FnMut(SeqNo) -> SendData) { + self.send_with_inner(app, true, packet_data) + } + pub fn try_send_with>(&self, app: TL, locked: bool, 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)) + seq.try_send(app, locked, packet_data(seq_no)) } pub fn service>(&self, app: TL) -> i64 { self.lock().service(app) From 88618287afd0e4aba617612bb2bb7c4783be5b13 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Fri, 18 Aug 2023 12:00:34 -0400 Subject: [PATCH 9/9] cargo fmt --- src/single_thread.rs | 9 +++++++-- src/sync.rs | 9 +++++++-- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/src/single_thread.rs b/src/single_thread.rs index 7e88d57..813754a 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -12,7 +12,7 @@ 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(self, packet_data: SendData) { + pub fn reply(self, packet_data: SendData) { self.reply_inner(false, |_, _| packet_data) } pub fn reply_locked(self, packet_data: SendData) { @@ -152,7 +152,12 @@ impl SeqEx { Err(e) => Err(e), } } - pub fn try_send_with(&mut self, mut app: impl TransportLayer, locked: bool, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), ()> { + pub fn try_send_with( + &mut self, + mut app: impl TransportLayer, + locked: bool, + packet_data: impl FnOnce(SeqNo) -> SendData, + ) -> Result<(), ()> { match self.try_send_direct_with(packet_data, app.time()) { Ok(mut p) => { p.set_locking(locked); diff --git a/src/sync.rs b/src/sync.rs index ebb2fd6..ecdf918 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -27,7 +27,7 @@ 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(self, packet_data: SendData) { + pub fn reply(self, packet_data: SendData) { self.reply_inner(false, |_, _| packet_data) } pub fn reply_locked(self, packet_data: SendData) { @@ -207,7 +207,12 @@ impl SeqExSync { pub fn send_locked_with(&self, app: impl TransportLayer, packet_data: impl FnMut(SeqNo) -> SendData) { self.send_with_inner(app, true, packet_data) } - pub fn try_send_with>(&self, app: TL, locked: bool, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> { + pub fn try_send_with>( + &self, + app: TL, + locked: bool, + packet_data: impl FnOnce(SeqNo) -> SendData, + ) -> Result<(), SendData> { let mut seq = self.lock(); let seq_no = seq.seq_no(); seq.try_send(app, locked, packet_data(seq_no))