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

feat: concurrency limits for flows

Browse Source
This commit is contained in:
Ruben Fiszel
2023-08-28 16:02:40 +02:00
parent 9264d4f046
commit 11c8f90daf
7 changed files with 67 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/windmill-api/src/flows.rs
View File

@@ -749,6 +749,8 @@ mod tests {
timeout: None,
}),
same_worker: false,
concurrent_limit: None,
concurrency_time_window_s: None,
};
let expect = serde_json::json!({
"modules": [

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

@@ -76,6 +76,10 @@ pub struct FlowValue {
#[serde(default)]
#[serde(skip_serializing_if = "is_default")]
pub same_worker: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrent_limit: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub concurrency_time_window_s: Option<i32>,
}
#[derive(Deserialize, Serialize, Debug, Clone)]

31
backend/windmill-queue/src/jobs.rs
View File

@@ -145,16 +145,18 @@ pub async fn cancel_job<'c: 'async_recursion>(
|| (job_running.job_kind == JobKind::Flow || job_running.job_kind == JobKind::FlowPreview))
&& !force_cancel
{
sqlx::query!(
"UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 \
AND workspace_id = $4 ",
let id = sqlx::query_scalar!(
"UPDATE queue SET canceled = true, canceled_by = $1, canceled_reason = $2, scheduled_for = now(), suspend = 0 WHERE id = $3 AND workspace_id = $4 RETURNING id",
username,
reason,
id,
w_id
)
.execute(&mut *tx)
.fetch_optional(&mut *tx)
.await?;
if let Some(id) = id {
tracing::info!("Soft cancelling job {}", id);
}
} else {
let reason = reason
.clone()
@@ -953,7 +955,10 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
// concurrency check. If more than X jobs for this path are already running, we re-queue and pull another job from the queue
let pulled_job = job.unwrap();
if pulled_job.script_path.is_none() || pulled_job.concurrent_limit.is_none() {
if pulled_job.script_path.is_none()
|| pulled_job.concurrent_limit.is_none()
|| pulled_job.canceled
{
if *METRICS_ENABLED {
QUEUE_PULL_COUNT.inc();
}
@@ -979,12 +984,12 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
FROM
(SELECT script_path, MIN(started_at) as min_started_at, COUNT(*) as completed_count
FROM completed_job
WHERE script_path = $1 AND started_at + INTERVAL '1 MILLISECOND' * duration_ms > (now() - INTERVAL '1 second' * $2) AND workspace_id = $3
WHERE script_path = $1 AND job_kind != 'dependencies' AND started_at + INTERVAL '1 MILLISECOND' * duration_ms > (now() - INTERVAL '1 second' * $2) AND workspace_id = $3 AND canceled = false
GROUP BY script_path) as j
FULL OUTER JOIN
(SELECT script_path, MIN(started_at) as min_started_at, COUNT(*) as running_count
FROM queue
WHERE script_path = $1 AND running = true AND workspace_id = $3
WHERE script_path = $1 AND job_kind != 'dependencies' AND running = true AND workspace_id = $3 AND canceled = false
GROUP BY script_path) as q
ON q.script_path = j.script_path",
job_script_path,
@@ -1509,10 +1514,10 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
path,
None,
JobKind::FlowPreview,
Some(value),
None,
None,
Some(value.clone()),
None,
value.concurrent_limit.clone(),
value.concurrency_time_window_s,
),
JobPayload::Flow(flow) => {
let value_json = fetch_scalar_isolated!(
@@ -1534,10 +1539,10 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
Some(flow),
None,
JobKind::Flow,
Some(value),
None,
None,
Some(value.clone()),
None,
value.concurrent_limit.clone(),
value.concurrency_time_window_s,
)
}
JobPayload::Identity => (None, None, None, JobKind::Identity, None, None, None, None),

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

@@ -1376,6 +1376,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
modules: (*modules).clone(),
failure_module: fm.clone(),
same_worker: flow.same_worker,
concurrent_limit: None,
concurrency_time_window_s: None,
},
path: Some(format!("{}/forloop", flow_job.script_path())),
},
@@ -1891,6 +1893,8 @@ async fn compute_next_flow_transform(
modules: (*modules).clone(),
failure_module: fm,
same_worker: flow.same_worker,
concurrent_limit: None,
concurrency_time_window_s: None,
},
path: Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)),
},
@@ -1962,6 +1966,8 @@ async fn compute_next_flow_transform(
modules,
failure_module: fm,
same_worker: flow.same_worker,
concurrent_limit: None,
concurrency_time_window_s: None,
},
path: Some(format!(
"{}/branchone-{}",
@@ -2002,6 +2008,8 @@ async fn compute_next_flow_transform(
modules: b.modules.clone(),
failure_module: fm.clone(),
same_worker: flow.same_worker,
concurrent_limit: None,
concurrency_time_window_s: None,
},
path: Some(format!(
"{}/branchall-{}",
@@ -2073,6 +2081,8 @@ async fn compute_next_flow_transform(
modules,
failure_module: fm.clone(),
same_worker: flow.same_worker,
concurrent_limit: None,
concurrency_time_window_s: None,
},
path: Some(format!(
"{}/branchall-{}",

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

@@ -53,7 +53,7 @@
</div>
{:else if job && 'running' in job && 'scheduled_for' in job && job.scheduled_for && forLater(job.scheduled_for)}
<div>
<Badge>
<Badge color="blue">
<Icon data={faCalendar} scale={SMALL_ICON_SCALE} class="mr-2" />
Scheduled for {displayDate(job.scheduled_for)}
</Badge>

29
frontend/src/lib/components/flows/content/FlowSettings.svelte
View File

@@ -9,7 +9,7 @@
import FlowCard from '../common/FlowCard.svelte'
import FlowSchedules from './FlowSchedules.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import { Alert } from '$lib/components/common'
import { Alert, Button, SecondsInput } from '$lib/components/common'
import { getContext } from 'svelte'
import type { FlowEditorContext } from '../types'
import autosize from 'svelte-autosize'
@@ -51,6 +51,7 @@
<Tab value="settings-schedule">Schedule</Tab>
<Tab value="settings-same-worker">Shared Directory</Tab>
<Tab value="settings-worker-group">Worker Group</Tab>
<Tab value="settings-concurrency">Concurrency</Tab>
<svelte:fragment slot="content">
<TabContent value="settings-metadata" class="p-4 h-full">
@@ -270,6 +271,32 @@
<Loader2 class="animate-spin" />
{/if}
</TabContent>
<TabContent value="settings-concurrency" class="p-4 flex flex-col">
<div>
<h2 class="pb-4">
Concurrency Limits
<Tooltip>Allowed concurrency within a given timeframe</Tooltip>
</h2>
</div>
<div>
<div class="text-xs font-bold !mt-2"
>Max number of executions within the time window</div
>
<div class="flex flex-row gap-2 max-w-sm"
><input bind:value={$flowStore.value.concurrent_limit} type="number" />
<Button
size="sm"
color="light"
on:click={() => {
$flowStore.value.concurrent_limit = undefined
}}
variant="border">Remove Limits</Button
></div
>
<div class="text-xs font-bold !mt-2">Time window in seconds</div>
<SecondsInput bind:seconds={$flowStore.value.concurrency_time_window_s} />
</div>
</TabContent>
</svelte:fragment>
</Tabs>
</div>

5
openflow.openapi.yaml
View File

@@ -46,7 +46,10 @@ components:
$ref: "#/components/schemas/FlowModule"
same_worker:
type: boolean
concurrent_limit:
type: number
concurrency_time_window_s:
type: number
required:
- modules