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

perf: parallelize more operations for deno jobs

Browse Source
This commit is contained in:
Ruben Fiszel
2023-04-10 10:55:56 +02:00
parent b7b26dc9c7
commit 415be7c70f
2 changed files with 102 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/jobs.rs
+4
View File
@@ -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
backend/windmill-worker/src/worker.rs
+98 -60
View File
@@ -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<serde_json::Value> = 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<serde_json::Value> {
// 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<HashMap<String, String>>
};
let (_, reserved_variables) = tokio::try_join!(args_and_out_f, reserved_variables_f)?;
Ok((reserved_variables, client.token)) as error::Result<(HashMap<String, String>, 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)]