diff --git a/Cargo.toml b/Cargo.toml index 227029b..9791f6a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,7 +28,7 @@ path = "lib/lib.rs" # macros "macros" = [] # scheduling -"sched" = ["dep:chrono"] +"sched" = ["dep:chrono", "utils.threads"] "sched.tokio" = ["dep:tokio", "tokio?/time", "tokio?/rt", "sched"] "sched.rule-recurrence" = ["sched"] "sched.rule-cron" = ["sched"] diff --git a/examples/sched.rs b/examples/sched.rs new file mode 100644 index 0000000..3e7e800 --- /dev/null +++ b/examples/sched.rs @@ -0,0 +1,31 @@ +use lool::sched::{recur, ruleset, Scheduler}; + +fn my_action() { + let now = chrono::Local::now(); + println!("I'm running at {}", now.format("%Y-%m-%d %H:%M:%S")); +} + +fn main() { + let mut sched = Scheduler::new(); + + let handler = sched.schedule("test-task", my_action, recur(ruleset().at_second(0))); + + std::thread::sleep(std::time::Duration::from_secs(1)); + + loop { + { + let task = handler.task.lock().unwrap(); + + let is_running = task.is_running(); + let last_run = task.get_last_run(); + let name = task.name(); + + println!( + "task {} |--> is running: {}, last run: {:?}", + name, is_running, last_run + ); + } + + std::thread::sleep(std::time::Duration::from_secs(2)); + } +} diff --git a/lib/sched/README.md b/lib/sched/README.md index c963b43..2a04e73 100644 --- a/lib/sched/README.md +++ b/lib/sched/README.md @@ -5,7 +5,9 @@

-lool » sched is a utility library that provides a way to schedule tasks in various ways. Supports std::thread and the tokio runtime (as a feature flag). +lool » sched is a utility library that provides a way to schedule tasks in +various ways. Supports std::thread and the tokio runtime +(as a feature flag).

