From 829d8e5963d8ba9e369c63b4075eb137f0831905 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Fri, 28 Jul 2023 12:44:44 -0400 Subject: [PATCH] refactored and added helpers --- Cargo.lock | 2 +- Cargo.toml | 6 +++- src/lib.rs | 9 +++++- src/multi_thread.rs | 62 ++++++++++++++++++++++++++++++++++++++++++ src/seq_queue.rs | 31 +++++---------------- src/single_thread.rs | 31 ++++++++++++++++----- src/transport_layer.rs | 18 ++++++++++++ 7 files changed, 125 insertions(+), 34 deletions(-) create mode 100644 src/multi_thread.rs create mode 100644 src/transport_layer.rs diff --git a/Cargo.lock b/Cargo.lock index 15fd827..0872f66 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,4 +4,4 @@ version = 3 [[package]] name = "seq_ex" -version = "0.0.1" +version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index c7ff57d..160fabd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "seq_ex" -version = "0.0.1" +version = "0.1.0" authors = ["Monica Moniot"] edition = "2021" @@ -8,3 +8,7 @@ edition = "2021" name = "seq_ex" path = "src/lib.rs" doc = true + +[features] +default = ["std"] +std = [] diff --git a/src/lib.rs b/src/lib.rs index 699295e..5c18faf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,7 +1,14 @@ -#![no_std] +//#![no_std] +//#![warn(missing_docs, rust_2018_idioms)] + +mod transport_layer; +pub use transport_layer::*; mod seq_queue; pub use seq_queue::*; mod single_thread; pub use single_thread::*; + +#[cfg(feature = "std")] +pub mod multi_thread; diff --git a/src/multi_thread.rs b/src/multi_thread.rs new file mode 100644 index 0000000..6b8b370 --- /dev/null +++ b/src/multi_thread.rs @@ -0,0 +1,62 @@ +use std::sync::{Mutex, MutexGuard}; + +use crate::{Error, SeqEx, SeqNo, TransportLayer}; + +pub struct SeqExLock(pub Mutex>); + +pub struct ReplyGuard<'a, TL: TransportLayer>(&'a SeqExLock, TL, SeqNo); +impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { + pub fn reply(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) { + let mut seq = self.0 .0.lock().unwrap(); + let p = packet_data(seq.seq_no(), self.2); + seq.reply_raw(self.1.clone(), self.2, p); + core::mem::forget(self); + } +} +impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { + fn drop(&mut self) { + let mut seq = self.0 .0.lock().unwrap(); + seq.reply_empty_raw(self.1.clone(), self.2); + } +} + +impl SeqExLock { + pub fn receive, T>( + &self, + app: TL, + seq_no: SeqNo, + reply_no: Option, + packet: P, + ) -> Result<(ReplyGuard<'_, TL>, P, Option), Error> { + let mut seq = self.lock(); + seq.receive_raw(app.clone(), seq_no, reply_no, packet) + .map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + } + pub fn pump(&self, app: TL) -> Result<(ReplyGuard<'_, TL>, TL::RecvData, Option), Error> { + let mut seq = self.lock(); + seq.pump_raw() + .map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + } + + pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self { + Self(Mutex::new(SeqEx::new(retry_interval, initial_seq_no))) + } + pub fn send(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> TL::SendData) -> bool { + let mut seq = self.lock(); + let p = packet_data(seq.seq_no()); + seq.send(app, p) + } + pub fn receive_ack(&self, reply_no: SeqNo) { + self.lock().receive_ack(reply_no) + } + pub fn receive_empty_reply(&self, reply_no: SeqNo) -> Option { + self.lock().receive_empty_reply(reply_no) + } + pub fn service(&self, app: TL) -> i64 { + self.lock().service(app) + } + + pub fn lock(&self) -> MutexGuard> { + self.0.lock().unwrap() + } +} diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 5331f11..275e5ce 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -33,32 +33,15 @@ //! //! ## Examples //! -#![forbid(unsafe_code)] -//#![warn(missing_docs, rust_2018_idioms)] -const MAX_CONCURRENCY: usize = 24; + +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. pub type SeqNo = u32; -/// A trait for giving an instance of SeqEx access to the transport layer. -/// -/// The implementor is free to choose how to define the generic types based on how they want to -/// manage memory. -/// It is possible through these generics to implement SeqEx to be no-alloc and zero-copy, but otherwise -/// a lot of them are most easily implemented as tuples of custom enums and Vec. -pub trait TransportLayer: Sized + Clone { - type RecvData; - type SendData; - - fn time(&self) -> i64; - - fn send(&self, data: &Self::SendData); - fn send_ack(&self, reply_no: SeqNo); - fn send_empty_reply(&self, reply_no: SeqNo); -} - +const MAX_CONCURRENCY: usize = 24; pub struct SeqEx { /// The interval at which packets will be resent if they have not yet been acknowledged by the /// remote peer. @@ -187,7 +170,7 @@ impl SeqEx { true } - pub fn receive>( + pub fn receive_raw>( &mut self, app: TL, seq_no: SeqNo, @@ -287,7 +270,7 @@ impl SeqEx { None } } - pub fn pump(&mut self) -> Result<(SeqNo, TL::RecvData, Option), Error> { + pub fn pump_raw(&mut self) -> Result<(SeqNo, TL::RecvData, Option), Error> { let next_seq_no = self.pre_recv_seq_no.wrapping_add(1); let i = next_seq_no as usize % self.recv_window.len(); @@ -324,7 +307,7 @@ impl SeqEx { next_activity - current_time } - pub fn reply(&mut self, app: TL, reply_no: SeqNo, packet_data: TL::SendData) { + pub fn reply_raw(&mut self, 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); @@ -341,7 +324,7 @@ impl SeqEx { app.send(&entry.data); } } - pub fn reply_empty(&mut self, app: TL, reply_no: SeqNo) { + pub fn reply_empty_raw(&mut self, app: TL, reply_no: SeqNo) { if self.remove_reservation(reply_no) { app.send_empty_reply(reply_no); } diff --git a/src/single_thread.rs b/src/single_thread.rs index 67893d7..40a09cf 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,20 +1,37 @@ use crate::{Error, SeqEx, SeqNo, TransportLayer}; pub struct ReplyGuard<'a, TL: TransportLayer>(&'a mut SeqEx, TL, SeqNo); -//impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { -// pub fn reply_no(&self) { - -// } -//} +impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { + pub fn seq_no(&self) -> SeqNo { + self.0.seq_no() + } + pub fn reply_no(&self) -> SeqNo { + self.2 + } + pub fn reply(self, packet_data: TL::SendData) { + self.0.reply_raw(self.1.clone(), self.2, packet_data); + core::mem::forget(self); + } +} +impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { + fn drop(&mut self) { + self.0.reply_empty_raw(self.1.clone(), self.2) + } +} impl SeqEx { - pub fn receive_guarded, T>( + pub fn receive, T>( &mut self, app: TL, seq_no: SeqNo, reply_no: Option, packet: P, ) -> Result<(ReplyGuard<'_, TL>, P, Option), Error> { - self.receive(app.clone(), seq_no, reply_no, packet).map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + self.receive_raw(app.clone(), seq_no, reply_no, packet) + .map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + } + pub fn pump(&mut self, app: TL) -> Result<(ReplyGuard<'_, TL>, TL::RecvData, Option), Error> { + self.pump_raw() + .map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) } } diff --git a/src/transport_layer.rs b/src/transport_layer.rs new file mode 100644 index 0000000..f3615e6 --- /dev/null +++ b/src/transport_layer.rs @@ -0,0 +1,18 @@ +use crate::SeqNo; + +/// A trait for giving an instance of SeqEx access to the transport layer. +/// +/// The implementor is free to choose how to define the generic types based on how they want to +/// manage memory. +/// It is possible through these generics to implement SeqEx to be no-alloc and zero-copy, but otherwise +/// a lot of them are most easily implemented as tuples of custom enums and Vec. +pub trait TransportLayer: Sized + Clone { + type RecvData; + type SendData; + + fn time(&self) -> i64; + + fn send(&self, data: &Self::SendData); + fn send_ack(&self, reply_no: SeqNo); + fn send_empty_reply(&self, reply_no: SeqNo); +}