From 946756ae83deb4e7a93111edddbd4d98c596d5a9 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 8 Sep 2026 15:21:10 +0200 Subject: [PATCH] perf: reduce shared worker debug polling frames (#11028) * fix: reduce php parser stack use in debug workers * test: document php stack regression expression depth * fix: offload php signature parsing from async workers * fix: address php parser review nits * perf: reduce shared worker debug polling frames * test: refresh agent volume fixtures --- backend/tests/agent_workers.rs | 27 ++++++++++++++++++++------- backend/tests/volume_tests.rs | 13 +++++++++++-- backend/windmill-worker/src/worker.rs | 25 +++++++++++++------------ 3 files changed, 44 insertions(+), 21 deletions(-) diff --git a/backend/tests/agent_workers.rs b/backend/tests/agent_workers.rs index 791556b035..4d6fffa346 100644 --- a/backend/tests/agent_workers.rs +++ b/backend/tests/agent_workers.rs @@ -293,6 +293,7 @@ async fn test_agent_worker_volume_e2e(db: Pool) -> anyhow::Result<()> let lfs_config = json!({ "type": "FilesystemStorage", "root_path": storage_root, + "volume_storage": "primary", "public_resource": null, "advanced_permissions": null }); @@ -306,7 +307,11 @@ async fn test_agent_worker_volume_e2e(db: Pool) -> anyhow::Result<()> .await?; // 2. Pre-populate the volume with a file - let vol_dir = storage_dir.path().join("volumes").join("test-vol"); + let vol_dir = storage_dir + .path() + .join("volumes") + .join("test-workspace") + .join("test-vol"); std::fs::create_dir_all(&vol_dir)?; std::fs::write(vol_dir.join("hello.txt"), b"hello from volume")?; @@ -343,6 +348,7 @@ async fn test_agent_worker_volume_e2e(db: Pool) -> anyhow::Result<()> // 4. GET /file/* — download the existing file let resp = http .get(format!("{vol_base}/file/hello.txt")) + .query(&[("worker_name", "test-worker-1")]) .send() .await?; assert!( @@ -360,6 +366,7 @@ async fn test_agent_worker_volume_e2e(db: Pool) -> anyhow::Result<()> // 5. PUT /file/* — upload a new file let resp = http .put(format!("{vol_base}/file/output.txt")) + .query(&[("worker_name", "test-worker-1")]) .body(b"written by agent worker".to_vec()) .send() .await?; @@ -432,7 +439,8 @@ async fn test_agent_worker_volume_http_worker_e2e(db: Pool) -> anyhow: "type": "FilesystemStorage", "root_path": storage_root, "public_resource": null, - "advanced_permissions": null + "advanced_permissions": null, + "volume_storage": "primary" }); sqlx::query!( @@ -444,20 +452,24 @@ async fn test_agent_worker_volume_http_worker_e2e(db: Pool) -> anyhow: .await?; // 2. Pre-populate the volume with a file - let vol_dir = storage_dir.path().join("volumes").join("test-vol"); + let vol_dir = storage_dir + .path() + .join("volumes") + .join("test-workspace") + .join("test-vol"); std::fs::create_dir_all(&vol_dir)?; std::fs::write(vol_dir.join("hello.txt"), b"hello from volume")?; // 3. Push the job, then run worker with HTTP connection (bun tag) - let code = r#"// volume: test-vol /tmp/data + let code = r#"// volume: test-vol data import { readFileSync, writeFileSync, existsSync } from "fs"; export function main() { - const content = readFileSync("/tmp/data/hello.txt", "utf-8"); - writeFileSync("/tmp/data/output.txt", "written by agent worker"); + const content = readFileSync("data/hello.txt", "utf-8"); + writeFileSync("data/output.txt", "written by agent worker"); return { read_content: content, - output_exists: existsSync("/tmp/data/output.txt"), + output_exists: existsSync("data/output.txt"), }; }"#; @@ -528,6 +540,7 @@ async fn test_agent_worker_volume_release(db: Pool) -> anyhow::Result< let lfs_config = json!({ "type": "FilesystemStorage", "root_path": storage_root, + "volume_storage": "primary", "public_resource": null, "advanced_permissions": null }); diff --git a/backend/tests/volume_tests.rs b/backend/tests/volume_tests.rs index fd780e0be1..e93979e9c7 100644 --- a/backend/tests/volume_tests.rs +++ b/backend/tests/volume_tests.rs @@ -549,7 +549,7 @@ export function main() { output_exists: existsSync("data/output.txt"), }; }"#; - run_volume_sql_worker_e2e(db, ScriptLang::Bun, code).await + run_volume_with_default_stack(db, ScriptLang::Bun, code).await } #[cfg(all(feature = "parquet", feature = "private", feature = "php"))] @@ -568,6 +568,15 @@ function main() { ]; }"#; + run_volume_with_default_stack(db, ScriptLang::Php, code).await +} + +#[cfg(feature = "parquet")] +async fn run_volume_with_default_stack( + db: Pool, + language: ScriptLang, + code: &'static str, +) -> anyhow::Result<()> { // CI raises RUST_MIN_STACK; keep the worker at Tokio's default to catch regressions. tokio::task::spawn_blocking(move || { tokio::runtime::Builder::new_multi_thread() @@ -575,7 +584,7 @@ function main() { .thread_stack_size(2 * 1024 * 1024) .enable_all() .build()? - .block_on(run_volume_sql_worker_e2e(db, ScriptLang::Php, code)) + .block_on(run_volume_sql_worker_e2e(db, language, code)) }) .await? } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index c882e413e3..d893ec071b 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -3867,7 +3867,8 @@ pub async fn run_worker( let job_result = windmill_common::log_context::with_log_context( log_ctx, async { - let result = handle_queued_job( + // Keep large job-phase futures boxed to limit debug polling frames. + let result = Box::pin(handle_queued_job( arc_job.clone(), raw_code, raw_lock, @@ -3888,7 +3889,7 @@ pub async fn run_worker( flow_runners, #[cfg(feature = "benchmark")] &mut bench, - ) + )) .await; record_job_span_status(&result); result @@ -5527,7 +5528,7 @@ async fn handle_code_execution_job( .await?; let language = language.clone(); - let result = run_language_executor( + let result = Box::pin(run_language_executor( job, conn, client, @@ -5552,7 +5553,7 @@ async fn handle_code_execution_job( &modules, false, in_pipeline, - ) + )) .await; record_declared_warehouse_write(job, conn, code, &result).await; result @@ -6466,7 +6467,7 @@ mount {{ .await; if let Connection::Sql(db) = conn { - volume_setup = crate::volume_oss::setup_volumes_sql_worker( + volume_setup = Box::pin(crate::volume_oss::setup_volumes_sql_worker( &volume_mounts, db, &job.workspace_id, @@ -6479,10 +6480,10 @@ mount {{ language, &mut envs, &mut shared_mount, - ) + )) .await?; } else if let Connection::Http(http) = conn { - volume_setup = crate::volume_oss::setup_volumes_http_worker( + volume_setup = Box::pin(crate::volume_oss::setup_volumes_http_worker( &volume_mounts, http, &job.workspace_id, @@ -6495,7 +6496,7 @@ mount {{ language, &mut envs, &mut shared_mount, - ) + )) .await?; } } @@ -6991,7 +6992,7 @@ mount {{ if let Some(ref vol_client) = volume_setup.client { if let Connection::Sql(db) = conn { - crate::volume_oss::sync_volumes_sql_worker( + Box::pin(crate::volume_oss::sync_volumes_sql_worker( &volume_setup.states, &volume_setup.writable, vol_client, @@ -7001,13 +7002,13 @@ mount {{ worker_name, conn, result.is_ok(), - ) + )) .await; } } if let Connection::Http(http) = conn { - crate::volume_oss::sync_volumes_http_worker( + Box::pin(crate::volume_oss::sync_volumes_http_worker( &volume_setup.states, &volume_setup.writable, http, @@ -7016,7 +7017,7 @@ mount {{ worker_name, conn, result.is_ok(), - ) + )) .await; }