kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

enable is_simple optimization for FlowScript (#4799)

Browse Source 
* backend: remove some allocations in `worker_flow.rs`

* feat(backend): enable `is_simple` optimization for `FlowScript`

* fix: fix node retrieval for `flowscript` inner step
This commit is contained in:
Lucas Abel
2024-11-27 10:06:19 +01:00
committed by GitHub
parent 41a7fbb80f
commit 60f2e9e242
5 changed files with 108 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/windmill-api/openapi.yaml
View File

@@ -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

17
backend/windmill-common/src/flows.rs
View File

@@ -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<String> {
pub fn get_type(&self) -> anyhow::Result<&str> {
#[derive(Deserialize)]
pub struct FlowModuleValueType<'a> {
pub r#type: &'a str,
}
serde_json::from_str::<FlowModuleValueType>(self.value.get())
.map_err(crate::error::to_anyhow)
.map(|x| x.type_)
.map(|x| x.r#type)
}
}

10
backend/windmill-worker/src/js_eval.rs
View File

@@ -173,7 +173,7 @@ pub async fn eval_timeout(
transform_context: HashMap<String, Arc<Box<RawValue>>>,
flow_input: Option<mappable_rc::Marc<HashMap<String, Box<RawValue>>>>,
authed_client: Option<&AuthedClient>,
by_id: Option<IdContext>,
by_id: Option<&IdContext>,
#[allow(unused_variables)] ctx: Option<Vec<(String, String)>>,
) -> anyhow::Result<Box<RawValue>> {
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::<IsolateHandle>();
let has_client = authed_client.is_some();
let authed_client = authed_client.cloned();

166
backend/windmill-worker/src/worker_flow.rs
View File

@@ -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<HashMap<String, Box<RawValue>>>,
result: Arc<Box<RawValue>>,
all_iters: Option<Arc<Box<RawValue>>>,
by_id: Option<IdContext>,
by_id: Option<&IdContext>,
client: Option<&AuthedClient>,
resumes: Option<(Arc<Box<RawValue>>, Arc<Box<RawValue>>, Arc<Box<RawValue>>)>,
ctx: Option<Vec<(String, String)>>,
@@ -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<FlowModule>,
mut modules: Vec<FlowModule>,
flow_job: &QueuedJob,
is_simple: bool,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -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<FlowModule>, flow: &FlowValue) -> bool {
fn is_simple_modules(modules: &Vec<FlowModule>, failure_module: Option<&Box<FlowModule>>) -> 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<FlowModule>, 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<sqlx::Postgres>,
flow_job: &QueuedJob,
module: &FlowModule,
@@ -3673,7 +3669,7 @@ async fn payload_from_simple_module(
) -> Result<JobPayloadWithTag, Error> {
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<String>,
content: &String,
language: &windmill_common::scripts::ScriptLang,
lock: &Option<String>,
custom_concurrency_key: &Option<String>,
concurrent_limit: &Option<i32>,
concurrency_time_window_s: &Option<i32>,
content: String,
language: windmill_common::scripts::ScriptLang,
lock: Option<String>,
custom_concurrency_key: Option<String>,
concurrent_limit: Option<i32>,
concurrency_time_window_s: Option<i32>,
module: &FlowModule,
tag: &Option<String>,
delete_after_use: &bool,
tag: Option<String>,
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<windmill_common::scripts::ScriptHash>,
script_path: &String,
script_hash: Option<windmill_common::scripts::ScriptHash>,
script_path: String,
db: &sqlx::Pool<sqlx::Postgres>,
flow_job: &QueuedJob,
module: &FlowModule,
tag_override: &Option<String>,
tag_override: Option<String>,
) -> Result<JobPayloadWithTag, Error> {
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,

2
frontend/src/lib/components/FlowStatusViewerInner.svelte
View File

@@ -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