From 30202e6e5a2bc3c88cbef7ac964ae70ac76465c8 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 11 Jan 2023 01:39:48 -0500 Subject: [PATCH] Fix build errors. This is largely because `inference_job` is not done --- .../src/caching/cache_miss_strategizer.rs | 2 +- .../src/job_steps/job_dependencies.rs | 12 +++++- .../inference_job/src/job_steps/main_loop.rs | 10 ++--- ...{main_loop_old.rs => main_loop_old.rs.txt} | 0 .../job/inference_job/src/job_steps/mod.rs | 2 - .../src/job_steps/process_single_job.rs | 37 ++++++++++--------- ...b_old.rs => process_single_job_old.rs.txt} | 0 .../src/job_types/tts/handle_tts_inference.rs | 16 ++++---- crates/service/job/inference_job/src/main.rs | 9 +++-- .../endpoints/media_uploads/upload_media.rs | 2 +- 10 files changed, 49 insertions(+), 41 deletions(-) rename crates/service/job/inference_job/src/job_steps/{main_loop_old.rs => main_loop_old.rs.txt} (100%) rename crates/service/job/inference_job/src/job_steps/{process_single_job_old.rs => process_single_job_old.rs.txt} (100%) diff --git a/crates/service/job/inference_job/src/caching/cache_miss_strategizer.rs b/crates/service/job/inference_job/src/caching/cache_miss_strategizer.rs index 3d99c4fb72..88ee4e9ab9 100644 --- a/crates/service/job/inference_job/src/caching/cache_miss_strategizer.rs +++ b/crates/service/job/inference_job/src/caching/cache_miss_strategizer.rs @@ -122,7 +122,7 @@ impl CacheMissStrategizer { #[cfg(test)] mod tests { use chrono::{Duration, Utc, DateTime}; - use crate::{CacheMissStrategizer, CacheMissStrategy}; + use crate::caching::cache_miss_strategizer::{CacheMissStrategizer, CacheMissStrategy}; fn get_date(datetime: &str) -> DateTime { let datetime = DateTime::parse_from_rfc3339(datetime).unwrap(); diff --git a/crates/service/job/inference_job/src/job_steps/job_dependencies.rs b/crates/service/job/inference_job/src/job_steps/job_dependencies.rs index 30bac353d3..1c64cd21c0 100644 --- a/crates/service/job/inference_job/src/job_steps/job_dependencies.rs +++ b/crates/service/job/inference_job/src/job_steps/job_dependencies.rs @@ -13,6 +13,8 @@ use jobs_common::job_progress_reporter::job_progress_reporter::JobProgressReport use jobs_common::semi_persistent_cache_dir::SemiPersistentCacheDir; use memory_caching::multi_item_ttl_cache::MultiItemTtlCache; use newrelic_telemetry::Client as NewRelicClient; +use r2d2_redis::RedisConnectionManager; +use r2d2_redis::r2d2; use sqlx::MySqlPool; use std::path::PathBuf; @@ -22,6 +24,13 @@ pub struct JobDependencies { pub mysql_pool: MySqlPool, + // TODO(2023-01-11): We don't always connect to a Redis + // Typically this is for job status reporting, but we might also report on when users leave the + // site to proactively kill their inference jobs and save on worker quota. + // On local dev we probably don't care about Redis at all, and on on-prem workers, we cannot + // connect to production Redis easily (requires lots of setup - ghosttunnel or something + IP rules) + pub maybe_redis_pool: Option>, + pub job_progress_reporter: Box, pub private_bucket_client: BucketClient, @@ -64,8 +73,7 @@ pub struct JobDependencies { pub job_batch_wait_millis: u64, // Max job attempts before failure. - // NB: This is an i32 so we don't need to convert to db column type. - pub job_max_attempts: i32, + pub job_max_attempts: u16, // Number of jobs to dequeue at once. pub job_batch_size: u32, diff --git a/crates/service/job/inference_job/src/job_steps/main_loop.rs b/crates/service/job/inference_job/src/job_steps/main_loop.rs index 4389049778..434e6b202d 100644 --- a/crates/service/job/inference_job/src/job_steps/main_loop.rs +++ b/crates/service/job/inference_job/src/job_steps/main_loop.rs @@ -1,5 +1,4 @@ use crate::job_steps::job_dependencies::JobDependencies; -use crate::job_steps::process_single_job_old::process_single_job_old; use database_queries::queries::generic_inference::job::list_available_generic_inference_jobs::{AvailableInferenceJob, list_available_generic_inference_jobs, ListAvailableGenericInferenceJobArgs}; use database_queries::queries::generic_inference::job::mark_generic_inference_job_failure::mark_generic_inference_job_failure; use errors::AnyhowResult; @@ -77,19 +76,20 @@ pub async fn main_loop(job_dependencies: JobDependencies) { // TODO: A common interface/trait for each submodule (tts, webvc) to declare how to determine if the job is "ready". // This probably returns a struct or enum with some measure of how many GB need to be downloaded. -async fn process_job_batch(job_state: &JobState, jobs: Vec) -> AnyhowResult<()> { +async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec) -> AnyhowResult<()> { for job in jobs.into_iter() { - let result = process_single_job(job_state, &job).await; + let result = process_single_job(job_dependencies, &job).await; match result { Ok(_) => {}, Err(e) => { warn!("Failure to process job: {:?}", e); let failure_reason = ""; let _r = mark_generic_inference_job_failure( - &job_state.mysql_pool, + &job_dependencies.mysql_pool, &job, failure_reason, - job_state.job_max_attempts + failure_reason, + job_dependencies.job_max_attempts ).await; } } diff --git a/crates/service/job/inference_job/src/job_steps/main_loop_old.rs b/crates/service/job/inference_job/src/job_steps/main_loop_old.rs.txt similarity index 100% rename from crates/service/job/inference_job/src/job_steps/main_loop_old.rs rename to crates/service/job/inference_job/src/job_steps/main_loop_old.rs.txt diff --git a/crates/service/job/inference_job/src/job_steps/mod.rs b/crates/service/job/inference_job/src/job_steps/mod.rs index e1f0d557b6..9cc6e3c9fb 100644 --- a/crates/service/job/inference_job/src/job_steps/mod.rs +++ b/crates/service/job/inference_job/src/job_steps/mod.rs @@ -1,9 +1,7 @@ pub mod job_dependencies; pub mod job_stats; pub mod main_loop; -pub mod main_loop_old; pub mod process_single_job; pub mod process_single_job_error; -pub mod process_single_job_old; pub mod select_job_batch; pub mod sidecar_health_check_trap; diff --git a/crates/service/job/inference_job/src/job_steps/process_single_job.rs b/crates/service/job/inference_job/src/job_steps/process_single_job.rs index abcdb2631a..210eaa022d 100644 --- a/crates/service/job/inference_job/src/job_steps/process_single_job.rs +++ b/crates/service/job/inference_job/src/job_steps/process_single_job.rs @@ -1,19 +1,15 @@ -use anyhow::anyhow; -use config::is_bad_download_url::is_bad_download_url; -use container_common::filesystem::safe_delete_temp_directory::safe_delete_temp_directory; use crate::job_steps::job_dependencies::JobDependencies; use database_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; use database_queries::queries::generic_inference::job::mark_generic_inference_job_pending_and_grab_lock::mark_generic_inference_job_pending_and_grab_lock; use database_queries::queries::generic_inference::job::mark_generic_inference_job_successfully_done::mark_generic_inference_job_successfully_done; use enums::workers::generic_inference_type::GenericInferenceType; use errors::AnyhowResult; -use jobs_common::redis_job_status_logger::RedisJobStatusLogger; use log::{info, warn}; -use tempdir::TempDir; pub async fn process_single_job(job_dependencies: &JobDependencies, job: &AvailableInferenceJob) -> AnyhowResult<()> { - let mut redis = job_dependencies.redis_pool.get()?; - let mut redis_logger = RedisJobStatusLogger::new_generic_download(&mut redis, job.download_job_token.as_str()); + // TODO(bt, 2023-01-11): Restore an optional status logger + //let mut redis = job_dependencies.redis_pool.get()?; + //let mut redis_logger = RedisJobStatusLogger::new_generic_download(&mut redis, job.download_job_token.as_str()); // ==================== ATTEMPT TO GRAB JOB LOCK ==================== // @@ -24,13 +20,15 @@ pub async fn process_single_job(job_dependencies: &JobDependencies, job: &Availa return Ok(()) } - info!("Beginning work on {:?} = {} | {}", job.download_type, job.download_job_token, job.download_url); + info!("Beginning work on {:?} = {}", job.inference_type, job.inference_job_token); // ==================== SETUP TEMP DIRS ==================== // let temp_dir = format!("temp_{}", job.id.0); let temp_dir = job_dependencies.scoped_temp_dir_creator.new_tempdir(&temp_dir)?; + let _p = temp_dir.path(); // TODO: Just so the build doesn't complain about unused. Remove. + // ==================== HANDLE DIFFERENT INFERENCE TYPES ==================== // let mut entity_token : Option = None; @@ -39,6 +37,8 @@ pub async fn process_single_job(job_dependencies: &JobDependencies, job: &Availa match job.inference_type { GenericInferenceType::TextToSpeech => { // TODO + entity_type = Some("todo".to_string()); // TODO + entity_token = Some("todo".to_string()); // TODO } GenericInferenceType::VoiceConversion => { // TODO @@ -101,16 +101,19 @@ pub async fn process_single_job(job_dependencies: &JobDependencies, job: &Availa info!("Saved model record: {} - {}", job.id.0, &job.inference_job_token); - job_dependencies.firehose_publisher.publish_generic_download_finished( - &job.creator_user_token, - entity_token.as_deref()) - .await - .map_err(|e| { - warn!("error publishing event: {:?}", e); - anyhow!("error publishing event") - })?; + // TODO(bt, 2023-01-11): Need to publish that the job finished. + // Publish the *correct type* of event. + //job_dependencies.firehose_publisher.publish_generic_download_finished( + // &job.maybe_creator_user_token, + // entity_token.as_deref()) + // .await + // .map_err(|e| { + // warn!("error publishing event: {:?}", e); + // anyhow!("error publishing event") + // })?; - redis_logger.log_status("done")?; + // TODO(bt, 2023-01-11): Restore optional Redis logging + //redis_logger.log_status("done")?; info!("Job done: {}", job.id.0); diff --git a/crates/service/job/inference_job/src/job_steps/process_single_job_old.rs b/crates/service/job/inference_job/src/job_steps/process_single_job_old.rs.txt similarity index 100% rename from crates/service/job/inference_job/src/job_steps/process_single_job_old.rs rename to crates/service/job/inference_job/src/job_steps/process_single_job_old.rs.txt diff --git a/crates/service/job/inference_job/src/job_types/tts/handle_tts_inference.rs b/crates/service/job/inference_job/src/job_types/tts/handle_tts_inference.rs index 6e4a882ff7..b4371278e3 100644 --- a/crates/service/job/inference_job/src/job_types/tts/handle_tts_inference.rs +++ b/crates/service/job/inference_job/src/job_types/tts/handle_tts_inference.rs @@ -1,18 +1,16 @@ -use anyhow::anyhow; -use log::{info, warn}; -use tempdir::TempDir; use crate::job_steps::job_dependencies::JobDependencies; -use crate::job_steps::process_single_job_error::ProcessSingleJobError; use database_queries::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; use errors::AnyhowResult; -use tts_common::clean_symbols::clean_symbols; -use crate::util::download_file_from_bucket::maybe_download_file_from_bucket; pub async fn handle_tts_inference( - job_dependencies: &JobDependencies, - job: &AvailableInferenceJob, + _job_dependencies: &JobDependencies, + _job: &AvailableInferenceJob, ) -> AnyhowResult<()> { + /* + TODO(bt, 2023-01-11): This was just copied verbatim from tts-inference-job. + It's not ready to run from inference-job yet as it needs adjustments to the new schema. + let mut job_progress_reporter = job_dependencies .job_progress_reporter .new_tts_inference(job.inference_job_token.as_str()) @@ -339,6 +337,6 @@ pub async fn handle_tts_inference( job_iteration_span.set_attribute("status", "success"); job_iteration_span.set_duration(duration); - Ok((since_creation_span, job_iteration_span)) + */ Ok(()) } \ No newline at end of file diff --git a/crates/service/job/inference_job/src/main.rs b/crates/service/job/inference_job/src/main.rs index 5ace5dea1c..dae31d27cf 100644 --- a/crates/service/job/inference_job/src/main.rs +++ b/crates/service/job/inference_job/src/main.rs @@ -38,7 +38,6 @@ use crate::http_clients::tts_sidecar_health_check_client::TtsSidecarHealthCheckC use crate::job_steps::job_dependencies::JobWorkerDetails; use crate::job_steps::job_dependencies::{JobDependencies, JobCaches, JobHttpClients}; use crate::job_steps::job_stats::JobStats; -use crate::job_steps::main_loop_old::main_loop_old; use crate::script_execution::tacotron_inference_command::TacotronInferenceCommand; use crate::util::scoped_temp_dir_creator::ScopedTempDirCreator; use database_queries::mediators::firehose_publisher::FirehosePublisher; @@ -286,10 +285,11 @@ async fn main() -> AnyhowResult<()> { } }; - let inferencer = JobDependencies { + let _job_dependencies = JobDependencies { scoped_temp_dir_creator: ScopedTempDirCreator::for_directory(&temp_directory), download_temp_directory: temp_directory, mysql_pool, + maybe_redis_pool: None, // TODO(bt, 2023-01-11): See note in JobDependencies job_progress_reporter, public_bucket_client, private_bucket_client, @@ -320,7 +320,7 @@ async fn main() -> AnyhowResult<()> { hifigan_vocoder_model_filename, hifigan_superres_vocoder_model_filename, job_batch_wait_millis: common_env.job_batch_wait_millis, - job_max_attempts: common_env.job_max_attempts as i32, + job_max_attempts: common_env.job_max_attempts as u16, job_batch_size: common_env.job_batch_size, no_op_logger_millis: common_env.no_op_logger_millis, sidecar_max_synthesizer_models, @@ -328,7 +328,8 @@ async fn main() -> AnyhowResult<()> { maybe_minimum_priority, }; - main_loop_old(inferencer).await; + // TODO: Main loop start + //main_loop_old(inferencer).await; Ok(()) } diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/media_uploads/upload_media.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/media_uploads/upload_media.rs index 25287fcf56..130913e9bc 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/media_uploads/upload_media.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/media_uploads/upload_media.rs @@ -14,6 +14,7 @@ use crate::http_server::web_utils::response_error_helpers::to_simple_json_error; use crate::server_state::ServerState; use crate::validations::model_uploads::validate_model_title; use crate::validations::validate_idempotency_token_format::validate_idempotency_token_format; +use database_queries::payloads::media_upload_details::MediaUploadDetails; use database_queries::queries::media_uploads::insert_media_upload::{Args, insert_media_upload}; use enums::core::visibility::Visibility; use enums::files::media_upload_type::MediaUploadType; @@ -24,7 +25,6 @@ use http_server_common::response::serialize_as_json_error::serialize_as_json_err use log::{info, warn, log}; use media::decode_basic_audio_info::decode_basic_audio_info; use regex::Regex; -use reusable_types::db::payloads::MediaUploadDetails; use sqlx::error::DatabaseError; use sqlx::error::Error::Database; use sqlx::mysql::MySqlDatabaseError;