fix(#830): fix model delete with tasks

Fixed the delete of models, which have active tasks
running. The problem was a dead-lock.

Signed-off-by: Tobias Anker <tobias.anker@kitsunemimi.moe>

Author: kitsudaiki

CHANGELOG.md

@@ -52,6 +52,7 @@
 - fixed number of output-values in case of int and float output-type
 - task-abort-endpoint not works
 - task-abort-modal was fixed in the dashboard
+- fixed delete model with running tasks
 
 ## v0.10.0
 

src/binaries/hanami/src/api/http_endpoints/model/delete_model_v1_0.rs

@@ -54,17 +54,6 @@ pub async fn delete_model(
         .await
         .map_err(map_ainari_error_to_api_response)?;
 
-    // send request to torii to delete the proxy, which is connected to the model
-    proxy_clients::delete_proxy(
-        &endpoints.torii,
-        &context.token,
-        &config::INTERNAL_API_KEY,
-        &proxy_uuid,
-        config::CONFIG.skip_tls_verification,
-    )
-    .await
-    .map_err(map_ainari_error_to_api_response)?;
-
     // send request to sakura to delete the model
     model_clients::delete_model(
         &host_data.address,
@@ -80,5 +69,16 @@ pub async fn delete_model(
     meta_model_table::delete_meta_model(&model_uuid, &context)
         .map_err(|e| map_db_uuid_get_delete_error("model-meta", &model_uuid, e))?;
 
+    // send request to torii to delete the proxy, which is connected to the model
+    proxy_clients::delete_proxy(
+        &endpoints.torii,
+        &context.token,
+        &config::INTERNAL_API_KEY,
+        &proxy_uuid,
+        config::CONFIG.skip_tls_verification,
+    )
+    .await
+    .map_err(map_ainari_error_to_api_response)?;
+
     Ok(NoContent)
 }

src/binaries/sakura/src/api/http_endpoints/model/delete_model_internal_v1_0.rs

@@ -13,15 +13,16 @@
 // limitations under the License.
 
 use actix_web::web::Path;
+use ainari_api_structs::task_structs::TaskState;
 use apistos::actix::NoContent;
 use apistos::api_operation;
 use uuid::Uuid;
 
 use crate::core::model_handler;
 use crate::database::model_table;
+use crate::database::task_table;
 
-use ainari_api::common_functions::map_ainari_error_to_api_response;
-use ainari_api::common_functions::map_db_uuid_get_delete_error;
+use ainari_api::common_functions::*;
 use ainari_api::errors::ErrorResponse;
 use ainari_api_structs::user_context::UserContext;
 
@@ -38,13 +39,50 @@ pub async fn delete_model_internal(
     model_uuid: Path<Uuid>,
     context: UserContext,
 ) -> Result<NoContent, ErrorResponse> {
+    // list all tasks
+    let tasks = match task_table::list_tasks(&model_uuid, &context) {
+        Ok(tasks) => tasks,
+        Err(e) => {
+            log::error!("Failed to get list of tasks form database: '{e}'");
+            return Err(ErrorResponse::InternalError("Internal Error".to_string()));
+        }
+    };
+
+    // abort all open tasks
+    for task in tasks {
+        let uuid = convert_uuid(&task.uuid)?;
+        let task_state = super::task::convert_task_state(&task.task_state)?;
+
+        if task_state == TaskState::Created
+            || task_state == TaskState::Queued
+            || task_state == TaskState::Active
+        {
+            task_table::update_task_state(&uuid, &TaskState::Aborted)
+                .map_err(|e| map_db_uuid_get_delete_error("task", &uuid, e))?;
+        }
+    }
+
+    // prepare delete of model from core
+    let model_handle = model_handler::CLUSTER_HANDLER
+        .write()
+        .expect("mutex poisoned");
+    let model_interface = model_handle
+        .get_model_interface(&model_uuid)
+        .map_err(map_ainari_error_to_api_response)?;
+    drop(model_handle);
+
+    // stop the interface. This must be done outside of the model_handler to avoid a dead-lock
+    let mut interface = model_interface.lock().expect("mutex poisoned");
+    interface.stop();
+
     // delete model from core
     let mut model_handle = model_handler::CLUSTER_HANDLER
         .write()
         .expect("mutex poisoned");
     model_handle
         .delete_model(&model_uuid)
         .map_err(map_ainari_error_to_api_response)?;
+    drop(model_handle);
 
     // delete model from database
     model_table::delete_model(&model_uuid, &context)

src/binaries/sakura/src/api/http_endpoints/model/task/mod.rs

@@ -139,7 +139,7 @@ fn convert_task_type(task_type: &String) -> Result<TaskType, ErrorResponse> {
 ///
 /// # Returns
 /// * `Result<TaskState, ErrorResponse>` - The converted task state or an error
-fn convert_task_state(task_state: &String) -> Result<TaskState, ErrorResponse> {
+pub fn convert_task_state(task_state: &String) -> Result<TaskState, ErrorResponse> {
     let converted_task_state = TaskState::from_str(task_state.as_str()).map_err(|_| {
         log::error!("Failed to convert task-state '{task_state}'");
         ErrorResponse::InternalError("Internal Error".to_string())

src/binaries/sakura/src/core/model_handler.rs

@@ -607,7 +607,7 @@ impl ModelDataHandler {
     /// * `Result<(), AinariError>` - Ok if the model was deleted successfully, Err otherwise
     pub fn delete_model(&mut self, model_uuid: &Uuid) -> Result<(), AinariError> {
         if !self.models.contains_key(model_uuid) {
-            let msg = format!("Model with uuid '{model_uuid}' not found.");
+            let msg: String = format!("Model with uuid '{model_uuid}' not found.");
             return Err(AinariError::InvalidInput(msg));
         }
 
@@ -1042,8 +1042,8 @@ mod tests {
             assert_eq!(hexagons.len(), 1);
         }
 
-        assert!(root_handler.delete_model(&model_uuid).is_ok());
-        assert!(root_handler.delete_model(&model_uuid).is_err());
+        assert!(root_handler.prepare_delete_model(&model_uuid).is_ok());
+        assert!(root_handler.prepare_delete_model(&model_uuid).is_err());
     }
 
     #[test]

src/binaries/sakura/src/core/model_interface.rs

@@ -140,6 +140,14 @@ impl ModelInterface {
     ///
     /// This method sets the running flag to false and joins the worker thread.
     pub fn stop(&mut self) {
+        // remove all open tasks from the queue
+        let mut queue_handle = self.queue.lock().expect("mutex poisoned");
+        queue_handle.clear();
+        drop(queue_handle);
+
+        thread::sleep(std::time::Duration::from_millis(5));
+
+        // stop all threads
         self.running.store(false, Ordering::Relaxed);
         if let Some(handle) = self.handle.take() {
             let _ = handle.join();

src/binaries/sakura/src/core/processing/task_queue.rs

@@ -73,6 +73,11 @@ impl TaskQueue {
     pub fn len(&self) -> usize {
         self.queue.len()
     }
+
+    /// Removed all remaining entries from the queue
+    pub fn clear(&mut self) {
+        self.queue.clear();
+    }
 }
 
 /// Initializes a new empty task queue.

src/libs/rust/ainari_api_structs/src/task_structs.rs

@@ -54,7 +54,7 @@ impl FromStr for TaskType {
     }
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, JsonSchema, ApiComponent)]
+#[derive(Serialize, Deserialize, PartialEq, Debug, Clone, JsonSchema, ApiComponent)]
 pub enum TaskState {
     Created = 0,
     Queued = 1,

src/libs/rust/ainari_clients/src/lib.rs

@@ -64,7 +64,7 @@ pub fn prepare_client(address: &str, insecure: bool) -> Client {
             .finish()
     } else {
         // Return a regular HTTP client for non-HTTPS connections
-        Client::new()
+        Client::builder().timeout(Duration::from_secs(60)).finish()
     }
 }