From 60f2e9e2424cbd96d495739b8a50b47a141daae2 Mon Sep 17 00:00:00 2001 From: Lucas Abel <22837557+uael@users.noreply.github.com> Date: Wed, 27 Nov 2024 10:06:19 +0100 Subject: [PATCH] enable `is_simple` optimization for `FlowScript` (#4799) * backend: remove some allocations in `worker_flow.rs` * feat(backend): enable `is_simple` optimization for `FlowScript` * fix: fix node retrieval for `flowscript` inner step --- backend/windmill-api/openapi.yaml | 2 + backend/windmill-common/src/flows.rs | 17 +- backend/windmill-worker/src/js_eval.rs | 10 +- backend/windmill-worker/src/worker_flow.rs | 166 ++++++++++-------- .../components/FlowStatusViewerInner.svelte | 2 +- 5 files changed, 108 insertions(+), 89 deletions(-) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 4f8d042f30..4a5d070044 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -10873,6 +10873,7 @@ components: "identity", "deploymentcallback", "singlescriptflow", + "flowscript", ] schedule_path: type: string @@ -10992,6 +10993,7 @@ components: "identity", "deploymentcallback", "singlescriptflow", + "flowscript", ] schedule_path: type: string diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index c5f1b9dede..7b819f550a 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -283,12 +283,6 @@ pub struct SkipIf { pub expr: String, } -#[derive(Deserialize)] -pub struct FlowModuleValueType { - #[serde(rename = "type")] - pub type_: String, -} - #[derive(Deserialize)] pub struct FlowModuleValueWithParallel { #[serde(rename = "type")] @@ -344,13 +338,18 @@ impl FlowModule { pub fn is_simple(&self) -> bool { //todo: flow modules could also be simple execpt for the fact that the case of having single parallel flow approval step is not handled well (Create SuspendedTimeout) self.get_type() - .is_ok_and(|x| x == "script" || x == "rawscript") + .is_ok_and(|x| x == "script" || x == "rawscript" || x == "flowscript") } - pub fn get_type(&self) -> anyhow::Result { + pub fn get_type(&self) -> anyhow::Result<&str> { + #[derive(Deserialize)] + pub struct FlowModuleValueType<'a> { + pub r#type: &'a str, + } + serde_json::from_str::(self.value.get()) .map_err(crate::error::to_anyhow) - .map(|x| x.type_) + .map(|x| x.r#type) } } diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index 45ee1cefd2..b4b84e2e2b 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -173,7 +173,7 @@ pub async fn eval_timeout( transform_context: HashMap>>, flow_input: Option>>>, authed_client: Option<&AuthedClient>, - by_id: Option, + by_id: Option<&IdContext>, #[allow(unused_variables)] ctx: Option>, ) -> anyhow::Result> { let expr = expr.trim().to_string(); @@ -200,7 +200,7 @@ pub async fn eval_timeout( } } - let p_ids = by_id.as_ref().map(|x| { + let p_ids = by_id.map(|x| { [ format!("results.{}", x.previous_id), format!("results?.{}", x.previous_id), @@ -221,7 +221,7 @@ pub async fn eval_timeout( .clone()); } - if by_id.is_some() && authed_client.is_some() { + if let (Some(by_id), Some(authed_client)) = (by_id, authed_client) { if let Some((id, idx_o, rest)) = RE_FULL.captures(&expr).map(|x| { ( x.get(1).unwrap().as_str(), @@ -238,8 +238,7 @@ pub async fn eval_timeout( rest.map(|x| x.trim_start_matches('.').to_string()) }; return authed_client - .unwrap() - .get_result_by_id(&by_id.as_ref().unwrap().flow_job.to_string(), id, query) + .get_result_by_id(&by_id.flow_job.to_string(), id, query) .await; } } @@ -253,6 +252,7 @@ pub async fn eval_timeout( #[cfg(feature = "deno_core")] { let expr2 = expr.clone(); + let by_id = by_id.cloned(); let (sender, mut receiver) = oneshot::channel::(); let has_client = authed_client.is_some(); let authed_client = authed_client.cloned(); diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index e87030eb91..f339cd6868 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -333,7 +333,7 @@ pub async fn update_flow_status_after_job_completion_internal( Error::InternalErr(format!("retrieval of args from state: {e:#}")) })?; compute_bool_from_expr( - expr.to_string(), + &expr, Marc::new(args.args.unwrap_or_default().0), result.clone(), all_iters, @@ -862,7 +862,7 @@ pub async fn update_flow_status_after_job_completion_internal( })?; let should_stop = compute_bool_from_expr( - expr.to_string(), + &expr, Marc::new(args.args.unwrap_or_default().0), nresult.clone(), None, @@ -1298,11 +1298,11 @@ fn next_retry(retry: &Retry, status: &RetryStatus) -> Option<(u16, Duration)> { } async fn compute_bool_from_expr( - expr: String, + expr: &str, flow_args: Marc>>, result: Arc>, all_iters: Option>>, - by_id: Option, + by_id: Option<&IdContext>, client: Option<&AuthedClient>, resumes: Option<(Arc>, Arc>, Arc>)>, ctx: Option>, @@ -1465,7 +1465,7 @@ async fn transform_input( env.clone(), Some(flow_args.clone()), Some(client), - Some(by_id.clone()), + Some(by_id), None, ) .await @@ -1715,7 +1715,7 @@ async fn push_next_flow_job( } if let Some(skip_expr) = &flow.skip_expr { let skip = compute_bool_from_expr( - skip_expr.to_string(), + &skip_expr, arc_flow_job_args.clone(), Arc::new(to_raw_value(&json!("{}"))), None, @@ -2187,11 +2187,11 @@ async fn push_next_flow_job( let is_skipped = if let Some(skip_if) = &module.skip_if { let idcontext = get_transform_context(&flow_job, previous_id.as_str(), &status).await?; compute_bool_from_expr( - skip_if.expr.to_string(), + &skip_if.expr, arc_flow_job_args.clone(), arc_last_job_result.clone(), None, - Some(idcontext.clone()), + Some(&idcontext), Some(client), Some((resumes.clone(), resume.clone(), approvers.clone())), None, @@ -2998,7 +2998,7 @@ async fn compute_next_flow_transform( match module.get_value()? { FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity), FlowModuleValue::Flow { path, .. } => { - let payload = flow_to_payload(&path, &delete_after_use); + let payload = flow_to_payload(path, delete_after_use); Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(payload), NextStatus::NextStep, @@ -3006,7 +3006,7 @@ async fn compute_next_flow_transform( } FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => { let payload = - script_to_payload(&script_hash, &script_path, db, flow_job, module, &tag_override) + script_to_payload(script_hash, script_path, db, flow_job, module, tag_override) .await?; Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(payload), @@ -3037,15 +3037,15 @@ async fn compute_next_flow_transform( }); let payload = raw_script_to_payload( path, - &content, - &language, - &lock, - &custom_concurrency_key, - &concurrent_limit, - &concurrency_time_window_s, + content, + language, + lock, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, module, - &tag, - &delete_after_use, + tag, + delete_after_use, ); Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(payload), @@ -3082,7 +3082,7 @@ async fn compute_next_flow_transform( }, FlowModuleValue::WhileloopFlow { modules, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg - let is_simple = is_simple_modules(&modules, flow); + let is_simple = is_simple_modules(&modules, flow.failure_module.as_ref()); let (flow_jobs, flow_jobs_success) = match status_module { FlowStatusModule::InProgress { flow_jobs: Some(flow_jobs), @@ -3106,7 +3106,7 @@ async fn compute_next_flow_transform( }, while_loop: true, }, - &modules, + modules, flow_job, is_simple, db, @@ -3118,7 +3118,7 @@ async fn compute_next_flow_transform( /* forloop modules are expected set `iter: { value: Value, index: usize }` as job arguments */ FlowModuleValue::ForloopFlow { modules, iterator, parallel, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg - let is_simple = !parallel && is_simple_modules(&modules, flow); + let is_simple = !parallel && is_simple_modules(&modules, flow.failure_module.as_ref()); // if is_simple { // match value { @@ -3154,7 +3154,7 @@ async fn compute_next_flow_transform( flow, status, ns, - &modules, + modules, flow_job, is_simple, db, @@ -3233,11 +3233,11 @@ async fn compute_next_flow_transform( let idcontext = get_transform_context(&flow_job, previous_id, &status).await?; for (i, b) in branches.iter().enumerate() { let pred = compute_bool_from_expr( - b.expr.to_string(), + &b.expr, arc_flow_job_args.clone(), arc_last_job_result.clone(), None, - Some(idcontext.clone()), + Some(&idcontext), Some(client), Some((resumes.clone(), resume.clone(), approvers.clone())), None, @@ -3444,7 +3444,7 @@ async fn next_loop_iteration( flow: &FlowValue, status: &FlowStatus, ns: ForloopNextIteration, - modules: &Vec, + mut modules: Vec, flow_job: &QueuedJob, is_simple: bool, db: &sqlx::Pool, @@ -3456,29 +3456,25 @@ async fn next_loop_iteration( failure_module.id_append(&format!("{}-{}", status.step, ns.index)); fm = Some(failure_module); } - let mut modules = (*modules).clone(); add_virtual_items_if_necessary(&mut modules); let inner_path = Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)); if is_simple { - let value = &modules[0].get_value()?; + let mut value = modules[0].get_value()?; + let simple_input_transforms = match &mut value { + FlowModuleValue::Script { input_transforms, .. } + | FlowModuleValue::RawScript { input_transforms, .. } + | FlowModuleValue::FlowScript { input_transforms, .. } + | FlowModuleValue::Flow { input_transforms, .. } => { + Some(std::mem::take(input_transforms)) + }, + _ => None, + }; let payload = payload_from_simple_module(value, db, flow_job, module, inner_path).await?; Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(payload), NextStatus::NextLoopIteration { next: ns, - simple_input_transforms: if is_simple { - match value { - FlowModuleValue::Script { input_transforms, .. } - | FlowModuleValue::RawScript { input_transforms, .. } - | FlowModuleValue::FlowScript { input_transforms, .. } - | FlowModuleValue::Flow { input_transforms, .. } => { - Some(input_transforms.clone()) - } - _ => None, - } - } else { - None - }, + simple_input_transforms, }, )) } else { @@ -3510,7 +3506,7 @@ async fn next_loop_iteration( } } -fn is_simple_modules(modules: &Vec, flow: &FlowValue) -> bool { +fn is_simple_modules(modules: &Vec, failure_module: Option<&Box>) -> bool { let is_simple = modules.len() == 1 && modules[0].is_simple() && modules[0].sleep.is_none() @@ -3521,7 +3517,7 @@ fn is_simple_modules(modules: &Vec, flow: &FlowValue) -> bool { && modules[0].stop_after_all_iters_if.is_none() && modules[0].skip_if.is_none() && (modules[0].mock.is_none() || modules[0].mock.as_ref().is_some_and(|m| !m.enabled)) - && flow.failure_module.is_none(); + && failure_module.is_none(); is_simple } @@ -3565,7 +3561,7 @@ async fn next_forloop_status( context, Some(arc_flow_job_args), Some(client), - Some(by_id), + Some(&by_id), None, ) .await? @@ -3626,7 +3622,7 @@ async fn next_forloop_status( context, Some(arc_flow_job_args), Some(client), - Some(by_id), + Some(&by_id), None, ) .await? @@ -3665,7 +3661,7 @@ async fn next_forloop_status( } async fn payload_from_simple_module( - value: &FlowModuleValue, + value: FlowModuleValue, db: &sqlx::Pool, flow_job: &QueuedJob, module: &FlowModule, @@ -3673,7 +3669,7 @@ async fn payload_from_simple_module( ) -> Result { let delete_after_use = module.delete_after_use.unwrap_or(false); Ok(match value { - FlowModuleValue::Flow { path, .. } => flow_to_payload(path, &delete_after_use), + FlowModuleValue::Flow { path, .. } => flow_to_payload(path, delete_after_use), FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => { script_to_payload(script_hash, script_path, db, flow_job, module, tag_override).await? } @@ -3688,7 +3684,7 @@ async fn payload_from_simple_module( concurrency_time_window_s, .. } => raw_script_to_payload( - path.clone().or(inner_path), + path.or(inner_path), content, language, lock, @@ -3697,68 +3693,90 @@ async fn payload_from_simple_module( concurrency_time_window_s, module, tag, - &delete_after_use, + delete_after_use, ), + FlowModuleValue::FlowScript { + id, // flow_node(id). + tag, + language, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + .. + } => JobPayloadWithTag { + payload: JobPayload::FlowScript { + id, + language, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, + cache_ttl: module.cache_ttl.map(|x| x as i32), + dedicated_worker: None, + }, + tag, + delete_after_use, + timeout: module.timeout, + }, _ => unreachable!("is simple flow"), }) } fn raw_script_to_payload( path: Option, - content: &String, - language: &windmill_common::scripts::ScriptLang, - lock: &Option, - custom_concurrency_key: &Option, - concurrent_limit: &Option, - concurrency_time_window_s: &Option, + content: String, + language: windmill_common::scripts::ScriptLang, + lock: Option, + custom_concurrency_key: Option, + concurrent_limit: Option, + concurrency_time_window_s: Option, module: &FlowModule, - tag: &Option, - delete_after_use: &bool, + tag: Option, + delete_after_use: bool, ) -> JobPayloadWithTag { JobPayloadWithTag { payload: JobPayload::Code(RawCode { hash: None, path, - content: content.clone(), - language: language.clone(), - lock: lock.clone(), - custom_concurrency_key: custom_concurrency_key.clone(), - concurrent_limit: *concurrent_limit, - concurrency_time_window_s: *concurrency_time_window_s, + content, + language, + lock, + custom_concurrency_key, + concurrent_limit, + concurrency_time_window_s, cache_ttl: module.cache_ttl.map(|x| x as i32), dedicated_worker: None, }), - tag: tag.clone(), - delete_after_use: *delete_after_use, + tag, + delete_after_use, timeout: module.timeout, } } -fn flow_to_payload(path: &str, delete_after_use: &bool) -> JobPayloadWithTag { +fn flow_to_payload(path: String, delete_after_use: bool) -> JobPayloadWithTag { let payload = JobPayload::Flow { - path: path.to_string(), + path, dedicated_worker: None, apply_preprocessor: false, }; - JobPayloadWithTag { payload, tag: None, delete_after_use: *delete_after_use, timeout: None } + JobPayloadWithTag { payload, tag: None, delete_after_use, timeout: None } } async fn script_to_payload( - script_hash: &Option, - script_path: &String, + script_hash: Option, + script_path: String, db: &sqlx::Pool, flow_job: &QueuedJob, module: &FlowModule, - tag_override: &Option, + tag_override: Option, ) -> Result { let tag_override = if tag_override.as_ref().is_some_and(|x| x.trim().is_empty()) { None } else { - tag_override.clone() + tag_override }; let (payload, tag, delete_after_use, script_timeout) = if script_hash.is_none() { let (jp, tag, delete_after_use, script_timeout) = - script_path_to_payload(script_path, db, &flow_job.workspace_id, Some(true)).await?; + script_path_to_payload(&script_path, db, &flow_job.workspace_id, Some(true)).await?; ( jp, tag_override.to_owned().or(tag), @@ -3766,7 +3784,7 @@ async fn script_to_payload( script_timeout, ) } else { - let hash = script_hash.clone().unwrap(); + let hash = script_hash.unwrap(); let mut tx: sqlx::Transaction<'_, sqlx::Postgres> = db.begin().await?; let ( tag, @@ -3783,7 +3801,7 @@ async fn script_to_payload( ( JobPayload::ScriptHash { hash, - path: script_path.to_owned(), + path: script_path, custom_concurrency_key, concurrent_limit, concurrency_time_window_s, diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index e7fddf26ae..8878fcfeed 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -711,7 +711,7 @@ }) } - if (jobLoaded.job_kind == 'script' || jobLoaded.job_kind == 'preview') { + if (jobLoaded.job_kind == 'script' || jobLoaded.job_kind == 'flowscript' || jobLoaded.job_kind == 'preview') { let id: string | undefined = undefined if (innerModule?.type == 'forloopflow' && innerModule.modules.length == 1) { id = innerModule?.modules?.[0]?.id