v1.46.1 with kanal instead of flume

Author: Dav1dde

Cargo.lock

@@ -114,6 +114,7 @@ ed25519-dalek = { version = "2.2.0", features = ["digest"] }
 enumset = "1.1.10"
 flate2 = "1.1.4"
 flume = { version = "0.11.1", default-features = false }
+kanal = "0.1.1"
 futures = { version = "0.3.31", default-features = false, features = ["std"] }
 globset = "0.4.16"
 hash32 = "1.0.0"
@@ -205,7 +206,8 @@ sysinfo = "0.35.1"
 tempfile = "3.23.0"
 thiserror = "2.0.17"
 tikv-jemallocator = "0.6.0"
-tokio = { git = "https://github.com/dav1dde/tokio", rev = "b5d827e709e097fb5a9a391e2327880fce2a7759", default-features = false }
+# tokio = { git = "https://github.com/dav1dde/tokio", rev = "b5d827e709e097fb5a9a391e2327880fce2a7759", default-features = false }
+tokio = { version = "=1.46.1", default-features = false }
 tokio-util = { version = "0.7.13", default-features = false }
 tower = { version = "0.5.2", default-features = false }
 tower-http = { version = "0.6.6", default-features = false }
@@ -224,4 +226,4 @@ zstd = { version = "0.13.3", features = ["experimental"] }
 # This prevents stack overflows in span description normalization.
 [patch.crates-io]
 sqlparser_derive = { git = "https://github.com/apache/datafusion-sqlparser-rs", rev = "ade40826563451cc14c130af9d689f4050dbdb15" }
-tokio = { git = "https://github.com/dav1dde/tokio", rev = "b5d827e709e097fb5a9a391e2327880fce2a7759" }
+# tokio = { git = "https://github.com/dav1dde/tokio", rev = "b5d827e709e097fb5a9a391e2327880fce2a7759" }

Cargo.toml

@@ -11,6 +11,7 @@ publish = false
 
 [dependencies]
 flume = { workspace = true, features = ["async"] }
+kanal = { workspace = true }
 futures = { workspace = true }
 tokio = { workspace = true }
 pin-project-lite = { workspace = true }
@@ -24,4 +25,4 @@ tokio = { workspace = true, features = ["rt", "rt-multi-thread", "time", "sync",
 
 [[bench]]
 name = "pool"
-harness = false
+harness = false

relay-threading/Cargo.toml

@@ -23,7 +23,7 @@ pub struct AsyncPool<F> {
     /// Name of the pool.
     name: &'static str,
     /// Transmission containing all tasks.
-    tx: flume::Sender<F>,
+    tx: kanal::AsyncSender<F>,
     /// The maximum number of tasks that are expected to run concurrently at any point in time.
     max_tasks: u64,
     /// Vector containing all the metrics collected individually in each thread.
@@ -59,24 +59,34 @@ where
         S: ThreadSpawn,
     {
         let pool_name = builder.pool_name.unwrap_or(DEFAULT_POOL_NAME);
-        let (tx, rx) = flume::bounded(builder.num_threads * 2);
+        let (tx, rx) = kanal::bounded_async(builder.num_threads * 2);
         let mut threads_metrics = Vec::with_capacity(builder.num_threads);
 
         for thread_id in 0..builder.num_threads {
             let rx = rx.clone();
 
             let thread_name: Option<String> = builder.thread_name.as_mut().map(|f| f(thread_id));
             let metrics = Arc::new(ThreadMetrics::default());
-            let task = MonitoredFuture::wrap_with_metrics(
-                Multiplexed::new(
-                    pool_name,
-                    builder.max_concurrency,
-                    rx.into_stream(),
-                    builder.task_panic_handler.clone(),
-                    metrics.clone(),
-                ),
-                metrics.raw_metrics.clone(),
-            );
+
+            let task = {
+                let metrics = Arc::clone(&metrics);
+                let task_panic_handler = builder.task_panic_handler.clone();
+                async move {
+                    let stream = rx.stream();
+
+                    MonitoredFuture::wrap_with_metrics(
+                        Multiplexed::new(
+                            pool_name,
+                            builder.max_concurrency,
+                            stream,
+                            task_panic_handler,
+                            metrics.clone(),
+                        ),
+                        metrics.raw_metrics.clone(),
+                    )
+                    .await
+                }
+            };
 
             let thread = Thread {
                 id: thread_id,
@@ -110,7 +120,7 @@ where
     /// the pool panicked.
     pub fn spawn(&self, future: F) {
         assert!(
-            self.tx.send(future).is_ok(),
+            self.tx.as_sync().send(future).is_ok(),
             "failed to schedule task: all worker threads have terminated (either none were spawned or all have panicked)"
         );
     }
@@ -125,7 +135,7 @@ where
     /// the pool panicked.
     pub async fn spawn_async(&self, future: F) {
         assert!(
-            self.tx.send_async(future).await.is_ok(),
+            self.tx.send(future).await.is_ok(),
             "failed to schedule task: all worker threads have terminated (either none were spawned or all have panicked)"
         );
     }