replace flume in iroh-net with async_channel

Rationale: see #2536

Author: rklaehn

iroh-net/Cargo.toml

@@ -17,13 +17,13 @@ workspace = true
 
 [dependencies]
 anyhow = { version = "1" }
+async-channel = "2.3.1"
 base64 = "0.22.1"
 backoff = "0.4.0"
 bytes = "1"
 netdev = "0.30.0"
 der = { version = "0.7", features = ["alloc", "derive"] }
 derive_more = { version = "1.0.0-beta.6", features = ["debug", "display", "from", "try_into", "deref"] }
-flume = "0.11"
 futures-buffered = "0.2.4"
 futures-concurrency = "7.6.0"
 futures-lite = "2.3"

iroh-net/src/magicsock.rs

@@ -177,7 +177,7 @@ pub(crate) struct MagicSock {
     proxy_url: Option<Url>,
 
     /// Used for receiving relay messages.
-    relay_recv_receiver: flume::Receiver<RelayRecvResult>,
+    relay_recv_receiver: async_channel::Receiver<RelayRecvResult>,
     /// Stores wakers, to be called when relay_recv_ch receives new data.
     network_recv_wakers: parking_lot::Mutex<Option<Waker>>,
     network_send_wakers: parking_lot::Mutex<Option<Waker>>,
@@ -786,11 +786,11 @@ impl MagicSock {
                 break;
             }
             match self.relay_recv_receiver.try_recv() {
-                Err(flume::TryRecvError::Empty) => {
+                Err(async_channel::TryRecvError::Empty) => {
                     self.network_recv_wakers.lock().replace(cx.waker().clone());
                     break;
                 }
-                Err(flume::TryRecvError::Disconnected) => {
+                Err(async_channel::TryRecvError::Closed) => {
                     return Poll::Ready(Err(io::Error::new(
                         io::ErrorKind::NotConnected,
                         "connection closed",
@@ -1375,7 +1375,7 @@ impl Handle {
             insecure_skip_relay_cert_verify,
         } = opts;
 
-        let (relay_recv_sender, relay_recv_receiver) = flume::bounded(128);
+        let (relay_recv_sender, relay_recv_receiver) = async_channel::bounded(128);
 
         let (pconn4, pconn6) = bind(port)?;
         let port = pconn4.port();
@@ -1701,7 +1701,7 @@ struct Actor {
     relay_actor_sender: mpsc::Sender<RelayActorMessage>,
     relay_actor_cancel_token: CancellationToken,
     /// Channel to send received relay messages on, for processing.
-    relay_recv_sender: flume::Sender<RelayRecvResult>,
+    relay_recv_sender: async_channel::Sender<RelayRecvResult>,
     /// When set, is an AfterFunc timer that will call MagicSock::do_periodic_stun.
     periodic_re_stun_timer: time::Interval,
     /// The `NetInfo` provided in the last call to `net_info_func`. It's used to deduplicate calls to netInfoFunc.
@@ -1855,7 +1855,7 @@ impl Actor {
                 let passthroughs = self.process_relay_read_result(read_result);
                 for passthrough in passthroughs {
                     self.relay_recv_sender
-                        .send_async(passthrough)
+                        .send(passthrough)
                         .await
                         .expect("missing recv sender");
                     let mut wakers = self.msock.network_recv_wakers.lock();

iroh-net/src/magicsock/udp_conn.rs

@@ -192,15 +192,15 @@ mod tests {
         let (m2, _m2_key) = wrap_socket(m2)?;
 
         let m1_addr = SocketAddr::new(network.local_addr(), m1.local_addr()?.port());
-        let (m1_send, m1_recv) = flume::bounded(8);
+        let (m1_send, m1_recv) = async_channel::bounded(8);
 
         let m1_task = tokio::task::spawn(async move {
             if let Some(conn) = m1.accept().await {
                 let conn = conn.await?;
                 let (mut send_bi, mut recv_bi) = conn.accept_bi().await?;
 
                 let val = recv_bi.read_to_end(usize::MAX).await?;
-                m1_send.send_async(val).await?;
+                m1_send.send(val).await?;
                 send_bi.finish().await?;
             }
 
@@ -220,7 +220,7 @@ mod tests {
         drop(send_bi);
 
         // make sure the right values arrived
-        let val = m1_recv.recv_async().await?;
+        let val = m1_recv.recv().await?;
         assert_eq!(val, b"hello");
 
         m1_task.await??;

iroh-net/src/net/netmon/actor.rs

@@ -57,7 +57,7 @@ pub(super) struct Actor {
     /// OS specific monitor.
     #[allow(dead_code)]
     route_monitor: RouteMonitor,
-    mon_receiver: flume::Receiver<NetworkMessage>,
+    mon_receiver: async_channel::Receiver<NetworkMessage>,
     actor_receiver: mpsc::Receiver<ActorMessage>,
     actor_sender: mpsc::Sender<ActorMessage>,
     /// Callback registry.
@@ -84,7 +84,7 @@ impl Actor {
         let wall_time = Instant::now();
 
         // Use flume channels, as tokio::mpsc is not safe to use across ffi boundaries.
-        let (mon_sender, mon_receiver) = flume::bounded(MON_CHAN_CAPACITY);
+        let (mon_sender, mon_receiver) = async_channel::bounded(MON_CHAN_CAPACITY);
         let route_monitor = RouteMonitor::new(mon_sender)?;
         let (actor_sender, actor_receiver) = mpsc::channel(ACTOR_CHAN_CAPACITY);
 
@@ -129,7 +129,7 @@ impl Actor {
                         debounce_interval.reset_immediately();
                     }
                 }
-                Ok(_event) = self.mon_receiver.recv_async() => {
+                Ok(_event) = self.mon_receiver.recv() => {
                     trace!("network activity detected");
                     last_event.replace(false);
                     debounce_interval.reset_immediately();

iroh-net/src/net/netmon/bsd.rs

@@ -23,7 +23,7 @@ impl Drop for RouteMonitor {
 }
 
 impl RouteMonitor {
-    pub(super) fn new(sender: flume::Sender<NetworkMessage>) -> Result<Self> {
+    pub(super) fn new(sender: async_channel::Sender<NetworkMessage>) -> Result<Self> {
         let socket = socket2::Socket::new(libc::AF_ROUTE.into(), socket2::Type::RAW, None)?;
         socket.set_nonblocking(true)?;
         let socket_std: std::os::unix::net::UnixStream = socket.into();
@@ -44,7 +44,7 @@ impl RouteMonitor {
                         ) {
                             Ok(msgs) => {
                                 if contains_interesting_message(&msgs) {
-                                    sender.send_async(NetworkMessage::Change).await.ok();
+                                    sender.send(NetworkMessage::Change).await.ok();
                                 }
                             }
                             Err(err) => {