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

fix: improve que job indices for faster performances

Browse Source
This commit is contained in:
Ruben Fiszel
2025-02-17 21:26:46 +01:00
committed by Guilhem
parent 731802a186
commit be1fcc016e
10 changed files with 43 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-19cc8499f682ec34d54bc4f694cb281a9bd7f5431c646c6268513751fff95395.json → backend/.sqlx/query-0cb0e912bc942af2b1ef784455f3f073a79e300f3dd48f14122d1782eee663cd.json
Generated
+4 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT coalesce(COUNT(*) FILTER(WHERE suspend = 0 AND running = false), 0) as \"database_length!\", coalesce(COUNT(*) FILTER(WHERE suspend > 0), 0) as \"suspended!\" FROM v2_as_queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now()",
"query": "SELECT coalesce(COUNT(*) FILTER(WHERE suspend = 0 AND running = false), 0) as \"database_length!\", coalesce(COUNT(*) FILTER(WHERE suspend > 0), 0) as \"suspended!\" FROM v2_as_queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now() AND ($3::text[] IS NULL OR tag = ANY($3))",
"describe": {
"columns": [
{
@@ -17,7 +17,8 @@
"parameters": {
"Left": [
"Text",
"Bool"
"Bool",
"TextArray"
]
},
"nullable": [
@@ -25,5 +26,5 @@
null
]
},
"hash": "19cc8499f682ec34d54bc4f694cb281a9bd7f5431c646c6268513751fff95395"
"hash": "0cb0e912bc942af2b1ef784455f3f073a79e300f3dd48f14122d1782eee663cd"
}
backend/.sqlx/query-47455b0ebaf999ab58b2cba3d74cb4bdd64075939a4aa2c117e18372511ea7e0.json
Generated
-12
View File
@@ -1,12 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "VACUUM (skip_locked) v2_job_queue, v2_job_runtime, v2_job_status",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "47455b0ebaf999ab58b2cba3d74cb4bdd64075939a4aa2c117e18372511ea7e0"
}
backend/.sqlx/query-b9b38d63af3670d1f11d5cbb82a8008a9479bf4b3d7231371ebf26382ecde365.json
Generated
+12
View File
@@ -0,0 +1,12 @@
{
"db_name": "PostgreSQL",
"query": "VACUUM v2_job_queue, v2_job_runtime, v2_job_status",
"describe": {
"columns": [],
"parameters": {
"Left": []
},
"nullable": []
},
"hash": "b9b38d63af3670d1f11d5cbb82a8008a9479bf4b3d7231371ebf26382ecde365"
}
backend/.sqlx/query-ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927.json
Generated
+1 -1
View File
@@ -15,7 +15,7 @@
]
},
"nullable": [
null
true
]
},
"hash": "ddf2eccb78a310ed00c7d8b9c3f05d394a7cbcf0038c72a78add5c7b02ef5927"
backend/windmill-api/src/jobs.rs
+6 -1
View File
@@ -1647,6 +1647,7 @@ struct QueueStats {
#[derive(Deserialize)]
pub struct CountQueueJobsQuery {
all_workspaces: Option<bool>,
tags: Option<String>,
}
async fn count_queue_jobs(
@@ -1654,12 +1655,16 @@ async fn count_queue_jobs(
Path(w_id): Path<String>,
Query(cq): Query<CountQueueJobsQuery>,
) -> error::JsonResult<QueueStats> {
let tags = cq
.tags
.map(|t| t.split(',').map(|s| s.to_string()).collect::<Vec<_>>());
Ok(Json(
sqlx::query_as!(
QueueStats,
"SELECT coalesce(COUNT(*) FILTER(WHERE suspend = 0 AND running = false), 0) as \"database_length!\", coalesce(COUNT(*) FILTER(WHERE suspend > 0), 0) as \"suspended!\" FROM v2_as_queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now()",
"SELECT coalesce(COUNT(*) FILTER(WHERE suspend = 0 AND running = false), 0) as \"database_length!\", coalesce(COUNT(*) FILTER(WHERE suspend > 0), 0) as \"suspended!\" FROM v2_as_queue WHERE (workspace_id = $1 OR $2) AND scheduled_for <= now() AND ($3::text[] IS NULL OR tag = ANY($3))",
w_id,
w_id == "admins" && cq.all_workspaces.unwrap_or(false),
tags.as_ref().map(|v| v.as_slice())
)
.fetch_one(&db)
.await?,
backend/windmill-common/src/lib.rs
+3 -3
View File
@@ -292,9 +292,9 @@ pub async fn connect(
// .after_connect(move |conn, _| {
// if worker_mode {
// Box::pin(async move {
// sqlx::query("SET enable_seqscan = OFF;")
// .execute(conn)
// .await?;
// // sqlx::query("SET enable_seqscan = OFF;")
// // .execute(conn)
// // .await?;
// Ok(())
// })
// } else {
backend/windmill-queue/src/jobs.rs
+3
View File
@@ -2108,12 +2108,15 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<'c>(
for query in queries.iter() {
// tracing::info!("Pulling job with query: {}", query);
// let instant = std::time::Instant::now();
let r = sqlx::query_as::<_, PulledJob>(query)
.bind(worker_name)
.fetch_optional(db)
.await?;
if let Some(pulled_job) = r {
// tracing::info!("pulled job: {:?}", instant.elapsed().as_micros());
highest_priority_job = Some(pulled_job);
break;
}
backend/windmill-worker/src/worker.rs
+1 -1
View File
@@ -1277,7 +1277,7 @@ pub async fn run_worker(
tokio::task::spawn(
(async move {
tracing::info!(worker = %worker_name, hostname = %hostname, "vacuuming queue");
if let Err(e) = sqlx::query!("VACUUM (skip_locked) v2_job_queue, v2_job_runtime, v2_job_status")
if let Err(e) = sqlx::query!("VACUUM v2_job_queue, v2_job_runtime, v2_job_status")
.execute(&db2)
.await
{
benchmarks/benchmark_oneoff.ts
+12 -10
View File
@@ -37,6 +37,7 @@ async function verifyOutputs(uuids: string[], workspace: string) {
console.log(`Incorrect results: ${incorrectResults}`);
}
export const NON_TEST_TAGS = ["deno", "python", "go", "bash", "dedicated", "bun", "nativets", "flow"]
export async function main({
host,
email,
@@ -96,11 +97,11 @@ export async function main({
windmill.setClient(final_token, host);
const enc = (s: string) => new TextEncoder().encode(s);
async function getQueueCount() {
async function getQueueCount(tags?: string[]) {
return (
await (
await fetch(
config.server + "/api/w/" + config.workspace_id + "/jobs/queue/count",
config.server + "/api/w/" + config.workspace_id + "/jobs/queue/count" + (tags && tags.length > 0 ? "?tags=" + tags.join(",") : ""),
{ headers: { ["Authorization"]: "Bearer " + config.token } }
)
).json()
@@ -132,11 +133,11 @@ export async function main({
}
let pastJobs = 0;
async function getCompletedJobsCount(): Promise<number> {
async function getCompletedJobsCount(tags?: string[]): Promise<number> {
const completedJobs = (
await (
await fetch(
host + "/api/w/" + config.workspace_id + "/jobs/completed/count",
host + "/api/w/" + config.workspace_id + "/jobs/completed/count" + (tags && tags.length > 0 ? "?tags=" + tags.join(",") : ""),
{ headers: { ["Authorization"]: "Bearer " + config.token } }
)
).json()
@@ -208,8 +209,9 @@ export async function main({
}
let testOtherTag = false;
const otherTagTodo = 500000;
if (testOtherTag) {
const otherTagTodo = 2000000;
let parsed = JSON.parse(body);
parsed.tag = "test";
let nbody = JSON.stringify(parsed);
@@ -237,7 +239,7 @@ export async function main({
}
}
pastJobs = await getCompletedJobsCount();
pastJobs = await getCompletedJobsCount(NON_TEST_TAGS);
const response = await fetch(
config.server +
@@ -283,14 +285,14 @@ export async function main({
while (completedJobs < jobsSent) {
const loopStart = Date.now();
if (!didStart) {
const actual_queue = await getQueueCount();
if (actual_queue < jobsSent + (testOtherTag ? otherTagTodo : 0)) {
const actual_queue = await getQueueCount(NON_TEST_TAGS);
if (actual_queue < jobsSent) {
start = Date.now();
didStart = true;
}
} else {
const elapsed = start ? Date.now() - start : 0;
completedJobs = await getCompletedJobsCount();
completedJobs = await getCompletedJobsCount(NON_TEST_TAGS);
if (nStepsFlow > 0) {
completedJobs = Math.floor(completedJobs / (nStepsFlow + 1));
}
@@ -328,7 +330,7 @@ export async function main({
console.log(`avg. throughput (jobs/time): ${jobsSent / total_duration_sec}`);
console.log("completed jobs", completedJobs);
console.log("queue length:", await getQueueCount());
console.log("queue length:", await getQueueCount(NON_TEST_TAGS));
if (
!noVerify &&
benchmarks/benchmark_suite.ts
+1 -1
View File
@@ -27,7 +27,7 @@ async function warmUp(
token,
workspace,
kind: "noop",
jobs: 100000,
jobs: 50000,
});
}