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

cleared errors

Browse Source
This commit is contained in:
Monica Moniot
2023-08-17 15:11:07 -04:00
parent ac024d9da9
commit 604a9103a7
5 changed files with 110 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+12
View File
@@ -222,6 +222,18 @@ checksum = "17ed6077ed6cd6c74735e21f37eb16dc3935f96878b1fe961074089cc80893f9"
dependencies = [
"backtrace",
"pin-project-lite",
"tokio-macros",
]
[[package]]
name = "tokio-macros"
version = "2.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
Cargo.toml
+1 -1
View File
@@ -15,7 +15,7 @@ std = []
[dependencies]
serde = { version = "1.0.183", default-features = false, features = ["derive"], optional = true }
tokio = { version = "1.31.0", default-features = false, features = ["sync", "rt"], optional = true }
tokio = { version = "1.31.0", default-features = false, features = ["sync", "rt", "time", "macros"], optional = true }
[dev-dependencies]
rand_core = { version = "0.6.4", features = ["getrandom"]}
src/seq_queue.rs
-3
View File
@@ -183,7 +183,6 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
let next_resend_time = current_time + self.resend_interval;
if self.next_service_timestamp > next_resend_time {
self.next_service_timestamp = next_resend_time;
app.update_service_time(next_resend_time, current_time);
}
let slot = self.send_window_slot_mut(seq_no);
debug_assert!(slot.is_none());
@@ -336,7 +335,6 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
let next_resend_time = current_time + self.resend_interval;
if self.next_service_timestamp > next_resend_time {
self.next_service_timestamp = next_resend_time;
app.update_service_time(next_resend_time, current_time);
}
let slot = self.send_window_slot_mut(seq_no);
debug_assert!(slot.is_none());
@@ -382,7 +380,6 @@ impl<SendData, RecvData, const CAP: usize> SeqEx<SendData, RecvData, CAP> {
next_activity = next_activity.min(entry.next_resend_time);
}
self.next_service_timestamp = next_activity;
app.update_service_time(next_activity, current_time);
next_activity - current_time
} else {
self.next_service_timestamp - current_time
src/tokio.rs
+97 -60
View File
@@ -1,11 +1,9 @@
use std::{
ops::{Deref, DerefMut},
use std::
sync::{
Mutex, MutexGuard,
},
time::Instant,
};
use tokio::{task, sync::{Notify, oneshot}};
}
;
use tokio::{task, sync::{Notify, oneshot, mpsc}, time};
use crate::{Error, SeqEx, SeqNo, TransportLayer, DEFAULT_INITIAL_SEQ_NO, DEFAULT_RESEND_INTERVAL_MS, DEFAULT_WINDOW_CAP};
@@ -16,12 +14,12 @@ pub struct SeqExTokio<Packet, const CAP: usize = DEFAULT_WINDOW_CAP> {
send_block: Notify,
}
pub struct ReplyGuard<'a, TL: TransportLayer<SendData<Packet>>, Packet, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub struct ReplyGuard<'a, TL: TokioTransportLayer<Packet = Packet>, Packet, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: &'a SeqExTokio<Packet, CAP>,
app: Option<TL>,
app: Option<&'a TokioTransport<TL>>,
reply_no: SeqNo,
}
impl<'a, TL: TransportLayer<SendData<Packet>>, Packet, const CAP: usize> ReplyGuard<'a, TL, Packet, CAP> {
impl<'a, TL: TokioTransportLayer<Packet = Packet>, Packet, const CAP: usize> ReplyGuard<'a, TL, Packet, 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,
@@ -35,13 +33,19 @@ impl<'a, TL: TransportLayer<SendData<Packet>>, Packet, const CAP: usize> ReplyGu
let mut seq = self.origin.seq_ex.lock().unwrap();
let seq_no = seq.0.seq_no();
let app = self.app.take().unwrap();
seq.0.reply_raw(app.clone(), self.reply_no, (tx, packet(seq_no, self.reply_no)));
let pre_ts = seq.0.next_service_timestamp;
seq.0.reply_raw(app, self.reply_no, (tx, packet(seq_no, self.reply_no)));
if seq.0.next_service_timestamp != pre_ts {
let _ = app.update_queue.send(seq.0.next_service_timestamp).await;
}
let (packet, reply_no) = rx.await.ok()?;
let g = ReplyGuard { origin: self.origin, app: Some(app), reply_no };
Some((packet, g))
}
}
impl<'a, TL: TransportLayer<SendData<Packet>>, Packet, const CAP: usize> Drop for ReplyGuard<'a, TL, Packet, CAP> {
impl<'a, TL: TokioTransportLayer<Packet = Packet>, Packet, const CAP: usize> Drop for ReplyGuard<'a, TL, Packet, CAP> {
fn drop(&mut self) {
if let Some(app) = self.app.take() {
let mut seq = self.origin.seq_ex.lock().unwrap();
@@ -50,9 +54,9 @@ impl<'a, TL: TransportLayer<SendData<Packet>>, Packet, const CAP: usize> Drop fo
}
}
pub struct ReplyIter<'a, TL: TransportLayer<SendData<Packet>>, Packet, const CAP: usize = DEFAULT_WINDOW_CAP> {
pub struct ReplyIter<'a, TL: TokioTransportLayer<Packet = Packet>, Packet, const CAP: usize = DEFAULT_WINDOW_CAP> {
origin: Option<&'a SeqExTokio<Packet, CAP>>,
app: TL,
app: &'a TokioTransport<TL>,
first: Option<(Packet, ReplyGuard<'a, TL, Packet, CAP>)>,
}
@@ -69,6 +73,51 @@ pub struct ReplyIter<'a, TL: TransportLayer<SendData<Packet>>, Packet, const CAP
// &mut self.0 .0
// }
//}
#[derive(Clone)]
pub struct TokioTransport<TL: TokioTransportLayer> {
time: time::Instant,
update_queue: mpsc::Sender<i64>,
app: TL,
}
impl<TL: TokioTransportLayer> TokioTransport<TL> {
pub fn new<const CAP: usize, S: AsRef<SeqExTokio<TL::Packet, CAP>> + Send + 'static>(app: TL, seq: S) -> Self {
let (update_queue, mut recv) = mpsc::channel(4);
let ret = TokioTransport {
time: time::Instant::now(),
update_queue,
app,
};
let task_tl = ret.clone();
task::spawn(async move {
let mut update_ts = i64::MAX;
loop {
if update_ts < i64::MAX {
let diff = update_ts - task_tl.time.elapsed().as_millis() as i64;
let mut do_update = diff <= 0;
if diff > 0 {
let sleep = time::sleep(time::Duration::from_millis(diff as u64));
tokio::select!{
Some(up) = recv.recv() => {
update_ts = up;
}
_ = sleep => {
do_update = true;
}
};
}
if do_update {
let mut seq = seq.as_ref().seq_ex.lock().unwrap();
seq.0.service(&task_tl);
update_ts = seq.0.next_service_timestamp;
}
} else if let Some(up) = recv.recv().await {
update_ts = up;
}
}
});
ret
}
}
impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self {
@@ -78,8 +127,9 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
}
}
fn process<TL: TransportLayer<SendData<Packet>>>(
&self, app: TL,
fn process<'a, TL: TokioTransportLayer<Packet = Packet>>(
&'a self,
app: &'a TokioTransport<TL>,
mut seq: MutexGuard<'_, (SeqEx<SendData<Packet>, Packet, CAP>, usize)>,
result: Result<(SeqNo, Packet, Option<SendData<Packet>>), Error>
) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
@@ -88,7 +138,7 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
self.send_block.notify_one();
}
if let Some((tx, _)) = send_data {
if let Err(e) = tx.send((packet, reply_no)) {
if let Err(_) = tx.send((packet, reply_no)) {
// Allow the drop code to be run
seq.0.ack_raw(app, reply_no);
}
@@ -100,30 +150,30 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
None
}
}
pub fn receive<TL: TransportLayer<SendData<Packet>>>(
&self,
app: TL,
pub fn receive<'a, TL: TokioTransportLayer<Packet = Packet>>(
&'a self,
app: &'a TokioTransport<TL>,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: Packet,
) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
let mut seq = self.seq_ex.lock().unwrap();
let result = seq.0.receive_raw(app.clone(), seq_no, reply_no, packet);
let result = seq.0.receive_raw(app, seq_no, reply_no, packet);
self.process(app, seq, result)
}
pub fn pump<TL: TransportLayer<SendData<Packet>>>(&self, app: TL) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
pub fn pump<'a, TL: TokioTransportLayer<Packet = Packet>>(&'a self, app: &'a TokioTransport<TL>) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
let mut seq = self.seq_ex.lock().unwrap();
let result = seq.0.pump_raw();
self.process(app, seq, result)
}
pub fn receive_all<TL: TransportLayer<SendData<Packet>>>(
&self,
app: TL,
pub fn receive_all<'a, TL: TokioTransportLayer<Packet = Packet>>(
&'a self,
app: &'a TokioTransport<TL>,
seq_no: SeqNo,
reply_no: Option<SeqNo>,
packet: Packet,
) -> ReplyIter<'_, TL, Packet, CAP> {
if let Some(g) = self.receive(app.clone(), seq_no, reply_no, packet) {
if let Some(g) = self.receive(app, seq_no, reply_no, packet) {
ReplyIter { origin: Some(self), app, first: Some(g) }
} else {
ReplyIter { origin: None, app, first: None }
@@ -142,24 +192,29 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
// let mut seq = self.lock();
// seq.try_send(app, packet_data)
//}
async fn send_inner<TL: TransportLayer<SendData<Packet>>>(
async fn send_inner<TL: TokioTransportLayer<Packet = Packet>>(
&self,
mut seq: MutexGuard<'_, (SeqEx<SendData<Packet>, Packet, CAP>, usize)>,
app: TL,
app: &TokioTransport<TL>,
mut tx: oneshot::Sender<(Packet, SeqNo)>,
mut packet: Packet
) {
while let Err(e) = seq.0.try_send(app.clone(), (tx, packet)) {
let mut pre_ts = seq.0.next_service_timestamp;
while let Err(e) = seq.0.try_send(app, (tx, packet)) {
(tx, packet) = e;
seq.1 += 1;
drop(seq);
self.send_block.notified().await;
seq = self.seq_ex.lock().unwrap();
pre_ts = seq.0.next_service_timestamp;
seq.1 -= 1;
}
if seq.0.next_service_timestamp != pre_ts {
let _ = app.update_queue.send(seq.0.next_service_timestamp).await;
}
}
/// If this future is dropped then the remote peer's reply to this packet will also be dropped.
pub async fn send<TL: TransportLayer<SendData<Packet>>>(&self, app: TL, packet: Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
pub async fn send<'a, TL: TokioTransportLayer<Packet = Packet>>(&'a self, app: &'a TokioTransport<TL>, packet: Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
self.send_with(app, |_| packet).await
}
//pub fn try_send_with<TL: TransportLayer<SendData>>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> {
@@ -167,25 +222,15 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
// let seq_no = seq.seq_no();
// seq.try_send(app, packet_data(seq_no))
//}
pub async fn send_with<TL: TransportLayer<SendData<Packet>>>(&self, app: TL, packet: impl FnOnce(SeqNo) -> Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
pub async fn send_with<'a, TL: TokioTransportLayer<Packet = Packet>>(&'a self, app: &'a TokioTransport<TL>, packet: impl FnOnce(SeqNo) -> Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> {
let (tx, rx) = oneshot::channel();
let seq = self.seq_ex.lock().unwrap();
let seq_no = seq.0.seq_no();
self.send_inner(seq, app.clone(), tx, packet(seq_no)).await;
self.send_inner(seq, app, tx, packet(seq_no)).await;
// This can only return an error if the sender was dropped.
let (packet, reply_no) = rx.await.ok()?;
Some((packet, ReplyGuard { origin: self, app: Some(app), reply_no }))
}
pub async fn main<TL: TransportLayer<SendData<Packet>>>(&self, app: TL) {
let a = task::spawn(async{
});
}
pub fn service<TL: TransportLayer<SendData<Packet>>>(&self, app: TL) -> i64 {
self.seq_ex.lock().unwrap().0.service(app)
}
//pub fn lock(&self) -> SeqExGuard<'_, SendData, RecvData, CAP> {
// SeqExGuard(self.seq_ex.lock().unwrap())
//}
@@ -196,45 +241,37 @@ impl<Packet, const CAP: usize> SeqExTokio<Packet, CAP> {
// }
//}
impl<'a, TL: TransportLayer<SendData<Packet>>, Packet, const CAP: usize> Iterator for ReplyIter<'a, TL, Packet, CAP> {
impl<'a, TL: TokioTransportLayer<Packet = Packet>, Packet, const CAP: usize> Iterator for ReplyIter<'a, TL, Packet, CAP> {
type Item = (Packet, ReplyGuard<'a, TL, Packet, CAP>);
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())
origin.pump(self.app)
} else {
None
}
}
}
pub trait TokioTransport<Packet>: Clone {
fn time(&mut self) -> i64;
#[allow(unused)]
fn update_service_time(&mut self, timestamp: i64, current_time: i64) {}
pub trait TokioTransportLayer: Clone + Send + 'static {
type Packet;
fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Packet);
fn send_ack(&mut self, reply_no: SeqNo);
fn send(&self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &Self::Packet);
fn send_ack(&self, reply_no: SeqNo);
}
impl<Packet, Tl: TokioTransport<Packet>> TransportLayer<SendData<Packet>> for (Tl, ) {
impl<TL: TokioTransportLayer> TransportLayer<SendData<TL::Packet>> for &TokioTransport<TL> {
fn time(&mut self) -> i64 {
todo!()
self.time.elapsed().as_millis() as i64
}
fn update_service_time(&mut self, timestamp: i64, current_time: i64) {
fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, (_, payload): &SendData<TL::Packet>) {
self.app.send(seq_no, reply_no, payload);
}
fn send(&mut self, seq_no: SeqNo, reply_no: Option<SeqNo>, payload: &SendData<Packet>) {
todo!()
}
fn send_ack(&mut self, reply_no: SeqNo) {
todo!()
self.app.send_ack(reply_no)
}
}
//#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
//#[derive(Clone)]
//pub enum PacketType<Payload: Clone> {
src/transport_layer.rs
-2
View File
@@ -8,8 +8,6 @@ use crate::SeqNo;
/// they are most easily implemented as some combination of custom enums, `Vec<u8>` and `Arc<[u8]>`.
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: &SendData);
fn send_ack(&mut self, reply_no: SeqNo);