@@ -26,7 +28,8 @@ cargo add lool --registry=lugit --features sched {sub-feature} ### has subfeatures  sched.tokio -Enables the `tokio` runtime support, replacing the default behaviour, which implies a `std::thread` pool to run the tasks. +Enables the `tokio` runtime support, replacing the default behaviour, which implies a `std::thread` +pool to run the tasks. > [!WARNING] > diff --git a/lib/sched/mod.rs b/lib/sched/mod.rs index 3f3332e..e5ff6ba 100644 --- a/lib/sched/mod.rs +++ b/lib/sched/mod.rs @@ -1,4 +1,6 @@ mod rules; +mod scheduler; + pub mod utils; -pub use rules::*; +pub use {rules::*, scheduler::*}; diff --git a/lib/sched/rules.rs b/lib/sched/rules.rs index 5a38019..4335413 100644 --- a/lib/sched/rules.rs +++ b/lib/sched/rules.rs @@ -13,6 +13,7 @@ use chrono::{DateTime, Local}; /// - `Once`: runs only at a specific `chrono::DateTime` /// - `Repeat`: runs at specific intervals defined by a `RecurrenceRule` /// - `Cron`: runs at specific intervals defined by a cron expression +#[derive(Clone)] pub enum SchedulingRule { /// 🧉 » a scheduling rule that makes the task run only once at a specific `chrono::DateTime` Once(chrono::DateTime), @@ -54,3 +55,20 @@ impl SchedulingRule { } } } + +/// 🧉 » create a new `SchedulingRule` that runs at specific intervals defined by a cron expression +#[cfg(feature = "sched.rule-cron")] +pub fn cron(cron: &str) -> SchedulingRule { + SchedulingRule::Cron(cron.to_string()) +} + +/// 🧉 » create a new `SchedulingRule` that runs at specific intervals defined by a `RecurrenceRule` +#[cfg(feature = "sched.rule-recurrence")] +pub fn recur(rule: &RecurrenceRuleSet) -> SchedulingRule { + SchedulingRule::Repeat(rule.clone()) +} + +/// 🧉 » create a new `SchedulingRule` that runs only once at a specific `chrono::DateTime` +pub fn once(datetime: DateTime) -> SchedulingRule { + SchedulingRule::Once(datetime) +} diff --git a/lib/sched/rules/recurrent/rule_unit.rs b/lib/sched/rules/recurrent/rule_unit.rs index 9d97354..031de6b 100644 --- a/lib/sched/rules/recurrent/rule_unit.rs +++ b/lib/sched/rules/recurrent/rule_unit.rs @@ -3,6 +3,7 @@ use num_traits::PrimInt; /// 🧉 » a recurrence rule unit /// /// represents a single rule unit that can be used to match a value +#[derive(Clone)] pub enum Rule where T: PrimInt, diff --git a/lib/sched/rules/recurrent/ruleset.rs b/lib/sched/rules/recurrent/ruleset.rs index 6d79ab9..5dfcb48 100644 --- a/lib/sched/rules/recurrent/ruleset.rs +++ b/lib/sched/rules/recurrent/ruleset.rs @@ -11,6 +11,7 @@ use { /// sets rules that define a certain recurrence behavior /// /// use the builder pattern to create a new `RecurrenceRuleSet` +#[derive(Clone)] pub struct RecurrenceRuleSet { /// second of the minute (0..59) second: Option>, diff --git a/lib/sched/scheduler.rs b/lib/sched/scheduler.rs new file mode 100644 index 0000000..1f86c3d --- /dev/null +++ b/lib/sched/scheduler.rs @@ -0,0 +1,222 @@ +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, Mutex, +}; + +use { + super::SchedulingRule, + crate::utils::threads::threadpool::ThreadPool, + chrono::{DateTime, Local}, +}; + +// TODO: add logging (always as debug) + +type Action = Box; + +/// 🧉 » a scheduled task +/// +/// this structs represents a task that has been scheduled in the scheduler. +/// +/// this is returned by the `Scheduler::schedule` method, and can be used to check and control the +/// status of the task. +pub struct ScheduledTask { + #[allow(dead_code)] + index: usize, + name: String, + action: Action, + rules: Vec, + is_running: AtomicBool, + last_run: Arc>>>, // TODO: remaining limits +} + +impl ScheduledTask { + fn run(&mut self) { + let action = self.action.as_mut(); + action(); + } + + pub fn get_last_run(&self) -> Option> { + let last_run_lock = self.last_run.lock().unwrap(); + last_run_lock.as_ref().cloned() + } + + pub fn is_running(&self) -> bool { + self.is_running.load(Ordering::Relaxed) + } + + pub fn name(&self) -> &str { + &self.name + } +} + +/// 🧉 » a task scheduler. +/// +/// this struct is responsible for scheduling tasks to be executed at specific times, depending on +/// the rules provided for each task. +/// +/// Each task can have n rules, and the task will be executed when any of the rules is met. +pub struct Scheduler { + pool: ThreadPool, + tasks: Vec>>, +} + +impl Default for Scheduler { + fn default() -> Self { + Self::new() + } +} + +impl Scheduler { + /// 🧉 » create a new scheduler + /// + /// default constructor, sets the internal thread pool to have 5 threads at most. + pub fn new() -> Self { + Self { + tasks: vec![], + pool: ThreadPool::create(5).unwrap(), + } + } + + /// 🧉 » create a new scheduler + /// + /// creates a new scheduler, just like `Scheduler::new`, but with a specific capacity for the + /// internal thread pool. + pub fn with_capacity(capacity: usize) -> Self { + Self { + tasks: vec![], + pool: ThreadPool::create(capacity).unwrap(), + } + } + + /// 🧉 » schedule a task + /// + /// schedules a task to be executed at times determined by the provided rules. + pub fn schedule(&mut self, name: &str, action: F, rules: SchedulingRule) -> TaskHandler + where + F: FnMut() + Send + Sync + 'static, + { + self.schedule_many_rules(name, action, vec![rules]) + } + + /// 🧉 » schedule a task + /// + /// schedules a task to be executed at times determined by the provided rules. + pub fn schedule_many_rules( + &mut self, + name: &str, + action: F, + rules: Vec, + ) -> TaskHandler + where + F: FnMut() + Send + Sync + 'static, + { + let index = self.tasks.len(); + + let task = Arc::new(Mutex::new(ScheduledTask { + index, + name: name.to_string(), + action: Box::new(action), + rules, + is_running: AtomicBool::new(false), + last_run: Arc::new(Mutex::new(None)), + })); + + self.tasks.push(task.clone()); + + run_in_pool(task.clone(), &self.pool); + + TaskHandler { task: task.clone() } + } +} + +/// 🧉 » task handler +/// +/// returned by the `Scheduler::schedule` method, +/// this struct can be used to check and control +/// the status of the task. +pub struct TaskHandler { + // HACK: holding the task in the TaskHandler is a temporal hack + // TaskHandler should hold Atomic references to the important parts of the task + // instead. e.g. is_running, last_run, etc. + // the problem is with last_run, as its a DateTime, and not a primitive type + // we could get around this by instead of holding the DateTime, holding the i64 + // value (unix timestamp) and then converting it to a DateTime when needed. + pub task: Arc>, +} + +/// **main function to run the task in the thread pool** +/// +/// it spawns a new job in the thread pool to run the task until the task is no longer scheduled to +/// run. +fn run_in_pool(task_mutex: Arc>, pool: &ThreadPool) { + pool.execute(move || { + let (mut maybe_next_run, name) = { + let task = task_mutex.lock().unwrap(); + let rules = &task.rules; + + (get_next_run_time(rules, None), task.name.clone()) + }; + + while let Some(run_date) = maybe_next_run { + let now = Local::now(); + if run_date > now { + // if the next run is in the future, go to bed until then + let sleep_until = run_date - now; + println!( + "task {} will run in {} seconds", + name, + sleep_until.num_seconds() + ); + std::thread::sleep(sleep_until.to_std().unwrap()); + } else { + // if the next run is in the past, run the task immediately, probably missed the + // run time for a few nanos + println!("task will run in 0 seconds"); + } + + let mut task = task_mutex.lock().unwrap(); + + task.last_run = Arc::new(Mutex::new(Some(run_date))); + task.is_running.store(true, Ordering::SeqCst); + task.run(); + task.is_running.store(false, Ordering::SeqCst); + maybe_next_run = get_next_run_time(&task.rules, Some(run_date)); + } + }); +} + +/// **get next run time** +/// +/// this function takes a list of scheduling rules and a base time, and returns the next time the +/// task should run. +/// +/// to determine the next run time, it iterates over the list of rules and calculates the next run +/// time for each of them, returning the earliest of them all. +fn get_next_run_time( + rules: &Vec, + from: Option>, +) -> Option> { + let mut next_run_so_far: Option> = None; + + let base = if let Some(from) = from { + from + } else { + Local::now() + }; + + for rule in rules { + let rule_next_run = rule.next_from(base); + + if let Some(next_run) = rule_next_run { + if let Some(d) = next_run_so_far { + if next_run < d { + next_run_so_far = Some(next_run); + } + } else { + next_run_so_far = Some(next_run); + } + } + } + + next_run_so_far +} diff --git a/🧳 lool.code-workspace b/🧳 lool.code-workspace index d176ddc..8e7a9b9 100644 --- a/🧳 lool.code-workspace +++ b/🧳 lool.code-workspace @@ -11,10 +11,9 @@ }, "lucodear-icons.activeIconPack": "rust_ferris", "lucodear-icons.folders.associations": { - ".cargo": "rust", "stylize": "theme", "ruleset": "rules", - "recurrent": "generator" + "recurrent": "generator", }, "lucodear-icons.files.associations": {