diff --git a/backend/server/Cargo.toml b/backend/server/Cargo.toml index 481e539..1070414 100644 --- a/backend/server/Cargo.toml +++ b/backend/server/Cargo.toml @@ -7,8 +7,10 @@ edition = "2021" macros = { path = "../macros" } -tracing = "0.1.41" +tracing = { version = "0.1.41", features = ["log"] } git2 = "0.20.0" +chrono = "0.4.39" +uuid = { version = "1.13.2", features = ["v4"] } [build-dependencies] git2 = "0.20.0" diff --git a/backend/server/src/main.rs b/backend/server/src/main.rs index 5e6abc2..655a616 100644 --- a/backend/server/src/main.rs +++ b/backend/server/src/main.rs @@ -1,4 +1,5 @@ mod constants; +mod scheduler; const BUILD_ID: &str = env!("BUILD_ID"); diff --git a/backend/server/src/scheduler/mod.rs b/backend/server/src/scheduler/mod.rs new file mode 100644 index 0000000..71cb62e --- /dev/null +++ b/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>, + is_alive: AtomicBool, +} + +#[derive(Default)] +pub struct Scheduler { + tasks: Arc>>>>, + running: Arc>>>, +} + +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>>>, + id: Uuid, + task: Arc>, + ) -> 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> { + 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) -> Result { + let id = Uuid::new_v4(); + self.tasks.write()?.insert(id, Arc::new(task)); + Ok(id) + } + + pub fn insert_list(&self, tasks: Vec>) -> Result, 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> { + 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(()) + } +} diff --git a/backend/server/src/scheduler/scheduler_error.rs b/backend/server/src/scheduler/scheduler_error.rs new file mode 100644 index 0000000..d9783f7 --- /dev/null +++ b/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 From> for SchedulerError { + fn from(value: PoisonError) -> 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 {} diff --git a/backend/server/src/scheduler/sleeper.rs b/backend/server/src/scheduler/sleeper.rs new file mode 100644 index 0000000..abc085a --- /dev/null +++ b/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, 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> for SkippableSleeper { + fn from(value: DateTime) -> 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(); + } +} diff --git a/backend/server/src/scheduler/task_func.rs b/backend/server/src/scheduler/task_func.rs new file mode 100644 index 0000000..af27ded --- /dev/null +++ b/backend/server/src/scheduler/task_func.rs @@ -0,0 +1,3 @@ +pub trait TaskFunc: Fn() + Send + Sync + 'static {} + +impl TaskFunc for T where T: Fn() + Send + Sync + Clone + 'static {} diff --git a/backend/server/src/scheduler/tasks/basic.rs b/backend/server/src/scheduler/tasks/basic.rs new file mode 100644 index 0000000..71fc3a6 --- /dev/null +++ b/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, +} + +impl BasicTask { + pub fn new(func: F, cron_expr: &str) -> Result> + where + F: TaskFunc, + { + Ok(BasicTask { + schedule: Schedule::from_str(cron_expr)?, + func: Box::new(func), + }) + } + + pub fn to_task(self) -> Box { + Box::new(self) + } +} + +impl Task for BasicTask { + fn get_next_execution(&self) -> Option> { + self.schedule.upcoming(Utc).next() + } + + fn call(&self) { + (self.func)() + } +} diff --git a/backend/server/src/scheduler/tasks/mod.rs b/backend/server/src/scheduler/tasks/mod.rs new file mode 100644 index 0000000..7e3b87a --- /dev/null +++ b/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>; + fn call(&self); +}