From 06aa1f72a4348d235751c3bdea95a1b9a32e1de1 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Mon, 16 Oct 2023 13:41:33 -0400 Subject: [PATCH 1/3] refactored repo and added docs --- Cargo.lock | 2 +- Cargo.toml | 23 +++- README.md | 16 ++- examples/calculator.rs | 10 +- examples/file_download.rs | 10 +- examples/hello_world.rs | 10 +- examples/hello_world_tokio.rs | 10 +- phi_accural.rs | 145 --------------------- src/lib.rs | 54 +++++++- src/{seq_queue.rs => no_std.rs} | 175 +------------------------ src/result.rs | 113 ++++++++++++++++ src/single_thread.rs | 107 ++++++++------- src/sync.rs | 222 +++++++++++++++++++++++--------- src/tokio.rs | 104 +++++++++------ src/transport_layer.rs | 191 ++++++++++++++++++++++++++- 15 files changed, 688 insertions(+), 504 deletions(-) delete mode 100644 phi_accural.rs rename src/{seq_queue.rs => no_std.rs} (75%) create mode 100644 src/result.rs diff --git a/Cargo.lock b/Cargo.lock index 86229c1..8788b3f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -234,7 +234,7 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" [[package]] name = "seq-ex" -version = "0.1.0" +version = "1.0.0" dependencies = [ "rand_core", "serde", diff --git a/Cargo.toml b/Cargo.toml index 9acb63d..f6f7f85 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,16 +1,16 @@ [package] name = "seq-ex" -version = "0.1.0" +version = "1.0.0" authors = ["Monica Moniot"] edition = "2021" [lib] -name = "seq_ex" +name = "seqex" path = "src/lib.rs" doc = true [features] -default = ["serde", "tokio"] +default = ["serde", "std"] tokio = ["std", "dep:tokio"] std = [] @@ -22,3 +22,20 @@ tokio = { version = "1.31.0", default-features = false, features = ["sync", "tim rand_core = { version = "0.6.4", features = ["getrandom"]} serde_cbor = { version = "0.11.2" } tokio = { version = "1.31.0", default-features = false, features = ["full"] } + +[[example]] +name = "hello_world" +path = "examples/hello_world.rs" + +[[example]] +name = "hello_world_tokio" +path = "examples/hello_world_tokio.rs" +required-features = ["tokio"] + +[[example]] +name = "calculator" +path = "examples/calculator.rs" + +[[example]] +name = "file_download" +path = "examples/file_download.rs" diff --git a/README.md b/README.md index 8444cc6..cbccc0d 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,20 @@ # Sequential Exchange Protocol -The reference implementation of the **Sequential Exchange Protocol**, or SEP. +The reference implementation of the **Sequential Exchange Protocol**, or SEQEX. -SEP is a lightweight, peer-to-peer transport protocol that guarantees packets of data will be losslessly received by the remote peer, and can optionally guaranteed that specified packets arrive in the order that they were sent. In addition, SEP facilitates stateful exchanges between two peers, giving each peer the opportunity to "reply" to any packet sent by the remote peer. This makes SEP particularly well-suited for writing async-await code, because unlike TCP, SEP will handle multiplexing each reply to the correct awaiter. Even without async-await, a simple `match` statement is sufficient to correctly multiplex packets to their handling code. +SEQEX is a lightweight, peer-to-peer transport protocol that guarantees packets of data will be losslessly received by the remote peer, and can optionally guaranteed that specified packets arrive in the order that they were sent. In addition, SEQEX facilitates stateful exchanges between two peers, giving each peer the opportunity to "reply" to any packet sent by the remote peer. This makes SEQEX particularly well-suited for writing async-await code, because unlike TCP, SEQEX will handle multiplexing each reply to the correct awaiter. Even without async-await, a simple `match` statement is sufficient to correctly multiplex packets to their handling code. A "stateful exchange" is defined here as a sequence of packets, where the first packet initiates the exchange, and all subsequent packets are replies to the previous packet in the exchange. Every exchange can be thought of as a linked list, where the head node is a packet containing a normal payload, and all subsequent nodes are replies to previous nodes. The final node is always a simple acknowledgement packet, or Ack, that signals a given exchange is over. Both peers are guaranteed to agree upon the "topology" of these links. Links will never get crossed, replies will always be received and understood, a peer will never deadlock awaiting a reply, and in general it is much easier to write bug-free networking code. -SEP is a tiny, dead simple protocol and we have implemented it here in less than 500 lines of code. +SEQEX is a tiny, dead simple protocol and we have implemented it here in around a 1000 lines of code, depending upon how many features you enable. -SEP is transport agnostic, and will not take over an entire UDP socket. As such it is relatively easier to run SEP in parrallel with raw UDP, or even with itself. Multiple instances of SEP can be opened between two peers and communication over each can occur in parrallel, making it very easy to reduce or even eliminate front-of-line latency in performance critical applications. +SEQEX is transport agnostic. It does not require being run over a single UDP socket. This allows SEQEX to easily be run over an encrypted tunnel that itself can run over as many or as few UDP sockets as necessary, if indeed UDP is even available. An instance of the SEQEX protocol can be forced to persist through a connection reset event, avoiding common issues with TCP where connection resets can cause unrecoverable packet loss. + +SEQEX is serialization agnostic, meaning its packets have no pre-defined encoding format. Users of SEQEX are free to choose between serde, packed structs, tagged unions, or anything else as their prefered serialization format. This means SEQEX takes some additional effort to set up up-front, but it means that user have significantly more flexibility long-term. + +As such it is relatively easier to run SEQEX in parrallel with another raw UDP protocol, or even in parrallel with itself. Multiple instances of SEQEX can be opened between two peers, making it very easy to reduce or even eliminate head-of-line latency in performance critical applications. ## Why not TCP? @@ -23,9 +27,9 @@ it has a larger amount of metadata that must be transported with packets, and it features that slow down runtime regardless of whether or not they are used. A lot of this overhead owes to TCPs sizeable complexity. -That being said SEP does lack many of TCP's additional features, such as a dynamic resend timer, +That being said SEQEX does lack many of TCP's additional features, such as a dynamic resend timer, keep-alives, and fragmentation. This can be both a pro and a con, as it means there is a lot of efficiency to be gained if these features are not needed or are implemented at a different protocol layer. -Neither SEP nor TCP are cryptographically secure. +Neither SEQEX nor TCP are cryptographically secure. diff --git a/examples/calculator.rs b/examples/calculator.rs index bce0892..4fd9a40 100644 --- a/examples/calculator.rs +++ b/examples/calculator.rs @@ -1,7 +1,7 @@ use std::{sync::mpsc::Receiver, thread, time::Duration}; -use seq_ex::{ - sync::{MpscTransport, SeqExSync}, +use seqex::{ + sync::{MpscTransport, SeqEx}, Packet, }; @@ -19,7 +19,7 @@ fn drop_packet() -> bool { rand_core::OsRng.next_u32() & 1 > 0 } -fn receive(recv: &Receiver>, seq: &SeqExSync, transport: &MpscTransport, value: &mut f32) { +fn receive(recv: &Receiver>, seq: &SeqEx, transport: &MpscTransport, value: &mut f32) { while let Ok(packet) = recv.try_recv() { if drop_packet() { continue; @@ -42,8 +42,8 @@ fn receive(recv: &Receiver>, seq: &SeqExSync, 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 = SeqEx::new(5, 1); + let seq2 = SeqEx::new(5, 1); let mut value = 0.0; let mut remote_value = value; diff --git a/examples/file_download.rs b/examples/file_download.rs index 034352b..931570a 100644 --- a/examples/file_download.rs +++ b/examples/file_download.rs @@ -10,8 +10,8 @@ use std::{ }; use rand_core::{OsRng, RngCore}; -use seq_ex::{ - sync::{RecvOk, SeqExSync}, +use seqex::{ + sync::{RecvOk, SeqEx}, Packet, TransportLayer, }; use serde::{Deserialize, Serialize}; @@ -32,7 +32,7 @@ struct Transport { struct Peer { filesystem: Arc>>>, transport: Transport, - seqex: Arc>, + seqex: Arc>, receiver: Receiver>, } @@ -130,13 +130,13 @@ fn main() { let peer1 = Peer { filesystem: Arc::new(RwLock::new(HashMap::new())), - seqex: Arc::new(SeqExSync::new(5, 1)), + seqex: Arc::new(SeqEx::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)), + seqex: Arc::new(SeqEx::new(5, 1)), transport: Transport { time: Instant::now(), sender: send2 }, receiver: recv2, }; diff --git a/examples/hello_world.rs b/examples/hello_world.rs index 5b249e4..d084466 100644 --- a/examples/hello_world.rs +++ b/examples/hello_world.rs @@ -1,7 +1,7 @@ use std::sync::mpsc::Receiver; -use seq_ex::{ - sync::{MpscTransport, RecvOk, SeqExSync}, +use seqex::{ + sync::{MpscTransport, RecvOk, SeqEx}, Packet, }; @@ -14,7 +14,7 @@ enum Payload { } use Payload::*; -fn receive(recv: &Receiver>, seq: &SeqExSync, transport: &MpscTransport) { +fn receive(recv: &Receiver>, seq: &SeqEx, transport: &MpscTransport) { let packet = recv.recv().unwrap(); for recv_data in seq.receive_all(transport, packet) { match recv_data.consume() { @@ -48,8 +48,8 @@ fn receive(recv: &Receiver>, seq: &SeqExSync, fn main() { let (transport1, recv2) = MpscTransport::new(); let (transport2, recv1) = MpscTransport::new(); - let seq1 = SeqExSync::default(); - let seq2 = SeqExSync::default(); + let seq1 = SeqEx::default(); + let seq2 = SeqEx::default(); // We begin a "Hello World" exchange right here. seq1.send(&transport1, false, Payload::Hello); diff --git a/examples/hello_world_tokio.rs b/examples/hello_world_tokio.rs index 14df8a2..a181ce3 100644 --- a/examples/hello_world_tokio.rs +++ b/examples/hello_world_tokio.rs @@ -2,8 +2,8 @@ use std::sync::Arc; use tokio::{sync::mpsc, task}; -use seq_ex::{ - tokio::{MpscTransport, ReplyGuard, SeqExTokio}, +use seqex::{ + tokio::{MpscTransport, ReplyGuard, SeqEx}, Packet, }; @@ -42,7 +42,7 @@ async fn receive(reply_guard: ReplyGuard<'_, &MpscTransport, Payload, P Some(()) } -async fn say_hello(seq: &SeqExTokio, transport: &MpscTransport) -> Option<()> { +async fn say_hello(seq: &SeqEx, transport: &MpscTransport) -> Option<()> { let (reply_guard, payload) = seq.send(transport, false, Hello).await.ok()?; if payload != Space { return None; @@ -59,8 +59,8 @@ async fn say_hello(seq: &SeqExTokio, transport: &MpscTransport Some(()) } -fn peer_main(transport: MpscTransport, mut recv: mpsc::Receiver>) -> Arc> { - let (seq, mut service) = SeqExTokio::new_default(); +fn peer_main(transport: MpscTransport, mut recv: mpsc::Receiver>) -> Arc> { + let (seq, mut service) = SeqEx::new_default(); let peer = Arc::new(seq); let peer_weak = Arc::downgrade(&peer); let tl = transport.clone(); diff --git a/phi_accural.rs b/phi_accural.rs deleted file mode 100644 index c13d08a..0000000 --- a/phi_accural.rs +++ /dev/null @@ -1,145 +0,0 @@ -use std::sync::mpsc::Receiver; - -use seq_ex::sync::{MpscTransport, PacketType, RecvSuccess, ReplyGuard, SeqExSync}; - -#[derive(Clone, Debug)] -enum Packet { - Hello, - Space, - World, - Exclamation, -} -use Packet::*; - -fn process(guard: ReplyGuard<'_, &MpscTransport>, recv_packet: Packet, send_packet: Option) { - match (recv_packet, send_packet) { - (Hello, None) => { - print!("Hello"); - guard.reply(Space); - } - (Space, Some(Hello)) => { - print!(" "); - guard.reply(World); - } - (World, Some(Space)) => { - print!("World"); - guard.reply(Exclamation); - } - (Exclamation, Some(World)) => { - print!("!"); - } - (a, None) => { - print!("Unsolicited packet received: {:?}", a); - } - (a, Some(b)) => { - print!("Incorrect reply received: {:?}, was a reply to: {:?}", a, b); - } - } -} - -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"); - } - result.is_some() - } - PacketType::Payload { seq_no, reply_no, payload } => { - if let Ok(RecvSuccess { guard, packet, send_data }) = seq.receive(transport, seq_no, reply_no, payload) { - process(guard, packet, send_data); - true - } else { - false - } - } - }; - if do_pump { - while let Ok(RecvSuccess { guard, packet, send_data }) = seq.pump(transport) { - process(guard, packet, send_data); - } - } -} - -pub const WINDOW_SIZE: usize = 32; -pub const DEFAULT_ALLOWED_MISSES: f64 = 1.0; -pub const DEFAULT_ALLOWED_PROB: f64 = .01; - -/// This version of phi accural takes into account the possibility of packets being dropped uniformly at random from the network, and computes an approximation of that cdf. We do not attempt to dynamically compute the loss rate (since packet loss is not uniform or independent irl), but instead require the user preprogram an `allowed_misses` parameter. -/// `allowed_misses` is an estimation of the number of phi accural packets that the user thinks could possibly be dropped in a row given that both the network and the remote peer are still alive. -/// -/// The exact distribution we simulate is the probability that the peer is dead, given the amount of time since the last received phi accural packet, and given that `allowed_misses` number of phi accural packets have been or will be dropped from the network. -#[derive(Clone)] -pub struct PhiAccumulator { - pub allowed_misses: f64, - pub allowed_prob: f64, - head_idx: usize, - intervals: [f64; WINDOW_SIZE], - last_time: i64, - mean: f64, - std: f64, -} - -pub fn normal_cdf_apprx(position: f64, mean: f64, std: f64) -> f64 { - -} - -impl PhiAccumulator { - pub fn new(allowed_prob_of_failure: f64, allowed_misses: f64, expected_first_interval: f64, current_time: i64) -> Self { - PhiAccumulator { - allowed_misses, - allowed_prob: allowed_prob_of_failure, - head_idx: 0, - intervals: std::array::from_fn(|_| expected_first_interval), - last_time: current_time, - mean: expected_first_interval, - std: 0.0, - } - } - /// Returns false if it is likely that the remote peer is dead or unreachable. - pub fn check(&self, current_time: i64) -> bool { - let prob = normal_cdf_apprx((current_time - self.last_time) as f32, (self.allowed_misses + 1.0)*self.mean, self.std); - prob > self.allowed_prob - } - /// Updates the internal state to acknowledge a just received phi accural packet. - pub fn just_received_phi_packet(&mut self, current_time: i64) { - let new_interval = (current_time - self.last_time) as f64; - self.last_time = current_time; - let idx = self.head_idx; - self.head_idx += 1; - // We compute a rolling mean, which is fast but suceptible to rounding errors. Hence f64. - self.mean += (new_interval - self.intervals[idx])/WINDOW_SIZE as f64; - self.intervals[idx] = new_interval; - let mut std = 0.0; - for x in self.intervals { - let diff = (x - self.mean); - std += diff*diff; - } - /// - self.std = std.sqrt()/WINDOW_SIZE as f64; - } -} - - - -fn main() { - let (transport1, recv2) = MpscTransport::new(); - let (transport2, recv1) = MpscTransport::new(); - let seq1 = SeqExSync::default(); - let seq2 = SeqExSync::default(); - - // We begin a "Hello World" exchange right here. - seq1.send(&transport1, Packet::Hello); - - receive(&recv2, &seq2, &transport2); - receive(&recv1, &seq1, &transport1); - receive(&recv2, &seq2, &transport2); - receive(&recv1, &seq1, &transport1); - receive(&recv2, &seq2, &transport2); -} diff --git a/src/lib.rs b/src/lib.rs index 6a331a3..9184a5d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,17 +1,61 @@ -//#![no_std] -//#![warn(missing_docs, rust_2018_idioms)] +//! # Sequential Exchange Protocol +//! +//! The reference implementation of the **Sequential Exchange Protocol**, or SEQEX. +//! +//! SEQEX is a lightweight, peer-to-peer transport protocol that guarantees packets of data will be losslessly received by the remote peer, and can optionally guaranteed that specified packets arrive in the order that they were sent. In addition, SEQEX facilitates stateful exchanges between two peers, giving each peer the opportunity to "reply" to any packet sent by the remote peer. This makes SEQEX particularly well-suited for writing async-await code, because unlike TCP, SEQEX will handle multiplexing each reply to the correct awaiter. Even without async-await, a simple `match` statement is sufficient to correctly multiplex packets to their handling code. +//! +//! A "stateful exchange" is defined here as a sequence of packets, where the first packet +//! initiates the exchange, and all subsequent packets are replies to the previous packet in the +//! exchange. Every exchange can be thought of as a linked list, where the head node is a packet containing a normal payload, and all subsequent nodes are replies to previous nodes. The final node is always a simple acknowledgement packet, or Ack, that signals a given exchange is over. Both peers are guaranteed to agree upon the "topology" of these links. Links will never get crossed, replies will always be received and understood, a peer will never deadlock awaiting a reply, and in general it is much easier to write bug-free networking code. +//! +//! SEQEX is a tiny, dead simple protocol and we have implemented it here in around a 1000 lines of code, depending upon how many features you enable. +//! +//! SEQEX is transport agnostic. It does not require being run over a single UDP socket. This allows SEQEX to easily be run over an encrypted tunnel that itself can run over as many or as few UDP sockets as necessary, if indeed UDP is even available. An instance of the SEQEX protocol can be forced to persist through a connection reset event, avoiding common issues with TCP where connection resets can cause unrecoverable packet loss. +//! +//! SEQEX is serialization agnostic, meaning its packets have no pre-defined encoding format. Users of SEQEX are free to choose between serde, packed structs, tagged unions, or anything else as their prefered serialization format. This means SEQEX takes some additional effort to set up up-front, but it means that user have significantly more flexibility long-term. +//! +//! As such it is relatively easier to run SEQEX in parrallel with another raw UDP protocol, or even in parrallel with itself. Multiple instances of SEQEX can be opened between two peers, making it very easy to reduce or even eliminate head-of-line latency in performance critical applications. +//! +//! ## Why not TCP? +//! +//! TCP only guarantees packets will be received in the same order they were sent. +//! It has no inherent concept of "replying to a packet" and as such it cannot guarantee both sides +//! of a conversation have the same view of any stateful exchanges that take place. This must be implemented manually by the user of TCP. +//! +//! TCP is also much higher overhead. It requires a 1.5 RTT handshake to begin any connection, +//! it has a larger amount of metadata that must be transported with packets, and it has quite a few +//! features that slow down runtime regardless of whether or not they are used. +//! A lot of this overhead owes to TCPs sizeable complexity. +//! +//! That being said SEQEX does lack many of TCP's additional features, such as a dynamic resend timer, +//! keep-alives, and fragmentation. This can be both a pro and a con, as it means there is a +//! lot of efficiency to be gained if these features are not needed or are implemented at a +//! different protocol layer. +//! +//! Neither SEQEX nor TCP are cryptographically secure. +#![warn(missing_docs, rust_2018_idioms)] mod transport_layer; pub use transport_layer::*; -mod seq_queue; -pub use seq_queue::*; +pub mod result; +/// This module contains the API for using SEQEX in a no-std environment. +/// This API is low level and is the backbone of the `sync` and `tokio` implementations of SEQEX. +/// +/// It contains relatively little in the way of safety and correctness guarantees, +/// so it is not recommended to be used unless necessary. +pub mod no_std; +/// Contains a higher level API for the no_std version of SEQEX. +/// no_std by default is extremely low level. mod single_thread; -pub use single_thread::*; +/// This module contains the API for using SEQEX safely in a multithreaded environment. +/// +/// This is the recommended API for using SEQEX in non-async code. #[cfg(feature = "std")] pub mod sync; +/// This module contains the API for using SEQEX safely with tokio for async-await style code. #[cfg(feature = "tokio")] pub mod tokio; diff --git a/src/seq_queue.rs b/src/no_std.rs similarity index 75% rename from src/seq_queue.rs rename to src/no_std.rs index f98eca3..941686a 100644 --- a/src/seq_queue.rs +++ b/src/no_std.rs @@ -1,50 +1,11 @@ -//! The reference implementation of the **Sequential Exchange Protocol**, or SEP. -//! -//! SEP is a peer-to-peer transport protocol that guarantees packets of data will always be received -//! in the same order they were sent. In addition, it also guarantees the sequential consistency of -//! stateful exchanges between the two communicating peers. -//! -//! A "stateful exchange" is defined here as a sequence of packets, where the first packet -//! initiates the exchange, and all subsequent packets are replies to the previous packet in the -//! exchange. -//! -//! SEP guarantees both peers will agree upon which packets are members of which exchanges, -//! and it guarantees each packet is received by each peer in sequential order. -//! -//! SEP is a tiny, dead simple protocol and we have implemented it here in less than 500 lines of code. -//! -//! ## Why not TCP? -//! -//! TCP only guarantees packets will be received in the same order they were sent. -//! It has no inherent concept of "replying to a packet" and as such it cannot guarantee both sides -//! of a conversation have the same view of any stateful exchanges that take place. -//! -//! TCP is also much higher overhead. It requires a 1.5 RTT handshake to begin any connection, -//! it has a larger amount of metadata that must be transported with packets, and it has quite a few -//! features that slow down runtime regardless of whether or not they are used. -//! A lot of this overhead owes to TCPs sizeable complexity. -//! -//! That being said SEP does lack many of TCP's additional features, such as a dynamic resend timer, -//! keep-alives, and fragmentation. This can be both a pro and a con, as it means there is a -//! lot of efficiency to be gained if these features are not needed or are implemented at a -//! different protocol layer. -//! -//! Neither SEP nor TCP are cryptographically secure. -//! -//! ## Examples -//! +pub use crate::single_thread::*; -/// A 32-bit sequence number. Packets transported with SEP are expected to contain at least one -/// sequence number, and sometimes two. -/// All packets will either have a seq_no, a reply_no, or both. -pub type SeqNo = u32; - -/// The resend interval for a default instance of SeqEx. -pub const DEFAULT_RESEND_INTERVAL_MS: i64 = 250; -/// The initial sequence number for a default instance of SeqEx. -pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 0; - -pub const DEFAULT_WINDOW_CAP: usize = 64; +use crate::{ + result::{TryError, TryRecvError}, + transport_layer::SeqNo, + Packet, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, +}; +use Packet::*; #[derive(Debug)] pub struct SeqEx { @@ -103,128 +64,6 @@ impl SendEntry { } } -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum TryRecvError { - DroppedTooEarly, - DroppedDuplicate, - DroppedDuplicateResendAck(SeqNo), - WaitingForRecv, - WaitingForReply, -} -#[cfg(feature = "std")] -impl std::fmt::Display for TryRecvError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - TryRecvError::DroppedTooEarly => write!(f, "packet arrived too early"), - TryRecvError::DroppedDuplicate => write!(f, "packet was a duplicate"), - TryRecvError::DroppedDuplicateResendAck(_) => write!(f, "packet was a duplicate, resending ack"), - TryRecvError::WaitingForRecv => write!(f, "can't process until another packet is received"), - TryRecvError::WaitingForReply => write!(f, "can't process until a reply is finished"), - } - } -} -#[cfg(feature = "std")] -impl std::error::Error for TryRecvError {} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum TryError { - WaitingForRecv, - WaitingForReply, -} -#[cfg(feature = "std")] -impl std::fmt::Display for TryError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - TryError::WaitingForRecv => write!(f, "can't process until another packet is received"), - TryError::WaitingForReply => write!(f, "can't process until a reply is finished"), - } - } -} -#[cfg(feature = "std")] -impl std::error::Error for TryError {} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] -pub enum Packet { - Payload(SeqNo, RecvData), - SeqCstPayload(SeqNo, RecvData), - Reply(SeqNo, SeqNo, RecvData), - SeqCstReply(SeqNo, SeqNo, RecvData), - Ack(SeqNo), -} -use Packet::*; -impl Packet { - pub fn new_with_data(seq_no: SeqNo, reply_no: Option, seq_cst: bool, data: RecvData) -> Self { - Self::new(Some(seq_no), reply_no, seq_cst, Some(data)).unwrap() - } - pub fn new(seq_no: Option, reply_no: Option, seq_cst: bool, data: Option) -> Option { - match (seq_no, reply_no, seq_cst, data) { - (Some(s), None, false, Some(d)) => Some(Payload(s, d)), - (Some(s), None, true, Some(d)) => Some(SeqCstPayload(s, d)), - (Some(s), Some(r), false, Some(d)) => Some(Reply(s, r, d)), - (Some(s), Some(r), true, Some(d)) => Some(SeqCstReply(s, r, d)), - (None, Some(r), false, None) => Some(Ack(r)), - _ => None, - } - } - pub fn as_ref(&self) -> Packet<&RecvData> { - match self { - Payload(seq_no, data) => Payload(*seq_no, data), - SeqCstPayload(seq_no, data) => SeqCstPayload(*seq_no, data), - Reply(seq_no, reply_no, data) => Reply(*seq_no, *reply_no, data), - SeqCstReply(seq_no, reply_no, data) => SeqCstReply(*seq_no, *reply_no, data), - Ack(reply_no) => Ack(*reply_no), - } - } - pub fn map(self, f: impl FnOnce(RecvData) -> SendData) -> Packet { - match self { - Payload(seq_no, data) => Payload(seq_no, f(data)), - SeqCstPayload(seq_no, data) => SeqCstPayload(seq_no, f(data)), - Reply(seq_no, reply_no, data) => Reply(seq_no, reply_no, f(data)), - SeqCstReply(seq_no, reply_no, data) => SeqCstReply(seq_no, reply_no, f(data)), - Ack(reply_no) => Ack(reply_no), - } - } - pub fn payload(self) -> Option { - self.consume().ok() - } - pub fn consume(self) -> Result { - match self { - Payload(_, data) | SeqCstPayload(_, data) | Reply(_, _, data) | SeqCstReply(_, _, data) => Ok(data), - Ack(r) => Err(r), - } - } - pub fn is_seq_cst(&self) -> bool { - matches!(self, SeqCstPayload(..) | SeqCstReply(..)) - } - pub fn set_seq_cst(&mut self, seq_cst: bool) { - let mut tmp = Ack(0); - core::mem::swap(&mut tmp, self); - match tmp { - Payload(seq_no, data) | SeqCstPayload(seq_no, data) => { - *self = if seq_cst { - SeqCstPayload(seq_no, data) - } else { - Payload(seq_no, data) - } - } - Reply(seq_no, reply_no, data) | SeqCstReply(seq_no, reply_no, data) => { - *self = if seq_cst { - SeqCstReply(seq_no, reply_no, data) - } else { - Reply(seq_no, reply_no, data) - } - } - Ack(reply_no) => *self = Ack(reply_no), - } - } -} -impl Packet<&RecvData> { - pub fn cloned(&self) -> Packet { - self.map(|d| d.clone()) - } -} - #[derive(Clone, Debug)] pub enum RecvOkRaw { Payload { diff --git a/src/result.rs b/src/result.rs new file mode 100644 index 0000000..f7fa8fe --- /dev/null +++ b/src/result.rs @@ -0,0 +1,113 @@ +use crate::SeqNo; + +/// These are the error types that can be returned by a non-blocking SEQEX receive function. +/// +/// Some of these errors specify that SEQEX is waiting on some event to occur before it can proceed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TryRecvError { + /// This packet had to be dropped because it arrived far enough out-of-order that it was outside + /// the receive window. + /// This packet will eventually be resent, so no data will be lost. + DroppedTooEarly, + /// This packet was a duplicate of a previously received packet. It must have been resent before + /// the remote peer received the ack for the packet. + /// Since this packet is a duplicate, no data is lost by dropping it. + DroppedDuplicate, + /// This packet was a duplicate of a previously received packet. We need to resend an Ack packet + /// containing the reply number within this error instance. + /// + /// So `Packet::Ack(SeqNo)` should be sent to the remote peer immediately. + DroppedDuplicateResendAck(SeqNo), + /// In order to preserve losslessness or in-order transport, the received packet cannot be + /// process until some other packet is received. The packet was saved to the receive window. + WaitingForRecv, + /// Either the receive window is full, or the received packet is SeqCst and cannot be processed + /// yet. In either case some currently issued reply number must be returned to SEQEX to send + /// either an Ack or a Reply. If using reply guards, then some currently existing reply guard + /// must be dropped or consumed. + /// Until this occurs the received packet cannot be processed. + /// + /// If the receive window was full, the packet was dropped. + /// Otherwise if the packet is SeqCst, then the packet was saved to the receive window. + WaitingForReply, +} + +/// These are the error types that can be returned by a blocking SEQEX receive function. +/// +/// Some of these errors specify that SEQEX is waiting on some event to occur before it can proceed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RecvError { + /// This packet had to be dropped because it arrived far enough out-of-order that it was outside + /// the receive window. + /// This packet will eventually be resent, so no data will be lost. + DroppedTooEarly, + /// This packet was a duplicate of a previously received packet. It must have been resent before + /// the remote peer received the ack for the packet. + /// Since this packet is a duplicate, no data is lost by dropping it. + DroppedDuplicate, + /// In order to preserve losslessness or in-order transport, the received packet cannot be + /// process until some other packet is received. The packet was saved to the receive window. + WaitingForRecv, + /// Either the receive window is full, or the received packet is SeqCst and cannot enter the + /// critical section where it is processed. In either case there currently exists some reply + /// guard that must be dropped or consumed before this packet can be processed. + /// + /// If the receive window was full, the packet was dropped. + /// Otherwise if the packet is SeqCst, then the packet was saved to the receive window. + WaitingForReply, +} + +/// A generic error that can be returned by a `try_send` or `try_pump` function. +/// They specify what event must occur before a future call to `try_send` or `try_pump` can succeed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TryError { + /// The packet could not be sent or processed at this time. + /// Some other packet must be received from the remote peer first. + WaitingForRecv, + /// Some currently issued reply number must be returned to SEQEX to send either an Ack or a + /// Reply. If using reply guards, then some currently existing reply guard must be dropped or + /// consumed. + /// Until this occurs the packet cannot be sent or processed. + WaitingForReply, +} + +#[cfg(feature = "std")] +impl std::fmt::Display for TryRecvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + TryRecvError::DroppedTooEarly => write!(f, "packet arrived too early"), + TryRecvError::DroppedDuplicate => write!(f, "packet was a duplicate"), + TryRecvError::DroppedDuplicateResendAck(_) => write!(f, "packet was a duplicate, resending ack"), + TryRecvError::WaitingForRecv => write!(f, "can't process until another packet is received"), + TryRecvError::WaitingForReply => write!(f, "can't process until a reply is finished"), + } + } +} +#[cfg(feature = "std")] +impl std::error::Error for TryRecvError {} + +#[cfg(feature = "std")] +impl std::fmt::Display for RecvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RecvError::DroppedTooEarly => write!(f, "packet arrived too early"), + RecvError::DroppedDuplicate => write!(f, "packet was a duplicate"), + RecvError::WaitingForRecv => write!(f, "can't process until another packet is received"), + RecvError::WaitingForReply => write!(f, "can't process until a reply is finished"), + } + } +} +#[cfg(feature = "std")] +impl std::error::Error for RecvError {} + +#[cfg(feature = "std")] +impl std::fmt::Display for TryError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + TryError::WaitingForRecv => write!(f, "can't process until another packet is received"), + TryError::WaitingForReply => write!(f, "can't process until a reply is finished"), + } + } +} +#[cfg(feature = "std")] +impl std::error::Error for TryError {} diff --git a/src/single_thread.rs b/src/single_thread.rs index 53ddb0a..d3f9bdc 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,15 +1,32 @@ -use crate::{Packet, RecvOkRaw, SeqEx, SeqNo, TransportLayer, TryError, TryRecvError, DEFAULT_WINDOW_CAP}; +use crate::no_std::{RecvOkRaw, SeqEx}; +use crate::result::{RecvError, TryError, TryRecvError}; +use crate::{Packet, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { seq: &'a mut SeqEx, - app: Option, + tl: TL, reply_no: SeqNo, is_holding_lock: bool, + has_replied: bool, } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { + pub fn get_tl(&self) -> &TL { + &self.tl + } + pub fn get_tl_mut(&mut self) -> &mut TL { + &mut self.tl + } + pub fn has_replied(&self) -> bool { + self.has_replied + } + pub fn is_seq_cst(&self) -> bool { + self.is_holding_lock + } + pub fn ack(&mut self) { - if let Some(app) = self.app.take() { - self.seq.ack_raw(app, self.reply_no, self.is_holding_lock); + if !self.has_replied { + self.has_replied = true; + self.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock); } } /// If you need to reply more than once, say to fragment a large file, then include in your @@ -25,20 +42,24 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// # Panic /// This function will panic if `ack` has been called. fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { - let app = self.app.take().expect("Cannot reply after an ack has been sent"); + assert!(!self.has_replied, "Cannot reply after an ack has been sent"); + self.has_replied = true; let seq_no = self.seq.seq_no(); self.seq - .reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); + .reply_raw(self.tl, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); core::mem::forget(self); } - pub fn to_components(self) -> (SeqNo, bool) { + pub unsafe fn to_components(self) -> (SeqNo, bool) { let ret = (self.reply_no, self.is_holding_lock); core::mem::forget(self); ret } - pub unsafe fn from_components(seq: &'a mut SeqEx, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { - ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock } + fn new(seq: &'a mut SeqEx, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { + ReplyGuard { seq, tl, reply_no, is_holding_lock, has_replied: false } + } + pub unsafe fn from_components(seq: &'a mut SeqEx, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { + Self::new(seq, tl, reply_no, is_holding_lock) } } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { @@ -55,27 +76,6 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std } } -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum RecvError { - DroppedTooEarly, - DroppedDuplicate, - WaitingForRecv, - WaitingForReply, -} -#[cfg(feature = "std")] -impl std::fmt::Display for RecvError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - RecvError::DroppedTooEarly => write!(f, "packet arrived too early"), - RecvError::DroppedDuplicate => write!(f, "packet was a duplicate"), - RecvError::WaitingForRecv => write!(f, "can't process until another packet is received"), - RecvError::WaitingForReply => write!(f, "can't process until a reply is finished"), - } - } -} -#[cfg(feature = "std")] -impl std::error::Error for RecvError {} - pub enum RecvOk<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { Payload { reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, @@ -114,14 +114,14 @@ macro_rules! impl_recvok { } } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> $recv<'a, TL, SendData, RecvData, CAP> { - fn from_raw(seq: $seq_ex, app: TL, value: RecvOkRaw) -> Self { + fn from_raw(seq: $seq_ex, tl: TL, value: RecvOkRaw) -> Self { match value { RecvOkRaw::Payload { reply_no, seq_cst, recv_data } => Self::Payload { - reply_guard: ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock: seq_cst }, + reply_guard: ReplyGuard::new(seq, tl, reply_no, seq_cst), recv_data, }, RecvOkRaw::Reply { reply_no, seq_cst, recv_data, send_data } => Self::Reply { - reply_guard: ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock: seq_cst }, + reply_guard: ReplyGuard::new(seq, tl, reply_no, seq_cst), recv_data, send_data, }, @@ -151,10 +151,10 @@ pub(crate) use impl_recvok; impl SeqEx { /// Can mutate `next_service_timestamp`. - pub fn try_send(&mut self, mut app: impl TransportLayer, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> { - match self.try_send_direct(app.time(), seq_cst, packet_data) { + pub fn try_send(&mut self, mut tl: impl TransportLayer, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> { + match self.try_send_direct(tl.time(), seq_cst, packet_data) { Ok(p) => { - app.send(p); + tl.send(p); Ok(()) } Err(e) => Err(e), @@ -163,13 +163,13 @@ impl SeqEx { /// Can mutate `next_service_timestamp`. pub fn try_send_with SendData>( &mut self, - mut app: impl TransportLayer, + mut tl: impl TransportLayer, seq_cst: bool, packet_data: F, ) -> Result<(), (TryError, F)> { - match self.try_send_direct_with(app.time(), seq_cst, packet_data) { + match self.try_send_direct_with(tl.time(), seq_cst, packet_data) { Ok(p) => { - app.send(p); + tl.send(p); Ok(()) } Err(e) => Err(e), @@ -178,13 +178,13 @@ impl SeqEx { /// If this returns `Ok` then `try_send` might succeed on next call. pub fn receive_raw>( &mut self, - mut app: impl TransportLayer, + mut tl: impl TransportLayer, packet: Packet

