From 11c8f90daf89422d5dc71b38b6ed87f35cf48bc1 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 28 Aug 2023 16:02:40 +0200 Subject: [PATCH] feat: concurrency limits for flows --- backend/windmill-api/src/flows.rs | 2 ++ backend/windmill-common/src/flows.rs | 4 +++ backend/windmill-queue/src/jobs.rs | 31 +++++++++++-------- backend/windmill-worker/src/worker_flow.rs | 10 ++++++ frontend/src/lib/components/JobStatus.svelte | 2 +- .../flows/content/FlowSettings.svelte | 29 ++++++++++++++++- openflow.openapi.yaml | 5 ++- 7 files changed, 67 insertions(+), 16 deletions(-) diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 9d210455d0..ef4addf56b 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -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": [ diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 0bda87ed59..4b270fb9b7 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -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, + #[serde(skip_serializing_if = "Option::is_none")] + pub concurrency_time_window_s: Option, } #[derive(Deserialize, Serialize, Debug, Clone)] diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 336fb2c119..53c71ffacc 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -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( // 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( 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), diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index fad83bc11a..1b8de80866 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1376,6 +1376,8 @@ async fn push_next_flow_job 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-{}", diff --git a/frontend/src/lib/components/JobStatus.svelte b/frontend/src/lib/components/JobStatus.svelte index 0c8ecb90be..b82393a607 100644 --- a/frontend/src/lib/components/JobStatus.svelte +++ b/frontend/src/lib/components/JobStatus.svelte @@ -53,7 +53,7 @@ {:else if job && 'running' in job && 'scheduled_for' in job && job.scheduled_for && forLater(job.scheduled_for)}
- + Scheduled for {displayDate(job.scheduled_for)} diff --git a/frontend/src/lib/components/flows/content/FlowSettings.svelte b/frontend/src/lib/components/flows/content/FlowSettings.svelte index 5fed564b41..205a2b26fb 100644 --- a/frontend/src/lib/components/flows/content/FlowSettings.svelte +++ b/frontend/src/lib/components/flows/content/FlowSettings.svelte @@ -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 @@ Schedule Shared Directory Worker Group + Concurrency @@ -270,6 +271,32 @@ {/if} + +
+

+ Concurrency Limits + Allowed concurrency within a given timeframe +

+
+
+
Max number of executions within the time window
+
+
+
Time window in seconds
+ +
+
diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 774fb41d3d..bbe8aa8b96 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -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