diff --git a/examples/file_download.rs b/examples/file_download.rs index 0319dfe..5356b69 100644 --- a/examples/file_download.rs +++ b/examples/file_download.rs @@ -70,7 +70,7 @@ fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload, Payl let mut i = 0; while i < file.len() { let j = file.len().min(i + FILE_CHUNK_SIZE); - seqex.send(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() }); + seqex.send_locked(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() }); i = j; } } diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 8ff0763..9f4f081 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -88,7 +88,7 @@ pub enum DirectError { /// invalid to process it right now. No action needs to be taken by the caller. OutOfSequence, /// The Send Window is currently full. The received packet cannot be processed right now because - /// it could cause the send window to overflow. No action needs to be taken by the caller. + /// it could cause the send window to overflow. WindowIsFull(Packet), WindowIsLocked(Packet), ResendAck(SeqNo), diff --git a/src/single_thread.rs b/src/single_thread.rs index f4c5eb4..7e88d57 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -12,15 +12,24 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// 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) { - self.reply_with(|_, _| packet_data) + pub fn reply(self, packet_data: SendData) { + self.reply_inner(false, |_, _| packet_data) } - pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_locked(self, packet_data: SendData) { + self.reply_inner(true, |_, _| packet_data) + } + pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + self.reply_inner(false, packet_data) + } + pub fn reply_locked_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + self.reply_inner(true, packet_data) + } + fn reply_inner(mut self, locked: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { let mut app = None; core::mem::swap(&mut app, &mut self.app); let seq_no = self.seq.seq_no(); self.seq - .reply_raw(app.unwrap(), self.reply_no, self.locked, packet_data(seq_no, self.reply_no)); + .reply_raw(app.unwrap(), self.reply_no, self.locked, locked, packet_data(seq_no, self.reply_no)); core::mem::forget(self); } } @@ -45,7 +54,7 @@ pub enum Error { /// invalid to process it right now. No action needs to be taken by the caller. OutOfSequence, /// The Send Window is currently full. The received packet cannot be processed right now because - /// it could cause the send window to overflow. No action needs to be taken by the caller. + /// it could cause the send window to overflow. WindowIsFull(Packet), WindowIsLocked(Packet), } @@ -133,18 +142,20 @@ impl_recvok!(RecvOk, &'a mut SeqEx); pub(crate) use impl_recvok; impl SeqEx { - 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, locked: bool, packet_data: SendData) -> Result<(), SendData> { match self.try_send_direct(packet_data, app.time()) { - Ok(p) => { + Ok(mut p) => { + p.set_locking(locked); app.send(p); Ok(()) } Err(e) => Err(e), } } - pub fn try_send_with(&mut self, mut app: impl TransportLayer, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), ()> { + pub fn try_send_with(&mut self, mut app: impl TransportLayer, locked: bool, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), ()> { match self.try_send_direct_with(packet_data, app.time()) { - Ok(p) => { + Ok(mut p) => { + p.set_locking(locked); app.send(p); Ok(()) } @@ -168,8 +179,9 @@ impl SeqEx { Err(DirectError::WindowIsLocked(p)) => Err(Error::WindowIsLocked(p)), } } - pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, unlock: bool, packet_data: SendData) { - if let Some(p) = self.reply_raw_and_direct(reply_no, unlock, packet_data, app.time()) { + pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, unlock: bool, locked_packet: bool, packet_data: SendData) { + if let Some(mut p) = self.reply_raw_and_direct(reply_no, unlock, packet_data, app.time()) { + p.set_locking(locked_packet); app.send(p) } } diff --git a/src/sync.rs b/src/sync.rs index 6f62973..ebb2fd6 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -12,7 +12,7 @@ use crate::{Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_ pub struct SeqExSync { seq_ex: Mutex<(SeqEx, usize)>, send_block: Condvar, - lock: Condvar, + recv_lock: Condvar, } pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { @@ -27,15 +27,28 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// 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) { - self.reply_with(|_, _| packet_data) + pub fn reply(self, packet_data: SendData) { + self.reply_inner(false, |_, _| packet_data) } - pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_locked(self, packet_data: SendData) { + self.reply_inner(true, |_, _| packet_data) + } + pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + self.reply_inner(false, packet_data) + } + pub fn reply_locked_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + self.reply_inner(true, packet_data) + } + fn reply_inner(mut self, locked: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { let mut app = None; core::mem::swap(&mut app, &mut self.app); let mut seq = self.seq.lock(); let seq_no = seq.seq_no(); - seq.reply_raw(app.unwrap(), self.reply_no, self.locked, packet_data(seq_no, self.reply_no)); + seq.reply_raw(app.unwrap(), self.reply_no, self.locked, locked, packet_data(seq_no, self.reply_no)); + drop(seq); + if self.locked { + self.seq.recv_lock.notify_all(); + } core::mem::forget(self); } } @@ -106,7 +119,7 @@ impl SeqExSync { Self { seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)), send_block: Condvar::default(), - lock: Condvar::default(), + recv_lock: Condvar::default(), } } @@ -127,7 +140,7 @@ impl SeqExSync { Err(crate::Error::OutOfSequence) => return Err(Error::OutOfSequence), Err(crate::Error::WindowIsFull(_)) => return Err(Error::WindowIsFull), Err(crate::Error::WindowIsLocked(p)) => { - seq = self.lock.wait(seq).unwrap(); + seq = self.recv_lock.wait(seq).unwrap(); packet = p; } } @@ -146,7 +159,7 @@ impl SeqExSync { Err(PumpError::OutOfSequence) => return Err(Error::OutOfSequence), Err(PumpError::WindowIsFull) => return Err(Error::WindowIsFull), Err(PumpError::WindowIsLocked) => { - seq = self.lock.wait(seq).unwrap(); + seq = self.recv_lock.wait(seq).unwrap(); } } } @@ -162,30 +175,42 @@ impl SeqExSync { ReplyIter { seq: None, app, first: None } } } - pub fn try_send>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> { - self.try_send_with(app, |_| packet_data) + pub fn try_send>(&self, app: TL, locked: bool, packet_data: SendData) -> Result<(), SendData> { + self.try_send_with(app, locked, |_| packet_data) } - pub fn send_with>(&self, app: TL, mut packet_data: impl FnMut(SeqNo) -> SendData) { + fn send_with_inner>(&self, app: TL, locked: bool, mut packet_data: impl FnMut(SeqNo) -> SendData) { let mut seq = self.seq_ex.lock().unwrap(); - while let Err(()) = seq.0.try_send_with(app.clone(), &mut packet_data) { + while let Err(()) = seq.0.try_send_with(app.clone(), locked, &mut packet_data) { seq.1 += 1; seq = self.send_block.wait(seq).unwrap(); seq.1 -= 1; } } - pub fn send>(&self, app: TL, mut packet_data: SendData) { + fn send_inner>(&self, app: TL, locked: bool, mut packet_data: SendData) { let mut seq = self.seq_ex.lock().unwrap(); - while let Err(p) = seq.0.try_send(app.clone(), packet_data) { + while let Err(p) = seq.0.try_send(app.clone(), locked, packet_data) { packet_data = p; seq.1 += 1; seq = self.send_block.wait(seq).unwrap(); seq.1 -= 1; } } - pub fn try_send_with>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> { + pub fn send(&self, app: impl TransportLayer, packet_data: SendData) { + self.send_inner(app, false, packet_data) + } + pub fn send_locked(&self, app: impl TransportLayer, packet_data: SendData) { + self.send_inner(app, true, packet_data) + } + pub fn send_with(&self, app: impl TransportLayer, packet_data: impl FnMut(SeqNo) -> SendData) { + self.send_with_inner(app, false, packet_data) + } + pub fn send_locked_with(&self, app: impl TransportLayer, packet_data: impl FnMut(SeqNo) -> SendData) { + self.send_with_inner(app, true, packet_data) + } + pub fn try_send_with>(&self, app: TL, locked: bool, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> { let mut seq = self.lock(); let seq_no = seq.seq_no(); - seq.try_send(app, packet_data(seq_no)) + seq.try_send(app, locked, packet_data(seq_no)) } pub fn service>(&self, app: TL) -> i64 { self.lock().service(app)