, ) -> Result<(RecvOkRaw, bool), RecvError> { match self.receive_raw_and_direct(packet) { Ok(a) => Ok(a), Err(TryRecvError::DroppedDuplicateResendAck(reply_no)) => { - app.send(Packet::Ack(reply_no)); + tl.send(Packet::Ack(reply_no)); Err(RecvError::DroppedDuplicate) } Err(TryRecvError::DroppedTooEarly) => Err(RecvError::DroppedTooEarly), @@ -197,41 +197,40 @@ impl SeqEx { /// If `unlock` is true and the return value is true pump may return new values. /// /// Only returns false if the reply number was incorrect or used twice. - pub fn reply_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, unlock: bool, seq_cst: bool, packet_data: SendData) -> bool { - if let Some(p) = self.reply_raw_and_direct(app.time(), reply_no, unlock, seq_cst, packet_data) { - app.send(p); + pub fn reply_raw(&mut self, mut tl: impl TransportLayer, reply_no: SeqNo, unlock: bool, seq_cst: bool, packet_data: SendData) -> bool { + if let Some(p) = self.reply_raw_and_direct(tl.time(), reply_no, unlock, seq_cst, packet_data) { + tl.send(p); true } else { false } } /// If `unlock` is true and the return value is true pump may return new values. - pub fn ack_raw(&mut self, mut app: impl TransportLayer, reply_no: SeqNo, unlock: bool) -> bool { + pub fn ack_raw(&mut self, mut tl: impl TransportLayer, reply_no: SeqNo, unlock: bool) -> bool { if let Some(p) = self.ack_raw_and_direct(reply_no, unlock) { - app.send(p); + tl.send(p); true } else { false } } /// Can mutate `next_service_timestamp`. - pub fn service(&mut self, mut app: impl TransportLayer) -> i64 { - let current_time = app.time(); + pub fn service(&mut self, mut tl: impl TransportLayer) -> i64 { + let current_time = tl.time(); let mut iter = None; while let Some(p) = self.service_direct(current_time, &mut iter) { - app.send(p) + tl.send(p) } self.resend_interval.min(self.next_service_timestamp - current_time) } pub fn receive>( &mut self, - app: TL, + tl: TL, packet: Packet, ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), RecvError> { - self.receive_raw(app.clone(), packet) - .map(|(r, do_pump)| (RecvOk::from_raw(self, app, r), do_pump)) + self.receive_raw(tl, packet).map(|(r, do_pump)| (RecvOk::from_raw(self, tl, r), do_pump)) } - pub fn try_pump>(&mut self, app: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> { - self.try_pump_raw().map(|(r, do_pump)| (RecvOk::from_raw(self, app, r), do_pump)) + pub fn try_pump>(&mut self, tl: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> { + self.try_pump_raw().map(|(r, do_pump)| (RecvOk::from_raw(self, tl, r), do_pump)) } } diff --git a/src/sync.rs b/src/sync.rs index 15baafa..4b1cd75 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -7,72 +7,171 @@ use std::{ }; use crate::{ - Packet, RecvError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, TryError, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, + no_std::RecvOkRaw, + result::{RecvError, TryError}, + Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; -pub struct SeqExSync { +/// The core thread-safe datastructure which manages the SEQEX protocol. +/// +/// `SendData` is some collection of data chosen by the user to define the contents, or payload, of +/// a packet. It can also be made to contain additional associated data that is not sent to the peer, +/// but instead is retained by SEQEX to locally track the current state of some exchange with the +/// remote peer. +/// Commonly uses include placing an enum within `SendData` that defines an asynchronous state machine. +/// The kind of state machine Rust would automatically generate to implement async-await code. +/// +/// Any instance of `SendData` passed into a `SeqEx` method can only be dropped if the +/// entire `SeqEx` instance is dropped, otherwise they will eventually be returned by some future +/// call into a `SeqEx` method. +/// +/// `RecvData` is another collection of data chosen by the user to define a payload of data that was +/// just received from the remote peer. Often this is just set to `Vec`, but it can also be +/// set to some other data-owning type. `RecvData` can also be made to contain packet metadata such +/// as what IP address and port the payload of data was received over. +pub struct SeqEx { inner: Mutex>, wait_on_recv: Condvar, wait_on_reply_sender: Condvar, wait_on_reply_receiver: Condvar, } struct SeqExInner { - seq: SeqEx, + seq: crate::no_std::SeqEx, recv_waiters: usize, reply_sender_waiters: bool, reply_receiver_waiters: bool, } - +/// A guard type that allows Rust's borrow-checker to guarantee that the invariants of the SEQEX +/// protocol are maintained. +/// +/// Every time a data-containing packet is received by SEQEX, it must be replied to by either an ack, +/// or a reply pacjet, but never both. This guard is created when data-containing packet is received, +/// and when it is dropped it will send an ack. However this guard contains the function `reply`. +/// This function consumes the guard and will cause a reply packet to be sent instead of an ack. +/// +/// In addition, if the data-containing packet was also a SeqCst packet. This guard acts like a +/// `MutexGuard`. All other received SeqCst packets will be blocked until this guard is dropped or +/// consumed. +/// +/// SeqCst packets almost always require a critical section to be processed correctly, otherwise +/// their in-order guarantee would be rendered pointless because of CPU scheduling non-determinism. +/// Reply guard make sure these critical sections are exist by default. pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { - seq: &'a SeqExSync, - app: Option, + seq: &'a SeqEx, + tl: TL, reply_no: SeqNo, is_holding_lock: bool, + has_replied: bool, } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { + /// Returns a reference to the `TransportLayer` instance this guard was created with. + pub fn get_tl(&self) -> &TL { + &self.tl + } + /// Returns a mutable reference to the `TransportLayer` instance this guard was created with. + /// + /// Keep in mind that this cannot be used to change how replies and acks are resent, since + /// resends are handled with a separate instance of `TransportLayer`. + pub fn get_tl_mut(&mut self) -> &mut TL { + &mut self.tl + } + /// Returns whether or not `ack` has already been called on this reply guard instance. + pub fn has_replied(&self) -> bool { + self.has_replied + } + /// Returns whether or nor this reply guard is for a SeqCst packet, and therefore is + /// holding a lock, preventing other SeqCst packets from being processed yet. + pub fn is_seq_cst(&self) -> bool { + self.is_holding_lock + } + /// Cause this reply guard to immediately send an ack to the remote peer, + /// preventing it from being used to send a reply. + /// + /// This allows advanced users to decouple locking the critical section for SeqCst packets + /// from sending acks to received packets. It is not recommended to use this function + /// except for this purpose. `drop` this reply guard instead. pub fn ack(&mut self) { - if let Some(app) = self.app.take() { + if !self.has_replied { + self.has_replied = true; let mut inner = self.seq.inner.lock().unwrap(); - inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock); + inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock); } } + /// Similar to `ReplyGuard::reply`, except the provided function is called to produce the data + /// to write to the send window. + /// + /// This function receives as its first argument the packet sequence number, and as its second + /// number the packet reply number. All calls to `TransportLayer::send` involving this packet + /// will receive the exact same sequence and reply number. /// # Panic - /// This function will panic if `ack` has been called. + /// This function will panic if `ack` has been called previously. pub fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { - let app = self.app.take().expect("Cannot reply after an ack has been sent"); + assert!(!self.has_replied, "Cannot reply after an ack has been sent"); + self.has_replied = true; let mut inner = self.seq.inner.lock().unwrap(); let seq_no = inner.seq.seq_no(); inner .seq - .reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); + .reply_raw(self.tl, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); self.seq.notify_reply(inner); core::mem::forget(self); } + /// Consume this reply guard to add `packet_data` to the send window and immediately send it + /// as a reply to the remote peer. Similar to `SeqEx::send`, except the remote peer will be + /// explicitly informed that this packet is indeed a reply to a packet they sent. + /// /// 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. /// # Panic - /// This function will panic if `ack` has been called. + /// This function will panic if `ack` has been called previously. pub fn reply(self, seq_cst: bool, packet_data: SendData) { self.reply_with(seq_cst, |_, _| packet_data) } - - pub fn to_components(self) -> (SeqNo, bool) { + /// Break down a `ReplyGuard` into its primitive components, without causing it to send an ack + /// or reply to the remote peer. + /// + /// The first return value is the packet reply number, and the second is the return value of + /// `is_seq_cst`, which states whether or not this `ReplyGuard` is holding a lock. + /// + /// This can be used in combination with `from_components` to move a `ReplyGuard` to a different + /// thread. + /// + /// # Safety + /// This function must never be called on a `ReplyGuard` that has had `ack` called on it. + /// + /// The caller must guarantee that `ReplyGuard::from_components` is eventually called on + /// the returned values. + /// + /// If these invariants are not maintained protocol deadlock is likely to occur, which will quickly + /// be followed by lock-based deadlock. + pub unsafe fn to_components(self) -> (SeqNo, bool) { + debug_assert!(!self.has_replied, "Cannot break down a ReplyGuard after an ack has been sent"); let ret = (self.reply_no, self.is_holding_lock); core::mem::forget(self); ret } - pub unsafe fn from_components(seq: &'a SeqExSync, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { - ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock } + fn new(seq: &'a SeqEx, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { + ReplyGuard { seq, tl, reply_no, is_holding_lock, has_replied: false } + } + /// Constructs a `ReplyGuard` object from the raw components returned by + /// `ReplyGuard::to_components`. + /// + /// # Safety + /// The caller must always pass values for `reply_no` and `is_holding_lock` that were + /// returned by `ReplyGuard::to_components`. Otherwise undefined behavior will occur. + pub unsafe fn from_components(seq: &'a SeqEx, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { + Self::new(seq, tl, reply_no, is_holding_lock) } } impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { let mut inner = self.seq.inner.lock().unwrap(); - if let Some(app) = self.app.take() { - inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock); + if !self.has_replied { + self.has_replied = true; + inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock); } self.seq.notify_reply(inner); } @@ -100,20 +199,29 @@ pub enum RecvOk<'a, TL: TransportLayer, SendData, RecvData, const CAP: send_data: SendData, }, } -crate::impl_recvok!(RecvOk, &'a SeqExSync); +crate::no_std::impl_recvok!(RecvOk, &'a SeqEx); pub struct RecvIter<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { - seq: Option<&'a SeqExSync>, - app: TL, + seq: Option<&'a SeqEx>, + tl: TL, first: Option>, blocking: bool, } -impl SeqExSync { +impl SeqEx { + /// Creates a new instance of SeqEx. + /// + /// `retry_interval` sets the interval at which SEQEX will resend unacknowledged packets. + /// + /// `initial_seq_no` sets the initial sequence number for this SEQEX session. This number + /// **must** be synchronized with the remote peer. + /// Setting `initial_seq_no` to a static value is the easiest way to synchronize it with the + /// emote peer, but this is only recommended if running SEQEX on top of an encrypted tunnel. + /// Otherwise it should be randomized. pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self { Self { inner: Mutex::new(SeqExInner { - seq: SeqEx::new(retry_interval, initial_seq_no), + seq: crate::no_std::SeqEx::new(retry_interval, initial_seq_no), recv_waiters: 0, reply_sender_waiters: false, reply_receiver_waiters: false, @@ -144,41 +252,39 @@ impl SeqExSync { pub fn try_receive>( &self, - app: TL, + tl: TL, packet: Packet, ) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), RecvError> { let mut inner = self.inner.lock().unwrap(); - match inner.seq.receive_raw(app.clone(), packet) { + match inner.seq.receive_raw(tl, packet) { Ok((r, do_pump)) => { self.notify_recv(inner); - Ok((RecvOk::from_raw(self, app, r), do_pump)) + Ok((RecvOk::from_raw(self, tl, r), do_pump)) } Err(e) => Err(e), } } - pub fn receive>( - &self, - app: TL, - packet: Packet, - ) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { - let result = self.try_receive(app.clone(), packet); + /// A SEQEX packet was just received and deserialized from the transport layer. + /// + pub fn receive>(&self, tl: TL, packet: Packet) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { + let result = self.try_receive(tl, packet); if let Err(RecvError::WaitingForReply) = result { - self.pump(app) + self.pump(tl) } else { result.ok() } } - pub fn try_pump>(&self, app: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> { + pub fn try_pump>(&self, tl: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> { let mut inner = self.inner.lock().unwrap(); match inner.seq.try_pump_raw() { Ok((r, do_pump)) => { self.notify_recv(inner); - Ok((RecvOk::from_raw(self, app, r), do_pump)) + Ok((RecvOk::from_raw(self, tl, r), do_pump)) } Err(e) => Err(e), } } - pub fn pump>(&self, app: TL) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { + pub fn pump>(&self, tl: TL) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { let mut inner = self.inner.lock().unwrap(); // Enforce that only one thread may wait to pump at a time. if inner.reply_receiver_waiters { @@ -188,7 +294,7 @@ impl SeqExSync { match inner.seq.try_pump_raw() { Ok((r, do_pump)) => { self.notify_recv(inner); - return Some((RecvOk::from_raw(self, app, r), do_pump)); + return Some((RecvOk::from_raw(self, tl, r), do_pump)); } Err(TryError::WaitingForRecv) => return None, Err(TryError::WaitingForReply) => { @@ -199,50 +305,50 @@ impl SeqExSync { } } - pub fn receive_all>(&self, app: TL, packet: Packet) -> RecvIter<'_, TL, SendData, RecvData, CAP> { - let ret = self.receive(app.clone(), packet); + pub fn receive_all>(&self, tl: TL, packet: Packet) -> RecvIter<'_, TL, SendData, RecvData, CAP> { + let ret = self.receive(tl, packet); if let Some((first, do_pump)) = ret { RecvIter { seq: do_pump.then_some(self), - app, + tl, first: Some(first), blocking: true, } } else { - RecvIter { seq: None, app, first: None, blocking: true } + RecvIter { seq: None, tl, first: None, blocking: true } } } - pub fn try_receive_all>(&self, app: TL, packet: Packet) -> RecvIter<'_, TL, SendData, RecvData, CAP> { - let ret = self.try_receive(app.clone(), packet); + pub fn try_receive_all>(&self, tl: TL, packet: Packet) -> RecvIter<'_, TL, SendData, RecvData, CAP> { + let ret = self.try_receive(tl, packet); if let Ok((first, do_pump)) = ret { RecvIter { seq: do_pump.then_some(self), - app, + tl, first: Some(first), blocking: false, } } else { - RecvIter { seq: None, app, first: None, blocking: false } + RecvIter { seq: None, tl, first: None, blocking: false } } } pub fn try_send_with, F: FnOnce(SeqNo) -> SendData>( &self, - app: TL, + tl: TL, seq_cst: bool, packet_data: F, ) -> Result<(), (TryError, F)> { let mut inner = self.inner.lock().unwrap(); - inner.seq.try_send_with(app, seq_cst, packet_data) + inner.seq.try_send_with(tl, seq_cst, packet_data) } - pub fn try_send>(&self, app: TL, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> { + pub fn try_send>(&self, tl: TL, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> { let mut inner = self.inner.lock().unwrap(); - inner.seq.try_send(app, seq_cst, packet_data) + inner.seq.try_send(tl, seq_cst, packet_data) } - pub fn send_with>(&self, app: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) { + pub fn send_with>(&self, tl: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) { let mut inner = self.inner.lock().unwrap(); - while let Err((e, p)) = inner.seq.try_send_with(app.clone(), seq_cst, packet_data) { + while let Err((e, p)) = inner.seq.try_send_with(tl, seq_cst, packet_data) { packet_data = p; match e { TryError::WaitingForRecv => { @@ -256,9 +362,9 @@ impl SeqExSync { } } } - pub fn send>(&self, app: TL, seq_cst: bool, mut packet_data: SendData) { + pub fn send>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) { let mut inner = self.inner.lock().unwrap(); - while let Err((e, p)) = inner.seq.try_send(app.clone(), seq_cst, packet_data) { + while let Err((e, p)) = inner.seq.try_send(tl, seq_cst, packet_data) { packet_data = p; match e { TryError::WaitingForRecv => { @@ -273,11 +379,11 @@ impl SeqExSync { } } - pub fn service>(&self, app: TL) -> i64 { - self.inner.lock().unwrap().seq.service(app) + pub fn service>(&self, tl: TL) -> i64 { + self.inner.lock().unwrap().seq.service(tl) } } -impl Default for SeqExSync { +impl Default for SeqEx { fn default() -> Self { Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO) } @@ -290,9 +396,9 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Ite Some(item) } else if let Some(origin) = self.seq { let ret = if self.blocking { - origin.pump(self.app.clone()) + origin.pump(self.tl) } else { - origin.try_pump(self.app.clone()).ok() + origin.try_pump(self.tl).ok() }; if let Some((item, do_pump)) = ret { if !do_pump { diff --git a/src/tokio.rs b/src/tokio.rs index 4460337..d34edfd 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -5,13 +5,15 @@ use tokio::{ }; use crate::{ - Packet, RecvError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, TryError, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, + no_std::RecvOkRaw, + result::{RecvError, TryError}, + Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; type Sender = (oneshot::Sender>, SendData); type Receiver = (oneshot::Sender<(SeqNo, bool, RecvData)>, RecvData); -pub struct SeqExTokio { +pub struct SeqEx { inner: Mutex>, wait_on_recv: Notify, wait_on_reply: Notify, @@ -19,38 +21,50 @@ pub struct SeqExTokio } struct SeqExInner { - seq: SeqEx, Receiver, CAP>, + seq: crate::no_std::SeqEx, Receiver, CAP>, recv_waiters: usize, reply_waiters: bool, } pub struct ReplyGuard<'a, TL: TokioLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { - seq: &'a SeqExTokio, - app: Option, + seq: &'a SeqEx, + tl: TL, reply_no: SeqNo, is_holding_lock: bool, + has_replied: bool, } impl<'a, TL: TokioLayer, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> { - fn new(seq: &'a SeqExTokio, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { - ReplyGuard { seq, app: Some(app), reply_no, is_holding_lock } + pub fn get_tl(&self) -> &TL { + &self.tl } - fn try_reply_with_inner(&mut self, app: TL, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> Sender) -> Option { + pub fn get_tl_mut(&mut self) -> &mut TL { + &mut self.tl + } + pub fn has_replied(&self) -> bool { + self.has_replied + } + pub fn is_seq_cst(&self) -> bool { + self.is_holding_lock + } + + fn try_reply_with_inner(&mut self, tl: TL, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> Sender) -> Option { let mut inner = self.seq.inner.lock().unwrap(); let seq_no = inner.seq.seq_no(); let pre_ts = inner.seq.next_service_timestamp; inner .seq - .reply_raw(app, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); + .reply_raw(tl, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); let ret = (pre_ts != inner.seq.next_service_timestamp).then_some(inner.seq.next_service_timestamp); self.seq.notify_reply(inner); ret } pub fn ack(&mut self) { - if let Some(app) = self.app.take() { + if !self.has_replied { + self.has_replied = true; let mut inner = self.seq.inner.lock().unwrap(); - inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock); + inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock); } } /// If you need to reply more than once, say to fragment a large file, then include in your @@ -70,34 +84,40 @@ impl<'a, TL: TokioLayer, SendData, RecvData, const CAP: usi seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData, ) -> Result<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, RecvData), AsyncError> { - let app = self.app.take().expect("Cannot reply after an ack has been sent"); + assert!(!self.has_replied, "Cannot reply after an ack has been sent"); + self.has_replied = true; + let tl = self.tl; + let seq = self.seq; let (tx, rx) = oneshot::channel(); - let update_ts = self.try_reply_with_inner(app.clone(), seq_cst, |s, r| (tx, packet_data(s, r))); - let seq = self.seq; + let update_ts = self.try_reply_with_inner(tl, seq_cst, |s, r| (tx, packet_data(s, r))); core::mem::forget(self); if let Some(update_ts) = update_ts { let _ = seq.update_queue.send(update_ts).await; } let (reply_no, seq_cst, recv_data) = rx.await.map_err(|_| AsyncError::SeqExClosed)?.ok_or(AsyncError::EndOfExchange)?; - Ok((Self::new(seq, app, reply_no, seq_cst), recv_data)) + Ok((Self::new(seq, tl, reply_no, seq_cst), recv_data)) } - pub fn to_components(self) -> (SeqNo, bool) { + pub unsafe fn to_components(self) -> (SeqNo, bool) { let ret = (self.reply_no, self.is_holding_lock); core::mem::forget(self); ret } - pub unsafe fn from_components(seq: &'a SeqExTokio, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { - Self::new(seq, app, reply_no, is_holding_lock) + fn new(seq: &'a SeqEx, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { + ReplyGuard { seq, tl, reply_no, is_holding_lock, has_replied: false } + } + pub unsafe fn from_components(seq: &'a SeqEx, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self { + Self::new(seq, tl, reply_no, is_holding_lock) } } impl<'a, TL: TokioLayer, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> { fn drop(&mut self) { let mut inner = self.seq.inner.lock().unwrap(); - if let Some(app) = self.app.take() { - inner.seq.ack_raw(app, self.reply_no, self.is_holding_lock); + if !self.has_replied { + self.has_replied = true; + inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock); } self.seq.notify_reply(inner); } @@ -159,13 +179,13 @@ impl<'a, RecvData> From> for Receiver { } } -impl SeqExTokio { +impl SeqEx { pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> (Self, ServiceState) { let (update_queue, recv_service_update) = mpsc::channel(8); ( Self { inner: Mutex::new(SeqExInner { - seq: SeqEx::new(retry_interval, initial_seq_no), + seq: crate::no_std::SeqEx::new(retry_interval, initial_seq_no), recv_waiters: 0, reply_waiters: false, }), @@ -189,11 +209,11 @@ impl SeqExTokio { } fn receive_inner>( &self, - app: TL, + tl: TL, packet: Packet>, ) -> Result<(ReplyGuard<'_, TL, SendData, RecvData, CAP>, RecvData), Option> { let mut inner = self.inner.lock().unwrap(); - return match inner.seq.receive_raw(app.clone(), packet) { + return match inner.seq.receive_raw(tl, packet) { Ok((recv_data, do_pump)) => { // pump first, handle return value second. let mut total_recv = 1; @@ -205,12 +225,12 @@ impl SeqExTokio { if recv_data.0.send((reply_no, seq_cst, recv_data.1)).is_err() { // Send an ack if no one is receiving the reply on the other end. // Could occur if the future holding the receiver is dropped. - inner.seq.ack_raw(app.clone(), reply_no, seq_cst); + inner.seq.ack_raw(tl, reply_no, seq_cst); } } RecvOkRaw::Reply { reply_no, seq_cst, recv_data, send_data } => { if send_data.0.send(Some((reply_no, seq_cst, recv_data.1))).is_err() { - inner.seq.ack_raw(app.clone(), reply_no, seq_cst); + inner.seq.ack_raw(tl, reply_no, seq_cst); } } RecvOkRaw::Ack { send_data: (tx, _) } => { @@ -224,12 +244,12 @@ impl SeqExTokio { } let ret = match recv_data { - RecvOkRaw::Payload { reply_no, seq_cst, recv_data } => Ok((ReplyGuard::new(self, app, reply_no, seq_cst), recv_data.0)), + RecvOkRaw::Payload { reply_no, seq_cst, recv_data } => Ok((ReplyGuard::new(self, tl, reply_no, seq_cst), recv_data.0)), RecvOkRaw::Reply { reply_no, seq_cst, recv_data, send_data: (tx, _) } => { if tx.send(Some((reply_no, seq_cst, recv_data.0))).is_err() { // Send an ack if no one is receiving the reply on the other end. // Could occur if the future holding the receiver is dropped. - inner.seq.ack_raw(app, reply_no, seq_cst); + inner.seq.ack_raw(tl, reply_no, seq_cst); } Err(Some(AsyncRecvError::AsyncReply)) } @@ -256,18 +276,18 @@ impl SeqExTokio { pub async fn receive>( &self, - app: TL, + tl: TL, packet: Packet, ) -> Result<(ReplyGuard<'_, TL, SendData, RecvData, CAP>, RecvData), AsyncRecvError> { let mut tx = None; let packet = packet.map(|r| IntoOneshot(r, &mut tx)); - match self.receive_inner(app.clone(), packet) { + match self.receive_inner(tl, packet) { Ok(ret) => Ok(ret), Err(Some(e)) => Err(e), Err(None) => { if let Some(tx) = tx { let (reply_no, is_holding_lock, data) = tx.await.map_err(|_| AsyncRecvError::SeqExClosed)?; - Ok((ReplyGuard::new(self, app, reply_no, is_holding_lock), data)) + Ok((ReplyGuard::new(self, tl, reply_no, is_holding_lock), data)) } else { Err(AsyncRecvError::DroppedDuplicate) } @@ -277,13 +297,13 @@ impl SeqExTokio { fn send_with_inner, F: FnOnce(SeqNo) -> Sender>( &self, - app: TL, + tl: TL, seq_cst: bool, packet_data: F, ) -> Result, (TryError, F)> { let mut inner = self.inner.lock().unwrap(); let pre_ts = inner.seq.next_service_timestamp; - let result = inner.seq.try_send_with(app, seq_cst, packet_data); + let result = inner.seq.try_send_with(tl, seq_cst, packet_data); match result { Err((TryError::WaitingForRecv, p)) => { inner.recv_waiters += 1; @@ -299,28 +319,28 @@ impl SeqExTokio { pub async fn send>( &self, - app: TL, + tl: TL, seq_cst: bool, packet_data: SendData, ) -> Result<(ReplyGuard<'_, TL, SendData, RecvData, CAP>, RecvData), AsyncError> { - self.send_with(app, seq_cst, |_| packet_data).await + self.send_with(tl, seq_cst, |_| packet_data).await } pub async fn send_with>( &self, - app: TL, + tl: TL, seq_cst: bool, packet_data: impl FnOnce(SeqNo) -> SendData, ) -> Result<(ReplyGuard<'_, TL, SendData, RecvData, CAP>, RecvData), AsyncError> { let (rx, tx) = oneshot::channel(); let mut pf = |s| (rx, packet_data(s)); loop { - match self.send_with_inner(app.clone(), seq_cst, pf) { + match self.send_with_inner(tl, seq_cst, pf) { Ok(update) => { if let Some(update) = update { let _ = self.update_queue.send(update).await; } let (reply_no, locked, recv_data) = tx.await.map_err(|_| AsyncError::SeqExClosed)?.ok_or(AsyncError::EndOfExchange)?; - return Ok((ReplyGuard::new(self, app, reply_no, locked), recv_data)); + return Ok((ReplyGuard::new(self, tl, reply_no, locked), recv_data)); } Err((TryError::WaitingForRecv, p)) => { pf = p; @@ -336,10 +356,10 @@ impl SeqExTokio { /// This function must be called with the same ServiceState instance returned upon creation of /// the given SeqExTokio instance. - pub async fn service_task>(&self, mut app: TL, state: &mut ServiceState) { + pub async fn service_task>(&self, mut tl: TL, state: &mut ServiceState) { let mut result = None; if state.next_service_timestamp < i64::MAX { - let diff = state.next_service_timestamp - app.time(); + let diff = state.next_service_timestamp - tl.time(); if diff > 0 { if let Ok(up) = time::timeout(time::Duration::from_millis(diff as u64), state.recv_service_update.recv()).await { result = up @@ -353,13 +373,13 @@ impl SeqExTokio { state.next_service_timestamp = state.next_service_timestamp.min(up); } else { let mut inner = self.inner.lock().unwrap(); - inner.seq.service(app.clone()); + inner.seq.service(tl); state.next_service_timestamp = inner.seq.next_service_timestamp; } } } -pub trait TokioLayer: Clone { +pub trait TokioLayer: Clone + Copy { type SendData; fn time(&mut self) -> i64; diff --git a/src/transport_layer.rs b/src/transport_layer.rs index ef67169..a728e2d 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -1,4 +1,183 @@ -use crate::Packet; +/// A 32-bit sequence number. Packets transported with SEQEX are expected to contain at least one +/// sequence number, and sometimes two. +/// All packets will either have a seq_no, a reply_no, or both. +pub type SeqNo = u32; + +/// The resend interval for a default instance of SeqEx. +pub const DEFAULT_RESEND_INTERVAL_MS: i64 = 250; +/// The initial sequence number for a default instance of SeqEx. +pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 0; +/// The default maximum capacity of the SEQEX send and receive window. +/// The larger the capacity, the longer packets can be sent through SEQEX without needing to wait +/// for acknowledgements to arrive. +/// +/// The memory usage of SEQEX increases linearly with this number. +pub const DEFAULT_WINDOW_CAP: usize = 64; + +/// SEQEX is serialization agnostic. The user is free to choose whatever serialization format they +/// want for packets originating from SEQEX. All that is required is that this enum can be +/// serialized and deserialized accurately on both ends of a connection. +/// +/// The easiest serialization implementation is to simply use serde to write this enum to a Vec +/// and send the resulting bytes over the wire to the other end. +/// The receiver can then deserialize with serde. +/// +/// However if more efficiency is required, this enum can easily be serialized as a "tagged union" +/// (https://en.wikipedia.org/wiki/Tagged_union). +/// +/// It is possible and even encouraged in real-time environments to include a "channel id" along +/// with the serialized packet. That way each end of a SEQEX connection can run multiple instances of +/// SEQEX in parrallel, each instance identified by the channel id. By running multiple in parallel, +/// Head-of-line blocking can be avoided even when SeqCst packets are being sent. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] +pub enum Packet { + /// This is a normal payload of data within SEQEX. + /// When it is sent to a remote peer, SEQEX guarantees lossless delivery, + /// meaning the remote peer is guaranteed to receive this payload exactly once. + /// + /// This payload is not guaranteed to be received in the order it was sent relative to other + /// payloads. + /// + /// The contained `SeqNo` is the packet sequence number. + Payload(SeqNo, RecvData), + /// This is a payload of data with the added guarantee that it is received in order relative to + /// other SeqCst payloads and replies. + /// When it is sent to a remote peer, SEQEX guarantees in-order, losslessness delivery. + /// + /// Any normal payload sent before a SeqCst payload will be received before the SeqCst payload + /// is received, however normal payloads sent after a SeqCst payload can be received in any + /// order. + /// + /// As a result of in-order delivery this payload usually exhibits higher latency than a normal + /// payload. + /// + /// The contained `SeqNo` is the packet sequence number. + SeqCstPayload(SeqNo, RecvData), + /// This is a normal reply to a received packet within SEQEX. + /// SEQEX provides the unique capability for the user to reply to any kind of payload, as well + /// as any kind of reply. SEQEX guarantees that replies are unambiguous, meaning that both the + /// sender and receiver of a reply will always agree upon which packet the reply is replying to. + /// + /// This essentially means that all conversations within SEQEX are linked-lists, with replies + /// acting as nodes with a link to the next node, and payloads being nodes with null links. + /// The most recently received packet is the "head" of the linked list, and the linked list can + /// be appended to by replying to that packet. + /// + /// Similar to a normal payload, when a normal reply is sent to a remote peer, + /// SEQEX guarantees lossless delivery. + /// + /// The first `SeqNo` is the packet sequence number, and the second is the packet reply number. + Reply(SeqNo, SeqNo, RecvData), + /// This is a reply to a received packet with the added guarantee that it is received in order + /// relative to other SeqCst payloads and replies. + /// + /// SEQEX guarantees that SeqCst replies are unambiguous, and are received in-order. + /// See the documentation for `Reply` and `SeqCstPayload` for further information about these + /// guarantees. + /// + /// The first `SeqNo` is the packet sequence number, and the second is the packet reply number. + SeqCstReply(SeqNo, SeqNo, RecvData), + /// This is an acknowledgement within SEQEX. + /// It is sent by default in response to any received packet, unless the user has chosen to + /// send a reply instead. + /// + /// To provide the transport guarantees that it does, all packets within SEQEX are sent over the + /// wire multiple times until they are acknowledged by either an ack or a reply. + /// + /// Within the linked list analogy, an ack is the last final node appended to the head of the + /// linked list. Nothing further can be appended afterwards. + /// + /// The contained `SeqNo` is the packet reply number. + Ack(SeqNo), +} +use Packet::*; +impl Packet { + /// Create a new payload or reply packet. + /// If `reply_no` is `None`, this will output a reply packet. + /// If it is `Some`, this will output a payload packet. + pub fn new_with_data(seq_no: SeqNo, reply_no: Option, seq_cst: bool, data: RecvData) -> Self { + Self::new(Some(seq_no), reply_no, seq_cst, Some(data)).unwrap() + } + /// Attempt to create a packet from raw parts, if the raw parts correctly specify some valid + /// packet type within SEQEX. + pub fn new(seq_no: Option, reply_no: Option, seq_cst: bool, data: Option) -> Option { + match (seq_no, reply_no, seq_cst, data) { + (Some(s), None, false, Some(d)) => Some(Payload(s, d)), + (Some(s), None, true, Some(d)) => Some(SeqCstPayload(s, d)), + (Some(s), Some(r), false, Some(d)) => Some(Reply(s, r, d)), + (Some(s), Some(r), true, Some(d)) => Some(SeqCstReply(s, r, d)), + (None, Some(r), false, None) => Some(Ack(r)), + _ => None, + } + } + ///Converts from &Packet to Packet<&RecvData>. + pub fn as_ref(&self) -> Packet<&RecvData> { + match self { + Payload(seq_no, data) => Payload(*seq_no, data), + SeqCstPayload(seq_no, data) => SeqCstPayload(*seq_no, data), + Reply(seq_no, reply_no, data) => Reply(*seq_no, *reply_no, data), + SeqCstReply(seq_no, reply_no, data) => SeqCstReply(*seq_no, *reply_no, data), + Ack(reply_no) => Ack(*reply_no), + } + } + /// Maps a Packet to Packet by applying a function to the contained data + /// (if this is a reply or payload variant) or by doing nothing (if this is the ack variant). + pub fn map(self, f: impl FnOnce(RecvData) -> T) -> Packet { + match self { + Payload(seq_no, data) => Payload(seq_no, f(data)), + SeqCstPayload(seq_no, data) => SeqCstPayload(seq_no, f(data)), + Reply(seq_no, reply_no, data) => Reply(seq_no, reply_no, f(data)), + SeqCstReply(seq_no, reply_no, data) => SeqCstReply(seq_no, reply_no, f(data)), + Ack(reply_no) => Ack(reply_no), + } + } + /// Returns any data contained within this packet (if this is a reply or payload variant). + pub fn payload(self) -> Option { + self.consume().ok() + } + /// Returns any data contained within this packet (if this is a reply or payload variant), or an + /// `Err` containing the reply_no of an acknowledgement (if this is an ack variant). + pub fn consume(self) -> Result { + match self { + Payload(_, data) | SeqCstPayload(_, data) | Reply(_, _, data) | SeqCstReply(_, _, data) => Ok(data), + Ack(r) => Err(r), + } + } + /// Returns true if this is a SeqCst payload or SeqCst reply. + pub fn is_seq_cst(&self) -> bool { + matches!(self, SeqCstPayload(..) | SeqCstReply(..)) + } + /// If this is a reply or payload variant, this function will change it to a SeqCst variant if + /// `seq_cst` is true, or to a normal variant if `seq_cst` is false. + pub fn set_seq_cst(&mut self, seq_cst: bool) { + let mut tmp = Ack(0); + core::mem::swap(&mut tmp, self); + match tmp { + Payload(seq_no, data) | SeqCstPayload(seq_no, data) => { + *self = if seq_cst { + SeqCstPayload(seq_no, data) + } else { + Payload(seq_no, data) + } + } + Reply(seq_no, reply_no, data) | SeqCstReply(seq_no, reply_no, data) => { + *self = if seq_cst { + SeqCstReply(seq_no, reply_no, data) + } else { + Reply(seq_no, reply_no, data) + } + } + Ack(reply_no) => *self = Ack(reply_no), + } + } +} +impl Packet<&RecvData> { + /// Maps a Packet<&RecvData> to Packet by cloning any data it containts. + pub fn cloned(&self) -> Packet { + self.map(|d| d.clone()) + } +} /// A trait for giving an instance of SeqEx access to the transport layer. /// @@ -6,8 +185,16 @@ use crate::Packet; /// manage memory. /// 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 { +pub trait TransportLayer: Clone + Copy { + /// A callback that should return the current time in milliseconds. + /// The source of this time does not have to be monotonic. + /// + /// The timestamp that this function returns can be the time at which this instance of + /// `TransportLayer` was created/passed into a SEQEX function, rather than the time at which + /// this function was called. fn time(&mut self) -> i64; + /// A callback that should attempt to serialize and send `packet` to the remote peer. + /// fn send(&mut self, packet: Packet<&SendData>); } From 0c9e248c583f9cbd3227d402d81d7b9d951c943e Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Mon, 16 Oct 2023 15:32:10 -0400 Subject: [PATCH 2/3] added lots of docs --- src/{result.rs => error.rs} | 0 src/lib.rs | 3 +- src/no_std.rs | 11 +- src/single_thread.rs | 10 +- src/sync.rs | 197 +++++++++++++++++++++++++++++++++--- 5 files changed, 194 insertions(+), 27 deletions(-) rename src/{result.rs => error.rs} (100%) diff --git a/src/result.rs b/src/error.rs similarity index 100% rename from src/result.rs rename to src/error.rs diff --git a/src/lib.rs b/src/lib.rs index 9184a5d..a800b13 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -38,7 +38,8 @@ mod transport_layer; pub use transport_layer::*; -pub mod result; +/// Module which contains the various error types that can be returned by SEQEX. +pub mod error; /// This module contains the API for using SEQEX in a no-std environment. /// This API is low level and is the backbone of the `sync` and `tokio` implementations of SEQEX. diff --git a/src/no_std.rs b/src/no_std.rs index 941686a..f6e3e7a 100644 --- a/src/no_std.rs +++ b/src/no_std.rs @@ -1,7 +1,7 @@ pub use crate::single_thread::*; use crate::{ - result::{TryError, TryRecvError}, + error::{TryError, TryRecvError}, transport_layer::SeqNo, Packet, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; @@ -197,14 +197,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. /// - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. pub fn try_send_direct(&mut self, current_time: i64, seq_cst: bool, packet_data: SendData) -> Result, (TryError, SendData)> { let mut tmp = Some(packet_data); self.try_send_direct_with(current_time, seq_cst, |_| tmp.take().unwrap()) .map_err(|e| e.0) .map_err(|e| (e, tmp.unwrap())) } - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. pub fn try_send_direct_with SendData>( &mut self, current_time: i64, @@ -392,7 +392,7 @@ impl SeqEx { /// and since each fragment will be received in order it will be trivial for them to reconstruct /// the original file. /// - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. /// If `unlock` is true and the return value is `Some` pump may return new values. #[must_use] pub fn reply_raw_and_direct( @@ -444,7 +444,8 @@ impl SeqEx { } } - /// Can mutate `next_service_timestamp`. + /// Can increase `next_service_timestamp`. + #[inline] pub fn service_direct(&mut self, current_time: i64, iter: &mut Option) -> Option> { if self.next_service_timestamp <= current_time { let iter = iter.get_or_insert(ServiceIter { diff --git a/src/single_thread.rs b/src/single_thread.rs index d3f9bdc..28985d2 100644 --- a/src/single_thread.rs +++ b/src/single_thread.rs @@ -1,5 +1,5 @@ use crate::no_std::{RecvOkRaw, SeqEx}; -use crate::result::{RecvError, TryError, TryRecvError}; +use crate::error::{RecvError, TryError, TryRecvError}; use crate::{Packet, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP}; pub struct ReplyGuard<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { @@ -150,7 +150,7 @@ impl_recvok!(RecvOk, &'a mut SeqEx); pub(crate) use impl_recvok; impl SeqEx { - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. pub fn try_send(&mut self, mut tl: impl TransportLayer, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> { match self.try_send_direct(tl.time(), seq_cst, packet_data) { Ok(p) => { @@ -160,7 +160,7 @@ impl SeqEx { Err(e) => Err(e), } } - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. pub fn try_send_with SendData>( &mut self, mut tl: impl TransportLayer, @@ -193,7 +193,7 @@ impl SeqEx { Err(TryRecvError::WaitingForReply) => Err(RecvError::WaitingForReply), } } - /// Can mutate `next_service_timestamp`. + /// Can decrease `next_service_timestamp`. /// If `unlock` is true and the return value is true pump may return new values. /// /// Only returns false if the reply number was incorrect or used twice. @@ -214,7 +214,7 @@ impl SeqEx { false } } - /// Can mutate `next_service_timestamp`. + /// Can increase `next_service_timestamp`. pub fn service(&mut self, mut tl: impl TransportLayer) -> i64 { let current_time = tl.time(); let mut iter = None; diff --git a/src/sync.rs b/src/sync.rs index 4b1cd75..157875e 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -8,7 +8,7 @@ use std::{ use crate::{ no_std::RecvOkRaw, - result::{RecvError, TryError}, + error::{RecvError, TryError}, Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, }; @@ -71,7 +71,7 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// Returns a mutable reference to the `TransportLayer` instance this guard was created with. /// /// Keep in mind that this cannot be used to change how replies and acks are resent, since - /// resends are handled with a separate instance of `TransportLayer`. + /// resends are handled with a separate instance of `TransportLayer` passed to `SeqEx::service`. pub fn get_tl_mut(&mut self) -> &mut TL { &mut self.tl } @@ -103,31 +103,42 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Rep /// This function receives as its first argument the packet sequence number, and as its second /// number the packet reply number. All calls to `TransportLayer::send` involving this packet /// will receive the exact same sequence and reply number. + /// + /// Returns the timestamp of when `service_ts` should be called next, only if it has decrease. + /// This can be safely ignored if `service_ts` is not being used. + /// /// # Panic /// This function will panic if `ack` has been called previously. - pub fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) { + pub fn reply_with(mut self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) -> Option { assert!(!self.has_replied, "Cannot reply after an ack has been sent"); self.has_replied = true; let mut inner = self.seq.inner.lock().unwrap(); + let pre_nst = inner.seq.next_service_timestamp; let seq_no = inner.seq.seq_no(); inner .seq .reply_raw(self.tl, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no)); + let nst = inner.seq.next_service_timestamp; self.seq.notify_reply(inner); core::mem::forget(self); + (pre_nst > nst).then_some(nst) } /// Consume this reply guard to add `packet_data` to the send window and immediately send it /// as a reply to the remote peer. Similar to `SeqEx::send`, except the remote peer will be /// explicitly informed that this packet is indeed a reply to a packet they sent. /// + /// Returns the timestamp of when `service_ts` should be called next, only if it has decrease. + /// This can be safely ignored if `service_ts` is not being used. + /// /// 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. + /// /// # Panic /// This function will panic if `ack` has been called previously. - pub fn reply(self, seq_cst: bool, packet_data: SendData) { + pub fn reply(self, seq_cst: bool, packet_data: SendData) -> Option { self.reply_with(seq_cst, |_, _| packet_data) } /// Break down a `ReplyGuard` into its primitive components, without causing it to send an ack @@ -185,22 +196,47 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> std } } +/// When a packet received through `SeqEx` is ready to be processed, +/// it is returned as an instance of this enum. +/// +/// This enum specifies what kind of packet was received, and any additional data associated with the packet. pub enum RecvOk<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { + /// The received packet is a payload. Payload { + /// This type of packet can be optionally replied to, otherwise we need to send an ack. + /// This guard instance guarantees exactly one of those happens. + /// See the documentation of `ReplyGuard` for more information. reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, + /// The data associated with the packet we just received from the remote peer. + /// It was moved from the receive window to this enum. recv_data: RecvData, }, + /// The received packet is a reply to some packet we sent previously. Reply { + /// This type of packet can be optionally replied to, otherwise we need to send an ack. + /// This guard instance guarantees exactly one of those happens. + /// See the documentation of `ReplyGuard` for more information. reply_guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>, + /// The data associated with the packet we just received from the remote peer. + /// It was moved from the receive window to this enum. recv_data: RecvData, + /// The data associated with a packet we sent to the remote peer. + /// The received packet is a reply to that packet. + /// As a result, this data was moved from the `SeqEx` send window to this enum. send_data: SendData, }, + /// The received packet is an acknowledgment of some packet we sent previously. Ack { + /// The data associated with a packet we sent to the remote peer. + /// The received packet was an ack of this packet. + /// As a result, this data was moved from the `SeqEx` send window to this enum. send_data: SendData, }, } crate::no_std::impl_recvok!(RecvOk, &'a SeqEx); +/// An iterator which will automatically call either `SeqEx::pump` or `SeqEx::try_pump` to go +/// through all packets that can currently be processed. pub struct RecvIter<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> { seq: Option<&'a SeqEx>, tl: TL, @@ -249,7 +285,14 @@ impl SeqEx { self.wait_on_recv.notify_one(); } } - + /// A SEQEX packet was just received and deserialized from the transport layer. + /// Passing it to this function officially writes it to the receive window so that it can be + /// eventually processed with the SEQEX transport guarantees. + /// + /// This function can only block to lock an internal mutex. + /// + /// If this function returns `Ok`, the contained boolean is true if `SeqEx::pump` should also + /// be called, because there are packets ready to be processed in the receive window. pub fn try_receive>( &self, tl: TL, @@ -265,7 +308,13 @@ impl SeqEx { } } /// A SEQEX packet was just received and deserialized from the transport layer. + /// Passing it to this function officially writes it to the receive window so that it can be + /// eventually processed with the SEQEX transport guarantees. /// + /// This function may block to preserve lossless or in-order transport. + /// + /// If this function returns `Some`, the contained boolean is true if `SeqEx::pump` should also + /// be called, because there are packets ready to be processed in the receive window. pub fn receive>(&self, tl: TL, packet: Packet) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { let result = self.try_receive(tl, packet); if let Err(RecvError::WaitingForReply) = result { @@ -274,6 +323,10 @@ impl SeqEx { result.ok() } } + /// Non-blocking variant of `SeqEx::pump`. + /// See the documentation for `SeqEx::pump` for more information. + /// + /// NOTE: This function can block for a short period of time to lock an internal mutex. pub fn try_pump>(&self, tl: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> { let mut inner = self.inner.lock().unwrap(); match inner.seq.try_pump_raw() { @@ -284,6 +337,16 @@ impl SeqEx { Err(e) => Err(e), } } + /// Check if there are any packets in the receive window that can be processed. + /// + /// This function may block to preserve lossless or in-order transport. + /// It will never block if the receive window is empty. + /// + /// If this function returns `Some`, the contained boolean is true if `SeqEx::pump` should be + /// called another time, because there are still packets ready to be processed in the receive window. + /// + /// It is recommended to assign the job of pumping `SeqEx` to a different thread so packets can + /// be processed in parallel. pub fn pump>(&self, tl: TL) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> { let mut inner = self.inner.lock().unwrap(); // Enforce that only one thread may wait to pump at a time. @@ -304,7 +367,8 @@ impl SeqEx { } } } - + /// Similar to `SeqEx::receive`, except this function will return an iterator which + /// automatically calls `SeqEx::pump` as many times as needed to empty the receive window. pub fn receive_all>(&self, tl: TL, packet: Packet) -> RecvIter<'_, TL, SendData, RecvData, CAP> { let ret = self.receive(tl, packet); if let Some((first, do_pump)) = ret { @@ -318,6 +382,9 @@ impl SeqEx { RecvIter { seq: None, tl, first: None, blocking: true } } } + /// Similar to `SeqEx::try_receive`, except this function will return an iterator which + /// automatically calls `SeqEx::try_pump` as many times as it can before either the receive + /// window is empty, or one of the calls would block. pub fn try_receive_all>(&self, tl: TL, packet: Packet) -> RecvIter<'_, TL, SendData, RecvData, CAP> { let ret = self.try_receive(tl, packet); if let Ok((first, do_pump)) = ret { @@ -332,56 +399,145 @@ impl SeqEx { } } - pub fn try_send_with, F: FnOnce(SeqNo) -> SendData>( + /// Non-blocking variant of `SeqEx::send` that allows for a provided function to write data to + /// the send window. + /// See the documentation for `SeqEx::send` for more information. + /// + /// This function receives the packet sequence number the packet will be assigned. + /// All calls to `TransportLayer::send` involving this packet will receive this exact same + /// sequence number. + /// + /// NOTE: This function can block for a short period of time to lock an internal mutex. + /// + /// A return value of `Ok` contains the timestamp of when `service_ts` should be called next, + /// only if it has decrease. This can be safely ignored if `service_ts` is not being used. + pub fn try_send_with_ts, F: FnOnce(SeqNo) -> SendData>( &self, tl: TL, seq_cst: bool, packet_data: F, - ) -> Result<(), (TryError, F)> { + ) -> Result, (TryError, F)> { let mut inner = self.inner.lock().unwrap(); - inner.seq.try_send_with(tl, seq_cst, packet_data) + let pre_nst = inner.seq.next_service_timestamp; + inner.seq.try_send_with(tl, seq_cst, packet_data).map(|()| { + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) + }) } - pub fn try_send>(&self, tl: TL, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> { + /// Non-blocking variant of `SeqEx::send`. + /// See the documentation for `SeqEx::send` for more information. + /// + /// NOTE: This function can block for a short period of time to lock an internal mutex. + /// + /// A return value of `Ok` contains the timestamp of when `service_ts` should be called next, + /// only if it has decrease. This can be safely ignored if `service_ts` is not being used. + pub fn try_send_ts>(&self, tl: TL, seq_cst: bool, packet_data: SendData) -> Result, (TryError, SendData)> { let mut inner = self.inner.lock().unwrap(); - inner.seq.try_send(tl, seq_cst, packet_data) + let pre_nst = inner.seq.next_service_timestamp; + inner.seq.try_send(tl, seq_cst, packet_data).map(|()| { + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) + }) } - pub fn send_with>(&self, tl: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) { + /// Variant of `SeqEx::send` that allows for a provided function to write data to the send + /// window. + /// See the documentation for `SeqEx::send` for more information. + /// + /// This function receives the packet sequence number the packet will be assigned. + /// All calls to `TransportLayer::send` involving this packet will receive this exact same + /// sequence number. + pub fn send_with_ts>(&self, tl: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) -> Option { let mut inner = self.inner.lock().unwrap(); + let mut pre_nst = inner.seq.next_service_timestamp; while let Err((e, p)) = inner.seq.try_send_with(tl, seq_cst, packet_data) { packet_data = p; match e { TryError::WaitingForRecv => { inner.recv_waiters += 1; inner = self.wait_on_recv.wait(inner).unwrap(); + pre_nst = inner.seq.next_service_timestamp; } TryError::WaitingForReply => { inner.reply_sender_waiters = true; inner = self.wait_on_reply_sender.wait(inner).unwrap(); + pre_nst = inner.seq.next_service_timestamp; } } } + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) } - pub fn send>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) { + /// Add the given `packet_data` to the send window, to be sent immediately, and to be resent + /// if the payload is not successfully received by the remote peer. + /// + /// If the either send or receive window is full, this function will block until they are not. + /// + /// By default this function guarantees lossless transport. When `seq_cst` is set to true, this + /// function also guarantees in-order transport. + /// + /// Lossless transport guarantees that all payloads will be received by the remote peer exactly + /// once, but not necessarily in the same order they were sent. + /// + /// In-order transport guarantees that all SeqCst payloads are received in the same order that + /// they were sent. + /// + /// Returns the timestamp of when `service_ts` should be called next, only if it has decrease. + /// This can be safely ignored if `service_ts` is not being used. + pub fn send>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) -> Option { let mut inner = self.inner.lock().unwrap(); + let mut pre_nst = inner.seq.next_service_timestamp; while let Err((e, p)) = inner.seq.try_send(tl, seq_cst, packet_data) { packet_data = p; match e { TryError::WaitingForRecv => { inner.recv_waiters += 1; inner = self.wait_on_recv.wait(inner).unwrap(); + pre_nst = inner.seq.next_service_timestamp; } TryError::WaitingForReply => { inner.reply_sender_waiters = true; inner = self.wait_on_reply_sender.wait(inner).unwrap(); + pre_nst = inner.seq.next_service_timestamp; } } } + let nst = inner.seq.next_service_timestamp; + (pre_nst > nst).then_some(nst) } - + /// Function which handles resending unacknowledged packets. + /// It returns the duration of time in milliseconds that should be waited + /// until calling this function again. + /// + /// This function should be called repeatedly in a loop, with the loop + /// sleeping the amount of time specified before calling again. pub fn service>(&self, tl: TL) -> i64 { self.inner.lock().unwrap().seq.service(tl) } + /// A variant of `SeqEx::service` for advanced users. + /// + /// Instead of returning a duration of time, this function returns the exact timestamp at which + /// this function should be called again, or i64::MAX if the send window is empty and nothing + /// currently needs to be resent. + /// + /// This can be used in combination with the return values of `SeqEx::send` and + /// `ReplyGuard::reply` to precisely schedule updates to `SeqEx`, allowing + /// updated to occur much less often. + /// + /// However this function can be very difficult to use correctly as it both requires a means + /// of dynamically scheduling calls, as well as discipline in correctly applying updates to + /// that schedule whenever `SeqEx::send`, `ReplyGuard::reply` or any of their variants are called. + /// + /// It is recommended to just use `SeqEx::service`. + pub fn service_ts(&mut self, mut tl: impl TransportLayer) -> i64 { + let mut inner = self.inner.lock().unwrap(); + let current_time = tl.time(); + let mut iter = None; + while let Some(p) = inner.seq.service_direct(current_time, &mut iter) { + tl.send(p) + } + inner.seq.next_service_timestamp + } } impl Default for SeqEx { fn default() -> Self { @@ -414,19 +570,28 @@ impl<'a, TL: TransportLayer, SendData, RecvData, const CAP: usize> Ite } } +/// An implementation of `TransportLayer` using std::sync::mpsc channels. +/// If you prefer channels instead of callbacks then you can use this. +/// +/// If you choose to use this with `SeqEx`, one instance of `MpscTransport` should be created, +/// and only that instance and clones of that instance should be used with `SeqEx` #[derive(Clone, Debug)] pub struct MpscTransport { + /// The sender channel of this instance. pub channel: Sender>, + /// A std::time::Instant for providing answers to `time()` callbacks. pub time: Instant, } impl MpscTransport { + /// Create a new instance of `MpscTransport`. pub fn new() -> (Self, Receiver>) { let (send, recv) = channel(); (Self { channel: send, time: std::time::Instant::now() }, recv) } - pub fn from_sender(send: Sender>) -> Self { - Self { channel: send, time: std::time::Instant::now() } + /// Create a new instance of `MpscTransport` using the provided `sender`. + pub fn from_sender(sender: Sender>) -> Self { + Self { channel: sender, time: std::time::Instant::now() } } } impl TransportLayer for &MpscTransport { From 0b86cd14e2726c069e0fd71e0c8d7a6a876dce90 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Mon, 16 Oct 2023 15:33:28 -0400 Subject: [PATCH 3/3] fixed error --- src/tokio.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tokio.rs b/src/tokio.rs index d34edfd..ef7ce43 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -6,7 +6,7 @@ use tokio::{ use crate::{ no_std::RecvOkRaw, - result::{RecvError, TryError}, + error::{RecvError, TryError}, Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, };