diff --git a/examples/hello_world_tokio.rs b/examples/hello_world_tokio.rs index 22704a0..14df8a2 100644 --- a/examples/hello_world_tokio.rs +++ b/examples/hello_world_tokio.rs @@ -96,6 +96,7 @@ async fn main() { say_hello(&peer1, &transport1).await; } + #[test] fn test() { main() diff --git a/src/seq_queue.rs b/src/seq_queue.rs index ed56e9e..f98eca3 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -91,6 +91,17 @@ struct SendEntry { next_resend_time: i64, data: SendData, } +impl SendEntry { + fn to_packet(&self) -> Packet<&SendData> { + let mut p = if let Some(reply_no) = self.reply_no { + Reply(self.seq_no, reply_no, &self.data) + } else { + Payload(self.seq_no, &self.data) + }; + p.set_seq_cst(self.seq_cst); + p + } +} #[derive(Debug, Clone, PartialEq, Eq)] pub enum TryRecvError { @@ -249,7 +260,7 @@ pub struct IterMut<'a, SendData>(core::slice::IterMut<'a, Option SeqEx { data: packet_data(seq_no), }); - let mut p = Payload(entry.seq_no, &entry.data); - p.set_seq_cst(seq_cst); - Ok(p) + Ok(entry.to_packet()) } fn fast_forward(&mut self) -> bool { @@ -577,9 +586,7 @@ impl SeqEx { data: packet_data, }); - let mut p = Reply(entry.seq_no, reply_no, &entry.data); - p.set_seq_cst(seq_cst); - Some(p) + Some(entry.to_packet()) } else { None } @@ -601,22 +608,19 @@ impl SeqEx { /// Can mutate `next_service_timestamp`. pub fn service_direct(&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 { + let iter = iter.get_or_insert(ServiceIter { + seq_no: self.next_send_seq_no.wrapping_sub(self.send_window.len() as u32), + next_time: i64::MAX, + }); + while iter.seq_no != self.next_send_seq_no { + let idx = iter.seq_no as usize % self.send_window.len(); + iter.seq_no = iter.seq_no.wrapping_add(1); + if let Some(entry) = &mut self.send_window[idx] { if entry.next_resend_time <= current_time { - let entry = self.send_window[iter.idx - 1].as_mut().unwrap(); entry.next_resend_time = current_time + self.resend_interval; iter.next_time = iter.next_time.min(entry.next_resend_time); - let mut p = if let Some(reply_no) = entry.reply_no { - Reply(entry.seq_no, reply_no, &entry.data) - } else { - Payload(entry.seq_no, &entry.data) - }; - p.set_seq_cst(entry.seq_cst); - return Some(p); + return Some(self.send_window[idx].as_ref().unwrap().to_packet()); } else { iter.next_time = iter.next_time.min(entry.next_resend_time); } diff --git a/src/single_thread.rs b/src/single_thread.rs index da09772..53ddb0a 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -27,13 +27,8 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { let app = self.app.take().expect("Cannot reply after an ack has been sent"); let seq_no = self.seq.seq_no(); - self.seq.reply_raw( - app, - self.reply_no, - self.is_holding_lock, - seq_cst, - packet_data(seq_no, self.reply_no), - ); + self.seq + .reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); core::mem::forget(self); }