Fix stream recreation crashes and thread starvation caused by MAVLink-related thread leak

src/main.rs

@@ -18,7 +18,7 @@ mod stream;
 mod video;
 mod video_stream;
 
-#[actix_web::main]
+#[tokio::main(flavor = "multi_thread", worker_threads = 10)]
 async fn main() -> Result<(), std::io::Error> {
     // CLI should be started before logger to allow control over verbosity
     cli::manager::init();
@@ -42,11 +42,13 @@ async fn main() -> Result<(), std::io::Error> {
         helper::develop::start_check_tasks_on_webrtc_reconnects();
     }
 
-    stream::webrtc::signalling_server::SignallingServer::default();
+    let _signalling_server = stream::webrtc::signalling_server::SignallingServer::default();
 
-    if let Err(error) = stream::manager::start_default() {
+    if let Err(error) = stream::manager::start_default().await {
         error!("Failed to start default streams. Reason: {error:?}")
     }
 
-    server::manager::run(&cli::manager::server_address()).await
+    server::manager::run(&cli::manager::server_address()).await?;
+
+    Ok(())
 }

src/mavlink/manager.rs

@@ -20,7 +20,7 @@ pub struct Manager {
 
 struct Connection {
     address: String,
-    connection: Box<dyn MavConnection<MavMessage> + Sync + Send>,
+    connection: Option<Box<dyn MavConnection<MavMessage> + Sync + Send>>,
     sender: broadcast::Sender<Message>,
 }
 
@@ -36,14 +36,12 @@ impl Default for Manager {
         let address =
             settings::manager::mavlink_endpoint().expect("No configured mavlink endpoint");
 
-        let connection = Connection::connect(&address);
-
         let (sender, _receiver) = broadcast::channel(100);
 
         let this = Self {
             connection: Arc::new(RwLock::new(Connection {
                 address,
-                connection,
+                connection: None,
                 sender,
             })),
             ids: Arc::new(RwLock::new(vec![])),
@@ -78,8 +76,15 @@ impl Manager {
             loop {
                 std::thread::sleep(std::time::Duration::from_millis(10));
 
+                let Ok(inner_guard) = inner.read() else {
+                    break; // Break to trigger reconnection
+                };
+                let Some(mavlink) = inner_guard.connection.as_deref() else {
+                    break; // Break to trigger reconnection
+                };
+
                 // Receive from the Mavlink network
-                let (header, message) = match inner.read().unwrap().connection.recv() {
+                let (header, message) = match mavlink.recv() {
                     Ok(message) => message,
                     Err(error) => {
                         trace!("Failed receiving from mavlink: {error:?}");
@@ -99,9 +104,7 @@ impl Manager {
                 trace!("Message received: {header:?}, {message:?}");
 
                 // Send the received message to the cameras
-                if let Err(error) = inner
-                    .read()
-                    .unwrap()
+                if let Err(error) = inner_guard
                     .sender
                     .send(Message::Received((header, message)))
                 {
@@ -113,7 +116,8 @@ impl Manager {
             // Reconnects
             {
                 let mut inner = inner.write().unwrap();
-                inner.connection = Connection::connect(&inner.address);
+                let address = inner.address.clone();
+                inner.connection.replace(Connection::connect(&address));
             }
 
             std::thread::sleep(std::time::Duration::from_millis(500));
@@ -141,20 +145,28 @@ impl Manager {
                     _ => continue,
                 };
 
+                let Ok(inner_guard) = inner.read() else {
+                    break; // Break to trigger reconnection
+                };
+                let Some(mavlink) = inner_guard.connection.as_deref() else {
+                    break; // Break to trigger reconnection
+                };
+
                 // Send the response from the cameras to the Mavlink network
-                if let Err(error) = inner.read().unwrap().connection.send(&header, &message) {
+                if let Err(error) = mavlink.send(&header, &message) {
                     error!("Failed sending message to Mavlink Connection: {error:?}");
 
                     break; // Break to trigger reconnection
                 }
 
-                trace!("Message sent: {header:?}, {message:?}");
+                debug!("Message sent: {header:?}, {message:?}");
             }
 
             // Reconnects
             {
                 let mut inner = inner.write().unwrap();
-                inner.connection = Connection::connect(&inner.address);
+                let address = inner.address.clone();
+                inner.connection.replace(Connection::connect(&address));
             }
 
             std::thread::sleep(std::time::Duration::from_millis(500));
@@ -213,7 +225,7 @@ impl Connection {
                     return connection;
                 }
                 Err(error) => {
-                    error!("Failed to connect, trying again in one second. Reason: {error:#?}.");
+                    error!("Failed to connect, trying again in one second. Reason: {error:?}.");
                 }
             }
         }

src/mavlink/mavlink_camera.rs

@@ -16,58 +16,61 @@ use super::manager::Message;
 use super::utils::*;
 
 #[derive(Debug)]
-pub struct MavlinkCameraHandle {
-    inner: Arc<MavlinkCamera>,
-    _runtime: tokio::runtime::Runtime,
-    heartbeat_handle: tokio::task::JoinHandle<()>,
-    messages_handle: tokio::task::JoinHandle<()>,
+pub struct MavlinkCamera {
+    inner: Arc<MavlinkCameraInner>,
+    heartbeat_handle: Option<tokio::task::JoinHandle<()>>,
+    messages_handle: Option<tokio::task::JoinHandle<()>>,
 }
 
 #[derive(Debug, Clone)]
-struct MavlinkCamera {
+struct MavlinkCameraInner {
     component: MavlinkCameraComponent,
     mavlink_stream_type: mavlink::common::VideoStreamType,
     video_stream_uri: Url,
     video_stream_name: String,
     video_source_type: VideoSourceType,
 }
 
-impl MavlinkCameraHandle {
+impl MavlinkCamera {
     #[instrument(level = "debug")]
-    pub fn try_new(video_and_stream_information: &VideoAndStreamInformation) -> Result<Self> {
-        let inner = Arc::new(MavlinkCamera::try_new(video_and_stream_information)?);
+    pub async fn try_new(video_and_stream_information: &VideoAndStreamInformation) -> Result<Self> {
+        let inner = Arc::new(MavlinkCameraInner::try_new(video_and_stream_information)?);
 
         let sender = crate::mavlink::manager::Manager::get_sender();
 
-        let runtime = tokio::runtime::Builder::new_multi_thread()
-            .on_thread_start(|| debug!("Thread started"))
-            .on_thread_stop(|| debug!("Thread stopped"))
-            .thread_name_fn(|| {
-                static ATOMIC_ID: std::sync::atomic::AtomicUsize =
-                    std::sync::atomic::AtomicUsize::new(0);
-                let id = ATOMIC_ID.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-                format!("MavlinkCamera-{id}")
-            })
-            .worker_threads(2)
-            .enable_all()
-            .build()
-            .expect("Failed building a new tokio runtime");
-
-        let heartbeat_handle =
-            runtime.spawn(MavlinkCamera::heartbeat_loop(inner.clone(), sender.clone()));
-        let messages_handle =
-            runtime.spawn(MavlinkCamera::messages_loop(inner.clone(), sender.clone()));
+        debug!("Starting MAVLink HeartBeat task...");
+
+        let inner_cloned = inner.clone();
+        let sender_cloned = sender.clone();
+        let heartbeat_handle = Some(tokio::spawn(async move {
+            debug!("MAVLink HeartBeat task started!");
+            match MavlinkCameraInner::heartbeat_loop(inner_cloned, sender_cloned).await {
+                Ok(_) => debug!("MAVLink HeartBeat task eneded with no errors"),
+                Err(error) => warn!("MAVLink HeartBeat task ended with error: {error:#?}"),
+            };
+        }));
+
+        debug!("Starting MAVLink Message task...");
+
+        let inner_cloned = inner.clone();
+        let sender_cloned = sender.clone();
+        let messages_handle = Some(tokio::spawn(async move {
+            debug!("MAVLink Message task started!");
+            match MavlinkCameraInner::messages_loop(inner_cloned, sender_cloned).await {
+                Ok(_) => debug!("MAVLink Message task eneded with no errors"),
+                Err(error) => warn!("MAVLink Message task ended with error: {error:#?}"),
+            };
+        }));
 
         Ok(Self {
             inner,
-            _runtime: runtime,
             heartbeat_handle,
             messages_handle,
         })
     }
 }
 
-impl MavlinkCamera {
+impl MavlinkCameraInner {
     #[instrument(level = "debug")]
     pub fn try_new(video_and_stream_information: &VideoAndStreamInformation) -> Result<Self> {
         let video_stream_uri = video_and_stream_information
@@ -128,7 +131,10 @@ impl MavlinkCamera {
 
     #[instrument(level = "trace", skip(sender))]
     #[instrument(level = "debug", skip_all, fields(component_id = camera.component.component_id))]
-    pub async fn heartbeat_loop(camera: Arc<MavlinkCamera>, sender: broadcast::Sender<Message>) {
+    pub async fn heartbeat_loop(
+        camera: Arc<MavlinkCameraInner>,
+        sender: broadcast::Sender<Message>,
+    ) -> Result<()> {
         let component_id = camera.component.component_id;
         let system_id = camera.component.system_id;
 
@@ -154,42 +160,33 @@ impl MavlinkCamera {
                 error!("Failed to send message: {error:?}");
                 continue;
             }
-
-            debug!("Heartbeat sent");
         }
     }
 
     #[instrument(level = "trace", skip(sender))]
     #[instrument(level = "debug", skip_all, fields(component_id = camera.component.component_id))]
-    pub async fn messages_loop(camera: Arc<MavlinkCamera>, sender: broadcast::Sender<Message>) {
+    pub async fn messages_loop(
+        camera: Arc<MavlinkCameraInner>,
+        sender: broadcast::Sender<Message>,
+    ) -> Result<()> {
         let mut receiver = sender.subscribe();
+        use crate::mavlink::mavlink_camera::Message::Received;
 
         loop {
-            let (header, message) = match receiver.recv().await {
-                Ok(Message::Received(message)) => message,
-                Err(broadcast::error::RecvError::Closed) => {
-                    unreachable!(
-                        "Closed channel: This should never happen, this channel is static!"
-                    );
-                }
-                Ok(Message::ToBeSent(_)) | Err(broadcast::error::RecvError::Lagged(_)) => continue,
-            };
+            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
 
-            trace!("Message received: {header:?}, {message:?}");
+            if let Ok(Received((header, message))) = receiver.recv().await {
+                trace!("Message received: {header:?}, {message:?}");
 
-            tokio::spawn(Self::handle_message(
-                camera.clone(),
-                sender.clone(),
-                header,
-                message,
-            ));
+                Self::handle_message(camera.clone(), sender.clone(), header, message).await;
+            }
         }
     }
 
     #[instrument(level = "trace", skip(sender))]
     #[instrument(level = "debug", skip(sender, camera), fields(component_id = camera.component.component_id))]
     async fn handle_message(
-        camera: Arc<MavlinkCamera>,
+        camera: Arc<MavlinkCameraInner>,
         sender: broadcast::Sender<Message>,
         header: MavHeader,
         message: MavMessage,
@@ -225,7 +222,7 @@ impl MavlinkCamera {
     #[instrument(level = "trace", skip(sender))]
     #[instrument(level = "debug", skip(sender, camera), fields(component_id = camera.component.component_id))]
     async fn handle_command_long(
-        camera: &MavlinkCamera,
+        camera: &MavlinkCameraInner,
         sender: broadcast::Sender<Message>,
         their_header: &MavHeader,
         data: &mavlink::common::COMMAND_LONG_DATA,
@@ -449,7 +446,7 @@ impl MavlinkCamera {
     #[instrument(level = "trace", skip(sender))]
     #[instrument(level = "debug", skip(sender, camera), fields(component_id = camera.component.component_id))]
     async fn handle_param_ext_set(
-        camera: &MavlinkCamera,
+        camera: &MavlinkCameraInner,
         sender: broadcast::Sender<Message>,
         header: &MavHeader,
         data: &mavlink::common::PARAM_EXT_SET_DATA,
@@ -510,7 +507,7 @@ impl MavlinkCamera {
     #[instrument(level = "trace", skip(sender))]
     #[instrument(level = "debug", skip(sender, camera), fields(component_id = camera.component.component_id))]
     async fn handle_param_ext_request_read(
-        camera: &MavlinkCamera,
+        camera: &MavlinkCameraInner,
         sender: broadcast::Sender<Message>,
         header: &MavHeader,
         data: &mavlink::common::PARAM_EXT_REQUEST_READ_DATA,
@@ -561,7 +558,7 @@ impl MavlinkCamera {
     #[instrument(level = "trace", skip(sender))]
     #[instrument(level = "debug", skip(sender, camera), fields(component_id = camera.component.component_id))]
     async fn handle_param_ext_request_list(
-        camera: &MavlinkCamera,
+        camera: &MavlinkCameraInner,
         sender: broadcast::Sender<Message>,
         header: &MavHeader,
         data: &mavlink::common::PARAM_EXT_REQUEST_LIST_DATA,
@@ -611,10 +608,37 @@ impl MavlinkCamera {
     }
 }
 
-impl Drop for MavlinkCameraHandle {
+impl Drop for MavlinkCamera {
+    #[instrument(level = "debug", skip(self))]
     fn drop(&mut self) {
-        self.heartbeat_handle.abort();
-        self.messages_handle.abort();
-        super::manager::Manager::drop_id(self.inner.component.component_id)
+        debug!("Dropping MavlinkCameraHandle...");
+
+        if let Some(handle) = self.heartbeat_handle.take() {
+            if !handle.is_finished() {
+                handle.abort();
+                tokio::spawn(async move {
+                    let _ = handle.await;
+                    debug!("Mavlink Heartbeat task aborted");
+                });
+            } else {
+                debug!("Mavlink Heartbeat task nicely finished!");
+            }
+        }
+
+        if let Some(handle) = self.messages_handle.take() {
+            if !handle.is_finished() {
+                handle.abort();
+                tokio::spawn(async move {
+                    let _ = handle.await;
+                    debug!("Mavlink Message task aborted");
+                });
+            } else {
+                debug!("Mavlink Message task nicely finished!");
+            }
+        }
+
+        super::manager::Manager::drop_id(self.inner.component.component_id);
+
+        debug!("MavlinkCameraHandle Dropped!");
     }
 }

src/server/pages.rs

@@ -232,7 +232,7 @@ pub fn v4l_post(json: web::Json<V4lControl>) -> HttpResponse {
 pub async fn reset_settings(query: web::Query<ResetSettings>) -> HttpResponse {
     if query.all.unwrap_or_default() {
         settings::manager::reset();
-        if let Err(error) = stream_manager::start_default() {
+        if let Err(error) = stream_manager::start_default().await {
             return HttpResponse::InternalServerError()
                 .content_type("text/plain")
                 .body(format!("{error:#?}"));
@@ -269,7 +269,7 @@ pub async fn streams() -> HttpResponse {
 
 #[api_v2_operation]
 /// Create a video stream
-pub fn streams_post(json: web::Json<PostStream>) -> HttpResponse {
+pub async fn streams_post(json: web::Json<PostStream>) -> HttpResponse {
     let json = json.into_inner();
 
     let video_source = match video_source::get_video_source(&json.source) {
@@ -285,7 +285,9 @@ pub fn streams_post(json: web::Json<PostStream>) -> HttpResponse {
         name: json.name,
         stream_information: json.stream_information,
         video_source,
-    }) {
+    })
+    .await
+    {
         return HttpResponse::NotAcceptable()
             .content_type("text/plain")
             .body(format!("{error:#?}"));