From 604a9103a7e75590104fdf59d7f2499a8ff92f59 Mon Sep 17 00:00:00 2001 From: Monica Moniot Date: Thu, 17 Aug 2023 15:11:07 -0400 Subject: [PATCH] cleared errors --- Cargo.lock | 12 ++++ Cargo.toml | 2 +- src/seq_queue.rs | 3 - src/tokio.rs | 157 +++++++++++++++++++++++++---------------- src/transport_layer.rs | 2 - 5 files changed, 110 insertions(+), 66 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ac844df..aa88eb2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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]] diff --git a/Cargo.toml b/Cargo.toml index 89688df..9c67fab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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"]} diff --git a/src/seq_queue.rs b/src/seq_queue.rs index ec57d96..01ca110 100644 --- a/src/seq_queue.rs +++ b/src/seq_queue.rs @@ -183,7 +183,6 @@ impl SeqEx { 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 SeqEx { 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 SeqEx { 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 diff --git a/src/tokio.rs b/src/tokio.rs index 6303bfe..814eac7 100644 --- a/src/tokio.rs +++ b/src/tokio.rs @@ -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 { send_block: Notify, } -pub struct ReplyGuard<'a, TL: TransportLayer>, Packet, const CAP: usize = DEFAULT_WINDOW_CAP> { +pub struct ReplyGuard<'a, TL: TokioTransportLayer, Packet, const CAP: usize = DEFAULT_WINDOW_CAP> { origin: &'a SeqExTokio, - app: Option, + app: Option<&'a TokioTransport>, reply_no: SeqNo, } -impl<'a, TL: TransportLayer>, Packet, const CAP: usize> ReplyGuard<'a, TL, Packet, CAP> { +impl<'a, TL: TokioTransportLayer, 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>, 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>, Packet, const CAP: usize> Drop for ReplyGuard<'a, TL, Packet, CAP> { +impl<'a, TL: TokioTransportLayer, 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>, Packet, const CAP: usize> Drop fo } } -pub struct ReplyIter<'a, TL: TransportLayer>, Packet, const CAP: usize = DEFAULT_WINDOW_CAP> { +pub struct ReplyIter<'a, TL: TokioTransportLayer, Packet, const CAP: usize = DEFAULT_WINDOW_CAP> { origin: Option<&'a SeqExTokio>, - app: TL, + app: &'a TokioTransport, first: Option<(Packet, ReplyGuard<'a, TL, Packet, CAP>)>, } @@ -69,6 +73,51 @@ pub struct ReplyIter<'a, TL: TransportLayer>, Packet, const CAP // &mut self.0 .0 // } //} +#[derive(Clone)] +pub struct TokioTransport { + time: time::Instant, + update_queue: mpsc::Sender, + app: TL, +} +impl TokioTransport { + pub fn new> + 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 SeqExTokio { pub fn new(retry_interval: i64, initial_seq_no: SeqNo) -> Self { @@ -78,8 +127,9 @@ impl SeqExTokio { } } - fn process>>( - &self, app: TL, + fn process<'a, TL: TokioTransportLayer>( + &'a self, + app: &'a TokioTransport, mut seq: MutexGuard<'_, (SeqEx, Packet, CAP>, usize)>, result: Result<(SeqNo, Packet, Option>), Error> ) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { @@ -88,7 +138,7 @@ impl SeqExTokio { 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 SeqExTokio { None } } - pub fn receive>>( - &self, - app: TL, + pub fn receive<'a, TL: TokioTransportLayer>( + &'a self, + app: &'a TokioTransport, seq_no: SeqNo, reply_no: Option, 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>>(&self, app: TL) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { + pub fn pump<'a, TL: TokioTransportLayer>(&'a self, app: &'a TokioTransport) -> 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>>( - &self, - app: TL, + pub fn receive_all<'a, TL: TokioTransportLayer>( + &'a self, + app: &'a TokioTransport, seq_no: SeqNo, reply_no: Option, 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 SeqExTokio { // let mut seq = self.lock(); // seq.try_send(app, packet_data) //} - async fn send_inner>>( + async fn send_inner>( &self, mut seq: MutexGuard<'_, (SeqEx, Packet, CAP>, usize)>, - app: TL, + app: &TokioTransport, 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>>(&self, app: TL, packet: Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { + pub async fn send<'a, TL: TokioTransportLayer>(&'a self, app: &'a TokioTransport, packet: Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { self.send_with(app, |_| packet).await } //pub fn try_send_with>(&self, app: TL, packet_data: impl FnOnce(SeqNo) -> SendData) -> Result<(), SendData> { @@ -167,25 +222,15 @@ impl SeqExTokio { // let seq_no = seq.seq_no(); // seq.try_send(app, packet_data(seq_no)) //} - pub async fn send_with>>(&self, app: TL, packet: impl FnOnce(SeqNo) -> Packet) -> Option<(Packet, ReplyGuard<'_, TL, Packet, CAP>)> { + pub async fn send_with<'a, TL: TokioTransportLayer>(&'a self, app: &'a TokioTransport, 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>>(&self, app: TL) { - let a = task::spawn(async{ - - }); - } - pub fn service>>(&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 SeqExTokio { // } //} -impl<'a, TL: TransportLayer>, Packet, const CAP: usize> Iterator for ReplyIter<'a, TL, Packet, CAP> { +impl<'a, TL: TokioTransportLayer, Packet, const CAP: usize> Iterator for ReplyIter<'a, TL, Packet, CAP> { type Item = (Packet, ReplyGuard<'a, TL, Packet, CAP>); fn next(&mut self) -> Option { 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: 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, payload: &Packet); - fn send_ack(&mut self, reply_no: SeqNo); + fn send(&self, seq_no: SeqNo, reply_no: Option, payload: &Self::Packet); + fn send_ack(&self, reply_no: SeqNo); } -impl> TransportLayer> for (Tl, ) { +impl TransportLayer> for &TokioTransport { 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, (_, payload): &SendData) { + self.app.send(seq_no, reply_no, payload); } - - fn send(&mut self, seq_no: SeqNo, reply_no: Option, payload: &SendData) { - 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 { diff --git a/src/transport_layer.rs b/src/transport_layer.rs index 9d7d9d0..53bc10d 100644 --- a/src/transport_layer.rs +++ b/src/transport_layer.rs @@ -8,8 +8,6 @@ use crate::SeqNo; /// they are most easily implemented as some combination of custom enums, `Vec` and `Arc<[u8]>`. pub trait TransportLayer: 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, payload: &SendData); fn send_ack(&mut self, reply_no: SeqNo);