zerotier/sequential-exchange
0
0
Fork 0
mirror of https://github.com/zerotier/sequential-exchange.git synced 2026-05-22 16:28:28 -07:00
Code Issues Packages Projects Releases Wiki Activity

fixed cbor glitch

Browse Source
This commit is contained in:
Monica Moniot
2023-08-18 12:25:33 -04:00
parent a2e76576a7
commit 9005ca760f
6 changed files with 60 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-24
View File
@@ -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"
Cargo.toml
-1
View File
@@ -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" }
async_file_share.rs
+45 -40
View File
@@ -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<u8> {
fn create_payload(seq_no: SeqNo, packet: &Payload<'_>) -> Vec<u8> {
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<u8> {
fn create_reply(seq_no: SeqNo, reply_no: SeqNo, packet: &Payload<'_>) -> Vec<u8> {
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<u8>)> for &Transport {
self.time.elapsed().as_millis() as i64
}
fn send(&mut self, _: SeqNo, _: Option<SeqNo>, (_, packet): &(SendData, Vec<u8>)) {
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<u8>)>) {
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<Peer>, transport: &Transport, guard: ReplyGuard<'_>, payload: Packet<'_>, data: Option<SendData>) -> Option<()> {
fn process(peer: &Arc<Peer>, transport: &Transport, guard: ReplyGuard<'_>, payload: Payload<'_>, data: Option<SendData>) -> 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<Peer>, 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<Peer>, 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<Peer>, 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<Peer>, 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<Peer>, transport: &Transport, receiver: &Receiver<Vec<u8>>
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>(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::<Payload>(&packet[payload_offset..]));
if let Ok(parsed_payload) = serde_cbor::from_slice::<Payload>(packet.get(payload_offset..)?) {
process(peer, transport, guard, parsed_packet, send_data.map(|d| d.0));
}
}
examples/file_download.rs
+3 -3
View File
@@ -42,7 +42,7 @@ impl TransportLayer<Payload> 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<Payload>>(&packet) {
if let Ok(parsed_packet) = serde_cbor::from_slice::<Packet<Payload>>(&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));
src/seq_queue.rs
+5 -2
View File
@@ -148,9 +148,12 @@ impl<RecvData> Packet<RecvData> {
}
}
pub fn payload(self) -> Option<RecvData> {
self.consume().ok()
}
pub fn consume(self) -> Result<RecvData, SeqNo> {
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 {
src/single_thread.rs
+7
View File
@@ -126,6 +126,13 @@ macro_rules! impl_recvok {
(None, None) => None,
}
}
pub fn map<R>(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<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize> $recv<'a, TL, P, SendData, RecvData, CAP> {
pub fn into(self) -> $recv<'a, TL, RecvData, SendData, RecvData, CAP> {