From 29848ae466b400e7d08eb72459fdff749137b364 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 19 Apr 2023 12:46:16 +0200 Subject: [PATCH] add barrier when num workers > 1 --- backend/src/main.rs | 19 +++++++++- backend/tests/worker.rs | 8 +++- backend/windmill-common/src/external_ip.rs | 8 +++- backend/windmill-worker/src/global_cache.rs | 31 ++++++++++------ backend/windmill-worker/src/lib.rs | 1 + backend/windmill-worker/src/worker.rs | 41 +++++++++++++-------- 6 files changed, 75 insertions(+), 33 deletions(-) diff --git a/backend/src/main.rs b/backend/src/main.rs index ac424a62bb..fba806f435 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -6,12 +6,17 @@ * LICENSE-AGPL for a copy of the license. */ -use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::{ + net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::{atomic::Ordering, Arc}, +}; use git_version::git_version; use monitor::handle_zombie_jobs_periodically; use sqlx::{Pool, Postgres}; -use windmill_common::{utils::rd_string, METRICS_ADDR}; +use tokio::sync::RwLock; +use windmill_common::{utils::rd_string, IS_READY, METRICS_ADDR}; +use windmill_worker::S3_CACHE_BUCKET; const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); const DEFAULT_NUM_WORKERS: usize = 3; @@ -266,6 +271,14 @@ pub async fn run_workers Result { - reqwest::get("https://hub.windmill.dev/getip") + reqwest::ClientBuilder::new() + .timeout(Duration::from_secs(3)) + .build()? + .get("https://hub.windmill.dev/getip") + .send() .await? .error_for_status()? .text() diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index 78acc7c56a..d906cdf70e 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -131,7 +131,7 @@ pub async fn copy_cache_to_bucket_as_tar(bucket: &str) { "copyto", &format!("{ROOT_TMP_CACHE_DIR}{TAR_CACHE_FILENAME}"), &format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}"), - "-vv", + "-v", "--size-only", "--fast-list", ], @@ -160,6 +160,17 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) { use tokio::fs::metadata; tracing::info!("Copying cache from bucket {bucket} as tar"); + + if metadata(&ROOT_TMP_CACHE_DIR).await.is_ok() { + if let Err(e) = tokio::fs::remove_dir_all(&ROOT_TMP_CACHE_DIR).await { + tracing::info!(error = %e, "Could not remove root tmp cache dir"); + } + } + + tokio::fs::create_dir_all(&ROOT_TMP_CACHE_DIR) + .await + .expect("Could not create root tmp cache dir"); + let elapsed = Instant::now(); if let Err(e) = execute_command( @@ -169,7 +180,7 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) { "copyto", &format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}"), &format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}"), - "-vv", + "-v", "--size-only", "--fast-list", ], @@ -191,6 +202,12 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) { return; } + if let Err(e) = + tokio::fs::remove_dir_all(format!("{ROOT_CACHE_DIR}deno/gen/file/tmp/windmill")).await + { + tracing::info!("Failed to remove tmp gen windmill. Error: {:?}", e); + }; + if let Err(e) = tokio::fs::remove_file(format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}")).await { tracing::info!("Failed to remove tar cache. Error: {:?}", e); return; @@ -201,16 +218,6 @@ pub async fn copy_cache_from_bucket_as_tar(bucket: &str) { elapsed.elapsed().as_secs() ); - if metadata(&ROOT_TMP_CACHE_DIR).await.is_ok() { - if let Err(e) = tokio::fs::remove_dir_all(&ROOT_TMP_CACHE_DIR).await { - tracing::info!(error = %e, "Could not remove root tmp cache dir"); - } - } - - tokio::fs::create_dir_all(&ROOT_TMP_CACHE_DIR) - .await - .expect("Could not create root tmp cache dir"); - for x in ["deno", "go", "pip"] { if let Err(e) = execute_command( TMP_DIR, diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 86cb82d4ef..ac0fdea068 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -7,4 +7,5 @@ mod python_executor; mod worker; mod worker_flow; +pub use global_cache::copy_cache_from_bucket_as_tar; pub use worker::*; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index cd89ee4400..f26692c71c 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -14,7 +14,7 @@ use sqlx::{Pool, Postgres}; use windmill_api_client::Client; use std::{ borrow::Borrow, collections::HashMap, io, os::unix::process::ExitStatusExt, panic, - process::Stdio, time::{Duration}, sync::atomic::Ordering, + process::Stdio, time::{Duration}, sync::{Arc}, }; use tracing::{trace_span, Instrument}; @@ -25,7 +25,7 @@ use windmill_common::{ flows::{FlowModuleValue, FlowValue}, scripts::{ScriptHash, ScriptLang}, utils::{rd_string}, - variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, IS_READY, METRICS_ENABLED, jobs::{JobKind, QueuedJob}, + variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, METRICS_ENABLED, jobs::{JobKind, QueuedJob}, }; use windmill_queue::{canceled_job_to_result, get_queued_job, pull, CLOUD_HOSTED}; @@ -36,7 +36,7 @@ use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, process::{Child, Command}, sync::{ - mpsc::{self, Sender}, watch, broadcast, RwLock + mpsc::{self, Sender}, watch, broadcast, RwLock, Barrier }, time::{interval, sleep, Instant, MissedTickBehavior} }; @@ -52,7 +52,7 @@ use async_recursion::async_recursion; use rand::Rng; #[cfg(feature = "enterprise")] -use crate::global_cache::{copy_cache_from_bucket_as_tar, copy_cache_to_tmp_cache, cache_global, copy_tmp_cache_to_cache}; +use crate::global_cache::{copy_cache_to_tmp_cache, cache_global, copy_tmp_cache_to_cache}; use crate::{ jobs::{add_completed_job, add_completed_job_error}, @@ -274,16 +274,17 @@ impl AuthedClient { } -#[tracing::instrument(skip(rsmq), level = "trace")] pub async fn run_worker( db: &Pool, worker_instance: &str, worker_name: String, i_worker: u64, + num_workers: u32, ip: &str, mut rx: tokio::sync::broadcast::Receiver<()>, base_internal_url: &str, rsmq: Option, + sync_barrier: RwLock>>, ) { #[cfg(not(feature = "enterprise"))] if !*DISABLE_NSJAIL { @@ -410,16 +411,6 @@ pub async fn run_worker> = None; - #[cfg(feature = "enterprise")] - if i_worker == 1 { - if let Some(ref s) = S3_CACHE_BUCKET.clone() { - // We try to download the entire cache as a tar, it is much faster over S3 - copy_cache_from_bucket_as_tar(&s).await; - } - }; - - IS_READY.store(true, Ordering::Relaxed); - tracing::info!(worker = %worker_name, "starting worker"); #[cfg(feature = "enterprise")] @@ -445,6 +436,7 @@ pub async fn run_worker 1 { + let read_barrier = sync_barrier.read().await; + let barrier = read_barrier.clone(); + if let Some(b) = barrier.as_ref() { + b.wait().await; + }; + } + let (do_break, next_job) = if first_run { (false, Ok(Some(QueuedJob::default()))) } else { @@ -509,6 +509,15 @@ pub async fn run_worker { + if num_workers > 1 { + let mut barrier = sync_barrier.write().await; + let arc_barrier = Arc::new(Some(tokio::sync::Barrier::new(num_workers as usize))); + *barrier = arc_barrier.clone(); + if let Some(b) = arc_barrier.as_ref() { + b.wait().await; + }; + } + //Arc::new(tokio::sync::Barrier::new(num_workers as usize + 1)); #[cfg(feature = "enterprise")] if let Err(e) = copy_tmp_cache_to_cache().await { tracing::error!(worker = %worker_name, "failed to sync tmp cache to cache: {}", e);