40 migrate to tokiosyncsemaphore for mutli tasks limit handling

Cargo.toml

@@ -1,17 +1,12 @@
 [package]
 name = "mangadex-desktop-api2"
-version = "0.5.5"
+version = "0.5.6-alpha.4"
 authors = ["tonymushah <tonymushahdev06@yahoo.com>"]
 license = "MIT OR Apache-2.0"
 description = "A Actix server for downloading manga, chapters, covers from Mangadex"
 edition = "2021"
 resolver = "2"
 
-[workspace]
-members = [
-    "test-tauri"
-]
-
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]

src/core.rs

@@ -1,5 +1,5 @@
 mod error;
 
-pub use error::Error;
+pub use error::{Error, ErrorType, WhenError};
 
 pub type ManagerCoreResult<T> = Result<T, Error>;

src/core/error.rs

@@ -4,7 +4,9 @@ use actix_web::ResponseError;
 use serde::Serialize;
 
 #[derive(serde::Serialize)]
+#[serde(rename_all = "camelCase")]
 pub struct WhenError {
+    type_: ErrorType,
     message: String,
     result: String,
 }
@@ -53,101 +55,169 @@ pub enum Error {
     TryIntError(#[from] TryFromIntError),
     #[error("An error occured when sending data between an oneshot channel \n Details: {0}")]
     OneshotRecvError(#[from] tokio::sync::oneshot::error::RecvError),
+    #[error("An error occured when acquiring a semaphore \n Details : {0}")]
+    AcquireError(#[from] tokio::sync::AcquireError),
+    #[error("The file transaction was been roolback due to an error. Details : {0}")]
+    RollBacked(String),
+    #[error("An RwLock occured \n Details : {0}")]
+    RwLockError(#[from] std::sync::PoisonError<String>),
+}
+
+#[derive(serde::Serialize, Debug, serde::Deserialize)]
+pub enum ErrorType {
+    Io,
+    ReqwestError,
+    MangadexAPIError,
+    TokioJoinError,
+    SerdeJsonError,
+    UuidError,
+    StringUtf8Error,
+    StringUTF16Error,
+    StringParseError,
+    Other,
+    ChapterDownloadBuilderError,
+    CoverDownloadBuilderError,
+    GetMangaBuilderError,
+    ListCoverBuilderError,
+    DownloadTaskLimitExceded,
+    TryIntError,
+    OneshotRecvError,
+    AcquireError,
+    RollBacked,
+    RwLockError,
 }
 
 impl ResponseError for Error {
     fn error_response(&self) -> actix_web::HttpResponse<actix_web::body::BoxBody> {
         match self {
             Error::Io(e) => actix_web::HttpResponse::InternalServerError().json(WhenError {
+                type_: ErrorType::Io,
                 message: e.to_string(),
                 result: "error".to_string(),
             }),
             Error::ReqwestError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::ReqwestError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::MangadexAPIError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::MangadexAPIError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::TokioJoinError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::TokioJoinError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::SerdeJsonError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::SerdeJsonError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::UuidError(e) => actix_web::HttpResponse::InternalServerError().json(WhenError {
+                type_: ErrorType::UuidError,
                 message: e.to_string(),
                 result: "error".to_string(),
             }),
             Error::StringUtf8Error(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::StringUtf8Error,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::StringUTF16Error(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::StringUTF16Error,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::StringParseError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::StringParseError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::Other(e) => actix_web::HttpResponse::InternalServerError().json(WhenError {
+                type_: ErrorType::Other,
                 message: e.to_string(),
                 result: "error".to_string(),
             }),
             Error::ChapterDownloadBuilderError(e) => actix_web::HttpResponse::InternalServerError()
                 .json(WhenError {
+                    type_: ErrorType::ChapterDownloadBuilderError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 }),
             Error::CoverDownloadBuilderError(e) => actix_web::HttpResponse::InternalServerError()
                 .json(WhenError {
+                    type_: ErrorType::CoverDownloadBuilderError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 }),
             Error::GetMangaBuilderError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::GetMangaBuilderError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::ListCoverBuilderError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::ListCoverBuilderError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::DownloadTaskLimitExceded { current, limit } => {
                 actix_web::HttpResponse::TooManyRequests().json(WhenError {
+                    type_: ErrorType::DownloadTaskLimitExceded,
                     message: format!("Download task limit exceded {current}/{limit}"),
                     result: "error".to_string(),
                 })
             }
             Error::TryIntError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::TryIntError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })
             }
             Error::OneshotRecvError(e) => {
                 actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::OneshotRecvError,
+                    message: e.to_string(),
+                    result: "error".to_string(),
+                })
+            }
+            Error::AcquireError(e) => {
+                actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::AcquireError,
+                    message: e.to_string(),
+                    result: "error".to_string(),
+                })
+            }
+            Error::RollBacked(e) => {
+                actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::RollBacked,
+                    message: e.to_string(),
+                    result: "error".to_string(),
+                })
+            }
+            Error::RwLockError(e) => {
+                actix_web::HttpResponse::InternalServerError().json(WhenError {
+                    type_: ErrorType::RwLockError,
                     message: e.to_string(),
                     result: "error".to_string(),
                 })

src/download.rs

@@ -1,4 +1,5 @@
 use futures::Future;
+use tokio::sync::{Semaphore, RwLock};
 use std::fmt::Debug;
 use std::{ops::Deref, sync::Arc};
 use tokio::sync::oneshot::channel;
@@ -17,13 +18,18 @@ pub mod manga;
 pub struct DownloadTaks {
     tasks: Arc<Mutex<JoinSet<()>>>,
     limit: u16,
+    sephamore : Arc<Semaphore>,
+    on_lock : Arc<RwLock<usize>>
 }
 
 impl Default for DownloadTaks {
     fn default() -> Self {
+        let limit = 20;
         Self {
             tasks: Arc::new(Mutex::new(JoinSet::default())),
-            limit: 20,
+            limit,
+            sephamore : Arc::new(Semaphore::new(limit.into())),
+            on_lock : Arc::new(RwLock::new(0))
         }
     }
 }
@@ -40,10 +46,12 @@ impl DownloadTaks {
         Self {
             tasks: Arc::new(Mutex::new(JoinSet::default())),
             limit,
+            sephamore : Arc::new(Semaphore::new(limit.into())),
+            on_lock : Arc::new(RwLock::new(0))
         }
     }
     pub async fn verify_limit(&self) -> bool {
-        self.lock().await.len() >= <u16 as Into<usize>>::into(self.limit)
+        self.sephamore.available_permits() == 0
     }
     pub async fn spawn<F>(&mut self, task: F) -> ManagerCoreResult<AbortHandle>
     where
@@ -55,21 +63,44 @@ impl DownloadTaks {
                 limit: self.limit,
             })
         } else {
-            Ok(self.tasks.lock().await.spawn(task))
+            let permit = self.sephamore.clone().acquire_owned().await?;
+            Ok(self.tasks.lock().await.spawn(async move {
+                task.await;
+                drop(permit)
+            }))
         }
     }
