diff --git a/crates/service/tts_inference_job/src/job_steps/job_args.rs b/crates/service/tts_inference_job/src/job_steps/job_args.rs index 92be25a5cc..38f8276748 100644 --- a/crates/service/tts_inference_job/src/job_steps/job_args.rs +++ b/crates/service/tts_inference_job/src/job_steps/job_args.rs @@ -2,6 +2,7 @@ use crate::caching::cache_miss_strategizer_multi::SyncMultiCacheMissStrategizer; use crate::caching::virtual_lfu_cache::SyncVirtualLfuCache; use crate::http_clients::tts_inference_sidecar_client::TtsInferenceSidecarClient; use crate::http_clients::tts_sidecar_health_check_client::TtsSidecarHealthCheckClient; +use crate::job_steps::job_stats::JobStats; use crate::script_execution::tacotron_inference_command::TacotronInferenceCommand; use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator; use database_queries::mediators::firehose_publisher::FirehosePublisher; @@ -33,6 +34,8 @@ pub struct JobArgs { pub http_clients: JobHttpClients, + pub job_stats: JobStats, + pub tts_inference_command: TacotronInferenceCommand, pub newrelic_client: NewRelicClient, diff --git a/crates/service/tts_inference_job/src/job_steps/job_stats.rs b/crates/service/tts_inference_job/src/job_steps/job_stats.rs new file mode 100644 index 0000000000..fe006d0b60 --- /dev/null +++ b/crates/service/tts_inference_job/src/job_steps/job_stats.rs @@ -0,0 +1,65 @@ +use std::sync::{Arc, LockResult, RwLock}; +use log::error; +use container_common::anyhow_result::AnyhowResult; + +/// Job stats uses interior mutability to be easy to copy around. +#[derive(Clone)] +pub struct JobStats { + inner: Arc>, +} + +/// Private inner implementation that may grow additional stats in the future. +#[derive(Default)] +struct JobStatsInner { + pub total_success_count: u64, + pub total_failure_count: u64, + pub consecutive_success_count: u64, + pub consecutive_failure_count: u64, +} + +/// Public result type. +#[derive(Default, Debug, Clone)] +pub struct SuccessAndFailureStats { + pub total_success_count: u64, + pub total_failure_count: u64, + pub consecutive_success_count: u64, + pub consecutive_failure_count: u64, +} + +impl JobStats { + pub fn new() -> Self { + Self { + inner: Arc::new(RwLock::new(JobStatsInner::default())), + } + } + + pub fn increment_failure_count(&self) -> AnyhowResult { + let mut lock = self.inner.write()?; + + lock.total_failure_count = lock.total_failure_count.saturating_add(1); + lock.consecutive_success_count = 0; + lock.consecutive_failure_count = lock.consecutive_failure_count.saturating_add(1); + + Ok(SuccessAndFailureStats { + total_success_count: lock.total_success_count, + total_failure_count: lock.total_failure_count, + consecutive_success_count: lock.consecutive_success_count, + consecutive_failure_count: lock.consecutive_failure_count, + }) + } + + pub fn increment_success_count(&self) -> AnyhowResult { + let mut lock = self.inner.write()?; + + lock.total_success_count = lock.total_success_count.saturating_add(1); + lock.consecutive_success_count = lock.consecutive_success_count.saturating_add(1); + lock.consecutive_failure_count = 0; + + Ok(SuccessAndFailureStats { + total_success_count: lock.total_success_count, + total_failure_count: lock.total_failure_count, + consecutive_success_count: lock.consecutive_success_count, + consecutive_failure_count: lock.consecutive_failure_count, + }) + } +} diff --git a/crates/service/tts_inference_job/src/job_steps/mod.rs b/crates/service/tts_inference_job/src/job_steps/mod.rs index 8c38dcce50..641c31aa42 100644 --- a/crates/service/tts_inference_job/src/job_steps/mod.rs +++ b/crates/service/tts_inference_job/src/job_steps/mod.rs @@ -1,5 +1,6 @@ pub mod download_file_from_bucket; pub mod health_check_trap; pub mod job_args; +pub mod job_stats; pub mod process_single_job; pub mod process_single_job_error; \ No newline at end of file diff --git a/crates/service/tts_inference_job/src/job_steps/process_single_job.rs b/crates/service/tts_inference_job/src/job_steps/process_single_job.rs index fb1c11d9d6..f2673e0b46 100644 --- a/crates/service/tts_inference_job/src/job_steps/process_single_job.rs +++ b/crates/service/tts_inference_job/src/job_steps/process_single_job.rs @@ -45,7 +45,7 @@ fn read_metadata_file(filename: &PathBuf) -> AnyhowResult { } pub async fn process_single_job( - inferencer: &JobArgs, + job_args: &JobArgs, job: &AvailableTtsInferenceJob, model_record: &TtsModelForInferenceRecord, ) -> Result<(Span, Span), ProcessSingleJobError> { @@ -71,7 +71,7 @@ pub async fn process_single_job( .attribute("user_token", maybe_user_token) .service_name("tts-inference-job"); - let mut job_progress_reporter = inferencer + let mut job_progress_reporter = job_args .job_progress_reporter .new_tts_inference(&job.inference_job_token) .map_err(|e| ProcessSingleJobError::Other(anyhow!(e)))?; @@ -81,7 +81,7 @@ pub async fn process_single_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) + mark_tts_inference_job_pending_and_grab_lock(&job_args.mysql_pool, job.id) .await .map_err(|e| ProcessSingleJobError::Other(e))?; @@ -103,19 +103,19 @@ pub async fn process_single_job( // ==================== CONFIRM OR DOWNLOAD WAVEGLOW VOCODER MODEL ==================== // let waveglow_vocoder_model_fs_path = { - let waveglow_vocoder_model_filename = inferencer.waveglow_vocoder_model_filename.clone(); - let waveglow_vocoder_model_fs_path = inferencer.semi_persistent_cache.tts_pretrained_vocoder_model_path(&waveglow_vocoder_model_filename); - let waveglow_vocoder_model_object_path = inferencer.bucket_path_unifier.tts_pretrained_vocoders_path(&waveglow_vocoder_model_filename); + let waveglow_vocoder_model_filename = job_args.waveglow_vocoder_model_filename.clone(); + let waveglow_vocoder_model_fs_path = job_args.semi_persistent_cache.tts_pretrained_vocoder_model_path(&waveglow_vocoder_model_filename); + let waveglow_vocoder_model_object_path = job_args.bucket_path_unifier.tts_pretrained_vocoders_path(&waveglow_vocoder_model_filename); maybe_download_file_from_bucket( "waveglow vocoder model", &waveglow_vocoder_model_fs_path, &waveglow_vocoder_model_object_path, - &inferencer.private_bucket_client, + &job_args.private_bucket_client, &mut job_progress_reporter, "downloading vocoder (1 of 3)", job.id.0, - &inferencer.scoped_temp_dir_creator, + &job_args.scoped_temp_dir_creator, ).await?; waveglow_vocoder_model_fs_path @@ -124,19 +124,19 @@ pub async fn process_single_job( // ==================== CONFIRM OR DOWNLOAD HIFIGAN (NORMAL) VOCODER MODEL ==================== // let pretrained_hifigan_vocoder_model_fs_path = { - let hifigan_vocoder_model_filename = inferencer.hifigan_vocoder_model_filename.clone(); - let hifigan_vocoder_model_fs_path = inferencer.semi_persistent_cache.tts_pretrained_vocoder_model_path(&hifigan_vocoder_model_filename); - let hifigan_vocoder_model_object_path = inferencer.bucket_path_unifier.tts_pretrained_vocoders_path(&hifigan_vocoder_model_filename); + let hifigan_vocoder_model_filename = job_args.hifigan_vocoder_model_filename.clone(); + let hifigan_vocoder_model_fs_path = job_args.semi_persistent_cache.tts_pretrained_vocoder_model_path(&hifigan_vocoder_model_filename); + let hifigan_vocoder_model_object_path = job_args.bucket_path_unifier.tts_pretrained_vocoders_path(&hifigan_vocoder_model_filename); maybe_download_file_from_bucket( "hifigan vocoder model", &hifigan_vocoder_model_fs_path, &hifigan_vocoder_model_object_path, - &inferencer.private_bucket_client, + &job_args.private_bucket_client, &mut job_progress_reporter, "downloading vocoder (2 of 3)", job.id.0, - &inferencer.scoped_temp_dir_creator, + &job_args.scoped_temp_dir_creator, ).await?; hifigan_vocoder_model_fs_path @@ -145,19 +145,19 @@ pub async fn process_single_job( // ==================== CONFIRM OR DOWNLOAD HIFIGAN (SUPERRES) VOCODER MODEL ==================== // let hifigan_superres_vocoder_model_fs_path = { - let hifigan_superres_vocoder_model_filename = inferencer.hifigan_superres_vocoder_model_filename.clone(); - let hifigan_superres_vocoder_model_fs_path = inferencer.semi_persistent_cache.tts_pretrained_vocoder_model_path(&hifigan_superres_vocoder_model_filename); - let hifigan_superres_vocoder_model_object_path = inferencer.bucket_path_unifier.tts_pretrained_vocoders_path(&hifigan_superres_vocoder_model_filename); + let hifigan_superres_vocoder_model_filename = job_args.hifigan_superres_vocoder_model_filename.clone(); + let hifigan_superres_vocoder_model_fs_path = job_args.semi_persistent_cache.tts_pretrained_vocoder_model_path(&hifigan_superres_vocoder_model_filename); + let hifigan_superres_vocoder_model_object_path = job_args.bucket_path_unifier.tts_pretrained_vocoders_path(&hifigan_superres_vocoder_model_filename); maybe_download_file_from_bucket( "hifigan superres vocoder model", &hifigan_superres_vocoder_model_fs_path, &hifigan_superres_vocoder_model_object_path, - &inferencer.private_bucket_client, + &job_args.private_bucket_client, &mut job_progress_reporter, "downloading vocoder (3 of 3)", job.id.0, - &inferencer.scoped_temp_dir_creator, + &job_args.scoped_temp_dir_creator, ).await?; hifigan_superres_vocoder_model_fs_path @@ -168,18 +168,18 @@ pub async fn process_single_job( let custom_vocoder_fs_path = match &model_record.maybe_custom_vocoder { None => None, Some(vocoder) => { - let custom_vocoder_fs_path = inferencer.semi_persistent_cache.custom_vocoder_model_path(&vocoder.vocoder_token); - let custom_vocoder_object_path = inferencer.bucket_path_unifier.vocoder_path(&vocoder.vocoder_private_bucket_hash); + let custom_vocoder_fs_path = job_args.semi_persistent_cache.custom_vocoder_model_path(&vocoder.vocoder_token); + let custom_vocoder_object_path = job_args.bucket_path_unifier.vocoder_path(&vocoder.vocoder_private_bucket_hash); maybe_download_file_from_bucket( "custom vocoder", &custom_vocoder_fs_path, &custom_vocoder_object_path, - &inferencer.private_bucket_client, + &job_args.private_bucket_client, &mut job_progress_reporter, "downloading user vocoder", job.id.0, - &inferencer.scoped_temp_dir_creator, + &job_args.scoped_temp_dir_creator, ).await?; Some(custom_vocoder_fs_path) @@ -189,18 +189,18 @@ pub async fn process_single_job( // ==================== CONFIRM OR DOWNLOAD TTS SYNTHESIZER MODEL ==================== // let tts_synthesizer_fs_path = { - let tts_synthesizer_fs_path = inferencer.semi_persistent_cache.tts_synthesizer_model_path(&model_record.model_token); - let tts_synthesizer_object_path = inferencer.bucket_path_unifier.tts_synthesizer_path(&model_record.private_bucket_hash); + let tts_synthesizer_fs_path = job_args.semi_persistent_cache.tts_synthesizer_model_path(&model_record.model_token); + let tts_synthesizer_object_path = job_args.bucket_path_unifier.tts_synthesizer_path(&model_record.private_bucket_hash); maybe_download_file_from_bucket( "synthesizer", &tts_synthesizer_fs_path, &tts_synthesizer_object_path, - &inferencer.private_bucket_client, + &job_args.private_bucket_client, &mut job_progress_reporter, "downloading synthesizer", job.id.0, - &inferencer.scoped_temp_dir_creator, + &job_args.scoped_temp_dir_creator, ).await?; tts_synthesizer_fs_path @@ -231,7 +231,7 @@ pub async fn process_single_job( .map_err(|e| ProcessSingleJobError::Other(e))?; // TODO: Fix this. - let maybe_unload_model_path = inferencer + let maybe_unload_model_path = job_args .virtual_model_lfu .insert_returning_replaced(tts_synthesizer_fs_path.to_str().unwrap_or("")) .map_err(|e| ProcessSingleJobError::Other(e))?; @@ -284,7 +284,7 @@ pub async fn process_single_job( }, }; - inferencer.http_clients.tts_inference_sidecar_client.request_inference( + job_args.http_clients.tts_inference_sidecar_client.request_inference( &cleaned_inference_text, &tts_synthesizer_fs_path, pretrained_vocoder, @@ -317,14 +317,14 @@ pub async fn process_single_job( job_progress_reporter.log_status("uploading result") .map_err(|e| ProcessSingleJobError::Other(e))?; - let audio_result_object_path = inferencer.bucket_path_unifier.tts_inference_wav_audio_output_path( + let audio_result_object_path = job_args.bucket_path_unifier.tts_inference_wav_audio_output_path( &job.uuid_idempotency_token); // TODO: Don't use this! info!("Audio destination bucket path: {:?}", &audio_result_object_path); info!("Uploading audio..."); - inferencer.public_bucket_client.upload_filename_with_content_type( + job_args.public_bucket_client.upload_filename_with_content_type( &audio_result_object_path, &output_audio_fs_path, "audio/wav") @@ -335,14 +335,14 @@ pub async fn process_single_job( // ==================== UPLOAD SPECTROGRAM TO BUCKETS ==================== // - let spectrogram_result_object_path = inferencer.bucket_path_unifier.tts_inference_spectrogram_output_path( + let spectrogram_result_object_path = job_args.bucket_path_unifier.tts_inference_spectrogram_output_path( &job.uuid_idempotency_token); // TODO: Don't use this! info!("Spectrogram destination bucket path: {:?}", &spectrogram_result_object_path); info!("Uploading spectrogram..."); - inferencer.public_bucket_client.upload_filename_with_content_type( + job_args.public_bucket_client.upload_filename_with_content_type( &spectrogram_result_object_path, &output_spectrogram_fs_path, "application/json") @@ -363,15 +363,15 @@ pub async fn process_single_job( // Default to showing the k8s node (machine) name, if possible, as this benefits // debugging on-prem workloads. - let worker_name = inferencer.worker_details.k8s_node_name.as_deref() - .or(inferencer.worker_details.k8s_pod_name.as_deref()) + let worker_name = job_args.worker_details.k8s_node_name.as_deref() + .or(job_args.worker_details.k8s_pod_name.as_deref()) .map(|name| name.to_string()) - .unwrap_or_else(|| inferencer.worker_details.worker_hostname.clone()); + .unwrap_or_else(|| job_args.worker_details.worker_hostname.clone()); info!("Saving tts inference record..."); let (id, inference_result_token) = insert_tts_result( - &inferencer.mysql_pool, + &job_args.mysql_pool, job, &text_hash, pretrained_vocoder, @@ -379,15 +379,15 @@ pub async fn process_single_job( &spectrogram_result_object_path, file_metadata.file_size_bytes, file_metadata.duration_millis.unwrap_or(0), - inferencer.worker_details.is_on_prem, + job_args.worker_details.is_on_prem, &worker_name, - inferencer.worker_details.is_debug_worker) + job_args.worker_details.is_debug_worker) .await .map_err(|e| ProcessSingleJobError::Other(e))?; info!("Marking job complete..."); mark_tts_inference_job_done( - &inferencer.mysql_pool, + &job_args.mysql_pool, job.id, true, Some(&inference_result_token)) @@ -396,7 +396,7 @@ pub async fn process_single_job( info!("TTS Done. Original text was: {}", &job.raw_inference_text); - inferencer.firehose_publisher.tts_inference_finished( + job_args.firehose_publisher.tts_inference_finished( job.maybe_creator_user_token.as_deref(), &model_record.model_token, &inference_result_token) diff --git a/crates/service/tts_inference_job/src/main.rs b/crates/service/tts_inference_job/src/main.rs index d8ad2fa2ad..b0ab6b93e6 100644 --- a/crates/service/tts_inference_job/src/main.rs +++ b/crates/service/tts_inference_job/src/main.rs @@ -36,6 +36,7 @@ use crate::http_clients::tts_sidecar_health_check_client::TtsSidecarHealthCheckC use crate::job_steps::health_check_trap::maybe_block_on_sidecar_health_check; use crate::job_steps::job_args::JobWorkerDetails; use crate::job_steps::job_args::{JobArgs, JobCaches, JobHttpClients}; +use crate::job_steps::job_stats::JobStats; use crate::job_steps::process_single_job::process_single_job; use crate::job_steps::process_single_job_error::ProcessSingleJobError; use crate::script_execution::tacotron_inference_command::TacotronInferenceCommand; @@ -306,6 +307,7 @@ async fn main() -> AnyhowResult<()> { tts_inference_sidecar_client, tts_sidecar_health_check_client, }, + job_stats: JobStats::new(), newrelic_client, newrelic_disabled, worker_details: JobWorkerDetails { @@ -344,10 +346,10 @@ async fn main() -> AnyhowResult<()> { const START_TIMEOUT_MILLIS : u64 = 500; const INCREASE_TIMEOUT_MILLIS : u64 = 1000; -async fn main_loop(inferencer: JobArgs) { +async fn main_loop(job_args: JobArgs) { let mut error_timeout_millis = START_TIMEOUT_MILLIS; - let mut noop_logger = NoOpLogger::new(inferencer.no_op_logger_millis as i64); + let mut noop_logger = NoOpLogger::new(job_args.no_op_logger_millis as i64); let mut span_batch = Vec::new(); @@ -357,30 +359,30 @@ async fn main_loop(inferencer: JobArgs) { let mut needs_health_check_at_start = true; // Run health check at startup. loop { - let num_records = inferencer.job_batch_size; + let num_records = job_args.job_batch_size; // Don't completely starve low-priority jobs - if sort_by_priority_count >= inferencer.low_priority_starvation_prevention_every_nth { + if sort_by_priority_count >= job_args.low_priority_starvation_prevention_every_nth { sort_by_priority_count = 0; sort_by_priority = false; } let maybe_available_jobs = - if let Some(minimum_priority) = inferencer.maybe_minimum_priority { + if let Some(minimum_priority) = job_args.maybe_minimum_priority { // Special high-priority workers list_available_tts_inference_jobs_with_minimum_priority( - &inferencer.mysql_pool, + &job_args.mysql_pool, minimum_priority, num_records, - inferencer.worker_details.is_debug_worker + job_args.worker_details.is_debug_worker ).await } else { // Normal path list_available_tts_inference_jobs( - &inferencer.mysql_pool, + &job_args.mysql_pool, sort_by_priority, num_records, - inferencer.worker_details.is_debug_worker + job_args.worker_details.is_debug_worker ).await }; @@ -400,14 +402,14 @@ async fn main_loop(inferencer: JobArgs) { if jobs.is_empty() { noop_logger.log_message_after_awhile("No TTS jobs picked up from database!"); - std::thread::sleep(Duration::from_millis(inferencer.job_batch_wait_millis)); + std::thread::sleep(Duration::from_millis(job_args.job_batch_wait_millis)); continue; } info!("Queried {} jobs from database", jobs.len()); let batch_result = process_jobs( - &inferencer, + &job_args, jobs, needs_health_check_at_start, ).await; @@ -426,18 +428,18 @@ async fn main_loop(inferencer: JobArgs) { } }; - if !inferencer.newrelic_disabled { + if !job_args.newrelic_disabled { span_batch.append(&mut spans); if span_batch.len() > 50 { let spans_to_send = span_batch.split_off(0).into(); - inferencer.newrelic_client.send_spans(spans_to_send).await; + job_args.newrelic_client.send_spans(spans_to_send).await; } } error_timeout_millis = START_TIMEOUT_MILLIS; // reset - std::thread::sleep(Duration::from_millis(inferencer.job_batch_wait_millis)); + std::thread::sleep(Duration::from_millis(job_args.job_batch_wait_millis)); } } @@ -557,6 +559,8 @@ async fn process_jobs( Err(e) => { warn!("Failure to process job: {:?}", e); + record_failure_and_maybe_slow_down(&inferencer.job_stats); + maybe_sidecar_health_issue = true; let failure_reason = ""; @@ -576,6 +580,7 @@ async fn process_jobs( delete_tts_synthesizers_from_cache(&inferencer.semi_persistent_cache)?; } } + } } } @@ -583,6 +588,31 @@ async fn process_jobs( Ok(batch_spans) } +fn record_failure_and_maybe_slow_down(job_stats: &JobStats) { + let stats = match job_stats.increment_failure_count() { + Ok(stats) => stats, + Err(e) => { + warn!("Error recording stats and reacting to repeated failures: {:?}", e); + return; // Can't really do anything. + } + }; + + let seconds_timeout = match stats.consecutive_failure_count { + t if t > 100 => 180, + t if t > 50 => 60, + t if t > 20 => 30, + t if t > 10 => 10, + t if t > 5 => 5, + _ => return, // No timeout + }; + + info!("Slowing down {} seconds due to significant repeated failures: {:?}", + seconds_timeout, + stats); + + std::thread::sleep(Duration::from_secs(seconds_timeout)); +} + /// Hack to delete locally cached TTS synthesizers to free up space from a full filesystem. /// This is not intelligent and doesn't use any LRU/LFU mechanic. /// This also relies on files not being read or written by concurrent workers while deleting.