Merge pull request #75 from pipeless-ai/issue_70

fix: Remove managers and unassign pipelines properly

pipeless/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pipeless-ai"
-version = "1.0.1"
+version = "1.0.2"
 edition = "2021"
 authors = ["Miguel A. Cabrera Minagorri"]
 description = "An open-source computer vision framework to build and deploy applications in minutes"

pipeless/src/config/streams.rs

@@ -179,6 +179,9 @@ impl StreamsTable {
     pub fn find_by_pipeline_id(&self, pipeline_id: uuid::Uuid) -> Option<&StreamsTableEntry> {
         self.table.iter().find(|entry| entry.get_pipeline() == Some(pipeline_id))
     }
+    pub fn find_by_pipeline_id_mut(&mut self, pipeline_id: uuid::Uuid) -> Option<&mut StreamsTableEntry> {
+        self.table.iter_mut().find(|entry| entry.get_pipeline() == Some(pipeline_id))
+    }
 
     pub fn update_by_entry_id(&mut self, entry_id: uuid::Uuid, input_uri: &str, output_uri: Option<String>, frame_path: Vec<String>) {
         if let Some(entry) = self.table.iter_mut().find(|entry| entry.id == entry_id) {

pipeless/src/dispatcher.rs

@@ -155,24 +155,32 @@ pub fn start(
                             }
                         }
 
+                        // When we have a running manager whose pipeline id is not in any entry, that means the entry was deleted, stop the manager
+                        // and remove it from the hash_map to be dropped
+                        let mut manager_to_remove = None;
                         {
-                            // When we have a running manager whose pipeline id is not in any entry, that means the entry was deleted, stop the manager
                             let managers_map_guard = running_managers.read().await;
                             for (pipeline_id, manager) in managers_map_guard.iter() {
                                 let streams_table = streams_table.read().await;
                                 if streams_table.find_by_pipeline_id(*pipeline_id).is_none() {
                                     info!("Stream config entry removed. Stopping associated pipeline");
                                     manager.stop().await;
+                                    manager_to_remove = Some(pipeline_id.clone());
                                 }
                             }
                         }
+                        if let Some(manager) = manager_to_remove {
+                            let mut managers_map_guard = running_managers.write().await;
+                            managers_map_guard.remove(&manager);
+                        }
                     }
                     DispatcherEvent::PipelineFinished(pipeline_id) => {
-                        let mut stream_entry;
+                        let stream_entry;
                         {
-                            let table_read_guard = streams_table.read().await;
-                            let stream_entry_option = table_read_guard.find_by_pipeline_id(pipeline_id);
+                            let mut table_write_guard = streams_table.write().await;
+                            let stream_entry_option = table_write_guard.find_by_pipeline_id_mut(pipeline_id);
                             if let Some(entry) = stream_entry_option {
+                                entry.unassign_pipeline();
                                 stream_entry = entry.clone();
                             } else {
                                 warn!("
@@ -184,8 +192,6 @@ pub fn start(
                             }
                         }
 
-                        stream_entry.unassign_pipeline();
-
                         let using_input_file = stream_entry.get_input_uri().starts_with("file://");
                         let using_output_file = match stream_entry.get_output_uri() {
                             Some(uri) => uri.starts_with("file://"),