diff --git a/backend/server/src/modules/blog/mod.rs b/backend/server/src/modules/blog/mod.rs index bdda532..3f17c69 100644 --- a/backend/server/src/modules/blog/mod.rs +++ b/backend/server/src/modules/blog/mod.rs @@ -3,7 +3,6 @@ mod cron; use crate::{ modules::{ - Module, blog::{ api::{ get_page_count, @@ -12,11 +11,12 @@ use crate::{ }, // cron::check_update, }, + Module, }, - scheduler::tasks::CronTask, + scheduler::tasks::{async_basic::AsyncBasic, CronTask}, }; -use axum::{Router, routing::get}; +use axum::{routing::get, Router}; use sqlx::PgPool; const MOUNT_POINT: &str = "/blog"; @@ -47,6 +47,9 @@ impl Module for Blog { // BasicTask::new(move || check_update::task(&db_handler), "0/15 * * * * *") // .unwrap() // .to_task(), + AsyncBasic::new(|| Box::pin(async { println!("test") }), "* * * * * *") + .unwrap() + .to_task(), ] } } diff --git a/backend/server/src/modules/mod.rs b/backend/server/src/modules/mod.rs index 32089d8..ab1e7dd 100644 --- a/backend/server/src/modules/mod.rs +++ b/backend/server/src/modules/mod.rs @@ -5,11 +5,12 @@ use std::num::ParseIntError; use crate::{ constants::config::server::{default_value, var_name}, - scheduler::{Scheduler, tasks::CronTask}, + scheduler::{tasks::CronTask, Scheduler}, }; use macros::error_panic; use axum::Router; +use tokio::net::TcpListener; use tracing::{info, warn}; // todo: merge `get_server_router` & `get_mount_point`, make Option @@ -51,13 +52,7 @@ impl ModuleTree { Err(_) => default_value::PORT, }; - let listener = match tokio::net::TcpListener::bind(format!( - "{}:{}", - default_value::HOST, - port - )) - .await - { + let listener = match TcpListener::bind(format!("{}:{}", default_value::HOST, port)).await { Ok(listener) => { info!("server started on {}:{}", default_value::HOST, port); listener @@ -77,6 +72,8 @@ impl ModuleTree { app = app.nest(module.get_mount_point(), module.get_server_router()); } + scheduler.start().unwrap(); + axum::serve(listener, app).await.unwrap_or_else(|e| { error_panic!("failed to start axum: {}", e.to_string()); }); diff --git a/backend/server/src/scheduler/mod.rs b/backend/server/src/scheduler/mod.rs index 692ab24..67a7b47 100644 --- a/backend/server/src/scheduler/mod.rs +++ b/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>, - is_alive: AtomicBool, -} - #[derive(Default)] pub struct Scheduler { - tasks: Arc>>>>, - running: Arc>>>, + running: Arc>>, + parking_pool: Arc>>>>, + 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>>>, - 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; - } + 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> { - 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) -> Result { - 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>) -> Result, 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)); diff --git a/backend/server/src/scheduler/runner.rs b/backend/server/src/scheduler/runner.rs new file mode 100644 index 0000000..71c09cd --- /dev/null +++ b/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, + is_running: AtomicBool, + expected_running: AtomicBool, +} + +pub struct TaskEntity { + func: Arc>, + status: TaskStatus, +} + +impl TaskEntity { + pub fn new(func: Arc>) -> 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>>, + 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>, + id: &Uuid, + ) -> Result<(), SchedulerError> { + self.task_list + .write()? + .insert(*id, TaskEntity::new(task).into()); + Ok(()) + } + + pub fn add(&self, task: Arc>) -> Result { + let id = Uuid::new_v4(); + self.add_with_id(task, &id)?; + Ok(id) + } + + pub fn add_list_with_id( + &self, + tasks: Vec<(&Uuid, Arc>)>, + ) -> Result<(), SchedulerError> { + tasks + .into_iter() + .try_for_each(|(id, task)| self.add_with_id(task, id)) + } + + pub fn add_list( + &self, + tasks: Vec>>, + ) -> Result, SchedulerError> { + tasks + .into_iter() + .map(|task| { + let id = Uuid::new_v4(); + self.add_with_id(task, &id).map(move |_| id) + }) + .collect::, _>>() + } + + pub fn terminate_remove(&self, id: Uuid) -> Result>, 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() + } +} diff --git a/backend/server/src/scheduler/task_func.rs b/backend/server/src/scheduler/task_func.rs index af27ded..ceafe54 100644 --- a/backend/server/src/scheduler/task_func.rs +++ b/backend/server/src/scheduler/task_func.rs @@ -1,3 +1,15 @@ +use std::pin::Pin; + pub trait TaskFunc: Fn() + Send + Sync + 'static {} impl TaskFunc for T where T: Fn() + Send + Sync + Clone + 'static {} + +pub trait AsyncTaskFunc: + Fn() -> Pin + Send>> + Send + Sync + 'static +{ +} + +impl AsyncTaskFunc for T where + T: Fn() -> Pin + Send>> + Send + Sync + Clone + 'static +{ +} diff --git a/backend/server/src/scheduler/tasks/async_basic.rs b/backend/server/src/scheduler/tasks/async_basic.rs new file mode 100644 index 0000000..3de6f37 --- /dev/null +++ b/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, +} + +impl AsyncBasic { + pub fn new(func: F, cron_expr: &str) -> Result> + where + F: AsyncTaskFunc, + { + Ok(AsyncBasic { + schedule: Schedule::from_str(cron_expr)?, + func: Box::new(func), + }) + } + + pub fn to_task(self) -> Box { + Box::new(self) + } +} + +impl CronTask for AsyncBasic { + fn get_next_execution(&self) -> Option> { + self.schedule.upcoming(Utc).next() + } + + fn call(&self) -> Pin + Send)>> { + Box::pin((self.func)()) + } +} diff --git a/backend/server/src/scheduler/tasks/basic.rs b/backend/server/src/scheduler/tasks/basic.rs index 4c69db7..cc02ed7 100644 --- a/backend/server/src/scheduler/tasks/basic.rs +++ b/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 + Send + '_>> { + Box::pin(async { (self.func)() }) } } diff --git a/backend/server/src/scheduler/tasks/mod.rs b/backend/server/src/scheduler/tasks/mod.rs index 8058bf6..0c58c99 100644 --- a/backend/server/src/scheduler/tasks/mod.rs +++ b/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>; - fn call(&self); + fn call(&self) -> Pin + Send + '_>>; }