-    pub async fn lock_spawn<F>(&mut self, task: F) -> AbortHandle
+    async fn add_something_to_lock_list(&self) -> ManagerCoreResult<()> {
+        let mut data = self.on_lock.write().await;
+        *data += 1;
+        Ok(())
+    }
+    async fn remove_something_to_lock_list(&self) -> ManagerCoreResult<()> {
+        let mut data = self.on_lock.write().await;
+        *data -= 1;
+        Ok(())
+    }
+    pub async fn lock_spawn<F>(&mut self, task: F) -> ManagerCoreResult<AbortHandle>
     where
         F: Future<Output = ()> + Send + 'static,
     {
         if self.verify_limit().await {
             let mut tasks = self.tasks.lock().await;
-            println!("Lock spawing");
+            self.add_something_to_lock_list().await?;
             tasks.join_next().await;
-            tasks.spawn(task)
+            self.remove_something_to_lock_list().await?;
+            let permit = self.sephamore.clone().acquire_owned().await?;
+            Ok(tasks.spawn(async move {
+                task.await;
+                drop(permit)
+            }))
         } else {
             let mut tasks = self.tasks.lock().await;
-            tasks.spawn(task)
+            let permit = self.sephamore.clone().acquire_owned().await?;
+            Ok(tasks.spawn(async move {
+                task.await;
+                drop(permit)
+            }))
         }
     }
     pub async fn spawn_with_data<T>(&mut self, task: T) -> ManagerCoreResult<T::Output>
