Portable delay queue

quinn-rs · Mar 4, 2023 · f6c21e2 · f6c21e2
1 parent aa92ba9
commit f6c21e2
Show file tree

Hide file tree

Showing 5 changed files with 642 additions and 19 deletions.
diff --git a/quinn/Cargo.toml b/quinn/Cargo.toml
@@ -42,10 +42,10 @@ rustc-hash = "1.1"
 pin-project-lite = "0.2"
 proto = { package = "quinn-proto", path = "../quinn-proto", version = "0.9", default-features = false }
 rustls = { version = "0.20.3", default-features = false, features = ["quic"], optional = true }
+slab = "0.4"
 thiserror = "1.0.21"
 tracing = "0.1.10"
 tokio = { version = "1.13.0", features = ["sync"] }
-tokio-util = { version = "0.6.9", features = ["time"] }
 udp = { package = "quinn-udp", path = "../quinn-udp", version = "0.3", default-features = false }
 webpki = { version = "0.22", default-features = false, optional = true }
 
@@ -54,6 +54,7 @@ anyhow = "1.0.22"
 crc = "3"
 bencher = "0.1.5"
 directories-next = "2"
+proptest = "1"
 rand = "0.8"
 rcgen = "0.10.0"
 rustls-pemfile = "1.0.0"

diff --git a/quinn/src/connection.rs b/quinn/src/connection.rs
@@ -15,14 +15,11 @@ use pin_project_lite::pin_project;
 use proto::{ConnectionError, ConnectionHandle, ConnectionStats, Dir, StreamEvent, StreamId};
 use rustc_hash::FxHashMap;
 use thiserror::Error;
-use tokio::{
-    sync::{futures::Notified, mpsc, oneshot, Notify},
-    time::Instant as TokioInstant,
-};
-use tokio_util::time::delay_queue;
+use tokio::sync::{futures::Notified, mpsc, oneshot, Notify};
 use tracing::debug_span;
 
 use crate::{
+    delay_queue::Timer,
     mutex::Mutex,
     recv_stream::RecvStream,
     send_stream::{SendStream, WriteError},
@@ -784,8 +781,8 @@ pub(crate) struct State {
     on_handshake_data: Option<oneshot::Sender<()>>,
     on_connected: Option<oneshot::Sender<bool>>,
     connected: bool,
-    pub(crate) timer_handle: Option<delay_queue::Key>,
-    pub(crate) timer_deadline: Option<TokioInstant>,
+    pub(crate) timer_handle: Option<Timer>,
+    pub(crate) timer_deadline: Option<Instant>,
     pub(crate) blocked_writers: FxHashMap<StreamId, Waker>,
     pub(crate) blocked_readers: FxHashMap<StreamId, Waker>,
     pub(crate) finishing: FxHashMap<StreamId, oneshot::Sender<Option<WriteError>>>,