diff --git a/examples/calculator.rs b/examples/calculator.rs index 8022289..b436d71 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -16,7 +16,7 @@ enum Packet { } fn drop_packet() -> bool { - static RNG: Mutex = Mutex::new(12); + static RNG: Mutex = Mutex::new(43); let mut rng = RNG.lock().unwrap(); *rng ^= *rng << 13; *rng ^= *rng >> 17; @@ -41,30 +41,17 @@ fn receive<'a>( transport: &'a MpscTransport, value: &mut f32, ) { - let packet = recv.try_recv(); - if !drop_packet() { - let do_pump = match packet { - Ok(PacketType::Ack { reply_no }) => { - seq.receive_ack(reply_no); - return; - } - Ok(PacketType::EmptyReply { reply_no }) => { - let result = seq.receive_empty_reply(reply_no); - result.is_some() - } - 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 + while let Ok(packet) = recv.try_recv() { + if !drop_packet() { + match packet { + PacketType::EmptyReply { reply_no } => { + let _ = seq.receive_empty_reply(reply_no); + } + PacketType::Payload { seq_no, reply_no, payload } => { + for RecvSuccess { guard, packet, send_data } in seq.receive_iter(transport, seq_no, reply_no, payload) { + process(guard, packet, send_data, value); + } } - } - _ => return, - }; - if do_pump { - 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 c3f3774..885290e 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -38,32 +38,19 @@ fn process(guard: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, s } fn receive<'a>(recv: &Receiver>, seq: &SeqExSync<&'a MpscTransport>, transport: &'a MpscTransport) { - let do_pump = match recv.recv().unwrap() { - PacketType::Ack { reply_no } => { - seq.receive_ack(reply_no); - return; - } + match recv.recv().unwrap() { PacketType::EmptyReply { reply_no } => { let result = seq.receive_empty_reply(reply_no); - if let Some(Exclamation) = &result { + if let Ok(Exclamation) = result { // Our Hello World exchange ends right here. print!("\n"); } - result.is_some() } 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 + for RecvSuccess {guard, packet, send_data} in seq.receive_iter(transport, seq_no, reply_no, payload) { + process(guard, packet, send_data) } } - }; - if do_pump { - 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 09d9322..8a94c16 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -46,10 +46,9 @@ pub const DEFAULT_RESEND_INTERVAL_MS: i64 = 200; /// The initial sequence number for a default instance of SeqEx. pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 1; -pub const DEFAULT_SEND_WINDOW_LEN: usize = 64; -pub const DEFAULT_RECV_WINDOW_LEN: usize = 64; +pub const DEFAULT_WINDOW_CAP: usize = 64; -pub struct SeqEx { +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. @@ -57,12 +56,16 @@ pub struct SeqEx>; SLEN], - recv_window: [Option>; RLEN], - /// The size of this array determines the maximum number of received packets that the - /// application may attempt to process concurrently before new received packets start being dropped. - concurrent_replies: [SeqNo; RLEN], + /// This could be made more efficient by changing to SoA format. + send_window: [Option>; CAP], + recv_window: [Option>; CAP], + /// The size of this array determines the maximum number of received packets that the application + /// may attempt to process concurrently before new received packets start being dropped. + concurrent_replies: [SeqNo; CAP], + /// The total number of concurrent replies being processed. When a packet is received, a reply + /// number is issued for that packet. That reply number reserves resources for itself, so that + /// when `reply_raw` or `reply_empty_raw` are called with it, they are guaranteed not to fail. + /// To accomplish this we must track all issued reply numbers. concurrent_replies_total: usize, } @@ -106,7 +109,7 @@ pub struct Iter<'a, TL: TransportLayer>(core::slice::Iter<'a, Option(core::slice::IterMut<'a, Option>>); -impl SeqEx { +impl SeqEx { /// Creates a new instance of `SeqEx` for a new remote peer. /// An instance of `SeqEx` expects to communicate with only exactly one other remote instance /// of `SeqEx`. @@ -140,7 +143,7 @@ impl SeqEx { pub fn is_full(&self) -> bool { // We claim that the window is full one entry before it is actually full for the sake of // making it always possible for both peers to process at least one reply at all times. - self.is_full_inner(1) + self.is_full_inner(true) } /// Returns the next sequence number to be attached to the next sent packet. /// This should be called before `SeqEx::send`, and the return value should be @@ -214,13 +217,13 @@ impl SeqEx { // If either are the case then we know we will eventually send a reply to the packet. // If neither are the case then we must send an empty reply so the remote peer can stop // resending the packet. - for entry in &self.send_window { - if entry.as_ref().map_or(false, |e| e.reply_no == Some(seq_no)) { + for entry in self.send_window.iter().flatten() { + if entry.reply_no == Some(seq_no) { return Err(Error::OutOfSequence); } } - for reply_no in self.concurrent_replies { - if reply_no == seq_no { + for i in 0..self.concurrent_replies_total { + if self.concurrent_replies[i] == seq_no { return Err(Error::OutOfSequence); } } @@ -232,7 +235,7 @@ impl SeqEx { // If the send window is full we cannot safely process received packets, // because there would be no way to reply. // We can only process this packet if processing it would make space in the send window. - let is_full = self.is_full_inner(0); + let is_full = self.is_full_inner(false); let i = seq_no as usize % self.recv_window.len(); if let Some(pre) = self.recv_window[i].as_mut() { @@ -240,7 +243,6 @@ impl SeqEx { if is_next && !is_full { self.recv_window[i] = None; } else { - app.send_ack(seq_no); return if is_full { Err(Error::WindowIsFull) } else { @@ -261,10 +263,6 @@ impl SeqEx { Ok((seq_no, packet, data)) } else { self.recv_window[i] = Some(RecvEntry { seq_no, reply_no, data: packet.into() }); - if let Some(reply_no) = reply_no { - self.receive_ack(reply_no); - } - app.send_ack(seq_no); if is_full { Err(Error::WindowIsFull) } else { @@ -272,29 +270,21 @@ impl SeqEx { } } } - pub fn receive_ack(&mut self, reply_no: SeqNo) { - let slot = self.send_window_slot_mut(reply_no); - if let Some(entry) = slot.as_mut() { - if entry.seq_no == reply_no { - entry.next_resend_time = i64::MAX; - } - } - } - pub fn receive_empty_reply(&mut self, reply_no: SeqNo) -> Option { + pub fn receive_empty_reply(&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(); - Some(entry.data) + Ok(entry.data) } else { - None + Err(Error::OutOfSequence) } } - fn is_full_inner(&self, bias: u32) -> bool { + fn is_full_inner(&self, reserve_one: bool) -> bool { if self.concurrent_replies_total >= self.concurrent_replies.len() { return true; } - for i in 0..self.concurrent_replies_total as u32 + 1 + bias { + for i in 0..self.concurrent_replies_total as u32 + 1 + reserve_one as u32 { let slot = self.send_window_slot(self.next_send_seq_no.wrapping_add(i)); if slot.is_some() { return true; @@ -316,7 +306,7 @@ impl SeqEx { let i = next_seq_no as usize % self.recv_window.len(); if self.recv_window[i].as_ref().map_or(false, |pre| pre.seq_no == next_seq_no) { - if self.is_full_inner(0) { + if self.is_full_inner(false) { return Err(Error::WindowIsFull); } @@ -350,7 +340,7 @@ impl SeqEx { self.next_service_timestamp = next_resend_time; app.update_service_time(next_resend_time, current_time); } - let slot = self.send_window_slot_mut(reply_no); + let slot = self.send_window_slot_mut(seq_no); debug_assert!(slot.is_none()); let entry = slot.insert(SendEntry { seq_no, diff --git a/src/sync.rs b/src/sync.rs index d5b9c90..e686b69 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -9,19 +9,19 @@ use std::{ }; use crate::{ - Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RECV_WINDOW_LEN, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_SEND_WINDOW_LEN, + Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_WINDOW_CAP, DEFAULT_RESEND_INTERVAL_MS, }; -pub struct SeqExSync { - seq_ex: Mutex>, +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, send_block: Condvar, } -pub struct ReplyGuard<'a, TL: TransportLayer>(&'a SeqExSync, TL, SeqNo); -impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { +pub struct ReplyGuard<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a SeqExSync, TL, SeqNo); +impl<'a, TL: TransportLayer, const CAP: usize> ReplyGuard<'a, TL, 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, @@ -33,23 +33,23 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { core::mem::forget(self); } } -impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { +impl<'a, TL: TransportLayer, const CAP: usize> Drop for ReplyGuard<'a, TL, CAP> { fn drop(&mut self) { let mut seq = self.0.seq_ex.lock().unwrap(); seq.reply_empty_raw(self.1.clone(), self.2); } } -pub struct RecvSuccess<'a, TL: TransportLayer, P> { - pub guard: ReplyGuard<'a, TL>, +pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, const CAP: usize = DEFAULT_WINDOW_CAP> { + pub guard: ReplyGuard<'a, TL, CAP>, pub packet: P, pub send_data: Option, } -impl SeqExSync { +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)), + seq_ex: Mutex::new(SeqEx::::new(retry_interval, initial_seq_no)), wait_count: UnsafeCell::new(0), send_block: Condvar::default(), } @@ -61,18 +61,32 @@ impl SeqExSync { seq_no: SeqNo, reply_no: Option, packet: P, - ) -> Result, Error> { + ) -> 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()); 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> { + 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, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } + pub fn receive_iter( + &self, + app: TL, + seq_no: SeqNo, + reply_no: Option, + packet: TL::RecvData, + ) -> ReplyIter<'_, TL, CAP> { + if let Ok(g) = self.receive(app.clone(), seq_no, reply_no, packet) { + ReplyIter { origin: Some(self), app, first: Some(g) } + } else { + ReplyIter { origin: None, app, first: None } + } + } #[inline] fn unblock(&self, is_ok: bool) { let has_waiting = unsafe { *self.wait_count.get() > 0 }; @@ -94,21 +108,16 @@ impl SeqExSync { } } - pub fn receive_ack(&self, reply_no: SeqNo) { - self.lock().receive_ack(reply_no) - } - pub fn receive_empty_reply(&self, reply_no: SeqNo) -> Option { + pub fn receive_empty_reply(&self, reply_no: SeqNo) -> Result { let ret = self.lock().receive_empty_reply(reply_no); - if ret.is_some() { - self.send_block.notify_one(); - } + self.unblock(ret.is_ok()); ret } pub fn service(&self, app: TL) -> i64 { self.lock().service(app) } - pub fn lock(&self) -> MutexGuard> { + pub fn lock(&self) -> MutexGuard> { self.seq_ex.lock().unwrap() } } @@ -118,6 +127,25 @@ impl Default for SeqExSync { } } +pub struct ReplyIter<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> { + origin: Option<&'a SeqExSync>, + app: TL, + first: Option> +} + +impl<'a, TL: TransportLayer> Iterator for ReplyIter<'a, TL> { + type Item = RecvSuccess<'a, TL, TL::RecvData>; + fn next(&mut self) -> Option { + if let Some(g) = self.first.take() { + Some(g) + } else if let Some(origin) = self.origin { + origin.pump(self.app.clone()).ok() + } else { + None + } + } +} + #[derive(Clone)] pub enum PacketType { Payload { @@ -125,9 +153,6 @@ pub enum PacketType { reply_no: Option, payload: Payload, }, - Ack { - reply_no: SeqNo, - }, EmptyReply { reply_no: SeqNo, }, @@ -158,9 +183,6 @@ impl TransportLayer for &MpscTransport { 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(&mut self, reply_no: SeqNo) { - let _ = self.channel.send(PacketType::Ack { reply_no }); - } 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 d2c0bd8..b621615 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -6,7 +6,7 @@ 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: Sized + Clone { +pub trait TransportLayer: Clone { type RecvData; type SendData; @@ -15,6 +15,5 @@ pub trait TransportLayer: Sized + Clone { 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_ack(&mut self, reply_no: SeqNo); fn send_empty_reply(&mut self, reply_no: SeqNo); }