From 65cf697cb0e077287ba587af867a9b0d91fb4b7d Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 20 Jun 2022 21:28:28 -0400 Subject: [PATCH] Attempt to enrich "filesystem full" errors for later handling --- .../storyteller_web/src/tts_inference_job.rs | 140 +++++++++++++----- 1 file changed, 102 insertions(+), 38 deletions(-) diff --git a/crates/service/storyteller_web/src/tts_inference_job.rs b/crates/service/storyteller_web/src/tts_inference_job.rs index 52271108c3..ee261f1ee2 100644 --- a/crates/service/storyteller_web/src/tts_inference_job.rs +++ b/crates/service/storyteller_web/src/tts_inference_job.rs @@ -64,7 +64,7 @@ use sqlx::MySqlPool; use sqlx::mysql::MySqlPoolOptions; use std::collections::HashMap; use std::fs::{File, metadata}; -use std::io::{BufReader, Read}; +use std::io::{BufReader, Read, ErrorKind}; use std::ops::Deref; use std::path::{PathBuf, Path}; use std::process::Command; @@ -500,7 +500,8 @@ async fn process_jobs( }; if !model_state.is_downloaded_to_filesystem || !model_state.is_in_memory_cache { - warn!("Model isn't ready (downloaded = {}), (in memory = {})", + warn!("Model isn't ready: {} (downloaded = {}), (in memory = {})", + &model_state.model_record.model_token, model_state.is_downloaded_to_filesystem, model_state.is_in_memory_cache); @@ -645,11 +646,44 @@ impl ModelState { } } +/// Error from processing a single job +#[derive(Debug)] +enum ProcessJobError { + /// The filesystem is out of space and we need to free it up. + FilesystemFull, + /// This is any other kind of error. + /// It might be important, we just haven't special cased it yet. + Other(anyhow::Error), +} + +impl ProcessJobError { + fn from_io_error(error: std::io::Error) -> Self { + match error.raw_os_error() { + // NB: We can't use err.kind() -> ErrorKind::StorageFull, because it's marked unstable: + // `io_error_more` is unstable [E0658] + Some(28) => ProcessJobError::FilesystemFull, + _ => ProcessJobError::Other(anyhow!(error)), + } + } + + fn from_anyhow_error(error: anyhow::Error) -> Self { + match error.downcast_ref::() { + Some(e) => match e.raw_os_error() { + // NB: We can't use err.kind() -> ErrorKind::StorageFull, because it's marked unstable: + // `io_error_more` is unstable [E0658] + Some(28) => ProcessJobError::FilesystemFull, + _ => ProcessJobError::Other(anyhow!(error)), + }, + None => ProcessJobError::Other(error), + } + } +} + async fn process_job( inferencer: &Inferencer, job: &AvailableTtsInferenceJob, model_record: &TtsModelForInferenceRecord, -) -> AnyhowResult<(Span, Span)> { +) -> Result<(Span, Span), ProcessJobError> { // TODO 1. Mark processing (DONE) @@ -686,7 +720,9 @@ async fn process_job( .attribute("user_token", maybe_user_token) .service_name("tts-inference-job"); - let mut redis = inferencer.redis_pool.get()?; + let mut redis = inferencer.redis_pool.get() + .map_err(|e| ProcessJobError::Other(anyhow!(e)))?; + let mut redis_logger = RedisJobStatusLogger::new_tts_inference( &mut redis, &job.inference_job_token); @@ -696,7 +732,9 @@ async fn process_job( info!("Attempting to grab lock for job: {}", job.inference_job_token); let lock_acquired = - mark_tts_inference_job_pending_and_grab_lock(&inferencer.mysql_pool, job.id).await?; + mark_tts_inference_job_pending_and_grab_lock(&inferencer.mysql_pool, job.id) + .await + .map_err(|e| ProcessJobError::Other(e))?; if !lock_acquired { warn!("Could not acquire job lock for: {:?}", &job.id); @@ -721,7 +759,8 @@ async fn process_job( if !waveglow_vocoder_model_fs_path.exists() { warn!("Waveglow vocoder model file does not exist: {:?}", &waveglow_vocoder_model_fs_path); - redis_logger.log_status("downloading vocoder (1 of 3)")?; + redis_logger.log_status("downloading vocoder (1 of 3)") + .map_err(|e| ProcessJobError::Other(e))?; let waveglow_vocoder_model_object_path = inferencer.bucket_path_unifier .tts_pretrained_vocoders_path(&waveglow_vocoder_model_filename); @@ -730,8 +769,9 @@ async fn process_job( inferencer.private_bucket_client.download_file_to_disk( &waveglow_vocoder_model_object_path, - &waveglow_vocoder_model_fs_path - ).await?; + &waveglow_vocoder_model_fs_path) + .await + .map_err(|e| ProcessJobError::Other(e))?; info!("Downloaded waveglow vocoder model from bucket!"); } @@ -744,7 +784,8 @@ async fn process_job( if !hifigan_vocoder_model_fs_path.exists() { warn!("Hifigan vocoder model file does not exist: {:?}", &hifigan_vocoder_model_fs_path); - redis_logger.log_status("downloading vocoder (2 of 3)")?; + redis_logger.log_status("downloading vocoder (2 of 3)") + .map_err(|e| ProcessJobError::Other(e))?; let hifigan_vocoder_model_object_path = inferencer.bucket_path_unifier .tts_pretrained_vocoders_path(&hifigan_vocoder_model_filename); @@ -753,8 +794,9 @@ async fn process_job( inferencer.private_bucket_client.download_file_to_disk( &hifigan_vocoder_model_object_path, - &hifigan_vocoder_model_fs_path - ).await?; + &hifigan_vocoder_model_fs_path) + .await + .map_err(|e| ProcessJobError::Other(e))?; info!("Downloaded hifigan vocoder model from bucket!"); } @@ -767,7 +809,8 @@ async fn process_job( if !hifigan_superres_vocoder_model_fs_path.exists() { warn!("Hifigan superres vocoder model file does not exist: {:?}", &hifigan_superres_vocoder_model_fs_path); - redis_logger.log_status("downloading vocoder (3 of 3)")?; + redis_logger.log_status("downloading vocoder (3 of 3)") + .map_err(|e| ProcessJobError::Other(e))?; let hifigan_superres_vocoder_model_object_path = inferencer.bucket_path_unifier .tts_pretrained_vocoders_path(&hifigan_superres_vocoder_model_filename); @@ -776,8 +819,9 @@ async fn process_job( inferencer.private_bucket_client.download_file_to_disk( &hifigan_superres_vocoder_model_object_path, - &hifigan_superres_vocoder_model_fs_path - ).await?; + &hifigan_superres_vocoder_model_fs_path) + .await + .map_err(|e| ProcessJobError::Other(e))?; info!("Downloaded hifigan superres vocoder model from bucket!"); } @@ -809,7 +853,8 @@ async fn process_job( if !tts_synthesizer_fs_path.exists() { info!("TTS synthesizer model file does not exist: {:?}", &tts_synthesizer_fs_path); - redis_logger.log_status("downloading synthesizer")?; + redis_logger.log_status("downloading synthesizer") + .map_err(|e| ProcessJobError::Other(e))?; let tts_synthesizer_object_path = inferencer.bucket_path_unifier .tts_synthesizer_path(&model_record.private_bucket_hash); @@ -818,8 +863,9 @@ async fn process_job( inferencer.private_bucket_client.download_file_to_disk( &tts_synthesizer_object_path, - &tts_synthesizer_fs_path - ).await?; + &tts_synthesizer_fs_path) + .await + .map_err(|e| ProcessJobError::from_anyhow_error(e))?; info!("Downloaded template media from bucket!"); } @@ -830,21 +876,29 @@ async fn process_job( // ==================== WRITE TEXT TO FILE ==================== // + info!("Creating tempdir for inference results."); + let temp_dir = format!("temp_tts_inference_{}", job.id.0); - let temp_dir = TempDir::new(&temp_dir)?; // NB: Exists until it goes out of scope. + + // NB: TempDir exists until it goes out of scope, at which point it should delete from filesystem. + let temp_dir = TempDir::new(&temp_dir) + .map_err(|e| ProcessJobError::from_io_error(e))?; let text_input_fs_path = temp_dir.path().join("inference_input.txt"); - std::fs::write(&text_input_fs_path, &cleaned_inference_text)?; + std::fs::write(&text_input_fs_path, &cleaned_inference_text) + .map_err(|e| ProcessJobError::from_io_error(e))?; // ==================== RUN INFERENCE ==================== // - redis_logger.log_status("running inference")?; + redis_logger.log_status("running inference") + .map_err(|e| ProcessJobError::Other(e))?; // TODO: Fix this. let maybe_unload_model_path = inferencer .virtual_model_lfu - .insert_returning_replaced(tts_synthesizer_fs_path.to_str().unwrap_or(""))?; + .insert_returning_replaced(tts_synthesizer_fs_path.to_str().unwrap_or("")) + .map_err(|e| ProcessJobError::Other(e))?; if let Some(model_path) = maybe_unload_model_path.as_deref() { warn!("Remove model from LFU cache: {:?}", model_path); @@ -876,7 +930,8 @@ async fn process_job( let mut pretrained_vocoder = VocoderType::HifiGanSuperResolution; if let Some(default_vocoder) = model_record.maybe_default_pretrained_vocoder.as_deref() { - pretrained_vocoder = VocoderType::from_str(default_vocoder)?; + pretrained_vocoder = VocoderType::from_str(default_vocoder) + .map_err(|e| ProcessJobError::Other(e))?; } info!("With pretrained vocoder: {:?}", pretrained_vocoder); @@ -891,24 +946,27 @@ async fn process_job( &output_audio_fs_path, &output_spectrogram_fs_path, &output_metadata_fs_path, - maybe_unload_model_path, - ).await?; + maybe_unload_model_path) + .await + .map_err(|e| ProcessJobError::Other(e))?; // ==================== CHECK ALL FILES EXIST AND GET METADATA ==================== // info!("Checking that output files exist..."); - check_file_exists(&output_audio_fs_path)?; - check_file_exists(&output_spectrogram_fs_path)?; - check_file_exists(&output_metadata_fs_path)?; + check_file_exists(&output_audio_fs_path).map_err(|e| ProcessJobError::Other(e))?; + check_file_exists(&output_spectrogram_fs_path).map_err(|e| ProcessJobError::Other(e))?; + check_file_exists(&output_metadata_fs_path).map_err(|e| ProcessJobError::Other(e))?; - let file_metadata = read_metadata_file(&output_metadata_fs_path)?; + let file_metadata = read_metadata_file(&output_metadata_fs_path) + .map_err(|e| ProcessJobError::Other(e))?; safe_delete_temp_file(&output_metadata_fs_path); // ==================== UPLOAD AUDIO TO BUCKET ==================== // - redis_logger.log_status("uploading result")?; + redis_logger.log_status("uploading result") + .map_err(|e| ProcessJobError::Other(e))?; let audio_result_object_path = inferencer.bucket_path_unifier.tts_inference_wav_audio_output_path( &job.uuid_idempotency_token); // TODO: Don't use this! @@ -921,7 +979,8 @@ async fn process_job( &audio_result_object_path, &output_audio_fs_path, "audio/wav") - .await?; + .await + .map_err(|e| ProcessJobError::Other(e))?; safe_delete_temp_file(&output_audio_fs_path); @@ -938,7 +997,8 @@ async fn process_job( &spectrogram_result_object_path, &output_spectrogram_fs_path, "application/json") - .await?; + .await + .map_err(|e| ProcessJobError::Other(e))?; safe_delete_temp_file(&output_spectrogram_fs_path); @@ -949,7 +1009,8 @@ async fn process_job( // ==================== SAVE RECORDS ==================== // - let text_hash = hash_string_sha2(&cleaned_inference_text)?; + let text_hash = hash_string_sha2(&cleaned_inference_text) + .map_err(|e| ProcessJobError::Other(e))?; info!("Saving tts inference record..."); let (id, inference_result_token) = insert_tts_result( @@ -962,16 +1023,18 @@ async fn process_job( file_metadata.file_size_bytes, file_metadata.duration_millis.unwrap_or(0), inferencer.worker_details.is_on_prem, - &inferencer.worker_details.worker_hostname, - ).await?; + &inferencer.worker_details.worker_hostname) + .await + .map_err(|e| ProcessJobError::Other(e))?; info!("Marking job complete..."); mark_tts_inference_job_done( &inferencer.mysql_pool, job.id, true, - Some(&inference_result_token) - ).await?; + Some(&inference_result_token)) + .await + .map_err(|e| ProcessJobError::Other(e))?; info!("TTS Done. Original text was: {}", &job.raw_inference_text); @@ -982,10 +1045,11 @@ async fn process_job( .await .map_err(|e| { error!("error publishing event: {:?}", e); - anyhow!("error publishing event") + ProcessJobError::Other(anyhow!("error publishing event")) })?; - redis_logger.log_status("done")?; + redis_logger.log_status("done") + .map_err(|e| ProcessJobError::Other(e))?; info!("Job {:?} complete success! Downloaded, ran inference, and uploaded. Saved model record: {}, Result Token: {}", job.id, id, &inference_result_token);