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

fix(backend): global cache synco only start if all piptars have been downloaded

Browse Source
This commit is contained in:
Ruben Fiszel
2023-04-21 22:20:57 +02:00
parent a235dfbe1f
commit a3487f18c4
2 changed files with 7 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/global_cache.rs
+4 -3
View File
@@ -161,7 +161,7 @@ pub async fn copy_cache_from_bucket(bucket: &str, tx: Sender<()>) -> error::Resu
)
.await
{
tracing::info!("Failed to to copy cache from bucket. Error: {:?}", e);
tracing::info!("Failed to copy cache from bucket. Error: {:?}", e);
return Err(e);
}
@@ -201,7 +201,7 @@ pub async fn copy_cache_to_bucket(bucket: &str) -> error::Result<()> {
)
.await
{
tracing::info!("Failed to to copy cache to bucket. Error: {:?}", e);
tracing::info!("Failed to copy cache to bucket. Error: {:?}", e);
return Err(e);
}
tracing::info!(
@@ -451,9 +451,10 @@ pub async fn extract_pip_tar(tar: &str, folder: &str) -> error::Result<()> {
use tokio::fs;
let start: Instant = Instant::now();
fs::create_dir(&folder).await?;
fs::create_dir_all(&folder).await?;
if let Err(e) = execute_command(&folder, "tar", vec!["-xpvf", tar]).await {
tracing::info!("Failed to untar piptar. Error: {:?}", e);
fs::remove_dir_all(&folder).await?;
return Err(e);
}
tracing::info!(
backend/windmill-worker/src/worker.rs
+3 -13
View File
@@ -38,7 +38,7 @@ use tokio::{
sync::{
mpsc::{self, Sender}, watch, broadcast, RwLock, Barrier
},
time::{interval, sleep, Instant, MissedTickBehavior}
time::{interval, sleep, Instant, MissedTickBehavior}, join
};
use futures::{
@@ -428,20 +428,10 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let worker_name2 = worker_name.clone();
handles.push(tokio::task::spawn(async move {
tracing::info!(worker = %worker_name2, "Started initial denogo tar sync in background");
copy_denogo_cache_from_bucket_as_tar(&bucket).await;
tracing::info!(worker = %worker_name2, "Started initial sync in background");
join!(copy_denogo_cache_from_bucket_as_tar(&bucket), copy_all_piptars_from_bucket(&bucket));
let _ = copy_to_bucket_tx2.send(()).await;
}));
let bucket = s.to_string();
let copy_to_bucket_tx = copy_to_bucket_tx.clone();
let worker_name = worker_name.clone();
handles.push(tokio::task::spawn(async move {
tracing::info!(worker = %worker_name, "Started initial piptars cache sync in background");
copy_all_piptars_from_bucket(&bucket).await;
let _ = copy_to_bucket_tx.send(()).await;
}));
}
}