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

incomplete updated

Browse Source
This commit is contained in:
Monica Moniot
2023-08-17 16:19:54 -04:00
parent fdc4b6c637
commit c2d09e6398
3 changed files with 46 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/seq_queue.rs
+4 -10
View File
@@ -201,7 +201,7 @@ 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_raw<P: Into<RecvData>>(
pub fn receive_direct<P: Into<RecvData>>(
&mut self,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
@@ -333,7 +333,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.
#[must_use]
pub fn reply_raw(&mut self, reply_no: SeqNo, packet_data: SendData, current_time: i64) -> Option<Payload<'_, SendData>> {
pub fn reply_direct(&mut self, reply_no: SeqNo, packet_data: SendData, current_time: i64) -> Option<Payload<'_, 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);
@@ -356,14 +356,8 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
None
}
}
pub fn ack_raw(&mut self, reply_no: SeqNo) -> Option<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.
Some(reply_no)
} else {
None
}
pub fn ack_direct(&mut self, reply_no: SeqNo) -> bool {
self.remove_reservation(reply_no)
}
fn remove_reservation(&mut self, reply_no: SeqNo) -> bool {
for i in 0..self.concurrent_replies_total {
src/single_thread.rs
+24 -6
View File
@@ -12,16 +12,14 @@ impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Rep
/// and since each fragment will be received in order it will be trivial for them to reconstruct
/// the original file.
pub fn reply(mut self, packet_data: SendData) {
if let Some(Payload { seq_no, reply_no, data }) = self.0.reply_raw(self.2, packet_data, self.1.time()) {
self.1.send(seq_no, reply_no, data)
}
self.0.reply_raw(self.1, self.2, packet_data);
core::mem::forget(self);
}
}
impl<'a, TL: TransportLayer<SendData>, SendData, RecvData, const CAP: usize> Drop for ReplyGuard<'a, TL, SendData, RecvData, CAP> {
fn drop(&mut self) {
if let Some(reply_no) = self.0.ack_raw(self.2) {
self.1.send_ack(reply_no)
if self.0.ack_direct(self.2) {
self.1.send_ack(self.2)
}
}
}
@@ -33,6 +31,26 @@ pub struct RecvSuccess<'a, TL: TransportLayer<SendData>, P: Into<RecvData>, Send
}
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_raw<P: Into<RecvData>>(
&mut self,
mut app: impl TransportLayer<SendData>,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<(SeqNo, P, Option<SendData>), Error> {
let ret = self.receive_direct(seq_no, reply_no, packet);
}
pub fn reply_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo, packet_data: SendData) {
if let Some(Payload { seq_no, reply_no, data }) = self.reply_direct(reply_no, packet_data, app.time()) {
app.send(seq_no, reply_no, data)
}
}
pub fn ack_raw(&mut self, mut app: impl TransportLayer<SendData>, reply_no: SeqNo) {
if self.ack_direct(reply_no) {
app.send_ack(reply_no)
}
}
pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&mut self,
app: TL,
@@ -40,7 +58,7 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
reply_no: Option<SeqNo>,
packet: P,
) -> Result<RecvSuccess<'_, TL, P, SendData, RecvData, CAP>, Error> {
self.receive_raw(seq_no, reply_no, packet)
self.receive_raw(app, 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>>(&mut self, app: TL) -> Result<RecvSuccess<'_, TL, RecvData, SendData, RecvData, CAP>, Error> {
src/sync.rs
+18 -32
View File
@@ -7,7 +7,7 @@ use std::{
time::Instant,
};
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP, Payload};
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>, usize)>,
@@ -25,24 +25,22 @@ 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(mut self, packet_data: SendData) {
self.reply_with(|_, _| packet_data)
pub fn reply(self, packet_data: SendData) {
let mut seq = self.origin.lock();
seq.reply_raw(self.app.clone(), self.reply_no, packet_data);
core::mem::forget(self);
}
pub fn reply_with(mut self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
pub fn reply_with(self, packet_data: impl FnOnce(SeqNo, SeqNo) -> SendData) {
let mut seq = self.origin.lock();
let seq_no = seq.seq_no();
if let Some(Payload { seq_no, reply_no, data }) = seq.reply_raw(self.reply_no, packet_data(seq_no, self.reply_no), self.app.time()) {
self.app.send(seq_no, reply_no, data)
}
seq.reply_raw(self.app.clone(), self.reply_no, packet_data(seq_no, self.reply_no));
core::mem::forget(self);
}
}
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.origin.lock();
if let Some(reply_no) = seq.ack_raw(self.reply_no) {
self.app.send_ack(reply_no)
}
seq.ack_raw(self.app.clone(), self.reply_no);
}
}
@@ -82,16 +80,13 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
pub fn receive<TL: TransportLayer<SendData>, P: Into<RecvData>>(
&self,
mut app: TL,
app: TL,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: P,
) -> Result<RecvSuccess<'_, TL, P, SendData, RecvData, CAP>, Error> {
let mut seq = self.seq_ex.lock().unwrap();
let ret = seq.0.receive_raw(seq_no, reply_no, packet);
if let Err(Error::ResendAck(ack_no)) = ret {
app.send_ack(ack_no);
}
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();
}
@@ -126,30 +121,21 @@ impl<SendData, RecvData, const CAP: usize> SeqExSync<SendData, RecvData, CAP> {
ReplyIter { origin: None, app, first: None }
}
}
pub fn try_send<TL: TransportLayer<SendData>>(&self, mut 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();
let ret = seq.try_send(packet_data, app.time());
if let Ok(Payload { seq_no, reply_no, data }) = ret {
app.send(seq_no, reply_no, data)
}
ret.map(|_| ())
seq.try_send(app, packet_data)
}
fn send_inner<TL: TransportLayer<SendData>>(
&self,
mut seq: MutexGuard<'_, (SeqEx<SendData, RecvData, CAP>, usize)>,
mut app: TL,
app: TL,
mut packet_data: SendData,
) {
loop {
let ret = seq.0.try_send(packet_data, app.time());
if let Err(p) = ret {
packet_data = p;
seq.1 += 1;
seq = self.send_block.wait(seq).unwrap();
seq.1 -= 1;
} else if let Ok(Payload { seq_no, reply_no, data }) = ret {
app.send(seq_no, reply_no, data)
}
while let Err(p) = seq.0.try_send(app.clone(), packet_data) {
packet_data = p;
seq.1 += 1;
seq = self.send_block.wait(seq).unwrap();
seq.1 -= 1;
}
}
pub fn send<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: SendData) {