diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 6d244fe229..e2134cb6c5 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -539,15 +539,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "atoi" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7c57d12312ff59c811c0643f4d80830505833c9ffaebd193d819392b265be8e" -dependencies = [ - "num-traits", -] - [[package]] name = "atoi" version = "2.0.0" @@ -1073,7 +1064,7 @@ dependencies = [ "async-trait", "futures-channel", "futures-util", - "parking_lot 0.12.1", + "parking_lot", "tokio", ] @@ -1879,7 +1870,7 @@ dependencies = [ "bitflags 2.4.1", "crossterm_winapi", "libc", - "parking_lot 0.12.1", + "parking_lot", "winapi", ] @@ -2010,7 +2001,7 @@ dependencies = [ "hashbrown 0.14.3", "lock_api", "once_cell", - "parking_lot_core 0.9.9", + "parking_lot_core", ] [[package]] @@ -2129,7 +2120,7 @@ dependencies = [ "libc", "log", "once_cell", - "parking_lot 0.12.1", + "parking_lot", "pin-project", "serde", "serde_json", @@ -2352,22 +2343,13 @@ dependencies = [ "subtle", ] -[[package]] -name = "dirs" -version = "4.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca3aa72a6f96ea37bbc5aa912f6788242832f75369bdfdadcb0e38423f100059" -dependencies = [ - "dirs-sys 0.3.7", -] - [[package]] name = "dirs" version = "5.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "44c45a9d03d6676652bcb5e724c7e988de1acad23a711b5217ab9cbecbec2225" dependencies = [ - "dirs-sys 0.4.1", + "dirs-sys", ] [[package]] @@ -2380,17 +2362,6 @@ dependencies = [ "dirs-sys-next", ] -[[package]] -name = "dirs-sys" -version = "0.3.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b1d1d91c932ef41c0f2663aa8b0ca0342d444d842c06914aa0a7e352d0bada6" -dependencies = [ - "libc", - "redox_users", - "winapi", -] - [[package]] name = "dirs-sys" version = "0.4.1" @@ -2945,17 +2916,6 @@ dependencies = [ "futures-util", ] -[[package]] -name = "futures-intrusive" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a604f7a68fbf8103337523b1fadc8ade7361ee3f112f7c680ad179651616aed5" -dependencies = [ - "futures-core", - "lock_api", - "parking_lot 0.11.2", -] - [[package]] name = "futures-intrusive" version = "0.5.0" @@ -2964,7 +2924,7 @@ checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f" dependencies = [ "futures-core", "lock_api", - "parking_lot 0.12.1", + "parking_lot", ] [[package]] @@ -3439,7 +3399,7 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b780635574b3d92f036890d8373433d6f9fc7abb320ee42a5c25897fc8ed732" dependencies = [ - "dirs 5.0.1", + "dirs", "indicatif", "log", "native-tls", @@ -3910,9 +3870,9 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" [[package]] name = "libc" -version = "0.2.151" +version = "0.2.152" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "302d7ab3130588088d277783b1e2d2e10c9e9e4a16dd9050e6ec93fb3e7048f4" +checksum = "13e3bf6590cbc649f4d1a3eefc9d5d6eb746f5200ffb04e5e142700b8faa56e7" [[package]] name = "libgit2-sys" @@ -4614,7 +4574,7 @@ dependencies = [ "humantime", "hyper", "itertools 0.11.0", - "parking_lot 0.12.1", + "parking_lot", "percent-encoding", "quick-xml 0.31.0", "rand 0.8.5", @@ -4753,17 +4713,6 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bb813b8af86854136c6922af0598d719255ecb2179515e6e7730d468f05c9cae" -[[package]] -name = "parking_lot" -version = "0.11.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99" -dependencies = [ - "instant", - "lock_api", - "parking_lot_core 0.8.6", -] - [[package]] name = "parking_lot" version = "0.12.1" @@ -4771,21 +4720,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f" dependencies = [ "lock_api", - "parking_lot_core 0.9.9", -] - -[[package]] -name = "parking_lot_core" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "60a2cfe6f0ad2bfc16aefa463b497d5c7a5ecd44a23efa72aa342d90177356dc" -dependencies = [ - "cfg-if", - "instant", - "libc", - "redox_syscall 0.2.16", - "smallvec", - "winapi", + "parking_lot_core", ] [[package]] @@ -4923,14 +4858,11 @@ dependencies = [ "archiver-rs", "async-trait", "bytes", - "dirs 5.0.1", + "dirs", "futures", "lazy_static", "log", - "reqwest", - "sqlx 0.6.3", "thiserror", - "tokio", "zip", ] @@ -5177,7 +5109,7 @@ checksum = "dd503430a6d9779b07915d858865fe998317ef3cfef8973881f578ac5d4baae7" dependencies = [ "ahash 0.8.7", "arrow-format", - "atoi 2.0.0", + "atoi", "atoi_simd", "bytemuck", "chrono", @@ -5445,7 +5377,7 @@ version = "0.35.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aae56f79e9cedd617773c1c8f5ca84a31a8b1d593714959d5f799e7bdd98fe51" dependencies = [ - "atoi 2.0.0", + "atoi", "chrono", "now", "once_cell", @@ -5764,7 +5696,7 @@ dependencies = [ "fnv", "lazy_static", "memchr", - "parking_lot 0.12.1", + "parking_lot", "thiserror", ] @@ -5870,7 +5802,7 @@ dependencies = [ "ahash 0.8.7", "equivalent", "hashbrown 0.14.3", - "parking_lot 0.12.1", + "parking_lot", ] [[package]] @@ -6043,15 +5975,6 @@ dependencies = [ "url", ] -[[package]] -name = "redox_syscall" -version = "0.2.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb5a58c1855b4b6819d59012155603f0b22ad30cad752600aadfcb695265519a" -dependencies = [ - "bitflags 1.3.2", -] - [[package]] name = "redox_syscall" version = "0.3.5" @@ -7294,82 +7217,19 @@ dependencies = [ "log", ] -[[package]] -name = "sqlx" -version = "0.6.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8de3b03a925878ed54a954f621e64bf55a3c1bd29652d0d1a17830405350188" -dependencies = [ - "sqlx-core 0.6.3", - "sqlx-macros 0.6.3", -] - [[package]] name = "sqlx" version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dba03c279da73694ef99763320dea58b51095dfe87d001b1d4b5fe78ba8763cf" dependencies = [ - "sqlx-core 0.7.3", - "sqlx-macros 0.7.3", + "sqlx-core", + "sqlx-macros", "sqlx-mysql", "sqlx-postgres", "sqlx-sqlite", ] -[[package]] -name = "sqlx-core" -version = "0.6.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa8241483a83a3f33aa5fff7e7d9def398ff9990b2752b6c6112b83c6d246029" -dependencies = [ - "ahash 0.7.7", - "atoi 1.0.0", - "base64 0.13.1", - "bitflags 1.3.2", - "byteorder", - "bytes", - "crc", - "crossbeam-queue", - "dirs 4.0.0", - "dotenvy", - "either", - "event-listener 2.5.3", - "futures-channel", - "futures-core", - "futures-intrusive 0.4.2", - "futures-util", - "hashlink", - "hex", - "hkdf", - "hmac", - "indexmap 1.9.3", - "itoa", - "libc", - "log", - "md-5 0.10.6", - "memchr", - "once_cell", - "paste", - "percent-encoding", - "rand 0.8.5", - "rustls 0.20.9", - "rustls-pemfile", - "serde", - "serde_json", - "sha1", - "sha2 0.10.8", - "smallvec", - "sqlformat", - "sqlx-rt", - "stringprep", - "thiserror", - "tokio-stream", - "url", - "webpki-roots 0.22.6", - "whoami", -] - [[package]] name = "sqlx-core" version = "0.7.3" @@ -7377,7 +7237,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d84b0a3c3739e220d94b3239fd69fb1f74bc36e16643423bd99de3b43c21bfbd" dependencies = [ "ahash 0.8.7", - "atoi 2.0.0", + "atoi", "bigdecimal 0.3.1", "byteorder", "bytes", @@ -7389,7 +7249,7 @@ dependencies = [ "event-listener 2.5.3", "futures-channel", "futures-core", - "futures-intrusive 0.5.0", + "futures-intrusive", "futures-io", "futures-util", "hashlink", @@ -7416,25 +7276,6 @@ dependencies = [ "webpki-roots 0.25.3", ] -[[package]] -name = "sqlx-macros" -version = "0.6.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9966e64ae989e7e575b19d7265cb79d7fc3cbbdf179835cb0d716f294c2049c9" -dependencies = [ - "dotenvy", - "either", - "heck", - "once_cell", - "proc-macro2", - "quote", - "sha2 0.10.8", - "sqlx-core 0.6.3", - "sqlx-rt", - "syn 1.0.109", - "url", -] - [[package]] name = "sqlx-macros" version = "0.7.3" @@ -7443,7 +7284,7 @@ checksum = "89961c00dc4d7dffb7aee214964b065072bff69e36ddb9e2c107541f75e4f2a5" dependencies = [ "proc-macro2", "quote", - "sqlx-core 0.7.3", + "sqlx-core", "sqlx-macros-core", "syn 1.0.109", ] @@ -7465,7 +7306,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.8", - "sqlx-core 0.7.3", + "sqlx-core", "sqlx-mysql", "sqlx-postgres", "sqlx-sqlite", @@ -7481,7 +7322,7 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e37195395df71fd068f6e2082247891bc11e3289624bbc776a0cdfa1ca7f1ea4" dependencies = [ - "atoi 2.0.0", + "atoi", "base64 0.21.5", "bigdecimal 0.3.1", "bitflags 2.4.1", @@ -7512,7 +7353,7 @@ dependencies = [ "sha1", "sha2 0.10.8", "smallvec", - "sqlx-core 0.7.3", + "sqlx-core", "stringprep", "thiserror", "tracing", @@ -7526,7 +7367,7 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6ac0ac3b7ccd10cc96c7ab29791a7dd236bd94021f31eec7ba3d46a74aa1c24" dependencies = [ - "atoi 2.0.0", + "atoi", "base64 0.21.5", "bigdecimal 0.3.1", "bitflags 2.4.1", @@ -7555,7 +7396,7 @@ dependencies = [ "sha1", "sha2 0.10.8", "smallvec", - "sqlx-core 0.7.3", + "sqlx-core", "stringprep", "thiserror", "tracing", @@ -7563,36 +7404,25 @@ dependencies = [ "whoami", ] -[[package]] -name = "sqlx-rt" -version = "0.6.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "804d3f245f894e61b1e6263c84b23ca675d96753b5abfd5cc8597d86806e8024" -dependencies = [ - "once_cell", - "tokio", - "tokio-rustls 0.23.4", -] - [[package]] name = "sqlx-sqlite" version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "210976b7d948c7ba9fced8ca835b11cbb2d677c59c79de41ac0d397e14547490" dependencies = [ - "atoi 2.0.0", + "atoi", "chrono", "flume", "futures-channel", "futures-core", "futures-executor", - "futures-intrusive 0.5.0", + "futures-intrusive", "futures-util", "libsqlite3-sys", "log", "percent-encoding", "serde", - "sqlx-core 0.7.3", + "sqlx-core", "tracing", "url", "urlencoding", @@ -7653,7 +7483,7 @@ checksum = "f91138e76242f575eb1d3b38b4f1362f10d3a43f47d182a5b359af488a02293b" dependencies = [ "new_debug_unreachable", "once_cell", - "parking_lot 0.12.1", + "parking_lot", "phf_shared 0.10.0", "precomputed-hash", "serde", @@ -7697,9 +7527,9 @@ dependencies = [ [[package]] name = "strsim" -version = "0.10.0" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623" +checksum = "ccbca6f34534eb78dbee83f6b2c9442fea7113f43d9e80ea320f0972ae5dc08d" [[package]] name = "strum" @@ -8399,7 +8229,7 @@ dependencies = [ "libc", "mio", "num_cpus", - "parking_lot 0.12.1", + "parking_lot", "pin-project-lite", "signal-hook-registry", "socket2 0.5.5", @@ -8453,7 +8283,7 @@ dependencies = [ "futures-channel", "futures-util", "log", - "parking_lot 0.12.1", + "parking_lot", "percent-encoding", "phf 0.11.2", "pin-project-lite", @@ -8661,7 +8491,7 @@ dependencies = [ "cookie", "futures-util", "http", - "parking_lot 0.12.1", + "parking_lot", "pin-project-lite", "tower-layer", "tower-service", @@ -9462,7 +9292,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.8", - "sqlx 0.7.3", + "sqlx", "tokio", "tokio-metrics", "tracing", @@ -9525,7 +9355,7 @@ dependencies = [ "serde_urlencoded", "sha2 0.10.8", "sql-builder", - "sqlx 0.7.3", + "sqlx", "tempfile", "time", "tinyvector", @@ -9574,7 +9404,7 @@ dependencies = [ "serde", "serde_json", "sql-builder", - "sqlx 0.7.3", + "sqlx", "tracing", "windmill-common", ] @@ -9602,7 +9432,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.8", - "sqlx 0.7.3", + "sqlx", "thiserror", "tokio", "tracing", @@ -9614,12 +9444,12 @@ dependencies = [ [[package]] name = "windmill-git-sync" -version = "1.235.0" +version = "1.235.1" dependencies = [ "rsmq_async", "serde", "serde_json", - "sqlx 0.7.3", + "sqlx", "tracing", "windmill-common", "windmill-queue", @@ -9690,7 +9520,7 @@ dependencies = [ "regex", "rustpython-parser", "serde_json", - "sqlx 0.7.3", + "sqlx", "windmill-common", "windmill-parser", ] @@ -9764,7 +9594,7 @@ dependencies = [ "serde_json", "serde_urlencoded", "sql-builder", - "sqlx 0.7.3", + "sqlx", "tokio", "tracing", "ulid", @@ -9815,7 +9645,7 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.8", - "sqlx 0.7.3", + "sqlx", "tiberius", "tokio", "tokio-postgres", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index fcbfb0b8db..afe42347b0 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -68,7 +68,7 @@ uuid.workspace = true gethostname.workspace = true serde_json.workspace = true serde.workspace = true -pg-embed = {git = "https://github.com/faokunega/pg-embed", optional = true} +pg-embed = {git = "https://github.com/faokunega/pg-embed", optional = true, default-features = false} [dev-dependencies] serde_json.workspace = true diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index ec9067aba7..d372f7c5cb 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -35,8 +35,8 @@ use windmill_common::{ BASE_URL, DB, METRICS_DEBUG_ENABLED, METRICS_ENABLED, }; use windmill_worker::{ - create_token_for_owner, handle_job_error, AuthedClient, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, - NPM_CONFIG_REGISTRY, PIP_EXTRA_INDEX_URL, SCRIPT_TOKEN_EXPIRY, + create_token_for_owner, handle_job_error, AuthedClient, SendResult, JOB_DEFAULT_TIMEOUT, + KEEP_JOB_DIR, NPM_CONFIG_REGISTRY, PIP_EXTRA_INDEX_URL, SCRIPT_TOKEN_EXPIRY, }; #[cfg(feature = "enterprise")] @@ -757,6 +757,7 @@ async fn handle_zombie_jobs // since the job is unrecoverable, the same worker queue should never be sent anything let (same_worker_tx_never_used, _same_worker_rx_never_used) = mpsc::channel::(1); + let (send_result_never_used, _send_result_rx_never_used) = mpsc::channel::(1); let token = create_token_for_owner( &db, @@ -795,6 +796,7 @@ async fn handle_zombie_jobs "", rsmq.clone(), worker_name, + send_result_never_used, ) .await; } diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 143a9ffe08..ee77635d9f 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -1805,7 +1805,7 @@ def main(): } #[sqlx::test(fixtures("base"))] -async fn test_empty_loop(db: Pool) { +async fn test_empty_loop_1(db: Pool) { initialize_tracing().await; let server = ApiServer::start(db.clone()).await; let port = server.addr.port(); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index e8a3ba2756..b0c9d31195 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -475,6 +475,7 @@ async fn handle_receive_completed_job< worker_name: &str, worker_save_completed_job_duration: Option>, worker_flow_transition_duration: Option>, + job_completed_tx: Sender, ) { let token = jc.token.clone(); let workspace = jc.job.workspace_id.clone(); @@ -497,6 +498,7 @@ async fn handle_receive_completed_job< worker_name, worker_save_completed_job_duration, worker_flow_transition_duration, + job_completed_tx.clone(), ) .await { @@ -512,6 +514,7 @@ async fn handle_receive_completed_job< &worker_dir, rsmq.clone(), worker_name, + job_completed_tx, ) .await; } @@ -519,7 +522,7 @@ async fn handle_receive_completed_job< #[derive(Clone)] pub struct JobCompletedSender( - Sender, + Sender, Option>>, Option>, ); @@ -528,12 +531,12 @@ impl JobCompletedSender { pub async fn send( &self, jc: JobCompleted, - ) -> Result<(), tokio::sync::mpsc::error::SendError> { + ) -> Result<(), tokio::sync::mpsc::error::SendError> { if let Some(wj) = self.1.as_ref() { wj.inc() } let timer = self.2.as_ref().map(|x| x.start_timer()); - let r = self.0.send(jc).await; + let r = self.0.send(SendResult::JobCompleted(jc)).await; timer.map(|x| x.stop_and_record()); r } @@ -879,7 +882,7 @@ pub async fn run_worker(5); - let (job_completed_tx, mut job_completed_rx) = mpsc::channel::(3); + let (job_completed_tx, mut job_completed_rx) = mpsc::channel::(3); let job_completed_tx = JobCompletedSender( job_completed_tx, @@ -967,119 +970,167 @@ pub async fn run_worker { + if let Some(wj) = worker_job_completed_channel_queue2.as_ref() { + wj.dec(); } - tokio::time::sleep(Duration::from_millis(3)).await; - } + let base_internal_url2 = base_internal_url2.clone(); + let worker_dir2 = worker_dir2.clone(); + let db2 = db2.clone(); + let same_worker_tx2 = same_worker_tx2.clone(); + let rsmq2 = rsmq2.clone(); + let worker_name = worker_name2.clone(); + if matches!(jc.job.job_kind, JobKind::Noop) || is_dedicated_worker { + thread_count.fetch_add(1, Ordering::SeqCst); + let thread_count = thread_count.clone(); - #[cfg(feature = "benchmark")] - let send_duration = send_duration2.clone(); - #[cfg(feature = "benchmark")] - let process_duration = process_duration.clone(); - #[cfg(feature = "benchmark")] - let completed_jobs = completed_jobs.clone(); - #[cfg(feature = "benchmark")] - let main_duration = main_duration2.clone(); - - let worker_save_completed_job_duration2 = - worker_save_completed_job_duration.clone(); - let worker_flow_transition_duration2 = worker_flow_transition_duration.clone(); - let killpill_tx = killpill_tx2.clone(); - tokio::spawn(async move { - #[cfg(feature = "benchmark")] - let process_start = Instant::now(); - - let is_dependency_job = matches!( - jc.job.job_kind, - JobKind::Dependencies | JobKind::FlowDependencies - ); - handle_receive_completed_job( - jc, - base_internal_url2, - db2.clone(), - worker_dir2, - same_worker_tx2, - rsmq2, - &worker_name, - worker_save_completed_job_duration2.clone(), - worker_flow_transition_duration2.clone(), - ) - .await; - #[cfg(feature = "benchmark")] - { - let n = completed_jobs.fetch_add(1, Ordering::SeqCst); - if (n + 1) % 1000 == 0 || n == (jobs - 1) as usize { - let duration_s = start.elapsed().as_secs_f64(); - let jobs_per_sec = n as f64 / duration_s; - tracing::info!( - "completed {} jobs in {}s, {} jobs/s", - n + 1, - duration_s, - jobs_per_sec - ); - - tracing::info!( - "main loop without send {}s", - main_duration.load(Ordering::SeqCst) as f64 / 1000.0 - ); - - tracing::info!( - "send job completed / send dedicated job duration {}s", - send_duration.load(Ordering::SeqCst) as f64 / 1000.0 - ); - - tracing::info!( - "job completed process duration {}s", - process_duration.load(Ordering::SeqCst) as f64 / 1000.0 - ); + loop { + if thread_count.load(Ordering::Relaxed) < 4 { + break; + } + tokio::time::sleep(Duration::from_millis(3)).await; } - process_duration.fetch_add( - process_start.elapsed().as_millis() as usize, - Ordering::SeqCst, - ); - } + #[cfg(feature = "benchmark")] + let send_duration = send_duration2.clone(); + #[cfg(feature = "benchmark")] + let process_duration = process_duration.clone(); + #[cfg(feature = "benchmark")] + let completed_jobs = completed_jobs.clone(); + #[cfg(feature = "benchmark")] + let main_duration = main_duration2.clone(); - thread_count.fetch_sub(1, Ordering::SeqCst); - if is_dependency_job && is_dedicated_worker { - tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted."); - sqlx::query!("UPDATE config SET config = config WHERE name = $1", format!("worker__{}", *WORKER_GROUP)) + let worker_save_completed_job_duration2 = + worker_save_completed_job_duration.clone(); + let worker_flow_transition_duration2 = + worker_flow_transition_duration.clone(); + let killpill_tx = killpill_tx2.clone(); + let job_completed_sender = job_completed_sender.clone(); + tokio::spawn(async move { + #[cfg(feature = "benchmark")] + let process_start = Instant::now(); + + let is_dependency_job = matches!( + jc.job.job_kind, + JobKind::Dependencies | JobKind::FlowDependencies + ); + handle_receive_completed_job( + jc, + base_internal_url2, + db2.clone(), + worker_dir2, + same_worker_tx2, + rsmq2, + &worker_name, + worker_save_completed_job_duration2.clone(), + worker_flow_transition_duration2.clone(), + job_completed_sender.clone(), + ) + .await; + #[cfg(feature = "benchmark")] + { + let n = completed_jobs.fetch_add(1, Ordering::SeqCst); + if (n + 1) % 1000 == 0 || n == (jobs - 1) as usize { + let duration_s = start.elapsed().as_secs_f64(); + let jobs_per_sec = n as f64 / duration_s; + tracing::info!( + "completed {} jobs in {}s, {} jobs/s", + n + 1, + duration_s, + jobs_per_sec + ); + + tracing::info!( + "main loop without send {}s", + main_duration.load(Ordering::SeqCst) as f64 / 1000.0 + ); + + tracing::info!( + "send job completed / send dedicated job duration {}s", + send_duration.load(Ordering::SeqCst) as f64 / 1000.0 + ); + + tracing::info!( + "job completed process duration {}s", + process_duration.load(Ordering::SeqCst) as f64 / 1000.0 + ); + } + + process_duration.fetch_add( + process_start.elapsed().as_millis() as usize, + Ordering::SeqCst, + ); + } + + thread_count.fetch_sub(1, Ordering::SeqCst); + if is_dependency_job && is_dedicated_worker { + tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted."); + sqlx::query!("UPDATE config SET config = config WHERE name = $1", format!("worker__{}", *WORKER_GROUP)) .execute(&db2) .await .expect("update config to trigger restart of all dedicated workers at that config"); - killpill_tx.send(()).unwrap_or_default(); + killpill_tx.send(()).unwrap_or_default(); + } + }); + } else { + handle_receive_completed_job( + jc, + base_internal_url2, + db2, + worker_dir2, + same_worker_tx2, + rsmq2, + &worker_name, + worker_save_completed_job_duration2.clone(), + worker_flow_transition_duration2.clone(), + job_completed_sender.clone(), + ) + .await; } - }); - } else { - handle_receive_completed_job( - jc, - base_internal_url2, - db2, - worker_dir2, - same_worker_tx2, - rsmq2, - &worker_name, - worker_save_completed_job_duration2.clone(), - worker_flow_transition_duration2.clone(), - ) - .await; + } + SendResult::UpdateFlow { + flow, + w_id, + success, + result, + worker_dir, + stop_early_override, + token, + } => { + // let r; + if let Err(e) = update_flow_status_after_job_completion( + &db2, + &AuthedClient { + base_internal_url: base_internal_url2.to_string(), + workspace: w_id.clone(), + token: token.clone(), + force_client: None, + }, + flow, + &Uuid::nil(), + &w_id, + success, + &result, + true, + same_worker_tx2.clone(), + &worker_dir, + stop_early_override, + rsmq2.clone(), + &worker_name2, + job_completed_sender.clone(), + ) + .await + { + tracing::error!("Error updating flow status after job completion: {e}"); + } + } + SendResult::Kill => { + break; + } } } @@ -1336,6 +1387,7 @@ pub async fn run_worker { @@ -1603,6 +1655,7 @@ pub async fn run_worker>, worker_flow_transition_duration: Option>, + job_completed_tx: Sender, ) -> windmill_common::error::Result<()> { if success { // println!("bef completed job{:?}", SystemTime::now()); @@ -2129,6 +2183,7 @@ pub async fn process_completed_job, worker_name: &str, + job_completed_tx: Sender, ) { let err = match err { Error::JsonErr(err) => err, @@ -2266,6 +2323,7 @@ pub async fn handle_job_error Box { ); } +pub enum SendResult { + JobCompleted(JobCompleted), + UpdateFlow { + flow: Uuid, + w_id: String, + success: bool, + result: Box, + worker_dir: String, + stop_early_override: Option, + token: String, + }, + Kill, +} + +// db: &DB, +// client: &AuthedClient, +// flow: uuid::Uuid, +// job_id_for_status: &Uuid, +// w_id: &str, +// success: bool, +// result: &'a RawValue, +// unrecoverable: bool, +// same_worker_tx: Sender, +// worker_dir: &str, +// stop_early_override: Option, +// rsmq: Option, +// worker_name: &str, + #[derive(Debug, Clone)] pub struct JobCompleted { pub job: Arc, @@ -2470,7 +2556,7 @@ async fn handle_queued_job( same_worker_tx, worker_dir, rsmq, - worker_name, + job_completed_tx.0.clone(), ) .await?; timer.map(|x| x.stop_and_record()); diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index aa71fa640b..5468407f69 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -15,9 +15,8 @@ use std::time::Duration; use crate::common::{hash_args, save_in_cache}; use crate::js_eval::{eval_timeout, IdContext}; -use crate::{AuthedClient, PreviousResult, KEEP_JOB_DIR}; +use crate::{AuthedClient, PreviousResult, SendResult, KEEP_JOB_DIR}; use anyhow::Context; -use async_recursion::async_recursion; use serde::Serialize; use serde_json::value::RawValue; use serde_json::{json, Value}; @@ -70,8 +69,10 @@ pub async fn update_flow_status_after_job_completion< stop_early_override: Option, rsmq: Option, worker_name: &str, + job_completed_tx: Sender, ) -> error::Result<()> { // this is manual tailrecursion because async_recursion blows up the stack + // todo!(); let mut rec = update_flow_status_after_job_completion_internal( db, client, @@ -87,6 +88,7 @@ pub async fn update_flow_status_after_job_completion< false, rsmq.clone(), worker_name, + job_completed_tx.clone(), ) .await?; while let Some(nrec) = rec { @@ -105,6 +107,7 @@ pub async fn update_flow_status_after_job_completion< nrec.skip_error_handler, rsmq.clone(), worker_name, + job_completed_tx.clone(), ) .await { @@ -125,6 +128,7 @@ pub async fn update_flow_status_after_job_completion< nrec.skip_error_handler, rsmq.clone(), worker_name, + job_completed_tx.clone(), ) .await? } @@ -167,6 +171,7 @@ pub async fn update_flow_status_after_job_completion_internal< skip_error_handler: bool, rsmq: Option, worker_name: &str, + job_completed_tx: Sender, ) -> error::Result> { let ( should_continue_flow, @@ -575,6 +580,8 @@ pub async fn update_flow_status_after_job_completion_internal< } tx.commit().await?; + tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow status updated"); + ( should_continue_flow, flow_job, @@ -686,6 +693,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( &flow_job, db, @@ -694,7 +702,7 @@ pub async fn update_flow_status_after_job_completion_internal< same_worker_tx.clone(), worker_dir, rsmq.clone(), - worker_name, + job_completed_tx, ) .await { @@ -1037,7 +1045,7 @@ pub async fn handle_flow( same_worker_tx: Sender, worker_dir: &str, rsmq: Option, - worker_name: &str, + job_completed_tx: Sender, ) -> anyhow::Result<()> { let flow = flow_job .parse_raw_flow() @@ -1056,7 +1064,7 @@ pub async fn handle_flow( same_worker_tx, worker_dir, rsmq, - worker_name, + job_completed_tx, ) .await?; Ok(()) @@ -1089,7 +1097,7 @@ pub struct RawArgs { pub args: Option>>>, } -#[async_recursion] +// #[async_recursion] // #[instrument(level = "trace", skip_all)] async fn push_next_flow_job( flow_job: &QueuedJob, @@ -1101,7 +1109,7 @@ async fn push_next_flow_job same_worker_tx: Sender, worker_dir: &str, rsmq: Option, - worker_name: &str, + job_completed_tx: Sender, ) -> error::Result<()> { let job_root = flow_job .root_job @@ -1122,29 +1130,51 @@ async fn push_next_flow_job // if this is an empty module of if the module has already been completed, successfully, update the parent flow if flow.modules.is_empty() || matches!(status_module, FlowStatusModule::Success { .. }) { - let r; - return update_flow_status_after_job_completion( - db, - client, - flow_job.id, - &Uuid::nil(), - flow_job.workspace_id.as_str(), - true, - if flow.modules.is_empty() { - r = to_raw_value(&flow_job_args); - &r - } else { - // it has to be an empty for loop event - serde_json::from_str("[]").unwrap() - }, - true, - same_worker_tx, - worker_dir, - None, - rsmq, - worker_name, - ) - .await; + job_completed_tx + .send(SendResult::UpdateFlow { + flow: flow_job.id, + success: true, + result: if flow.modules.is_empty() { + to_raw_value(&flow_job_args) + } else { + // it has to be an empty for loop event + serde_json::from_str("[]").unwrap() + }, + stop_early_override: None, + w_id: flow_job.workspace_id.clone(), + worker_dir: worker_dir.to_string(), + token: client.token.clone(), + }) + .await + .map_err(|e| { + Error::InternalErr(format!( + "error sending update flow message to job completed channel: {e}" + )) + })?; + // let r; + // return update_flow_status_after_job_completion( + // db, + // client, + // flow_job.id, + // &Uuid::nil(), + // flow_job.workspace_id.as_str(), + // true, + // if flow.modules.is_empty() { + // r = to_raw_value(&flow_job_args); + // &r + // } else { + // // it has to be an empty for loop event + // serde_json::from_str("[]").unwrap() + // }, + // true, + // same_worker_tx, + // worker_dir, + // None, + // rsmq, + // worker_name, + // ) + // .await; + return Ok(()); } let arc_flow_job_args = Arc::new(flow_job_args.clone()); @@ -1166,25 +1196,45 @@ async fn push_next_flow_job flow_job.id ).fetch_one(db).await?.unwrap_or(0); if count > 0 { - return update_flow_status_after_job_completion( - db, - client, - flow_job.id, - &Uuid::nil(), - flow_job.workspace_id.as_str(), - true, - serde_json::from_str( - "\"not allowed to overlap, scheduling next iteration\"", - ) - .unwrap(), - true, - same_worker_tx, - worker_dir, - Some(true), - rsmq, - worker_name, - ) - .await; + job_completed_tx + .send(SendResult::UpdateFlow { + flow: flow_job.id, + success: true, + result: serde_json::from_str( + "\"not allowed to overlap, scheduling next iteration\"", + ) + .unwrap(), + stop_early_override: Some(true), + w_id: flow_job.workspace_id.clone(), + worker_dir: worker_dir.to_string(), + token: client.token.clone(), + }) + .await + .map_err(|e| { + Error::InternalErr(format!( + "error sending update flow message to job completed channel: {e}" + )) + })?; + // return update_flow_status_after_job_completion( + // db, + // client, + // flow_job.id, + // &Uuid::nil(), + // flow_job.workspace_id.as_str(), + // true, + // serde_json::from_str( + // "\"not allowed to overlap, scheduling next iteration\"", + // ) + // .unwrap(), + // true, + // same_worker_tx, + // worker_dir, + // Some(true), + // rsmq, + // worker_name, + // ) + // .await; + return Ok(()); } } } @@ -1199,22 +1249,40 @@ async fn push_next_flow_job ) .await?; if skip { - return update_flow_status_after_job_completion( - db, - client, - flow_job.id, - &Uuid::nil(), - flow_job.workspace_id.as_str(), - true, - serde_json::from_str("\"stopped early\"").unwrap(), - true, - same_worker_tx, - worker_dir, - Some(true), - rsmq, - worker_name, - ) - .await; + job_completed_tx + .send(SendResult::UpdateFlow { + flow: flow_job.id, + success: true, + result: serde_json::from_str("\"stopped early\"").unwrap(), + stop_early_override: Some(true), + w_id: flow_job.workspace_id.clone(), + worker_dir: worker_dir.to_string(), + token: client.token.clone(), + }) + .await + .map_err(|e| { + Error::InternalErr(format!( + "error sending update flow message to job completed channel: {e}" + )) + })?; + + // return update_flow_status_after_job_completion( + // db, + // client, + // flow_job.id, + // &Uuid::nil(), + // flow_job.workspace_id.as_str(), + // true, + // serde_json::from_str("\"stopped early\"").unwrap(), + // true, + // same_worker_tx, + // worker_dir, + // Some(true), + // rsmq, + // worker_name, + // ) + // .await; + return Ok(()); } } } @@ -1436,22 +1504,39 @@ async fn push_next_flow_job .await?; if flow_job.is_flow_step { if let Some(parent_job) = flow_job.parent_job { - update_flow_status_after_job_completion( - db, - client, - parent_job, - &flow_job.id, - &flow_job.workspace_id, - true, - &to_raw_value(&result), - false, - same_worker_tx.clone(), - &worker_dir, - None, - rsmq, - worker_name, - ) - .await?; + job_completed_tx + .send(SendResult::UpdateFlow { + flow: parent_job, + success: true, + result: to_raw_value(&result), + stop_early_override: Some(true), + w_id: flow_job.workspace_id.clone(), + worker_dir: worker_dir.to_string(), + token: client.token.clone(), + }) + .await + .map_err(|e| { + Error::InternalErr(format!( + "error sending update flow message to job completed channel: {e}" + )) + })?; + // update_flow_status_after_job_completion( + // db, + // client, + // parent_job, + // &flow_job.id, + // &flow_job.workspace_id, + // true, + // &to_raw_value(&result), + // false, + // same_worker_tx.clone(), + // &worker_dir, + // None, + // rsmq, + // worker_name, + // ) + // .await?; + return Ok(()); } } return Ok(()); @@ -1689,6 +1774,7 @@ async fn push_next_flow_job _ => Ok(flow_job_args), } }; + tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow job args computed"); let next_flow_transform = compute_next_flow_transform( arc_flow_job_args.clone(), @@ -1889,6 +1975,8 @@ async fn push_next_flow_job ) .await?; + tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushed next flow job: {uuid}"); + if let FlowModuleValue::ForloopFlow { parallelism: Some(p), .. } = &module.value { if i as u16 >= *p { sqlx::query!( @@ -2211,6 +2299,8 @@ async fn compute_next_flow_transform( )) }; let delete_after_use = module.delete_after_use.unwrap_or(false); + + tracing::debug!(id = %flow_job.id, "computing next flow transform for {:?}", &module.value); match &module.value { FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity), FlowModuleValue::Flow { path, .. } => { diff --git a/frontend/src/routes/(root)/(logged)/user/(user)/workspaces/+page.svelte b/frontend/src/routes/(root)/(logged)/user/(user)/workspaces/+page.svelte index 8087449ec3..79d0c7e233 100644 --- a/frontend/src/routes/(root)/(logged)/user/(user)/workspaces/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/user/(user)/workspaces/+page.svelte @@ -136,12 +136,13 @@ workspace.

{/if} - {#each nonAdminWorkspaces as workspace} + {#each nonAdminWorkspaces as workspace (workspace.id)}