From ddd3e46afa0239e113fd9044ef04367fd69e41f3 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 29 Sep 2024 13:16:45 +0200 Subject: [PATCH] optimize flow update parallel queries --- ...60e26e644a9d939e74a8317c4a09335f110fe.json | 25 +++++ ...82c1df21ba601ac0989e175f834c569962d46.json | 23 +++++ ...d7db9b956abd88b94db3948f2c579c3826d0.json} | 4 +- ...be86b869b01ca1ab10aa17634bb6dae879f12.json | 14 --- ...59da0ef78b5252b6ac669a3c9e545d05c0d43.json | 23 ----- ...42e7285f4266a1471e5ffdefadf421a67e44b.json | 25 +++++ backend/windmill-worker/src/worker_flow.rs | 98 +++++++++++-------- 7 files changed, 131 insertions(+), 81 deletions(-) create mode 100644 backend/.sqlx/query-1e188d8e427cab25dbe18aa900260e26e644a9d939e74a8317c4a09335f110fe.json create mode 100644 backend/.sqlx/query-6e7f234267fbb4720b29f288fba82c1df21ba601ac0989e175f834c569962d46.json rename backend/.sqlx/{query-a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c.json => query-777190559e27c8c8fb6718b0a0c1d7db9b956abd88b94db3948f2c579c3826d0.json} (60%) delete mode 100644 backend/.sqlx/query-d0df57fc4cd0be7b541dee081ffbe86b869b01ca1ab10aa17634bb6dae879f12.json delete mode 100644 backend/.sqlx/query-d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43.json create mode 100644 backend/.sqlx/query-f916ec232837ece9323675e5f5142e7285f4266a1471e5ffdefadf421a67e44b.json diff --git a/backend/.sqlx/query-1e188d8e427cab25dbe18aa900260e26e644a9d939e74a8317c4a09335f110fe.json b/backend/.sqlx/query-1e188d8e427cab25dbe18aa900260e26e644a9d939e74a8317c4a09335f110fe.json new file mode 100644 index 0000000000..01fb19b1c2 --- /dev/null +++ b/backend/.sqlx/query-1e188d8e427cab25dbe18aa900260e26e644a9d939e74a8317c4a09335f110fe.json @@ -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" +} diff --git a/backend/.sqlx/query-6e7f234267fbb4720b29f288fba82c1df21ba601ac0989e175f834c569962d46.json b/backend/.sqlx/query-6e7f234267fbb4720b29f288fba82c1df21ba601ac0989e175f834c569962d46.json new file mode 100644 index 0000000000..e8db78adbd --- /dev/null +++ b/backend/.sqlx/query-6e7f234267fbb4720b29f288fba82c1df21ba601ac0989e175f834c569962d46.json @@ -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" +} diff --git a/backend/.sqlx/query-a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c.json b/backend/.sqlx/query-777190559e27c8c8fb6718b0a0c1d7db9b956abd88b94db3948f2c579c3826d0.json similarity index 60% rename from backend/.sqlx/query-a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c.json rename to backend/.sqlx/query-777190559e27c8c8fb6718b0a0c1d7db9b956abd88b94db3948f2c579c3826d0.json index 5906375f53..0c1d65bb6b 100644 --- a/backend/.sqlx/query-a94dbd1f7aab20682548471c5cc06c7a496edac36e3de537b00b76ad93c7556c.json +++ b/backend/.sqlx/query-777190559e27c8c8fb6718b0a0c1d7db9b956abd88b94db3948f2c579c3826d0.json @@ -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" } diff --git a/backend/.sqlx/query-d0df57fc4cd0be7b541dee081ffbe86b869b01ca1ab10aa17634bb6dae879f12.json b/backend/.sqlx/query-d0df57fc4cd0be7b541dee081ffbe86b869b01ca1ab10aa17634bb6dae879f12.json deleted file mode 100644 index 29227c3d8c..0000000000 --- a/backend/.sqlx/query-d0df57fc4cd0be7b541dee081ffbe86b869b01ca1ab10aa17634bb6dae879f12.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43.json b/backend/.sqlx/query-d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43.json deleted file mode 100644 index 7d7274d774..0000000000 --- a/backend/.sqlx/query-d6c25421bb6513908697ebe74c159da0ef78b5252b6ac669a3c9e545d05c0d43.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-f916ec232837ece9323675e5f5142e7285f4266a1471e5ffdefadf421a67e44b.json b/backend/.sqlx/query-f916ec232837ece9323675e5f5142e7285f4266a1471e5ffdefadf421a67e44b.json new file mode 100644 index 0000000000..2cf1fb31e9 --- /dev/null +++ b/backend/.sqlx/query-f916ec232837ece9323675e5f5142e7285f4266a1471e5ffdefadf421a67e44b.json @@ -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" +} diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index af9976c707..6474af6d46 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -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, job_id_for_status: &Uuid) -> Option { + 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>>, flow_jobs: &Vec, @@ -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",