feat(backend) fix deadlock issue, code cleanup

backend/server/src/scheduler/mod.rs

@@ -19,8 +19,8 @@ pub mod tasks;
 
 #[derive(Default)]
 pub struct Scheduler {
-    running: Arc<RwLock<HashSet<Uuid>>>,
-    parking_pool: Arc<RwLock<HashMap<Uuid, Arc<Box<dyn CronTask>>>>>,
+    running: RwLock<HashSet<Uuid>>,
+    parking_pool: RwLock<HashMap<Uuid, Arc<Box<dyn CronTask>>>>,
     runner: Runner,
 }
 
@@ -47,7 +47,9 @@ impl Scheduler {
         } else if self.parking_pool.read()?.contains_key(id) {
             let func = self.parking_pool.write()?.remove(id).unwrap();
             self.running.write()?.insert(*id);
-            self.runner.add_with_id(func, id)
+            self.runner
+                .add_with_id(func, id)
+                .and_then(|_| self.runner.run_id(id))
         } else {
             Err(IdNotFound)
         }
@@ -55,21 +57,14 @@ impl Scheduler {
 
     /// Stop a task with `id`
     pub fn stop(&self, id: &Uuid) -> Result<(), SchedulerError> {
-        dbg!("stop() called");
         if self.running.read()?.contains(id) {
-            dbg!("b1");
             let func = self.runner.terminate_remove(*id)?;
-            dbg!("b1a");
             self.running.write()?.remove(id);
-            dbg!("b1b");
             self.parking_pool.write()?.insert(*id, func.clone());
-            dbg!("b1c");
             Ok(())
         } else if self.parking_pool.read()?.get(id).is_some() {
-            dbg!("b2");
             Ok(())
         } else {
-            dbg!("b3");
             Err(IdNotFound)
         }
     }
@@ -98,18 +93,16 @@ impl Scheduler {
 
 #[cfg(test)]
 mod tests {
-    use crate::scheduler::tasks::basic::BasicTask;
-    use crate::scheduler::Scheduler;
-    use std::error::Error;
-    use std::thread;
-    use std::time::Duration;
+    use std::{error::Error, thread, time::Duration};
 
-    #[test]
-    fn test_run() -> Result<(), Box<dyn Error>> {
-        let scheduler = Scheduler::default();
-        let task = BasicTask::new(|| println!("test print"), "* * * * * *")?;
-        let id = scheduler.insert(Box::new(task))?;
-        scheduler.start()?;
+    use crate::scheduler::{
+        tasks::{async_basic::AsyncBasic, basic::BasicTask},
+        Scheduler,
+    };
+
+    use uuid::Uuid;
+
+    fn execution(scheduler: Scheduler, id: Uuid) -> Result<(), Box<dyn Error>> {
         for i in 1..=2 {
             println!("Loop {} start", i);
             println!("Run {}", i);
@@ -124,4 +117,25 @@ mod tests {
         }
         Ok(())
     }
+
+    #[test]
+    fn test_run() -> Result<(), Box<dyn Error>> {
+        let scheduler = Scheduler::default();
+        let task = BasicTask::new(|| println!("test print"), "* * * * * *")?;
+        let id = scheduler.insert(Box::new(task))?;
+        scheduler.start()?;
+        execution(scheduler, id)
+    }
+
+    #[test]
+    fn test_run_async() -> Result<(), Box<dyn Error>> {
+        let scheduler = Scheduler::default();
+        let task = AsyncBasic::new(
+            || Box::pin(async { println!("test async print") }),
+            "* * * * * *",
+        )?;
+        let id = scheduler.insert(Box::new(task))?;
+        scheduler.start()?;
+        execution(scheduler, id)
+    }
 }

backend/server/src/scheduler/runner.rs

@@ -41,9 +41,10 @@ impl TaskEntity {
 
     pub async fn run_blocking(&self) {
         loop {
+            self.status.sleeper.read().unwrap().start();
             if self.status.expected_running.load(Ordering::SeqCst) {
-                self.status.sleeper.read().unwrap().start();
                 self.func.call().await;
+
                 if let Some(next_execution) = self.func.get_next_execution() {
                     *(self.status.sleeper.write().unwrap()) = next_execution.into();
                 } else {
@@ -80,20 +81,33 @@ impl Runner {
         let list = &self.task_list.read()?;
         self.rt.block_on(async {
             for (_, entity) in list.iter() {
-                if !entity.status.is_running.load(Ordering::SeqCst)
-                    && entity.status.expected_running.load(Ordering::SeqCst)
-                {
-                    entity.status.is_running.store(true, Ordering::SeqCst);
-                    tokio::spawn({
-                        let entity_clone = entity.clone();
-                        async move { entity_clone.run_blocking().await }
-                    });
-                }
+                self.run_entity(entity.clone()).await
             }
         });
         Ok(())
     }
 
+    pub fn run_id(&self, id: &Uuid) -> Result<(), SchedulerError> {
+        if let Some(entity) = self.task_list.read()?.get(id).cloned() {
+            self.rt.block_on(async { self.run_entity(entity).await });
+            Ok(())
+        } else {
+            Err(IdNotFound)
+        }
+    }
+
+    pub async fn run_entity(&self, entity: Arc<TaskEntity>) {
+        if !entity.status.is_running.load(Ordering::SeqCst)
+            && entity.status.expected_running.load(Ordering::SeqCst)
+        {
+            entity.status.is_running.store(true, Ordering::SeqCst);
+            tokio::spawn({
+                let entity_clone = entity.clone();
+                async move { entity_clone.run_blocking().await }
+            });
+        }
+    }
+
     pub fn add_with_id(
         &self,
         task: Arc<Box<dyn CronTask>>,
@@ -134,11 +148,12 @@ impl Runner {
     }
 
     pub fn terminate_remove(&self, id: Uuid) -> Result<Arc<Box<dyn CronTask>>, SchedulerError> {
-        match self.task_list.read()?.get(&id) {
+        let mut list = self.task_list.write()?;
+        match list.get(&id) {
             None => Err(IdNotFound),
             Some(task) => {
                 task.terminate();
-                let func = &self.task_list.write()?.remove(&id).unwrap().func;
+                let func = list.remove(&id).unwrap().func.clone();
                 Ok(func.clone())
             }
         }