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: improve concurrency limit lock

Browse Source
This commit is contained in:
Ruben Fiszel
2023-10-25 00:17:38 +02:00
parent 9f6f7150cd
commit d4da2de5c0
8 changed files with 61 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json → backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json
Generated
+4 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $11)\n )\n INSERT INTO queue \n (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10 FROM uuid_table) \n RETURNING id",
"query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $11)\n )\n INSERT INTO queue \n (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13 FROM uuid_table) \n RETURNING id",
"describe": {
"columns": [
{
@@ -63,6 +63,8 @@
"Varchar",
"Timestamptz",
"Varchar",
"Int4",
"Int4",
"Int4"
]
},
@@ -70,5 +72,5 @@
false
]
},
"hash": "6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d"
"hash": "a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de"
}
backend/windmill-api/src/jobs.rs
+19 -7
View File
@@ -2283,14 +2283,22 @@ async fn add_batch_jobs(
) -> error::JsonResult<Vec<Uuid>> {
require_super_admin(&db, &authed.email).await?;
let (hash, path, job_kind, language, dedicated_worker) = match batch_info.kind.as_str() {
let (
hash,
path,
job_kind,
language,
dedicated_worker,
concurrent_limit,
concurrent_time_window_s,
) = match batch_info.kind.as_str() {
"script" => {
if let Some(path) = batch_info.path {
let (
script_hash,
_tag,
_concurrent_limit,
_concurrency_time_window_s,
concurrent_limit,
concurrency_time_window_s,
_cache_ttl,
language,
dedicated_worker,
@@ -2302,6 +2310,8 @@ async fn add_batch_jobs(
JobKind::Script,
Some(language),
dedicated_worker,
concurrent_limit,
concurrency_time_window_s,
)
} else {
Err(anyhow::anyhow!(
@@ -2360,7 +2370,7 @@ async fn add_batch_jobs(
}
return Ok(Json(uuids));
}
"noop" => (None, None, JobKind::Noop, None, None),
"noop" => (None, None, JobKind::Noop, None, None, None, None),
_ => {
return Err(error::Error::BadRequest(format!(
"Invalid batch kind: {}",
@@ -2386,8 +2396,8 @@ async fn add_batch_jobs(
select gen_random_uuid() as uuid from generate_series(1, $11)
)
INSERT INTO queue
(id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id)
(SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10 FROM uuid_table)
(id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s)
(SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13 FROM uuid_table)
RETURNING id"#,
hash.map(|h| h.0),
path,
@@ -2399,7 +2409,9 @@ async fn add_batch_jobs(
authed.email,
Utc::now(),
w_id,
n
n,
concurrent_limit,
concurrent_time_window_s
)
.fetch_all(&db)
.await?;
backend/windmill-queue/src/jobs.rs
+8
View File
@@ -1099,6 +1099,14 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
job_custom_concurrency_time_window_s
);
sqlx::query_scalar!(
"SELECT null FROM queue WHERE id = $1 FOR UPDATE",
pulled_job.id
)
.fetch_one(&mut tx)
.await
.context("lock job in queue")?;
let jobs_uuids_init_json_value = serde_json::from_str::<serde_json::Value>(
format!("{{\"{}\": {{}}}}", pulled_job.id.hyphenated().to_string()).as_str(),
)
benchmarks/main.ts
+1 -1
View File
@@ -489,7 +489,7 @@ if (import.meta.main) {
}
)
.option(
"--continous",
"-c --continuous",
"Run the benchmark forever. This effectively disables metric collection & exports. No zombie jobs will be tracked."
)
.option(
benchmarks/pulumi/index.ts
+9 -4
View File
@@ -198,7 +198,7 @@ const asg = new aws.autoscaling.Group("asg3", {
version: "$Latest",
},
minSize: 0,
maxSize: 15,
maxSize: 30,
vpcZoneIdentifiers: [subnetA.id],
tags: [
{
@@ -274,6 +274,8 @@ db.address.apply((address) => {
environment: [
{ name: "MODE", value: "worker" },
{ name: "NUM_WORKERS", value: "1" },
{ name: "DATABASE_CONNECTIONS", value: "5"},
{ name: "SLEEP_QUEUE", value: "300"},
// { name: "METRICS_ADDR", value: "true" },
{ name: "RUST_LOG", value: "info" },
{
@@ -325,6 +327,8 @@ db.address.apply((address) => {
environment: [
{ name: "WORKER_GROUP", value: "dedicated" },
{ name: "NUM_WORKERS", value: "10" },
{ name: "DATABASE_CONNECTIONS", value: "15"},
{ name: "SLEEP_QUEUE", value: "300"},
{ name: "MODE", value: "worker" },
// { name: "METRICS_ADDR", value: "true" },
{ name: "RUST_LOG", value: "info" },
@@ -372,6 +376,7 @@ db.address.apply((address) => {
{ name: "MODE", value: "server" },
// { name: "METRICS_ADDR", value: "true" },
{ name: "RUST_LOG", value: "info" },
{ name: "DATABASE_CONNECTIONS", value: "5"},
{
name: "DATABASE_URL",
value: `postgres://postgres:postgres@${address}/windmill?sslmode=disable`,
@@ -404,7 +409,7 @@ db.address.apply((address) => {
const service_server = new aws.ecs.Service("service-server", {
cluster: cluster.id,
taskDefinition: server_td.arn,
desiredCount: 2,
desiredCount: 4,
forceNewDeployment: true,
orderedPlacementStrategies: [
{
@@ -424,7 +429,7 @@ db.address.apply((address) => {
const service_worker = new aws.ecs.Service("service-worker", {
cluster: cluster.id,
taskDefinition: worker_td.arn,
desiredCount: 10,
desiredCount: 20,
forceNewDeployment: true,
orderedPlacementStrategies: [
{
@@ -437,7 +442,7 @@ db.address.apply((address) => {
const service_worker2 = new aws.ecs.Service("service-worker-2", {
cluster: cluster.id,
taskDefinition: worker_td2.arn,
desiredCount: 3,
desiredCount: 30,
forceNewDeployment: true,
orderedPlacementStrategies: [
{
frontend/src/lib/components/SimpleEditor.svelte
+1
View File
@@ -76,6 +76,7 @@
code = ncode
if (editor) {
editor.setValue(ncode)
console.log(editor, ncode)
}
}
frontend/src/lib/components/WorkspaceGroup.svelte
+7 -1
View File
@@ -25,6 +25,7 @@
cache_clear?: number
init_bash?: string
}
export let activeWorkers: number
let nconfig: any = config
? config.worker_tags != undefined || config.dedicated_worker != undefined
@@ -355,7 +356,7 @@
dirty = false
dirtyCode = false
}}
disabled={!dirty || !$enterpriseLicense}
disabled={(!dirty && nconfig?.dedicated_worker == undefined) || !$enterpriseLicense}
>
Apply changes
</Button>
@@ -399,4 +400,9 @@
>config <Tooltip>{JSON.stringify(config, null, 4)}</Tooltip></span
>
{/if}
{#if activeWorkers > 1}
<span class="ml-4 text-xs"
>{activeWorkers} Workers <Tooltip>Number of workers active in the last 10s</Tooltip></span
>
{/if}
</div>
frontend/src/routes/(root)/(logged)/workers/+page.svelte
+12 -5
View File
@@ -20,20 +20,21 @@
let workers: WorkerPing[] | undefined = undefined
let filteredWorkers: WorkerPing[] = []
let workerGroups: Record<string, any> | undefined = undefined
let groupedWorkers: [string, [[string, string], WorkerPing[]][]][] = []
let groupedWorkers: [string, [string, WorkerPing[]][]][] = []
let intervalId: NodeJS.Timer | undefined
const splitter = '_%%%_'
let globalCache = false
let customTags: string[] | undefined = []
$: filteredWorkers = (workers ?? []).filter((x) => (x.last_ping ?? 0) < 300)
$: groupedWorkers = groupBy(
groupBy(
filteredWorkers,
(wp: WorkerPing) => [wp.worker_instance, wp.worker_group],
(wp: WorkerPing) => wp.worker_instance + splitter + wp.worker_group,
(wp: WorkerPing) => wp.worker
),
(x) => x[0][1],
(x) => x[0][0]
(x) => x[0]?.split(splitter)?.[1],
(x) => x[0]?.split(splitter)?.[0]
)
const WORKER_S3_BUCKET_SYNC_SETTING = 'worker_s3_bucket_sync'
@@ -289,6 +290,9 @@
on:reload={() => {
loadWorkerGroups()
}}
activeWorkers={worker_group?.[1].flatMap((x) =>
x[1]?.filter((y) => (y.last_ping ?? 0) < 15)
)?.length ?? 0}
/>
<DataTable>
@@ -320,8 +324,11 @@
scope="colgroup"
class="bg-surface-secondary/60 py-2 border-b"
>
Instance: <Badge color="gray">{section[0]}</Badge>
Instance: <Badge color="gray">{section?.split(splitter)?.[0]}</Badge>
IP: <Badge color="gray">{workers[0].ip}</Badge>
{#if workers?.length > 1}
{workers?.length} Workers
{/if}
</Cell>
</tr>