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

improve stack size usage (#6936)

Browse Source 
* oom

* final

* final
This commit is contained in:
Ruben Fiszel
2025-10-24 22:08:54 +00:00
committed by GitHub
parent 9838d146d8
commit 9a75add1c0
9 changed files with 434 additions and 418 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.toml
+2
View File
@@ -161,6 +161,7 @@ bitflags.workspace = true
nom.workspace = true
globset.workspace = true
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemallocator = { optional = true, workspace = true }
tikv-jemalloc-sys = { optional = true, workspace = true }
@@ -265,6 +266,7 @@ aws-sigv4 = "^1.3.4"
aws-sdk-config = "=1.68.0"
async-trait = "0.1.88"
v8 = "=130.0.7" # Exact version NOTE: Do not forget to update version and hash in flake.nix
deno_fetch = "0.214.0"
deno_tls = "0.177.0"
backend/parsers/windmill-parser-py-imports/src/lib.rs
+1 -1
View File
@@ -170,10 +170,10 @@ fn parse_code_for_imports(code: &str, path: &str) -> error::Result<Vec<NImport>>
// This is needed because we've split off the real main function above
let code_with_fake_main = format!("{}\n\ndef main(): pass", code);
let ast = Suite::parse(&code_with_fake_main, "main.py").map_err(|e| {
error::Error::ExecutionErr(format!("Error parsing code for imports: {}", e.to_string()))
})?;
// Note: We're still using the original code for finding pins,
// as the TextRange values from the parsed AST would be based on code_with_fake_main
// but we want to match against the original code
backend/tests/worker.rs
+8 -7
View File
@@ -2860,13 +2860,6 @@ from wmill import task
import pandas as pd
import numpy as np
def main(n: int):
l = []
for i in range(n):
l.append(heavy_compute(i))
print(l)
return [send_result(sum(l), "example@example.com"), n]
@task()
def heavy_compute(n: int):
df = pd.DataFrame(np.random.randn(100, 4), columns=list('ABCD'))
@@ -2876,6 +2869,14 @@ def heavy_compute(n: int):
def send_result(res: int, email: str):
print(f"Sending result {res} to {email}")
return "OK"
def main(n: int):
l = []
for i in range(n):
l.append(heavy_compute(i))
print(l)
return [send_result(sum(l), "example@example.com"), n]
"#;
backend/windmill-common/src/worker.rs
+1 -1
View File
@@ -430,7 +430,7 @@ fn format_pull_query(peek: String) -> String {
",
peek
);
tracing::debug!("pull query: {}", r);
// tracing::debug!("pull query: {}", r);
r
}
backend/windmill-queue/src/jobs.rs
+2 -2
View File
@@ -1140,13 +1140,13 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
.unwrap_or(false);
if schedule_next_tick {
if let Err(err) = handle_maybe_scheduled_job(
if let Err(err) = Box::pin(handle_maybe_scheduled_job(
db,
queued_job,
&schedule,
&script_path,
&queued_job.workspace_id,
)
))
.await
{
match err {
backend/windmill-worker/src/python_executor.rs
+3 -2
View File
@@ -546,6 +546,7 @@ pub async fn handle_python_job(
precomputed_agent_info: Option<PrecomputedAgentInfo>,
has_stream: &mut bool,
) -> windmill_common::error::Result<Box<RawValue>> {
let script_path = crate::common::use_flow_root_path(job.runnable_path());
let annotations = PythonAnnotations::parse(inner_content);
@@ -1181,13 +1182,13 @@ async fn handle_python_deps(
let (v, requirements_lines, error_hint) = match conn {
Connection::Sql(db) => {
let mut version_specifiers = vec![];
let (r, h) = windmill_parser_py_imports::parse_python_imports(
let (r, h) = Box::pin(windmill_parser_py_imports::parse_python_imports(
inner_content,
w_id,
script_path,
db,
&mut version_specifiers,
)
))
.await?;
let v = PyV::resolve(
backend/windmill-worker/src/worker.rs
+407 -395
View File
File diff suppressed because it is too large Load Diff
backend/windmill-worker/src/worker_flow.rs
+8 -8
View File
@@ -104,7 +104,7 @@ pub async fn update_flow_status_after_job_completion(
let mut unrecoverable = unrecoverable;
loop {
potentially_crash_for_testing();
let nrec = match update_flow_status_after_job_completion_internal(
let nrec = match Box::pin(update_flow_status_after_job_completion_internal(
db,
client,
rec.flow,
@@ -122,13 +122,13 @@ pub async fn update_flow_status_after_job_completion(
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
bench,
)
))
.await
{
Ok(j) => j,
Err(e) => {
tracing::error!("Error while updating flow status of {} after completion of {}, updating flow status again with error: {e:#}", rec.flow, &rec.job_id_for_status);
update_flow_status_after_job_completion_internal(
Box::pin(update_flow_status_after_job_completion_internal(
db,
client,
rec.flow,
@@ -148,7 +148,7 @@ pub async fn update_flow_status_after_job_completion(
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
bench,
)
))
.await?
}
};
@@ -1508,7 +1508,7 @@ pub async fn update_flow_status_after_job_completion_internal(
true
} else {
tracing::debug!(id = %flow_job.id, "start handle flow");
match handle_flow(
match Box::pin(handle_flow(
flow_job.clone(),
&flow_data,
db,
@@ -1518,7 +1518,7 @@ pub async fn update_flow_status_after_job_completion_internal(
worker_dir,
job_completed_tx,
worker_name,
)
))
.warn_after_seconds(10)
.await
{
@@ -1963,7 +1963,7 @@ pub async fn handle_flow(
let mut rec = PushNextFlowJobRec { flow_job: flow_job, status: status };
loop {
let PushNextFlowJobRec { flow_job, status } = rec;
let next = push_next_flow_job(
let next = Box::pin(push_next_flow_job(
flow_job,
status,
flow,
@@ -1973,7 +1973,7 @@ pub async fn handle_flow(
same_worker_tx,
worker_dir,
worker_name,
)
))
.warn_after_seconds(10)
.await?;
match next {
backend/windmill-worker/src/worker_lockfiles.rs
+2 -2
View File
@@ -417,7 +417,7 @@ pub async fn process_relative_imports(
// Script might have no relative imports but still be referenced by someone else.
match timeout(
core::time::Duration::from_secs(60),
trigger_dependents_to_recompute_dependencies(
Box::pin(trigger_dependents_to_recompute_dependencies(
w_id,
script_path,
deployment_message,
@@ -427,7 +427,7 @@ pub async fn process_relative_imports(
permissioned_as,
db,
already_visited,
),
)),
)
.warn_after_seconds(10)
.await