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

finalized new api

Browse Source
This commit is contained in:
Monica Moniot
2023-08-18 12:00:25 -04:00
parent 55b19e3af8
commit bdd85155a3
4 changed files with 66 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/file_download.rs
+1 -1
View File
@@ -70,7 +70,7 @@ fn process(peer: &Peer, recv_data: RecvOk<'_, &Transport, Payload, Payload, Payl
let mut i = 0;
while i < file.len() {
let j = file.len().min(i + FILE_CHUNK_SIZE);
seqex.send(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() });
seqex.send_locked(&transport, FileDownload { filename: filename.clone(), file_chunk: file[i..j].to_vec() });
i = j;
}
}
src/seq_queue.rs
+1 -1
View File
@@ -88,7 +88,7 @@ pub enum DirectError<RecvData> {
/// invalid to process it right now. No action needs to be taken by the caller.
OutOfSequence,
/// The Send Window is currently full. The received packet cannot be processed right now because
/// it could cause the send window to overflow. No action needs to be taken by the caller.
/// it could cause the send window to overflow.
WindowIsFull(Packet<RecvData>),
WindowIsLocked(Packet<RecvData>),
ResendAck(SeqNo),
src/single_thread.rs
+23 -11
View File
@@ -12,15 +12,24 @@ 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.
pub fn reply(self, packet_data: SendData) {
self.reply_with(|_, _| packet_data)
pub fn reply(self, packet_data: SendData) {
self.reply_inner(false, |_, _| packet_data)
}
pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
pub fn reply_locked(self, packet_data: SendData) {
self.reply_inner(true, |_, _| packet_data)
}
pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
self.reply_inner(false, packet_data)
}
pub fn reply_locked_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
self.reply_inner(true, packet_data)
}
fn reply_inner(mut self, locked: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
let mut app = None;
core::mem::swap(&mut app, &mut self.app);
let seq_no = self.seq.seq_no();
self.seq
.reply_raw(app.unwrap(), self.reply_no, self.locked, packet_data(seq_no, self.reply_no));
.reply_raw(app.unwrap(), self.reply_no, self.locked, locked, packet_data(seq_no, self.reply_no));
core::mem::forget(self);
}
}
@@ -45,7 +54,7 @@ pub enum Error<RecvData> {
/// invalid to process it right now. No action needs to be taken by the caller.
OutOfSequence,
/// The Send Window is currently full. The received packet cannot be processed right now because
/// it could cause the send window to overflow. No action needs to be taken by the caller.
/// it could cause the send window to overflow.
WindowIsFull(Packet<RecvData>),
WindowIsLocked(Packet<RecvData>),
}
@@ -133,18 +142,20 @@ impl_recvok!(RecvOk, &'a mut SeqEx<SendData, RecvData, CAP>);
pub(crate) use impl_recvok;
impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, packet_data: SendData) -> Result<(), SendData> {
pub fn try_send(&mut self, mut app: impl TransportLayer<SendData>, locked: bool, packet_data: SendData) -> Result<(), SendData> {
match self.try_send_direct(packet_data, app.time()) {
Ok(p) => {
Ok(mut p) => {
p.set_locking(locked);
app.send(p);
Ok(())
}
Err(e) => Err(e),
}
}
pub fn try_send_with(&mut self, mut app: impl TransportLayer<SendData>, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), ()> {
pub fn try_send_with(&mut self, mut app: impl TransportLayer<SendData>, locked: bool, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), ()> {
match self.try_send_direct_with(packet_data, app.time()) {
Ok(p) => {
Ok(mut p) => {
p.set_locking(locked);
app.send(p);
Ok(())
}
@@ -168,8 +179,9 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
Err(DirectError::WindowIsLocked(p)) => Err(Error::WindowIsLocked(p)),
}
}
pub fn reply_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, unlock: bool, packet_data: SendData) {
if let Some(p) = self.reply_raw_and_direct(reply_no, unlock, packet_data, app.time()) {
pub fn reply_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, unlock: bool, locked_packet: bool, packet_data: SendData) {
if let Some(mut p) = self.reply_raw_and_direct(reply_no, unlock, packet_data, app.time()) {
p.set_locking(locked_packet);
app.send(p)
}
}
src/sync.rs
+41 -16
View File
@@ -12,7 +12,7 @@ use crate::{Packet, PumpError, RecvOkRaw, SeqEx, SeqNo, TransportLayer, DEFAULT_
pub struct SeqExSync<SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
seq_ex: Mutex<(SeqEx<SendData, RecvData, CAP>, usize)>,
send_block: Condvar,
lock: Condvar,
recv_lock: Condvar,
}
pub struct ReplyGuard<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize = DEFAULT_WINDOW_CAP> {
@@ -27,15 +27,28 @@ 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.
pub fn reply(self, packet_data: SendData) {
self.reply_with(|_, _| packet_data)
pub fn reply(self, packet_data: SendData) {
self.reply_inner(false, |_, _| packet_data)
}
pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
pub fn reply_locked(self, packet_data: SendData) {
self.reply_inner(true, |_, _| packet_data)
}
pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
self.reply_inner(false, packet_data)
}
pub fn reply_locked_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
self.reply_inner(true, packet_data)
}
fn reply_inner(mut self, locked: bool, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
let mut app = None;
core::mem::swap(&mut app, &mut self.app);
let mut seq = self.seq.lock();
let seq_no = seq.seq_no();
seq.reply_raw(app.unwrap(), self.reply_no, self.locked, packet_data(seq_no, self.reply_no));
seq.reply_raw(app.unwrap(), self.reply_no, self.locked, locked, packet_data(seq_no, self.reply_no));
drop(seq);
if self.locked {
self.seq.recv_lock.notify_all();
}
core::mem::forget(self);
}
}
@@ -106,7 +119,7 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
Self {
seq_ex: Mutex::new((SeqEx::new(retry_interval, initial_seq_no), 0)),
send_block: Condvar::default(),
lock: Condvar::default(),
recv_lock: Condvar::default(),
}
}
@@ -127,7 +140,7 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
Err(crate::Error::OutOfSequence) => return Err(Error::OutOfSequence),
Err(crate::Error::WindowIsFull(_)) => return Err(Error::WindowIsFull),
Err(crate::Error::WindowIsLocked(p)) => {
seq = self.lock.wait(seq).unwrap();
seq = self.recv_lock.wait(seq).unwrap();
packet = p;
}
}
@@ -146,7 +159,7 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
Err(PumpError::OutOfSequence) => return Err(Error::OutOfSequence),
Err(PumpError::WindowIsFull) => return Err(Error::WindowIsFull),
Err(PumpError::WindowIsLocked) => {
seq = self.lock.wait(seq).unwrap();
seq = self.recv_lock.wait(seq).unwrap();
}
}
}
@@ -162,30 +175,42 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
ReplyIter { seq: None, app, first: None }
}
}
pub fn try_send<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: SendData) -> Result<(), SendData> {
self.try_send_with(app, |_| packet_data)
pub fn try_send<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, packet_data: SendData) -> Result<(), SendData> {
self.try_send_with(app, locked, |_| packet_data)
}
pub fn send_with<TL: TransportLayer<SendData>>(&self, app: TL, mut packet_data: impl FnMut(SeqNo) -> SendData) {
fn send_with_inner<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, mut packet_data: impl FnMut(SeqNo) -> SendData) {
let mut seq = self.seq_ex.lock().unwrap();
while let Err(()) = seq.0.try_send_with(app.clone(), &mut packet_data) {
while let Err(()) = seq.0.try_send_with(app.clone(), locked, &mut packet_data) {
seq.1 += 1;
seq = self.send_block.wait(seq).unwrap();
seq.1 -= 1;
}
}
pub fn send<TL: TransportLayer<SendData>>(&self, app: TL, mut packet_data: SendData) {
fn send_inner<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, mut packet_data: SendData) {
let mut seq = self.seq_ex.lock().unwrap();
while let Err(p) = seq.0.try_send(app.clone(), packet_data) {
while let Err(p) = seq.0.try_send(app.clone(), locked, packet_data) {
packet_data = p;
seq.1 += 1;
seq = self.send_block.wait(seq).unwrap();
seq.1 -= 1;
}
}
pub fn try_send_with<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> {
pub fn send(&self, app: impl TransportLayer<SendData>, packet_data: SendData) {
self.send_inner(app, false, packet_data)
}
pub fn send_locked(&self, app: impl TransportLayer<SendData>, packet_data: SendData) {
self.send_inner(app, true, packet_data)
}
pub fn send_with(&self, app: impl TransportLayer<SendData>, packet_data: impl FnMut(SeqNo) -> SendData) {
self.send_with_inner(app, false, packet_data)
}
pub fn send_locked_with(&self, app: impl TransportLayer<SendData>, packet_data: impl FnMut(SeqNo) -> SendData) {
self.send_with_inner(app, true, packet_data)
}
pub fn try_send_with<TL: TransportLayer<SendData>>(&self, app: TL, locked: bool, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> {
let mut seq = self.lock();
let seq_no = seq.seq_no();
seq.try_send(app, packet_data(seq_no))
seq.try_send(app, locked, packet_data(seq_no))
}
pub fn service<TL: TransportLayer<SendData>>(&self, app: TL) -> i64 {
self.lock().service(app)