diff --git a/backend/server/src/scheduler/mod.rs b/backend/server/src/scheduler/mod.rs index 67a7b47..456dc76 100644 --- a/backend/server/src/scheduler/mod.rs +++ b/backend/server/src/scheduler/mod.rs @@ -19,8 +19,8 @@ pub mod tasks; #[derive(Default)] pub struct Scheduler { - running: Arc>>, - parking_pool: Arc>>>>, + running: RwLock>, + parking_pool: RwLock>>>, 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> { - 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> { 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> { + 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> { + 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) + } } diff --git a/backend/server/src/scheduler/runner.rs b/backend/server/src/scheduler/runner.rs index 71c09cd..15a3181 100644 --- a/backend/server/src/scheduler/runner.rs +++ b/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) { + 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>, @@ -134,11 +148,12 @@ impl Runner { } pub fn terminate_remove(&self, id: Uuid) -> Result>, 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()) } }