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

improved API and fixed protocol bug

Browse Source
This commit is contained in:
Monica Moniot
2023-08-16 15:14:26 -04:00
parent eeabe072c0
commit b324584286
5 changed files with 90 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/calculator.rs
+11 -24
View File
@@ -16,7 +16,7 @@ enum Packet {
}
fn drop_packet() -> bool {
static RNG: Mutex<u32> = Mutex::new(12);
static RNG: Mutex<u32> = Mutex::new(43);
let mut rng = RNG.lock().unwrap();
*rng ^= *rng << 13;
*rng ^= *rng >> 17;
@@ -41,30 +41,17 @@ fn receive<'a>(
transport: &'a MpscTransport<Packet>,
value: &mut f32,
) {
let packet = recv.try_recv();
if !drop_packet() {
let do_pump = match packet {
Ok(PacketType::Ack { reply_no }) => {
seq.receive_ack(reply_no);
return;
}
Ok(PacketType::EmptyReply { reply_no }) => {
let result = seq.receive_empty_reply(reply_no);
result.is_some()
}
Ok(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, value);
true
} else {
false
while let Ok(packet) = recv.try_recv() {
if !drop_packet() {
match packet {
PacketType::EmptyReply { reply_no } => {
let _ = seq.receive_empty_reply(reply_no);
}
PacketType::Payload { seq_no, reply_no, payload } => {
for RecvSuccess { guard, packet, send_data } in seq.receive_iter(transport, seq_no, reply_no, payload) {
process(guard, packet, send_data, value);
}
}
}
_ => return,
};
if do_pump {
while let Ok(RecvSuccess { guard, packet, send_data }) = seq.pump(transport) {
process(guard, packet, send_data, value);
}
}
}
examples/hello_world.rs
+4 -17
View File
@@ -38,32 +38,19 @@ fn process(guard: ReplyGuard<'_, &MpscTransport<Packet>>, recv_packet: Packet, s
}
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;
}
match recv.recv().unwrap() {
PacketType::EmptyReply { reply_no } => {
let result = seq.receive_empty_reply(reply_no);
if let Some(Exclamation) = &result {
if let Ok(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
for RecvSuccess {guard, packet, send_data} in seq.receive_iter(transport, seq_no, reply_no, payload) {
process(guard, packet, send_data)
}
}
};
if do_pump {
while let Ok(RecvSuccess { guard, packet, send_data }) = seq.pump(transport) {
process(guard, packet, send_data);
}
}
}
src/seq_queue.rs
+26 -36
View File
@@ -46,10 +46,9 @@ pub const DEFAULT_RESEND_INTERVAL_MS: i64 = 200;
/// The initial sequence number for a default instance of SeqEx.
pub const DEFAULT_INITIAL_SEQ_NO: SeqNo = 1;
pub const DEFAULT_SEND_WINDOW_LEN: usize = 64;
pub const DEFAULT_RECV_WINDOW_LEN: usize = 64;
pub const DEFAULT_WINDOW_CAP: usize = 64;
pub struct SeqEx<TL: TransportLayer, const SLEN: usize = DEFAULT_SEND_WINDOW_LEN, const RLEN: usize = DEFAULT_RECV_WINDOW_LEN> {
pub struct SeqEx<TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
/// The interval at which packets will be resent if they have not yet been acknowledged by the
/// remote peer.
/// It can be statically or dynamically set, it is up to the user to decide.
@@ -57,12 +56,16 @@ pub struct SeqEx<TL: TransportLayer, const SLEN: usize = DEFAULT_SEND_WINDOW_LEN
pub next_service_timestamp: i64,
next_send_seq_no: SeqNo,
pre_recv_seq_no: SeqNo,
/// This could be made more efficient in SoA format.
send_window: [Option<SendEntry<TL>>; SLEN],
recv_window: [Option<RecvEntry<TL>>; RLEN],
/// The size of this array determines the maximum number of received packets that the
/// application may attempt to process concurrently before new received packets start being dropped.
concurrent_replies: [SeqNo; RLEN],
/// This could be made more efficient by changing to SoA format.
send_window: [Option<SendEntry<TL>>; CAP],
recv_window: [Option<RecvEntry<TL>>; CAP],
/// The size of this array determines the maximum number of received packets that the application
/// may attempt to process concurrently before new received packets start being dropped.
concurrent_replies: [SeqNo; CAP],
/// The total number of concurrent replies being processed. When a packet is received, a reply
/// number is issued for that packet. That reply number reserves resources for itself, so that
/// when `reply_raw` or `reply_empty_raw` are called with it, they are guaranteed not to fail.
/// To accomplish this we must track all issued reply numbers.
concurrent_replies_total: usize,
}
@@ -106,7 +109,7 @@ pub struct Iter<'a, TL: TransportLayer>(core::slice::Iter<'a, Option<SendEntry<T
/// the packet.
pub struct IterMut<'a, TL: TransportLayer>(core::slice::IterMut<'a, Option<SendEntry<TL>>>);
impl<TL: TransportLayer> SeqEx<TL> {
impl<TL: TransportLayer, const CAP: usize> SeqEx<TL, CAP> {
/// Creates a new instance of `SeqEx` for a new remote peer.
/// An instance of `SeqEx` expects to communicate with only exactly one other remote instance
/// of `SeqEx`.
@@ -140,7 +143,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
pub fn is_full(&self) -> bool {
// We claim that the window is full one entry before it is actually full for the sake of
// making it always possible for both peers to process at least one reply at all times.
self.is_full_inner(1)
self.is_full_inner(true)
}
/// Returns the next sequence number to be attached to the next sent packet.
/// This should be called before `SeqEx::send`, and the return value should be
@@ -214,13 +217,13 @@ impl<TL: TransportLayer> SeqEx<TL> {
// If either are the case then we know we will eventually send a reply to the packet.
// If neither are the case then we must send an empty reply so the remote peer can stop
// resending the packet.
for entry in &self.send_window {
if entry.as_ref().map_or(false, |e| e.reply_no == Some(seq_no)) {
for entry in self.send_window.iter().flatten() {
if entry.reply_no == Some(seq_no) {
return Err(Error::OutOfSequence);
}
}
for reply_no in self.concurrent_replies {
if reply_no == seq_no {
for i in 0..self.concurrent_replies_total {
if self.concurrent_replies[i] == seq_no {
return Err(Error::OutOfSequence);
}
}
@@ -232,7 +235,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
// If the send window is full we cannot safely process received packets,
// because there would be no way to reply.
// We can only process this packet if processing it would make space in the send window.
let is_full = self.is_full_inner(0);
let is_full = self.is_full_inner(false);
let i = seq_no as usize % self.recv_window.len();
if let Some(pre) = self.recv_window[i].as_mut() {
@@ -240,7 +243,6 @@ impl<TL: TransportLayer> SeqEx<TL> {
if is_next && !is_full {
self.recv_window[i] = None;
} else {
app.send_ack(seq_no);
return if is_full {
Err(Error::WindowIsFull)
} else {
@@ -261,10 +263,6 @@ impl<TL: TransportLayer> SeqEx<TL> {
Ok((seq_no, packet, data))
} else {
self.recv_window[i] = Some(RecvEntry { seq_no, reply_no, data: packet.into() });
if let Some(reply_no) = reply_no {
self.receive_ack(reply_no);
}
app.send_ack(seq_no);
if is_full {
Err(Error::WindowIsFull)
} else {
@@ -272,29 +270,21 @@ impl<TL: TransportLayer> SeqEx<TL> {
}
}
}
pub fn receive_ack(&mut self, reply_no: SeqNo) {
let slot = self.send_window_slot_mut(reply_no);
if let Some(entry) = slot.as_mut() {
if entry.seq_no == reply_no {
entry.next_resend_time = i64::MAX;
}
}
}
pub fn receive_empty_reply(&mut self, reply_no: SeqNo) -> Option<TL::SendData> {
pub fn receive_empty_reply(&mut self, reply_no: SeqNo) -> Result<TL::SendData, Error> {
let slot = self.send_window_slot_mut(reply_no);
if slot.as_ref().map_or(false, |e| e.seq_no == reply_no) {
let entry = slot.take().unwrap();
Some(entry.data)
Ok(entry.data)
} else {
None
Err(Error::OutOfSequence)
}
}
fn is_full_inner(&self, bias: u32) -> bool {
fn is_full_inner(&self, reserve_one: bool) -> bool {
if self.concurrent_replies_total >= self.concurrent_replies.len() {
return true;
}
for i in 0..self.concurrent_replies_total as u32 + 1 + bias {
for i in 0..self.concurrent_replies_total as u32 + 1 + reserve_one as u32 {
let slot = self.send_window_slot(self.next_send_seq_no.wrapping_add(i));
if slot.is_some() {
return true;
@@ -316,7 +306,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
let i = next_seq_no as usize % self.recv_window.len();
if self.recv_window[i].as_ref().map_or(false, |pre| pre.seq_no == next_seq_no) {
if self.is_full_inner(0) {
if self.is_full_inner(false) {
return Err(Error::WindowIsFull);
}
@@ -350,7 +340,7 @@ impl<TL: TransportLayer> SeqEx<TL> {
self.next_service_timestamp = next_resend_time;
app.update_service_time(next_resend_time, current_time);
}
let slot = self.send_window_slot_mut(reply_no);
let slot = self.send_window_slot_mut(seq_no);
debug_assert!(slot.is_none());
let entry = slot.insert(SendEntry {
seq_no,
src/sync.rs
+48 -26
View File
@@ -9,19 +9,19 @@ use std::{
};
use crate::{
Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RECV_WINDOW_LEN, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_SEND_WINDOW_LEN,
Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_WINDOW_CAP, DEFAULT_RESEND_INTERVAL_MS,
};
pub struct SeqExSync<TL: TransportLayer, const SLEN: usize = DEFAULT_SEND_WINDOW_LEN, const RLEN: usize = DEFAULT_RECV_WINDOW_LEN> {
seq_ex: Mutex<SeqEx<TL, SLEN, RLEN>>,
pub struct SeqExSync<TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq_ex: Mutex<SeqEx<TL, CAP>>,
/// The mutex above is always held when this value changes, hence it is safe to mutate.
/// We don't pack this as a component of the mutex to avoid having to reimplement MutexGuard.
wait_count: UnsafeCell<usize>,
send_block: Condvar,
}
pub struct ReplyGuard<'a, TL: TransportLayer>(&'a SeqExSync<TL>, TL, SeqNo);
impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
pub struct ReplyGuard<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP>(&'a SeqExSync<TL, CAP>, TL, SeqNo);
impl<'a, TL: TransportLayer, const CAP: usize> ReplyGuard<'a, TL, CAP> {
/// 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,
@@ -33,23 +33,23 @@ impl<'a, TL: TransportLayer> ReplyGuard<'a, TL> {
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer> Drop for ReplyGuard<'a, TL> {
impl<'a, TL: TransportLayer, const CAP: usize> Drop for ReplyGuard<'a, TL, CAP> {
fn drop(&mut self) {
let mut seq = self.0.seq_ex.lock().unwrap();
seq.reply_empty_raw(self.1.clone(), self.2);
}
}
pub struct RecvSuccess<'a, TL: TransportLayer, P> {
pub guard: ReplyGuard<'a, TL>,
pub struct RecvSuccess<'a, TL: TransportLayer, P: Into<TL::RecvData>, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub guard: ReplyGuard<'a, TL, CAP>,
pub packet: P,
pub send_data: Option<TL::SendData>,
}
impl<TL: TransportLayer> SeqExSync<TL> {
impl<TL: TransportLayer, const CAP: usize> SeqExSync<TL, CAP> {
pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
Self {
seq_ex: Mutex::new(SeqEx::new(retry_interval, initial_seq_no)),
seq_ex: Mutex::new(SeqEx::<TL, CAP>::new(retry_interval, initial_seq_no)),
wait_count: UnsafeCell::new(0),
send_block: Condvar::default(),
}
@@ -61,18 +61,32 @@ impl<TL: TransportLayer> SeqExSync<TL> {
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<RecvSuccess<'_, TL, P>, Error> {
) -> Result<RecvSuccess<'_, TL, P, CAP>, Error> {
let mut seq = self.lock();
let ret = seq.receive_raw(app.clone(), seq_no, reply_no, packet);
// TODO: double check blocking.
self.unblock(ret.is_ok());
ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
pub fn pump(&self, app: TL) -> Result<RecvSuccess<'_, TL, TL::RecvData>, Error> {
pub fn pump(&self, app: TL) -> Result<RecvSuccess<'_, TL, TL::RecvData, CAP>, Error> {
let mut seq = self.lock();
let ret = seq.pump_raw();
self.unblock(ret.is_ok());
ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
pub fn receive_iter(
&self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: TL::RecvData,
) -> ReplyIter<'_, TL, CAP> {
if let Ok(g) = self.receive(app.clone(), seq_no, reply_no, packet) {
ReplyIter { origin: Some(self), app, first: Some(g) }
} else {
ReplyIter { origin: None, app, first: None }
}
}
#[inline]
fn unblock(&self, is_ok: bool) {
let has_waiting = unsafe { *self.wait_count.get() > 0 };
@@ -94,21 +108,16 @@ impl<TL: TransportLayer> SeqExSync<TL> {
}
}
pub fn receive_ack(&self, reply_no: SeqNo) {
self.lock().receive_ack(reply_no)
}
pub fn receive_empty_reply(&self, reply_no: SeqNo) -> Option<TL::SendData> {
pub fn receive_empty_reply(&self, reply_no: SeqNo) -> Result<TL::SendData, Error> {
let ret = self.lock().receive_empty_reply(reply_no);
if ret.is_some() {
self.send_block.notify_one();
}
self.unblock(ret.is_ok());
ret
}
pub fn service(&self, app: TL) -> i64 {
self.lock().service(app)
}
pub fn lock(&self) -> MutexGuard<SeqEx<TL>> {
pub fn lock(&self) -> MutexGuard<SeqEx<TL, CAP>> {
self.seq_ex.lock().unwrap()
}
}
@@ -118,6 +127,25 @@ impl<TL: TransportLayer> Default for SeqExSync<TL> {
}
}
pub struct ReplyIter<'a, TL: TransportLayer, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: Option<&'a SeqExSync<TL, CAP>>,
app: TL,
first: Option<RecvSuccess<'a, TL, TL::RecvData, CAP>>
}
impl<'a, TL: TransportLayer> Iterator for ReplyIter<'a, TL> {
type Item = RecvSuccess<'a, TL, TL::RecvData>;
fn next(&mut self) -> Option<Self::Item> {
if let Some(g) = self.first.take() {
Some(g)
} else if let Some(origin) = self.origin {
origin.pump(self.app.clone()).ok()
} else {
None
}
}
}
#[derive(Clone)]
pub enum PacketType<Payload: Clone> {
Payload {
@@ -125,9 +153,6 @@ pub enum PacketType<Payload: Clone> {
reply_no: Option<SeqNo>,
payload: Payload,
},
Ack {
reply_no: SeqNo,
},
EmptyReply {
reply_no: SeqNo,
},
@@ -158,9 +183,6 @@ impl<Payload: Clone> TransportLayer for &MpscTransport<Payload> {
fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Payload) {
let _ = self.channel.send(PacketType::Payload { seq_no, reply_no, payload: payload.clone() });
}
fn send_ack(&mut self, reply_no: SeqNo) {
let _ = self.channel.send(PacketType::Ack { reply_no });
}
fn send_empty_reply(&mut self, reply_no: SeqNo) {
let _ = self.channel.send(PacketType::EmptyReply { reply_no });
}
src/transport_layer.rs
+1 -2
View File
@@ -6,7 +6,7 @@ use crate::SeqNo;
/// 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: Sized + Clone {
pub trait TransportLayer: Clone {
type RecvData;
type SendData;
@@ -15,6 +15,5 @@ pub trait TransportLayer: Sized + Clone {
fn update_service_time(&mut self, timestamp: i64, current_time: i64) {}
fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Self::SendData);
fn send_ack(&mut self, reply_no: SeqNo);
fn send_empty_reply(&mut self, reply_no: SeqNo);
}