From 7b84a63a14877d15bf98b930c440ef301dcf7ff9 Mon Sep 17 00:00:00 2001 From: Lumither Tao Date: Wed, 12 Mar 2025 18:34:49 -0700 Subject: [PATCH] feat(backend::scheduler) deadlock spotted --- .../check_update/error.rs | 0 .../check_update/index.rs | 20 +-- .../blog/components/check_update/mod.rs | 78 +++++++++ .../blog/components/check_update/utils.rs | 81 +++++++++ .../server/src/modules/blog/components/mod.rs | 2 + .../src/modules/blog/components/rag/mod.rs | 0 .../src/modules/blog/cron/check_update/mod.rs | 148 ---------------- .../modules/blog/cron/check_update/utils.rs | 88 ---------- backend/server/src/modules/blog/cron/mod.rs | 1 - backend/server/src/modules/blog/mod.rs | 17 +- backend/server/src/modules/mod.rs | 12 +- backend/server/src/scheduler/mod.rs | 158 +++++++++--------- backend/server/src/scheduler/runner.rs | 88 ++++++---- backend/server/src/scheduler/sleeper.rs | 101 +++++------ backend/server/src/utils/git.rs | 9 +- 15 files changed, 375 insertions(+), 428 deletions(-) rename backend/server/src/modules/blog/{cron => components}/check_update/error.rs (100%) rename backend/server/src/modules/blog/{cron => components}/check_update/index.rs (97%) create mode 100644 backend/server/src/modules/blog/components/check_update/mod.rs create mode 100644 backend/server/src/modules/blog/components/check_update/utils.rs create mode 100644 backend/server/src/modules/blog/components/mod.rs create mode 100644 backend/server/src/modules/blog/components/rag/mod.rs delete mode 100644 backend/server/src/modules/blog/cron/check_update/mod.rs delete mode 100644 backend/server/src/modules/blog/cron/check_update/utils.rs delete mode 100644 backend/server/src/modules/blog/cron/mod.rs diff --git a/backend/server/src/modules/blog/cron/check_update/error.rs b/backend/server/src/modules/blog/components/check_update/error.rs similarity index 100% rename from backend/server/src/modules/blog/cron/check_update/error.rs rename to backend/server/src/modules/blog/components/check_update/error.rs diff --git a/backend/server/src/modules/blog/cron/check_update/index.rs b/backend/server/src/modules/blog/components/check_update/index.rs similarity index 97% rename from backend/server/src/modules/blog/cron/check_update/index.rs rename to backend/server/src/modules/blog/components/check_update/index.rs index fabd478..9cf590c 100644 --- a/backend/server/src/modules/blog/cron/check_update/index.rs +++ b/backend/server/src/modules/blog/components/check_update/index.rs @@ -1,12 +1,10 @@ use std::error::Error; use std::path::Path; -use crate::{ - modules::blog::cron::check_update::utils::get_md_file_basename, - modules::blog::cron::check_update::{error::PostIdxError, utils::parse_date}, -}; use markdown::MdFile; +use crate::modules::blog::components::check_update::utils::get_md_file_basename; +use crate::modules::blog::components::check_update::{error::PostIdxError, utils::parse_date}; use sqlx::{query, PgPool, Pool, Postgres, Row}; use tokio::runtime::Runtime; use tracing::{info, warn}; @@ -18,7 +16,7 @@ async fn base(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> return Err(PostIdxError::MissingField { field: "title".to_string(), filename: post.filename.clone(), - }) + }); } Some(title) => title.as_str().unwrap_or_else(|| { warn!("invalid title for `{}`", &post.filename); @@ -41,7 +39,7 @@ async fn base(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> return Err(PostIdxError::MissingField { field: "category".to_string(), filename: post.filename.clone(), - }) + }); } Some(title) => title.as_str().unwrap_or_else(|| { warn!("invalid category for `{}`", &post.filename); @@ -58,7 +56,7 @@ async fn base(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> err_field: last_update.to_string(), filename: post.filename.clone(), msg: e.to_string(), - }) + }); } } } else { @@ -70,7 +68,7 @@ async fn base(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> return Err(PostIdxError::MissingField { field: "date".to_string(), filename: post.filename.clone(), - }) + }); } Some(first_update) => match parse_date(first_update.as_str().unwrap()) { Ok(date) => date, @@ -80,7 +78,7 @@ async fn base(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> err_field: first_update.to_string(), filename: post.filename.clone(), msg: e.to_string(), - }) + }); } }, }; @@ -137,7 +135,7 @@ async fn tag(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> return Err(PostIdxError::MissingField { field: "tags".to_string(), filename: post.filename.clone(), - }) + }); } Some(val) => val, }; @@ -173,7 +171,7 @@ async fn tag(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> db_table: "Tag".to_string(), id: post_id.to_string(), msg: e.to_string(), - }) + }); } } } diff --git a/backend/server/src/modules/blog/components/check_update/mod.rs b/backend/server/src/modules/blog/components/check_update/mod.rs new file mode 100644 index 0000000..6e9f5dd --- /dev/null +++ b/backend/server/src/modules/blog/components/check_update/mod.rs @@ -0,0 +1,78 @@ +use crate::constants::config::general::{default_value, var_name}; +use macros::panic_with_log; +use std::sync::Arc; +use std::{ + env, + path::{Path, PathBuf}, +}; + +use crate::modules::blog::components::check_update::{ + // index::drop_index, + // utils::{expand_path, index_files, search_md}, + utils::expand_path, +}; +use crate::scheduler::task_func::AsyncTaskFunc; +use sqlx::PgPool; +use tracing::{info, warn, Level}; + +// mod error; +// mod index; +mod utils; + +#[derive(Debug)] +struct Args { + git_work_dir: PathBuf, + git_url: String, + git_remote_name: String, + git_remote_branch: String, +} + +pub fn task(db_connection: PgPool) -> impl AsyncTaskFunc { + // init + let git_work_dir = { + let path = if let Ok(default_log_path) = env::var(var_name::WORK_DIR) { + expand_path(default_log_path) + } else { + let fallback_path = expand_path(default_value::WORK_DIR.to_string()); + warn!( + "unset env var `{}`, using default path: {}/blog_git", + var_name::WORK_DIR, + fallback_path + ); + fallback_path + }; + Path::new(&path).join("blog_git") + }; + + // clone & check update + + let git_url = match env::var(var_name::BLOG_GIT_URL) { + Ok(url) => url, + Err(e) => { + panic_with_log!( + Level::ERROR, + "failed to read `{}`: {}", + var_name::BLOG_GIT_URL, + e + ) + } + }; + let git_remote_name = env::var(var_name::BLOG_GIT_REMOTE_NAME) + .unwrap_or(default_value::BLOG_GIT_REMOTE_NAME.to_string()); + let git_remote_branch = env::var(var_name::BLOG_GIT_REMOTE_BRANCH) + .unwrap_or(default_value::BLOG_GIT_REMOTE_BRANCH.to_string()); + + let arc_args = Arc::new(Args { + git_work_dir, + git_url, + git_remote_name, + git_remote_branch, + }); + + move || Box::pin(test(arc_args.clone())) +} + +async fn test(args: Arc) { + info!("test"); + dbg!(&args); +} diff --git a/backend/server/src/modules/blog/components/check_update/utils.rs b/backend/server/src/modules/blog/components/check_update/utils.rs new file mode 100644 index 0000000..9f00fe5 --- /dev/null +++ b/backend/server/src/modules/blog/components/check_update/utils.rs @@ -0,0 +1,81 @@ +use crate::constants::{ + config::general::{default_value, var_name}, + time::DEFAULT_DATE_FORMAT, +}; + +// use crate::modules::blog::components::check_update::index::index_file; +use chrono::{DateTime, FixedOffset, ParseResult}; +use dirs::home_dir; +use tracing::warn; + +// pub fn search_md(entry: PathBuf) -> Vec { +// if entry.ends_with(".git") { +// return vec![]; +// } +// if entry.is_dir() { +// if let Ok(files) = entry.read_dir() { +// files +// .flat_map(|file| { +// let file_path = file.unwrap().path(); +// search_md(file_path) +// }) +// .collect() +// } else { +// vec![] +// } +// } else if entry.extension().is_some_and(|ext| ext == "md") { +// vec![entry] +// } else { +// vec![] +// } +// } + +// pub fn index_files(files: &[PathBuf], db_conn: &Pool) { +// let rt = Runtime::new().unwrap(); +// rt.block_on(async { +// let files_clone = files.to_owned(); +// let handlers = files_clone +// .into_iter() +// .map(|file: PathBuf| { +// let conn_clone = db_conn.clone(); +// spawn(async move { +// if let Err(e) = index_file(&file, &conn_clone).await { +// error!("failed to index markdown file `{}`: {}", file.display(), e); +// } +// }) +// }) +// .collect::>(); +// +// block_on(join_all(handlers)); +// }); +// } + +pub fn expand_path(path: String) -> String { + if path.starts_with('~') { + home_dir() + .map(|dir| { + dir.join(path.strip_prefix("~/").unwrap()) + .into_os_string() + .into_string() + .unwrap() + }) + .unwrap_or_else(|| { + warn!( + "failed to parse env var `{}`, using default value `{}`", + var_name::WORK_DIR, + default_value::WORK_DIR + ); + default_value::WORK_DIR.to_string() + }) + } else { + path + } +} + +pub fn parse_date(date_str: &str) -> ParseResult> { + DateTime::parse_from_str(format!("{}T120000", date_str).as_str(), DEFAULT_DATE_FORMAT) +} + +// pub fn get_md_file_basename(f_name: &str) -> String { +// f_name.replace(".md", "") +// } diff --git a/backend/server/src/modules/blog/components/mod.rs b/backend/server/src/modules/blog/components/mod.rs new file mode 100644 index 0000000..2683687 --- /dev/null +++ b/backend/server/src/modules/blog/components/mod.rs @@ -0,0 +1,2 @@ +pub mod check_update; +mod rag; diff --git a/backend/server/src/modules/blog/components/rag/mod.rs b/backend/server/src/modules/blog/components/rag/mod.rs new file mode 100644 index 0000000..e69de29 diff --git a/backend/server/src/modules/blog/cron/check_update/mod.rs b/backend/server/src/modules/blog/cron/check_update/mod.rs deleted file mode 100644 index bc02d9c..0000000 --- a/backend/server/src/modules/blog/cron/check_update/mod.rs +++ /dev/null @@ -1,148 +0,0 @@ -use std::{ - env, - fs::remove_dir_all, - path::{Path, PathBuf}, -}; - -use crate::{ - constants::config::general::{default_value, var_name}, - modules::blog::cron::check_update::{ - index::drop_index, - utils::{expand_path, get_md_file_basename, index_files, search_md}, - }, - utils::git, -}; -use macros::panic_with_log; - -use git2::{Delta, Repository}; -use sqlx::{PgPool, Pool, Postgres}; -use tracing::{error, info, warn, Level}; - -mod error; -mod index; -mod utils; - -pub fn task(db_connection: &PgPool) { - // init - let git_work_dir = { - let path = if let Ok(default_log_path) = env::var(var_name::WORK_DIR) { - expand_path(default_log_path) - } else { - let fallback_path = expand_path(default_value::WORK_DIR.to_string()); - warn!( - "unset env var `{}`, using default path: {}/blog_git", - var_name::WORK_DIR, - fallback_path - ); - fallback_path - }; - Path::new(&path).join("blog_git") - }; - - // clone & check update - - let git_url = match env::var(var_name::BLOG_GIT_URL) { - Ok(url) => url, - Err(e) => { - panic_with_log!( - Level::ERROR, - "failed to read `{}`: {}", - var_name::BLOG_GIT_URL, - e - ) - } - }; - let git_remote_name = env::var(var_name::BLOG_GIT_REMOTE_NAME) - .unwrap_or(default_value::BLOG_GIT_REMOTE_NAME.to_string()); - let git_remote_branch = env::var(var_name::BLOG_GIT_REMOTE_BRANCH) - .unwrap_or(default_value::BLOG_GIT_REMOTE_BRANCH.to_string()); - - match Repository::open(&git_work_dir) { - Ok(repo) => { - // prevent git remote url mismatch (clean cache) - match repo.find_remote(&git_remote_name) { - Ok(remote) => { - if remote.url().is_some_and(|url| url != git_url) { - remove_dir_all(&git_work_dir).unwrap(); - warn!("blog content git({}) url mismatch, reinitialized", &git_url); - return; - } - } - Err(e) => { - remove_dir_all(&git_work_dir).unwrap(); - error!( - "unexpected error on indexing blog content git({}), reinitialized: {}", - &git_url, e - ); - } - } - - match git::sync(&git_remote_name, &git_remote_branch, &repo) { - Ok(updates) => { - if !updates.is_empty() { - // incremental update - info!( - "blog content git({}) update detected: {:?}", - &git_url, updates - ); - - let mut update_md: Vec = vec![]; - - for update in updates { - if [Delta::Deleted, Delta::Untracked].contains(&update.status) { - if let Some(path) = &update.old_path { - if drop_index( - &get_md_file_basename( - path.file_name().unwrap().to_str().unwrap(), - ), - db_connection, - ) - .is_ok() - { - info!( - "blog content git({}) file `{}` is detected as deleted/untracked, index dropped", - &git_url, - &path.display() - ); - } else { - error!( - "blog content git({}) file `{}` is detected as deleted/untracked, failed to drop index", - &git_url, - &path.display() - ) - } - } - } else if let Some(path) = &update.new_path { - if path.components().any(|comp| comp.as_os_str() == "posts") - && path.extension().is_some_and(|ext| ext == "md") - { - update_md.push(path.to_path_buf()); - } - } - } - - index_files(&update_md, db_connection); - } - } - Err(e) => { - error!( - "failed to sync blog content git({}) updates: {}", - &git_url, e - ); - } - } - } - Err(_) => { - match Repository::clone(&git_url, &git_work_dir) { - Ok(_repo) => { - // full repo index - let files = search_md(git_work_dir.join("posts")); - index_files(&files, db_connection); - } - Err(e) => { - error!("failed to clone blog content git({}): {}", &git_url, e); - } - } - } - }; -} diff --git a/backend/server/src/modules/blog/cron/check_update/utils.rs b/backend/server/src/modules/blog/cron/check_update/utils.rs deleted file mode 100644 index 85086ca..0000000 --- a/backend/server/src/modules/blog/cron/check_update/utils.rs +++ /dev/null @@ -1,88 +0,0 @@ -use std::path::PathBuf; - -use crate::{ - constants::{ - config::general::{default_value, var_name}, - time::DEFAULT_DATE_FORMAT, - }, - modules::blog::cron::check_update::index::index_file, -}; - -use chrono::{DateTime, FixedOffset, ParseResult}; -use dirs::home_dir; -use futures::{executor::block_on, future::join_all}; -use sqlx::{Pool, Postgres}; -use tokio::{runtime::Runtime, spawn}; -use tracing::{error, warn}; - -pub fn search_md(entry: PathBuf) -> Vec { - if entry.ends_with(".git") { - return vec![]; - } - if entry.is_dir() { - if let Ok(files) = entry.read_dir() { - files - .flat_map(|file| { - let file_path = file.unwrap().path(); - search_md(file_path) - }) - .collect() - } else { - vec![] - } - } else if entry.extension().is_some_and(|ext| ext == "md") { - vec![entry] - } else { - vec![] - } -} - -pub fn index_files(files: &[PathBuf], db_conn: &Pool) { - let rt = Runtime::new().unwrap(); - rt.block_on(async { - let files_clone = files.to_owned(); - let handlers = files_clone - .into_iter() - .map(|file: PathBuf| { - let conn_clone = db_conn.clone(); - spawn(async move { - if let Err(e) = index_file(&file, &conn_clone).await { - error!("failed to index markdown file `{}`: {}", file.display(), e); - } - }) - }) - .collect::>(); - - block_on(join_all(handlers)); - }); -} - -pub fn expand_path(path: String) -> String { - if path.starts_with('~') { - home_dir() - .map(|dir| { - dir.join(path.strip_prefix("~/").unwrap()) - .into_os_string() - .into_string() - .unwrap() - }) - .unwrap_or_else(|| { - warn!( - "failed to parse env var `{}`, using default value `{}`", - var_name::WORK_DIR, - default_value::WORK_DIR - ); - default_value::WORK_DIR.to_string() - }) - } else { - path - } -} - -pub fn parse_date(date_str: &str) -> ParseResult> { - DateTime::parse_from_str(format!("{}T120000", date_str).as_str(), DEFAULT_DATE_FORMAT) -} - -pub fn get_md_file_basename(f_name: &str) -> String { - f_name.replace(".md", "") -} diff --git a/backend/server/src/modules/blog/cron/mod.rs b/backend/server/src/modules/blog/cron/mod.rs deleted file mode 100644 index f5d8ad7..0000000 --- a/backend/server/src/modules/blog/cron/mod.rs +++ /dev/null @@ -1 +0,0 @@ -// pub mod check_update; diff --git a/backend/server/src/modules/blog/mod.rs b/backend/server/src/modules/blog/mod.rs index 3f17c69..213e816 100644 --- a/backend/server/src/modules/blog/mod.rs +++ b/backend/server/src/modules/blog/mod.rs @@ -1,5 +1,5 @@ mod api; -mod cron; +mod components; use crate::{ modules::{ @@ -13,9 +13,12 @@ use crate::{ }, Module, }, - scheduler::tasks::{async_basic::AsyncBasic, CronTask}, + scheduler::tasks::CronTask, }; +use crate::modules::blog::components::check_update; +use crate::scheduler::tasks::async_basic::AsyncBasic; +use crate::scheduler::tasks::basic::BasicTask; use axum::{routing::get, Router}; use sqlx::PgPool; @@ -42,12 +45,12 @@ impl Module for Blog { } fn get_cron_tasks(&self) -> Vec> { - // let db_handler = self.db_handler.clone(); + let db_handler = self.db_handler.clone(); vec![ - // BasicTask::new(move || check_update::task(&db_handler), "0/15 * * * * *") - // .unwrap() - // .to_task(), - AsyncBasic::new(|| Box::pin(async { println!("test") }), "* * * * * *") + AsyncBasic::new(check_update::task(db_handler.clone()), "* * * * * *") + .unwrap() + .to_task(), + BasicTask::new(|| println!("test lol"), "* * * * * *") .unwrap() .to_task(), ] diff --git a/backend/server/src/modules/mod.rs b/backend/server/src/modules/mod.rs index ab1e7dd..3ff1c0a 100644 --- a/backend/server/src/modules/mod.rs +++ b/backend/server/src/modules/mod.rs @@ -1,7 +1,6 @@ pub mod blog; -use std::env; -use std::num::ParseIntError; +use std::{env, num::ParseIntError, thread}; use crate::{ constants::config::server::{default_value, var_name}, @@ -10,6 +9,7 @@ use crate::{ use macros::error_panic; use axum::Router; +use futures::executor::block_on; use tokio::net::TcpListener; use tracing::{info, warn}; @@ -68,11 +68,15 @@ impl ModuleTree { }; for module in &self.modules { - let _ = scheduler.insert_list(module.get_cron_tasks()); + let _ = scheduler.insert_list(module.get_cron_tasks()).await; app = app.nest(module.get_mount_point(), module.get_server_router()); } - scheduler.start().unwrap(); + let _ = thread::spawn(move || { + let fut = scheduler.start(); + block_on(fut).unwrap() + }) + .join(); 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 67f09a1..63913c3 100644 --- a/backend/server/src/scheduler/mod.rs +++ b/backend/server/src/scheduler/mod.rs @@ -1,6 +1,6 @@ use std::{ collections::{HashMap, HashSet}, - sync::{Arc, RwLock}, + sync::Arc, }; use crate::scheduler::{ @@ -9,12 +9,13 @@ use crate::scheduler::{ tasks::CronTask, }; +use tokio::sync::RwLock; use uuid::Uuid; pub mod error; mod runner; mod sleeper; -mod task_func; +pub mod task_func; pub mod tasks; #[derive(Default)] @@ -26,43 +27,42 @@ pub struct Scheduler { impl Scheduler { /// Start all tasks in the scheduler - pub fn start(&self) -> Result<(), SchedulerError> { - self.runner.run_all() + pub async fn start(&self) -> Result<(), SchedulerError> { + self.runner.run_all().await } /// Terminate all tasks in the scheduler - 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()); + pub async fn halt(&self) -> Result<(), SchedulerError> { + for id in self.running.read().await.iter() { + let func = &self.runner.terminate_remove(*id).await?; + self.running.write().await.remove(id); + self.parking_pool.write().await.insert(*id, func.clone()); } Ok(()) } /// Run a task with `id` - pub fn run(&self, id: &Uuid) -> Result<(), SchedulerError> { - if self.running.read()?.contains(id) { + pub async fn run(&self, id: &Uuid) -> Result<(), SchedulerError> { + if self.running.read().await.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) - .and_then(|_| self.runner.run_id(id)) + } else if self.parking_pool.read().await.contains_key(id) { + let func = self.parking_pool.write().await.remove(id).unwrap(); + self.running.write().await.insert(*id); + self.runner.add_with_id(func, id).await?; + self.runner.run_id(id).await } else { Err(IdNotFound) } } /// Stop a task with `id` - pub fn stop(&self, id: &Uuid) -> Result<(), SchedulerError> { - if self.running.read()?.contains(id) { - let func = self.runner.terminate_remove(*id)?; - self.running.write()?.remove(id); - self.parking_pool.write()?.insert(*id, func.clone()); + pub async fn stop(&self, id: &Uuid) -> Result<(), SchedulerError> { + if self.running.read().await.contains(id) { + let func = self.runner.terminate_remove(*id).await?; + self.running.write().await.remove(id); + self.parking_pool.write().await.insert(*id, func.clone()); Ok(()) - } else if self.parking_pool.read()?.get(id).is_some() { + } else if self.parking_pool.read().await.get(id).is_some() { Ok(()) } else { Err(IdNotFound) @@ -72,17 +72,21 @@ impl Scheduler { // pub fn now(&self, id: &Uuid) -> Result<(), SchedulerError> { // } - pub fn insert(&self, task: Box) -> Result { - let id = self.runner.add(task.into())?; - self.running.write()?.insert(id); + pub async fn insert(&self, task: Box) -> Result { + let id = self.runner.add(task.into()).await?; + self.running.write().await.insert(id); Ok(id) } - pub fn insert_list(&self, tasks: Vec>) -> Result, SchedulerError> { + pub async fn insert_list( + &self, + tasks: Vec>, + ) -> Result, SchedulerError> { let ids = self .runner - .add_list(tasks.into_iter().map(|t| t.into()).collect())?; - self.running.write()?.extend(&ids); + .add_list(tasks.into_iter().map(|t| t.into()).collect()) + .await?; + self.running.write().await.extend(&ids); Ok(ids) } @@ -91,51 +95,51 @@ impl Scheduler { } } -#[cfg(test)] -mod tests { - use std::{error::Error, thread, time::Duration}; - - 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); - 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)); - println!("Loop {} finished", i); - } - 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) - } -} +// #[cfg(test)] +// mod tests { +// use std::{error::Error, thread, time::Duration}; +// +// use crate::scheduler::{ +// tasks::{async_basic::AsyncBasic, basic::BasicTask}, +// Scheduler, +// }; +// +// use uuid::Uuid; +// +// async fn execution(scheduler: Scheduler, id: Uuid) -> Result<(), Box> { +// for i in 1..=2 { +// println!("Loop {} start", i); +// println!("Run {}", i); +// scheduler.run(&id).await?; +// 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)); +// println!("Loop {} finished", i); +// } +// Ok(()) +// } +// +// #[tokio::test] +// async fn test_run() -> Result<(), Box> { +// let scheduler = Scheduler::default(); +// let task = BasicTask::new(|| println!("test print"), "* * * * * *")?; +// let id = scheduler.insert(Box::new(task)).await?; +// scheduler.start().await?; +// execution(scheduler, id) +// } +// +// #[tokio::test] +// async 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)).await?; +// scheduler.start().await?; +// execution(scheduler, id) +// } +// } diff --git a/backend/server/src/scheduler/runner.rs b/backend/server/src/scheduler/runner.rs index b3ca00b..69dbe2c 100644 --- a/backend/server/src/scheduler/runner.rs +++ b/backend/server/src/scheduler/runner.rs @@ -1,20 +1,22 @@ -use std::{ - collections::HashMap, - sync::{ - atomic::{AtomicBool, Ordering}, Arc, - RwLock, - }, -}; - use crate::scheduler::{ error::{SchedulerError, SchedulerError::IdNotFound}, sleeper::SkippableSleeper, tasks::CronTask, }; +use futures::future; +use std::{ + collections::HashMap, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, +}; use tokio::runtime::Runtime; +use tokio::sync::RwLock; use uuid::Uuid; +#[derive(Debug)] struct TaskStatus { sleeper: RwLock, is_running: AtomicBool, @@ -40,13 +42,22 @@ impl TaskEntity { } pub async fn run_blocking(&self) { + dbg!("entered"); + dbg!("chkpt 1"); loop { - self.status.sleeper.read().unwrap().start(); + dbg!("chkpt 2"); + // deadlock spotted + self.status.sleeper.read().await.start().await; + dbg!("chkpt 3"); if self.status.expected_running.load(Ordering::SeqCst) { + dbg!("chkpt 4"); self.func.call().await; + dbg!("chkpt 5"); if let Some(next_execution) = self.func.get_next_execution() { - *(self.status.sleeper.write().unwrap()) = next_execution.into(); + dbg!("chkpt 6a"); + *(self.status.sleeper.write().await) = next_execution.into(); } else { + dbg!("chkpt 6b"); self.status.is_running.store(false, Ordering::SeqCst); break; } @@ -56,10 +67,10 @@ impl TaskEntity { } } - pub fn terminate(&self) { + pub async 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(); + self.status.sleeper.read().await.skip(); } } @@ -76,8 +87,8 @@ impl Runner { } } - pub fn run_all(&self) -> Result<(), SchedulerError> { - let list = &self.task_list.read()?; + pub async fn run_all(&self) -> Result<(), SchedulerError> { + let list = &self.task_list.read().await; self.rt.block_on(async { for (_, entity) in list.iter() { self.run_entity(entity.clone()).await @@ -86,8 +97,8 @@ impl Runner { Ok(()) } - pub fn run_id(&self, id: &Uuid) -> Result<(), SchedulerError> { - if let Some(entity) = self.task_list.read()?.get(id).cloned() { + pub async fn run_id(&self, id: &Uuid) -> Result<(), SchedulerError> { + if let Some(entity) = self.task_list.read().await.get(id).cloned() { self.rt.block_on(async { self.run_entity(entity).await }); Ok(()) } else { @@ -100,60 +111,67 @@ impl Runner { && entity.status.expected_running.load(Ordering::SeqCst) { entity.status.is_running.store(true, Ordering::SeqCst); - tokio::spawn({ + let fut = { let entity_clone = entity.clone(); async move { entity_clone.run_blocking().await } - }) - .await - .expect("tokio runtime error"); + }; + dbg!("spawn"); + tokio::spawn(fut); } } - pub fn add_with_id( + pub async fn add_with_id( &self, task: Arc>, id: &Uuid, ) -> Result<(), SchedulerError> { self.task_list - .write()? + .write() + .await .insert(*id, TaskEntity::new(task).into()); Ok(()) } - pub fn add(&self, task: Arc>) -> Result { + pub async fn add(&self, task: Arc>) -> Result { let id = Uuid::new_v4(); - self.add_with_id(task, &id)?; + self.add_with_id(task, &id).await?; Ok(id) } - pub fn add_list_with_id( + pub async fn add_list_with_id( &self, tasks: Vec<(&Uuid, Arc>)>, ) -> Result<(), SchedulerError> { - tasks + let fut = tasks .into_iter() - .try_for_each(|(id, task)| self.add_with_id(task, id)) + .map(async |(id, entity)| self.add_with_id(entity, id).await) + .collect::>(); + future::join_all(fut).await.into_iter().collect() } - pub fn add_list( + pub async fn add_list( &self, tasks: Vec>>, ) -> Result, SchedulerError> { - tasks + let fut = tasks .into_iter() - .map(|task| { + .map(async |task| { let id = Uuid::new_v4(); - self.add_with_id(task, &id).map(move |_| id) + self.add_with_id(task, &id).await.map(move |_| id) }) - .collect::, _>>() + .collect::>(); + future::join_all(fut).await.into_iter().collect() } - pub fn terminate_remove(&self, id: Uuid) -> Result>, SchedulerError> { - let mut list = self.task_list.write()?; + pub async fn terminate_remove( + &self, + id: Uuid, + ) -> Result>, SchedulerError> { + let mut list = self.task_list.write().await; match list.get(&id) { None => Err(IdNotFound), Some(task) => { - task.terminate(); + task.terminate().await; let func = list.remove(&id).unwrap().func.clone(); Ok(func.clone()) } diff --git a/backend/server/src/scheduler/sleeper.rs b/backend/server/src/scheduler/sleeper.rs index abc085a..93b913f 100644 --- a/backend/server/src/scheduler/sleeper.rs +++ b/backend/server/src/scheduler/sleeper.rs @@ -1,80 +1,81 @@ -use std::{ - sync::{Arc, Condvar, Mutex}, - thread, - time::Duration, -}; - use chrono::{DateTime, Utc}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::{sync::Arc, time::Duration}; +use tokio::sync::Notify; -#[derive(Clone, Debug, Default)] +#[derive(Debug, Clone)] pub struct SkippableSleeper { - pub duration: Duration, - state: Arc<(Mutex, Condvar)>, + duration: Duration, + skipped: Arc, + notify: Arc, } impl SkippableSleeper { pub fn new(duration: Duration) -> Self { - SkippableSleeper { + Self { duration, - state: Arc::new((Mutex::new(false), Condvar::new())), + skipped: Arc::new(AtomicBool::new(false)), + notify: Arc::new(Notify::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(); + pub async fn start(&self) { + if self.skipped.load(Ordering::Acquire) { + return; + } + + let sleep = tokio::time::sleep(self.duration); + let notify = self.notify.notified(); + + tokio::select! { + _ = sleep => {} + _ = notify => { + self.skipped.store(true, Ordering::Release); + } } } - // Skip the blocking delay pub fn skip(&self) { - let (lock, cvar) = &*self.state; - let mut received = lock.lock().unwrap(); - *received = true; - cvar.notify_one(); + self.notify.notify_waiters(); } } 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)) + let now = Utc::now(); + let duration = value.signed_duration_since(now); + let duration = duration.to_std().unwrap_or(Duration::ZERO); + Self::new(duration) } } #[cfg(test)] -mod test { - use crate::scheduler::sleeper::SkippableSleeper; - use std::thread; - use std::time::Duration; +mod tests { + use super::*; + use tokio::time::{sleep, 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!"); - } + #[tokio::test] + async fn skip_sleeper() { + let sleeper = SkippableSleeper::new(Duration::from_secs(10)); + let sleeper_clone = sleeper.clone(); + + let handle = tokio::spawn(async move { + println!("Sleeping..."); + sleeper_clone.start().await; + println!("Woke up!"); }); - println!("Start Sleeping..."); - thread::sleep(Duration::from_secs(4)); + + println!("Start waiting..."); + sleep(Duration::from_secs(4)).await; println!("Skipping sleep..."); sleeper.skip(); + + handle.await.unwrap(); + } + + #[tokio::test] + async fn normal_sleeper() { + let sleeper = SkippableSleeper::new(Duration::from_secs(4)); + sleeper.start().await; } } diff --git a/backend/server/src/utils/git.rs b/backend/server/src/utils/git.rs index 9c48fc0..94b00cc 100644 --- a/backend/server/src/utils/git.rs +++ b/backend/server/src/utils/git.rs @@ -3,11 +3,12 @@ /// https://github.com/rust-lang/git2-rs/commit/f3b87baed1e33d6c2d94fe1fa6aa6503a071d837 /// use git2::{ - AnnotatedCommit, AutotagOption, Delta, FetchOptions, Reference, Remote, Repository, build, + build, AnnotatedCommit, AutotagOption, Delta, FetchOptions, Reference, Remote, Repository, }; use std::path::PathBuf; use tracing::{error, warn}; + #[derive(Debug, Eq, PartialEq)] pub struct FileDelta { pub old_path: Option, @@ -15,7 +16,6 @@ pub struct FileDelta { pub status: Delta, } -#[allow(dead_code)] fn do_fetch<'a>( repo: &'a Repository, refs: &[&str], @@ -30,7 +30,6 @@ fn do_fetch<'a>( repo.reference_to_annotated_commit(&fetch_head) } -#[allow(dead_code)] fn fast_forward( repo: &Repository, lb: &mut Reference, @@ -48,7 +47,6 @@ fn fast_forward( Ok(()) } -#[allow(dead_code)] fn normal_merge( repo: &Repository, local: &AnnotatedCommit, @@ -85,7 +83,6 @@ fn normal_merge( Ok(()) } -#[allow(dead_code)] fn do_merge<'a>( repo: &'a Repository, remote_branch: &str, @@ -122,7 +119,6 @@ fn do_merge<'a>( calculate_diff(repo, &head_commit, &fetch_commit) } -#[allow(dead_code)] fn calculate_diff( repo: &Repository, local: &AnnotatedCommit, @@ -153,7 +149,6 @@ fn calculate_diff( Ok(updated_files.into_iter().collect()) } -#[allow(dead_code)] pub fn sync( remote_name: &str, remote_branch: &str,