diff --git a/examples/calculator.rs b/examples/calculator.rs index e4f8192..aa89dba 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -1,29 +1,18 @@ use std::{ - sync::{ - mpsc::{channel, Receiver, Sender}, - Mutex, - }, + sync::{mpsc::Receiver, Mutex}, thread, time::Duration, }; -use seq_ex::{ReplyGuard, SeqEx, SeqNo}; -use std::time::Instant; +use seq_ex::sync::{MpscTransport, PacketType, ReplyGuard, SeqExSync}; #[derive(Clone)] -enum RawPacket { - Ack(SeqNo), - EmptyReply(SeqNo), - Send(SeqNo, SendPacket), -} - -#[derive(Clone)] -enum SendPacket { - Add(f64), - Sub(f64), - Mul(f64), - Div(f64), - Mod(f64), +enum Packet { + Add(f32), + Sub(f32), + Mul(f32), + Div(f32), + Mod(f32), } fn drop_packet() -> bool { @@ -32,43 +21,11 @@ fn drop_packet() -> bool { *rng ^= *rng << 13; *rng ^= *rng >> 17; *rng ^= *rng << 5; - *rng & 3 == 0 + *rng & 1 == 0 } -struct Transport { - channel: Sender, - time: Instant, - value: Mutex, -} - -impl seq_ex::TransportLayer for &Transport { - type RecvData = SendPacket; - - type SendData = RawPacket; - - fn time(&self) -> i64 { - self.time.elapsed().as_millis() as i64 - } - - fn send(&self, data: &Self::SendData) { - if drop_packet() { - let _ = self.channel.send(data.clone()); - } - } - fn send_ack(&self, reply_no: SeqNo) { - if drop_packet() { - let _ = self.channel.send(RawPacket::Ack(reply_no)); - } - } - fn send_empty_reply(&self, reply_no: SeqNo) { - if drop_packet() { - let _ = self.channel.send(RawPacket::EmptyReply(reply_no)); - } - } -} -fn process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option) -> Self::RecvReturn { - let mut value = self.value.lock().unwrap(); - use SendPacket::*; +fn process(_: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, _: Option, value: &mut f32) { + use Packet::*; match recv_packet { Add(n) => *value = *value + n, Sub(n) => *value = *value - n, @@ -78,47 +35,66 @@ fn process(&self, _: ReplyGuard<'_, Self>, recv_packet: &SendPacket, _: Option(recv: &Receiver, seq: &mut SeqEx<&'a Transport>, transport: &'a Transport) { - match recv.try_recv() { - Ok(RawPacket::Ack(reply_no)) => seq.receive_ack(reply_no), - Ok(RawPacket::EmptyReply(reply_no)) => { - seq.receive_empty_reply(reply_no); - while let Ok(()) = seq.pump(transport) {} +fn receive<'a>( + recv: &Receiver>, + seq: &SeqExSync<&'a MpscTransport>, + 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::Data { seq_no, reply_no, payload }) => { + if let Ok((guard, recv_packet, send_packet)) = seq.receive(transport, seq_no, reply_no, payload) { + process(guard, recv_packet, send_packet, value); + true + } else { + false + } + } + _ => return, + }; + if do_pump { + while let Ok((guard, recv_packet, send_packet)) = seq.pump(transport) { + process(guard, recv_packet, send_packet, value); + } } - Ok(RawPacket::Send(seq_no, packet)) => match seq.receive(transport, seq_no, None, packet) { - Ok(()) => while let Ok(()) = seq.pump(transport) {}, - Err(_) => {} - }, - _ => {} } } -fn main() { - let (send1, recv1) = channel(); - let (send2, recv2) = channel(); - let transport1 = Transport { channel: send2, time: Instant::now(), value: Mutex::new(0.0) }; - let transport2 = Transport { channel: send1, time: Instant::now(), value: Mutex::new(0.0) }; - let mut seq1 = SeqEx::new(5, 1); - let mut seq2 = SeqEx::new(5, 1); +fn main() { + let (transport1, recv2) = MpscTransport::new(); + let (transport2, recv1) = MpscTransport::new(); + let seq1 = SeqExSync::new(5, 1); + let seq2 = SeqExSync::new(5, 1); let mut value = 0.0; - assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Add(1.0)))); + let mut remote_value = value; + + seq1.send(&transport1, Packet::Add(1.0)); value += 1.0; - assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Sub(2.0)))); + seq1.send(&transport1, Packet::Sub(2.0)); value -= 2.0; - assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mul(3.0)))); + seq1.send(&transport1, Packet::Mul(3.0)); value *= 3.0; - assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Div(4.0)))); + seq1.send(&transport1, Packet::Div(4.0)); value /= 4.0; - assert!(seq1.try_send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mod(5.0)))); + seq1.send(&transport1, Packet::Mod(5.0)); value %= 5.0; for _ in 0..30 { - receive(&recv1, &mut seq1, &transport1); - receive(&recv2, &mut seq2, &transport2); + receive(&recv1, &seq1, &transport1, &mut value); + receive(&recv2, &seq2, &transport2, &mut remote_value); thread::sleep(Duration::from_millis(5)); seq1.service(&transport1); seq2.service(&transport2); } - let remote_value = transport2.value.lock().unwrap(); - assert_eq!(value, remote_value.clone()); + assert_eq!(value, remote_value); } diff --git a/examples/hello_world.rs b/examples/hello_world.rs index 0be719c..ba2ae46 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -1,7 +1,6 @@ use std::sync::mpsc::Receiver; -use seq_ex::sync::{MpscTransport, PacketType}; -use seq_ex::{ReplyGuard, SeqEx}; +use seq_ex::sync::{MpscTransport, PacketType, ReplyGuard, SeqExSync}; #[derive(Clone, Debug)] enum Packet { @@ -38,27 +37,27 @@ fn process(guard: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, s } } -fn receive<'a>(recv: &Receiver>, seq: &mut SeqEx<&'a MpscTransport>, transport: &'a MpscTransport) { - let do_pump = { - let result = match recv.recv().unwrap() { - PacketType::Ack { reply_no } => { - seq.receive_ack(reply_no); - return; +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; + } + PacketType::EmptyReply { reply_no } => { + let result = seq.receive_empty_reply(reply_no); + if let Some(Exclamation) = result { + // Our Hello World exchange ends right here. + print!("\n"); } - PacketType::EmptyReply { reply_no } => { - if let Some(Exclamation) = seq.receive_empty_reply(reply_no) { - // Our Hello World exchange ends right here. - print!("\n"); - } - return; + result.is_some() + } + PacketType::Data { seq_no, reply_no, payload } => { + if let Ok((guard, recv_packet, send_packet)) = seq.receive(transport, seq_no, reply_no, payload) { + process(guard, recv_packet, send_packet); + true + } else { + false } - 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); - true - } else { - false } }; if do_pump { @@ -71,15 +70,15 @@ fn receive<'a>(recv: &Receiver>, seq: &mut SeqEx<&'a MpscTran fn main() { let (transport1, recv2) = MpscTransport::new(); let (transport2, recv1) = MpscTransport::new(); - let mut seq1 = SeqEx::default(); - let mut seq2 = SeqEx::default(); + let seq1 = SeqExSync::default(); + let seq2 = SeqExSync::default(); // We begin a "Hello World" exchange right here. - assert!(seq1.try_send(&transport1, Packet::Hello).is_ok()); + seq1.send(&transport1, Packet::Hello); - 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); + receive(&recv2, &seq2, &transport2); + receive(&recv1, &seq1, &transport1); + receive(&recv2, &seq2, &transport2); + receive(&recv1, &seq1, &transport1); + receive(&recv2, &seq2, &transport2); } diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 64e2518..cd80297 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -46,8 +46,11 @@ 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 = 32; + const MAX_CONCURRENCY: usize = 24; -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. @@ -141,11 +144,11 @@ impl SeqEx { } /// Sends the given packet to the remote peer and adds it to the send window. /// - /// If the return value is `false` the queue is full and the packet will not be 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_empty_reply` returns `Some` and try again. + /// `receive` or `receive_empty_reply` returns `Ok` and try again. /// - /// If true is returned then the packet was successfully sent. + /// If `Ok` is returned then the packet was successfully sent. /// /// `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. diff --git a/src/sync.rs b/src/sync.rs index e46ce44..3402553 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -8,10 +8,12 @@ use std::{ time::Instant, }; -use crate::{Error, SeqEx, SeqNo, TransportLayer}; +use crate::{ + Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RECV_WINDOW_LEN, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_SEND_WINDOW_LEN, +}; -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, @@ -34,7 +36,15 @@ impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { } impl SeqExSync { - pub fn receive, T>( + 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 receive>( &self, app: TL, seq_no: SeqNo, @@ -54,21 +64,11 @@ impl SeqExSync { } #[inline] fn unblock(&self, is_ok: bool) { - let has_waiting = unsafe { - *self.wait_count.get() > 0 - }; + 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) @@ -77,9 +77,9 @@ impl SeqExSync { 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} + unsafe { *self.wait_count.get() += 1 } seq = self.send_block.wait(seq).unwrap(); - unsafe { *self.wait_count.get() -= 1} + unsafe { *self.wait_count.get() -= 1 } } } @@ -101,6 +101,11 @@ impl SeqExSync { self.seq_ex.lock().unwrap() } } +impl Default for SeqExSync { + fn default() -> Self { + Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO) + } +} #[derive(Clone)] pub enum PacketType { @@ -139,7 +144,7 @@ impl TransportLayer for &MpscTransport { self.time.elapsed().as_millis() as i64 } - fn send(&self, seq_no: SeqNo, reply_no: Option, payload: &Self::SendData) { + fn send(&self, seq_no: SeqNo, reply_no: Option, payload: &Payload) { let _ = self.channel.send(PacketType::Data { seq_no, reply_no, payload: payload.clone() }); } fn send_ack(&self, reply_no: SeqNo) {