diff --git a/Cargo.lock b/Cargo.lock index cc1b8a4..b49f1c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -19,12 +19,36 @@ dependencies = [ "wasi", ] +[[package]] +name = "itoa" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af150ab688ff2122fcef229be89cb50dd66af9e01a4ff320cc137eecc9bacc38" + [[package]] name = "libc" version = "0.2.147" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4668fb0ea861c1df094127ac5f1da3409a82116a4ba74fca2e58ef927159bb3" +[[package]] +name = "proc-macro2" +version = "1.0.66" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18fb31db3f9bddb2ea821cde30a9f70117e3f119938b5ee630b7403aa6e2ead9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50f3b39ccfb720540debaa0164757101c08ecb8d326b15358ce76a62c7e85965" +dependencies = [ + "proc-macro2", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -34,13 +58,69 @@ dependencies = [ "getrandom", ] +[[package]] +name = "ryu" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad4cc8da4ef723ed60bced201181d83791ad433213d8c24efffda1eec85d741" + [[package]] name = "seq_ex" version = "0.1.0" dependencies = [ "rand_core", + "serde", + "serde_json", ] +[[package]] +name = "serde" +version = "1.0.183" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32ac8da02677876d532745a130fc9d8e6edfa81a269b107c5b00829b91d8eb3c" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.183" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aafe972d60b0b9bee71a91b92fee2d4fb3c9d7e8f6b179aa99f27203d99a4816" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.105" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "693151e1ac27563d6dbcec9dee9fbd5da8539b20fa14ad3752b2e6d363ace360" +dependencies = [ + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "syn" +version = "2.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04361975b3f5e348b2189d8dc55bc942f278b2d482a6a0365de5bdd62d351567" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "unicode-ident" +version = "1.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "301abaae475aa91687eb82514b328ab47a211a533026cb25fc3e519b86adfc3c" + [[package]] name = "wasi" version = "0.11.0+wasi-snapshot-preview1" diff --git a/Cargo.toml b/Cargo.toml index 8cf1979..c97516e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,8 +10,12 @@ path = "src/lib.rs" doc = true [features] -default = ["std"] +default = ["std", "serde"] std = [] +[dependencies] +serde = { version = "1.0.183", default-features = false, features = ["derive"], optional = true } + [dev-dependencies] rand_core = { version = "0.6.4", features = ["getrandom"]} +serde_json = { version = "1.0.104" } diff --git a/examples/calculator.rs b/examples/calculator.rs index fab1805..5955e54 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -1,6 +1,6 @@ use std::{sync::mpsc::Receiver, thread, time::Duration}; -use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync}; +use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, MpscGuard, MpscSeqEx}; #[derive(Clone)] enum Packet { @@ -16,7 +16,7 @@ fn drop_packet() -> bool { rand_core::OsRng.next_u32() & 1 > 0 } -fn process(_: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, _: Option, value: &mut f32) { +fn process(_: MpscGuard<'_, Packet>, recv_packet: Packet, _: Option, value: &mut f32) { use Packet::*; match recv_packet { Add(n) => *value = *value + n, @@ -29,17 +29,17 @@ fn process(_: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, _: Op fn receive<'a>( recv: &Receiver>, - seq: &SeqExSync<&'a MpscTransport>, + seq: &MpscSeqEx, transport: &'a MpscTransport, value: &mut f32, ) { while let Ok(packet) = recv.try_recv() { if !drop_packet() { match packet { - PacketType::Ack { reply_no } => { + PacketType::Ack ( reply_no ) => { let _ = seq.receive_ack(reply_no); } - PacketType::Payload { seq_no, reply_no, payload } => { + PacketType::Payload ( seq_no, reply_no, payload ) => { for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) { process(guard, packet, send_data, value); } @@ -52,8 +52,8 @@ fn receive<'a>( 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 seq1 = MpscSeqEx::new(5, 1); + let seq2 = MpscSeqEx::new(5, 1); let mut value = 0.0; let mut remote_value = value; @@ -68,7 +68,7 @@ fn main() { seq1.send(&transport1, Packet::Mod(5.0)); value %= 5.0; - for _ in 0..30 { + for _ in 0..16 { receive(&recv1, &seq1, &transport1, &mut value); receive(&recv2, &seq2, &transport2, &mut remote_value); thread::sleep(Duration::from_millis(5)); diff --git a/examples/file_download.rs b/examples/file_download.rs new file mode 100644 index 0000000..2ace598 --- /dev/null +++ b/examples/file_download.rs @@ -0,0 +1,157 @@ +use std::{sync::{mpsc::{Receiver, Sender, channel}, Arc, RwLock}, thread, time::{Duration, Instant}, collections::HashMap, ops::Deref}; + +use seq_ex::{sync::{PacketType, RecvSuccess, ReplyGuard, SeqExSync}, TransportLayer, SeqNo}; +use rand_core::{RngCore, OsRng}; +use serde::{Serialize, Deserialize}; + +const FILE_CHUNK_SIZE: usize = 1000; +#[derive(Clone, Debug, Serialize, Deserialize)] +enum Packet { + RequestFile { + filename: String, + }, + ConfirmRequestFile { + filesize: u64, + }, + FileDownload { + filename: String, + file_chunk: Vec, + } +} + +#[derive(Clone)] +struct Transport { + sender: Sender>, + time: Instant, +} +struct Peer { + filesystem: Arc>>>, + transport: Transport, + seqex: Arc>, + receiver: Receiver>, +} + +impl TransportLayer for &Transport { + type SendData = Packet; + + fn time(&mut self) -> i64 { + self.time.elapsed().as_millis() as i64 + } + + fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &Packet) { + let p = PacketType::Payload(seq_no, reply_no, payload.clone()); + if let Ok(p) = serde_json::to_vec(&p) { + let _ = self.sender.send(p); + } + } + + fn send_ack(&mut self, reply_no: SeqNo) { + let p = PacketType::::Ack(reply_no); + if let Ok(p) = serde_json::to_vec(&p) { + let _ = self.sender.send(p); + } + } +} + + +fn drop_packet() -> bool { + OsRng.next_u32() >= (u32::MAX / 4 * 3) +} + +fn process<'a>(peer: &Peer, guard: ReplyGuard<'_, &Transport, Packet, Packet>, recv_packet: Packet, sent_packet: Option) { + match (recv_packet, sent_packet) { + (Packet::RequestFile { filename }, None) => { + let filesystem = peer.filesystem.clone(); + let transport = peer.transport.clone(); + let seqex = peer.seqex.clone(); + if let Some(file) = filesystem.read().unwrap().get(&filename) { + guard.reply(Packet::ConfirmRequestFile { filesize: file.len() as u64 }); + } + thread::spawn(move || { + let filesystem = filesystem.read().unwrap(); + // NOTE: in a real application you need to explicitly handle the situation where the + // file is missing. + if let Some(file) = filesystem.get(&filename) { + let mut i = 0; + while i < file.len() { + let j = file.len().min(i + FILE_CHUNK_SIZE); + seqex.send(&transport, Packet::FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() }); + i = j; + } + } + }); + } + (Packet::ConfirmRequestFile { filesize }, Some(Packet::RequestFile { filename })) => { + let mut filesystem = peer.filesystem.write().unwrap(); + let file = Vec::with_capacity(filesize as usize); + filesystem.insert(filename, file); + } + (Packet::FileDownload { filename, file_chunk }, None) => { + let mut filesystem = peer.filesystem.write().unwrap(); + if let Some(file) = filesystem.get_mut(&filename) { + if file.len() + file_chunk.len() <= file.capacity() { + file.extend(&file_chunk); + } + } + } + _ => { + assert!(false); + } + } +} + +fn receive<'a>( + peer: &Peer, +) { + while let Ok(packet) = peer.receiver.try_recv() { + if drop_packet() { + continue; + } + let parsed_packet = serde_json::from_slice::>(&packet); + match parsed_packet { + Ok(PacketType::Ack ( reply_no )) => { + let _ = peer.seqex.receive_ack(reply_no); + } + Ok(PacketType::Payload ( seq_no, reply_no, payload )) => { + for RecvSuccess { guard, packet, send_data } in peer.seqex.receive_all(&peer.transport, seq_no, reply_no, payload) { + process(peer, guard, packet, send_data); + } + } + _ => {} + } + } +} + + +fn main() { + let mut filesystem2 = HashMap::new(); + let mut file = Vec::from([0u8; 1 << 16]); + OsRng.fill_bytes(&mut file); + filesystem2.insert("File1".to_string(), file); + let mut file = Vec::from([0u8; 1 << 18]); + OsRng.fill_bytes(&mut file); + filesystem2.insert("File2".to_string(), file); + let mut file = Vec::from([0u8; 1 << 20]); + OsRng.fill_bytes(&mut file); + filesystem2.insert("File3".to_string(), file); + + let (send1, recv2) = channel(); + let (send2, recv1) = channel(); + + let peer1 = Peer { filesystem: Arc::new(RwLock::new(HashMap::new())), seqex: Arc::new(SeqExSync::new(5, 1)), transport: Transport{time: Instant::now(), sender: send1}, receiver: recv1 }; + let peer2 = Peer { filesystem: Arc::new(RwLock::new(filesystem2)), seqex: Arc::new(SeqExSync::new(5, 1)), transport: Transport{time: Instant::now(), sender: send2}, receiver: recv2 }; + + peer1.seqex.send(&peer1.transport, Packet::RequestFile{filename: "File1".to_string()}); + peer1.seqex.send(&peer1.transport, Packet::RequestFile{filename: "File3".to_string()}); + peer1.seqex.send(&peer1.transport, Packet::RequestFile{filename: "File2".to_string()}); + + for _ in 0..300 { + receive(&peer1); + receive(&peer2); + thread::sleep(Duration::from_millis(1)); + peer1.seqex.service(&peer1.transport); + peer2.seqex.service(&peer2.transport); + } + + assert_eq!(peer1.filesystem.read().unwrap().deref(), peer2.filesystem.read().unwrap().deref()); +} diff --git a/examples/hello_world.rs b/examples/hello_world.rs index dc291af..cf13b22 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -1,6 +1,6 @@ use std::sync::mpsc::Receiver; -use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync}; +use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, MpscGuard, MpscSeqEx}; #[derive(Clone, Debug)] enum Packet { @@ -11,7 +11,7 @@ enum Packet { } use Packet::*; -fn process(guard: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, send_packet: Option) { +fn process(guard: MpscGuard<'_, Packet>, recv_packet: Packet, send_packet: Option) { match (recv_packet, send_packet) { (Hello, None) => { print!("Hello"); @@ -37,16 +37,16 @@ fn process(guard: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, s } } -fn receive<'a>(recv: &Receiver>, seq: &SeqExSync<&'a MpscTransport>, transport: &'a MpscTransport) { +fn receive(recv: &Receiver>, seq: &MpscSeqEx, transport: &MpscTransport) { match recv.recv().unwrap() { - PacketType::Ack { reply_no } => { + PacketType::Ack ( reply_no ) => { let result = seq.receive_ack(reply_no); if let Ok(Exclamation) = result { // Our Hello World exchange ends right here. print!("\n"); } } - PacketType::Payload { seq_no, reply_no, payload } => { + PacketType::Payload ( seq_no, reply_no, payload ) => { for RecvSuccess { guard, packet, send_data } in seq.receive_all(transport, seq_no, reply_no, payload) { process(guard, packet, send_data) } @@ -57,8 +57,8 @@ fn receive<'a>(recv: &Receiver>, seq: &SeqExSync<&'a MpscTran fn main() { let (transport1, recv2) = MpscTransport::new(); let (transport2, recv1) = MpscTransport::new(); - let seq1 = SeqExSync::default(); - let seq2 = SeqExSync::default(); + let seq1 = MpscSeqEx::default(); + let seq2 = MpscSeqEx::default(); // We begin a "Hello World" exchange right here. seq1.send(&transport1, Packet::Hello); diff --git a/src/seq_queue.rs b/src/seq_queue.rs index fb334c3..57d1cc7 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -48,7 +48,7 @@ pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 1; 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,8 +57,8 @@ pub struct SeqEx { next_send_seq_no: SeqNo, pre_recv_seq_no: SeqNo, /// This could be made more efficient by changing to SoA format. - send_window: [Option>; CAP], - recv_window: [Option>; CAP], + 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], @@ -69,17 +69,17 @@ pub struct SeqEx { concurrent_replies_total: usize, } -struct RecvEntry { +struct RecvEntry { seq_no: SeqNo, reply_no: Option, - data: TL::RecvData, + data: RecvData, } -struct SendEntry { +struct SendEntry { seq_no: SeqNo, reply_no: Option, next_resend_time: i64, - data: TL::SendData, + data: SendData, } /// The error type for when a packet has been received, but for whatever reason could not be @@ -97,7 +97,7 @@ pub enum Error { /// An iterator over all packets in the send window. It will iterate over all packets currently /// being sent to the remote peer. /// These packets are awaiting a reply from the remote peer. -pub struct Iter<'a, TL: TransportLayer>(core::slice::Iter<'a, Option>>); +pub struct Iter<'a, SendData>(core::slice::Iter<'a, Option>>); /// A mutable iterator over all packets in the send window. /// /// The user is able to mutate the contents of the packet being sent to the remote peer, as well as @@ -107,9 +107,9 @@ pub struct Iter<'a, TL: TransportLayer>(core::slice::Iter<'a, Option(core::slice::IterMut<'a, Option>>); +pub struct IterMut<'a, SendData>(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`. @@ -131,10 +131,10 @@ impl SeqEx { concurrent_replies_total: 0, } } - fn send_window_slot_mut(&mut self, seq_no: SeqNo) -> &mut Option> { + fn send_window_slot_mut(&mut self, seq_no: SeqNo) -> &mut Option> { &mut self.send_window[seq_no as usize % self.send_window.len()] } - fn send_window_slot(&self, seq_no: SeqNo) -> &Option> { + fn send_window_slot(&self, seq_no: SeqNo) -> &Option> { &self.send_window[seq_no as usize % self.send_window.len()] } @@ -171,7 +171,7 @@ 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 try_send(&mut self, mut app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> { + pub fn try_send(&mut self, mut app: impl TransportLayer, packet_data: SendData) -> Result<(), SendData> { if self.is_full() { return Err(packet_data); } @@ -192,13 +192,13 @@ impl SeqEx { Ok(()) } - pub fn receive_raw>( + pub fn receive_raw>( &mut self, - mut app: TL, + mut app: impl TransportLayer, seq_no: SeqNo, reply_no: Option, packet: P, - ) -> Result<(SeqNo, P, Option), Error> { + ) -> Result<(SeqNo, P, Option), Error> { // We only want to accept packets with sequence numbers 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 @@ -270,7 +270,7 @@ impl SeqEx { } } } - pub fn receive_ack(&mut self, reply_no: SeqNo) -> Result { + pub fn receive_ack(&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(); @@ -293,7 +293,7 @@ impl SeqEx { false } - fn take_send(&mut self, reply_no: SeqNo) -> Option { + fn take_send(&mut self, reply_no: SeqNo) -> Option { let slot = self.send_window_slot_mut(reply_no); if slot.as_ref().map_or(false, |e| e.seq_no == reply_no) { slot.take().map(|e| e.data) @@ -301,7 +301,7 @@ impl SeqEx { None } } - pub fn pump_raw(&mut self) -> Result<(SeqNo, TL::RecvData, Option), Error> { + pub fn pump_raw(&mut self) -> Result<(SeqNo, RecvData, Option), Error> { let next_seq_no = self.pre_recv_seq_no.wrapping_add(1); let i = next_seq_no as usize % self.recv_window.len(); @@ -329,7 +329,7 @@ impl SeqEx { /// 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_raw(&mut self, mut app: TL, reply_no: SeqNo, packet_data: TL::SendData) { + pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, packet_data: SendData) { if self.remove_reservation(reply_no) { let seq_no = self.next_send_seq_no; self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1); @@ -352,7 +352,7 @@ impl SeqEx { app.send(entry.seq_no, entry.reply_no, &entry.data); } } - pub fn ack_raw(&mut self, mut app: TL, reply_no: SeqNo) { + pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo) { if self.remove_reservation(reply_no) { // Acks are only sent once. There is code in `receive_raw` to handle resending // an ack in the event that the first one here was dropped by the network. @@ -371,7 +371,7 @@ impl SeqEx { false } - pub fn service(&mut self, mut app: TL) -> i64 { + pub fn service(&mut self, mut app: impl TransportLayer) -> i64 { let current_time = app.time(); let next_interval = current_time + self.resend_interval; let mut next_activity = i64::MAX; @@ -388,29 +388,29 @@ impl SeqEx { self.resend_interval.min(next_activity - current_time) } - pub fn iter(&self) -> Iter<'_, TL> { + pub fn iter(&self) -> Iter<'_, SendData> { Iter(self.send_window.iter()) } - pub fn iter_mut(&mut self) -> IterMut<'_, TL> { + pub fn iter_mut(&mut self) -> IterMut<'_, SendData> { IterMut(self.send_window.iter_mut()) } } -impl Default for SeqEx { +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>; +impl<'a, SendData, RecvData, const CAP: usize> IntoIterator for &'a SeqEx { + type Item = &'a SendData; + type IntoIter = Iter<'a, SendData>; fn into_iter(self) -> Self::IntoIter { self.iter() } } -impl<'a, TL: TransportLayer> IntoIterator for &'a mut SeqEx { - type Item = &'a mut TL::SendData; - type IntoIter = IterMut<'a, TL>; +impl<'a, SendData, RecvData, const CAP: usize> IntoIterator for &'a mut SeqEx { + type Item = &'a mut SendData; + type IntoIter = IterMut<'a, SendData>; fn into_iter(self) -> Self::IntoIter { self.iter_mut() @@ -419,8 +419,8 @@ impl<'a, TL: TransportLayer> IntoIterator for &'a mut SeqEx { macro_rules! iterator { ($iter:ident, {$( $mut:tt )?}) => { - impl<'a, TL: TransportLayer> Iterator for $iter<'a, TL> { - type Item = &'a $($mut)? TL::SendData; + impl<'a, SendData> Iterator for $iter<'a, SendData> { + type Item = &'a $($mut)? SendData; fn next(&mut self) -> Option { while let Some(entry) = self.0.next() { if let Some(entry) = entry { @@ -434,7 +434,7 @@ macro_rules! iterator { (0, Some(self.0.len())) } } - impl<'a, TL: TransportLayer> DoubleEndedIterator for $iter<'a, TL> { + impl<'a, SendData> DoubleEndedIterator for $iter<'a, SendData> { fn next_back(&mut self) -> Option { while let Some(entry) = self.0.next_back() { if let Some(entry) = entry { diff --git a/src/single_thread.rs b/src/single_thread.rs index a96eddb..ecc6f84 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,41 +1,41 @@ use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; -pub struct ReplyGuard<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a mut SeqEx, TL, SeqNo); -impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> { +pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a mut SeqEx, TL, SeqNo); +impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, 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, /// 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: TL::SendData) { + pub fn reply(self, packet_data: SendData) { self.0.reply_raw(self.1.clone(), self.2, packet_data); core::mem::forget(self); } } -impl<'a, TL: TransportLayer, const CAP: usize> Drop for ReplyGuard<'a, TL, CAP> { +impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { self.0.ack_raw(self.1.clone(), self.2) } } -pub struct RecvSuccess<'a, TL: TransportLayer, P, const CAP: usize = DEFAULT_WINDOW_CAP> { - pub guard: ReplyGuard<'a, TL, CAP>, +pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { + pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, pub packet: P, - pub send_data: Option, + pub send_data: Option, } -impl SeqEx { - pub fn receive>( +impl SeqEx { + pub fn receive, P: Into>( &mut self, app: TL, seq_no: SeqNo, reply_no: Option, packet: P, - ) -> Result, Error> { + ) -> Result, Error> { self.receive_raw(app.clone(), seq_no, reply_no, packet) .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } - pub fn pump(&mut self, app: TL) -> Result, Error> { + pub fn pump>(&mut self, app: TL) -> Result, Error> { self.pump_raw() .map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data }) } diff --git a/src/sync.rs b/src/sync.rs index 1aed663..153a7cd 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -10,75 +10,75 @@ use std::{ use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP}; -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, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a SeqExSync, TL, SeqNo); -impl<'a, TL: TransportLayer, const CAP: usize> ReplyGuard<'a, TL, CAP> { +pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a SeqExSync, TL, SeqNo); +impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, 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, /// 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: TL::SendData) { + pub fn reply(self, packet_data: 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, const CAP: usize> Drop for ReplyGuard<'a, TL, CAP> { +impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { let mut seq = self.0.seq_ex.lock().unwrap(); seq.ack_raw(self.1.clone(), self.2); } } -pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, const CAP: usize = DEFAULT_WINDOW_CAP> { - pub guard: ReplyGuard<'a, TL, CAP>, +pub struct RecvSuccess<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { + pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, pub packet: P, - pub send_data: Option, + pub send_data: Option, } -pub struct ReplyIter<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> { - origin: Option<&'a SeqExSync>, +pub struct ReplyIter<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { + origin: Option<&'a SeqExSync>, app: TL, - first: Option>, + first: 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(), } } - pub fn receive>( + pub fn receive, P: Into>( &self, app: TL, 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_all(&self, app: TL, seq_no: SeqNo, reply_no: Option, packet: TL::RecvData) -> ReplyIter<'_, TL, CAP> { + pub fn receive_all>(&self, app: TL, seq_no: SeqNo, reply_no: Option, packet: RecvData) -> ReplyIter<'_, TL, SendData, RecvData, CAP> { if let Ok(g) = self.receive(app.clone(), seq_no, reply_no, packet) { ReplyIter { origin: Some(self), app, first: Some(g) } } else { @@ -92,11 +92,11 @@ impl SeqExSync { self.send_block.notify_one(); } } - pub fn try_send(&self, app: TL, packet_data: TL::SendData) -> Result<(), TL::SendData> { + pub fn try_send>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> { let mut seq = self.lock(); seq.try_send(app, packet_data) } - pub fn send(&self, app: TL, mut packet_data: TL::SendData) { + pub fn send>(&self, app: TL, mut packet_data: SendData) { let mut seq = self.lock(); while let Err(p) = seq.try_send(app.clone(), packet_data) { packet_data = p; @@ -106,27 +106,29 @@ impl SeqExSync { } } - pub fn receive_ack(&self, reply_no: SeqNo) -> Result { + pub fn receive_ack(&self, reply_no: SeqNo) -> Result { let ret = self.lock().receive_ack(reply_no); self.unblock(ret.is_ok()); ret } - pub fn service(&self, app: TL) -> i64 { + 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() } } -impl Default for SeqExSync { +impl Default for SeqExSync { fn default() -> Self { Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO) } } +unsafe impl Send for SeqExSync {} +unsafe impl Sync for SeqExSync {} -impl<'a, TL: TransportLayer, const CAP: usize> Iterator for ReplyIter<'a, TL, CAP> { - type Item = RecvSuccess<'a, TL, TL::RecvData, CAP>; +impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Iterator for ReplyIter<'a, TL, SendData, RecvData, CAP> { + type Item = RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>; fn next(&mut self) -> Option { if let Some(g) = self.first.take() { Some(g) @@ -138,16 +140,11 @@ impl<'a, TL: TransportLayer, const CAP: usize> Iterator for ReplyIter<'a, TL, CA } } +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] #[derive(Clone)] pub enum PacketType { - Payload { - seq_no: SeqNo, - reply_no: Option, - payload: Payload, - }, - Ack { - reply_no: SeqNo, - }, + Payload(SeqNo, Option, Payload), + Ack(SeqNo), } #[derive(Clone)] @@ -155,6 +152,9 @@ pub struct MpscTransport { pub channel: Sender>, pub time: Instant, } +pub type MpscGuard<'a, Packet> = ReplyGuard<'a, &'a MpscTransport, Packet, Packet>; +pub type MpscSeqEx = SeqExSync; + impl MpscTransport { pub fn new() -> (Self, Receiver>) { let (send, recv) = channel(); @@ -165,7 +165,6 @@ impl MpscTransport { } } impl TransportLayer for &MpscTransport { - type RecvData = Payload; type SendData = Payload; fn time(&mut self) -> i64 { @@ -173,9 +172,9 @@ 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() }); + let _ = self.channel.send(PacketType::Payload(seq_no, reply_no, payload.clone() )); } fn send_ack(&mut self, reply_no: SeqNo) { - let _ = self.channel.send(PacketType::Ack { reply_no }); + let _ = self.channel.send(PacketType::Ack ( reply_no )); } } diff --git a/src/transport_layer.rs b/src/transport_layer.rs index 4e3c312..00782fe 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -7,7 +7,6 @@ use crate::SeqNo; /// 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: Clone { - type RecvData; type SendData; fn time(&mut self) -> i64;