Fix turn relay client fd leak (#514)

* fix: missing close relay conn when happen error

* fix: make sure that the listening loop must exit when the client closes

Author: honsunrise

Cargo.lock

@@ -820,8 +820,8 @@ impl Agent {
                     return Ok(());
                 }
 
-                let relay_conn = match client.allocate().await {
-                    Ok(conn) => conn,
+                let relay_conn: Arc<dyn Conn + Send + Sync> = match client.allocate().await {
+                    Ok(conn) => Arc::new(conn),
                     Err(err) => {
                         let _ = client.close().await;
                         log::warn!(
@@ -841,7 +841,7 @@ impl Agent {
                         address: raddr.ip().to_string(),
                         port: raddr.port(),
                         component: COMPONENT_RTP,
-                        conn: Some(Arc::new(relay_conn)),
+                        conn: Some(Arc::clone(&relay_conn)),
                         ..CandidateBaseConfig::default()
                     },
                     rel_addr,
@@ -853,6 +853,7 @@ impl Agent {
                     match relay_config.new_candidate_relay() {
                         Ok(candidate) => Arc::new(candidate),
                         Err(err) => {
+                            let _ = relay_conn.close().await;
                             let _ = client.close().await;
                             log::warn!(
                                 "[{}]: Failed to create relay candidate: {} {}: {}",

ice/src/agent/agent_gather.rs

@@ -14,6 +14,7 @@ util = { version = "0.8", path = "../util", package = "webrtc-util", default-fea
 stun = { version = "0.5", path = "../stun" }
 
 tokio = { version = "1.32.0", features = ["full"] }
+tokio-util = "0.7"
 futures = "0.3"
 async-trait = "0.1"
 log = "0.4"

turn/Cargo.toml

@@ -24,7 +24,10 @@ use stun::integrity::*;
 use stun::message::*;
 use stun::textattrs::*;
 use stun::xoraddr::*;
+use tokio::pin;
+use tokio::select;
 use tokio::sync::{mpsc, Mutex};
+use tokio_util::sync::CancellationToken;
 use transaction::*;
 use util::conn::*;
 use util::vnet::net::*;
@@ -78,6 +81,7 @@ struct ClientInternal {
     binding_mgr: Arc<Mutex<BindingManager>>,
     rto_in_ms: u16,
     read_ch_tx: Arc<Mutex<Option<mpsc::Sender<InboundData>>>>,
+    close_notify: CancellationToken,
 }
 
 #[async_trait]
@@ -210,6 +214,7 @@ impl ClientInternal {
             },
             integrity: MessageIntegrity::new_short_term_integrity(String::new()),
             read_ch_tx: Arc::new(Mutex::new(None)),
+            close_notify: CancellationToken::new(),
         })
     }
 
@@ -227,33 +232,51 @@ impl ClientInternal {
         let tr_map = Arc::clone(&self.tr_map);
         let read_ch_tx = Arc::clone(&self.read_ch_tx);
         let binding_mgr = Arc::clone(&self.binding_mgr);
+        let close_notify = self.close_notify.clone();
 
         tokio::spawn(async move {
             let mut buf = vec![0u8; MAX_DATA_BUFFER_SIZE];
+            let wait_cancel = close_notify.cancelled();
+            pin!(wait_cancel);
+
             loop {
-                //TODO: gracefully exit loop
-                let (n, from) = match conn.recv_from(&mut buf).await {
-                    Ok((n, from)) => (n, from),
-                    Err(err) => {
-                        log::debug!("exiting read loop: {}", err);
+                let (n, from) = select! {
+                    biased;
+
+                    _ = &mut wait_cancel => {
+                        log::debug!("exiting read loop");
                         break;
+                    },
+                    result = conn.recv_from(&mut buf) => match result {
+                        Ok((n, from)) => (n, from),
+                        Err(err) => {
+                            log::debug!("exiting read loop: {}", err);
+                            break;
+                        }
                     }
                 };
-
                 log::debug!("received {} bytes of udp from {}", n, from);
 
-                if let Err(err) = ClientInternal::handle_inbound(
-                    &read_ch_tx,
-                    &buf[..n],
-                    from,
-                    &stun_serv_str,
-                    &tr_map,
-                    &binding_mgr,
-                )
-                .await
-                {
-                    log::debug!("exiting read loop: {}", err);
-                    break;
+                select! {
+                    biased;
+
+                    _ = &mut wait_cancel => {
+                        log::debug!("exiting read loop");
+                        break;
+                    },
+                    result = ClientInternal::handle_inbound(
+                        &read_ch_tx,
+                        &buf[..n],
+                        from,
+                        &stun_serv_str,
+                        &tr_map,
+                        &binding_mgr,
+                    ) => {
+                        if let Err(err) = result {
+                            log::debug!("exiting read loop: {}", err);
+                            break;
+                        }
+                    }
                 }
             }
         });
@@ -430,6 +453,7 @@ impl ClientInternal {
 
     /// Closes this client.
     async fn close(&mut self) {
+        self.close_notify.cancel();
         {
             let mut read_ch_tx = self.read_ch_tx.lock().await;
             read_ch_tx.take();