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

Browse Source
This commit is contained in:
Monica Moniot
2023-08-17 10:08:25 -04:00
parent a6b3f85c8e
commit 79d34d623d
7 changed files with 92 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/calculator.rs
+5
View File
@@ -72,3 +72,8 @@ fn main() {
}
assert_eq!(value, remote_value);
}
#[test]
fn test() {
main()
}
examples/file_download.rs
+9 -6
View File
@@ -36,9 +36,7 @@ struct Peer {
receiver: Receiver<Vec<u8>>,
}
impl TransportLayer for &Transport {
type SendData = Packet;
impl TransportLayer<Packet> for &Transport {
fn time(&mut self) -> i64 {
self.time.elapsed().as_millis() as i64
}
@@ -129,13 +127,13 @@ fn receive(peer: &Peer) {
fn main() {
let mut filesystem2 = HashMap::new();
let mut file = Vec::from([0u8; 1 << 16]);
let mut file = vec![0; 1 << 16];
OsRng.fill_bytes(&mut file);
filesystem2.insert("File1".to_string(), file);
let mut file = Vec::from([0u8; 1 << 18]);
let mut file = vec![0; 1 << 18];
OsRng.fill_bytes(&mut file);
filesystem2.insert("File2".to_string(), file);
let mut file = Vec::from([0u8; 1 << 20]);
let mut file = vec![0; 1 << 20];
OsRng.fill_bytes(&mut file);
filesystem2.insert("File3".to_string(), file);
@@ -169,3 +167,8 @@ fn main() {
assert_eq!(peer1.filesystem.read().unwrap().deref(), peer2.filesystem.read().unwrap().deref());
}
#[test]
fn test() {
main()
}
examples/hello_world.rs
+5
View File
@@ -69,3 +69,8 @@ fn main() {
receive(&recv1, &seq1, &transport1);
receive(&recv2, &seq2, &transport2);
}
#[test]
fn test() {
main()
}
src/seq_queue.rs
+12 -15
View File
@@ -156,11 +156,11 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
}
/// Sends the given packet to the remote peer and adds it to the send window.
///
/// If `Ok` is returned then the packet was successfully sent.
///
/// If the return value is `Err` the queue is full and the packet will not be sent.
/// The caller must either cancel sending, abort the connection, or wait until a call to
/// `receive` or `receive_ack` returns `Ok` and try again.
///
/// If `Ok` is returned then the packet was successfully sent.
/// `receive`, `receive_ack` or `pump` returns `Ok` and try again.
///
/// `packet_data` should contain both the packet to be sent as well as any local metadata the
/// caller wants to store with the packet. This metadata allows the exchange to be stateful.
@@ -171,7 +171,7 @@ 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.
#[must_use = "The queue might be full causing the packet to not be sent"]
pub fn try_send(&mut self, mut app: impl TransportLayer<SendData = SendData>, packet_data: SendData) -> Result<(), SendData> {
pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, packet_data: SendData) -> Result<(), SendData> {
if self.is_full() {
return Err(packet_data);
}
@@ -192,9 +192,10 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
Ok(())
}
/// 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 = SendData>,
mut app: impl TransportLayer<SendData>,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
@@ -270,14 +271,9 @@ 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_ack(&mut self, reply_no: SeqNo) -> Result<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();
Ok(entry.data)
} else {
Err(Error::OutOfSequence)
}
self.take_send(reply_no).ok_or(Error::OutOfSequence)
}
fn is_full_inner(&self, reserve_one: bool) -> bool {
@@ -301,6 +297,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
None
}
}
/// If this returns `Ok` then `try_send` might succeed on next call.
pub fn pump_raw(&mut self) -> Result<(SeqNo, RecvData, Option<SendData>), Error> {
let next_seq_no = self.pre_recv_seq_no.wrapping_add(1);
let i = next_seq_no as usize % self.recv_window.len();
@@ -329,7 +326,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
/// 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.
pub fn reply_raw(&mut self, mut app: impl TransportLayer<SendData = SendData>, reply_no: SeqNo, packet_data: SendData) {
pub fn reply_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, packet_data: SendData) {
if self.remove_reservation(reply_no) {
let seq_no = self.next_send_seq_no;
self.next_send_seq_no = self.next_send_seq_no.wrapping_add(1);
@@ -352,7 +349,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
app.send(entry.seq_no, entry.reply_no, &entry.data);
}
}
pub fn ack_raw(&mut self, mut app: impl TransportLayer<SendData = SendData>, reply_no: SeqNo) {
pub fn ack_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo) {
if self.remove_reservation(reply_no) {
// Acks are only sent once. There is code in `receive_raw` to handle resending
// an ack in the event that the first one here was dropped by the network.
@@ -371,7 +368,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
false
}
pub fn service(&mut self, mut app: impl TransportLayer<SendData = SendData>) -> i64 {
pub fn service(&mut self, mut app: impl TransportLayer<SendData>) -> i64 {
let current_time = app.time();
let next_interval = current_time + self.resend_interval;
let mut next_activity = i64::MAX;
src/single_thread.rs
+6 -9
View File
@@ -1,11 +1,11 @@
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_WINDOW_CAP};
pub struct ReplyGuard<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(
&'a mut SeqEx<SendData, RecvData, CAP>,
TL,
SeqNo,
);
impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, 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,
@@ -16,20 +16,20 @@ impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP:
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
fn drop(&mut self) {
self.0.ack_raw(self.1.clone(), self.2)
}
}
pub struct RecvSuccess<'a, TL: TransportLayer<SendData = SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub struct RecvSuccess<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
pub packet: P,
pub send_data: Option<SendData>,
}
impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
pub fn receive<TL: TransportLayer<SendData = SendData>, P: Into<RecvData>>(
pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&mut self,
app: TL,
seq_no: SeqNo,
@@ -39,10 +39,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
self.receive_raw(app.clone(), seq_no, reply_no, packet)
.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
pub fn pump<TL: TransportLayer<SendData = SendData>>(
&mut self,
app: TL,
) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
pub fn pump<TL: TransportLayer<SendData>>(&mut self, app: TL) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
self.pump_raw()
.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
src/sync.rs
+53 -48
View File
@@ -1,5 +1,5 @@
use std::{
cell::UnsafeCell,
ops::{Deref, DerefMut},
sync::{
mpsc::{channel, Receiver, Sender},
Condvar, Mutex, MutexGuard,
@@ -10,78 +10,91 @@ use std::{
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP};
pub struct SeqExSync<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq_ex: Mutex<SeqEx<SendData, RecvData, 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>,
seq_ex: Mutex<(SeqEx<SendData, RecvData, CAP>, usize)>,
send_block: Condvar,
}
pub struct ReplyGuard<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP>(
&'a SeqExSync<SendData, RecvData, CAP>,
TL,
SeqNo,
);
impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, CAP> {
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> ReplyGuard<'a, TL, SendData, RecvData, 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,
/// and since each fragment will be received in order it will be trivial for them to reconstruct
/// the original file.
pub fn reply(self, packet_data: SendData) {
let mut seq = self.0.seq_ex.lock().unwrap();
let mut seq = self.0.lock();
seq.reply_raw(self.1.clone(), self.2, packet_data);
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
fn drop(&mut self) {
let mut seq = self.0.seq_ex.lock().unwrap();
let mut seq = self.0.lock();
seq.ack_raw(self.1.clone(), self.2);
}
}
pub struct RecvSuccess<'a, TL: TransportLayer<SendData = SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub struct RecvSuccess<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub guard: ReplyGuard<'a, TL, SendData, RecvData, CAP>,
pub packet: P,
pub send_data: Option<SendData>,
}
pub struct ReplyIter<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub struct ReplyIter<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: Option<&'a SeqExSync<SendData, RecvData, CAP>>,
app: TL,
first: Option<RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>>,
}
pub struct SeqExGuard<'a, SendData, RecvData, const CAP: usize>(MutexGuard<'a, (SeqEx<SendData, RecvData, CAP>, usize)>);
impl<'a, SendData, RecvData, const CAP: usize> Deref for SeqExGuard<'a, SendData, RecvData, CAP> {
type Target = SeqEx<SendData, RecvData, CAP>;
fn deref(&self) -> &Self::Target {
&self.0 .0
}
}
impl<'a, SendData, RecvData, const CAP: usize> DerefMut for SeqExGuard<'a, SendData, RecvData, CAP> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0 .0
}
}
impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
Self {
seq_ex: Mutex::new(SeqEx::new(retry_interval, initial_seq_no)),
wait_count: UnsafeCell::new(0),
seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)),
send_block: Condvar::default(),
}
}
pub fn receive<TL: TransportLayer<SendData = SendData>, P: Into<RecvData>>(
pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&self,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<RecvSuccess<'_, TL, P, SendData, RecvData, 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());
let mut seq = self.seq_ex.lock().unwrap();
let ret = seq.0.receive_raw(app.clone(), seq_no, reply_no, packet);
if seq.1 > 0 && ret.is_ok() {
self.send_block.notify_one();
}
ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
pub fn pump<TL: TransportLayer<SendData = SendData>>(&self, app: TL) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
let mut seq = self.lock();
let ret = seq.pump_raw();
self.unblock(ret.is_ok());
pub fn pump<TL: TransportLayer<SendData>>(&self, app: TL) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
let mut seq = self.seq_ex.lock().unwrap();
let ret = seq.0.pump_raw();
if seq.1 > 0 && ret.is_ok() {
self.send_block.notify_one();
}
ret.map(|(reply_no, packet, send_data)| RecvSuccess { guard: ReplyGuard(self, app, reply_no), packet, send_data })
}
pub fn receive_all<TL: TransportLayer<SendData = SendData>>(
pub fn receive_all<TL: TransportLayer<SendData>>(
&self,
app: TL,
seq_no: SeqNo,
@@ -94,38 +107,34 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
ReplyIter { origin: None, app, first: None }
}
}
#[inline]
fn unblock(&self, is_ok: bool) {
let has_waiting = unsafe { *self.wait_count.get() > 0 };
if has_waiting && is_ok {
self.send_block.notify_one();
}
}
pub fn try_send<TL: TransportLayer<SendData = SendData>>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> {
pub fn try_send<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> {
let mut seq = self.lock();
seq.try_send(app, packet_data)
}
pub fn send<TL: TransportLayer<SendData = SendData>>(&self, app: TL, mut packet_data: SendData) {
let mut seq = self.lock();
while let Err(p) = seq.try_send(app.clone(), packet_data) {
pub fn send<TL: TransportLayer<SendData>>(&self, app: TL, mut packet_data: SendData) {
let mut seq = self.seq_ex.lock().unwrap();
while let Err(p) = seq.0.try_send(app.clone(), packet_data) {
packet_data = p;
unsafe { *self.wait_count.get() += 1 }
seq.1 += 1;
seq = self.send_block.wait(seq).unwrap();
unsafe { *self.wait_count.get() -= 1 }
seq.1 -= 1;
}
}
pub fn receive_ack(&self, reply_no: SeqNo) -> Result<SendData, Error> {
let ret = self.lock().receive_ack(reply_no);
self.unblock(ret.is_ok());
let mut seq = self.seq_ex.lock().unwrap();
let ret = seq.0.receive_ack(reply_no);
if seq.1 > 0 && ret.is_ok() {
self.send_block.notify_one();
}
ret
}
pub fn service<TL: TransportLayer<SendData = SendData>>(&self, app: TL) -> i64 {
pub fn service<TL: TransportLayer<SendData>>(&self, app: TL) -> i64 {
self.lock().service(app)
}
pub fn lock(&self) -> MutexGuard<SeqEx<SendData, RecvData, CAP>> {
self.seq_ex.lock().unwrap()
pub fn lock(&self) -> SeqExGuard<'_, SendData, RecvData, CAP> {
SeqExGuard(self.seq_ex.lock().unwrap())
}
}
impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvData, CAP> {
@@ -133,10 +142,8 @@ impl<SendData, RecvData, const CAP: usize> Default for SeqExSync<SendData, RecvD
Self::new(DEFAULT_RESEND_INTERVAL_MS, DEFAULT_INITIAL_SEQ_NO)
}
}
unsafe impl<SendData, RecvData, const CAP: usize> Send for SeqExSync<SendData, RecvData, CAP> {}
unsafe impl<SendData, RecvData, const CAP: usize> Sync for SeqExSync<SendData, RecvData, CAP> {}
impl<'a, TL: TransportLayer<SendData = SendData>, SendData, RecvData, const CAP: usize> Iterator for ReplyIter<'a, TL, SendData, RecvData, CAP> {
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Iterator for ReplyIter<'a, TL, SendData, RecvData, CAP> {
type Item = RecvSuccess<'a, TL, RecvData, SendData, RecvData, CAP>;
fn next(&mut self) -> Option<Self::Item> {
if let Some(g) = self.first.take() {
@@ -173,9 +180,7 @@ impl<Payload: Clone> MpscTransport<Payload> {
Self { channel: send, time: std::time::Instant::now() }
}
}
impl<Payload: Clone> TransportLayer for &MpscTransport<Payload> {
type SendData = Payload;
impl<Payload: Clone> TransportLayer<Payload> for &MpscTransport<Payload> {
fn time(&mut self) -> i64 {
self.time.elapsed().as_millis() as i64
}
src/transport_layer.rs
+2 -4
View File
@@ -6,13 +6,11 @@ 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: Clone {
type SendData;
pub trait TransportLayer<SendData>: Clone {
fn time(&mut self) -> i64;
#[allow(unused)]
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(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &SendData);
fn send_ack(&mut self, reply_no: SeqNo);
}