kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: batch commit completed jobs in single transaction

Browse Source 
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-04-08 18:28:46 +00:00
parent fc5e479424
commit d1b8d5427b
3 changed files with 239 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
backend/windmill-queue/src/jobs.rs
View File

@@ -462,6 +462,73 @@ pub async fn batch_pull(
Ok(jobs)
}
/// Batch-commit multiple completed jobs in a single transaction.
/// Only for simple jobs (no flows, no schedules, no concurrency limits).
/// Returns (job_id, duration_ms) for each committed job.
pub async fn batch_commit_completed_jobs(
db: &Pool<Postgres>,
jobs: &[(Uuid, bool, &serde_json::value::RawValue, i32, Option<i64>)], // (id, success, result, mem_peak, duration)
) -> windmill_common::error::Result<Vec<(Uuid, i64)>> {
if jobs.is_empty() {
return Ok(vec![]);
}
let mut tx = db.begin().await.map_err(|e| {
windmill_common::error::Error::InternalErr(format!("batch commit begin: {e:#}"))
})?;
let mut results = Vec::with_capacity(jobs.len());
for &(job_id, success, result, mem_peak, duration) in jobs {
let status = if success { "success" } else { "failure" };
let duration_ms: Option<i64> = sqlx::query_scalar(
"INSERT INTO v2_job_completed AS cj
(workspace_id, id, started_at, duration_ms, result,
flow_status, workflow_as_code_status,
memory_peak, status, worker)
SELECT q.workspace_id, q.id, started_at,
COALESCE($3::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000),
$2::jsonb, flow_status, workflow_as_code_status,
$4, $5::job_status, q.worker
FROM v2_job_queue q LEFT JOIN v2_job_status USING (id)
WHERE q.id = $1
ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $2::jsonb
RETURNING duration_ms",
)
.bind(job_id)
.bind(result.get())
.bind(duration)
.bind(if mem_peak > 0 { Some(mem_peak) } else { None })
.bind(status)
.fetch_optional(&mut *tx)
.await
.map_err(|e| {
windmill_common::error::Error::InternalErr(format!(
"batch commit insert job {job_id}: {e:#}"
))
})?;
if let Some(dur) = duration_ms {
sqlx::query!("DELETE FROM v2_job_queue WHERE id = $1", job_id)
.execute(&mut *tx)
.await
.map_err(|e| {
windmill_common::error::Error::InternalErr(format!(
"batch commit delete job {job_id}: {e:#}"
))
})?;
results.push((job_id, dur));
} else {
tracing::warn!("batch commit: job {job_id} not found in queue, skipping");
}
}
tx.commit()
.await
.map_err(|e| windmill_common::error::Error::InternalErr(format!("batch commit: {e:#}")))?;
Ok(results)
}
pub const PERIODIC_SCRIPT_TAG: &str = "periodic_bash_script";
pub const INIT_SCRIPT_TAG: &str = "init_script";
pub const INIT_SCRIPT_PATH_PREFIX: &str = "init_script_";

211
backend/windmill-worker/src/result_processor.rs
View File