@@ -96,10 +127,16 @@ impl DownloadTaks {
         self.lock_spawn(async {
             let _ = sender.send(task.await);
         })
-        .await;
+        .await?;
         Ok(receiver.await?)
     }
     pub fn get_limit(&self) -> u16 {
         self.limit
     }
+    pub fn get_running_tasks(&self) -> usize {
+        <u16 as Into<usize>>::into(self.limit) - self.sephamore.available_permits()
+    }
+    pub async fn get_locked_tasks(&self) -> usize {
+        *self.on_lock.read().await
+    }
 }

src/download/chapter.rs

@@ -13,6 +13,7 @@ use uuid::Uuid;
 
 use crate::core::Error;
 use crate::server::traits::{AccessDownloadTasks, AccessHistory};
+use crate::settings::file_history::history_w_file::traits::{NoLFAsyncAutoCommitRollbackRemove, NoLFAsyncAutoCommitRollbackInsert};
 use crate::settings::files_dirs::DirsOptions;
 use crate::utils::chapter::{ChapterUtils, ChapterUtilsWithID};
 use crate::{
@@ -87,10 +88,10 @@ impl ChapterDownload {
         self.download_json_data(task_manager).await?;
         if let Ok(data) = chapter_utils.is_manga_there() {
             if !data {
-                chapter_utils.patch_manga(history, task_manager).await?;
+                (chapter_utils).patch_manga(history, task_manager).await?;
             }
         } else {
-            chapter_utils.patch_manga(history, task_manager).await?;
+            (chapter_utils).patch_manga(history, task_manager).await?;
         }
         Ok(())
     }
@@ -106,42 +107,18 @@ impl ChapterDownload {
             chapter_id,
             mangadex_api_types_rust::RelationshipType::Chapter,
         );
-        match history.insert_in_history(&history_entry).await {
-            Ok(_) => (),
-            Err(error_) => {
-                if let Error::Io(error) = error_ {
-                    if error.kind() != std::io::ErrorKind::AlreadyExists {
-                        return Err(crate::core::Error::Io(error));
-                    }
-                } else {
-                    return Err(error_);
-                }
-            }
-        };
-        history.commit_rel(history_entry.get_data_type()).await?;
+        <dyn AccessHistory as NoLFAsyncAutoCommitRollbackInsert<HistoryEntry>>::insert(history, history_entry).await?;
         Ok(history_entry)
     }
     pub async fn end_transation<'a, H>(
-        &self,
+        &'a self,
         entry: HistoryEntry,
         history: &'a mut H,
     ) -> ManagerCoreResult<()>
     where
         H: AccessHistory,
     {
-        if let Err(error) = history.remove_in_history(&entry).await {
-            if let Error::Io(io) = error{
-                if io.kind() == std::io::ErrorKind::NotFound {
-                    history.commit_rel(entry.get_data_type()).await?;
-                }else{
-                    return Err(Error::Io(io))
-                }
-            }else {
-                return Err(error);
-            }
-        }else{
-            history.commit_rel(entry.get_data_type()).await?;
-        }
+        <dyn AccessHistory as NoLFAsyncAutoCommitRollbackRemove<HistoryEntry>>::remove(history, entry).await?;
         Ok(())
     }
     pub async fn download_chapter<'a, H, D>(

src/lib.rs

@@ -9,7 +9,7 @@ use log::{info, warn};
 use server::AppState;
 mod r#core;
 
-pub use r#core::Error;
+pub use r#core::{Error, ErrorType};
 
 pub mod download;
 #[cfg(feature = "feeds")]