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

Merge pull request #4 from zerotier/dev

Browse Source 
Improved type-safety for advanced users
This commit is contained in:
Monica Moniot
2023-10-25 12:39:00 -04:00
committed by GitHub
parent f0168fde39 d4aed2dcc6
commit d9fc950d40
5 changed files with 351 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib.rs
+1 -1
View File
@@ -33,7 +33,7 @@
//! different protocol layer.
//!
//! Neither SEQEX nor TCP are cryptographically secure.
#![warn(missing_docs, rust_2018_idioms)]
//#![warn(missing_docs, rust_2018_idioms)]
mod transport_layer;
pub use transport_layer::*;
src/no_std.rs
+14
View File
@@ -13,6 +13,8 @@ pub struct SeqEx<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
/// remote peer.
/// It can be statically or dynamically set.
pub resend_interval: i64,
/// The timestamp at which `service_direct` should be called again.
/// This can be `i64::MAX` does not currently need to be called again.
pub next_service_timestamp: i64,
next_send_seq_no: SeqNo,
next_recv_seq_no: SeqNo,
@@ -431,6 +433,11 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
}
}
/// If `unlock` is true and the return value is `Some` pump may return new values.
///
/// # Panics
/// This function will debug panic if `unlock` is `true` while this instance of `SeqEx` is not
/// locked. This would be equivalent to double-unlocking a mutex, and is a sign the user is
/// not making correct use of SeqCst packets.
#[must_use]
pub fn ack_raw_and_direct(&mut self, reply_no: SeqNo, unlock: bool) -> Option<Packet<&SendData>> {
if self.remove_reservation(reply_no) {
@@ -443,6 +450,13 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
None
}
}
/// If this is called, pump may return new values.
///
/// If this function is called then this instance of SeqEx will no longer be considered "locked".
/// Any calls to `ack_raw_and_direct` should take this into account and not set `unlock` to `true`.
pub fn unlock_raw(&mut self) {
self.is_locked = false;
}
/// Can increase `next_service_timestamp`.
#[inline]
src/single_thread.rs
+87 -21
View File
@@ -1,5 +1,5 @@
use crate::no_std::{RecvOkRaw, SeqEx};
use crate::error::{RecvError, TryError, TryRecvError};
use crate::no_std::{RecvOkRaw, SeqEx};
use crate::{Packet, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
@@ -7,64 +7,130 @@ pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, cons
tl: TL,
reply_no: SeqNo,
is_holding_lock: bool,
has_replied: bool,
}
pub struct SeqCstGuard<'a, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a mut SeqEx<SendData, RecvData, CAP>,
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
/// Returns a mutable reference to the `TransportLayer` instance that created this guard.
pub fn get_seqex(&'a mut self) -> &'a mut SeqEx<SendData, RecvData, CAP> {
self.seq
}
/// 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.
pub fn get_tl_mut(&mut self) -> &mut TL {
&mut self.tl
}
pub fn has_replied(&self) -> bool {
self.has_replied
}
/// Returns whether or nor this reply guard is currently holding the SeqCst lock,
/// preventing other SeqCst packets from being processed.
///
/// When this returns `true`, it means the current thread is within the critical section for
/// processing SeqCst packets. SeqCst packets can only enter this critical section in the same
/// order they were sent.
pub fn is_seq_cst(&self) -> bool {
self.is_holding_lock
}
pub fn ack(&mut self) {
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
/// 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.
pub fn reply(self, seq_cst: bool, packet_data: SendData) {
self.reply_with(seq_cst, |_, _| packet_data)
}
/// # 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) {
assert!(!self.has_replied, "Cannot reply after an ack has been sent");
self.has_replied = true;
pub fn reply_with(self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
let seq_no = self.seq.seq_no();
self.seq
.reply_raw(self.tl, self.reply_no, self.is_holding_lock, seq_cst, packet_data(seq_no, self.reply_no));
core::mem::forget(self);
}
fn consume_lock(self) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
let ret = if self.is_holding_lock {
let seq = self.seq as *mut SeqEx<SendData, RecvData, CAP>;
Some(SeqCstGuard { seq: unsafe { seq.as_mut().unwrap_unchecked() } })
} else {
None
};
core::mem::forget(self);
ret
}
pub fn ack(self) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
self.seq.ack_raw(self.tl, self.reply_no, false);
self.consume_lock()
}
pub fn unlock(&mut self) -> bool {
if self.is_holding_lock {
self.is_holding_lock = false;
self.seq.unlock_raw();
true
} else {
false
}
}
pub fn reply_stay_locked(self, seq_cst: bool, packet_data: SendData) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
self.reply_with_stay_locked(seq_cst, |_, _| packet_data)
}
pub fn reply_with_stay_locked(
self,
seq_cst: bool,
packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData,
) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
let seq_no = self.seq.seq_no();
self.seq
.reply_raw(self.tl, self.reply_no, false, seq_cst, packet_data(seq_no, self.reply_no));
self.consume_lock()
}
/// 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 the SeqCst lock.
///
/// This can be used in combination with `from_components` to move a `ReplyGuard` to a different
/// thread.
///
/// # Safety
/// The caller must guarantee that `ReplyGuard::from_components` is eventually called on
/// the returned values.
///
/// If this does not happen the SEQEX protocol will enter a deadlocked state.
pub unsafe fn to_components(self) -> (SeqNo, bool) {
let ret = (self.reply_no, self.is_holding_lock);
core::mem::forget(self);
ret
}
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 }
ReplyGuard { seq, tl, reply_no, is_holding_lock }
}
/// 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
/// originally returned by consuming a `ReplyGuard` instance with `to_components`.
///
/// `seq` must be the exact same instance of `SeqEx` that issued the original `ReplyGuard`.
///
/// Otherwise undefined behavior will occur.
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> {
fn drop(&mut self) {
self.ack();
self.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock);
}
}
impl<'a, SendData, RecvData, const CAP: usize> Drop for SeqCstGuard<'a, SendData, RecvData, CAP> {
fn drop(&mut self) {
self.seq.unlock_raw();
}
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> std::fmt::Debug for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
src/sync.rs
+144 -52
View File
@@ -7,8 +7,8 @@ use std::{
};
use crate::{
no_std::RecvOkRaw,
error::{RecvError, TryError},
no_std::RecvOkRaw,
Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP,
};
@@ -45,58 +45,49 @@ struct SeqExInner<SendData, RecvData, const CAP: usize> {
/// 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,
/// or a reply packet, 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.
/// In addition, if the data-containing packet was also a SeqCst packet, then this guard will act
/// like a `MutexGuard`. All other received SeqCst packets will be blocked until this guard is
/// dropped.
/// This creates a critical section for SeqCst packets that guarantees they are processed in the
/// exact same order they were sent.
///
/// 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 SeqEx<SendData, RecvData, CAP>,
tl: TL,
reply_no: SeqNo,
is_holding_lock: bool,
has_replied: bool,
}
/// A lock guard for maintaining the critical section for processing SeqCst packets. SeqCst packets
/// can only enter this critical section in the order they were sent by the remote peer.
pub struct SeqCstGuard<'a, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a SeqEx<SendData, RecvData, CAP>,
}
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` passed to `SeqEx::service`.
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.
/// Returns whether or nor this reply guard is currently holding the SeqCst lock,
/// preventing other SeqCst packets from being processed.
///
/// When this returns `true`, it means the current thread is within the critical section for
/// processing SeqCst packets. SeqCst packets can only enter this critical section in the same
/// order they were sent.
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 !self.has_replied {
self.has_replied = true;
let mut inner = self.seq.inner.lock().unwrap();
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.
///
@@ -106,12 +97,7 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Rep
///
/// Returns the timestamp of when `service_scheduled` should be called next, only if it has decrease.
/// This can be safely ignored if `service_scheduled` 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) -> Option<i64> {
assert!(!self.has_replied, "Cannot reply after an ack has been sent");
self.has_replied = true;
pub fn reply_with(self, seq_cst: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) -> Option<i64> {
let mut inner = self.seq.inner.lock().unwrap();
let pre_nst = inner.seq.next_service_timestamp;
let seq_no = inner.seq.seq_no();
@@ -135,44 +121,138 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Rep
/// 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) -> Option<i64> {
self.reply_with(seq_cst, |_, _| packet_data)
}
fn consume_lock(self, inner: MutexGuard<'a, SeqExInner<SendData, RecvData, CAP>>) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
let ret = if self.is_holding_lock {
Some(SeqCstGuard { seq: self.seq })
} else {
self.seq.notify_reply(inner);
None
};
core::mem::forget(self);
ret
}
/// Cause this reply guard to immediately send an ack to the remote peer,
/// consuming it and returning a new guard. This `SeqCstGuard` prevents SeqCst packets from
/// being processed by SeqEx until it is dropped.
///
/// If this guard was not for a SeqCst packet (or the user called `unlock` previously),
/// then it will return `None` instead of a `SeqCstGuard`.
///
/// This allows advanced users to de-couple locking the critical section for SeqCst packets
/// from sending acks to received packets.
/// In cases where the critical section has to be maintained for a long period of time this can
/// save bandwidth that would otherwise be wasted on resends.
/// It is not recommended to use this function except for this purpose.
///
/// Casual users are recommended to simply `drop` this reply guard instead of calling this function.
pub fn ack(self) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
let mut inner = self.seq.inner.lock().unwrap();
inner.seq.ack_raw(self.tl, self.reply_no, false);
self.consume_lock(inner)
}
/// If this guard is holding the SeqCst lock, and preventing other SeqCst packets from
/// being processed, then this function will force it to drop this lock.
/// If the next SeqCst packet is waiting in the receive window, then this will unblock a thread
/// blocked on `SeqEx::receive` or `SeqEx::pump` to process it.
///
/// Returns `true` if this guard was holding the SeqCst lock,
/// similar to function `ReplyGuard::is_seq_cst`.
///
/// This allows advanced users to de-couple locking the critical section for SeqCst packets
/// from sending acks to received packets.
/// In cases where a critical section is no longer needed, this function can allow more than one
/// thread to process SeqCst packets in parallel, improving performance.
/// It is not recommended to use this function except for this purpose.
///
/// Casual users are recommended to simply `drop` this reply guard instead of calling this function.
pub fn unlock(&mut self) -> bool {
if self.is_holding_lock {
self.is_holding_lock = false;
let mut inner = self.seq.inner.lock().unwrap();
inner.seq.unlock_raw();
self.seq.notify_reply(inner);
true
} else {
false
}
}
/// Send a reply to the remote peer that was created by the provided function.
/// Similar to `ReplyGuard::reply_with`, except this will not drop the SeqCst lock if
/// this guard is holding it.
///
/// See the documentation for `ReplyGuard::reply_stay_locked` for more information.
pub fn reply_with_stay_locked(
self,
seq_cst: bool,
packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData,
) -> (Option<SeqCstGuard<'a, SendData, RecvData, CAP>>, Option<i64>) {
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, false, seq_cst, packet_data(seq_no, self.reply_no));
let nst = inner.seq.next_service_timestamp;
(self.consume_lock(inner), (pre_nst > nst).then_some(nst))
}
/// Send a reply to the remote peer, similar to `ReplyGuard::reply`, except without dropping the
/// SeqCst lock.
///
/// If this guard was holding the SeqCst lock, then it will return a `SeqCstGuard`, which will
/// continue to hold the lock until it is dropped. If this guard was not holding the lock, it
/// will return `None` instead.
///
/// This allows advanced users to de-couple locking the critical section for SeqCst packets
/// from sending acks to received packets.
/// In cases where the critical section has to be maintained for a long period of time this can
/// save bandwidth that would otherwise be wasted on resends.
/// It is not recommended to use this function except for this purpose.
///
/// Casual users are recommended to use `ReplyGuard::reply` instead of calling this function.
///
/// The second return value is again the timestamp of when `service_scheduled` should be called
/// next, only if it has decrease.
pub fn reply_stay_locked(self, seq_cst: bool, packet_data: SendData) -> (Option<SeqCstGuard<'a, SendData, RecvData, CAP>>, Option<i64>) {
self.reply_with_stay_locked(seq_cst, |_, _| packet_data)
}
/// 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.
/// `is_seq_cst`, which states whether or not this `ReplyGuard` is holding the SeqCst 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.
/// If this does not happen the SEQEX protocol will enter a deadlocked state,
/// which is likely to cause threads to permanently block.
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
}
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 }
ReplyGuard { seq, tl, reply_no, is_holding_lock }
}
/// 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.
/// originally returned by consuming a `ReplyGuard` instance with `to_components`.
///
/// `seq` must be the exact same instance of `SeqEx` that issued the original `ReplyGuard`.
///
/// 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)
}
@@ -180,10 +260,15 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Rep
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 !self.has_replied {
self.has_replied = true;
inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock);
}
inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock);
self.seq.notify_reply(inner);
}
}
impl<'a, SendData, RecvData, const CAP: usize> Drop for SeqCstGuard<'a, SendData, RecvData, CAP> {
fn drop(&mut self) {
let mut inner = self.seq.inner.lock().unwrap();
// It is guaranteed at this point for SeqEx to be locked because it was locked on guard creation.
inner.seq.unlock_raw();
self.seq.notify_reply(inner);
}
}
@@ -312,9 +397,16 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// eventually processed with the SEQEX transport guarantees.
///
/// This function may block to preserve lossless or in-order transport.
/// In particular this function will block to guarantee SeqCst packets are processed in the same
/// order that they were sent. See `ReplyGuard` for more information.
///
/// 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.
/// If we are waiting to receive packets from the remote peer, this function will not block
/// and instead return `None`.
///
/// This function returns an option, where the first contained value is received data that
/// the caller can now process, and the second value is a boolean specifying if there is yet
/// more received data to be processed. This boolean is true if `SeqEx::pump` should be called,
/// because there are more 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 {
src/tokio.rs
+105 -23
View File
@@ -5,8 +5,8 @@ use tokio::{
};
use crate::{
no_std::RecvOkRaw,
error::{RecvError, TryError},
no_std::RecvOkRaw,
Packet, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP,
};
@@ -31,18 +31,27 @@ pub struct ReplyGuard<'a, TL: TokioLayer<SendData = SendData>, SendData, RecvDat
tl: TL,
reply_no: SeqNo,
is_holding_lock: bool,
has_replied: bool,
}
pub struct SeqCstGuard<'a, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq: &'a SeqEx<SendData, RecvData, CAP>,
}
impl<'a, TL: TokioLayer<SendData = 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.
pub fn get_tl_mut(&mut self) -> &mut TL {
&mut self.tl
}
pub fn has_replied(&self) -> bool {
self.has_replied
}
/// Returns whether or nor this reply guard is currently holding the SeqCst lock,
/// preventing other SeqCst packets from being processed.
///
/// When this returns `true`, it means the current thread is within the critical section for
/// processing SeqCst packets. SeqCst packets can only enter this critical section in the same
/// order they were sent.
pub fn is_seq_cst(&self) -> bool {
self.is_holding_lock
}
@@ -60,32 +69,19 @@ impl<'a, TL: TokioLayer<SendData = SendData>, SendData, RecvData, const CAP: usi
self.seq.notify_reply(inner);
ret
}
pub fn ack(&mut self) {
if !self.has_replied {
self.has_replied = true;
let mut inner = self.seq.inner.lock().unwrap();
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
/// 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.
pub async fn reply(self, seq_cst: bool, packet_data: SendData) -> Result<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, RecvData), AsyncError> {
self.reply_with(seq_cst, |_, _| packet_data).await
}
/// # Panic
/// This function will panic if `ack` has been called.
pub async fn reply_with(
mut self,
seq_cst: bool,
packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData,
) -> Result<(ReplyGuard<'a, TL, SendData, RecvData, CAP>, RecvData), AsyncError> {
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();
@@ -100,14 +96,95 @@ impl<'a, TL: TokioLayer<SendData = SendData>, SendData, RecvData, const CAP: usi
Ok((Self::new(seq, tl, reply_no, seq_cst), recv_data))
}
fn consume_lock(self, inner: MutexGuard<'a, SeqExInner<SendData, RecvData, CAP>>) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
let ret = if self.is_holding_lock {
Some(SeqCstGuard { seq: self.seq })
} else {
self.seq.notify_reply(inner);
None
};
core::mem::forget(self);
ret
}
/// Cause this reply guard to immediately send an ack to the remote peer,
/// consuming it and returning a new guard. This `SeqCstGuard` prevents SeqCst packets from
/// being processed by SeqEx until it is dropped.
///
/// If this guard was not for a SeqCst packet (or the user called `unlock` previously),
/// then it will return `None` instead of a `SeqCstGuard`.
///
/// This allows advanced users to de-couple locking the critical section for SeqCst packets
/// from sending acks to received packets.
/// In cases where the critical section has to be maintained for a long period of time this can
/// save bandwidth that would otherwise be wasted on resends.
/// It is not recommended to use this function except for this purpose.
///
/// Casual users are recommended to simply `drop` this reply guard instead of calling this function.
pub fn ack(self) -> Option<SeqCstGuard<'a, SendData, RecvData, CAP>> {
let mut inner = self.seq.inner.lock().unwrap();
inner.seq.ack_raw(self.tl, self.reply_no, false);
self.consume_lock(inner)
}
/// If this guard is holding the SeqCst lock, and preventing other SeqCst packets from
/// being processed, then this function will force it to drop this lock.
/// If the next SeqCst packet is waiting in the receive window, then this will unblock a thread
/// blocked on `SeqEx::receive` or `SeqEx::pump` to process it.
///
/// Returns `true` if this guard was holding the SeqCst lock,
/// similar to function `ReplyGuard::is_seq_cst`.
///
/// This allows advanced users to de-couple locking the critical section for SeqCst packets
/// from sending acks to received packets.
/// In cases where a critical section is no longer needed, this function can allow more than one
/// thread to process SeqCst packets in parallel, improving performance.
/// It is not recommended to use this function except for this purpose.
///
/// Casual users are recommended to simply `drop` this reply guard instead of calling this function.
pub fn unlock(&mut self) -> bool {
if self.is_holding_lock {
self.is_holding_lock = false;
let mut inner = self.seq.inner.lock().unwrap();
inner.seq.unlock_raw();
self.seq.notify_reply(inner);
true
} else {
false
}
}
/// 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 the SeqCst lock.
///
/// This can be used in combination with `from_components` to move a `ReplyGuard` to a different
/// thread.
///
/// # Safety
/// The caller must guarantee that `ReplyGuard::from_components` is eventually called on
/// the returned values.
///
/// If this does not happen the SEQEX protocol will enter a deadlocked state,
/// which is likely to cause threads to permanently block.
pub unsafe fn to_components(self) -> (SeqNo, bool) {
let ret = (self.reply_no, self.is_holding_lock);
core::mem::forget(self);
ret
}
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 }
ReplyGuard { seq, tl, reply_no, is_holding_lock }
}
/// 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
/// originally returned by consuming a `ReplyGuard` instance with `to_components`.
///
/// `seq` must be the exact same instance of `SeqEx` that issued the original `ReplyGuard`.
///
/// 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)
}
@@ -115,10 +192,15 @@ impl<'a, TL: TokioLayer<SendData = SendData>, SendData, RecvData, const CAP: usi
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 !self.has_replied {
self.has_replied = true;
inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock);
}
inner.seq.ack_raw(self.tl, self.reply_no, self.is_holding_lock);
self.seq.notify_reply(inner);
}
}
impl<'a, SendData, RecvData, const CAP: usize> Drop for SeqCstGuard<'a, SendData, RecvData, CAP> {
fn drop(&mut self) {
let mut inner = self.seq.inner.lock().unwrap();
// It is guaranteed at this point for SeqEx to be locked because it was locked on guard creation.
inner.seq.unlock_raw();
self.seq.notify_reply(inner);
}
}