From b5e226b977e6d24ebd28bc1e7c867cb4888f77b2 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 28 Sep 2024 15:43:06 +0200 Subject: [PATCH] fix: improve suspend_first behavior and frequency --- backend/Cargo.toml | 2 +- backend/plot.py | 54 ++- backend/src/monitor.rs | 2 + backend/windmill-common/Cargo.toml | 1 + backend/windmill-common/src/bench.rs | 213 +++++++++ backend/windmill-common/src/lib.rs | 13 + backend/windmill-queue/Cargo.toml | 2 +- backend/windmill-queue/src/jobs.rs | 40 +- backend/windmill-worker/Cargo.toml | 2 +- backend/windmill-worker/src/bench.rs | 111 ----- .../windmill-worker/src/dedicated_worker.rs | 2 +- backend/windmill-worker/src/lib.rs | 3 +- .../windmill-worker/src/result_processor.rs | 191 +++++++- backend/windmill-worker/src/worker.rs | 440 ++++++++---------- backend/windmill-worker/src/worker_flow.rs | 28 +- 15 files changed, 691 insertions(+), 413 deletions(-) create mode 100644 backend/windmill-common/src/bench.rs delete mode 100644 backend/windmill-worker/src/bench.rs diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 651a1f3f67..cb6b291e77 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -44,7 +44,7 @@ default = [] enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise", "windmill-git-sync/enterprise", "windmill-common/prometheus", "windmill-common/enterprise", "windmill-indexer/enterprise"] enterprise_saml = ["windmill-api/enterprise_saml"] stripe = ["windmill-api/stripe"] -benchmark = ["windmill-api/benchmark", "windmill-worker/benchmark", "windmill-queue/benchmark"] +benchmark = ["windmill-api/benchmark", "windmill-worker/benchmark", "windmill-queue/benchmark", "windmill-common/benchmark"] flamegraph = ["windmill-common/flamegraph", "windmill-worker/flamegraph"] loki = ["windmill-common/loki"] pg_embed = ["dep:pg-embed"] diff --git a/backend/plot.py b/backend/plot.py index ab6e630916..2b02984a62 100644 --- a/backend/plot.py +++ b/backend/plot.py @@ -11,14 +11,19 @@ def load_json_data(filepath): def plot_two_arrays_of_subarrays(arrays1, arrays2): # Function to calculate sum of durations for each step def calculate_sums(arrays): - steps = [step for step, _ in arrays[0]] # Extract steps from the first iteration - sums = {step: 0 for step in steps} # Initialize sums dictionary with step names + steps = [step for step, _ in arrays[0]['timings']] # Extract steps from the first iteration + sums = {step: 0.0 for step in steps} # Initialize sums dictionary with step names # Sum up the durations for each step across all subarrays for subarray in arrays: - for step_name, duration in subarray: + for step_name, duration in subarray['timings']: + if step_name not in sums: + sums[step_name] = 0 sums[step_name] += duration + for step_name, duration in sums.items(): + sums[step_name] = duration / 1000000000 + # Convert the sums dictionary to two lists (for plotting) step_names = list(sums.keys()) durations = list(sums.values()) @@ -29,21 +34,23 @@ def plot_two_arrays_of_subarrays(arrays1, arrays2): step_names2, sums2 = calculate_sums(arrays2) # Create two subplots, one on top of the other - fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8)) + fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 12)) # First plot (top) for the first array of subarrays - ax1.plot(step_names1, sums1, marker='o', linestyle='-', color='b') + ax1.bar(step_names1, sums1, color='b') ax1.set_title('Total Duration per Step - Main Loop') ax1.set_xlabel('Step Name') - ax1.set_ylabel('Total Duration') - ax1.grid(True) + ax1.set_ylabel('Total Duration (s)') + ax1.grid(True, axis='y') + ax1.tick_params(axis='x', rotation=45) # Second plot (bottom) for the second array of subarrays - ax2.plot(step_names2, sums2, marker='o', linestyle='-', color='r') + ax2.bar(step_names2, sums2, color='r') ax2.set_title('Total Duration per Step - Result Processor') ax2.set_xlabel('Step Name') - ax2.set_ylabel('Total Duration') - ax2.grid(True) + ax2.set_ylabel('Total Duration (s)') + ax2.grid(True, axis='y') + ax2.tick_params(axis='x', rotation=45) # Adjust layout so the plots don't overlap plt.tight_layout() @@ -52,8 +59,31 @@ def plot_two_arrays_of_subarrays(arrays1, arrays2): plt.show() # Load arrays from the JSON files -arrays1 = load_json_data('/tmp/windmill/profiling_main.json') -arrays2 = load_json_data('/tmp/windmill/profiling_result_processor.json') +main = load_json_data('/tmp/windmill/profiling_main.json') +result_processor = load_json_data('/tmp/windmill/profiling_result_processor.json') + +arrays1 = main['timings'] +arrays2 = result_processor['timings'] + +total_duration1 = main['total_duration']/1000 +total_duration2 = result_processor['total_duration']/1000 + +print(f"Total duration for main: {total_duration1}s") +print(f"Total duration for result processor: {total_duration2}s") + +iterations_total = sum(main['iter_durations']) / 1000000000 +iterations_total2 = sum(result_processor['iter_durations']) / 1000000000 + +print(f"Number of iterations: {len(main['iter_durations'])}") +print(f"Total iterations for main: {iterations_total}s") +print(f"Total iterations for result processor: {iterations_total2}s") + +# Calculate RPS +rps1 = len(main['iter_durations']) / total_duration1 +rps2 = len(result_processor['iter_durations']) / total_duration2 + +print(f"RPS for main: {rps1}") +print(f"RPS for result processor: {rps2}") # Plot the data plot_two_arrays_of_subarrays(arrays1, arrays2) \ No newline at end of file diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 0e58589813..92f69a0d64 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1368,6 +1368,8 @@ async fn handle_zombie_jobs rsmq.clone(), worker_name, send_result_never_used, + #[cfg(feature = "benchmark")] + &mut windmill_common::bench::BenchmarkIter::new(), ) .await; } diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 5f45e051d4..09acf22157 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -11,6 +11,7 @@ jemalloc = ["dep:tikv-jemalloc-ctl"] prometheus = ["dep:prometheus"] flamegraph = ["dep:tracing-flame"] loki = ["dep:tracing-loki"] +benchmark = [] parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"] [lib] diff --git a/backend/windmill-common/src/bench.rs b/backend/windmill-common/src/bench.rs new file mode 100644 index 0000000000..0e89af571c --- /dev/null +++ b/backend/windmill-common/src/bench.rs @@ -0,0 +1,213 @@ +use crate::{ + worker::{write_file, TMP_DIR}, + DB, +}; +use serde::Serialize; +use tokio::time::Instant; + +#[derive(Serialize)] +pub struct BenchmarkInfo { + #[serde(skip)] + pub start: Instant, + #[serde(skip)] + pub iters: u64, + timings: Vec, + pub iter_durations: Vec, + pub total_duration: Option, +} + +impl BenchmarkInfo { + pub fn new() -> Self { + BenchmarkInfo { + iters: 0, + timings: vec![], + start: Instant::now(), + iter_durations: vec![], + total_duration: None, + } + } + + pub fn add_iter(&mut self, bench: BenchmarkIter, inc_iters: bool) { + if inc_iters { + self.iters += 1; + } + let elapsed_total = bench.start.elapsed().as_nanos() as u64; + self.timings.push(bench); + self.iter_durations.push(elapsed_total); + } + + pub fn write_to_file(&mut self, path: &str) -> anyhow::Result<()> { + let total_duration = self.start.elapsed().as_millis() as u64; + self.total_duration = Some(total_duration as u64); + + println!( + "Writing benchmark {path}, duration of benchmark: {total_duration}s and RPS: {}", + self.iters as f64 / total_duration as f64 + ); + write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling"); + Ok(()) + } +} + +#[derive(Serialize)] +pub struct BenchmarkIter { + #[serde(skip)] + pub start: Instant, + #[serde(skip)] + last_instant: Instant, + last_step: String, + timings: Vec<(String, u32)>, +} + +impl BenchmarkIter { + pub fn new() -> Self { + BenchmarkIter { + last_instant: Instant::now(), + timings: vec![], + start: Instant::now(), + last_step: String::new(), + } + } + + pub fn add_timing(&mut self, name: &str) { + let elapsed = self.last_instant.elapsed().as_nanos() as u32; + self.timings + .push((format!("{}->{}", self.last_step, name), elapsed)); + self.last_instant = Instant::now(); + self.last_step = name.to_string(); + } +} + +pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) { + use crate::{jobs::JobKind, scripts::ScriptLang}; + + let benchmark_kind = std::env::var("BENCHMARK_KIND").unwrap_or("noop".to_string()); + + if benchmark_jobs > 0 { + match benchmark_kind.as_str() { + "dedicated" => { + // you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47 + let hash = sqlx::query_scalar!( + "SELECT hash FROM script WHERE path = $1 AND workspace_id = $2", + "f/benchmarks/dedicated", + "admins" + ) + .fetch_one(db) + .await + .unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")); + sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))", + hash, + "f/benchmarks/dedicated", + JobKind::Script as JobKind, + ScriptLang::Bun as ScriptLang, + "admins:f/benchmarks/dedicated", + "admin", + "u/admin", + "admin@windmill.dev", + chrono::Utc::now(), + "admins", + benchmark_jobs + ) + .execute(db) + .await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")); + } + "parallelflow" => { + //create dedicated script + sqlx::query!("INSERT INTO script (summary, description, dedicated_worker, content, workspace_id, path, hash, language, tag, created_by, lock) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '') ON CONFLICT (workspace_id, hash) DO NOTHING", + "export async function main() { + console.log('hello world'); + }", + "admins", + "u/admin/parallelflow", + 1234567890, + ScriptLang::Deno as ScriptLang, + "flow", + "admin", + ) + .execute(db) + .await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs {_e:#}")); + sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, raw_flow, flow_status) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12 FROM generate_series(1, 1))", + None::, + None::, + JobKind::FlowPreview as JobKind, + ScriptLang::Deno as ScriptLang, + "flow", + "admin", + "u/admin", + "admin@windmill.dev", + chrono::Utc::now(), + "admins", + serde_json::from_str::(r#" +{ + "modules": [ + { + "id": "a", + "value": { + "type": "forloopflow", + "modules": [ + { + "id": "b", + "value": { + "path": "u/admin/parallelflow", + "type": "script", + "tag_override": "", + "input_transforms": {} + }, + "summary": "calctest" + } + ], + "iterator": { + "expr": "[...new Array(300)]", + "type": "javascript" + }, + "parallel": true, + "parallelism": 10, + "skip_failures": true + } + } + ], + "preprocessor_module": null +} + "#).unwrap(), + serde_json::from_str::(r#" +{ + "step": 0, + "modules": [ + { + "id": "a", + "type": "WaitingForPriorSteps" + } + ], + "cleanup_module": {}, + "failure_module": { + "id": "failure", + "type": "WaitingForPriorSteps" + }, + "preprocessor_module": null + } + + "#).unwrap() + ) + .execute(db) + .await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs")); + } + _ => { + sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))", + None::, + None::, + JobKind::Noop as JobKind, + ScriptLang::Deno as ScriptLang, + "deno", + "admin", + "u/admin", + "admin@windmill.dev", + chrono::Utc::now(), + "admins", + benchmark_jobs + ) + .execute(db) + .await.unwrap_or_else(|_e| panic!("failed to insert noop jobs")); + } + } + } +} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 48c31fbebb..cd8059251e 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -17,6 +17,8 @@ use scripts::ScriptLang; use sqlx::{Pool, Postgres}; pub mod apps; +#[cfg(feature = "benchmark")] +pub mod bench; pub mod db; pub mod ee; pub mod error; @@ -51,6 +53,17 @@ pub const DEFAULT_MAX_CONNECTIONS_INDEXER: u32 = 5; pub const DEFAULT_HUB_BASE_URL: &str = "https://hub.windmill.dev"; +#[macro_export] +macro_rules! add_time { + ($bench:expr, $name:expr) => { + #[cfg(feature = "benchmark")] + { + $bench.add_timing($name); + // println!("{}: {:?}", $z, $y.elapsed()); + } + }; +} + lazy_static::lazy_static! { pub static ref METRICS_PORT: u16 = std::env::var("METRICS_PORT") .ok() diff --git a/backend/windmill-queue/Cargo.toml b/backend/windmill-queue/Cargo.toml index ae89cb7c6c..435b7c71f7 100644 --- a/backend/windmill-queue/Cargo.toml +++ b/backend/windmill-queue/Cargo.toml @@ -12,7 +12,7 @@ path = "src/lib.rs" default = [] enterprise = ["windmill-common/enterprise"] cloud = [] -benchmark = [] +benchmark = ["windmill-common/benchmark"] prometheus = ["dep:prometheus"] [dependencies] diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 8734dd9d24..eca874aa63 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -39,6 +39,7 @@ use windmill_audit::audit_ee::{audit_log, AuditAuthor}; use windmill_audit::ActionKind; use windmill_common::{ + add_time, auth::{fetch_authed_from_permissioned_as, permissioned_as_to_username}, db::{Authed, UserDB}, error::{self, to_anyhow, Error}, @@ -178,6 +179,8 @@ pub async fn cancel_single_job<'c>( rsmq.clone(), "server", false, + #[cfg(feature = "benchmark")] + &mut windmill_common::bench::BenchmarkIter::new(), ) .await; @@ -495,6 +498,7 @@ pub async fn add_completed_job_error, _worker_name: &str, flow_is_done: bool, + #[cfg(feature = "benchmark")] bench: &mut windmill_common::bench::BenchmarkIter, ) -> Result { #[cfg(feature = "prometheus")] register_metric( @@ -532,6 +536,8 @@ pub async fn add_completed_job_error, rsmq: Option, flow_is_done: bool, + #[cfg(feature = "benchmark")] bench: &mut windmill_common::bench::BenchmarkIter, ) -> Result { // tracing::error!("Start"); // let start = tokio::time::Instant::now(); + add_time!(bench, "add_completed_job start"); if !result.is_valid_json() { return Err(Error::InternalErr( "Result of job is invalid json (empty)".to_string(), @@ -566,6 +574,7 @@ pub async fn add_completed_job< } let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into(); + let job_id = queued_job.id; // tracing::error!("1 {:?}", start.elapsed()); @@ -576,6 +585,7 @@ pub async fn add_completed_job< ); let mem_peak = mem_peak.max(queued_job.mem_peak.unwrap_or(0)); + add_time!(bench, "add_completed_job query START"); let _duration: i64 = sqlx::query_scalar!( "INSERT INTO completed_job AS cj ( workspace_id @@ -647,6 +657,8 @@ pub async fn add_completed_job< .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?; // tracing::error!("2 {:?}", start.elapsed()); + add_time!(bench, "add_completed_job query END"); + if !queued_job.is_flow_step { if _duration > 500 && (queued_job.job_kind == JobKind::Script || queued_job.job_kind == JobKind::Preview) @@ -1766,9 +1778,9 @@ pub async fn pull( db: &Pool, rsmq: Option, suspend_first: bool, -) -> windmill_common::error::Result> { +) -> windmill_common::error::Result<(Option, bool)> { loop { - let job = pull_single_job_and_mark_as_running_no_concurrency_limit( + let (job, suspended) = pull_single_job_and_mark_as_running_no_concurrency_limit( db, rsmq.clone(), suspend_first, @@ -1776,7 +1788,7 @@ pub async fn pull( .await?; if job.is_none() { - return Ok(None); + return Ok((None, suspended)); } let has_concurent_limit = job.as_ref().unwrap().concurrent_limit.is_some(); @@ -1796,7 +1808,7 @@ pub async fn pull( if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { QUEUE_PULL_COUNT.inc(); } - return Ok(Option::Some(pulled_job)); + return Ok((Option::Some(pulled_job), suspended)); } let itx = db.begin().await?; @@ -1897,7 +1909,7 @@ pub async fn pull( QUEUE_PULL_COUNT.inc(); } tx.commit().await?; - return Ok(Option::Some(pulled_job)); + return Ok((Option::Some(pulled_job), suspended)); } let x = sqlx::query_scalar!( "UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1 RETURNING (SELECT COUNT(*) FROM jsonb_object_keys(job_uuids))", @@ -2024,8 +2036,8 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< db: &Pool, rsmq: Option, suspend_first: bool, -) -> windmill_common::error::Result> { - let job: Option = if let Some(mut rsmq) = rsmq { +) -> windmill_common::error::Result<(Option, bool)> { + let job_and_suspended: (Option, bool) = if let Some(mut rsmq) = rsmq { #[cfg(feature = "benchmark")] let instant = Instant::now(); @@ -2083,9 +2095,9 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< #[cfg(feature = "benchmark")] println!("rsmq 2: {:?}", instant.elapsed()); - m2r + (m2r, false) } else { - None + (None, false) } } else { /* Jobs can be started if they: @@ -2099,7 +2111,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< if query.is_empty() { tracing::warn!("No suspended pull queries available"); - return Ok(None); + return Ok((None, false)); } let r = if suspend_first { @@ -2119,7 +2131,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< if queries.is_empty() { tracing::warn!("No pull queries available"); - return Ok(None); + return Ok((None, false)); } for query in queries.iter() { @@ -2137,12 +2149,12 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< // #[cfg(feature = "benchmark")] // println!("pull query: {:?}", instant.elapsed()); - highest_priority_job + (highest_priority_job, false) } else { - r + (r, true) } }; - Ok(job) + Ok(job_and_suspended) } pub async fn custom_concurrency_key( diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 1529866492..c05bc26fc3 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -12,7 +12,7 @@ path = "src/lib.rs" default = [] prometheus = ["dep:prometheus", "windmill-common/prometheus"] enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:pem", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"] -benchmark = ["windmill-queue/benchmark"] +benchmark = ["windmill-queue/benchmark", "windmill-common/benchmark"] flamegraph = [] parquet = ["windmill-common/parquet", "dep:object_store"] flow_testing = [] diff --git a/backend/windmill-worker/src/bench.rs b/backend/windmill-worker/src/bench.rs deleted file mode 100644 index 0725940a04..0000000000 --- a/backend/windmill-worker/src/bench.rs +++ /dev/null @@ -1,111 +0,0 @@ -use serde::Serialize; -use tokio::time::Instant; -use windmill_common::{ - worker::{write_file, TMP_DIR}, - DB, -}; - -pub struct BenchmarkInfo { - iters: u64, - timings: Vec, -} - -impl Serialize for BenchmarkInfo { - fn serialize(&self, serializer: S) -> Result - where - S: serde::Serializer, - { - let timings: Vec> = self - .timings - .iter() - .map(|x| x.timings.clone()) - .collect::>>(); - //serialize timings as vec of vec of tuples - timings.serialize(serializer) - } -} - -impl BenchmarkInfo { - pub fn new() -> Self { - BenchmarkInfo { iters: 0, timings: vec![] } - } - - pub fn add_iter(&mut self, bench: BenchmarkIter) { - self.iters += 1; - self.timings.push(bench); - } - - pub fn write_to_file(&self, path: &str) -> anyhow::Result<()> { - println!("Writing benchmark {path}"); - write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling"); - Ok(()) - } -} - -pub struct BenchmarkIter { - last_instant: Instant, - timings: Vec<(String, u32)>, -} - -impl BenchmarkIter { - pub fn new() -> Self { - BenchmarkIter { last_instant: Instant::now(), timings: vec![] } - } - - pub fn add_timing(&mut self, name: &str) { - let elapsed = self.last_instant.elapsed().as_nanos() as u32; - self.timings.push((name.to_string(), elapsed)); - self.last_instant = Instant::now(); - } -} - -pub async fn benchmark_init(is_dedicated_worker: bool, db: &DB) { - use windmill_common::{jobs::JobKind, scripts::ScriptLang}; - - let benchmark_jobs: i32 = std::env::var("BENCHMARK_JOBS_AT_INIT") - .unwrap_or("5000".to_string()) - .parse::() - .unwrap(); - if is_dedicated_worker { - // you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47 - let hash = sqlx::query_scalar!( - "SELECT hash FROM script WHERE path = $1 AND workspace_id = $2", - "f/benchmarks/dedicated", - "admins" - ) - .fetch_one(db) - .await - .unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")); - sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))", - hash, - "f/benchmarks/dedicated", - JobKind::Script as JobKind, - ScriptLang::Bun as ScriptLang, - "admins:f/benchmarks/dedicated", - "admin", - "u/admin", - "admin@windmill.dev", - chrono::Utc::now(), - "admins", - benchmark_jobs - ) - .execute(db) - .await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs")); - } else { - sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))", - None::, - None::, - JobKind::Noop as JobKind, - ScriptLang::Deno as ScriptLang, - "deno", - "admin", - "u/admin", - "admin@windmill.dev", - chrono::Utc::now(), - "admins", - benchmark_jobs - ) - .execute(db) - .await.unwrap_or_else(|_e| panic!("failed to insert noop jobs")); - } -} diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index 08d2ccdc54..da3e25a699 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -128,7 +128,7 @@ pub async fn handle_dedicated_process( if let Err(e) = process_status(status) { tracing::error!("child exit status was not success: {e:#}"); } else { - tracing::info!("child exist status was success"); + tracing::info!("child exit status was success"); } }); diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index b30a2a5068..f1c5bc3926 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -7,8 +7,7 @@ mod snowflake_executor; mod ansible_executor; mod bash_executor; -#[cfg(feature = "benchmark")] -mod bench; + mod bun_executor; pub mod common; mod config; diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 5b417c804f..21ed761d2c 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -1,13 +1,21 @@ use serde::Serialize; use sqlx::{types::Json, Pool, Postgres}; -use std::{collections::HashMap, sync::Arc}; +use std::{ + collections::HashMap, + sync::{ + atomic::{AtomicBool, AtomicU16, Ordering}, + Arc, + }, +}; use uuid::Uuid; use windmill_common::{ + add_time, + bench::BenchmarkInfo, error::{self, Error}, - jobs::QueuedJob, - worker::to_raw_value, + jobs::{JobKind, QueuedJob}, + worker::{to_raw_value, WORKER_GROUP}, DB, }; @@ -18,7 +26,13 @@ use windmill_queue::register_metric; use serde_json::{json, value::RawValue}; -use tokio::sync::mpsc::Sender; +use tokio::{ + sync::{ + self, + mpsc::{Receiver, Sender}, + }, + task::JoinHandle, +}; use windmill_queue::{add_completed_job, add_completed_job_error}; @@ -26,13 +40,151 @@ use crate::{ bash_executor::ANSI_ESCAPE_RE, common::{read_result, save_in_cache}, worker_flow::update_flow_status_after_job_completion, - AuthedClient, JobCompleted, JobCompletedSender, SameWorkerSender, SendResult, + AuthedClient, JobCompleted, JobCompletedSender, SameWorkerSender, SendResult, INIT_SCRIPT_TAG, }; -use crate::add_time; - #[cfg(feature = "benchmark")] -use crate::bench::BenchmarkIter; +use windmill_common::bench::BenchmarkIter; + +pub fn start_background_processor( + mut job_completed_rx: Receiver, + job_completed_sender: Sender, + same_worker_queue_size: Arc, + job_completed_processor_is_done: Arc, + base_internal_url: String, + db: DB, + worker_dir: String, + same_worker_tx: SameWorkerSender, + rsmq: Option, + worker_name: String, + killpill_tx: sync::broadcast::Sender<()>, + is_dedicated_worker: bool, +) -> JoinHandle<()> +where + R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static, +{ + tokio::spawn(async move { + let mut has_been_killed = false; + + #[cfg(feature = "benchmark")] + let mut infos = BenchmarkInfo::new(); + + //if we have been killed, we want to drain the queue of jobs + while let Some(sr) = { + if has_been_killed && same_worker_queue_size.load(Ordering::SeqCst) == 0 { + job_completed_rx.try_recv().ok() + } else { + job_completed_rx.recv().await + } + } { + #[cfg(feature = "benchmark")] + let mut bench = BenchmarkIter::new(); + + match sr { + SendResult::JobCompleted(jc) => { + let rsmq = rsmq.clone(); + + let is_init_script_and_failure = + !jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG; + let is_dependency_job = matches!( + jc.job.job_kind, + JobKind::Dependencies | JobKind::FlowDependencies + ); + + handle_receive_completed_job( + jc, + &base_internal_url, + &db, + &worker_dir, + &same_worker_tx, + rsmq, + &worker_name, + job_completed_sender.clone(), + #[cfg(feature = "benchmark")] + &mut bench, + ) + .await; + + if is_init_script_and_failure { + tracing::error!("init script errored, exiting"); + killpill_tx.send(()).unwrap_or_default(); + break; + } + if is_dependency_job && is_dedicated_worker { + tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted."); + sqlx::query!( + "UPDATE config SET config = config WHERE name = $1", + format!("worker__{}", *WORKER_GROUP) + ) + .execute(&db) + .await + .expect("update config to trigger restart of all dedicated workers at that config"); + killpill_tx.send(()).unwrap_or_default(); + } + add_time!(bench, "job completed processed"); + + #[cfg(feature = "benchmark")] + { + infos.add_iter(bench, true); + } + } + SendResult::UpdateFlow { + flow, + w_id, + success, + result, + worker_dir, + stop_early_override, + token, + } => { + // let r; + tracing::info!(parent_flow = %flow, "updating flow status"); + if let Err(e) = update_flow_status_after_job_completion( + &db, + &AuthedClient { + base_internal_url: base_internal_url.to_string(), + workspace: w_id.clone(), + token: token.clone(), + force_client: None, + }, + flow, + &Uuid::nil(), + &w_id, + success, + Arc::new(result), + true, + same_worker_tx.clone(), + &worker_dir, + stop_early_override, + rsmq.clone(), + &worker_name, + job_completed_sender.clone(), + #[cfg(feature = "benchmark")] + &mut bench, + ) + .await + { + tracing::error!("Error updating flow status after job completion for {flow} on {worker_name}: {e:#}"); + } + } + SendResult::Kill => { + has_been_killed = true; + } + } + } + + job_completed_processor_is_done.store(true, Ordering::SeqCst); + + tracing::info!("finished processing all completed jobs"); + + #[cfg(feature = "benchmark")] + { + infos + .write_to_file("profiling_result_processor.json") + .expect("write to file profiling"); + } + }) +} async fn send_job_completed( job_completed_tx: JobCompletedSender, @@ -214,6 +366,8 @@ pub async fn handle_receive_completed_job< rsmq.clone(), worker_name, job_completed_tx, + #[cfg(feature = "benchmark")] + bench, ) .await; } @@ -242,8 +396,6 @@ pub async fn process_completed_job, worker_name: &str, job_completed_tx: Sender, + #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) { let err = match err { Error::JsonErr(err) => err, @@ -361,6 +522,8 @@ pub async fn handle_job_error { - #[cfg(feature = "benchmark")] - { - $bench.add_timing($name); - // println!("{}: {:?}", $z, $y.elapsed()); - } - }; -} +use windmill_common::add_time; pub async fn create_token_for_owner_in_bg( db: &Pool, @@ -552,17 +561,13 @@ impl AuthedClient { } } - - #[allow(dead_code)] #[derive(Clone)] pub struct JobCompletedSender(Sender); - #[derive(Clone)] pub struct SameWorkerSender(pub Sender, pub Arc); - pub struct SameWorkerPayload { pub job_id: Uuid, pub recoverable: bool, @@ -587,7 +592,6 @@ impl SameWorkerSender { } } - // on linux, we drop caches every DROP_CACHE_PERIOD to avoid OOM killer believing we are using too much memory just because we create lots of files when executing jobs #[cfg(any(target_os = "linux"))] pub async fn drop_cache() { @@ -674,8 +678,6 @@ fn add_outstanding_wait_time( }.in_current_span()); } - - #[tracing::instrument(name = "worker", level = "info", skip_all, fields(worker = %worker_name, hostname = %hostname))] pub async fn run_worker( db: &Pool, @@ -780,8 +782,6 @@ pub async fn run_worker() + .unwrap(); + + #[cfg(feature = "benchmark")] + benchmark_init(benchmark_jobs, &db).await; #[cfg(feature = "prometheus")] if let Some(ws) = WORKER_STARTED.as_ref() { @@ -930,158 +934,30 @@ pub async fn run_worker(5); - let (job_completed_tx, mut job_completed_rx) = mpsc::channel::(3); + let (job_completed_tx, job_completed_rx) = mpsc::channel::(3); - let job_completed_tx = JobCompletedSender( - job_completed_tx, - ); + let job_completed_tx = JobCompletedSender(job_completed_tx); let same_worker_queue_size = Arc::new(AtomicU16::new(0)); let same_worker_tx = SameWorkerSender(same_worker_tx, same_worker_queue_size.clone()); - - let db2 = db.clone(); - let base_internal_url2 = base_internal_url.to_string(); - let same_worker_tx2 = same_worker_tx.clone(); - let rsmq2 = rsmq.clone(); - let worker_dir2 = worker_dir.clone(); - - - - let worker_name2 = worker_name.clone(); - let killpill_tx2 = killpill_tx.clone(); - let job_completed_sender = job_completed_tx.0.clone(); - let job_completed_processor_is_done = Arc::new(AtomicBool::new(false)); - let job_completed_processor_is_done2 = job_completed_processor_is_done.clone(); - let same_worker_queue_size2 = same_worker_queue_size.clone(); - - - let send_result = tokio::spawn( - (async move { - let mut has_been_killed = false; - - #[cfg(feature = "benchmark")] - let mut infos = BenchmarkInfo::new(); - - - //if we have been killed, we want to drain the queue of jobs - while let Some(sr) = { - if has_been_killed && same_worker_queue_size2.load(Ordering::SeqCst) == 0 { - job_completed_rx.try_recv().ok() - } else { - job_completed_rx.recv().await - } - } { - - #[cfg(feature = "benchmark")] - let mut bench = BenchmarkIter::new(); - - match sr { - SendResult::JobCompleted(jc) => { - let rsmq2 = rsmq2.clone(); - - let is_init_script_and_failure = !jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG; - let is_dependency_job = matches!( - jc.job.job_kind, - JobKind::Dependencies | JobKind::FlowDependencies - ); - add_time!(bench, "pre handle_receive_completed_job"); - - handle_receive_completed_job( - jc, - &base_internal_url2, - &db2, - &worker_dir2, - &same_worker_tx2, - rsmq2, - &worker_name2, - job_completed_sender.clone(), - #[cfg(feature = "benchmark")] - &mut bench - ) - .await; - - if is_init_script_and_failure { - tracing::error!("init script errored, exiting"); - killpill_tx2.send(()).unwrap_or_default(); - break; - } - if is_dependency_job && is_dedicated_worker { - tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted."); - sqlx::query!( - "UPDATE config SET config = config WHERE name = $1", - format!("worker__{}", *WORKER_GROUP) - ) - .execute(&db2) - .await - .expect("update config to trigger restart of all dedicated workers at that config"); - killpill_tx2.send(()).unwrap_or_default(); - } - add_time!(bench, "post handle_receive_completed_job"); - - #[cfg(feature = "benchmark")] { - infos.add_iter(bench); - } - } - SendResult::UpdateFlow { - flow, - w_id, - success, - result, - worker_dir, - stop_early_override, - token, - } => { - // let r; - tracing::info!(parent_flow = %flow, "updating flow status"); - if let Err(e) = update_flow_status_after_job_completion( - &db2, - &AuthedClient { - base_internal_url: base_internal_url2.to_string(), - workspace: w_id.clone(), - token: token.clone(), - force_client: None, - }, - flow, - &Uuid::nil(), - &w_id, - success, - Arc::new(result), - true, - same_worker_tx2.clone(), - &worker_dir, - stop_early_override, - rsmq2.clone(), - &worker_name2, - job_completed_sender.clone(), - ) - .await - { - tracing::error!("Error updating flow status after job completion for {flow} on {worker_name2}: {e:#}"); - } - } - SendResult::Kill => { - has_been_killed = true; - } - } - } - - job_completed_processor_is_done2.store(true, Ordering::SeqCst); - - tracing::info!("finished processing all completed jobs"); - - #[cfg(feature = "benchmark")] - { - infos.write_to_file("profiling_result_processor.json").expect("write to file profiling"); - } - - }) - .instrument(tracing::Span::current()), + let send_result = start_background_processor( + job_completed_rx, + job_completed_tx.0.clone(), + same_worker_queue_size.clone(), + job_completed_processor_is_done.clone(), + base_internal_url.to_string(), + db.clone(), + worker_dir.clone(), + same_worker_tx.clone(), + rsmq.clone(), + worker_name.clone(), + killpill_tx.clone(), + is_dedicated_worker, ); let mut last_executed_job: Option = None; - let mut last_checked_suspended = Instant::now(); #[cfg(feature = "benchmark")] let mut started = false; @@ -1155,8 +1031,10 @@ pub async fn run_worker = vec![false; 30]; + let mut last_suspend_first = Instant::now(); let mut killed_but_draining_same_worker_jobs = false; + loop { #[cfg(feature = "benchmark")] let mut bench = BenchmarkIter::new(); @@ -1185,7 +1063,6 @@ pub async fn run_worker NUM_SECS_READINGS { @@ -1195,9 +1072,9 @@ pub async fn run_worker 0 && infos.iters == benchmark_jobs as u64 { + tracing::info!("benchmark finished, exiting"); + job_completed_tx + .0 + .send(SendResult::Kill) + .await + .expect("send kill to job completed tx"); + break; + } else { + tracing::info!("benchmark not finished, still pulling jobs {}", infos.iters); + } + let next_job = { // println!("2: {:?}", instant.elapsed()); #[cfg(feature = "benchmark")] @@ -1275,7 +1165,11 @@ pub async fn run_worker 3 { - last_checked_suspended = Instant::now(); - true - } else { - false - }; - add_time!(bench, "pre pull"); + let likelihood_of_suspend = + (1.0 + last_30jobs_suspended.iter().filter(|&&x| x).count() as f64) / 31.0; + let suspend_first = suspend_first_success + || rand::random::() < likelihood_of_suspend + || last_suspend_first.elapsed().as_secs_f64() > 5.0; + if suspend_first { + last_suspend_first = Instant::now(); + } + let job = pull(&db, rsmq.clone(), suspend_first).await; - add_time!(bench, "post pull"); + + add_time!(bench, "job pulled from DB"); let duration_pull_s = pull_time.elapsed().as_secs_f64(); let err_pull = job.is_ok(); - let empty = job.as_ref().is_ok_and(|x| x.is_none()); - suspend_first_success = suspend_first && !empty; + // let empty = job.as_ref().is_ok_and(|x| x.is_none()); + if !agent_mode && duration_pull_s > 0.5 { + let empty = job.as_ref().is_ok_and(|x| x.0.is_none()); tracing::warn!("pull took more than 0.5s ({duration_pull_s}), this is a sign that the database is VERY undersized for this load. empty: {empty}, err: {err_pull}"); #[cfg(feature = "prometheus")] if empty { @@ -1323,6 +1224,7 @@ pub async fn run_worker 0.1 { + let empty = job.as_ref().is_ok_and(|x| x.0.is_none()); tracing::warn!("pull took more than 0.1s ({duration_pull_s}) this is a sign that the database is undersized for this load. empty: {empty}, err: {err_pull}"); #[cfg(feature = "prometheus")] if empty { @@ -1334,8 +1236,16 @@ pub async fn run_worker 30 { + last_30jobs_suspended.remove(0); + } + } + suspend_first_success = suspend_first && suspend_success; + #[cfg(feature = "prometheus")] if j.is_some() { if let Some(wp) = worker_pull_duration_counter.as_ref() { wp.inc_by(duration_pull_s); @@ -1352,7 +1262,7 @@ pub async fn run_worker { - - last_executed_job = None; jobs_executed += 1; tracing::debug!("started handling of job {}", job.id); - if matches!(job.job_kind, JobKind::Script | JobKind::Preview) { - if !dedicated_workers.is_empty() { let key_o = if is_flow_worker { job.flow_step_id.as_ref().map(|x| x.to_string()) @@ -1386,11 +1289,16 @@ pub async fn run_worker { handle_job_error( - db, - &authed_client.get_authed().await, - arc_job.as_ref(), - 0, - None, - err, - false, - same_worker_tx.clone(), - &worker_dir, - rsmq.clone(), - &worker_name, - (&job_completed_tx.0).clone(), - ) - .await; - if is_init_script { - tracing::error!("init script job failed (in handler), exiting"); - update_worker_ping_for_failed_init_script(db, &worker_name, arc_job.id).await; + db, + &authed_client.get_authed().await, + arc_job.as_ref(), + 0, + None, + err, + false, + same_worker_tx.clone(), + &worker_dir, + rsmq.clone(), + &worker_name, + (&job_completed_tx.0).clone(), + #[cfg(feature = "benchmark")] + &mut bench, + ) + .await; + if is_init_script { + tracing::error!("init script job failed (in handler), exiting"); + update_worker_ping_for_failed_init_script( + db, + &worker_name, + arc_job.id, + ) + .await; + break; + } + } + Ok(false) if is_init_script => { + tracing::error!("init script job failed, exiting"); + update_worker_ping_for_failed_init_script(db, &worker_name, arc_job.id) + .await; break; } - }, - Ok(false) if is_init_script => { - tracing::error!("init script job failed, exiting"); - update_worker_ping_for_failed_init_script(db, &worker_name, arc_job.id).await; - break; - + _ => {} } - _ => {} - } #[cfg(feature = "prometheus")] if let Some(duration) = _timer.map(|x| x.stop_and_record()) { @@ -1594,13 +1511,12 @@ pub async fn run_worker( @@ -1741,7 +1661,6 @@ pub enum SendResult { Kill, } - #[derive(Debug, Clone)] pub struct JobCompleted { pub job: Arc, @@ -1810,6 +1729,7 @@ async fn handle_queued_job( rsmq: Option, job_completed_tx: JobCompletedSender, occupancy_metrics: &mut OccupancyMetrics, + #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) -> windmill_common::error::Result { if job.canceled { return Err(Error::JsonErr(canceled_job_to_result(&job))); @@ -1906,7 +1826,11 @@ async fn handle_queued_job( ) .fetch_one(db) .await - .map_err(|e| Error::InternalErr(format!("Fetching script path from queue for caching purposes: {e:#}")))? + .map_err(|e| { + Error::InternalErr(format!( + "Fetching script path from queue for caching purposes: {e:#}" + )) + })? .ok_or_else(|| Error::InternalErr(format!("Expected script_path")))?; let step = match step.unwrap() { Step::Step(i) => i.to_string(), @@ -2082,7 +2006,8 @@ async fn handle_queued_job( occupancy_metrics, ) .await; - occupancy_metrics.total_duration_of_running_jobs += metric_timer.elapsed().as_secs_f32(); + occupancy_metrics.total_duration_of_running_jobs += + metric_timer.elapsed().as_secs_f32(); r } }; @@ -2116,7 +2041,6 @@ async fn handle_queued_job( } } - pub fn build_envs( envs: Option>, ) -> windmill_common::error::Result> { @@ -2446,7 +2370,11 @@ async fn handle_code_execution_job( ); let shared_mount = if job.same_worker && job.language != Some(ScriptLang::Deno) { - let folder = if job.language == Some(ScriptLang::Go) { "/go" } else { "" }; + let folder = if job.language == Some(ScriptLang::Go) { + "/go" + } else { + "" + }; format!( r#" mount {{ @@ -2618,7 +2546,6 @@ mount {{ .await } Some(ScriptLang::Ansible) => { - handle_ansible_job( requirements_o, job_dir, @@ -2634,7 +2561,8 @@ mount {{ base_internal_url, envs, occupancy_metrics, - ).await + ) + .await } _ => panic!("unreachable, language is not supported: {language:#?}"), }; diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 8c1161cc3e..b8085e65fe 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -29,7 +29,10 @@ use sqlx::FromRow; use tokio::sync::mpsc::Sender; use tracing::instrument; use uuid::Uuid; +use windmill_common::add_time; use windmill_common::auth::JobPerms; +#[cfg(feature = "benchmark")] +use windmill_common::bench::BenchmarkIter; use windmill_common::db::Authed; use windmill_common::flow_status::{ ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult, @@ -76,6 +79,7 @@ pub async fn update_flow_status_after_job_completion< rsmq: Option, worker_name: &str, job_completed_tx: Sender, + #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) -> error::Result<()> { // this is manual tailrecursion because async_recursion blows up the stack // todo!(); @@ -97,6 +101,8 @@ pub async fn update_flow_status_after_job_completion< rsmq.clone(), worker_name, job_completed_tx.clone(), + #[cfg(feature = "benchmark")] + bench, ) .await?; while let Some(nrec) = rec { @@ -117,6 +123,8 @@ pub async fn update_flow_status_after_job_completion< rsmq.clone(), worker_name, job_completed_tx.clone(), + #[cfg(feature = "benchmark")] + bench, ) .await { @@ -141,6 +149,8 @@ pub async fn update_flow_status_after_job_completion< rsmq.clone(), worker_name, job_completed_tx.clone(), + #[cfg(feature = "benchmark")] + bench, ) .await? } @@ -193,7 +203,9 @@ pub async fn update_flow_status_after_job_completion_internal< rsmq: Option, worker_name: &str, job_completed_tx: Sender, + #[cfg(feature = "benchmark")] bench: &mut BenchmarkIter, ) -> error::Result> { + add_time!(bench, "update flow status internal START"); let ( should_continue_flow, flow_job, @@ -376,6 +388,8 @@ pub async fn update_flow_status_after_job_completion_internal< })?; } + add_time!(bench, "process module status START"); + let (inc_step_counter, new_status) = match module_status { FlowStatusModule::InProgress { iterator, @@ -517,9 +531,9 @@ pub async fn update_flow_status_after_job_completion_internal< tracing::info!( "parallel iteration {job_id_for_status} of flow {flow} has finished", ); - (true, Some(new_status)) } else { + add_time!(bench, "handle parallel flow start"); tx.commit().await?; if parallelism.is_some() { @@ -562,7 +576,7 @@ pub async fn update_flow_status_after_job_completion_internal< r.unwrap() ); } - + add_time!(bench, "non final parallel flow finished"); return Ok(None); } } @@ -964,6 +978,8 @@ pub async fn update_flow_status_after_job_completion_internal< rsmq.clone(), worker_name, true, + #[cfg(feature = "benchmark")] + bench, ) .await?; } else { @@ -994,6 +1010,8 @@ pub async fn update_flow_status_after_job_completion_internal< let success = success && (!is_failure_step || result_has_recover_true(nresult.clone())) && !skip_error_handler; + + add_time!(bench, "flow status update 1"); if success { add_completed_job( db, @@ -1005,6 +1023,8 @@ pub async fn update_flow_status_after_job_completion_internal< None, rsmq.clone(), true, + #[cfg(feature = "benchmark")] + bench, ) .await?; } else { @@ -1022,6 +1042,8 @@ pub async fn update_flow_status_after_job_completion_internal< None, rsmq.clone(), true, + #[cfg(feature = "benchmark")] + bench, ) .await?; } @@ -1059,6 +1081,8 @@ pub async fn update_flow_status_after_job_completion_internal< rsmq.clone(), worker_name, true, + #[cfg(feature = "benchmark")] + bench, ) .await; true