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

refactored repo and added docs

Browse Source
This commit is contained in:
Monica Moniot
2023-10-16 13:41:33 -04:00
parent 05e8a62ab3
commit 06aa1f72a4
15 changed files with 688 additions and 504 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -234,7 +234,7 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
[[package]]
name = "seq-ex"
version = "0.1.0"
version = "1.0.0"
dependencies = [
"rand_core",
"serde",
Cargo.toml
+20 -3
View File
@@ -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"
README.md
+10 -6
View File
@@ -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.
examples/calculator.rs
+5 -5
View File
@@ -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<Packet<Payload>>, seq: &SeqExSync<Payload, Payload>, transport: &MpscTransport<Payload>, value: &mut f32) {
fn receive(recv: &Receiver<Packet<Payload>>, seq: &SeqEx<Payload, Payload>, transport: &MpscTransport<Payload>, value: &mut f32) {
while let Ok(packet) = recv.try_recv() {
if drop_packet() {
continue;
@@ -42,8 +42,8 @@ fn receive(recv: &Receiver<Packet<Payload>>, seq: &SeqExSync<Payload, Payload>,
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;
examples/file_download.rs
+5 -5
View File
@@ -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<RwLock<HashMap<String, Vec<u8>>>>,
transport: Transport,
seqex: Arc<SeqExSync<Payload, Payload>>,
seqex: Arc<SeqEx<Payload, Payload>>,
receiver: Receiver<Vec<u8>>,
}
@@ -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,
};
examples/hello_world.rs
+5 -5
View File
@@ -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<Packet<Payload>>, seq: &SeqExSync<Payload, Payload>, transport: &MpscTransport<Payload>) {
fn receive(recv: &Receiver<Packet<Payload>>, seq: &SeqEx<Payload, Payload>, transport: &MpscTransport<Payload>) {
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<Packet<Payload>>, seq: &SeqExSync<Payload, Payload>,
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);
examples/hello_world_tokio.rs
+5 -5
View File
@@ -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>, Payload, P
Some(())
}
async fn say_hello(seq: &SeqExTokio<Payload, Payload>, transport: &MpscTransport<Payload>) -> Option<()> {
async fn say_hello(seq: &SeqEx<Payload, Payload>, transport: &MpscTransport<Payload>) -> 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<Payload, Payload>, transport: &MpscTransport
Some(())
}
fn peer_main(transport: MpscTransport<Payload>, mut recv: mpsc::Receiver<Packet<Payload>>) -> Arc<SeqExTokio<Payload, Payload>> {
let (seq, mut service) = SeqExTokio::new_default();
fn peer_main(transport: MpscTransport<Payload>, mut recv: mpsc::Receiver<Packet<Payload>>) -> Arc<SeqEx<Payload, Payload>> {
let (seq, mut service) = SeqEx::new_default();
let peer = Arc::new(seq);
let peer_weak = Arc::downgrade(&peer);
let tl = transport.clone();
phi_accural.rs
-145
View File
@@ -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<Packet>>, recv_packet: Packet, send_packet: Option<Packet>) {
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<PacketType<Packet>>, seq: &SeqExSync<&'a MpscTransport<Packet>>, transport: &'a MpscTransport<Packet>) {
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);
}
src/lib.rs
+49 -5
View File
@@ -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;
src/seq_queue.rs → src/no_std.rs
+7 -168
View File
@@ -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<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
@@ -103,128 +64,6 @@ impl<SendData> SendEntry<SendData> {
}
}
#[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<RecvData> {
Payload(SeqNo, RecvData),
SeqCstPayload(SeqNo, RecvData),
Reply(SeqNo, SeqNo, RecvData),
SeqCstReply(SeqNo, SeqNo, RecvData),
Ack(SeqNo),
}
use Packet::*;
impl<RecvData> Packet<RecvData> {
pub fn new_with_data(seq_no: SeqNo, reply_no: Option<SeqNo>, seq_cst: bool, data: RecvData) -> Self {
Self::new(Some(seq_no), reply_no, seq_cst, Some(data)).unwrap()
}
pub fn new(seq_no: Option<SeqNo>, reply_no: Option<SeqNo>, seq_cst: bool, data: Option<RecvData>) -> Option<Self> {
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<SendData>(self, f: impl FnOnce(RecvData) -> SendData) -> Packet<SendData> {
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<RecvData> {
self.consume().ok()
}
pub fn consume(self) -> Result<RecvData, SeqNo> {
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<RecvData: Clone> Packet<&RecvData> {
pub fn cloned(&self) -> Packet<RecvData> {
self.map(|d| d.clone())
}
}
#[derive(Clone, Debug)]
pub enum RecvOkRaw<SendData, RecvData> {
Payload {
src/result.rs
+113
View File
@@ -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 {}
src/single_thread.rs
+53 -54
View File
@@ -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>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a mut SeqEx<SendData, RecvData, CAP>,
app: Option<TL>,
tl: TL,
reply_no: SeqNo,
is_holding_lock: bool,
has_replied: bool,
}
impl<'a, TL: TransportLayer<SendData>, 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>, 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<SendData, RecvData, CAP>, 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<SendData, RecvData, CAP>, 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<SendData, RecvData, CAP>, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self {
Self::new(seq, tl, reply_no, is_holding_lock)
}
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
@@ -55,27 +76,6 @@ impl<'a, TL: TransportLayer<SendData>, 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>, 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>, SendData, RecvData, const CAP: usize> $recv<'a, TL, SendData, RecvData, CAP> {
fn from_raw(seq: $seq_ex, app: TL, value: RecvOkRaw<SendData, RecvData>) -> Self {
fn from_raw(seq: $seq_ex, tl: TL, value: RecvOkRaw<SendData, RecvData>) -> 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// Can mutate `next_service_timestamp`.
pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, 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<SendData>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// Can mutate `next_service_timestamp`.
pub fn try_send_with<F: FnOnce(SeqNo) -> SendData>(
&mut self,
mut app: impl TransportLayer<SendData>,
mut tl: impl TransportLayer<SendData>,
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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// If this returns `Ok` then `try_send` might succeed on next call.
pub fn receive_raw<P: Into<RecvData>>(
&mut self,
mut app: impl TransportLayer<SendData>,
mut tl: impl TransportLayer<SendData>,
packet: Packet<P>,
) -> Result<(RecvOkRaw<SendData, P>, 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// 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<SendData>, 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<SendData>, 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<SendData>, reply_no: SeqNo, unlock: bool) -> bool {
pub fn ack_raw(&mut self, mut tl: impl TransportLayer<SendData>, 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<SendData>) -> i64 {
let current_time = app.time();
pub fn service(&mut self, mut tl: impl TransportLayer<SendData>) -> 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<TL: TransportLayer<SendData>>(
&mut self,
app: TL,
tl: TL,
packet: Packet<RecvData>,
) -> 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<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&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))
}
}
src/sync.rs
+164 -58
View File
@@ -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<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
/// 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<u8>`, 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<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
inner: Mutex<SeqExInner<SendData, RecvData, CAP>>,
wait_on_recv: Condvar,
wait_on_reply_sender: Condvar,
wait_on_reply_receiver: Condvar,
}
struct SeqExInner<SendData, RecvData, const CAP: usize> {
seq: SeqEx<SendData, RecvData, CAP>,
seq: crate::no_std::SeqEx<SendData, RecvData, CAP>,
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>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a SeqExSync<SendData, RecvData, CAP>,
app: Option<TL>,
seq: &'a SeqEx<SendData, RecvData, CAP>,
tl: TL,
reply_no: SeqNo,
is_holding_lock: bool,
has_replied: bool,
}
impl<'a, TL: TransportLayer<SendData>, 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<SendData, RecvData, CAP>, 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<SendData, RecvData, CAP>, 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<SendData, RecvData, CAP>, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self {
Self::new(seq, tl, reply_no, is_holding_lock)
}
}
impl<'a, TL: TransportLayer<SendData>, 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>, SendData, RecvData, const CAP:
send_data: SendData,
},
}
crate::impl_recvok!(RecvOk, &'a SeqExSync<SendData, RecvData, CAP>);
crate::no_std::impl_recvok!(RecvOk, &'a SeqEx<SendData, RecvData, CAP>);
pub struct RecvIter<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
app: TL,
seq: Option<&'a SeqEx<SendData, RecvData, CAP>>,
tl: TL,
first: Option<RecvOk<'a, TL, SendData, RecvData, CAP>>,
blocking: bool,
}
impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// 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<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
pub fn try_receive<TL: TransportLayer<SendData>>(
&self,
app: TL,
tl: TL,
packet: Packet<RecvData>,
) -> 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<TL: TransportLayer<SendData>>(
&self,
app: TL,
packet: Packet<RecvData>,
) -> 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<TL: TransportLayer<SendData>>(&self, tl: TL, packet: Packet<RecvData>) -> 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<TL: TransportLayer<SendData>>(&self, app: TL) -> Result<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool), TryError> {
pub fn try_pump<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&self, app: TL) -> Option<(RecvOk<'_, TL, SendData, RecvData, CAP>, bool)> {
pub fn pump<TL: TransportLayer<SendData>>(&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<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
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<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
}
}
pub fn receive_all<TL: TransportLayer<SendData>>(&self, app: TL, packet: Packet<RecvData>) -> RecvIter<'_, TL, SendData, RecvData, CAP> {
let ret = self.receive(app.clone(), packet);
pub fn receive_all<TL: TransportLayer<SendData>>(&self, tl: TL, packet: Packet<RecvData>) -> 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<TL: TransportLayer<SendData>>(&self, app: TL, packet: Packet<RecvData>) -> RecvIter<'_, TL, SendData, RecvData, CAP> {
let ret = self.try_receive(app.clone(), packet);
pub fn try_receive_all<TL: TransportLayer<SendData>>(&self, tl: TL, packet: Packet<RecvData>) -> 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<TL: TransportLayer<SendData>, 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<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, packet_data: SendData) -> Result<(), (TryError, SendData)> {
pub fn try_send<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) {
pub fn send_with<TL: TransportLayer<SendData>>(&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<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
}
}
}
pub fn send<TL: TransportLayer<SendData>>(&self, app: TL, seq_cst: bool, mut packet_data: SendData) {
pub fn send<TL: TransportLayer<SendData>>(&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<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
}
}
pub fn service<TL: TransportLayer<SendData>>(&self, app: TL) -> i64 {
self.inner.lock().unwrap().seq.service(app)
pub fn service<TL: TransportLayer<SendData>>(&self, tl: TL) -> i64 {
self.inner.lock().unwrap().seq.service(tl)
}
}
impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvData, CAP> {
impl<SendData, RecvData, const CAP: usize> Default for SeqEx<SendData, RecvData, CAP> {
fn default() -> Self {
Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO)
}
@@ -290,9 +396,9 @@ impl<'a, TL: TransportLayer<SendData>, 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 {
src/tokio.rs
+62 -42
View File
@@ -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<SendData, RecvData> = (oneshot::Sender<Option<(SeqNo, bool, RecvData)>>, SendData);
type Receiver<RecvData> = (oneshot::Sender<(SeqNo, bool, RecvData)>, RecvData);
pub struct SeqExTokio<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub struct SeqEx<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
inner: Mutex<SeqExInner<SendData, RecvData, CAP>>,
wait_on_recv: Notify,
wait_on_reply: Notify,
@@ -19,38 +21,50 @@ pub struct SeqExTokio<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>
}
struct SeqExInner<SendData, RecvData, const CAP: usize> {
seq: SeqEx<Sender<SendData, RecvData>, Receiver<RecvData>, CAP>,
seq: crate::no_std::SeqEx<Sender<SendData, RecvData>, Receiver<RecvData>, CAP>,
recv_waiters: usize,
reply_waiters: bool,
}
pub struct ReplyGuard<'a, TL: TokioLayer<SendData = SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a SeqExTokio<SendData, RecvData, CAP>,
app: Option<TL>,
seq: &'a SeqEx<SendData, RecvData, CAP>,
tl: TL,
reply_no: SeqNo,
is_holding_lock: bool,
has_replied: bool,
}
impl<'a, TL: TokioLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
fn new(seq: &'a SeqExTokio<SendData, RecvData, CAP>, 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<SendData, RecvData>) -> Option<i64> {
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<SendData, RecvData>) -> Option<i64> {
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 = SendData>, 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<SendData, RecvData, CAP>, app: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self {
Self::new(seq, app, reply_no, is_holding_lock)
fn new(seq: &'a SeqEx<SendData, RecvData, CAP>, 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<SendData, RecvData, CAP>, tl: TL, reply_no: SeqNo, is_holding_lock: bool) -> Self {
Self::new(seq, tl, reply_no, is_holding_lock)
}
}
impl<'a, TL: TokioLayer<SendData = SendData>, 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<IntoOneshot<'a, RecvData>> for Receiver<RecvData> {
}
}
impl<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
}
fn receive_inner<TL: TokioLayer<SendData = SendData>>(
&self,
app: TL,
tl: TL,
packet: Packet<IntoOneshot<'_, RecvData>>,
) -> Result<(ReplyGuard<'_, TL, SendData, RecvData, CAP>, RecvData), Option<AsyncRecvError>> {
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<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
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<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
}
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<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
pub async fn receive<TL: TokioLayer<SendData = SendData>>(
&self,
app: TL,
tl: TL,
packet: Packet<RecvData>,
) -> 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<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
fn send_with_inner<TL: TokioLayer<SendData = SendData>, F: FnOnce(SeqNo) -> Sender<SendData, RecvData>>(
&self,
app: TL,
tl: TL,
seq_cst: bool,
packet_data: F,
) -> Result<Option<i64>, (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<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
pub async fn send<TL: TokioLayer<SendData = SendData>>(
&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<TL: TokioLayer<SendData = SendData>>(
&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<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
/// This function must be called with the same ServiceState instance returned upon creation of
/// the given SeqExTokio instance.
pub async fn service_task<TL: TokioLayer<SendData = SendData>>(&self, mut app: TL, state: &mut ServiceState) {
pub async fn service_task<TL: TokioLayer<SendData = SendData>>(&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<SendData, RecvData, const CAP: usize> SeqExTokio<SendData, RecvData, CAP> {
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;
src/transport_layer.rs
+189 -2
View File
@@ -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<u8>
/// 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<RecvData> {
/// 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<RecvData> Packet<RecvData> {
/// 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<SeqNo>, 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<SeqNo>, reply_no: Option<SeqNo>, seq_cst: bool, data: Option<RecvData>) -> Option<Self> {
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<RecvData> 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<RecvData> to Packet<T> 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<T>(self, f: impl FnOnce(RecvData) -> T) -> Packet<T> {
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<RecvData> {
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<RecvData, SeqNo> {
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<RecvData: Clone> Packet<&RecvData> {
/// Maps a Packet<&RecvData> to Packet<RecvData> by cloning any data it containts.
pub fn cloned(&self) -> Packet<RecvData> {
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<u8>` and `Arc<[u8]>`.
pub trait TransportLayer<SendData>: Clone {
pub trait TransportLayer<SendData>: 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>);
}