From 415be7c70fdd71f841cc6548f5ef4cfb995b601a Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 10 Apr 2023 10:55:56 +0200 Subject: [PATCH] perf: parallelize more operations for deno jobs --- backend/windmill-worker/src/jobs.rs | 4 + backend/windmill-worker/src/worker.rs | 158 ++++++++++++++++---------- 2 files changed, 102 insertions(+), 60 deletions(-) diff --git a/backend/windmill-worker/src/jobs.rs b/backend/windmill-worker/src/jobs.rs index 24b444c619..bd518ed470 100644 --- a/backend/windmill-worker/src/jobs.rs +++ b/backend/windmill-worker/src/jobs.rs @@ -6,6 +6,8 @@ * LICENSE-AGPL for a copy of the license. */ +use std::time::SystemTime; + use sqlx::{Pool, Postgres, Transaction}; use tracing::instrument; use uuid::Uuid; @@ -157,6 +159,8 @@ pub async fn add_completed_job( .execute(&mut tx) .await .map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e}")))?; + println!("{:?}", SystemTime::now()); + let _ = delete_job(db, &queued_job.workspace_id, job_id).await?; if !queued_job.is_flow_step && queued_job.job_kind != JobKind::Flow diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 3f410a2ded..09260911cf 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -17,7 +17,8 @@ use windmill_api_client::Client; use windmill_parser_go::parse_go_imports; use std::{ borrow::Borrow, collections::HashMap, io, os::unix::process::ExitStatusExt, panic, - process::Stdio, time::Duration, sync::{atomic::Ordering, Arc}, + process::Stdio, time::{Duration, SystemTime}, sync::atomic::Ordering, + sync::{Arc}, }; use tracing::{trace_span, Instrument}; use uuid::Uuid; @@ -768,6 +769,7 @@ pub async fn run_worker( } match next_job { Ok(Some(job)) => { + println!("{:?}", SystemTime::now()); let token = create_token_for_owner_in_bg(&db, &job).await; let label_values = [ @@ -800,6 +802,7 @@ pub async fn run_worker( .create(&job_dir) .await .expect("could not create job dir"); + println!("dir creation {:?}", SystemTime::now()); let same_worker = job.same_worker; @@ -824,6 +827,7 @@ pub async fn run_worker( .await .expect("could not create shared dir"); } + println!("bef token: {:?}", SystemTime::now()); let authed_client = AuthedClientBackgroundTask { base_internal_url: base_internal_url.to_string(), token: token, workspace: job.workspace_id.to_string(), client: OnceCell::new() }; let is_flow = job.job_kind == JobKind::Flow || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::FlowDependencies; @@ -1019,6 +1023,7 @@ async fn handle_queued_job( } _ => { let mut logs = "".to_string(); + println!("handle queue {:?}", SystemTime::now()); if let Some(log_str) = &job.logs { logs.push_str(&log_str); } @@ -1078,6 +1083,7 @@ async fn handle_queued_job( let client = &client.get_authed().await; match result { Ok(r) => { + println!("bef completed job{:?}", SystemTime::now()); add_completed_job(db, &job, true, false, r.clone(), logs).await?; if job.is_flow_step { if let Some(parent_job) = job.parent_job { @@ -1277,6 +1283,8 @@ mount {{ "".to_string() }; + println!("handle lang job {:?}", SystemTime::now()); + let result: error::Result = match language { None => { return Err(Error::ExecutionErr( @@ -1351,6 +1359,8 @@ mount {{ &lang_str, job.id ); + println!("handled job: {:?}", SystemTime::now()); + result } @@ -1672,67 +1682,99 @@ async fn handle_deno_job( base_internal_url: &str, worker_name: &str ) -> error::Result { + // let mut start = Instant::now(); logs.push_str("\n\n--- DENO CODE EXECUTION ---\n"); - set_logs(logs, &job.id, db).await; + + let logs_to_set = logs.clone(); + let id = job.id.clone(); + let db2 = db.clone(); + + let set_logs_f = async { + set_logs(&logs_to_set, &id, &db2).await; + Ok(()) as error::Result<()> + }; - // logs.push_str(format!("st: {:?}\n", start.elapsed().as_millis()).as_str()); - // start = Instant::now(); - let _ = write_file(job_dir, "main.ts", inner_content).await?; + let write_main_f = write_file(job_dir, "main.ts", inner_content); - let sig = trace_span!("parse_deno_signature") - .in_scope(|| windmill_parser_ts::parse_deno_signature(inner_content, true))?; - let client = client.get_authed().await; - create_args_and_out_file(&client, job, job_dir).await?; + let write_wrapper_f = async { + // let mut start = Instant::now(); + let spread = windmill_parser_ts::parse_deno_signature(inner_content, true)?.args.into_iter().map(|x| x.name).join(","); + // logs.push_str(format!("infer args: {:?}\n", start.elapsed().as_micros()).as_str()); + let wrapper_content: String = format!( + r#" + import {{ main }} from "./main.ts"; - let spread = sig.args.into_iter().map(|x| x.name).join(","); - let wrapper_content: String = format!( - r#" -import {{ main }} from "./main.ts"; + const args = await Deno.readTextFile("args.json") + .then(JSON.parse) + .then(({{ {spread} }}) => [ {spread} ]) -const args = await Deno.readTextFile("args.json") - .then(JSON.parse) - .then(({{ {spread} }}) => [ {spread} ]) - -async function run() {{ - let res: any = await main(...args); - const res_json = JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value); - await Deno.writeTextFile("result.json", res_json); - Deno.exit(0); -}} -run().catch(async (e) => {{ - await Deno.writeTextFile("result.json", JSON.stringify({{ message: e.message, name: e.name, stack: e.stack }})); - Deno.exit(1); -}}); -"#, - ); - let w_id = job.workspace_id.clone(); - let script_path_split = job.script_path().split("/"); - let script_path_parts_len = script_path_split.clone().count(); - let mut relative_mounts = "".to_string(); - for c in 0..script_path_parts_len { - relative_mounts += ",\n "; - relative_mounts += &format!("\"./{}\": \"{base_internal_url}/api/w/{w_id}/scripts/raw/p/{}{}\"", - (0..c).map(|_| "../").join(""), - &script_path_split.clone().take(script_path_parts_len - c - 1).join("/"), - if c == script_path_parts_len - 1 { "" } else { "/" }, + async function run() {{ + let res: any = await main(...args); + const res_json = JSON.stringify(res ?? null, (key, value) => typeof value === 'undefined' ? null : value); + await Deno.writeTextFile("result.json", res_json); + Deno.exit(0); + }} + run().catch(async (e) => {{ + await Deno.writeTextFile("result.json", JSON.stringify({{ message: e.message, name: e.name, stack: e.stack }})); + Deno.exit(1); + }}); + "#, ); - } - write_file(job_dir, "wrapper.ts", &wrapper_content).await?; - let import_map = format!( - r#"{{ - "imports": {{ - "/": "{base_internal_url}/api/w/{w_id}/scripts/raw/p/", - "./wrapper.ts": "./wrapper.ts", - "./main.ts": "./main.ts"{relative_mounts} - }} - }}"#, - ); - write_file(job_dir, "import_map.json", &import_map).await?; - let mut reserved_variables = get_reserved_variables(job, &client.token, db).await?; - reserved_variables.insert("RUST_LOG".to_string(), "info".to_string()); + write_file(job_dir, "wrapper.ts", &wrapper_content).await?; + Ok(()) as error::Result<()> + }; + + let write_import_map_f = async { + let w_id = job.workspace_id.clone(); + let script_path_split = job.script_path().split("/"); + let script_path_parts_len = script_path_split.clone().count(); + let mut relative_mounts = "".to_string(); + for c in 0..script_path_parts_len { + relative_mounts += ",\n "; + relative_mounts += &format!("\"./{}\": \"{base_internal_url}/api/w/{w_id}/scripts/raw/p/{}{}\"", + (0..c).map(|_| "../").join(""), + &script_path_split.clone().take(script_path_parts_len - c - 1).join("/"), + if c == script_path_parts_len - 1 { "" } else { "/" }, + ); + } + let import_map = format!( + r#"{{ + "imports": {{ + "/": "{base_internal_url}/api/w/{w_id}/scripts/raw/p/", + "./wrapper.ts": "./wrapper.ts", + "./main.ts": "./main.ts"{relative_mounts} + }} + }}"#, + ); + write_file(job_dir, "import_map.json", &import_map).await?; + Ok(()) as error::Result<()> + }; + + let reserved_variables_args_out_f = async { + let client = client.get_authed().await; + let args_and_out_f = async { + create_args_and_out_file(&client, job, job_dir).await?; + Ok(()) as Result<()> + }; + let reserved_variables_f = async { + let mut vars = get_reserved_variables(job, &client.token, db).await?; + vars.insert("RUST_LOG".to_string(), "info".to_string()); + Ok(vars) as Result> + }; + let (_, reserved_variables) = tokio::try_join!(args_and_out_f, reserved_variables_f)?; + Ok((reserved_variables, client.token)) as error::Result<(HashMap, String)> + }; + + let (_, (reserved_variables, token), _, _, _) = tokio::try_join!( + set_logs_f, + reserved_variables_args_out_f, + write_main_f, + write_wrapper_f, + write_import_map_f)?; + + let common_deno_proc_envs = get_common_deno_proc_envs(&token, base_internal_url); - let common_deno_proc_envs = get_common_deno_proc_envs(&client.token, base_internal_url); //do not cache local dependencies let reload = format!("--reload={base_internal_url}"); let child = async { @@ -1768,16 +1810,12 @@ run().catch(async (e) => {{ .stderr(Stdio::piped()) .spawn() } - .instrument(trace_span!("create_deno_jail")) .await?; - // logs.push_str(format!("prepare: {:?}\n", start.elapsed().as_millis()).as_str()); + // logs.push_str(format!("prepare: {:?}\n", start.elapsed().as_micros()).as_str()); // start = Instant::now(); handle_child(&job.id, db, logs, child, false, worker_name, &job.workspace_id).await?; // logs.push_str(format!("execute: {:?}\n", start.elapsed().as_millis()).as_str()); - // start = Instant::now(); - let r = read_result(job_dir).await; - // logs.push_str(format!("rr: {:?}\n", start.elapsed().as_millis()).as_str()); - r + read_result(job_dir).await } #[tracing::instrument(level = "trace", skip_all)]