Revert "v1.46.1 with kanal instead of flume"

This reverts commit d2c1b36.

Author: Dav1dde

Cargo.lock

@@ -114,7 +114,6 @@ 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"
@@ -206,8 +205,7 @@ 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 = { version = "=1.46.1", default-features = false }
+tokio = { git = "https://github.com/dav1dde/tokio", rev = "b5d827e709e097fb5a9a391e2327880fce2a7759", 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 }
@@ -226,4 +224,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,7 +11,6 @@ publish = false
 
 [dependencies]
 flume = { workspace = true, features = ["async"] }
-kanal = { workspace = true }
 futures = { workspace = true }
 tokio = { workspace = true }
 pin-project-lite = { workspace = true }
@@ -25,4 +24,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: kanal::AsyncSender<F>,
+    tx: flume::Sender<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,34 +59,24 @@ where
         S: ThreadSpawn,
     {
         let pool_name = builder.pool_name.unwrap_or(DEFAULT_POOL_NAME);
-        let (tx, rx) = kanal::bounded_async(builder.num_threads * 2);
+        let (tx, rx) = flume::bounded(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 = {
-                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 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 thread = Thread {
                 id: thread_id,
@@ -120,7 +110,7 @@ where
     /// the pool panicked.
     pub fn spawn(&self, future: F) {
         assert!(
-            self.tx.as_sync().send(future).is_ok(),
+            self.tx.send(future).is_ok(),
             "failed to schedule task: all worker threads have terminated (either none were spawned or all have panicked)"
         );
     }
@@ -135,7 +125,7 @@ where
     /// the pool panicked.
     pub async fn spawn_async(&self, future: F) {
         assert!(
-            self.tx.send(future).await.is_ok(),
+            self.tx.send_async(future).await.is_ok(),
             "failed to schedule task: all worker threads have terminated (either none were spawned or all have panicked)"
         );
     }