From 5a25995be5d5f81fd792700dfdcee795e3b24c8e Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Fri, 6 Sep 2024 22:29:32 -0400 Subject: [PATCH] refactor job so we can use multiple threads --- Cargo.lock | 1 + crates/service/job/analytics_job/Cargo.toml | 1 + .../job/analytics_job/src/job_state.rs | 2 + crates/service/job/analytics_job/src/main.rs | 43 +++++++++++++------ .../calculate_old_model_analytics_loop.rs} | 2 +- .../calculate_old_model_analytics/mod.rs | 1 + crates/service/job/es-update-job/Cargo.toml | 2 +- 7 files changed, 36 insertions(+), 16 deletions(-) rename crates/service/job/analytics_job/src/{main_loop.rs => tasks/calculate_old_model_analytics/calculate_old_model_analytics_loop.rs} (87%) diff --git a/Cargo.lock b/Cargo.lock index 0681e4cc95..d513477f8c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -473,6 +473,7 @@ dependencies = [ "easyenv", "enums", "errors", + "futures 0.3.29", "hostname 0.4.0", "log", "my-workspace-hack", diff --git a/crates/service/job/analytics_job/Cargo.toml b/crates/service/job/analytics_job/Cargo.toml index 41611410e1..e659118262 100644 --- a/crates/service/job/analytics_job/Cargo.toml +++ b/crates/service/job/analytics_job/Cargo.toml @@ -42,6 +42,7 @@ sqlx = { version = "0.7.2", features = [ "mysql", "runtime-tokio", "tls-rustls", # Runtime # NB: We switched to Tokio from `async-std` for broader ecosystem support. tokio = { version = "1.38.0", features = ["macros"] } +futures = "0.3.29" # Serialization serde = "1.0.125" diff --git a/crates/service/job/analytics_job/src/job_state.rs b/crates/service/job/analytics_job/src/job_state.rs index 3c53b14075..ad7c5bc273 100644 --- a/crates/service/job/analytics_job/src/job_state.rs +++ b/crates/service/job/analytics_job/src/job_state.rs @@ -1,5 +1,6 @@ use sqlx::MySqlPool; +#[derive(Clone)] pub struct JobState { pub mysql_pool: MySqlPool, @@ -7,6 +8,7 @@ pub struct JobState { } /// Use sleep to not overload the database. +#[derive(Clone)] pub struct SleepConfigs { // How long to wait between individual "jobs". pub between_job_wait_millis: u64, diff --git a/crates/service/job/analytics_job/src/main.rs b/crates/service/job/analytics_job/src/main.rs index ec7543084c..6bff5514ba 100644 --- a/crates/service/job/analytics_job/src/main.rs +++ b/crates/service/job/analytics_job/src/main.rs @@ -17,6 +17,7 @@ //#![forbid(warnings)] use log::info; +use sqlx::{MySql, Pool}; use sqlx::mysql::MySqlPoolOptions; use config::shared_constants::DEFAULT_MYSQL_CONNECTION_STRING; @@ -24,11 +25,10 @@ use config::shared_constants::DEFAULT_RUST_LOG; use errors::AnyhowResult; use crate::job_state::{JobState, SleepConfigs}; -use crate::main_loop::main_loop; +use crate::tasks::calculate_old_model_analytics::calculate_old_model_analytics_loop::calculate_old_model_analytics_loop; pub mod job_state; pub mod tasks; -pub mod main_loop; #[tokio::main] async fn main() -> AnyhowResult<()> { @@ -46,17 +46,7 @@ async fn main() -> AnyhowResult<()> { info!("Hostname: {}", &server_hostname); - let db_connection_string = - easyenv::get_env_string_or_default( - "MYSQL_URL", - DEFAULT_MYSQL_CONNECTION_STRING); - - info!("Connecting to database..."); - - let mysql_pool = MySqlPoolOptions::new() - .max_connections(5) - .connect(&db_connection_string) - .await?; + let mysql_pool = get_mysql_pool().await?; let job_state = JobState { mysql_pool, @@ -68,7 +58,32 @@ async fn main() -> AnyhowResult<()> { }, }; - let _r = main_loop(job_state).await; + let job_state_1 = job_state.clone(); + + let handle_1 = tokio::task::spawn(async move { + let _r = calculate_old_model_analytics_loop(job_state_1).await; + }); + + futures::future::join_all([ + handle_1, + ]).await; + Ok(()) } + +async fn get_mysql_pool() -> AnyhowResult> { + info!("Connecting to MySQL database..."); + + let db_connection_string = + easyenv::get_env_string_or_default( + "MYSQL_URL", + DEFAULT_MYSQL_CONNECTION_STRING); + + let mysql_pool = MySqlPoolOptions::new() + .max_connections(5) + .connect(&db_connection_string) + .await?; + + Ok(mysql_pool) +} diff --git a/crates/service/job/analytics_job/src/main_loop.rs b/crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/calculate_old_model_analytics_loop.rs similarity index 87% rename from crates/service/job/analytics_job/src/main_loop.rs rename to crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/calculate_old_model_analytics_loop.rs index 6a37a70dea..54656c5098 100644 --- a/crates/service/job/analytics_job/src/main_loop.rs +++ b/crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/calculate_old_model_analytics_loop.rs @@ -7,7 +7,7 @@ use errors::AnyhowResult; use crate::job_state::JobState; use crate::tasks::calculate_old_model_analytics::calculate_old_model_analytics::calculate_old_model_analytics; -pub async fn main_loop(job_state: JobState) -> AnyhowResult<()> { +pub async fn calculate_old_model_analytics_loop(job_state: JobState) -> AnyhowResult<()> { loop { info!("Begin analytics job batch."); diff --git a/crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/mod.rs b/crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/mod.rs index bf6bd303e6..143285f729 100644 --- a/crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/mod.rs +++ b/crates/service/job/analytics_job/src/tasks/calculate_old_model_analytics/mod.rs @@ -1 +1,2 @@ pub mod calculate_old_model_analytics; +pub mod calculate_old_model_analytics_loop; diff --git a/crates/service/job/es-update-job/Cargo.toml b/crates/service/job/es-update-job/Cargo.toml index 3b7dd91ca7..bd666f2a84 100644 --- a/crates/service/job/es-update-job/Cargo.toml +++ b/crates/service/job/es-update-job/Cargo.toml @@ -49,9 +49,9 @@ elasticsearch = "8.5.0-alpha.1" # Runtime # NB: We switched to Tokio from `async-std` for broader ecosystem support. tokio = { version = "1.38.0", features = ["macros"] } +futures = "0.3.29" # Serialization serde = "1.0.125" serde_derive = "1.0.125" serde_json = "1.0.108" -futures = "0.3.29"