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

feat: parallel loop as expr (#6743)

Browse Source 
* ok

* revert

* backward compatible

* nits

* nits

* perf

* ok

* nits
This commit is contained in:
dieriba
2025-10-08 08:57:22 +02:00
committed by GitHub
parent c9dae9580a
commit 27d7809ef9
5 changed files with 190 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/flows.rs
+15 -4
View File
@@ -405,14 +405,24 @@ pub struct FlowModuleValueWithParallel {
#[serde(rename = "type")]
pub type_: String,
pub parallel: Option<bool>,
pub parallelism: Option<u16>,
#[serde(
default,
deserialize_with = "raw_value_to_input_transform::<_, u16>",
skip_serializing_if = "Option::is_none"
)]
pub parallelism: Option<InputTransform>,
}
#[derive(Deserialize)]
pub struct FlowModuleValueWithSkipFailures {
pub skip_failures: Option<bool>,
pub parallel: Option<bool>,
pub parallelism: Option<u16>,
#[serde(
default,
deserialize_with = "raw_value_to_input_transform::<_, u16>",
skip_serializing_if = "Option::is_none"
)]
pub parallelism: Option<InputTransform>,
}
#[derive(Deserialize)]
@@ -627,7 +637,7 @@ pub enum FlowModuleValue {
skip_failures: bool,
parallel: bool,
#[serde(skip_serializing_if = "Option::is_none")]
parallelism: Option<u16>,
parallelism: Option<InputTransform>,
},
/// While loop node
@@ -730,7 +740,8 @@ struct UntaggedFlowModuleValue {
modules: Option<Vec<FlowModule>>,
skip_failures: Option<bool>,
parallel: Option<bool>,
parallelism: Option<u16>,
#[serde(default, deserialize_with = "raw_value_to_input_transform::<_, u16>")]
parallelism: Option<InputTransform>,
branches: Option<Vec<Branch>>,
default: Option<Vec<FlowModule>>,
content: Option<String>,
backend/windmill-worker/src/worker_flow.rs
+20 -5
View File
@@ -372,7 +372,7 @@ pub async fn update_flow_status_after_job_completion_internal(
.as_ref()
.and_then(|x| x.skip_failures)
.unwrap_or(false),
value.as_ref().and_then(|x| x.parallelism),
value.and_then(|x| x.parallelism),
*parallel,
)
} else {
@@ -3172,17 +3172,32 @@ async fn push_next_flow_job(
tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushed next flow job: {uuid}");
if value_with_parallel.type_ == "forloopflow" {
if let Some(p) = value_with_parallel.parallelism {
tracing::debug!(id = %flow_job.id, root_id = %job_root, "updating suspend for forloopflow job {uuid}");
if let Some(parallelism_transform) = &value_with_parallel.parallelism {
tracing::debug!(id = %flow_job.id, root_id = %job_root, "evaluating parallelism expression for forloopflow job {uuid}");
if i as u16 >= p {
let ctx = get_transform_context(&flow_job, &previous_id, &status)
.warn_after_seconds(3)
.await?;
let evaluated_parallelism = evaluate_input_transform::<u16>(
parallelism_transform,
arc_last_job_result.clone(),
Some(arc_flow_job_args.clone()),
Some(client),
Some(&ctx),
)
.await?;
tracing::debug!(id = %flow_job.id, root_id = %job_root, "updating suspend for forloopflow job {uuid} with parallelism {evaluated_parallelism}");
if i as u16 >= evaluated_parallelism {
sqlx::query!(
"UPDATE v2_job_queue SET
suspend = $1,
suspend_until = now() + interval '14 day',
running = true
WHERE id = $2",
(i as u16 - p + 1) as i32,
(i as u16 - evaluated_parallelism + 1) as i32,
uuid,
)
.execute(&mut *inner_tx)
frontend/src/lib/components/copilot/chat/flow/FlowAIChat.svelte
+5 -2
View File
@@ -532,8 +532,11 @@
module.value.parallelism = undefined
} else if (module.value.parallel || opts.parallel === true) {
// Only set parallelism if parallel is enabled
const n = Math.max(1, Math.floor(Math.abs(opts.parallelism)))
module.value.parallelism = n
const n = Math.max(1, Math.floor(Math.abs(opts.parallelism)));
module.value.parallelism = {
type: 'static',
value: n
}
}
}
frontend/src/lib/components/flows/content/FlowLoop.svelte
+148 -11
View File
@@ -16,8 +16,8 @@
import FlowModuleSleep from './FlowModuleSleep.svelte'
import FlowModuleMock from './FlowModuleMock.svelte'
import { Play } from 'lucide-svelte'
import type { FlowModule, Job } from '$lib/gen'
import { Play, FunctionSquare } from 'lucide-svelte'
import type { FlowModule, ForloopFlow, Job } from '$lib/gen'
import FlowLoopIterationPreview from '$lib/components/FlowLoopIterationPreview.svelte'
import FlowModuleDeleteAfterUse from './FlowModuleDeleteAfterUse.svelte'
import IteratorGen from '$lib/components/copilot/IteratorGen.svelte'
@@ -28,6 +28,10 @@
import type { PropPickerContext } from '$lib/components/prop_picker'
import TabsV2 from '$lib/components/common/tabs/TabsV2.svelte'
import { useUiIntent } from '$lib/components/copilot/chat/flow/useUiIntent'
import { emptySchema } from '$lib/utils'
import { slide } from 'svelte/transition'
import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
const { previewArgs, flowStateStore, flowStore, currentEditor } =
getContext<FlowEditorContext>('FlowEditorContext')
@@ -49,7 +53,30 @@
}: Props = $props()
let editor: SimpleEditor | undefined = $state(undefined)
let parallelismEditor: SimpleEditor | undefined = $state(undefined)
let selected: string = $state('early-stop')
let parallelismType: 'static' | 'javascript' | undefined = $state(
mod.value.type === 'forloopflow'
? mod.value.parallelism?.type === 'javascript'
? 'javascript'
: 'static'
: undefined
)
let parallelismSchema = $state(emptySchema())
parallelismSchema.properties['parallelism'] = {
type: 'number'
}
if (mod.value.type === 'forloopflow') {
const forloopValue = mod.value as ForloopFlow
if (typeof forloopValue.parallelism === 'number') {
forloopValue.parallelism = {
type: 'static',
value: forloopValue.parallelism
}
}
}
// UI Intent handling for AI tool control
useUiIntent(`forloopflow-${mod.id}`, {
@@ -139,8 +166,8 @@
<Splitpanes horizontal class="h-full">
<Pane size={50} minSize={20} class="p-4">
{#if mod.value.type === 'forloopflow'}
<div class="flex flex-row gap-8 mt-2 mb-6">
<div>
<div class="flex flex-row gap-6 mt-2 mb-6">
<div class="flex-shrink-0">
<div class="mb-2 text-sm font-bold"
>Skip failures <Tooltip
documentationLink="https://www.windmill.dev/docs/flows/flow_loops"
@@ -154,30 +181,140 @@
options={{
right: 'Skip failures'
}}
class="whitespace-nowrap"
/>
</div>
<div>
<div class="flex-shrink-0">
<div class="mb-2 text-sm font-bold">Run in parallel</div>
<Toggle
bind:checked={mod.value.parallel}
options={{
right: 'All iterations run in parallel'
}}
class="whitespace-nowrap"
/>
</div>
<div>
<div class="flex-shrink-0">
<div class="mb-2 text-sm font-bold"
>Parallelism <Tooltip
>Assign a maximum number of branches run in parallel to control huge for-loops.</Tooltip
>
</div>
<input
type="number"
disabled={!mod.value.parallel}
bind:value={mod.value.parallelism}
/>
<div class="flex gap-2 items-center">
<input
type="number"
min="1"
class="w-20 px-2 py-1 text-sm border border-gray-200 dark:border-gray-700 rounded bg-surface"
disabled={!mod.value.parallel || parallelismType === 'javascript'}
placeholder={parallelismType === 'javascript' ? 'Expression' : ''}
bind:value={
() => {
const parallelismExpr = (mod.value as ForloopFlow).parallelism
return parallelismExpr && parallelismExpr.type === 'static'
? parallelismExpr.value
: ''
},
(value) => {
;(mod.value as ForloopFlow).parallelism = {
type: 'static',
value
}
}
}
/>
<ToggleButtonGroup
disabled={!mod.value.parallel}
bind:selected={parallelismType}
on:selected={(e) => {
const forLoopFlow = mod.value as ForloopFlow
if (e.detail == parallelismType) return
if (e.detail === 'javascript') {
if (!forLoopFlow.parallelism || forLoopFlow.parallelism.type !== 'javascript') {
;(mod.value as ForloopFlow).parallelism = {
type: 'javascript',
expr: ''
}
}
} else {
if (!forLoopFlow.parallelism || forLoopFlow.parallelism.type !== 'static') {
;(mod.value as ForloopFlow).parallelism = {
type: 'static',
value: 0
}
}
}
}}
class="h-6"
>
{#snippet children({ item })}
<ToggleButton light small label="static" value="static" {item} />
<ToggleButton
small
light
tooltip="JavaScript expression ('flow_input' or 'results')."
value="javascript"
icon={FunctionSquare}
{item}
/>
{/snippet}
</ToggleButtonGroup>
</div>
</div>
</div>
{#if mod.value.type === 'forloopflow' && mod.value.parallel && mod.value.parallelism?.type == 'javascript'}
<div class="my-2 flex flex-row gap-2 items-center">
<div class="text-sm font-bold whitespace-nowrap">
Parallelism expression
<Tooltip>
JavaScript expression that defines the maximum number of parallel executions.
Example: flow_input.max_parallel || 3
</Tooltip>
</div>
</div>
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class="border w-full mb-2 h-full max-h-[250px]"
id="flow-editor-parallel-expression"
transition:slide={{ duration: 300 }}
>
<PropPickerWrapper
notSelectable
flow_input={stepPropPicker.pickableProperties.flow_input}
pickableProperties={stepPropPicker.pickableProperties}
on:select={({ detail }) => {
parallelismEditor?.insertAtCursor(detail)
parallelismEditor?.focus()
}}
>
<SimpleEditor
bind:this={parallelismEditor}
autofocus
lang="javascript"
bind:code={
() => {
const parallelismExpr = (mod.value as ForloopFlow).parallelism
return parallelismExpr && parallelismExpr.type === 'javascript'
? parallelismExpr.expr
: ''
},
(expr) => {
;(mod.value as ForloopFlow).parallelism = {
type: 'javascript',
expr
}
}
}
class="small-editor"
shouldBindKey={false}
extraLib={stepPropPicker.extraLib}
/>
</PropPickerWrapper>
</div>
{/if}
<div class="my-2 flex flex-row gap-2 items-center">
<div class="text-sm font-bold whitespace-nowrap">
Iterator expression
openflow.openapi.yaml
+2 -2
View File
@@ -373,7 +373,7 @@ components:
parallel:
type: boolean
parallelism:
type: integer
$ref: "#/components/schemas/InputTransform"
required:
- modules
- iterator
@@ -396,7 +396,7 @@ components:
parallel:
type: boolean
parallelism:
type: integer
$ref: "#/components/schemas/InputTransform"
required:
- modules
- skip_failures