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

added lots of docs

Browse Source
This commit is contained in:
Monica Moniot
2023-10-16 15:32:10 -04:00
parent 06aa1f72a4
commit 0c9e248c58
5 changed files with 194 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/result.rs → src/error.rs
View File
src/lib.rs
+2 -1
View File
@@ -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.
src/no_std.rs
+6 -5
View File
@@ -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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// 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<Packet<&SendData>, (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<F: FnOnce(SeqNo) -> SendData>(
&mut self,
current_time: i64,
@@ -392,7 +392,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// 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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
}
}
/// Can mutate `next_service_timestamp`.
/// Can increase `next_service_timestamp`.
#[inline]
pub fn service_direct(&mut self, current_time: i64, iter: &mut Option<ServiceIter>) -> Option<Packet<&SendData>> {
if self.next_service_timestamp <= current_time {
let iter = iter.get_or_insert(ServiceIter {
src/single_thread.rs
+5 -5
View File
@@ -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>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
@@ -150,7 +150,7 @@ impl_recvok!(RecvOk, &'a mut SeqEx<SendData, RecvData, CAP>);
pub(crate) use impl_recvok;
impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// Can mutate `next_service_timestamp`.
/// Can decrease `next_service_timestamp`.
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) => {
@@ -160,7 +160,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
Err(e) => Err(e),
}
}
/// Can mutate `next_service_timestamp`.
/// Can decrease `next_service_timestamp`.
pub fn try_send_with<F: FnOnce(SeqNo) -> SendData>(
&mut self,
mut tl: impl TransportLayer<SendData>,
@@ -193,7 +193,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
false
}
}
/// Can mutate `next_service_timestamp`.
/// Can increase `next_service_timestamp`.
pub fn service(&mut self, mut tl: impl TransportLayer<SendData>) -> i64 {
let current_time = tl.time();
let mut iter = None;
src/sync.rs
+181 -16
View File
@@ -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>, 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>, 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<i64> {
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<i64> {
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>, 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>, 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<SendData, RecvData, CAP>);
/// 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>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: Option<&'a SeqEx<SendData, RecvData, CAP>>,
tl: TL,
@@ -249,7 +285,14 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<TL: TransportLayer<SendData>>(
&self,
tl: TL,
@@ -265,7 +308,13 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
}
}
/// 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<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 {
@@ -274,6 +323,10 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<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() {
@@ -284,6 +337,16 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<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.
@@ -304,7 +367,8 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
}
}
}
/// 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<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 {
@@ -318,6 +382,9 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
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<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 {
@@ -332,56 +399,145 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
}
}
pub fn try_send_with<TL: TransportLayer<SendData>, 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<TL: TransportLayer<SendData>, F: FnOnce(SeqNo) -> SendData>(
&self,
tl: TL,
seq_cst: bool,
packet_data: F,
) -> Result<(), (TryError, F)> {
) -> Result<Option<i64>, (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<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&self, tl: TL, seq_cst: bool, packet_data: SendData) -> Result<Option<i64>, (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<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&self, tl: TL, seq_cst: bool, mut packet_data: impl FnOnce(SeqNo) -> SendData) -> Option<i64> {
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<TL: TransportLayer<SendData>>(&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<TL: TransportLayer<SendData>>(&self, tl: TL, seq_cst: bool, mut packet_data: SendData) -> Option<i64> {
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<TL: TransportLayer<SendData>>(&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<SendData>) -> 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<SendData, RecvData, const CAP: usize> Default for SeqEx<SendData, RecvData, CAP> {
fn default() -> Self {
@@ -414,19 +570,28 @@ impl<'a, TL: TransportLayer<SendData>, 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<Payload: Clone> {
/// The sender channel of this instance.
pub channel: Sender<Packet<Payload>>,
/// A std::time::Instant for providing answers to `time()` callbacks.
pub time: Instant,
}
impl<Payload: Clone> MpscTransport<Payload> {
/// Create a new instance of `MpscTransport`.
pub fn new() -> (Self, Receiver<Packet<Payload>>) {
let (send, recv) = channel();
(Self { channel: send, time: std::time::Instant::now() }, recv)
}
pub fn from_sender(send: Sender<Packet<Payload>>) -> 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<Packet<Payload>>) -> Self {
Self { channel: sender, time: std::time::Instant::now() }
}
}
impl<Payload: Clone> TransportLayer<Payload> for &MpscTransport<Payload> {