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

optimize flow update parallel queries

Browse Source
This commit is contained in:
Ruben Fiszel
2024-09-29 13:16:45 +02:00
parent 827113c797
commit ddd3e46afa
7 changed files with 131 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-1e188d8e427cab25dbe18aa900260e26e644a9d939e74a8317c4a09335f110fe.json
Generated
+25
View File
@@ -0,0 +1,25 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue\n SET flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb),\n last_ping = NULL\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "int4",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Int4",
"Uuid",
"Text",
"Jsonb"
]
},
"nullable": [
null
]
},
"hash": "1e188d8e427cab25dbe18aa900260e26e644a9d939e74a8317c4a09335f110fe"
}
backend/.sqlx/query-6e7f234267fbb4720b29f288fba82c1df21ba601ac0989e175f834c569962d46.json
Generated
+23
View File
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb),\n last_ping = NULL\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "int4",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Int4",
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "6e7f234267fbb4720b29f288fba82c1df21ba601ac0989e175f834c569962d46"
}
backend/.sqlx/query-a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c.json → backend/.sqlx/query-777190559e27c8c8fb6718b0a0c1d7db9b956abd88b94db3948f2c579c3826d0.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb)\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
"query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb),\n last_ping = NULL\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
"describe": {
"columns": [
{
@@ -19,5 +19,5 @@
null
]
},
"hash": "a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c"
"hash": "777190559e27c8c8fb6718b0a0c1d7db9b956abd88b94db3948f2c579c3826d0"
}
backend/.sqlx/query-d0df57fc4cd0be7b541dee081ffbe86b869b01ca1ab10aa17634bb6dae879f12.json
Generated
-14
View File
@@ -1,14 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue\n SET last_ping = null\n WHERE id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "d0df57fc4cd0be7b541dee081ffbe86b869b01ca1ab10aa17634bb6dae879f12"
}
backend/.sqlx/query-d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43.json
Generated
-23
View File
@@ -1,23 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue\n SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb)\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "int4",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Int4",
"Uuid"
]
},
"nullable": [
null
]
},
"hash": "d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43"
}
backend/.sqlx/query-f916ec232837ece9323675e5f5142e7285f4266a1471e5ffdefadf421a67e44b.json
Generated
+25
View File
@@ -0,0 +1,25 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue\n SET flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'iterator', 'index'],\n ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n ),\n last_ping = NULL\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "int4",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Int4",
"Uuid",
"Text",
"Jsonb"
]
},
"nullable": [
null
]
},
"hash": "f916ec232837ece9323675e5f5142e7285f4266a1471e5ffdefadf421a67e44b"
}
backend/windmill-worker/src/worker_flow.rs
+56 -42
View File
@@ -436,25 +436,38 @@ pub async fn update_flow_status_after_job_completion_internal<
} if *parallel => {
let (nindex, len) = match (iterator, branchall) {
(Some(Iterator { itered, .. }), _) => {
set_success_in_flow_job_success(
flow_jobs_success,
jobs,
job_id_for_status,
&old_status,
flow,
success,
&mut tx,
)
.await?;
let position = if flow_jobs_success.is_some() {
find_flow_job_index(jobs, job_id_for_status)
} else {
None
};
let nindex = sqlx::query_scalar!(
let nindex = if let Some(position) = position {
sqlx::query_scalar!(
"UPDATE queue
SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb)
SET flow_status = JSONB_SET(
JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),
ARRAY['modules', $1::TEXT, 'iterator', 'index'],
((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb
),
last_ping = NULL
WHERE id = $2
RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
old_status.step,
flow
)
flow,
position as i32,
json!(success)
)} else {
sqlx::query_scalar!(
"UPDATE queue
SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb),
last_ping = NULL
WHERE id = $2
RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int",
old_status.step,
flow
)
}
.fetch_one(&mut tx)
.await.map_err(|e| {
Error::InternalErr(format!(
@@ -470,24 +483,35 @@ pub async fn update_flow_status_after_job_completion_internal<
(nindex, itered.len() as i32)
}
(_, Some(BranchAllStatus { len, .. })) => {
set_success_in_flow_job_success(
flow_jobs_success,
jobs,
job_id_for_status,
&old_status,
flow,
success,
&mut tx,
)
.await?;
let nindex = sqlx::query_scalar!(
let position = if flow_jobs_success.is_some() {
find_flow_job_index(jobs, job_id_for_status)
} else {
None
};
let nindex = if let Some(position) = position {
sqlx::query_scalar!(
"UPDATE queue
SET flow_status = JSONB_SET(
JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),
ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb),
last_ping = NULL
WHERE id = $2
RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
old_status.step,
flow,
position as i32,
json!(success)
)
} else { sqlx::query_scalar!(
"UPDATE queue
SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb)
SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb),
last_ping = NULL
WHERE id = $2
RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int",
old_status.step,
flow
)
)}
.fetch_one(&mut tx)
.await
.map_err(|e| {
@@ -586,18 +610,6 @@ pub async fn update_flow_status_after_job_completion_internal<
})?;
}
sqlx::query!(
"UPDATE queue
SET last_ping = null
WHERE id = $1",
flow
)
.execute(db)
.await
.map_err(|e| {
Error::InternalErr(format!("error while setting last ping to null: {e:#}"))
})?;
let r = sqlx::query_scalar!(
"DELETE FROM parallel_monitor_lock WHERE parent_flow_id = $1 and job_id = $2 RETURNING last_ping",
flow,
@@ -1146,6 +1158,10 @@ pub async fn update_flow_status_after_job_completion_internal<
}
}
fn find_flow_job_index(flow_jobs: &Vec<Uuid>, job_id_for_status: &Uuid) -> Option<usize> {
flow_jobs.iter().position(|x| x == job_id_for_status)
}
async fn set_success_in_flow_job_success<'c, R: rsmq_async::RsmqConnection + Send>(
flow_jobs_success: &Option<Vec<Option<bool>>>,
flow_jobs: &Vec<Uuid>,
@@ -1155,10 +1171,8 @@ async fn set_success_in_flow_job_success<'c, R: rsmq_async::RsmqConnection + Sen
success: bool,
tx: &mut QueueTransaction<'c, R>,
) -> error::Result<()> {
let flow_jobs_success = flow_jobs_success.clone();
if flow_jobs_success.is_some() {
let position = flow_jobs.iter().position(|x| x == job_id_for_status);
let position = find_flow_job_index(flow_jobs, job_id_for_status);
if let Some(position) = position {
sqlx::query!(
"UPDATE queue SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4) WHERE id = $2",