refactor: use tokio mpsc for worker errors

src/main.rs

@@ -32,7 +32,6 @@ use crate::reflector_selector::ReflectorSelector;
 use crate::settings::{ReflectorDefinition, ReflectorProtocol, Settings};
 use crate::util::RwLockExt;
 use ::log::{info, warn};
-use flume::RecvTimeoutError;
 use std::collections::HashMap;
 use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
@@ -40,6 +39,7 @@ use std::sync::{Arc, RwLock};
 use std::thread;
 use std::thread::{JoinHandle, sleep};
 use std::time::{Duration, Instant};
+use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use tokio::sync::watch;
 
 struct ReflectorSetup {
@@ -267,7 +267,7 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
 
 fn spawn_task<F, E>(
     rt: &tokio::runtime::Runtime,
-    error_tx: &flume::Sender<anyhow::Error>,
+    error_tx: &UnboundedSender<anyhow::Error>,
     task: F,
 ) -> tokio::task::JoinHandle<()>
 where
@@ -285,7 +285,7 @@ where
 
 fn spawn_worker<F>(
     name: &str,
-    error_tx: &flume::Sender<anyhow::Error>,
+    error_tx: &UnboundedSender<anyhow::Error>,
     worker: F,
 ) -> anyhow::Result<JoinHandle<()>>
 where
@@ -303,16 +303,16 @@ where
         .map_err(|e| anyhow::anyhow!(e))
 }
 
-fn wait_for_exit(error_rx: &flume::Receiver<anyhow::Error>) -> anyhow::Result<()> {
+async fn wait_for_exit(error_rx: &mut UnboundedReceiver<anyhow::Error>) -> anyhow::Result<()> {
     loop {
-        match error_rx.recv_timeout(Duration::from_secs(1)) {
+        match tokio::time::timeout(Duration::from_secs(1), error_rx.recv()).await {
-            Ok(error) => {
+            Ok(Some(error)) => {
                 return Err(anyhow::anyhow!("worker exited with error: {error}"));
             }
-            Err(RecvTimeoutError::Disconnected) => {
+            Ok(None) => {
                 return Ok(());
             }
-            Err(RecvTimeoutError::Timeout) => {
+            Err(_) => {
                 if SHUTDOWN.load(Ordering::Relaxed) {
                     info!("Received shutdown signal");
                     return Ok(());
@@ -347,7 +347,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     } = setup_reflectors(settings)?;
 
     let (control_snapshot_tx, control_snapshot_rx) = watch::channel(None);
-    let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
+    let (error_tx, mut error_rx) = tokio::sync::mpsc::unbounded_channel::<anyhow::Error>();
     let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded();
     #[cfg(feature = "irtt")]
     let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded();
@@ -495,7 +495,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     // Drop the original sender so the channel disconnects if all workers exit cleanly.
     drop(error_tx);
 
-    let result = wait_for_exit(&error_rx);
+    let result = rt.block_on(wait_for_exit(&mut error_rx));
 
     // Make the error path request shutdown too, rather than leaving the other workers running.
     SHUTDOWN.store(true, Ordering::Relaxed);