diff --git a/examples/calculator.rs b/examples/calculator.rs new file mode 100644 index 0000000..85e1c38 --- /dev/null +++ b/examples/calculator.rs @@ -0,0 +1,131 @@ +use std::{ + sync::{ + mpsc::{channel, Receiver, Sender}, + Mutex, + }, + time::Duration, thread, +}; + +use seq_ex::{ReplyGuard, SeqEx, SeqNo}; +use std::time::Instant; + +#[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), +} + +fn drop_packet() -> bool { + static RNG: Mutex = Mutex::new(12); + let mut rng = RNG.lock().unwrap(); + *rng ^= *rng << 13; + *rng ^= *rng >> 17; + *rng ^= *rng << 5; + *rng & 1 > 0 +} + +struct Transport { + channel: Sender, + time: Instant, + value: Mutex, +} + +impl seq_ex::TransportLayer for &Transport { + type RecvData = SendPacket; + type RecvDataRef<'a> = &'a SendPacket; + type RecvReturn = (); + + 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 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 receive<'a>(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) {} + } + 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); + + let mut value = 0.0; + assert!(seq1.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)))); + value -= 2.0; + assert!(seq1.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)))); + value /= 4.0; + assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), SendPacket::Mod(5.0)))); + value %= 5.0; + + for _ in 0..20 { + receive(&recv1, &mut seq1, &transport1); + receive(&recv2, &mut seq2, &transport2); + 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()); +} diff --git a/examples/hello_world.rs b/examples/hello_world.rs new file mode 100644 index 0000000..1b6fe8e --- /dev/null +++ b/examples/hello_world.rs @@ -0,0 +1,96 @@ +use std::sync::mpsc::{channel, Receiver, Sender}; + +use seq_ex::{ReplyGuard, SeqEx, SeqNo}; +use std::time::Instant; + +#[derive(Clone)] +enum RawPacket { + Ack(SeqNo), + EmptyReply(SeqNo), + Send(SeqNo, Packet), + Reply(SeqNo, SeqNo, Packet), +} +#[derive(Clone)] +enum Packet { + Hello, + Space, + World, + Exclamation, +} + +struct Transport { + channel: Sender, + time: Instant, +} + +impl seq_ex::TransportLayer for &Transport { + type RecvData = Packet; + type RecvDataRef<'a> = &'a Packet; + type RecvReturn = (); + + type SendData = RawPacket; + + 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(RawPacket::Ack(reply_no)); + } + fn send_empty_reply(&self, reply_no: SeqNo) { + let _ = self.channel.send(RawPacket::EmptyReply(reply_no)); + } + + fn deserialize<'a>(data: &'a Self::RecvData) -> Self::RecvDataRef<'a> { + data + } + fn process(&self, reply_cx: ReplyGuard<'_, Self>, recv_packet: &Packet, _: Option) -> Self::RecvReturn { + use RawPacket::Reply; + match recv_packet { + Packet::Hello => { + print!("Hello"); + reply_cx.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Packet::Space)); + } + Packet::Space => { + print!(" "); + reply_cx.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Packet::World)); + } + Packet::World => { + print!("World"); + reply_cx.reply_with(|seq_no, reply_no| Reply(seq_no, reply_no, Packet::Exclamation)); + } + Packet::Exclamation => { + println!("!"); + } + } + } +} + +fn receive<'a>(recv: &Receiver, seq: &mut SeqEx<&'a Transport>, transport: &'a Transport) { + match recv.recv().unwrap() { + RawPacket::Ack(reply_no) => seq.receive_ack(reply_no), + RawPacket::EmptyReply(reply_no) => { + seq.receive_empty_reply(reply_no); + } + RawPacket::Send(seq_no, packet) => seq.receive(transport, seq_no, None, packet).unwrap(), + RawPacket::Reply(seq_no, reply_no, packet) => seq.receive(transport, seq_no, Some(reply_no), packet).unwrap(), + } +} +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); + + assert!(seq1.send(&transport1, RawPacket::Send(seq1.seq_no(), Packet::Hello))); + + receive(&recv2, &mut seq2, &transport2); + receive(&recv1, &mut seq1, &transport1); + receive(&recv2, &mut seq2, &transport2); + receive(&recv1, &mut seq1, &transport1); +} diff --git a/src/lib.rs b/src/lib.rs index c434abc..ffd8dcb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,6 +55,8 @@ pub trait TransportLayer: Sized { type SendData; + fn time(&self) -> i64; + fn send(&self, data: &Self::SendData); fn send_ack(&self, reply_no: SeqNo); fn send_empty_reply(&self, reply_no: SeqNo); @@ -194,14 +196,14 @@ 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, current_time: i64) -> bool { + pub fn send(&mut self, app: TL, packet_data: TL::SendData) -> bool { if self.is_full() { return false; } let seq_no = self.next_send_seq_no; self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1); - let next_resent_time = current_time + self.resend_interval; + let next_resent_time = app.time() + self.resend_interval; let entry = self.send_window[seq_no as usize % self.send_window.len()].insert(SendEntry { seq_no, reply_no: None, @@ -213,13 +215,7 @@ impl SeqEx { true } - pub fn receive( - &mut self, - app: TL, - seq_no: SeqNo, - reply_no: Option, - packet: impl IntoRecvData, - ) -> Result { + pub fn receive(&mut self, app: TL, seq_no: SeqNo, reply_no: Option, packet: impl IntoRecvData) -> Result { // We only want to accept packets with seq_nos in the range: // `self.pre_recv_seq_no < seq_no <= self.pre_recv_seq_no + self.recv_window.len()`. // To check that range we compute `seq_no - (self.pre_recv_seq_no + 1)` and check @@ -328,7 +324,8 @@ impl SeqEx { } } - pub fn service(&mut self, app: TL, current_time: i64) -> i64 { + pub fn service(&mut self, app: TL) -> i64 { + let current_time = app.time(); let next_interval = current_time + self.resend_interval; let mut next_activity = next_interval; for item in self.send_window.iter_mut() { @@ -375,14 +372,14 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { pub fn reply_no(&self) -> SeqNo { self.reply_no } - pub fn reply(mut self, packet_data: TL::SendData, current_time: i64) { + pub fn reply(mut self, packet_data: TL::SendData) { if let Some(app) = self.app { let seq_queue = &mut self.seq_queue; let seq_no = seq_queue.next_send_seq_no; seq_queue.next_send_seq_no = seq_queue.next_send_seq_no.wrapping_add(1); let i = seq_no as usize % seq_queue.send_window.len(); - let next_resent_time = current_time + seq_queue.resend_interval; + let next_resent_time = app.time() + seq_queue.resend_interval; let entry = seq_queue.send_window[i].insert(SendEntry { seq_no, reply_no: Some(self.reply_no), @@ -394,6 +391,10 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { self.app = None; } } + pub fn reply_with(self, creator: impl FnOnce(SeqNo, SeqNo) -> TL::SendData) { + let packet_data = creator(self.seq_no(), self.reply_no()); + self.reply(packet_data) + } } impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> { fn drop(&mut self) {