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

add barrier when num workers > 1

Browse Source
This commit is contained in:
Ruben Fiszel
2023-04-19 12:46:25 +02:00
parent 36acbf793b
commit 29848ae466
6 changed files with 75 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main.rs
+17 -2
View File
@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
let mut handles = Vec::with_capacity(num_workers as usize);
#[cfg(feature = "enterprise")]
if let Some(ref s) = S3_CACHE_BUCKET.clone() {
// We donwload the entire cache as tar
windmill_worker::copy_cache_from_bucket_as_tar(&s).await;
}
IS_READY.store(true, Ordering::Relaxed);
for i in 1..(num_workers + 1) {
let db1 = db.clone();
let instance_name = instance_name.clone();
@@ -281,10 +294,12 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
&instance_name,
worker_name,
i as u64,
num_workers as u32,
&ip,
rx,
&base_internal_url,
rsmq2,
RwLock::new(Arc::new(None)),
)
.await
})));
backend/tests/worker.rs
+6 -2
View File
@@ -1,7 +1,10 @@
use std::sync::Arc;
use futures::{stream, Stream};
use serde::Deserialize;
use serde_json::json;
use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres, Transaction};
use tokio::sync::RwLock;
use windmill_api::jobs::{CompletedJob, Job};
use windmill_common::{
flow_status::{FlowStatus, FlowStatusModule},
@@ -911,7 +914,6 @@ fn spawn_test_worker(
let db = db.to_owned();
let worker_instance: &str = "test worker instance";
let worker_name: String = next_worker_name();
let i_worker: u64 = Default::default();
let ip: &str = Default::default();
let future = async move {
let base_internal_url = format!("http://localhost:{}", port);
@@ -919,11 +921,13 @@ fn spawn_test_worker(
&db,
worker_instance,
worker_name,
i_worker,
1,
1,
ip,
rx,
&base_internal_url,
None,
RwLock::new(Arc::new(None)),
)
.await
};
backend/windmill-common/src/external_ip.rs
+7 -1
View File
@@ -11,10 +11,16 @@
//! For users writing scripts to access their infrastructure with firewalls requiring incoming
//! connections to be from whitelisted IP addresses.
use std::time::Duration;
use reqwest::Result;
pub async fn get_ip() -> Result<String> {
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()
backend/windmill-worker/src/global_cache.rs
+19 -12
View File
@@ -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,
backend/windmill-worker/src/lib.rs
+1
View File
@@ -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::*;
backend/windmill-worker/src/worker.rs
+25 -16
View File
@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
db: &Pool<Postgres>,
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<R>,
sync_barrier: RwLock<Arc<Option<Barrier>>>,
) {
#[cfg(not(feature = "enterprise"))]
if !*DISABLE_NSJAIL {
@@ -410,16 +411,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let mut copy_cache_from_bucket_handle: Option<tokio::task::JoinHandle<()>> = 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let mut first_run = true;
// let mut barrier = Arc::new();
loop {
if *METRICS_ENABLED {
worker_busy.set(0);
@@ -492,7 +484,15 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
}
#[cfg(feature = "enterprise")]
if num_workers > 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
(true, Ok(None))
},
_ = copy_to_bucket_rx.recv() => {
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);