From fdc4b6c63760ec11dc6b71e9ac7d05fbab2d6c85 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 17 Aug 2023 16:01:18 -0400 Subject: [PATCH] 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); }