feat(backend) move module scheduler

backend/server/src/scheduler/mod.rs

@@ -0,0 +1,202 @@
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, RwLock,
+    },
+    thread,
+};
+
+use crate::scheduler::{
+    scheduler_error::{
+        SchedulerError, SchedulerError::IdAlreadyRunning, SchedulerError::IdNotFound,
+    },
+    sleeper::SkippableSleeper,
+    tasks::Task,
+};
+
+use tracing::log::error;
+use uuid::Uuid;
+
+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 Task>>>>>,
+    running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
+}
+
+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(())
+    }
+
+    /// 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 Task>>,
+    ) -> 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;
+            }
+        }
+        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));
+            Ok(())
+        } 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(())
+            }
+        }
+    }
+
+    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 insert(&self, task: Box<dyn Task>) -> Result<Uuid, SchedulerError> {
+        let id = Uuid::new_v4();
+        self.tasks.write()?.insert(id, Arc::new(task));
+        Ok(id)
+    }
+
+    pub fn insert_list(&self, tasks: Vec<Box<dyn Task>>) -> Result<Vec<Uuid>, SchedulerError> {
+        tasks.into_iter().map(|t| self.insert(t)).collect()
+    }
+
+    pub fn new() -> Self {
+        Self::default()
+    }
+}
+
+#[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;
+
+    #[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))?;
+        for i in 1..=2 {
+            println!("Loop {} start", i);
+            println!("Run {}", i);
+            scheduler.run_non_blocking(&id)?;
+            thread::sleep(Duration::from_secs(4));
+            scheduler.stop(&id)?;
+            println!("Stop {}", i);
+            thread::sleep(Duration::from_secs(2));
+            println!("Loop {} finished", i);
+        }
+        Ok(())
+    }
+}

backend/server/src/scheduler/scheduler_error.rs

@@ -0,0 +1,47 @@
+use std::{
+    error::Error,
+    fmt::{Debug, Display, Formatter},
+    sync::PoisonError,
+};
+
+use crate::scheduler::scheduler_error::SchedulerError::MutexPoison;
+
+pub enum SchedulerError {
+    IdNotFound,
+    IdAlreadyRunning,
+    MutexPoison(String),
+}
+
+impl<T> From<PoisonError<T>> for SchedulerError {
+    fn from(value: PoisonError<T>) -> Self {
+        MutexPoison(value.to_string())
+    }
+}
+
+impl Debug for SchedulerError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self)
+    }
+}
+
+impl Display for SchedulerError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                SchedulerError::IdNotFound => {
+                    "SchedulerError::IdNotFound".to_string()
+                }
+                SchedulerError::IdAlreadyRunning => {
+                    "SchedulerError::IdAlreadyRunning".to_string()
+                }
+                MutexPoison(str) => {
+                    format!("SchedulerError::MutexPoison: {}", str)
+                }
+            }
+        )
+    }
+}
+
+impl Error for SchedulerError {}

backend/server/src/scheduler/sleeper.rs

@@ -0,0 +1,80 @@
+use std::{
+    sync::{Arc, Condvar, Mutex},
+    thread,
+    time::Duration,
+};
+
+use chrono::{DateTime, Utc};
+
+#[derive(Clone, Debug, Default)]
+pub struct SkippableSleeper {
+    pub duration: Duration,
+    state: Arc<(Mutex<bool>, Condvar)>,
+}
+
+impl SkippableSleeper {
+    pub fn new(duration: Duration) -> Self {
+        SkippableSleeper {
+            duration,
+            state: Arc::new((Mutex::new(false), Condvar::new())),
+        }
+    }
+
+    // Blocks until the duration has passed or skip() is called
+    pub fn start(&self) {
+        let (lock, cvar) = &*self.state;
+        let mut received = lock.lock().unwrap();
+        let state_clone = Arc::clone(&self.state);
+        let duration_clone = self.duration;
+        thread::spawn(move || {
+            thread::sleep(duration_clone);
+            let (lock, cvar) = &*state_clone;
+            let mut received = lock.lock().unwrap();
+            *received = true;
+            cvar.notify_one();
+        });
+        while !*received {
+            received = cvar.wait(received).unwrap();
+        }
+    }
+
+    // Skip the blocking delay
+    pub fn skip(&self) {
+        let (lock, cvar) = &*self.state;
+        let mut received = lock.lock().unwrap();
+        *received = true;
+        cvar.notify_one();
+    }
+}
+
+impl From<DateTime<Utc>> for SkippableSleeper {
+    fn from(value: DateTime<Utc>) -> Self {
+        let from_utc_timestamp = value.timestamp() as u64;
+        let curr_utc_timestamp = Utc::now().timestamp() as u64;
+        SkippableSleeper::new(Duration::from_secs(from_utc_timestamp - curr_utc_timestamp))
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::scheduler::sleeper::SkippableSleeper;
+    use std::thread;
+    use std::time::Duration;
+
+    #[test]
+    fn usage() {
+        let sleeper = SkippableSleeper::new(Duration::new(10, 0));
+        thread::spawn({
+            let sleeper = sleeper.clone();
+            move || {
+                println!("Sleeping...");
+                sleeper.start();
+                println!("Woke up!");
+            }
+        });
+        println!("Start Sleeping...");
+        thread::sleep(Duration::from_secs(4));
+        println!("Skipping sleep...");
+        sleeper.skip();
+    }
+}

backend/server/src/scheduler/task_func.rs

@@ -0,0 +1,3 @@
+pub trait TaskFunc: Fn() + Send + Sync + 'static {}
+
+impl<T> TaskFunc for T where T: Fn() + Send + Sync + Clone + 'static {}

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

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

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

@@ -0,0 +1,8 @@
+use chrono::{DateTime, Utc};
+
+pub mod basic;
+
+pub trait Task: Send + Sync {
+    fn get_next_execution(&self) -> Option<DateTime<Utc>>;
+    fn call(&self);
+}