starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-21 08:02:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: handle better single step parallel flows

Browse Source
This commit is contained in:
Ruben Fiszel
2024-06-19 08:28:15 +02:00
parent 9c152d96c9
commit 6d4b8a2e4f
3 changed files with 41 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/worker_flow.rs
+33 -21
View File
@@ -2595,8 +2595,17 @@ 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 = is_simple_modules(modules, flow);
let is_simple = !parallel && is_simple_modules(modules, flow);
// if is_simple {
// match value {
// FlowModuleValue::Script { input_transforms, .. }
// | FlowModuleValue::RawScript { input_transforms, .. }
// | FlowModuleValue::Flow { input_transforms, .. } => {
// Some(input_transforms.clone())
// }
// _ => None,
// }
let next_loop_status = next_forloop_status(
status_module,
by_id,
@@ -2634,14 +2643,18 @@ async fn compute_next_flow_transform(
if modules.is_empty() {
return Ok(NextFlowTransform::EmptyInnerFlows);
} else {
let inner_path = Some(format!("{}/loop-parallel", flow_job.script_path(),));
let value = &modules[0].get_value()?;
let continue_payload = if is_simple {
let payload =
payload_from_simple_module(value, db, flow_job, module, inner_path)
.await?;
ContinuePayload::ForloopJobs { n: itered.len(), payload: payload }
} else {
// let inner_path = Some(format!("{}/loop-parallel", flow_job.script_path(),));
// let value = &modules[0].get_value()?;
// we removed the is_simple_case
// if is_simple {
// let payload =
// payload_from_simple_module(value, db, flow_job, module, inner_path)
// .await?;
// ContinuePayload::ForloopJobs { n: itered.len(), payload: payload }
// } else {
let continue_payload = {
let payload = {
JobPayloadWithTag {
payload: JobPayload::RawFlow {
@@ -2675,18 +2688,17 @@ async fn compute_next_flow_transform(
index: 0,
itered,
}),
simple_input_transforms: if is_simple {
match value {
FlowModuleValue::Script { input_transforms, .. }
| FlowModuleValue::RawScript { input_transforms, .. }
| FlowModuleValue::Flow { input_transforms, .. } => {
Some(input_transforms.clone())
}
_ => None,
}
} else {
None
},
// we removed the is_simple_case for simple_input_transforms
// if is_simple {
// match value {
// FlowModuleValue::Script { input_transforms, .. }
// | FlowModuleValue::RawScript { input_transforms, .. }
// | FlowModuleValue::Flow { input_transforms, .. } => {
// Some(input_transforms.clone())
// }
// _ => None,
// }
simple_input_transforms: None,
},
))
}
frontend/src/lib/components/FlowStatusViewerInner.svelte
+7 -4
View File
@@ -237,11 +237,14 @@
timeout && clearTimeout(timeout)
innerModules = []
if (flowJobIds) {
let common = {
iteration_from: Math.max(flowJobIds.flowJobs.length - 20, 0),
iteration_total: flowJobIds?.length
}
let modId = flowJobIds?.moduleId ?? ''
let common = {
iteration_from:
$localDurationStatuses?.[modId]?.iteration_from ??
Math.max(flowJobIds.flowJobs.length - 20, 0),
iteration_total: $localDurationStatuses?.[modId]?.iteration_total ?? flowJobIds?.length
}
$localDurationStatuses[modId] = {
...($localDurationStatuses[modId] ?? { byJob: {} }),
...common
frontend/src/lib/components/schema/SchemaFormDND.svelte
+1 -1
View File
@@ -25,7 +25,7 @@
function computeItems() {
return (
(schema.order ?? Object.keys(schema.properties ?? {}) ?? []).map((key) => ({
(schema?.order ?? Object.keys(schema?.properties ?? {}) ?? []).map((key) => ({
id: key,
value: key
})) ?? []