From f35113504361b829091a289c753e0b0b6c7fcc28 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Wed, 9 Aug 2023 07:43:38 -0400 Subject: [PATCH] added to API --- examples/calculator.rs | 12 +++---- examples/hello_world.rs | 14 ++++---- src/seq_queue.rs | 71 +++++++++++++++++++++++++---------------- src/single_thread.rs | 14 +++++--- src/sync.rs | 26 +++++++++------ src/transport_layer.rs | 10 +++--- 6 files changed, 88 insertions(+), 59 deletions(-) diff --git a/examples/calculator.rs b/examples/calculator.rs index aa89dba..8022289 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -4,7 +4,7 @@ use std::{ time::Duration, }; -use seq_ex::sync::{MpscTransport, PacketType, ReplyGuard, SeqExSync}; +use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync}; #[derive(Clone)] enum Packet { @@ -52,9 +52,9 @@ fn receive<'a>( let result = seq.receive_empty_reply(reply_no); result.is_some() } - Ok(PacketType::Data { seq_no, reply_no, payload }) => { - if let Ok((guard, recv_packet, send_packet)) = seq.receive(transport, seq_no, reply_no, payload) { - process(guard, recv_packet, send_packet, value); + Ok(PacketType::Payload { seq_no, reply_no, payload }) => { + if let Ok(RecvSuccess { guard, packet, send_data }) = seq.receive(transport, seq_no, reply_no, payload) { + process(guard, packet, send_data, value); true } else { false @@ -63,8 +63,8 @@ fn receive<'a>( _ => return, }; if do_pump { - while let Ok((guard, recv_packet, send_packet)) = seq.pump(transport) { - process(guard, recv_packet, send_packet, value); + while let Ok(RecvSuccess { guard, packet, send_data }) = seq.pump(transport) { + process(guard, packet, send_data, value); } } } diff --git a/examples/hello_world.rs b/examples/hello_world.rs index ba2ae46..c3f3774 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -1,6 +1,6 @@ use std::sync::mpsc::Receiver; -use seq_ex::sync::{MpscTransport, PacketType, ReplyGuard, SeqExSync}; +use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync}; #[derive(Clone, Debug)] enum Packet { @@ -45,15 +45,15 @@ fn receive<'a>(recv: &Receiver>, seq: &SeqExSync<&'a MpscTran } PacketType::EmptyReply { reply_no } => { let result = seq.receive_empty_reply(reply_no); - if let Some(Exclamation) = result { + if let Some(Exclamation) = &result { // Our Hello World exchange ends right here. print!("\n"); } result.is_some() } - PacketType::Data { seq_no, reply_no, payload } => { - if let Ok((guard, recv_packet, send_packet)) = seq.receive(transport, seq_no, reply_no, payload) { - process(guard, recv_packet, send_packet); + PacketType::Payload { seq_no, reply_no, payload } => { + if let Ok(RecvSuccess { guard, packet, send_data }) = seq.receive(transport, seq_no, reply_no, payload) { + process(guard, packet, send_data); true } else { false @@ -61,8 +61,8 @@ fn receive<'a>(recv: &Receiver>, seq: &SeqExSync<&'a MpscTran } }; if do_pump { - while let Ok((guard, recv_packet, send_packet)) = seq.pump(transport) { - process(guard, recv_packet, send_packet); + while let Ok(RecvSuccess { guard, packet, send_data }) = seq.pump(transport) { + process(guard, packet, send_data); } } } diff --git a/src/seq_queue.rs b/src/seq_queue.rs index cd80297..a0df572 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -55,6 +55,7 @@ pub struct SeqEx>; SLEN], @@ -72,7 +73,7 @@ struct RecvEntry { struct SendEntry { seq_no: SeqNo, reply_no: Option, - next_resent_time: i64, + next_resend_time: i64, data: TL::SendData, } @@ -116,6 +117,7 @@ impl SeqEx { pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self { Self { resend_interval: retry_interval, + next_service_timestamp: i64::MIN, next_send_seq_no: initial_seq_no, pre_recv_seq_no: initial_seq_no.wrapping_sub(1), recv_window: core::array::from_fn(|_| None), @@ -159,18 +161,25 @@ 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, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> { + pub fn try_send(&mut self, mut app: TL, packet_data: TL::SendData) -> Result<(), TL::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 next_resent_time = app.time() + self.resend_interval; - let entry = self.send_window[seq_no as usize % self.send_window.len()].insert(SendEntry { + 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; + app.update_service_time(current_time, next_resend_time); + } + let i = seq_no as usize % self.send_window.len(); + debug_assert!(self.send_window[i].is_none()); + let entry = self.send_window[i].insert(SendEntry { seq_no, reply_no: None, - next_resent_time, + next_resend_time, data: packet_data, }); @@ -180,7 +189,7 @@ impl SeqEx { pub fn receive_raw>( &mut self, - app: TL, + mut app: TL, seq_no: SeqNo, reply_no: Option, packet: P, @@ -256,7 +265,7 @@ impl SeqEx { let i = reply_no as usize % self.send_window.len(); if let Some(entry) = self.send_window[i].as_mut() { if entry.seq_no == reply_no { - entry.next_resent_time = i64::MAX; + entry.next_resend_time = i64::MAX; } } } @@ -298,41 +307,30 @@ impl SeqEx { } } - pub fn service(&mut self, app: TL) -> i64 { - let current_time = app.time(); - let next_interval = current_time + self.resend_interval; - let mut next_activity = next_interval; - for item in self.send_window.iter_mut() { - if let Some(entry) = item { - if entry.next_resent_time <= current_time { - entry.next_resent_time = next_interval; - app.send(entry.seq_no, entry.reply_no, &entry.data); - } else { - next_activity = next_activity.min(entry.next_resent_time); - } - } - } - next_activity - current_time - } - - pub fn reply_raw(&mut self, app: TL, reply_no: SeqNo, packet_data: TL::SendData) { + pub fn reply_raw(&mut self, mut app: TL, reply_no: SeqNo, packet_data: TL::SendData) { if self.remove_reservation(reply_no) { let seq_no = self.next_send_seq_no; self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1); let i = seq_no as usize % self.send_window.len(); - let next_resent_time = app.time() + self.resend_interval; + 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; + app.update_service_time(current_time, next_resend_time); + } + debug_assert!(self.send_window[i].is_none()); let entry = self.send_window[i].insert(SendEntry { seq_no, reply_no: Some(reply_no), - next_resent_time, + next_resend_time, data: packet_data, }); app.send(entry.seq_no, entry.reply_no, &entry.data); } } - pub fn reply_empty_raw(&mut self, app: TL, reply_no: SeqNo) { + pub fn reply_empty_raw(&mut self, mut app: TL, reply_no: SeqNo) { if self.remove_reservation(reply_no) { app.send_empty_reply(reply_no); } @@ -348,6 +346,23 @@ impl SeqEx { false } + pub fn service(&mut self, mut app: TL) -> i64 { + let current_time = app.time(); + let next_interval = 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_interval; + app.send(entry.seq_no, entry.reply_no, &entry.data); + } else { + next_activity = next_activity.min(entry.next_resend_time); + } + } + self.next_service_timestamp = next_activity; + app.update_service_time(current_time, next_activity); + self.resend_interval.min(next_activity - current_time) + } + pub fn iter(&self) -> Iter<'_, TL> { Iter(self.send_window.iter()) } diff --git a/src/single_thread.rs b/src/single_thread.rs index b2664bb..d8cce40 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -19,6 +19,12 @@ impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { } } +pub struct RecvSuccess<'a, TL: TransportLayer, P> { + pub guard: ReplyGuard<'a, TL>, + pub packet: P, + pub send_data: Option, +} + impl SeqEx { pub fn receive>( &mut self, @@ -26,12 +32,12 @@ impl SeqEx { seq_no: SeqNo, reply_no: Option, packet: P, - ) -> Result<(ReplyGuard<'_, TL>, P, Option), Error> { + ) -> Result, Error> { self.receive_raw(app.clone(), seq_no, reply_no, packet) - .map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } - pub fn pump(&mut self, app: TL) -> Result<(ReplyGuard<'_, TL>, TL::RecvData, Option), Error> { + pub fn pump(&mut self, app: TL) -> Result, Error> { self.pump_raw() - .map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } } diff --git a/src/sync.rs b/src/sync.rs index 3402553..d3d3585 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -35,6 +35,12 @@ impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { } } +pub struct RecvSuccess<'a, TL: TransportLayer, P> { + pub guard: ReplyGuard<'a, TL>, + pub packet: P, + pub send_data: Option, +} + impl SeqExSync { pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self { Self { @@ -50,17 +56,17 @@ impl SeqExSync { seq_no: SeqNo, reply_no: Option, packet: P, - ) -> Result<(ReplyGuard<'_, TL>, P, Option), Error> { + ) -> Result, Error> { let mut seq = self.lock(); let ret = seq.receive_raw(app.clone(), seq_no, reply_no, packet); self.unblock(ret.is_ok()); - ret.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } - pub fn pump(&self, app: TL) -> Result<(ReplyGuard<'_, TL>, TL::RecvData, Option), Error> { + pub fn pump(&self, app: TL) -> Result, Error> { let mut seq = self.lock(); let ret = seq.pump_raw(); self.unblock(ret.is_ok()); - ret.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } #[inline] fn unblock(&self, is_ok: bool) { @@ -109,7 +115,7 @@ impl Default for SeqExSync { #[derive(Clone)] pub enum PacketType { - Data { + Payload { seq_no: SeqNo, reply_no: Option, payload: Payload, @@ -140,17 +146,17 @@ impl TransportLayer for &MpscTransport { type RecvData = Payload; type SendData = Payload; - fn time(&self) -> i64 { + fn time(&mut self) -> i64 { self.time.elapsed().as_millis() as i64 } - fn send(&self, seq_no: SeqNo, reply_no: Option, payload: &Payload) { - let _ = self.channel.send(PacketType::Data { seq_no, reply_no, payload: payload.clone() }); + fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &Payload) { + let _ = self.channel.send(PacketType::Payload { seq_no, reply_no, payload: payload.clone() }); } - fn send_ack(&self, reply_no: SeqNo) { + fn send_ack(&mut self, reply_no: SeqNo) { let _ = self.channel.send(PacketType::Ack { reply_no }); } - fn send_empty_reply(&self, reply_no: SeqNo) { + fn send_empty_reply(&mut self, reply_no: SeqNo) { let _ = self.channel.send(PacketType::EmptyReply { reply_no }); } } diff --git a/src/transport_layer.rs b/src/transport_layer.rs index 254a3dd..482aaa2 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -10,9 +10,11 @@ pub trait TransportLayer: Sized + Clone { type RecvData; type SendData; - fn time(&self) -> i64; + fn time(&mut self) -> i64; + #[allow(unused)] + fn update_service_time(&mut self, current_time: i64, timestamp: i64) {} - fn send(&self, seq_no: SeqNo, reply_no: Option, payload: &Self::SendData); - fn send_ack(&self, reply_no: SeqNo); - fn send_empty_reply(&self, reply_no: SeqNo); + fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &Self::SendData); + fn send_ack(&mut self, reply_no: SeqNo); + fn send_empty_reply(&mut self, reply_no: SeqNo); }