From 79d34d623d7e798f576f96871e1021b78e5d5902 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 17 Aug 2023 10:08:25 -0400 Subject: [PATCH] improved API --- examples/calculator.rs | 5 ++ examples/file_download.rs | 15 +++--- examples/hello_world.rs | 5 ++ src/seq_queue.rs | 27 +++++----- src/single_thread.rs | 15 +++--- src/sync.rs | 101 ++++++++++++++++++++------------------ src/transport_layer.rs | 6 +-- 7 files changed, 92 insertions(+), 82 deletions(-) diff --git a/examples/calculator.rs b/examples/calculator.rs index e3abcc0..63ca26d 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -72,3 +72,8 @@ fn main() { } assert_eq!(value, remote_value); } + +#[test] +fn test() { + main() +} diff --git a/examples/file_download.rs b/examples/file_download.rs index 79052ca..8d2be99 100644 --- a/examples/file_download.rs +++ b/examples/file_download.rs @@ -36,9 +36,7 @@ struct Peer { receiver: Receiver>, } -impl TransportLayer for &Transport { - type SendData = Packet; - +impl TransportLayer for &Transport { fn time(&mut self) -> i64 { self.time.elapsed().as_millis() as i64 } @@ -129,13 +127,13 @@ fn receive(peer: &Peer) { fn main() { let mut filesystem2 = HashMap::new(); - let mut file = Vec::from([0u8; 1 << 16]); + let mut file = vec![0; 1 << 16]; OsRng.fill_bytes(&mut file); filesystem2.insert("File1".to_string(), file); - let mut file = Vec::from([0u8; 1 << 18]); + let mut file = vec![0; 1 << 18]; OsRng.fill_bytes(&mut file); filesystem2.insert("File2".to_string(), file); - let mut file = Vec::from([0u8; 1 << 20]); + let mut file = vec![0; 1 << 20]; OsRng.fill_bytes(&mut file); filesystem2.insert("File3".to_string(), file); @@ -169,3 +167,8 @@ fn main() { assert_eq!(peer1.filesystem.read().unwrap().deref(), peer2.filesystem.read().unwrap().deref()); } + +#[test] +fn test() { + main() +} diff --git a/examples/hello_world.rs b/examples/hello_world.rs index 33e782f..7202448 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -69,3 +69,8 @@ fn main() { receive(&recv1, &seq1, &transport1); receive(&recv2, &seq2, &transport2); } + +#[test] +fn test() { + main() +} diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 57d1cc7..4543d77 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -156,11 +156,11 @@ impl SeqEx { } /// Sends the given packet to the remote peer and adds it to the send window. /// + /// If `Ok` is returned then the packet was successfully sent. + /// /// If the return value is `Err` the queue is full and the packet will not be sent. /// The caller must either cancel sending, abort the connection, or wait until a call to - /// `receive` or `receive_ack` returns `Ok` and try again. - /// - /// If `Ok` is returned then the packet was successfully sent. + /// `receive`, `receive_ack` or `pump` returns `Ok` and try again. /// /// `packet_data` should contain both the packet to be sent as well as any local metadata the /// caller wants to store with the packet. This metadata allows the exchange to be stateful. @@ -171,7 +171,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, mut app: impl TransportLayer, packet_data: SendData) -> Result<(), SendData> { + pub fn try_send(&mut self, mut app: impl TransportLayer, packet_data: SendData) -> Result<(), SendData> { if self.is_full() { return Err(packet_data); } @@ -192,9 +192,10 @@ impl SeqEx { Ok(()) } + /// If this returns `Ok` then `try_send` might succeed on next call. pub fn receive_raw>( &mut self, - mut app: impl TransportLayer, + mut app: impl TransportLayer, seq_no: SeqNo, reply_no: Option, packet: P, @@ -270,14 +271,9 @@ impl SeqEx { } } } + /// If this returns `Ok` then `try_send` might succeed on next call. pub fn receive_ack(&mut self, reply_no: SeqNo) -> Result { - let slot = self.send_window_slot_mut(reply_no); - if slot.as_ref().map_or(false, |e| e.seq_no == reply_no) { - let entry = slot.take().unwrap(); - Ok(entry.data) - } else { - Err(Error::OutOfSequence) - } + self.take_send(reply_no).ok_or(Error::OutOfSequence) } fn is_full_inner(&self, reserve_one: bool) -> bool { @@ -301,6 +297,7 @@ impl SeqEx { 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); let i = next_seq_no as usize % self.recv_window.len(); @@ -329,7 +326,7 @@ 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) { + pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, packet_data: 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); @@ -352,7 +349,7 @@ impl SeqEx { app.send(entry.seq_no, entry.reply_no, &entry.data); } } - pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo) { + pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo) { 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. @@ -371,7 +368,7 @@ impl SeqEx { false } - pub fn service(&mut self, mut app: impl TransportLayer) -> i64 { + pub fn service(&mut self, mut app: impl TransportLayer) -> i64 { let current_time = app.time(); let next_interval = current_time + self.resend_interval; let mut next_activity = i64::MAX; diff --git a/src/single_thread.rs b/src/single_thread.rs index 0699092..0faae3c 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,11 +1,11 @@ use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; -pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>( +pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>( &'a mut SeqEx, TL, SeqNo, ); -impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { +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. /// The identifier will tell the remote peer which packets contain fragments of the file, @@ -16,20 +16,20 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: core::mem::forget(self); } } -impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { +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) } } -pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { +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 SeqEx { - pub fn receive, P: Into>( + pub fn receive, P: Into>( &mut self, app: TL, seq_no: SeqNo, @@ -39,10 +39,7 @@ impl SeqEx { self.receive_raw(app.clone(), 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> { + 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 }) } diff --git a/src/sync.rs b/src/sync.rs index 982d1c0..33a8253 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -1,5 +1,5 @@ use std::{ - cell::UnsafeCell, + ops::{Deref, DerefMut}, sync::{ mpsc::{channel, Receiver, Sender}, Condvar, Mutex, MutexGuard, @@ -10,78 +10,91 @@ use std::{ use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; pub struct SeqExSync { - seq_ex: Mutex>, - /// The mutex above is always held when this value changes, hence it is safe to mutate. - /// We don't pack this as a component of the mutex to avoid having to reimplement MutexGuard. - wait_count: UnsafeCell, + seq_ex: Mutex<(SeqEx, usize)>, send_block: Condvar, } -pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>( +pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>( &'a SeqExSync, TL, SeqNo, ); -impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { +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. /// 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.0.seq_ex.lock().unwrap(); + let mut seq = self.0.lock(); seq.reply_raw(self.1.clone(), 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> { +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.0.seq_ex.lock().unwrap(); + let mut seq = self.0.lock(); seq.ack_raw(self.1.clone(), self.2); } } -pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { +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, } -pub struct ReplyIter<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { +pub struct ReplyIter<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { origin: Option<&'a SeqExSync>, app: TL, first: Option>, } +pub struct SeqExGuard<'a, SendData, RecvData, const CAP: usize>(MutexGuard<'a, (SeqEx, usize)>); +impl<'a, SendData, RecvData, const CAP: usize> Deref for SeqExGuard<'a, SendData, RecvData, CAP> { + type Target = SeqEx; + + fn deref(&self) -> &Self::Target { + &self.0 .0 + } +} +impl<'a, SendData, RecvData, const CAP: usize> DerefMut for SeqExGuard<'a, SendData, RecvData, CAP> { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.0 .0 + } +} + impl SeqExSync { pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self { Self { - seq_ex: Mutex::new(SeqEx::new(retry_interval, initial_seq_no)), - wait_count: UnsafeCell::new(0), + seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)), send_block: Condvar::default(), } } - pub fn receive, P: Into>( + pub fn receive, P: Into>( &self, app: TL, seq_no: SeqNo, reply_no: Option, packet: P, ) -> Result, Error> { - let mut seq = self.lock(); - let ret = seq.receive_raw(app.clone(), seq_no, reply_no, packet); - // TODO: double check blocking. - self.unblock(ret.is_ok()); + let mut seq = self.seq_ex.lock().unwrap(); + 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(); + } ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } - pub fn pump>(&self, app: TL) -> Result, Error> { - let mut seq = self.lock(); - let ret = seq.pump_raw(); - self.unblock(ret.is_ok()); + 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(self, app, reply_no), packet, send_data }) } - pub fn receive_all>( + pub fn receive_all>( &self, app: TL, seq_no: SeqNo, @@ -94,38 +107,34 @@ impl SeqExSync { ReplyIter { origin: None, app, first: None } } } - #[inline] - fn unblock(&self, is_ok: bool) { - let has_waiting = unsafe { *self.wait_count.get() > 0 }; - if has_waiting && is_ok { - self.send_block.notify_one(); - } - } - pub fn try_send>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> { + pub fn try_send>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> { let mut seq = self.lock(); seq.try_send(app, packet_data) } - pub fn send>(&self, app: TL, mut packet_data: SendData) { - let mut seq = self.lock(); - while let Err(p) = seq.try_send(app.clone(), packet_data) { + 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; - unsafe { *self.wait_count.get() += 1 } + seq.1 += 1; seq = self.send_block.wait(seq).unwrap(); - unsafe { *self.wait_count.get() -= 1 } + seq.1 -= 1; } } pub fn receive_ack(&self, reply_no: SeqNo) -> Result { - let ret = self.lock().receive_ack(reply_no); - self.unblock(ret.is_ok()); + 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 { + pub fn service>(&self, app: TL) -> i64 { self.lock().service(app) } - pub fn lock(&self) -> MutexGuard> { - self.seq_ex.lock().unwrap() + pub fn lock(&self) -> SeqExGuard<'_, SendData, RecvData, CAP> { + SeqExGuard(self.seq_ex.lock().unwrap()) } } impl Default for SeqExSync { @@ -133,10 +142,8 @@ impl Default for SeqExSync Send for SeqExSync {} -unsafe impl Sync for SeqExSync {} -impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Iterator for ReplyIter<'a, TL, SendData, RecvData, CAP> { +impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Iterator for ReplyIter<'a, TL, SendData, RecvData, CAP> { type Item = RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>; fn next(&mut self) -> Option { if let Some(g) = self.first.take() { @@ -173,9 +180,7 @@ impl MpscTransport { Self { channel: send, time: std::time::Instant::now() } } } -impl TransportLayer for &MpscTransport { - type SendData = Payload; - +impl TransportLayer for &MpscTransport { fn time(&mut self) -> i64 { self.time.elapsed().as_millis() as i64 } diff --git a/src/transport_layer.rs b/src/transport_layer.rs index 00782fe..9d7d9d0 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -6,13 +6,11 @@ use crate::SeqNo; /// manage memory. /// It is possible through these generics to make SeqEx no-alloc and zero-copy, but otherwise /// they are most easily implemented as some combination of custom enums, `Vec` and `Arc<[u8]>`. -pub trait TransportLayer: Clone { - type SendData; - +pub trait TransportLayer: Clone { fn time(&mut self) -> i64; #[allow(unused)] fn update_service_time(&mut self, timestamp: i64, current_time: i64) {} - fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &Self::SendData); + fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &SendData); fn send_ack(&mut self, reply_no: SeqNo); }