Vendoring DelayQueue to enable customizations

Author: AhmedSoliman

Cargo.lock

@@ -22,10 +22,11 @@ futures = { workspace = true }
 hashbrown = { version = "0.16"  }
 pin-project = { workspace = true }
 rocksdb = { workspace = true }
+slab = { version = "0.4" }
 smallvec = { workspace = true }
 strum = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["sync"] }
 tokio-stream = { workspace = true }
-tokio-util = { workspace = true, features = ["time"] }
+tokio-util = { workspace = true }
 tracing = { workspace = true }

crates/vqueues/Cargo.toml

@@ -9,6 +9,7 @@
 // by the Apache License, Version 2.0.
 
 mod cache;
+mod delay_queue;
 pub mod scheduler;
 mod vqueue_config;
 

crates/vqueues/src/lib.rs

@@ -17,7 +17,6 @@ use std::task::Waker;
 use hashbrown::HashMap;
 use hashbrown::hash_map;
 use pin_project::pin_project;
-use tokio_util::time::DelayQueue;
 use tracing::{info, trace};
 
 use restate_futures_util::concurrency::Concurrency;
@@ -30,6 +29,7 @@ use restate_types::vqueue::VQueueId;
 use crate::EventDetails;
 use crate::VQueueEvent;
 use crate::VQueuesMeta;
+use crate::delay_queue::DelayQueue;
 use crate::scheduler::Assignments;
 use crate::scheduler::vqueue_state::Eligibility;
 

crates/vqueues/src/scheduler/drr.rs

@@ -11,7 +11,6 @@
 use std::time::Duration;
 
 use hashbrown::HashSet;
-use tokio_util::time::{DelayQueue, delay_queue};
 use tracing::trace;
 
 use restate_storage_api::StorageError;
@@ -21,6 +20,7 @@ use restate_types::clock::UniqueTimestamp;
 use restate_types::vqueue::VQueueId;
 
 use super::Assignments;
+use crate::delay_queue::{self, DelayQueue};
 use crate::scheduler::Action;
 use crate::vqueue_config::VQueueConfig;
 

crates/vqueues/src/scheduler/vqueue_state.rs

@@ -127,7 +127,7 @@ sync_wrapper = { version = "1", default-features = false, features = ["futures"]
 time = { version = "0.3", features = ["formatting", "local-offset", "macros", "parsing"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "ring", "tls12"] }
 tokio-stream = { version = "0.1", features = ["net", "sync"] }
-tokio-util = { version = "0.7", features = ["codec", "io-util", "net", "rt", "time"] }
+tokio-util = { version = "0.7", features = ["codec", "io-util", "net", "rt"] }
 toml_datetime = { version = "0.6", default-features = false, features = ["serde"] }
 toml_edit = { version = "0.22", features = ["serde"] }
 tonic = { version = "0.14", features = ["gzip", "tls-native-roots", "tls-ring", "zstd"] }
@@ -263,7 +263,7 @@ sync_wrapper = { version = "1", default-features = false, features = ["futures"]
 time = { version = "0.3", features = ["formatting", "local-offset", "macros", "parsing"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "ring", "tls12"] }
 tokio-stream = { version = "0.1", features = ["net", "sync"] }
-tokio-util = { version = "0.7", features = ["codec", "io-util", "net", "rt", "time"] }
+tokio-util = { version = "0.7", features = ["codec", "io-util", "net", "rt"] }
 toml_datetime = { version = "0.6", default-features = false, features = ["serde"] }
 toml_edit = { version = "0.22", features = ["serde"] }
 tonic = { version = "0.14", features = ["gzip", "tls-native-roots", "tls-ring", "zstd"] }