feat(backend) scheduler rewrite (deadlock for now)

backend/server/src/scheduler/mod.rs

@@ -1,172 +1,94 @@
 use std::{
-    collections::HashMap,
-    sync::{
-        atomic::{AtomicBool, Ordering}, Arc,
-        RwLock,
-    },
-    thread,
+    collections::{HashMap, HashSet},
+    sync::{Arc, RwLock},
 };
 
 use crate::scheduler::{
-    scheduler_error::{
-        SchedulerError, SchedulerError::IdAlreadyRunning, SchedulerError::IdNotFound,
-    },
-    sleeper::SkippableSleeper,
+    runner::Runner,
+    scheduler_error::{SchedulerError, SchedulerError::IdNotFound},
     tasks::CronTask,
 };
 
-use tracing::log::error;
 use uuid::Uuid;
 
+mod runner;
 pub mod scheduler_error;
 mod sleeper;
 mod task_func;
 pub mod tasks;
 
-struct TaskStatus {
-    sleeper: Option<Arc<SkippableSleeper>>,
-    is_alive: AtomicBool,
-}
-
 #[derive(Default)]
 pub struct Scheduler {
-    tasks: Arc<RwLock<HashMap<Uuid, Arc<Box<dyn CronTask>>>>>,
-    running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
+    running: Arc<RwLock<HashSet<Uuid>>>,
+    parking_pool: Arc<RwLock<HashMap<Uuid, Arc<Box<dyn CronTask>>>>>,
+    runner: Runner,
 }
 
 impl Scheduler {
     /// Start all tasks in the scheduler
     pub fn start(&self) -> Result<(), SchedulerError> {
-        for (k, _v) in self.tasks.read()?.iter() {
-            self.run_non_blocking(k)?
-        }
-        Ok(())
+        self.runner.run_all()
     }
 
     /// Terminate all tasks in the scheduler
-    pub fn halt() {}
-
-    fn run_task(
-        running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
-        id: Uuid,
-        task: Arc<Box<dyn CronTask>>,
-    ) -> Result<(), SchedulerError> {
-        if running.read()?.get(&id).is_some() {
-            return Err(IdAlreadyRunning);
-        } else {
-            running.write()?.insert(
-                id,
-                TaskStatus {
-                    sleeper: None,
-                    is_alive: AtomicBool::new(true),
-                }
-                .into(),
-            );
-        }
-        loop {
-            if let Some(next_execution) = task.get_next_execution() {
-                // guarantee is alive
-                let task_status = {
-                    let running_reading_guard = running.read()?;
-                    running_reading_guard.get(&id).cloned().clone()
-                };
-                if let Some(task_status) = task_status {
-                    if !&task_status.is_alive.load(Ordering::SeqCst) {
-                        {
-                            let mut running_lock_guard = running.write()?;
-                            running_lock_guard.remove(&id);
-                        }
-                        return Ok(());
-                    } else {
-                        let sleeper = Arc::from(SkippableSleeper::from(next_execution));
-                        {
-                            let mut running_lock_guard = running.write()?;
-                            running_lock_guard.insert(
-                                id,
-                                TaskStatus {
-                                    sleeper: Some(sleeper.clone()),
-                                    is_alive: AtomicBool::new(true),
-                                }
-                                .into(),
-                            );
-                        }
-                        sleeper.start();
-                        task.call();
-                    }
-                }
-            } else {
-                error!("unexpected error ");
-                break;
-            }
+    pub fn halt(&self) -> Result<(), SchedulerError> {
+        for id in self.running.read()?.iter() {
+            let func = &self.runner.terminate_remove(*id)?;
+            self.running.write()?.remove(id);
+            self.parking_pool.write()?.insert(*id, func.clone());
         }
         Ok(())
     }
 
     /// Run a task with `id`
     pub fn run(&self, id: &Uuid) -> Result<(), SchedulerError> {
-        let id = id.to_owned();
-        if let Some(task) = self.tasks.read()?.get(&id) {
-            Self::run_task(self.running.clone(), id, task.clone())
-        } else {
-            Err(IdNotFound)
-        }
-    }
-
-    pub fn run_non_blocking(&self, id: &Uuid) -> Result<(), SchedulerError> {
-        let id = id.to_owned();
-        if let Some(task) = self.tasks.read()?.get(&id) {
-            let running_clone = self.running.clone();
-            let task_clone = task.clone();
-            thread::spawn(move || Self::run_task(running_clone, id, task_clone));
+        if self.running.read()?.contains(id) {
             Ok(())
+        } 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)
         } else {
             Err(IdNotFound)
         }
     }
 
-    fn get_running(&self, id: &Uuid) -> Option<Arc<TaskStatus>> {
-        let ret = self.running.read().unwrap().get(id).cloned();
-        ret
-    }
-
     /// Stop a task with `id`
     pub fn stop(&self, id: &Uuid) -> Result<(), SchedulerError> {
-        match self.get_running(id) {
-            None => Err(IdNotFound),
-            Some(status) => {
-                status.is_alive.store(false, Ordering::SeqCst);
-                if let Some(s) = &status.sleeper {
-                    s.skip();
-                } else {
-                    return Err(IdAlreadyRunning);
-                }
-                Ok(())
-            }
+        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)
         }
     }
 
-    pub fn now(&self, id: &Uuid) -> Result<(), SchedulerError> {
-        match self.get_running(id) {
-            None => Err(IdNotFound),
-            Some(status) => {
-                if let Some(s) = &status.sleeper {
-                    s.skip()
-                } else {
-                    return Err(IdAlreadyRunning);
-                }
-                Ok(())
-            }
-        }
-    }
+    // pub fn now(&self, id: &Uuid) -> Result<(), SchedulerError> {
+    // }
 
     pub fn insert(&self, task: Box<dyn CronTask>) -> Result<Uuid, SchedulerError> {
-        let id = Uuid::new_v4();
-        self.tasks.write()?.insert(id, Arc::new(task));
+        let id = self.runner.add(task.into())?;
+        self.running.write()?.insert(id);
         Ok(id)
     }
 
     pub fn insert_list(&self, tasks: Vec<Box<dyn CronTask>>) -> Result<Vec<Uuid>, SchedulerError> {
-        tasks.into_iter().map(|t| self.insert(t)).collect()
+        let ids = self
+            .runner
+            .add_list(tasks.into_iter().map(|t| t.into()).collect())?;
+        self.running.write()?.extend(&ids);
+        Ok(ids)
     }
 
     pub fn new() -> Self {
@@ -187,11 +109,14 @@ mod tests {
         let scheduler = Scheduler::default();
         let task = BasicTask::new(|| println!("test print"), "* * * * * *")?;
         let id = scheduler.insert(Box::new(task))?;
+        scheduler.start()?;
         for i in 1..=2 {
             println!("Loop {} start", i);
             println!("Run {}", i);
-            scheduler.run_non_blocking(&id)?;
+            scheduler.run(&id)?;
+            println!("Sleep {}", i);
             thread::sleep(Duration::from_secs(4));
+            println!("Sleep {} finished", i);
             scheduler.stop(&id)?;
             println!("Stop {}", i);
             thread::sleep(Duration::from_secs(2));

backend/server/src/scheduler/runner.rs

@@ -0,0 +1,152 @@
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicBool, Ordering}, Arc,
+        RwLock,
+    },
+};
+
+use crate::scheduler::{
+    scheduler_error::{SchedulerError, SchedulerError::IdNotFound},
+    sleeper::SkippableSleeper,
+    tasks::CronTask,
+};
+
+use tokio::runtime::Runtime;
+use uuid::Uuid;
+
+struct TaskStatus {
+    sleeper: RwLock<SkippableSleeper>,
+    is_running: AtomicBool,
+    expected_running: AtomicBool,
+}
+
+pub struct TaskEntity {
+    func: Arc<Box<dyn CronTask>>,
+    status: TaskStatus,
+}
+
+impl TaskEntity {
+    pub fn new(func: Arc<Box<dyn CronTask>>) -> Self {
+        let next_execution: SkippableSleeper = func.get_next_execution().unwrap().into();
+        Self {
+            func,
+            status: TaskStatus {
+                sleeper: next_execution.into(),
+                is_running: false.into(),
+                expected_running: true.into(),
+            },
+        }
+    }
+
+    pub async fn run_blocking(&self) {
+        loop {
+            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 {
+                    self.status.is_running.store(false, Ordering::SeqCst);
+                    break;
+                }
+            } else {
+                break;
+            }
+        }
+    }
+
+    pub fn terminate(&self) {
+        self.status.expected_running.store(false, Ordering::SeqCst);
+        self.status.is_running.store(false, Ordering::SeqCst);
+        self.status.sleeper.read().unwrap().skip();
+    }
+}
+
+pub struct Runner {
+    task_list: RwLock<HashMap<Uuid, Arc<TaskEntity>>>,
+    rt: Runtime,
+}
+
+impl Runner {
+    pub fn new() -> Self {
+        Self {
+            task_list: HashMap::new().into(),
+            rt: Runtime::new().unwrap(),
+        }
+    }
+
+    pub fn run_all(&self) -> Result<(), SchedulerError> {
+        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 }
+                    });
+                }
+            }
+        });
+        Ok(())
+    }
+
+    pub fn add_with_id(
+        &self,
+        task: Arc<Box<dyn CronTask>>,
+        id: &Uuid,
+    ) -> Result<(), SchedulerError> {
+        self.task_list
+            .write()?
+            .insert(*id, TaskEntity::new(task).into());
+        Ok(())
+    }
+
+    pub fn add(&self, task: Arc<Box<dyn CronTask>>) -> Result<Uuid, SchedulerError> {
+        let id = Uuid::new_v4();
+        self.add_with_id(task, &id)?;
+        Ok(id)
+    }
+
+    pub fn add_list_with_id(
+        &self,
+        tasks: Vec<(&Uuid, Arc<Box<dyn CronTask>>)>,
+    ) -> Result<(), SchedulerError> {
+        tasks
+            .into_iter()
+            .try_for_each(|(id, task)| self.add_with_id(task, id))
+    }
+
+    pub fn add_list(
+        &self,
+        tasks: Vec<Arc<Box<dyn CronTask>>>,
+    ) -> Result<Vec<Uuid>, SchedulerError> {
+        tasks
+            .into_iter()
+            .map(|task| {
+                let id = Uuid::new_v4();
+                self.add_with_id(task, &id).map(move |_| id)
+            })
+            .collect::<Result<Vec<_>, _>>()
+    }
+
+    pub fn terminate_remove(&self, id: Uuid) -> Result<Arc<Box<dyn CronTask>>, SchedulerError> {
+        match self.task_list.read()?.get(&id) {
+            None => Err(IdNotFound),
+            Some(task) => {
+                task.terminate();
+                let func = &self.task_list.write()?.remove(&id).unwrap().func;
+                Ok(func.clone())
+            }
+        }
+    }
+}
+
+impl Default for Runner {
+    fn default() -> Self {
+        Self::new()
+    }
+}

