feat: improved dedicated benchmarks + buffer fix (#2313)
* feat: improved dedicated benchmarks + buffer fix * fixes + limit task spawning to noop/dedicated * fix: cargo test --------- Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
HugoCasa
@@ -37,7 +37,6 @@
|
||||
"bash",
|
||||
"postgresql",
|
||||
"nativets",
|
||||
"Nativets",
|
||||
"bun",
|
||||
"mysql",
|
||||
"bigquery",
|
||||
|
||||
@@ -67,7 +67,6 @@
|
||||
"bash",
|
||||
"postgresql",
|
||||
"nativets",
|
||||
"Nativets",
|
||||
"bun",
|
||||
"mysql",
|
||||
"bigquery",
|
||||
|
||||
@@ -28,7 +28,6 @@
|
||||
"bash",
|
||||
"postgresql",
|
||||
"nativets",
|
||||
"Nativets",
|
||||
"bun",
|
||||
"mysql",
|
||||
"bigquery",
|
||||
|
||||
@@ -42,7 +42,6 @@
|
||||
"bash",
|
||||
"postgresql",
|
||||
"nativets",
|
||||
"Nativets",
|
||||
"bun",
|
||||
"mysql",
|
||||
"bigquery",
|
||||
|
||||
@@ -42,7 +42,6 @@
|
||||
"bash",
|
||||
"postgresql",
|
||||
"nativets",
|
||||
"Nativets",
|
||||
"bun",
|
||||
"mysql",
|
||||
"bigquery",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) 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)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10 FROM uuid_table) \n RETURNING id",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -46,7 +46,6 @@
|
||||
"bash",
|
||||
"postgresql",
|
||||
"nativets",
|
||||
"Nativets",
|
||||
"bun",
|
||||
"mysql",
|
||||
"bigquery",
|
||||
@@ -70,5 +69,5 @@
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "2de52e1f3226ca9281b6e25f74d4d05f4509cb87c875234bfc7b310a012e4d40"
|
||||
"hash": "6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d"
|
||||
}
|
||||
@@ -60,7 +60,6 @@
|
||||
"bash",
|
||||
"postgresql",
|
||||
"nativets",
|
||||
"Nativets",
|
||||
"bun",
|
||||
"mysql",
|
||||
"bigquery",
|
||||
|
||||
@@ -42,7 +42,6 @@
|
||||
"bash",
|
||||
"postgresql",
|
||||
"nativets",
|
||||
"Nativets",
|
||||
"bun",
|
||||
"mysql",
|
||||
"bigquery",
|
||||
|
||||
@@ -63,7 +63,7 @@ pub async fn initial_load(
|
||||
db: &Pool<Postgres>,
|
||||
tx: tokio::sync::broadcast::Sender<()>,
|
||||
worker_mode: bool,
|
||||
server_mode: bool,
|
||||
server_mode: bool
|
||||
) {
|
||||
let reload_worker_config_f = async {
|
||||
if worker_mode {
|
||||
|
||||
@@ -2332,7 +2332,6 @@ struct BatchInfo {
|
||||
kind: String,
|
||||
flow_value: Option<FlowValue>,
|
||||
path: Option<String>,
|
||||
dedicated_worker: Option<bool>,
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
@@ -2362,7 +2361,7 @@ async fn add_batch_jobs(
|
||||
batch_info.path,
|
||||
JobKind::Script,
|
||||
Some(script.language),
|
||||
batch_info.dedicated_worker,
|
||||
script.dedicated_worker,
|
||||
)
|
||||
}
|
||||
"flow" => {
|
||||
@@ -2433,15 +2432,22 @@ async fn add_batch_jobs(
|
||||
format!("{}", language.as_str())
|
||||
};
|
||||
|
||||
let uuids = sqlx::query_scalar!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id",
|
||||
let uuids = sqlx::query_scalar!(
|
||||
r#"WITH uuid_table as (
|
||||
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)
|
||||
RETURNING id"#,
|
||||
hash.map(|h| h.0),
|
||||
path,
|
||||
job_kind.clone() as JobKind,
|
||||
language as ScriptLang,
|
||||
tag,
|
||||
authed.username,
|
||||
authed.email,
|
||||
username_to_permissioned_as(&authed.username),
|
||||
authed.email,
|
||||
Utc::now(),
|
||||
w_id,
|
||||
n
|
||||
|
||||
@@ -161,6 +161,7 @@ pub struct Script {
|
||||
pub concurrent_limit: Option<i32>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub concurrency_time_window_s: Option<i32>,
|
||||
pub dedicated_worker: Option<bool>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
|
||||
@@ -464,6 +464,10 @@ pub async fn start_worker(
|
||||
mut jobs_rx: Receiver<QueuedJob>,
|
||||
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
|
||||
) -> Result<()> {
|
||||
use std::task::Poll;
|
||||
|
||||
use futures::{future, Future};
|
||||
|
||||
let mut logs = "".to_string();
|
||||
let _ = write_file(job_dir, "main.ts", inner_content).await?;
|
||||
let common_bun_proc_envs: HashMap<String, String> =
|
||||
@@ -684,6 +688,21 @@ plugin(p)
|
||||
// let mut i = 0;
|
||||
// let mut j = 0;
|
||||
let mut alive = true;
|
||||
|
||||
fn conditional_polling<T>(
|
||||
fut: impl Future<Output = T>,
|
||||
predicate: bool,
|
||||
) -> impl Future<Output = T> {
|
||||
let mut fut = Box::pin(fut);
|
||||
future::poll_fn(move |cx| {
|
||||
if predicate {
|
||||
fut.as_mut().poll(cx)
|
||||
} else {
|
||||
Poll::Pending
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
@@ -711,8 +730,8 @@ plugin(p)
|
||||
tracing::info!("dedicated worker process exited");
|
||||
break;
|
||||
}
|
||||
}
|
||||
job = jobs_rx.recv(), if alive && jobs.len() < MAX_BUFFERED_DEDICATED_JOBS => {
|
||||
},
|
||||
job = conditional_polling(jobs_rx.recv(), alive && jobs.len() < MAX_BUFFERED_DEDICATED_JOBS) => {
|
||||
// i += 1;
|
||||
if let Some(job) = job {
|
||||
tracing::debug!("received job");
|
||||
|
||||
@@ -10,12 +10,16 @@ use anyhow::Result;
|
||||
use const_format::concatcp;
|
||||
use itertools::Itertools;
|
||||
use once_cell::sync::OnceCell;
|
||||
use prometheus::core::{AtomicU64, GenericCounter};
|
||||
#[cfg(feature = "benchmark")]
|
||||
use serde::Serialize;
|
||||
use sqlx::{Pool, Postgres};
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
sync::{atomic::Ordering, Arc},
|
||||
sync::{
|
||||
atomic::{AtomicUsize, Ordering},
|
||||
Arc,
|
||||
},
|
||||
time::Duration,
|
||||
};
|
||||
use windmill_api_client::Client;
|
||||
@@ -337,6 +341,52 @@ macro_rules! add_time {
|
||||
};
|
||||
}
|
||||
|
||||
async fn handle_receive_completed_job<
|
||||
R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static,
|
||||
>(
|
||||
jc: JobCompleted,
|
||||
worker_execution_failed: HashMap<Option<ScriptLang>, GenericCounter<AtomicU64>>,
|
||||
base_internal_url: String,
|
||||
db: Pool<Postgres>,
|
||||
worker_dir: String,
|
||||
same_worker_tx: Sender<Uuid>,
|
||||
rsmq: Option<R>,
|
||||
) {
|
||||
let metrics = build_language_metrics(&worker_execution_failed.clone(), &jc.job.language);
|
||||
let token = jc.token.clone();
|
||||
let workspace = jc.job.workspace_id.clone();
|
||||
let client = AuthedClient {
|
||||
base_internal_url: base_internal_url.to_string(),
|
||||
workspace,
|
||||
token,
|
||||
client: OnceCell::new(),
|
||||
};
|
||||
if let Err(err) = process_completed_job(
|
||||
&jc,
|
||||
&client,
|
||||
&db,
|
||||
&worker_dir,
|
||||
metrics.clone(),
|
||||
same_worker_tx.clone(),
|
||||
rsmq.clone(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
handle_job_error(
|
||||
&db,
|
||||
&client,
|
||||
&jc.job,
|
||||
err,
|
||||
metrics,
|
||||
false,
|
||||
same_worker_tx.clone(),
|
||||
&worker_dir,
|
||||
rsmq.clone(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
|
||||
db: &Pool<Postgres>,
|
||||
worker_instance: &str,
|
||||
@@ -549,43 +599,168 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
|
||||
let rsmq2 = rsmq.clone();
|
||||
let worker_dir2 = worker_dir.clone();
|
||||
let worker_execution_failed2 = worker_execution_failed.clone();
|
||||
let thread_count = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let is_dedicated_worker = WORKER_CONFIG.read().await.dedicated_worker.is_some();
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
let jobs = 25000;
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
{
|
||||
if is_dedicated_worker {
|
||||
// you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47
|
||||
let hash = sqlx::query_scalar!(
|
||||
"SELECT hash FROM script WHERE path = $1 AND workspace_id = $2",
|
||||
"f/benchmarks/dedicated",
|
||||
"admins"
|
||||
)
|
||||
.fetch_one(db)
|
||||
.await
|
||||
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
|
||||
sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))",
|
||||
hash,
|
||||
"f/benchmarks/dedicated",
|
||||
JobKind::Script as JobKind,
|
||||
ScriptLang::Bun as ScriptLang,
|
||||
"admins:f/benchmarks/dedicated",
|
||||
"admin",
|
||||
"u/admin",
|
||||
"admin@windmill.dev",
|
||||
chrono::Utc::now(),
|
||||
"admins",
|
||||
jobs
|
||||
)
|
||||
.execute(db)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
|
||||
} else {
|
||||
sqlx::query!("INSERT INTO queue (id, script_hash, script_path, job_kind, language, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11))",
|
||||
None::<i64>,
|
||||
None::<String>,
|
||||
JobKind::Noop as JobKind,
|
||||
ScriptLang::Deno as ScriptLang,
|
||||
"deno",
|
||||
"admin",
|
||||
"u/admin",
|
||||
"admin@windmill.dev",
|
||||
chrono::Utc::now(),
|
||||
"admins",
|
||||
jobs
|
||||
)
|
||||
.execute(db)
|
||||
.await.unwrap_or_else(|_e| panic!("failed to insert noop jobs"));
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
let completed_jobs = Arc::new(AtomicUsize::new(0));
|
||||
#[cfg(feature = "benchmark")]
|
||||
let start = Instant::now();
|
||||
#[cfg(feature = "benchmark")]
|
||||
let main_duration = Arc::new(AtomicUsize::new(0));
|
||||
#[cfg(feature = "benchmark")]
|
||||
let send_duration = Arc::new(AtomicUsize::new(0));
|
||||
#[cfg(feature = "benchmark")]
|
||||
let process_duration = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
let main_duration2 = main_duration.clone();
|
||||
#[cfg(feature = "benchmark")]
|
||||
let send_duration2 = send_duration.clone();
|
||||
|
||||
let send_result = tokio::spawn(async move {
|
||||
while let Some(jc) = job_completed_rx.recv().await {
|
||||
let metrics = build_language_metrics(&worker_execution_failed2, &jc.job.language);
|
||||
let token = jc.token.clone();
|
||||
let workspace = jc.job.workspace_id.clone();
|
||||
let client = AuthedClient {
|
||||
base_internal_url: base_internal_url2.to_string(),
|
||||
workspace,
|
||||
token,
|
||||
client: OnceCell::new(),
|
||||
};
|
||||
if let Err(err) = process_completed_job(
|
||||
&jc,
|
||||
&client,
|
||||
&db2,
|
||||
&worker_dir2,
|
||||
metrics.clone(),
|
||||
same_worker_tx2.clone(),
|
||||
rsmq2.clone(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
handle_job_error(
|
||||
&db2,
|
||||
&client,
|
||||
&jc.job,
|
||||
err,
|
||||
metrics,
|
||||
false,
|
||||
same_worker_tx2.clone(),
|
||||
&worker_dir2,
|
||||
rsmq2.clone(),
|
||||
let base_internal_url2 = base_internal_url2.clone();
|
||||
let worker_execution_failed2 = worker_execution_failed2.clone();
|
||||
let worker_dir2 = worker_dir2.clone();
|
||||
let db2 = db2.clone();
|
||||
let same_worker_tx2 = same_worker_tx2.clone();
|
||||
let rsmq2 = rsmq2.clone();
|
||||
|
||||
if matches!(jc.job.job_kind, JobKind::Noop) || is_dedicated_worker {
|
||||
thread_count.fetch_add(1, Ordering::SeqCst);
|
||||
let thread_count = thread_count.clone();
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
let send_duration = send_duration2.clone();
|
||||
#[cfg(feature = "benchmark")]
|
||||
let process_duration = process_duration.clone();
|
||||
#[cfg(feature = "benchmark")]
|
||||
let completed_jobs = completed_jobs.clone();
|
||||
#[cfg(feature = "benchmark")]
|
||||
let main_duration = main_duration2.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
#[cfg(feature = "benchmark")]
|
||||
let process_start = Instant::now();
|
||||
|
||||
handle_receive_completed_job(
|
||||
jc,
|
||||
worker_execution_failed2,
|
||||
base_internal_url2,
|
||||
db2,
|
||||
worker_dir2,
|
||||
same_worker_tx2,
|
||||
rsmq2,
|
||||
)
|
||||
.await;
|
||||
#[cfg(feature = "benchmark")]
|
||||
{
|
||||
let n = completed_jobs.fetch_add(1, Ordering::SeqCst);
|
||||
if (n + 1) % 1000 == 0 || n == (jobs - 1) as usize {
|
||||
let duration_s = start.elapsed().as_secs_f64();
|
||||
let jobs_per_sec = n as f64 / duration_s;
|
||||
tracing::info!(
|
||||
"completed {} jobs in {}s, {} jobs/s",
|
||||
n + 1,
|
||||
duration_s,
|
||||
jobs_per_sec
|
||||
);
|
||||
|
||||
tracing::info!(
|
||||
"main loop without send {}s",
|
||||
main_duration.load(Ordering::SeqCst) as f64 / 1000.0
|
||||
);
|
||||
|
||||
tracing::info!(
|
||||
"send job completed / send dedicated job duration {}s",
|
||||
send_duration.load(Ordering::SeqCst) as f64 / 1000.0
|
||||
);
|
||||
|
||||
tracing::info!(
|
||||
"job completed process duration {}s",
|
||||
process_duration.load(Ordering::SeqCst) as f64 / 1000.0
|
||||
);
|
||||
}
|
||||
|
||||
process_duration.fetch_add(
|
||||
process_start.elapsed().as_millis() as usize,
|
||||
Ordering::SeqCst,
|
||||
);
|
||||
}
|
||||
|
||||
thread_count.fetch_sub(1, Ordering::SeqCst);
|
||||
});
|
||||
} else {
|
||||
handle_receive_completed_job(
|
||||
jc,
|
||||
worker_execution_failed2,
|
||||
base_internal_url2,
|
||||
db2,
|
||||
worker_dir2,
|
||||
same_worker_tx2,
|
||||
rsmq2,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
tracing::info!("stopped processing new completed jobs");
|
||||
while thread_count.load(Ordering::SeqCst) > 0 {
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
tracing::info!("finished processing all completed jobs");
|
||||
|
||||
// if let Err(e) =
|
||||
// add_completed_job(&db2, &job, success, false, result, logs, rsmq2.clone()).await
|
||||
// {
|
||||
@@ -719,6 +894,9 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
|
||||
(None, None) as (Option<Sender<QueuedJob>>, Option<JoinHandle<()>>)
|
||||
};
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
tracing::info!("pre loop time {}s", start.elapsed().as_secs_f64());
|
||||
|
||||
loop {
|
||||
#[cfg(feature = "benchmark")]
|
||||
let loop_start = Instant::now();
|
||||
@@ -889,11 +1067,27 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
|
||||
jobs_executed += 1;
|
||||
|
||||
if let Some(dedicated_worker_tx) = dedicated_worker_tx.clone() {
|
||||
#[cfg(feature = "benchmark")]
|
||||
main_duration
|
||||
.fetch_add(loop_start.elapsed().as_millis() as usize, Ordering::SeqCst);
|
||||
#[cfg(feature = "benchmark")]
|
||||
let send_start = Instant::now();
|
||||
|
||||
if let Err(e) = dedicated_worker_tx.send(job.clone()).await {
|
||||
tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}");
|
||||
}
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
send_duration
|
||||
.fetch_add(send_start.elapsed().as_millis() as usize, Ordering::SeqCst);
|
||||
continue;
|
||||
} else if matches!(job.job_kind, JobKind::Noop) {
|
||||
#[cfg(feature = "benchmark")]
|
||||
main_duration
|
||||
.fetch_add(loop_start.elapsed().as_millis() as usize, Ordering::SeqCst);
|
||||
#[cfg(feature = "benchmark")]
|
||||
let send_start = Instant::now();
|
||||
|
||||
job_completed_tx
|
||||
.send(JobCompleted {
|
||||
job,
|
||||
@@ -905,6 +1099,10 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
|
||||
})
|
||||
.await
|
||||
.expect("send job completed");
|
||||
|
||||
#[cfg(feature = "benchmark")]
|
||||
send_duration
|
||||
.fetch_add(send_start.elapsed().as_millis() as usize, Ordering::SeqCst);
|
||||
} else {
|
||||
let token = create_token_for_owner_in_bg(&db, &job).await;
|
||||
|
||||
|
||||
@@ -11,6 +11,31 @@ import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts";
|
||||
|
||||
import { VERSION, createBenchScript, getFlowPayload, login } from "./lib.ts";
|
||||
|
||||
async function verifyOutputs(uuids: string[], workspace: string) {
|
||||
console.log("Verifying outputs");
|
||||
let incorrectResults = 0;
|
||||
for (const uuid of uuids) {
|
||||
try {
|
||||
const job = await windmill.JobService.getCompletedJob({
|
||||
workspace,
|
||||
id: uuid,
|
||||
});
|
||||
if (!job.success) {
|
||||
console.log(`Job ${uuid} did not complete`);
|
||||
incorrectResults++;
|
||||
}
|
||||
if (job.result !== uuid) {
|
||||
console.log(`Job ${uuid} did not output the correct value`);
|
||||
incorrectResults++;
|
||||
}
|
||||
} catch (err) {
|
||||
console.log(`Job ${uuid} did not complete`);
|
||||
incorrectResults++;
|
||||
}
|
||||
}
|
||||
console.log(`Incorrect results: ${incorrectResults}`);
|
||||
}
|
||||
|
||||
export async function main({
|
||||
host,
|
||||
email,
|
||||
@@ -19,6 +44,7 @@ export async function main({
|
||||
workspace,
|
||||
kind,
|
||||
jobs,
|
||||
noVerify,
|
||||
}: {
|
||||
host: string;
|
||||
email?: string;
|
||||
@@ -27,6 +53,7 @@ export async function main({
|
||||
workspace: string;
|
||||
kind: string;
|
||||
jobs: number;
|
||||
noVerify: boolean;
|
||||
}) {
|
||||
windmill.setClient("", host);
|
||||
|
||||
@@ -65,11 +92,37 @@ export async function main({
|
||||
windmill.setClient(final_token, host);
|
||||
const enc = (s: string) => new TextEncoder().encode(s);
|
||||
|
||||
async function getQueueCount() {
|
||||
return (
|
||||
await (
|
||||
await fetch(
|
||||
config.server + "/api/w/" + config.workspace_id + "/jobs/queue/count",
|
||||
{ headers: { ["Authorization"]: "Bearer " + config.token } }
|
||||
)
|
||||
).json()
|
||||
).database_length;
|
||||
}
|
||||
|
||||
let pastJobs = 0;
|
||||
async function getCompletedJobsCount(): Promise<number> {
|
||||
const completedJobs = (
|
||||
await (
|
||||
await fetch(
|
||||
host + "/api/w/" + config.workspace_id + "/jobs/completed/count",
|
||||
{ headers: { ["Authorization"]: "Bearer " + config.token } }
|
||||
)
|
||||
).json()
|
||||
).database_length;
|
||||
return completedJobs - pastJobs;
|
||||
}
|
||||
|
||||
if (["deno", "python", "go", "bash", "dedicated", "bun"].includes(kind)) {
|
||||
await createBenchScript(kind, workspace);
|
||||
}
|
||||
|
||||
let jobsSent = jobs;
|
||||
pastJobs = await getCompletedJobsCount();
|
||||
|
||||
const jobsSent = jobs;
|
||||
console.log(`Bulk creating ${jobsSent} jobs`);
|
||||
|
||||
const start_create = Date.now();
|
||||
@@ -84,9 +137,8 @@ export async function main({
|
||||
body = JSON.stringify({
|
||||
kind: "script",
|
||||
path: "f/benchmarks/" + kind,
|
||||
dedicated_worker: kind === "dedicated",
|
||||
});
|
||||
} else if (["2steps", "onebranch", "branchallparrallel"].includes(kind)) {
|
||||
} else if (["2steps"].includes(kind)) {
|
||||
const payload = getFlowPayload(kind);
|
||||
body = JSON.stringify({
|
||||
kind: "flow",
|
||||
@@ -113,6 +165,7 @@ export async function main({
|
||||
if (!response.ok) {
|
||||
throw new Error("Failed to create jobs: " + response.statusText);
|
||||
}
|
||||
const uuids = await response.json();
|
||||
const end_create = Date.now();
|
||||
const create_duration = end_create - start_create;
|
||||
console.log(
|
||||
@@ -122,69 +175,61 @@ export async function main({
|
||||
);
|
||||
let start = Date.now();
|
||||
|
||||
let queue_length = jobsSent;
|
||||
let completedJobs = 0;
|
||||
let lastElapsed = 0;
|
||||
let lastQueueLength = queue_length;
|
||||
const updateState = setInterval(async () => {
|
||||
const elapsed = start ? Date.now() - start : 0;
|
||||
queue_length = (
|
||||
await (
|
||||
await fetch(
|
||||
host + "/api/w/" + config.workspace_id + "/jobs/queue/count",
|
||||
{ headers: { ["Authorization"]: "Bearer " + config.token } }
|
||||
let lastCompletedJobs = 0;
|
||||
|
||||
let didStart = false;
|
||||
while (completedJobs < jobsSent) {
|
||||
if (!didStart) {
|
||||
const actual_queue = await getQueueCount();
|
||||
if (actual_queue < jobsSent) {
|
||||
start = Date.now();
|
||||
didStart = true;
|
||||
}
|
||||
} else {
|
||||
const elapsed = start ? Date.now() - start : 0;
|
||||
completedJobs = await getCompletedJobsCount();
|
||||
if (kind === "2steps") {
|
||||
completedJobs = Math.floor(completedJobs / 3);
|
||||
}
|
||||
const avgThr = ((completedJobs / elapsed) * 1000).toFixed(2);
|
||||
const instThr =
|
||||
lastElapsed > 0
|
||||
? (
|
||||
((completedJobs - lastCompletedJobs) / (elapsed - lastElapsed)) *
|
||||
1000
|
||||
).toFixed(2)
|
||||
: 0;
|
||||
|
||||
lastElapsed = elapsed;
|
||||
lastCompletedJobs = completedJobs;
|
||||
|
||||
await Deno.stdout.write(
|
||||
enc(
|
||||
`elapsed: ${(elapsed / 1000).toFixed(
|
||||
2
|
||||
)} | jobs executed: ${completedJobs}/${jobsSent} (thr: inst ${instThr} - avg ${avgThr}) | remaining: ${
|
||||
jobsSent - completedJobs
|
||||
} \r`
|
||||
)
|
||||
).json()
|
||||
).database_length;
|
||||
const avgThr = (((jobsSent - queue_length) / elapsed) * 1000).toFixed(2);
|
||||
const instThr =
|
||||
lastElapsed > 0
|
||||
? (
|
||||
((lastQueueLength - queue_length) / (elapsed - lastElapsed)) *
|
||||
1000
|
||||
).toFixed(2)
|
||||
: 0;
|
||||
|
||||
lastElapsed = elapsed;
|
||||
lastQueueLength = queue_length;
|
||||
|
||||
await Deno.stdout.write(
|
||||
enc(
|
||||
`elapsed: ${(elapsed / 1000).toFixed(2)} | jobs executed: ${
|
||||
jobsSent - queue_length
|
||||
}/${jobsSent} (thr: inst ${instThr} - avg ${avgThr}) | queue: ${queue_length} \r`
|
||||
)
|
||||
);
|
||||
}, 10);
|
||||
|
||||
while (queue_length > 0) {
|
||||
if (queue_length < jobsSent && jobsSent === jobs) {
|
||||
// reset start time to when the first job was picked up
|
||||
start = Date.now();
|
||||
jobsSent = queue_length;
|
||||
);
|
||||
}
|
||||
await sleep(0.01);
|
||||
}
|
||||
|
||||
clearInterval(updateState);
|
||||
|
||||
const total_duration_sec = (Date.now() - start) / 1000.0;
|
||||
|
||||
await sleep(0.1);
|
||||
console.log(`\njobs: ${jobsSent}`);
|
||||
console.log(`duration: ${total_duration_sec}s`);
|
||||
console.log(`avg. throughput (jobs/time): ${jobsSent / total_duration_sec}`);
|
||||
|
||||
console.log(
|
||||
"queue length:",
|
||||
(
|
||||
await (
|
||||
await fetch(
|
||||
host + "/api/w/" + config.workspace_id + "/jobs/queue/count",
|
||||
{ headers: { ["Authorization"]: "Bearer " + config.token } }
|
||||
)
|
||||
).json()
|
||||
).database_length
|
||||
);
|
||||
console.log("completed jobs", await getCompletedJobsCount());
|
||||
console.log("queue length:", await getQueueCount());
|
||||
|
||||
if (!noVerify && kind !== "noop") {
|
||||
await verifyOutputs(uuids, config.workspace_id);
|
||||
}
|
||||
|
||||
console.log("done");
|
||||
|
||||
return {
|
||||
@@ -229,7 +274,7 @@ if (import.meta.main) {
|
||||
)
|
||||
.option(
|
||||
"--kind <kind:string>",
|
||||
"Specifiy the benchmark kind among: deno, identity, python, go, bash, dedicated, bun, noop, 2steps, onebranch, branchallparrallel",
|
||||
"Specifiy the benchmark kind among: deno, identity, python, go, bash, dedicated, bun, noop, 2steps",
|
||||
{
|
||||
required: true,
|
||||
}
|
||||
@@ -237,6 +282,7 @@ if (import.meta.main) {
|
||||
.option("-j --jobs <jobs:number>", "Number of jobs to create.", {
|
||||
default: 10000,
|
||||
})
|
||||
.option("--no-verify", "Do not verify the output of the jobs.")
|
||||
.action(main)
|
||||
.command(
|
||||
"upgrade",
|
||||
|
||||
@@ -32,16 +32,16 @@ async function waitForDedicatedWorker(workspace: string, path: string) {
|
||||
const query = windmill.JobService.runWaitResultScriptByPath({
|
||||
workspace,
|
||||
path,
|
||||
requestBody: {
|
||||
args: {},
|
||||
},
|
||||
requestBody: {},
|
||||
});
|
||||
const timeout = new Promise((_, reject) => {
|
||||
setTimeout(() => {
|
||||
let timeout;
|
||||
const timeoutPromise = new Promise((_, reject) => {
|
||||
timeout = setTimeout(() => {
|
||||
reject("Timeout");
|
||||
}, 15000);
|
||||
});
|
||||
await Promise.race([query, timeout]);
|
||||
await Promise.race([query, timeoutPromise]);
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
|
||||
export async function createBenchScript(
|
||||
@@ -63,6 +63,7 @@ export async function createBenchScript(
|
||||
|
||||
let scriptContent: string;
|
||||
let language: string;
|
||||
let schemaProperties = {};
|
||||
if (scriptPattern === "python") {
|
||||
scriptContent =
|
||||
'import os\n\ndef main():\n return os.environ.get("WM_JOB_ID")';
|
||||
@@ -74,9 +75,15 @@ export async function createBenchScript(
|
||||
} else if (scriptPattern === "bash") {
|
||||
scriptContent = "echo $WM_JOB_ID";
|
||||
language = "bash";
|
||||
} else if (scriptPattern === "dedicated" || scriptPattern === "bun") {
|
||||
} else if (scriptPattern === "bun") {
|
||||
scriptContent = 'export function main(){ return Bun.env["WM_JOB_ID"]; }';
|
||||
language = "bun";
|
||||
} else if (scriptPattern === "dedicated") {
|
||||
scriptContent = "export function main(uuid){ return uuid; }";
|
||||
language = "bun";
|
||||
schemaProperties = {
|
||||
uuid: { default: null, description: "", type: "string" },
|
||||
};
|
||||
} else if (scriptPattern === "deno") {
|
||||
scriptContent =
|
||||
'export function main(){ return Deno.env.get("WM_JOB_ID"); }';
|
||||
@@ -96,6 +103,12 @@ export async function createBenchScript(
|
||||
description: "",
|
||||
language: language as api.NewScript.language,
|
||||
dedicated_worker: scriptPattern === "dedicated",
|
||||
schema: {
|
||||
$schema: "https://json-schema.org/draft/2020-12/schema",
|
||||
properties: schemaProperties,
|
||||
required: [],
|
||||
type: "object",
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
|
||||
Reference in New Issue
Block a user