From 61e7c04b950237c5869463d4f9caffc264c0ca19 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 24 Aug 2023 11:29:39 -0400 Subject: [PATCH] added the ability to ack early --- src/seq_queue.rs | 2 +- src/single_thread.rs | 26 +++++++++++++++-------- src/sync.rs | 23 +++++++++++++++----- src/tokio.rs | 50 ++++++++++++++++++++++++++++---------------- 4 files changed, 68 insertions(+), 33 deletions(-) diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 1162a4f..a26aaa5 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -50,7 +50,7 @@ pub const DEFAULT_WINDOW_CAP: usize = 64; pub struct SeqEx { /// The interval at which packets will be resent if they have not yet been acknowledged by the /// remote peer. - /// It can be statically or dynamically set, it is up to the user to decide. + /// It can be statically or dynamically set. pub resend_interval: i64, pub next_service_timestamp: i64, next_send_seq_no: SeqNo, diff --git a/src/single_thread.rs b/src/single_thread.rs index d2468de..da09772 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -7,28 +7,40 @@ pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, cons is_holding_lock: bool, } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { + pub fn ack(&mut self) { + if let Some(app) = self.app.take() { + self.seq.ack_raw(app, self.reply_no, self.is_holding_lock); + } + } /// 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. + /// # Panic + /// This function will panic if `ack` has been called. pub fn reply(self, seq_cst: bool, packet_data: SendData) { self.reply_with(seq_cst, |_, _| packet_data) } + /// # Panic + /// This function will panic if `ack` has been called. fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { - let app = self.app.take(); + 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.unwrap(), + app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no), ); + core::mem::forget(self); } - pub fn to_components(mut self) -> (TL, SeqNo, bool) { - (self.app.take().unwrap(), self.reply_no, self.is_holding_lock) + pub fn to_components(self) -> (SeqNo, bool) { + let ret = (self.reply_no, self.is_holding_lock); + core::mem::forget(self); + ret } pub unsafe fn from_components(seq: &'a mut SeqEx, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock } @@ -36,11 +48,7 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { - if let Some(app) = &mut self.app { - if let Some(p) = self.seq.ack_raw_and_direct(self.reply_no, self.is_holding_lock) { - app.send(p) - } - } + self.ack(); } } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> { diff --git a/src/sync.rs b/src/sync.rs index 5b72151..15baafa 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -30,26 +30,39 @@ pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, cons is_holding_lock: bool, } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { + pub fn ack(&mut self) { + if let Some(app) = self.app.take() { + let mut inner = self.seq.inner.lock().unwrap(); + inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock); + } + } + /// # Panic + /// This function will panic if `ack` has been called. pub fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { - let app = self.app.take().unwrap(); + let app = self.app.take().expect("Cannot reply after an ack has been sent"); let mut inner = self.seq.inner.lock().unwrap(); let seq_no = inner.seq.seq_no(); inner .seq .reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); self.seq.notify_reply(inner); + core::mem::forget(self); } /// 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. + /// # Panic + /// This function will panic if `ack` has been called. pub fn reply(self, seq_cst: bool, packet_data: SendData) { self.reply_with(seq_cst, |_, _| packet_data) } - pub fn to_components(mut self) -> (TL, SeqNo, bool) { - (self.app.take().unwrap(), self.reply_no, self.is_holding_lock) + pub fn to_components(self) -> (SeqNo, bool) { + let ret = (self.reply_no, self.is_holding_lock); + core::mem::forget(self); + ret } pub unsafe fn from_components(seq: &'a SeqExSync, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock } @@ -57,11 +70,11 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { + let mut inner = self.seq.inner.lock().unwrap(); if let Some(app) = self.app.take() { - let mut inner = self.seq.inner.lock().unwrap(); inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock); - self.seq.notify_reply(inner); } + self.seq.notify_reply(inner); } } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> { diff --git a/src/tokio.rs b/src/tokio.rs index 49ab326..4460337 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -12,7 +12,7 @@ type Sender = (oneshot::Sender = (oneshot::Sender<(SeqNo, bool, RecvData)>, RecvData); pub struct SeqExTokio { - seq_ex: Mutex>, + inner: Mutex>, wait_on_recv: Notify, wait_on_reply: Notify, update_queue: mpsc::Sender, @@ -35,7 +35,7 @@ impl<'a, TL: TokioLayer, SendData, RecvData, const CAP: usi ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock } } fn try_reply_with_inner(&mut self, app: TL, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> Sender) -> Option { - let mut inner = self.seq.seq_ex.lock().unwrap(); + let mut inner = self.seq.inner.lock().unwrap(); let seq_no = inner.seq.seq_no(); let pre_ts = inner.seq.next_service_timestamp; @@ -47,33 +47,47 @@ impl<'a, TL: TokioLayer, SendData, RecvData, const CAP: usi self.seq.notify_reply(inner); ret } - ///// 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 ack(&mut self) { + if let Some(app) = self.app.take() { + let mut inner = self.seq.inner.lock().unwrap(); + inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock); + } + } + /// 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. + /// # Panic + /// This function will panic if `ack` has been called. pub async fn reply(self, seq_cst: bool, packet_data: SendData) -> Result<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, RecvData), AsyncError> { self.reply_with(seq_cst, |_, _| packet_data).await } + /// # Panic + /// This function will panic if `ack` has been called. pub async fn reply_with( mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData, ) -> Result<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, RecvData), AsyncError> { - let app = self.app.take().unwrap(); + let app = self.app.take().expect("Cannot reply after an ack has been sent"); let (tx, rx) = oneshot::channel(); let update_ts = self.try_reply_with_inner(app.clone(), seq_cst, |s, r| (tx, packet_data(s, r))); + let seq = self.seq; + core::mem::forget(self); if let Some(update_ts) = update_ts { - let _ = self.seq.update_queue.send(update_ts).await; + let _ = seq.update_queue.send(update_ts).await; } let (reply_no, seq_cst, recv_data) = rx.await.map_err(|_| AsyncError::SeqExClosed)?.ok_or(AsyncError::EndOfExchange)?; - Ok((Self::new(self.seq, app, reply_no, seq_cst), recv_data)) + Ok((Self::new(seq, app, reply_no, seq_cst), recv_data)) } - pub fn to_components(mut self) -> (TL, SeqNo, bool) { - (self.app.take().unwrap(), self.reply_no, self.is_holding_lock) + pub fn to_components(self) -> (SeqNo, bool) { + let ret = (self.reply_no, self.is_holding_lock); + core::mem::forget(self); + ret } pub unsafe fn from_components(seq: &'a SeqExTokio, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { Self::new(seq, app, reply_no, is_holding_lock) @@ -81,11 +95,11 @@ impl<'a, TL: TokioLayer, SendData, RecvData, const CAP: usi } impl<'a, TL: TokioLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { + let mut inner = self.seq.inner.lock().unwrap(); if let Some(app) = self.app.take() { - let mut inner = self.seq.seq_ex.lock().unwrap(); inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock); - self.seq.notify_reply(inner); } + self.seq.notify_reply(inner); } } impl<'a, TL: TokioLayer, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> { @@ -150,7 +164,7 @@ impl SeqExTokio { let (update_queue, recv_service_update) = mpsc::channel(8); ( Self { - seq_ex: Mutex::new(SeqExInner { + inner: Mutex::new(SeqExInner { seq: SeqEx::new(retry_interval, initial_seq_no), recv_waiters: 0, reply_waiters: false, @@ -178,7 +192,7 @@ impl SeqExTokio { app: TL, packet: Packet>, ) -> Result<(ReplyGuard<'_, TL, SendData, RecvData, CAP>, RecvData), Option> { - let mut inner = self.seq_ex.lock().unwrap(); + let mut inner = self.inner.lock().unwrap(); return match inner.seq.receive_raw(app.clone(), packet) { Ok((recv_data, do_pump)) => { // pump first, handle return value second. @@ -267,7 +281,7 @@ impl SeqExTokio { seq_cst: bool, packet_data: F, ) -> Result, (TryError, F)> { - let mut inner = self.seq_ex.lock().unwrap(); + let mut inner = self.inner.lock().unwrap(); let pre_ts = inner.seq.next_service_timestamp; let result = inner.seq.try_send_with(app, seq_cst, packet_data); match result { @@ -338,7 +352,7 @@ impl SeqExTokio { if let Some(up) = result { state.next_service_timestamp = state.next_service_timestamp.min(up); } else { - let mut inner = self.seq_ex.lock().unwrap(); + let mut inner = self.inner.lock().unwrap(); inner.seq.service(app.clone()); state.next_service_timestamp = inner.seq.next_service_timestamp; }