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

fix: respect FIFO order for concurrency limit

Browse Source
This commit is contained in:
Ruben Fiszel
2023-07-24 18:11:23 +02:00
parent 0fc8f0bb44
commit 0f48c19d5a
6 changed files with 42 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20230724145302_improve_pull_with_index.down.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add down migration script here
DROP INDEX index_queue_on_main_filter;
backend/migrations/20230724145302_improve_pull_with_index.up.sql
+1
View File
@@ -0,0 +1 @@
-- Add up migration script here
backend/src/main.rs
+6 -2
View File
@@ -160,8 +160,6 @@ Windmill Community Edition {GIT_VERSION}
"PIP_LOCAL_DEPENDENCIES",
"ADDITIONAL_PYTHON_PATHS",
"INCLUDE_HEADERS",
"WHITELIST_WORKSPACES",
"BLACKLIST_WORKSPACES",
"INSTANCE_EVENTS_WEBHOOK",
"CLOUD_HOSTED",
"GLOBAL_CACHE_INTERVAL",
@@ -181,6 +179,12 @@ Windmill Community Edition {GIT_VERSION}
"GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE",
]);
if std::env::var("WHITELIST_WORKSPACES").is_ok()
|| std::env::var("BLACKLIST_WORKSPACES").is_ok()
{
panic!("WHITELIST_WORKSPACES and BLACKLIST_WORKSPACES have been removed, please use Worker Groups instead");
}
tracing::info!("Loading OAuth providers...: {:#?}", *OAUTH_CLIENTS);
if let Some(ref smtp) = *SMTP_CLIENT {
tracing::info!("Smtp client defined. Testing connection...");
backend/windmill-queue/src/jobs.rs
+29 -64
View File
@@ -82,8 +82,30 @@ lazy_static::lazy_static! {
"hub".to_string(),
"other".to_string()]);
pub static ref ACCEPTED_TAGS_FILTER: String = format!(" AND ({})",
ACCEPTED_TAGS.clone().into_iter().map(|x| format!("(tag = '{x}')")).join(" OR "));
pub static ref PULL_QUERY: String = format!(
"UPDATE queue
SET running = true
, started_at = coalesce(started_at, now())
, last_ping = now()
, suspend_until = null
WHERE id = (
SELECT id
FROM queue
WHERE ((running = false
AND scheduled_for <= now())
OR (suspend_until IS NOT NULL
AND ( suspend <= 0
OR suspend_until <= now())))
{}
ORDER BY scheduled_for, created_at
FOR UPDATE SKIP LOCKED
LIMIT 1
) LIMIT 1
RETURNING *",
format!(" AND ({})",
ACCEPTED_TAGS.clone().into_iter().map(|x| format!("(tag = '{x}')")).join(" OR "))
);
// When compiled in 'benchmark' mode, this flags is exposed via the /workers/toggle endpoint
// and make it possible to disable to current active workers (such that they don't pull any)
@@ -660,49 +682,15 @@ async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c
pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
db: &Pool<Postgres>,
whitelist_workspaces: Option<Vec<String>>,
blacklist_workspaces: Option<Vec<String>>,
rsmq: Option<R>,
) -> windmill_common::error::Result<Option<QueuedJob>> {
let mut workspaces_filter = String::new();
if let Some(whitelist) = whitelist_workspaces {
workspaces_filter.push_str(&format!(
" AND workspace_id IN ({})",
whitelist
.into_iter()
.map(|x| format!("'{x}'"))
.collect::<Vec<String>>()
.join(",")
));
if let Some(_rsmq) = rsmq {
todo!("REDIS: Implement workspace filters for redis");
}
}
if let Some(blacklist) = blacklist_workspaces {
workspaces_filter.push_str(&format!(
" AND workspace_id NOT IN ({})",
blacklist
.into_iter()
.map(|x| format!("'{x}'"))
.collect::<Vec<String>>()
.join(",")
));
if let Some(_rsmq) = rsmq {
todo!("REDIS: Implement workspace filters for redis");
}
}
// let rs = rd_string(2);
// let instant = Instant::now();
loop {
let tx: QueueTransaction<'_, _> = (rsmq.clone(), db.clone().begin().await?).into();
let (job, mut tx) = pull_single_job_and_mark_as_running_no_concurrency_limit(
tx,
workspaces_filter.as_str(),
rsmq.clone(),
)
.await?;
let (job, mut tx) =
pull_single_job_and_mark_as_running_no_concurrency_limit(tx, rsmq.clone()).await?;
if job.is_none() {
return Ok(None);
@@ -842,7 +830,6 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
R: rsmq_async::RsmqConnection + Send + Clone,
>(
mut tx: QueueTransaction<'c, R>,
workspaces_filter: &str,
rsmq: Option<R>,
) -> windmill_common::error::Result<(Option<QueuedJob>, QueueTransaction<'c, R>)> {
let job: Option<QueuedJob> = if let Some(mut rsmq) = rsmq {
@@ -876,7 +863,6 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
None
}
} else {
let accepted_tags_filter = &*ACCEPTED_TAGS_FILTER;
/* Jobs can be started if they:
* - haven't been started before,
* running = false
@@ -884,30 +870,9 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
* suspend_until is non-null
* and suspend = 0 when the resume messages are received
* or suspend_until <= now() if it has timed out */
sqlx::query_as::<_, QueuedJob>(&format!(
"UPDATE queue
SET running = true
, started_at = coalesce(started_at, now())
, last_ping = now()
, suspend_until = null
WHERE id = (
SELECT id
FROM queue
WHERE ((running = false
AND scheduled_for <= now())
OR (suspend_until IS NOT NULL
AND ( suspend <= 0
OR suspend_until <= now())))
{workspaces_filter}
{accepted_tags_filter}
ORDER BY scheduled_for
FOR UPDATE SKIP LOCKED
LIMIT 1
)
RETURNING *"
))
.fetch_optional(&mut tx)
.await?
sqlx::query_as::<_, QueuedJob>(&PULL_QUERY)
.fetch_optional(&mut tx)
.await?
};
Ok((job, tx))
}
backend/windmill-worker/src/worker.rs
+1 -8
View File
@@ -212,13 +212,6 @@ lazy_static::lazy_static! {
.ok()
.map(|x| x.split(',').map(|x| (x.to_string(), std::env::var(x).unwrap_or("".to_string()))).collect());
static ref WHITELIST_WORKSPACES: Option<Vec<String>> = std::env::var("WHITELIST_WORKSPACES")
.ok()
.map(|x| x.split(',').map(|x| x.to_string()).collect());
static ref BLACKLIST_WORKSPACES: Option<Vec<String>> = std::env::var("BLACKLIST_WORKSPACES")
.ok()
.map(|x| x.split(',').map(|x| x.to_string()).collect());
pub static ref TAR_CACHE_RATE: i32 = std::env::var("TAR_CACHE_RATE")
.ok()
.and_then(|x| x.parse::<i32>().ok())
@@ -610,7 +603,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
},
(job, timer) = {
let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None };
pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone(), rsmq.clone()).map(|x| (x, timer))
pull(&db, rsmq.clone()).map(|x| (x, timer))
} => {
timer.map(|timer| {
let duration_pull_s = timer.stop_and_record();
benchmarks/worker.ts
+2 -5
View File
@@ -272,16 +272,14 @@ while (cont) {
language: "postgresql",
args: {
query: "SELECT email FROM usr",
database_url:
"postgres://postgres:changeme@localhost:5432/windmill",
database_url: "postgres://postgres:changeme@localhost:5432/windmill",
},
};
} else {
payload = {
path: "denosimple",
language: api.Preview.language.DENO,
content:
'export function main(){ return Deno.env.get("WM_JOB_ID"); }',
content: 'export function main(){ return Deno.env.get("WM_JOB_ID"); }',
args: {},
};
}
@@ -300,7 +298,6 @@ while (cont) {
clearInterval(updateStatusInterval);
const end_time = Date.now() + complete_timeout;
let incorrect_results = 0;