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

perf: reduce shared worker debug polling frames (#11028)

Browse Source 
* 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
This commit is contained in:
Ruben Fiszel
2026-09-08 15:21:10 +02:00
committed by GitHub
parent 2cb02e3b33
commit 946756ae83
3 changed files with 44 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/agent_workers.rs
+20 -7
View File
@@ -293,6 +293,7 @@ async fn test_agent_worker_volume_e2e(db: Pool<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> anyhow::Result<
let lfs_config = json!({
"type": "FilesystemStorage",
"root_path": storage_root,
"volume_storage": "primary",
"public_resource": null,
"advanced_permissions": null
});
backend/tests/volume_tests.rs
+11 -2
View File
@@ -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<Postgres>,
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?
}
backend/windmill-worker/src/worker.rs
+13 -12
View File
@@ -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;
}