diff --git a/Cargo.lock b/Cargo.lock index 0872f66..8006a3d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,38 @@ # It is not intended for manual editing. version = 3 +[[package]] +name = "futex" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d370e15a8972dee506ad50638e331183109e85ff99f349f19e04f288dc6cef3" +dependencies = [ + "integer-atomics", + "libc", + "lock-wrappers", +] + +[[package]] +name = "integer-atomics" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c33cd4d18b4ade167caace0e92364e8568c1e47c193738397b4b48a3e414139" + +[[package]] +name = "libc" +version = "0.2.147" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4668fb0ea861c1df094127ac5f1da3409a82116a4ba74fca2e58ef927159bb3" + +[[package]] +name = "lock-wrappers" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d654f44a90e266c873afdcf93f4506d9d6fb036b211e2be63a09695ecb0a07a" + [[package]] name = "seq_ex" version = "0.1.0" +dependencies = [ + "futex", +] diff --git a/Cargo.toml b/Cargo.toml index 160fabd..0aaba56 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,4 +11,7 @@ doc = true [features] default = ["std"] -std = [] +std = ["futex"] + +[dependencies] +futex = {version = "0.1.3", optional = true} diff --git a/examples/calculator.rs b/examples/calculator.rs index 46426af..e4f8192 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -43,8 +43,6 @@ struct Transport { impl seq_ex::TransportLayer for &Transport { type RecvData = SendPacket; - type RecvDataRef<'a> = &'a SendPacket; - type RecvReturn = (); type SendData = RawPacket; @@ -67,20 +65,16 @@ impl seq_ex::TransportLayer for &Transport { let _ = self.channel.send(RawPacket::EmptyReply(reply_no)); } } - - fn deserialize<'a>(data: &'a Self::RecvData) -> Self::RecvDataRef<'a> { - data - } - fn process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option) -> Self::RecvReturn { - let mut value = self.value.lock().unwrap(); - use SendPacket::*; - match recv_packet { - Add(n) => *value = *value + n, - Sub(n) => *value = *value - n, - Mul(n) => *value = *value * n, - Div(n) => *value = *value / n, - Mod(n) => *value = *value % n, - } +} +fn process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option) -> Self::RecvReturn { + let mut value = self.value.lock().unwrap(); + use SendPacket::*; + match recv_packet { + Add(n) => *value = *value + n, + Sub(n) => *value = *value - n, + Mul(n) => *value = *value * n, + Div(n) => *value = *value / n, + Mod(n) => *value = *value % n, } } @@ -107,15 +101,15 @@ fn main() { let mut seq2 = SeqEx::new(5, 1); let mut value = 0.0; - assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Add(1.0)))); + assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Add(1.0)))); value += 1.0; - assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Sub(2.0)))); + assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Sub(2.0)))); value -= 2.0; - assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mul(3.0)))); + assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mul(3.0)))); value *= 3.0; - assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Div(4.0)))); + assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Div(4.0)))); value /= 4.0; - assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mod(5.0)))); + assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mod(5.0)))); value %= 5.0; for _ in 0..30 { diff --git a/examples/hello_world.rs b/examples/hello_world.rs index c3b495f..0be719c 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -1,94 +1,58 @@ -use std::sync::mpsc::{channel, Receiver, Sender}; +use std::sync::mpsc::Receiver; -use seq_ex::{ReplyGuard, SeqEx, SeqNo}; -use std::time::Instant; +use seq_ex::sync::{MpscTransport, PacketType}; +use seq_ex::{ReplyGuard, SeqEx}; #[derive(Clone, Debug)] enum Packet { - Ack(SeqNo), - EmptyReply(SeqNo), - Hello(SeqNo), - Reply(SeqNo, SeqNo, ReplyPacket), -} - -#[derive(Clone, Debug)] -enum ReplyPacket { + Hello, Space, World, Exclamation, } +use Packet::*; -struct Transport { - channel: Sender, - time: Instant, -} - -impl seq_ex::TransportLayer for &Transport { - type RecvData = Packet; - type SendData = Packet; - - fn time(&self) -> i64 { - self.time.elapsed().as_millis() as i64 - } - - fn send(&self, data: &Self::SendData) { - let _ = self.channel.send(data.clone()); - } - fn send_ack(&self, reply_no: SeqNo) { - let _ = self.channel.send(Packet::Ack(reply_no)); - } - fn send_empty_reply(&self, reply_no: SeqNo) { - let _ = self.channel.send(Packet::EmptyReply(reply_no)); - } -} - -fn process(guard: ReplyGuard<'_, &Transport>, recv_packet: Packet, send_packet: Option) { - use Packet::*; - use ReplyPacket::*; +fn process(guard: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, send_packet: Option) { match (recv_packet, send_packet) { - (Hello(_), None) => { + (Hello, None) => { print!("Hello"); - guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Space)); + guard.reply(Space); + } + (Space, Some(Hello)) => { + print!(" "); + guard.reply(World); + } + (World, Some(Space)) => { + print!("World"); + guard.reply(Exclamation); + } + (Exclamation, Some(World)) => { + print!("!"); } - (Reply(_, _, r), Some(p)) => match (r, p) { - (Space, Hello(_)) => { - print!(" "); - guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, World)); - } - (World, Reply(_, _, Space)) => { - print!("World"); - guard.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Exclamation)); - } - (Exclamation, Reply(_, _, World)) => { - println!("!"); - } - (a, b) => { - println!("Unsolicited reply received: {:?}, was a reply to: {:?}", a, b); - } - }, (a, None) => { - println!("Unsolicited packet received: {:?}", a); + print!("Unsolicited packet received: {:?}", a); } (a, Some(b)) => { - println!("Unsolicited reply received: {:?}, was a reply to: {:?}", a, b); + print!("Incorrect reply received: {:?}, was a reply to: {:?}", a, b); } } } -fn receive<'a>(recv: &Receiver, seq: &mut SeqEx<&'a Transport>, transport: &'a Transport) { - use Packet::*; +fn receive<'a>(recv: &Receiver>, seq: &mut SeqEx<&'a MpscTransport>, transport: &'a MpscTransport) { let do_pump = { let result = match recv.recv().unwrap() { - Ack(reply_no) => { + PacketType::Ack { reply_no } => { seq.receive_ack(reply_no); return; } - EmptyReply(reply_no) => { - seq.receive_empty_reply(reply_no); + PacketType::EmptyReply { reply_no } => { + if let Some(Exclamation) = seq.receive_empty_reply(reply_no) { + // Our Hello World exchange ends right here. + print!("\n"); + } return; } - Hello(seq_no) => seq.receive(transport, seq_no, None, Hello(seq_no)), - Reply(seq_no, reply_no, p) => seq.receive(transport, seq_no, Some(reply_no), Reply(seq_no, reply_no, p)), + PacketType::Data { seq_no, reply_no, payload } => seq.receive(transport, seq_no, reply_no, payload), }; if let Ok((guard, recv_packet, send_packet)) = result { process(guard, recv_packet, send_packet); @@ -105,17 +69,17 @@ fn receive<'a>(recv: &Receiver, seq: &mut SeqEx<&'a Transport>, transpor } fn main() { - let (send1, recv1) = channel(); - let (send2, recv2) = channel(); - let transport1 = Transport { channel: send2, time: Instant::now() }; - let transport2 = Transport { channel: send1, time: Instant::now() }; - let mut seq1 = SeqEx::new(100, 1); - let mut seq2 = SeqEx::new(100, 1); + let (transport1, recv2) = MpscTransport::new(); + let (transport2, recv1) = MpscTransport::new(); + let mut seq1 = SeqEx::default(); + let mut seq2 = SeqEx::default(); - assert!(seq1.send_with(&transport1, |seq_no| Packet::Hello(seq_no))); + // We begin a "Hello World" exchange right here. + assert!(seq1.try_send(&transport1, Packet::Hello).is_ok()); receive(&recv2, &mut seq2, &transport2); receive(&recv1, &mut seq1, &transport1); receive(&recv2, &mut seq2, &transport2); receive(&recv1, &mut seq1, &transport1); + receive(&recv2, &mut seq2, &transport2); } diff --git a/src/lib.rs b/src/lib.rs index 5c18faf..08a29e1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -11,4 +11,4 @@ mod single_thread; pub use single_thread::*; #[cfg(feature = "std")] -pub mod multi_thread; +pub mod sync; diff --git a/src/multi_thread.rs b/src/multi_thread.rs deleted file mode 100644 index ed1bd77..0000000 --- a/src/multi_thread.rs +++ /dev/null @@ -1,62 +0,0 @@ -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_with(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 8cd0f60..64e2518 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -41,6 +41,11 @@ use crate::TransportLayer; /// All packets will either have a seq_no, a reply_no, or both. pub type SeqNo = u32; +/// The resend interval for a default instance of SeqEx. +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; + 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 @@ -151,9 +156,9 @@ 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 send(&mut self, app: TL, packet_data: TL::SendData) -> bool { + pub fn try_send(&mut self, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> { if self.is_full() { - return false; + return Err(packet_data); } let seq_no = self.next_send_seq_no; self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1); @@ -166,13 +171,8 @@ impl SeqEx { data: packet_data, }); - app.send(&entry.data); - true - } - #[must_use = "The queue might be full causing the packet to not be sent"] - pub fn send_with(&mut self, app: TL, create_data: impl FnOnce(SeqNo) -> TL::SendData) -> bool { - let p = create_data(self.seq_no()); - self.send(app, p) + app.send(entry.seq_no, entry.reply_no, &entry.data); + Ok(()) } pub fn receive_raw>( @@ -303,7 +303,7 @@ impl SeqEx { if let Some(entry) = item { if entry.next_resent_time <= current_time { entry.next_resent_time = next_interval; - app.send(&entry.data); + app.send(entry.seq_no, entry.reply_no, &entry.data); } else { next_activity = next_activity.min(entry.next_resent_time); } @@ -326,7 +326,7 @@ impl SeqEx { data: packet_data, }); - app.send(&entry.data); + app.send(entry.seq_no, entry.reply_no, &entry.data); } } pub fn reply_empty_raw(&mut self, app: TL, reply_no: SeqNo) { @@ -352,6 +352,11 @@ impl SeqEx { IterMut(self.send_window.iter_mut()) } } +impl Default for SeqEx { + fn default() -> Self { + Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO) + } +} impl<'a, TL: TransportLayer> IntoIterator for &'a SeqEx { type Item = &'a TL::SendData; type IntoIter = Iter<'a, TL>; diff --git a/src/single_thread.rs b/src/single_thread.rs index 9f5e4ce..b2664bb 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -12,10 +12,6 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { self.0.reply_raw(self.1.clone(), self.2, packet_data); core::mem::forget(self); } - pub fn reply_with(self, create_data: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) { - let p = create_data(self.seq_no(), self.reply_no()); - self.reply(p) - } } impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { fn drop(&mut self) { diff --git a/src/sync.rs b/src/sync.rs new file mode 100644 index 0000000..e46ce44 --- /dev/null +++ b/src/sync.rs @@ -0,0 +1,151 @@ +use std::cell::UnsafeCell; +use std::sync::Condvar; +use std::{ + sync::{ + mpsc::{channel, Receiver, Sender}, + Mutex, MutexGuard, + }, + time::Instant, +}; + +use crate::{Error, SeqEx, SeqNo, TransportLayer}; + +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 fn reply(self, packet_data: TL::SendData) { + let mut seq = self.0.seq_ex.lock().unwrap(); + seq.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) { + let mut seq = self.0.seq_ex.lock().unwrap(); + seq.reply_empty_raw(self.1.clone(), self.2); + } +} + +impl SeqExSync { + 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(); + let ret = seq.receive_raw(app.clone(), seq_no, reply_no, packet); + self.unblock(ret.is_ok()); + ret.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(); + let ret = seq.pump_raw(); + self.unblock(ret.is_ok()); + ret.map(|(reply_no, packet, data)| (ReplyGuard(self, app, reply_no), packet, data)) + } + #[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 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), + send_block: Condvar::default(), + } + } + pub fn try_send(&self, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> { + let mut seq = self.lock(); + seq.try_send(app, packet_data) + } + pub fn send(&self, app: TL, mut packet_data: TL::SendData) { + let mut seq = self.lock(); + while let Err(p) = seq.try_send(app.clone(), packet_data) { + packet_data = p; + unsafe { *self.wait_count.get() += 1} + seq = self.send_block.wait(seq).unwrap(); + unsafe { *self.wait_count.get() -= 1} + } + } + + pub fn receive_ack(&self, reply_no: SeqNo) { + self.lock().receive_ack(reply_no) + } + pub fn receive_empty_reply(&self, reply_no: SeqNo) -> Option { + let ret = self.lock().receive_empty_reply(reply_no); + if ret.is_some() { + self.send_block.notify_one(); + } + ret + } + pub fn service(&self, app: TL) -> i64 { + self.lock().service(app) + } + + pub fn lock(&self) -> MutexGuard> { + self.seq_ex.lock().unwrap() + } +} + +#[derive(Clone)] +pub enum PacketType { + Data { + seq_no: SeqNo, + reply_no: Option, + payload: Payload, + }, + Ack { + reply_no: SeqNo, + }, + EmptyReply { + reply_no: SeqNo, + }, +} + +#[derive(Clone)] +pub struct MpscTransport { + pub channel: Sender>, + pub time: Instant, +} +impl MpscTransport { + pub fn new() -> (Self, Receiver>) { + let (send, recv) = channel(); + (Self { channel: send, time: std::time::Instant::now() }, recv) + } + pub fn from_sender(send: Sender>) -> Self { + Self { channel: send, time: std::time::Instant::now() } + } +} +impl TransportLayer for &MpscTransport { + type RecvData = Payload; + type SendData = Payload; + + fn time(&self) -> i64 { + self.time.elapsed().as_millis() as i64 + } + + fn send(&self, seq_no: SeqNo, reply_no: Option, payload: &Self::SendData) { + let _ = self.channel.send(PacketType::Data { seq_no, reply_no, payload: payload.clone() }); + } + fn send_ack(&self, reply_no: SeqNo) { + let _ = self.channel.send(PacketType::Ack { reply_no }); + } + fn send_empty_reply(&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 f3615e6..254a3dd 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -4,15 +4,15 @@ use crate::SeqNo; /// /// 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. +/// 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 { type RecvData; type SendData; fn time(&self) -> i64; - fn send(&self, data: &Self::SendData); + fn send(&self, seq_no: SeqNo, reply_no: Option, payload: &Self::SendData); fn send_ack(&self, reply_no: SeqNo); fn send_empty_reply(&self, reply_no: SeqNo); }