@@ -218,6 +218,22 @@ enum JobCompletedRx {
WakeUp,
}
fn is_batchable(jc: &JobCompleted) -> bool {
jc.success
&& !jc.job.is_flow_step()
&& jc.job.flow_step_id.as_deref() != Some("preprocessor")
&& jc.preprocessed_args.is_none()
&& jc.job.tag.as_str() != INIT_SCRIPT_TAG
&& !matches!(
jc.job.kind,
JobKind::Dependencies | JobKind::FlowDependencies
)
&& jc.canceled_by.is_none()
&& jc.cached_res_path.is_none()
&& jc.job.concurrent_limit.is_none()
&& jc.job.schedule_path().is_none()
}
pub fn start_background_processor(
job_completed_rx: JobCompletedReceiver,
job_completed_sender: JobCompletedSender,
@@ -264,6 +280,8 @@ pub fn start_background_processor(
}
});
let mut batch_result_buffer: Vec<windmill_queue::JobCompleted> = Vec::new();
//if we have been killed, we want to drain the queue of jobs
while let Some(sr) = {
if has_been_killed {
@@ -303,61 +321,152 @@ pub fn start_background_processor(
result: SendResultPayload::JobCompleted(jc),
time,
}) => {
let is_init_script_and_failure =
!jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG;
let is_dependency_job = matches!(
jc.job.kind,
JobKind::Dependencies | JobKind::FlowDependencies
);
#[cfg(feature = "benchmark")]
let bench_job_id = jc.job.id;
#[cfg(feature = "benchmark")]
let is_top_level_job = jc.job.parent_job.is_none();
let batch_mode = windmill_common::utils::MODE_AND_ADDONS.mode
== windmill_common::utils::Mode::AgentBatch;
process_jc(
jc,
&worker_name,
&base_internal_url,
&db,
&worker_dir,
Some(&same_worker_tx),
&job_completed_sender,
&stats_map,
&killpill_rx,
#[cfg(feature = "benchmark")]
&mut bench,
#[cfg(feature = "benchmark")]
&mut infos,
)
.warn_after_seconds(10)
.await;
if batch_mode && is_batchable(&jc) {
// Accumulate for batch commit
batch_result_buffer.push(jc);
if is_init_script_and_failure {
tracing::error!("init script errored, exiting");
killpill_tx.send();
break;
}
if is_dependency_job && is_dedicated_worker {
tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted.");
sqlx::query!(
"UPDATE config SET config = config WHERE name = $1",
format!("worker__{}", *WORKER_GROUP)
)
.execute(&db)
.await
.expect("update config to trigger restart of all dedicated workers at that config");
killpill_tx.send();
}
add_time!(bench, "job completed processed");
#[cfg(feature = "benchmark")]
{
if infos.add_iter(bench, bench_job_id, is_top_level_job) {
infos.shared_iters.fetch_add(1, Ordering::Relaxed);
// Drain any additional ready results
while let Ok(SendResult {
result: SendResultPayload::JobCompleted(jc2),
..
}) = bounded_rx.try_recv()
{
if is_batchable(&jc2) {
batch_result_buffer.push(jc2);
} else {
process_jc(
jc2,
&worker_name,
&base_internal_url,
&db,
&worker_dir,
Some(&same_worker_tx),
&job_completed_sender,
&stats_map,
&killpill_rx,
#[cfg(feature = "benchmark")]
&mut bench,
#[cfg(feature = "benchmark")]
&mut infos,
)
.warn_after_seconds(10)
.await;
}
if batch_result_buffer.len() >= 50 {
break;
}
}
// Flush batch
if !batch_result_buffer.is_empty() {
let batch_items: Vec<_> = batch_result_buffer
.iter()
.map(|jc| {
(
jc.job.id,
jc.success,
jc.result.as_ref() as &serde_json::value::RawValue,
jc.mem_peak,
jc.duration,
)
})
.collect();
match windmill_queue::batch_commit_completed_jobs(&db, &batch_items)
.await
{
Ok(committed) => {
tracing::debug!("batch committed {} jobs", committed.len());
}
Err(e) => {
tracing::error!(
"batch commit failed, falling back to individual: {e:#}"
);
for jc in batch_result_buffer.drain(..) {
process_jc(
jc,
&worker_name,
&base_internal_url,
&db,
&worker_dir,
Some(&same_worker_tx),
&job_completed_sender,
&stats_map,
&killpill_rx,
#[cfg(feature = "benchmark")]
&mut bench,
#[cfg(feature = "benchmark")]
&mut infos,
)
.warn_after_seconds(10)
.await;
}
}
}
batch_result_buffer.clear();
}
last_processing_duration
.store(time.elapsed().as_secs() as u16, Ordering::SeqCst);
} else {
let is_init_script_and_failure =
!jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG;
let is_dependency_job = matches!(
jc.job.kind,
JobKind::Dependencies | JobKind::FlowDependencies
);
#[cfg(feature = "benchmark")]
let bench_job_id = jc.job.id;
#[cfg(feature = "benchmark")]
let is_top_level_job = jc.job.parent_job.is_none();
process_jc(
jc,
&worker_name,
&base_internal_url,
&db,
&worker_dir,
Some(&same_worker_tx),
&job_completed_sender,
&stats_map,
&killpill_rx,
#[cfg(feature = "benchmark")]
&mut bench,
#[cfg(feature = "benchmark")]
&mut infos,
)
.warn_after_seconds(10)
.await;
if is_init_script_and_failure {
tracing::error!("init script errored, exiting");
killpill_tx.send();
break;
}
if is_dependency_job && is_dedicated_worker {
tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted.");
sqlx::query!(
"UPDATE config SET config = config WHERE name = $1",
format!("worker__{}", *WORKER_GROUP)
)
.execute(&db)
.await
.expect("update config to trigger restart of all dedicated workers at that config");
killpill_tx.send();
}
add_time!(bench, "job completed processed");
#[cfg(feature = "benchmark")]
{
if infos.add_iter(bench, bench_job_id, is_top_level_job) {
infos.shared_iters.fetch_add(1, Ordering::Relaxed);
}
}
last_processing_duration
.store(time.elapsed().as_secs() as u16, Ordering::SeqCst);
}
last_processing_duration
.store(time.elapsed().as_secs() as u16, Ordering::SeqCst);
}
JobCompletedRx::JobCompleted(SendResult {
result:

13
backend/windmill-worker/src/worker.rs
View File

@@ -2075,7 +2075,18 @@ pub async fn run_worker(
let mut last_suspend_first = Instant::now();
let mut killed_but_draining_same_worker_jobs = false;
let batch_pull_size = *windmill_common::worker::BATCH_PULL_SIZE;
let is_agent_batch =
windmill_common::utils::MODE_AND_ADDONS.mode == windmill_common::utils::Mode::AgentBatch;
let batch_pull_size = if is_agent_batch {
let s = *windmill_common::worker::BATCH_PULL_SIZE;
if s > 0 {
s
} else {
100
}
} else {
0
};
let mut batch_pull_buffer: std::collections::VecDeque<windmill_queue::PulledJob> =
std::collections::VecDeque::new();
let mut agent_batch_buffer: std::collections::VecDeque<windmill_queue::JobAndPerms> =