diff --git a/Cargo.lock b/Cargo.lock index 05f29ad..887d4c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -25,12 +25,6 @@ version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eabb4a44450da02c90444cf74558da904edde8fb4e9035a9a6a4e15445af0bd7" -[[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" @@ -64,12 +58,6 @@ 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" @@ -77,7 +65,6 @@ dependencies = [ "rand_core", "serde", "serde_cbor", - "serde_json", ] [[package]] @@ -110,17 +97,6 @@ dependencies = [ "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" diff --git a/Cargo.toml b/Cargo.toml index 6b7e6b5..44a7e00 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,5 +19,4 @@ serde = { version = "1.0.183", default-features = false, features = ["derive"], [dev-dependencies] rand_core = { version = "0.6.4", features = ["getrandom"]} -serde_json = { version = "1.0.104" } serde_cbor = { version = "0.11.2" } diff --git a/async_file_share.rs b/async_file_share.rs index 0c54955..f9ea0ae 100644 --- a/async_file_share.rs +++ b/async_file_share.rs @@ -13,7 +13,7 @@ use std::{ }; use rand_core::{OsRng, RngCore}; -use seq_ex::{sync::RecvSuccess, SeqNo, TransportLayer}; +use seq_ex::{SeqNo, TransportLayer, Packet}; use serde::{Deserialize, Serialize}; /// serde_cbor minimal format is both smaller and faster than default format. @@ -41,7 +41,7 @@ enum SendData { } #[derive(Serialize, Deserialize)] -enum Packet<'a> { +enum Payload<'a> { RequestFile { filename: &'a str }, ConfirmFileSize { filesize: u64 }, ConfirmDownload { fileid: u64 }, @@ -52,16 +52,18 @@ enum Packet<'a> { } const PACKET_TYPE_PAYLOAD: u8 = 0; -const PACKET_TYPE_REPLY: u8 = 1; -const PACKET_TYPE_ACK: u8 = 2; +const PACKET_TYPE_LOCK_PAYLOAD: u8 = 1; +const PACKET_TYPE_REPLY: u8 = 2; +const PACKET_TYPE_LOCK_REPLY: u8 = 3; +const PACKET_TYPE_ACK: u8 = 4; -fn create_payload(seq_no: SeqNo, packet: &Packet<'_>) -> Vec { +fn create_payload(seq_no: SeqNo, packet: &Payload<'_>) -> Vec { let mut p = vec![PACKET_TYPE_PAYLOAD]; p.extend(&seq_no.to_be_bytes()); to_writer_minimal(packet, &mut p); p } -fn create_reply(seq_no: SeqNo, reply_no: SeqNo, packet: &Packet<'_>) -> Vec { +fn create_reply(seq_no: SeqNo, reply_no: SeqNo, packet: &Payload<'_>) -> Vec { let mut p = vec![PACKET_TYPE_REPLY]; p.extend(&seq_no.to_be_bytes()); p.extend(&reply_no.to_be_bytes()); @@ -99,29 +101,33 @@ impl TransportLayer<(SendData, Vec)> for &Transport { self.time.elapsed().as_millis() as i64 } - fn send(&mut self, _: SeqNo, _: Option, (_, packet): &(SendData, Vec)) { - let _ = self.sender.send(packet.clone()); - } - fn send_ack(&mut self, reply_no: SeqNo) { - let mut p = vec![PACKET_TYPE_ACK]; - p.extend(&reply_no.to_be_bytes()); - self.sender.send(p); + fn send(&mut self, packet: Packet<&(SendData, Vec)>) { + let p = match packet.consume() { + Ok((_, packet)) => packet.clone(), + Err(reply_no) => { + let mut p = Vec::with_capacity(5); + p.push(PACKET_TYPE_ACK); + p.extend(&reply_no.to_be_bytes()); + p + } + }; + let _ = self.sender.send(p); } } -fn process(peer: &Arc, transport: &Transport, guard: ReplyGuard<'_>, payload: Packet<'_>, data: Option) -> Option<()> { +fn process(peer: &Arc, transport: &Transport, guard: ReplyGuard<'_>, payload: Payload<'_>, data: Option) -> Option<()> { match (payload, data) { - (Packet::RequestFile { filename }, None) => { + (Payload::RequestFile { filename }, None) => { let file = File::open(peer.home_dir.join(filename)).ok()?; let metadata = file.metadata().ok()?; let filesize = metadata.len(); reply!( guard, SendData::ConfirmFileSize { file }, - Packet::ConfirmFileSize { filesize } + Payload::ConfirmFileSize { filesize } ); } - (Packet::ConfirmFileSize { filesize }, Some(SendData::RequestFile { filename })) => { + (Payload::ConfirmFileSize { filesize }, Some(SendData::RequestFile { filename })) => { let path = peer.home_dir.join(filename); if filesize > DOWNLOAD_LIMIT { return None; @@ -129,9 +135,9 @@ fn process(peer: &Arc, transport: &Transport, guard: ReplyGuard<'_>, paylo let file = File::create(&path).ok()?; let fileid = OsRng.next_u64(); peer.downloads_in_progress.insert(fileid, file); - reply!(guard, SendData::ConfirmDownload, Packet::ConfirmDownload { fileid }); + reply!(guard, SendData::ConfirmDownload, Payload::ConfirmDownload { fileid }); } - (Packet::ConfirmDownload { fileid }, Some(SendData::ConfirmFileSize { mut file })) => { + (Payload::ConfirmDownload { fileid }, Some(SendData::ConfirmFileSize { mut file })) => { let peer = peer.clone(); let transport = transport.clone(); thread::spawn(move || { @@ -148,7 +154,7 @@ fn process(peer: &Arc, transport: &Transport, guard: ReplyGuard<'_>, paylo peer, &transport, SendData::FileDownload, - Packet::FileDownload { fileid, file_chunk: &buffer[i..j] } + Payload::FileDownload { fileid, file_chunk: &buffer[i..j] } ); i = j; } @@ -157,18 +163,18 @@ fn process(peer: &Arc, transport: &Transport, guard: ReplyGuard<'_>, paylo peer, &transport, SendData::FileDownload, - Packet::FileDownloadComplete { fileid } + Payload::FileDownloadComplete { fileid } ); }); } - (Packet::FileDownload { fileid, file_chunk }, None) => { + (Payload::FileDownload { fileid, file_chunk }, None) => { let mut file = peer.downloads_in_progress.get(&fileid)?; let result = file.write_all(file_chunk); } - (Packet::FileDownloadComplete { fileid }, None) => { + (Payload::FileDownloadComplete { fileid }, None) => { } - (Packet::ReadDir, None) => { + (Payload::ReadDir, None) => { let dir = read_dir(&peer.home_dir).ok()?; let mut filenames = Vec::new(); for entry in dir { @@ -183,9 +189,9 @@ fn process(peer: &Arc, transport: &Transport, guard: ReplyGuard<'_>, paylo } } let filenames: Vec<&str> = filenames.iter().map(|f| f.as_str()).collect(); - reply!(guard, SendData::DirContents, Packet::DirContents { filenames }); + reply!(guard, SendData::DirContents, Payload::DirContents { filenames }); } - (Packet::DirContents { filenames }, Some(SendData::ReadDir { download_missing: true })) => { + (Payload::DirContents { filenames }, Some(SendData::ReadDir { download_missing: true })) => { drop(guard); for filename in filenames {} } @@ -199,30 +205,29 @@ fn receive(peer: &Arc, transport: &Transport, receiver: &Receiver> if drop_packet() { continue; } - let iter = match *packet.get(0)? { + let pt = *packet.get(0)?; + let (mut parsed_packet, payload_offset) = match pt { PACKET_TYPE_ACK => { let reply_no = SeqNo::from_be_bytes(packet.get(1..5)?.try_into().ok()?); - let _ = peer.seqex.receive_ack(reply_no); - return None; + (Packet::Ack(reply_no), 5) } - PACKET_TYPE_PAYLOAD => { + PACKET_TYPE_PAYLOAD | PACKET_TYPE_LOCK_PAYLOAD => { let seq_no = SeqNo::from_be_bytes(packet.get(1..5)?.try_into().ok()?); - peer.seqex.receive_all(transport, seq_no, None, packet) + (Packet::Payload(seq_no, packet), 5) } - PACKET_TYPE_REPLY => { + PACKET_TYPE_REPLY | PACKET_TYPE_LOCK_REPLY => { let seq_no = SeqNo::from_be_bytes(packet.get(1..5)?.try_into().ok()?); let reply_no = SeqNo::from_be_bytes(packet.get(5..9)?.try_into().ok()?); - peer.seqex.receive_all(transport, seq_no, Some(reply_no), packet) + (Packet::Reply(seq_no, reply_no, packet), 9) } _ => return None, }; - for RecvSuccess { guard, packet, send_data } in iter { - let offset = match packet[0] { - PACKET_TYPE_PAYLOAD => 5, - PACKET_TYPE_REPLY => 9, - _ => return None, - }; - if let Ok(parsed_packet) = serde_cbor::from_slice::(packet.get(offset..)?) { + if pt == PACKET_TYPE_LOCK_PAYLOAD || pt == PACKET_TYPE_LOCK_REPLY { + parsed_packet.set_locking(true); + } + for recv_data in peer.seqex.receive_all(transport, parsed_packet) { + let recv_data = recv_data.map(|packet| serde_cbor::from_slice::(&packet[payload_offset..])); + if let Ok(parsed_payload) = serde_cbor::from_slice::(packet.get(payload_offset..)?) { process(peer, transport, guard, parsed_packet, send_data.map(|d| d.0)); } } diff --git a/examples/file_download.rs b/examples/file_download.rs index 5356b69..c39bb96 100644 --- a/examples/file_download.rs +++ b/examples/file_download.rs @@ -42,7 +42,7 @@ impl TransportLayer for &Transport { } fn send(&mut self, packet: Packet<&Payload>) { - if let Ok(p) = serde_json::to_vec(&packet) { + if let Ok(p) = serde_cbor::to_vec(&packet) { let _ = self.sender.send(p); } } @@ -101,7 +101,7 @@ fn receive(peer: &Peer) { if drop_packet() { continue; } - if let Ok(parsed_packet) = serde_json::from_slice::>(&packet) { + if let Ok(parsed_packet) = serde_cbor::from_slice::>(&packet) { for recv_data in peer.seqex.receive_all(&peer.transport, parsed_packet) { process(peer, recv_data); } @@ -141,7 +141,7 @@ fn main() { peer1.seqex.send(&peer1.transport, Payload::RequestFile { filename: "File3".to_string() }); peer1.seqex.send(&peer1.transport, Payload::RequestFile { filename: "File2".to_string() }); - for _ in 0..300 { + for _ in 0..500 { receive(&peer1); receive(&peer2); thread::sleep(Duration::from_millis(1)); diff --git a/src/seq_queue.rs b/src/seq_queue.rs index 9f4f081..a876696 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -148,9 +148,12 @@ impl Packet { } } pub fn payload(self) -> Option { + self.consume().ok() + } + pub fn consume(self) -> Result { match self { - Payload(_, data) | LockPayload(_, data) | Reply(_, _, data) | LockReply(_, _, data) => Some(data), - Ack(_) => None, + Payload(_, data) | LockPayload(_, data) | Reply(_, _, data) | LockReply(_, _, data) => Ok(data), + Ack(r) => Err(r), } } pub fn is_locking(&self) -> bool { diff --git a/src/single_thread.rs b/src/single_thread.rs index 813754a..34d30e9 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -126,6 +126,13 @@ macro_rules! impl_recvok { (None, None) => None, } } + pub fn map(self, f: impl FnOnce(P) -> R) -> $recv<'a, TL, R, SendData, RecvData, CAP> { + match self { + Self::Payload { reply_guard, recv_data } => $recv::Payload { reply_guard, recv_data: f(recv_data) }, + Self::Reply { reply_guard, recv_data, send_data } => $recv::Reply { reply_guard, recv_data: f(recv_data), send_data }, + Self::Ack { send_data } => $recv::Ack { send_data }, + } + } } impl<'a, TL: TransportLayer, P: Into, SendData, RecvData, const CAP: usize> $recv<'a, TL, P, SendData, RecvData, CAP> { pub fn into(self) -> $recv<'a, TL, RecvData, SendData, RecvData, CAP> {