backend/server/src/scheduler/task_func.rs

@@ -1,3 +1,15 @@
+use std::pin::Pin;
+
 pub trait TaskFunc: Fn() + Send + Sync + 'static {}
 
 impl<T> TaskFunc for T where T: Fn() + Send + Sync + Clone + 'static {}
+
+pub trait AsyncTaskFunc:
+    Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static
+{
+}
+
+impl<T> AsyncTaskFunc for T where
+    T: Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + Clone + 'static
+{
+}

backend/server/src/scheduler/tasks/async_basic.rs

@@ -0,0 +1,37 @@
+use std::{error::Error, pin::Pin, str::FromStr};
+
+use crate::{scheduler::task_func::AsyncTaskFunc, scheduler::tasks::CronTask};
+
+use chrono::{DateTime, Utc};
+use cron::Schedule;
+
+pub struct AsyncBasic {
+    schedule: Schedule,
+    func: Box<dyn AsyncTaskFunc>,
+}
+
+impl AsyncBasic {
+    pub fn new<F>(func: F, cron_expr: &str) -> Result<Self, Box<dyn Error>>
+    where
+        F: AsyncTaskFunc,
+    {
+        Ok(AsyncBasic {
+            schedule: Schedule::from_str(cron_expr)?,
+            func: Box::new(func),
+        })
+    }
+
+    pub fn to_task(self) -> Box<dyn CronTask> {
+        Box::new(self)
+    }
+}
+
+impl CronTask for AsyncBasic {
+    fn get_next_execution(&self) -> Option<DateTime<Utc>> {
+        self.schedule.upcoming(Utc).next()
+    }
+
+    fn call(&self) -> Pin<Box<(dyn Future<Output = ()> + Send)>> {
+        Box::pin((self.func)())
+    }
+}

backend/server/src/scheduler/tasks/basic.rs

@@ -1,5 +1,4 @@
-use std::error::Error;
-use std::str::FromStr;
+use std::{error::Error, pin::Pin, str::FromStr};
 
 use crate::scheduler::{task_func::TaskFunc, tasks::CronTask};
 
@@ -32,7 +31,7 @@ impl CronTask for BasicTask {
         self.schedule.upcoming(Utc).next()
     }
 
-    fn call(&self) {
-        (self.func)()
+    fn call(&self) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
+        Box::pin(async { (self.func)() })
     }
 }

backend/server/src/scheduler/tasks/mod.rs

@@ -1,8 +1,11 @@
+use std::pin::Pin;
+
 use chrono::{DateTime, Utc};
 
+pub mod async_basic;
 pub mod basic;
 
 pub trait CronTask: Send + Sync {
     fn get_next_execution(&self) -> Option<DateTime<Utc>>;
-    fn call(&self);
+    fn call(&self) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>;
 }