Slow down jobs on repeated failures

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,

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<RwLock<JobStatsInner>>,
+}
+
+/// 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<SuccessAndFailureStats> {
+      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<SuccessAndFailureStats> {
+      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,
+      })
+   }
+}

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;

crates/service/tts_inference_job/src/job_steps/process_single_job.rs

@@ -45,7 +45,7 @@ fn read_metadata_file(filename: &PathBuf) -> AnyhowResult<FileMetadata> {
 }
 
 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)

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.