From 1a3e111dd6967933452f1b1c6edf369e490d02b7 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 29 Nov 2023 17:30:03 -0500 Subject: [PATCH 1/2] Fix job pathing for outage and to make easier to manage --- .../src/filesystem/check_directory_exists.rs | 14 ++-- .../src/job/job_loop/process_single_job.rs | 19 ++--- .../lipsync/sad_talker/process_job.rs | 2 +- .../download_static_dependencies.rs | 6 +- .../tacotron2_v2_early_fakeyou/process_job.rs | 4 +- .../tts/vall_e_x/process_job_create_voice.rs | 2 +- .../vall_e_x/process_job_inference_voice.rs | 2 +- .../src/job/job_types/tts/vits/process_job.rs | 2 +- .../job/job_types/vc/rvc_v2/process_job.rs | 8 +-- .../job_types/vc/so_vits_svc/process_job.rs | 2 +- .../job/inference_job/src/job_dependencies.rs | 24 ++++--- crates/service/job/inference_job/src/main.rs | 71 ++++++++++++------- .../src/util/scoped_temp_dir_creator.rs | 6 +- 13 files changed, 97 insertions(+), 65 deletions(-) diff --git a/crates/lib/deprecated/container_common/src/filesystem/check_directory_exists.rs b/crates/lib/deprecated/container_common/src/filesystem/check_directory_exists.rs index 43487de769..8e770b171f 100644 --- a/crates/lib/deprecated/container_common/src/filesystem/check_directory_exists.rs +++ b/crates/lib/deprecated/container_common/src/filesystem/check_directory_exists.rs @@ -1,15 +1,17 @@ -use std::path::PathBuf; +use std::path::Path; use anyhow::bail; use crate::anyhow_result::AnyhowResult; -pub fn check_directory_exists(path: &PathBuf) -> AnyhowResult<()> { - if !path.exists() { - bail!("Path doesn't exist: {:?}", path); +pub fn check_directory_exists>(path: P) -> AnyhowResult<()> { + let check_path = path.as_ref(); + + if !check_path.exists() { + bail!("Path doesn't exist: {:?}", check_path); } - if !path.is_dir() { - bail!("Path isn't a directory: {:?}", path); + if !check_path.is_dir() { + bail!("Path isn't a directory: {:?}", check_path); } Ok(()) } diff --git a/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs b/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs index 401fdd76dd..4309db6e60 100644 --- a/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs +++ b/crates/service/job/inference_job/src/job/job_loop/process_single_job.rs @@ -160,18 +160,19 @@ async fn do_process_single_job( } } - // ==================== SETUP TEMP DIRS ==================== // - - let temp_dir = format!("temp_{}", job.id.0); - let temp_dir = job_dependencies.fs.scoped_temp_dir_creator_for_downloads.new_tempdir(&temp_dir) - .map_err(|err| ProcessSingleJobError::Other(anyhow!("filesystem error: {:?}", err)))?; - - let _p = temp_dir.path(); // TODO: Just so the build doesn't complain about unused. Remove. +// NB(bt,2023-11-29): This looks dead, but hopefully isn't leaned on for downstream side effects. +// I'll leave this commented out while dealing with the outage, but this can probably be removed +// in a short while after we redeploy and verify all the jobs. +// // ==================== SETUP TEMP DIRS ==================== // +// +// let temp_dir = format!("temp_{}", job.id.0); +// let temp_dir = job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads.new_tempdir(&temp_dir) +// .map_err(|err| ProcessSingleJobError::Other(anyhow!("filesystem error: {:?}", err)))?; +// +// let _p = temp_dir.path(); // TODO: Just so the build doesn't complain about unused. Remove. // ==================== HANDLE DIFFERENT INFERENCE TYPES ==================== // - let mut maybe_result_entity : Option; - let job_success_result = match job.inference_category { InferenceCategory::LipsyncAnimation => { process_single_lipsync_job(job_dependencies, job).await? diff --git a/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs b/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs index 4c424abeec..009fb63b28 100644 --- a/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs @@ -74,7 +74,7 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result( &mut job_progress_reporter, "downloading vocoder (1 of 3)", job.id.0, - &job_dependencies.fs.scoped_temp_dir_creator_for_downloads, + &job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads, ).await?; waveglow_vocoder_model_fs_path @@ -61,7 +61,7 @@ pub async fn download_static_dependencies<'a>( &mut job_progress_reporter, "downloading vocoder (2 of 3)", job.id.0, - &job_dependencies.fs.scoped_temp_dir_creator_for_downloads, + &job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads, ).await?; hifigan_vocoder_model_fs_path @@ -83,7 +83,7 @@ pub async fn download_static_dependencies<'a>( &mut job_progress_reporter, "downloading vocoder (3 of 3)", job.id.0, - &job_dependencies.fs.scoped_temp_dir_creator_for_downloads, + &job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads, ).await?; hifigan_superres_vocoder_model_fs_path diff --git a/crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/process_job.rs b/crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/process_job.rs index 8428d099c1..177791f92c 100644 --- a/crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/process_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/process_job.rs @@ -127,7 +127,7 @@ async fn process_job_with_cleanup( &mut job_progress_reporter, "downloading user vocoder", job.id.0, - &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads, + &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads, ).await?; Some(custom_vocoder_fs_path) @@ -148,7 +148,7 @@ async fn process_job_with_cleanup( &mut job_progress_reporter, "downloading synthesizer", job.id.0, - &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads, + &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads, ).await?; tts_synthesizer_fs_path diff --git a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_create_voice.rs b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_create_voice.rs index dbc9980393..e91582ac01 100644 --- a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_create_voice.rs +++ b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_create_voice.rs @@ -177,7 +177,7 @@ pub async fn process_create_voice( for downloader in model_dependencies.downloaders.all_downloaders() { let result = downloader.download_if_not_on_filesystem( &args.job_dependencies.buckets.private_bucket_client, - &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads + &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads ).await; if let Err(e) = result { error!("could not download: {:?}", e); diff --git a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_inference_voice.rs b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_inference_voice.rs index cfbbb2852d..a9aabac937 100644 --- a/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_inference_voice.rs +++ b/crates/service/job/inference_job/src/job/job_types/tts/vall_e_x/process_job_inference_voice.rs @@ -82,7 +82,7 @@ pub async fn process_inference_voice( for downloader in model_dependencies.downloaders.all_downloaders() { let result = downloader.download_if_not_on_filesystem( &args.job_dependencies.buckets.private_bucket_client, - &args.job_dependencies.fs.scoped_temp_dir_creator_for_downloads + &args.job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads ).await; if let Err(e) = result { diff --git a/crates/service/job/inference_job/src/job/job_types/tts/vits/process_job.rs b/crates/service/job/inference_job/src/job/job_types/tts/vits/process_job.rs index efad6075e4..d42cff9da9 100644 --- a/crates/service/job/inference_job/src/job/job_types/tts/vits/process_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/tts/vits/process_job.rs @@ -78,7 +78,7 @@ pub async fn process_job(args: VitsProcessJobArgs<'_>) -> Result) -> Result) -> Result) -> Result) -> Result) -> Result, - // TODO: Rename + /// Directory to store long-term downloads (models) + /// (In prod, typically model files from GCS / NFS PVC mount) + pub directory_long_lived_downloads: PathBuf, + + /// Temporary directory for storing downloads /// Creates temp directories for scratch files - pub scoped_temp_dir_creator_for_downloads: ScopedTempDirCreator, + pub scoped_temp_dir_creator_for_short_lived_downloads: ScopedTempDirCreator, + + /// Temporary directory for storing downloads + /// (In prod, typically model files from GCS / NFS PVC mount) + pub scoped_temp_dir_creator_for_long_lived_downloads: ScopedTempDirCreator, + + /// Temporary directory for storing work + /// (In prod, typically python inference outputs / emptyDir mount) pub scoped_temp_dir_creator_for_work: ScopedTempDirCreator, - // TODO: Rename /// Organizes downloaded files pub semi_persistent_cache: SemiPersistentCacheDir, } diff --git a/crates/service/job/inference_job/src/main.rs b/crates/service/job/inference_job/src/main.rs index d6b7e66565..d7dc736356 100644 --- a/crates/service/job/inference_job/src/main.rs +++ b/crates/service/job/inference_job/src/main.rs @@ -124,31 +124,16 @@ async fn main() -> AnyhowResult<()> { PathBuf::from("/tmp/downloads") ); - // Where we store scratch files for workloads (temporary processing). - let temp_directory_work = easyenv::get_env_pathbuf_or_default( - "TEMP_DIR_WORK", - PathBuf::from("/tmp/work") - ); + // //// Where we store scratch files for workloads (temporary processing). + // let temp_directory_work = easyenv::get_env_pathbuf_or_default( + // "TEMP_DIR_WORK", + // PathBuf::from("/tmp/work") + // ); - if !container_environment.server_environment.is_deployed_in_production() { - warn!("Creating temporary directories for non-production / development only!"); - create_dir_all_if_missing(&temp_directory_downloads)?; - create_dir_all_if_missing(&temp_directory_work)?; - } - - check_directory_exists(&temp_directory_downloads)?; - check_directory_exists(&temp_directory_work)?; let semi_persistent_cache = SemiPersistentCacheDir::configured_root(&temp_directory_downloads); - // TODO(bt,2023.05.22): create_all_paths() for some subset of jobs. - info!("Creating pod semi-persistent cache dirs..."); - semi_persistent_cache.create_custom_vocoder_model_path()?; - semi_persistent_cache.create_tts_pretrained_vocoder_model_path()?; - semi_persistent_cache.create_tts_synthesizer_model_path()?; - semi_persistent_cache.create_voice_conversion_model_path()?; - let db_connection_string = easyenv::get_env_string_or_default( "MYSQL_URL", @@ -242,11 +227,18 @@ async fn main() -> AnyhowResult<()> { maybe_keepalive_redis_pool, }, fs: FileSystemDetails { - temp_directory_downloads: temp_directory_downloads.clone(), - temp_directory_work: temp_directory_work.clone(), maybe_pause_file: easyenv::get_env_pathbuf_optional("PAUSE_FILE"), - scoped_temp_dir_creator_for_downloads: ScopedTempDirCreator::for_directory(&temp_directory_downloads), - scoped_temp_dir_creator_for_work: ScopedTempDirCreator::for_directory(&temp_directory_work), + directory_long_lived_downloads: easyenv::get_env_pathbuf_or_default( + "DIR_LONG_LIVED_DOWNLOADS", PathBuf::from("/tmp/downloads_long_lived")), + scoped_temp_dir_creator_for_short_lived_downloads: ScopedTempDirCreator::for_directory( + easyenv::get_env_pathbuf_or_default( + "SCOPED_TEMP_DIR_SHORT_LIVED_DOWNLOADS", PathBuf::from("/tmp/downloads_long_lived"))), + scoped_temp_dir_creator_for_long_lived_downloads: ScopedTempDirCreator::for_directory( + easyenv::get_env_pathbuf_or_default( + "SCOPED_TEMP_DIR_LONG_LIVED_DOWNLOADS", PathBuf::from("/tmp/downloads_long_lived"))), + scoped_temp_dir_creator_for_work: ScopedTempDirCreator::for_directory( + easyenv::get_env_pathbuf_or_default( + "SCOPED_TEMP_DIR_WORK", PathBuf::from("/tmp/downloads_long_lived"))), semi_persistent_cache, }, buckets: BucketDependencies { @@ -306,6 +298,8 @@ async fn main() -> AnyhowResult<()> { }, }; + set_up_directories(&job_dependencies)?; + std::thread::spawn(move || { let actix_runtime = actix_web::rt::System::new(); let http_server_handle = launch_http_server(create_server_args); @@ -322,3 +316,32 @@ async fn main() -> AnyhowResult<()> { Ok(()) } +fn set_up_directories(job_dependencies: &JobDependencies) -> AnyhowResult<()> { + info!("Setting up and checking file system paths for temporary work, ephemeral and persistent downloads, etc..."); + + let fs = &job_dependencies.fs; + + if !job_dependencies.job.info.container.server_environment.is_deployed_in_production() { + warn!("Creating directories for non-production / development only!"); + + create_dir_all_if_missing(&fs.directory_long_lived_downloads)?; + create_dir_all_if_missing(&fs.scoped_temp_dir_creator_for_long_lived_downloads.get_base_dir())?; + create_dir_all_if_missing(&fs.scoped_temp_dir_creator_for_short_lived_downloads.get_base_dir())?; + create_dir_all_if_missing(&fs.scoped_temp_dir_creator_for_work.get_base_dir())?; + } + + check_directory_exists(&fs.directory_long_lived_downloads)?; + check_directory_exists(fs.scoped_temp_dir_creator_for_long_lived_downloads.get_base_dir())?; + check_directory_exists(fs.scoped_temp_dir_creator_for_short_lived_downloads.get_base_dir())?; + check_directory_exists(fs.scoped_temp_dir_creator_for_work.get_base_dir())?; + + // TODO(bt,2023.05.22): create_all_paths() for some subset of jobs. + info!("Creating semi-persistent cache dirs..."); + + fs.semi_persistent_cache.create_custom_vocoder_model_path()?; + fs.semi_persistent_cache.create_tts_pretrained_vocoder_model_path()?; + fs.semi_persistent_cache.create_tts_synthesizer_model_path()?; + fs.semi_persistent_cache.create_voice_conversion_model_path()?; + + Ok(()) +} diff --git a/crates/service/job/inference_job/src/util/scoped_temp_dir_creator.rs b/crates/service/job/inference_job/src/util/scoped_temp_dir_creator.rs index 0c5dec1a03..e6e0dc4077 100644 --- a/crates/service/job/inference_job/src/util/scoped_temp_dir_creator.rs +++ b/crates/service/job/inference_job/src/util/scoped_temp_dir_creator.rs @@ -15,7 +15,11 @@ impl ScopedTempDirCreator { } } + pub fn get_base_dir(&self) -> &Path { + &self.base_dir + } + pub fn new_tempdir(&self, name: &str) -> std::io::Result { TempDir::new_in(&self.base_dir, name) } -} \ No newline at end of file +} From 37c016cecbd8a990b8c165d61ddac0a43b673191 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 29 Nov 2023 17:48:29 -0500 Subject: [PATCH 2/2] "semi persistent dir" needs cleanup, this is confusing --- .../download_static_dependencies.rs | 6 +++--- crates/service/job/inference_job/src/main.rs | 16 +++------------- .../src/util/maybe_download_file_from_bucket.rs | 12 ++++++------ 3 files changed, 12 insertions(+), 22 deletions(-) diff --git a/crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/download_static_dependencies.rs b/crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/download_static_dependencies.rs index 10a529a897..f0b3d2a0cd 100644 --- a/crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/download_static_dependencies.rs +++ b/crates/service/job/inference_job/src/job/job_types/tts/tacotron2_v2_early_fakeyou/download_static_dependencies.rs @@ -39,7 +39,7 @@ pub async fn download_static_dependencies<'a>( &mut job_progress_reporter, "downloading vocoder (1 of 3)", job.id.0, - &job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads, + &job_dependencies.fs.scoped_temp_dir_creator_for_long_lived_downloads, ).await?; waveglow_vocoder_model_fs_path @@ -61,7 +61,7 @@ pub async fn download_static_dependencies<'a>( &mut job_progress_reporter, "downloading vocoder (2 of 3)", job.id.0, - &job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads, + &job_dependencies.fs.scoped_temp_dir_creator_for_long_lived_downloads, ).await?; hifigan_vocoder_model_fs_path @@ -83,7 +83,7 @@ pub async fn download_static_dependencies<'a>( &mut job_progress_reporter, "downloading vocoder (3 of 3)", job.id.0, - &job_dependencies.fs.scoped_temp_dir_creator_for_short_lived_downloads, + &job_dependencies.fs.scoped_temp_dir_creator_for_long_lived_downloads, ).await?; hifigan_superres_vocoder_model_fs_path diff --git a/crates/service/job/inference_job/src/main.rs b/crates/service/job/inference_job/src/main.rs index d7dc736356..3cd39e5e90 100644 --- a/crates/service/job/inference_job/src/main.rs +++ b/crates/service/job/inference_job/src/main.rs @@ -119,20 +119,10 @@ async fn main() -> AnyhowResult<()> { )?; // Where we download models and resources to (typically a shared NFS volume in prod). - let temp_directory_downloads = easyenv::get_env_pathbuf_or_default( - "TEMP_DIR_DOWNLOADS", - PathBuf::from("/tmp/downloads") - ); - // //// Where we store scratch files for workloads (temporary processing). - // let temp_directory_work = easyenv::get_env_pathbuf_or_default( - // "TEMP_DIR_WORK", - // PathBuf::from("/tmp/work") - // ); - - - let semi_persistent_cache = - SemiPersistentCacheDir::configured_root(&temp_directory_downloads); + let semi_persistent_cache = SemiPersistentCacheDir::configured_root( + easyenv::get_env_pathbuf_or_default("SEMI_PERSISTENT_DIR_ROOT", + PathBuf::from("/tmp/persistent"))); let db_connection_string = easyenv::get_env_string_or_default( diff --git a/crates/service/job/inference_job/src/util/maybe_download_file_from_bucket.rs b/crates/service/job/inference_job/src/util/maybe_download_file_from_bucket.rs index bd294c51cb..98a49512ca 100644 --- a/crates/service/job/inference_job/src/util/maybe_download_file_from_bucket.rs +++ b/crates/service/job/inference_job/src/util/maybe_download_file_from_bucket.rs @@ -13,7 +13,7 @@ use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator; pub async fn maybe_download_file_from_bucket( name_or_description_of_file: &str, - file_path: &Path, + final_filesystem_file_path: &Path, bucket_object_path: &Path, bucket_client: &BucketClient, job_progress_reporter: &mut Box, @@ -22,12 +22,12 @@ pub async fn maybe_download_file_from_bucket( scoped_tempdir_creator: &ScopedTempDirCreator, ) -> Result<(), ProcessSingleJobError> { - if file_path.exists() { + if final_filesystem_file_path.exists() { // TODO(bt, 2022-07-15): Check signature of file return Ok(()) } - warn!("{} does not exist at path: {:?}", name_or_description_of_file, &file_path); + warn!("{} does not exist at path: {:?}", name_or_description_of_file, &final_filesystem_file_path); job_progress_reporter.log_status(job_progress_update_description) .map_err(|e| ProcessSingleJobError::Other(e))?; @@ -54,15 +54,15 @@ pub async fn maybe_download_file_from_bucket( info!("Downloaded {} from bucket!", name_or_description_of_file); info!("Renaming {} temp file from {:?} to {:?}!", - name_or_description_of_file, &temp_path, &file_path); + name_or_description_of_file, &temp_path, &final_filesystem_file_path); - std::fs::rename(&temp_path, &file_path) + std::fs::rename(&temp_path, &final_filesystem_file_path) .map_err(|e| { safe_delete_temp_directory(&temp_dir); ProcessSingleJobError::from_io_error(e) })?; - info!("Finished downloading {} file to {:?}", name_or_description_of_file, &file_path); + info!("Finished downloading {} file to {:?}", name_or_description_of_file, &final_filesystem_file_path); safe_delete_temp_directory(&temp_dir);