Poll voice conversion model info; pull from cache

This commit is contained in:
Brandon Thomas committed 2023-07-23 19:12:16 -04:00
1 parent a601998fa5
commit 31c88d6d64
22 files changed
+299 -60

No files matched your search

@@ -92,7 +92,7 @@ WHERE vc.token = ?
return Ok(None);
},
_ => {
warn!("tts model query error: {:?}", err);
warn!("voice conversion model query error: {:?}", err);
return Err(VoiceConversionModelForInferenceError::DatabaseError {
reason: format!("Mysql error: {:?}", err)
});
@@ -1,3 +1,4 @@
pub mod inference;
pub mod model_info_lite;
pub mod models;
pub mod results;
@@ -0,0 +1,61 @@
use anyhow::anyhow;
use crate::queries::voice_conversion::model_info_lite::model_info_lite::VoiceConversionModelInfoLite;
use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType;
use errors::AnyhowResult;
use log::warn;
use sqlx::pool::PoolConnection;
use sqlx::{MySql, MySqlPool};
use tokens::voice_conversion::model::VoiceConversionModelToken;
pub async fn get_voice_conversion_model_info_lite(
token: &str,
mysql_pool: &MySqlPool,
) -> AnyhowResult<Option<VoiceConversionModelInfoLite>> {
let mut connection = mysql_pool.acquire().await?;
get_voice_conversion_model_info_lite_with_connection(token, &mut connection).await
}
pub async fn get_voice_conversion_model_info_lite_with_connection(
token: &str,
mysql_connection: &mut PoolConnection<MySql>,
) -> AnyhowResult<Option<VoiceConversionModelInfoLite>> {
let maybe_model= sqlx::query_as!(
RawVoiceConversionModelInfoLite,
r#"
SELECT
vc.token as `token: tokens::voice_conversion::model::VoiceConversionModelToken`,
vc.model_type as `model_type: enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType`
FROM voice_conversion_models as vc
WHERE vc.token = ?
"#,
token
)
.fetch_one(mysql_connection)
.await;
let model : RawVoiceConversionModelInfoLite = match maybe_model {
Ok(model) => model,
Err(err) => {
return match err {
sqlx::Error::RowNotFound => {
Ok(None)
},
_ => {
warn!("voice conversion model query error: {:?}", err);
Err(anyhow!("error querying voce conversion model info lite: {:?}", err))
}
}
}
};
Ok(Some(VoiceConversionModelInfoLite {
token: model.token,
model_type: model.model_type,
}))
}
struct RawVoiceConversionModelInfoLite {
pub token: VoiceConversionModelToken,
pub model_type: VoiceConversionModelType,
}
@@ -1,17 +1,10 @@
use crate::queries::voice_conversion::model_info_lite::model_info_lite::VoiceConversionModelInfoLite;
use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType;
use errors::AnyhowResult;
use sqlx::pool::PoolConnection;
use sqlx::{MySql, MySqlPool};
use tokens::voice_conversion::model::VoiceConversionModelToken;
/// This is meant to be used for quick in-memory caches,
/// particularly the one that serves the voice conversion enqueue API.
#[derive(Serialize)]
pub struct VoiceConversionModelInfoLite {
pub token: VoiceConversionModelToken,
pub model_type: VoiceConversionModelType,
}
pub async fn list_voice_conversion_model_info_lite(
mysql_pool: &MySqlPool,
) -> AnyhowResult<Vec<VoiceConversionModelInfoLite>> {
@@ -0,0 +1,3 @@
pub mod get_voice_conversion_model_info_lite;
pub mod list_voice_conversion_model_info_lite;
pub mod model_info_lite;
@@ -0,0 +1,10 @@
use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType;
use tokens::voice_conversion::model::VoiceConversionModelToken;
/// This is meant to be used for quick in-memory caches,
/// particularly the one that serves the voice conversion enqueue API.
#[derive(Serialize)]
pub struct VoiceConversionModelInfoLite {
pub token: VoiceConversionModelToken,
pub model_type: VoiceConversionModelType,
}
@@ -1,3 +1,2 @@
pub mod insert_voice_conversion_model_from_download_job;
pub mod list_voice_conversion_model_info_lite;
pub mod list_voice_conversion_models;
@@ -40,7 +40,7 @@ pub async fn list_fully_computed_assigned_tts_categories_handler(
_http_request: HttpRequest,
server_state: web::Data<Arc<ServerState>>) -> Result<HttpResponse, ListFullyComputedAssignedTtsCategoriesError>
{
let maybe_category_assignments = server_state.caches.tts_model_category_assignments.grab_copy_without_bump_if_unexpired()
let maybe_category_assignments = server_state.caches.ephemeral.tts_model_category_assignments.grab_copy_without_bump_if_unexpired()
.map_err(|e| {
error!("Error consulting cache: {:?}", e);
ListFullyComputedAssignedTtsCategoriesError::ServerError
@@ -53,11 +53,11 @@ pub async fn list_fully_computed_assigned_tts_categories_handler(
},
None => {
let category_assignments = query_and_construct_payload(
&server_state.caches.database_tts_category_list,
&server_state.caches.ephemeral.database_tts_category_list,
&server_state.mysql_pool)
.await?;
server_state.caches.tts_model_category_assignments.store_copy(&category_assignments)
server_state.caches.ephemeral.tts_model_category_assignments.store_copy(&category_assignments)
.map_err(|e| {
error!("Error storing cache: {:?}", e);
ListFullyComputedAssignedTtsCategoriesError::ServerError
@@ -95,7 +95,7 @@ pub async fn list_tts_categories_handler(
server_state: web::Data<Arc<ServerState>>) -> Result<HttpResponse, ListTtsCategoriesError>
{
let database_categories = list_cached_tts_categories_for_public_dropdown_db_pool(
&server_state.caches.database_tts_category_list, &server_state.mysql_pool)
&server_state.caches.ephemeral.database_tts_category_list, &server_state.mysql_pool)
.await
.map_err(|e| {
error!("error querying categories from db/cache: {:?}", e);
@@ -65,7 +65,7 @@ pub async fn get_pending_inference_job_count_handler(
});
}
let maybe_cached = server_state.caches.inference_queue_length.grab_copy_without_bump_if_unexpired()
let maybe_cached = server_state.caches.ephemeral.inference_queue_length.grab_copy_without_bump_if_unexpired()
.map_err(|e| {
error!("error consulting cache: {:?}", e);
GetPendingInferenceJobCountError::ServerError
@@ -88,7 +88,7 @@ pub async fn get_pending_inference_job_count_handler(
Err(err) => {
warn!("error querying database / inserting into cache: {:?}", err);
let maybe_cached = server_state.caches.inference_queue_length.grab_even_expired_and_bump()
let maybe_cached = server_state.caches.ephemeral.inference_queue_length.grab_even_expired_and_bump()
.map_err(|err| {
error!("error consulting cache (even expired): {:?}", err);
GetPendingInferenceJobCountError::ServerError
@@ -102,7 +102,7 @@ pub async fn get_pending_inference_job_count_handler(
// Happy path...
Ok(count_result) => {
server_state.caches.inference_queue_length.store_copy(&count_result)
server_state.caches.ephemeral.inference_queue_length.store_copy(&count_result)
.map_err(|e| {
error!("error storing cache: {:?}", e);
GetPendingInferenceJobCountError::ServerError
@@ -84,7 +84,7 @@ pub async fn leaderboard_handler(
server_state: web::Data<Arc<ServerState>>
) -> Result<HttpResponse, LeaderboardErrorResponse> {
let maybe_cached = server_state.caches.leaderboard.grab_copy_without_bump_if_unexpired()
let maybe_cached = server_state.caches.ephemeral.leaderboard.grab_copy_without_bump_if_unexpired()
.map_err(|e| {
error!("error consulting cache: {:?}", e);
LeaderboardErrorResponse::server_error()
@@ -108,7 +108,7 @@ pub async fn leaderboard_handler(
Err(err) => {
warn!("error querying database / inserting into cache: {:?}", err);
let maybe_cached = server_state.caches.leaderboard.grab_even_expired_and_bump()
let maybe_cached = server_state.caches.ephemeral.leaderboard.grab_even_expired_and_bump()
.map_err(|err| {
error!("error consulting cache (even expired): {:?}", err);
LeaderboardErrorResponse::server_error()
@@ -122,7 +122,7 @@ pub async fn leaderboard_handler(
// Happy path...
Ok(leaderboard_info) => {
server_state.caches.leaderboard.store_copy(&leaderboard_info)
server_state.caches.ephemeral.leaderboard.store_copy(&leaderboard_info)
.map_err(|e| {
error!("error storing cache: {:?}", e);
LeaderboardErrorResponse::server_error()
@@ -79,7 +79,7 @@ pub async fn get_unified_queue_stats_handler(
});
}
let maybe_cached = server_state.caches.queue_stats.grab_copy_without_bump_if_unexpired()
let maybe_cached = server_state.caches.ephemeral.queue_stats.grab_copy_without_bump_if_unexpired()
.map_err(|e| {
error!("error consulting cache: {:?}", e);
GetUnifiedQueueStatsError::ServerError
@@ -102,7 +102,7 @@ pub async fn get_unified_queue_stats_handler(
Err(err) => {
warn!("error querying database / inserting into cache: {:?}", err);
let maybe_cached = server_state.caches.queue_stats.grab_even_expired_and_bump()
let maybe_cached = server_state.caches.ephemeral.queue_stats.grab_even_expired_and_bump()
.map_err(|err| {
error!("error consulting cache (even expired): {:?}", err);
GetUnifiedQueueStatsError::ServerError
@@ -116,7 +116,7 @@ pub async fn get_unified_queue_stats_handler(
// Happy path...
Ok(stats_result) => {
server_state.caches.queue_stats.store_copy(&stats_result)
server_state.caches.ephemeral.queue_stats.store_copy(&stats_result)
.map_err(|e| {
error!("error storing cache: {:?}", e);
GetUnifiedQueueStatsError::ServerError
@@ -66,7 +66,7 @@ pub async fn get_pending_tts_inference_job_count_handler(
});
}
let maybe_cached = server_state.caches.tts_queue_length.grab_copy_without_bump_if_unexpired()
let maybe_cached = server_state.caches.ephemeral.tts_queue_length.grab_copy_without_bump_if_unexpired()
.map_err(|e| {
error!("error consulting cache: {:?}", e);
GetPendingTtsInferenceJobCountError::ServerError
@@ -89,7 +89,7 @@ pub async fn get_pending_tts_inference_job_count_handler(
Err(err) => {
warn!("error querying database / inserting into cache: {:?}", err);
let maybe_cached = server_state.caches.tts_queue_length.grab_even_expired_and_bump()
let maybe_cached = server_state.caches.ephemeral.tts_queue_length.grab_even_expired_and_bump()
.map_err(|err| {
error!("error consulting cache (even expired): {:?}", err);
GetPendingTtsInferenceJobCountError::ServerError
@@ -103,7 +103,7 @@ pub async fn get_pending_tts_inference_job_count_handler(
// Happy path...
Ok(count_result) => {
server_state.caches.tts_queue_length.store_copy(&count_result)
server_state.caches.ephemeral.tts_queue_length.store_copy(&count_result)
.map_err(|e| {
error!("error storing cache: {:?}", e);
GetPendingTtsInferenceJobCountError::ServerError
@@ -105,7 +105,7 @@ pub async fn list_tts_models_handler(
});
}
let maybe_models = server_state.caches.tts_model_list.grab_copy_without_bump_if_unexpired()
let maybe_models = server_state.caches.ephemeral.tts_model_list.grab_copy_without_bump_if_unexpired()
.map_err(|e| {
error!("Error consulting cache: {:?}", e);
ListTtsModelsError::ServerError
@@ -139,7 +139,7 @@ pub async fn list_tts_models_handler(
maybe_mysql_connection = Some(mysql_connection);
server_state.caches.tts_model_list.store_copy(&models)
server_state.caches.ephemeral.tts_model_list.store_copy(&models)
.map_err(|e| {
error!("Error storing cache: {:?}", e);
ListTtsModelsError::ServerError
@@ -9,20 +9,25 @@ use actix_web::{web, HttpResponse, HttpRequest};
use crate::configs::plans::get_correct_plan_for_session::get_correct_plan_for_session;
use crate::http_server::endpoints::investor_demo::demo_cookie::request_has_demo_cookie;
use crate::http_server::web_utils::response_error_helpers::to_simple_json_error;
use crate::memory_cache::model_token_to_info_cache::{ModelInfoLite, ModelTokenToInfoCache};
use crate::server_state::ServerState;
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType;
use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType;
use enums::common::visibility::Visibility;
use http_server_common::request::get_request_header_optional::get_request_header_optional;
use http_server_common::request::get_request_ip::get_request_ip;
use log::{info, warn};
use log::{error, info, warn};
use mysql_queries::payloads::generic_inference_args::{FundamentalFrequencyMethodForJob, GenericInferenceArgs, InferenceCategoryAbbreviated, PolymorphicInferenceArgs};
use mysql_queries::queries::generic_inference::web::insert_generic_inference_job::{InsertGenericInferenceArgs, insert_generic_inference_job};
use mysql_queries::queries::voice_conversion::model_info_lite::get_voice_conversion_model_info_lite::get_voice_conversion_model_info_lite_with_connection;
use r2d2_redis::redis::Commands;
use redis_common::redis_keys::RedisKeys;
use sqlx::MySql;
use sqlx::pool::PoolConnection;
use std::fmt;
use std::sync::Arc;
use enums::by_table::generic_inference_jobs::inference_input_source_token_type::InferenceInputSourceTokenType;
use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
use tokens::files::media_upload::MediaUploadToken;
use tokens::jobs::inference::InferenceJobToken;
use tokens::users::user::UserToken;
@@ -213,16 +218,24 @@ pub async fn enqueue_voice_conversion_inference_handler(
}
}
// ==================== CHECK AND ENQUEUE VOICE CONVERSION ==================== //
// ==================== LOOK UP MODEL INFO ==================== //
// TODO(bt): CHECK DATABASE!
let model_token = request.voice_conversion_model_token.clone();
let media_token = request.source_media_upload_token.clone();
let model_info_lite = lookup_model_info(
&model_token,
&server_state.caches.durable.model_token_info,
&mut mysql_connection
).await?;
// ==================== CHECK AND ENQUEUE VOICE CONVERSION ==================== //
let mut redis = server_state.redis_pool
.get()
.map_err(|e| {
warn!("redis error: {:?}", e);
error!("redis error: {:?}", e);
EnqueueVoiceConversionInferenceError::ServerError
})?;
@@ -268,7 +281,7 @@ pub async fn enqueue_voice_conversion_inference_handler(
let query_result = insert_generic_inference_job(InsertGenericInferenceArgs {
uuid_idempotency_token: &request.uuid_idempotency_token,
inference_category: InferenceCategory::VoiceConversion,
maybe_model_type: Some(InferenceModelType::SoVitsSvc), // TODO/FIXME: This is stupidly wrong.
maybe_model_type: Some(model_info_lite.model_type),
maybe_model_token: Some(model_token.as_str()),
maybe_input_source_token: Some(media_token.as_str()),
maybe_input_source_token_type: Some(InferenceInputSourceTokenType::MediaUpload),
@@ -316,3 +329,64 @@ pub async fn enqueue_voice_conversion_inference_handler(
.content_type("application/json")
.body(body))
}
async fn lookup_model_info(
model_token: &VoiceConversionModelToken,
cache: &ModelTokenToInfoCache,
mysql_connection: &mut PoolConnection<MySql>,
) -> Result<ModelInfoLite, EnqueueVoiceConversionInferenceError> {
let maybe_model_token_info = cache
.get_info(model_token.as_str())
.map_err(|err| {
error!("in-memory cache error: {:?}", err);
EnqueueVoiceConversionInferenceError::ServerError
})?;
let model_token_info = match maybe_model_token_info {
Some(info) => info,
None => {
let result = get_voice_conversion_model_info_lite_with_connection(
model_token.as_str(), mysql_connection)
.await
.map_err(|err| {
error!("model lookup error: {:?}", err);
EnqueueVoiceConversionInferenceError::ServerError
})?;
match result {
Some(info) => {
let model_type = match info.model_type {
VoiceConversionModelType::RvcV2 => InferenceModelType::RvcV2,
VoiceConversionModelType::SoVitsSvc => InferenceModelType::SoVitsSvc,
VoiceConversionModelType::SoftVc => {
// SoftVC is unsupported
return Err(EnqueueVoiceConversionInferenceError::BadInput("wrong model type".to_string()));
}
};
let model_info = ModelInfoLite {
inference_category: InferenceCategory::VoiceConversion,
model_type,
};
cache.insert_one(model_token.as_str(), &model_info)
.map_err(|err| {
error!("cache insertion error: {:?}", err);
EnqueueVoiceConversionInferenceError::ServerError
})?;
model_info
}
None => {
// NB: Fail open for now.
ModelInfoLite {
inference_category: InferenceCategory::VoiceConversion,
model_type: InferenceModelType::SoVitsSvc,
}
}
}
}
};
Ok(model_token_info)
}
@@ -113,7 +113,7 @@ pub async fn list_voice_conversion_models_handler(
});
}
let maybe_models = server_state.caches.voice_conversion_model_list.grab_copy_without_bump_if_unexpired()
let maybe_models = server_state.caches.ephemeral.voice_conversion_model_list.grab_copy_without_bump_if_unexpired()
.map_err(|e| {
error!("Error consulting cache: {:?}", e);
ListVoiceConversionModelsError::ServerError
@@ -146,7 +146,7 @@ pub async fn list_voice_conversion_models_handler(
maybe_mysql_connection = Some(mysql_connection);
server_state.caches.voice_conversion_model_list.store_copy(&models)
server_state.caches.ephemeral.voice_conversion_model_list.store_copy(&models)
.map_err(|e| {
error!("Error storing cache: {:?}", e);
ListVoiceConversionModelsError::ServerError
@@ -61,6 +61,7 @@ pub async fn list_w2l_templates_handler(
let maybe_templates = server_state
.caches
.ephemeral
.w2l_template_list
.grab_copy_without_bump_if_unexpired()
.map_err(|e| {
@@ -89,7 +90,7 @@ pub async fn list_w2l_templates_handler(
}
};
server_state.caches.w2l_template_list.store_copy(&templates)
server_state.caches.ephemeral.w2l_template_list.store_copy(&templates)
.map_err(|e| {
error!("error storing cache: {:?}", e);
ListW2lTemplatesError::ServerError
+34 -16
View File
@@ -62,7 +62,7 @@ use crate::configs::static_api_tokens::{StaticApiTokenConfig, StaticApiTokens, S
use crate::http_server::middleware::pushback_filter_middleware::PushbackFilter;
use crate::http_server::web_utils::redis_rate_limiter::RedisRateLimiter;
use crate::routes::add_routes;
use crate::server_state::{ServerState, EnvConfig, TwitchOauthSecrets, TwitchOauth, RedisRateLimiters, InMemoryCaches, StripeSettings, ServerInfo, StaticFeatureFlags, TrollBans};
use crate::server_state::{ServerState, EnvConfig, TwitchOauthSecrets, TwitchOauth, RedisRateLimiters, InMemoryCaches, StripeSettings, ServerInfo, StaticFeatureFlags, TrollBans, DurableInMemoryCaches, EphemeralInMemoryCaches};
use crate::threads::db_health_checker_thread::db_health_check_status::HealthCheckStatus;
use crate::threads::db_health_checker_thread::db_health_checker_thread::db_health_checker_thread;
use crate::threads::poll_ip_banlist_thread::poll_ip_bans;
@@ -95,6 +95,8 @@ use twitch_common::twitch_secrets::TwitchSecrets;
use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
use users_component::utils::session_checker::SessionChecker;
use users_component::utils::session_cookie_manager::SessionCookieManager;
use crate::memory_cache::model_token_to_info_cache::ModelTokenToInfoCache;
use crate::threads::poll_model_token_info_thread::poll_model_token_info_thread;
const DEFAULT_BIND_ADDRESS : &'static str = "0.0.0.0:12345";
@@ -330,12 +332,16 @@ async fn main() -> AnyhowResult<()> {
let user_token_troll_bans = load_troll_user_token_bans();
let ip_address_troll_bans = load_ip_address_troll_bans();
let model_token_info_cache = ModelTokenToInfoCache::new();
let model_token_info_cache2 = model_token_info_cache.clone();
// Background jobs.
let health_check_status = HealthCheckStatus::new();
let health_check_status2 = health_check_status.clone();
let mysql_pool3 = pool.clone();
let mysql_pool4 = pool.clone();
let mysql_pool5 = pool.clone();
let tokio_runtime = Runtime::new()?;
@@ -355,6 +361,13 @@ async fn main() -> AnyhowResult<()> {
poll_ip_bans(ip_ban_list2, mysql_pool4).await;
});
info!("Spawning token info cache polling thread.");
tokio_runtime.spawn(async {
poll_model_token_info_thread(model_token_info_cache2, mysql_pool5).await;
});
let stripe_configs = StripeConfig {
checkout: StripeCheckoutConfigs {
success_url: FullUrlOrPath::Path(easyenv::get_env_string_required("STRIPE_CHECKOUT_SUCCESS_URL_PATH")?),
@@ -448,21 +461,26 @@ async fn main() -> AnyhowResult<()> {
sort_key_crypto,
static_api_token_set,
caches: InMemoryCaches {
tts_model_list: voice_list_cache,
voice_conversion_model_list: SingleItemTtlCache::create_with_duration(
easyenv::get_env_duration_seconds_or_default(
"VOICE_CONVERSION_MODEL_LIST_CACHE_TTL_SECONDS",
Duration::from_secs(60))),
w2l_template_list: w2l_template_cache,
database_tts_category_list: database_tts_category_list_cache,
tts_queue_length: tts_queue_length_cache,
tts_model_category_assignments: tts_model_category_assignments_cache,
leaderboard: leaderboard_cache,
inference_queue_length: inference_queue_length_cache,
queue_stats: SingleItemTtlCache::create_with_duration(
easyenv::get_env_duration_seconds_or_default(
"QUEUE_STATS_CACHE_TTL_SECONDS",
Duration::from_secs(60))),
durable: DurableInMemoryCaches {
model_token_info: model_token_info_cache,
},
ephemeral: EphemeralInMemoryCaches {
tts_model_list: voice_list_cache,
voice_conversion_model_list: SingleItemTtlCache::create_with_duration(
easyenv::get_env_duration_seconds_or_default(
"VOICE_CONVERSION_MODEL_LIST_CACHE_TTL_SECONDS",
Duration::from_secs(60))),
w2l_template_list: w2l_template_cache,
database_tts_category_list: database_tts_category_list_cache,
tts_queue_length: tts_queue_length_cache,
tts_model_category_assignments: tts_model_category_assignments_cache,
leaderboard: leaderboard_cache,
inference_queue_length: inference_queue_length_cache,
queue_stats: SingleItemTtlCache::create_with_duration(
easyenv::get_env_duration_seconds_or_default(
"QUEUE_STATS_CACHE_TTL_SECONDS",
Duration::from_secs(60))),
}
},
twitch_oauth: TwitchOauth {
secrets: TwitchOauthSecrets {
@@ -25,22 +25,22 @@ impl ModelTokenToInfoCache {
}
}
pub fn insert_one(&self, token: &str, info: ModelInfoLite) -> AnyhowResult<()> {
pub fn insert_one(&self, token: &str, info: &ModelInfoLite) -> AnyhowResult<()> {
match self.database.write() {
Err(err) => return Err(anyhow!("lock err: {:?}", err)),
Ok(mut lock) => {
lock.insert(token.to_string(), info);
lock.insert(token.to_string(), info.clone());
}
}
Ok(())
}
pub fn insert_many(&self, records: Vec<(&str, ModelInfoLite)>) -> AnyhowResult<()> {
pub fn insert_many(&self, records: Vec<(String, ModelInfoLite)>) -> AnyhowResult<()> {
match self.database.write() {
Err(err) => return Err(anyhow!("lock err: {:?}", err)),
Ok(mut lock) => {
for (token, info) in records.into_iter() {
lock.insert(token.to_string(), info);
lock.insert(token, info);
}
}
}
@@ -27,6 +27,7 @@ use sqlx::MySqlPool;
use url_config::third_party_url_redirector::ThirdPartyUrlRedirector;
use users_component::utils::session_checker::SessionChecker;
use users_component::utils::session_cookie_manager::SessionCookieManager;
use crate::memory_cache::model_token_to_info_cache::ModelTokenToInfoCache;
/// State that is injected into every endpoint.
pub struct ServerState {
@@ -134,9 +135,23 @@ pub struct RedisRateLimiters {
pub model_upload: RedisRateLimiter,
}
/// In-memory caches with TTL-based eviction.
/// In-memory caches of several types.
#[derive(Clone)]
pub struct InMemoryCaches {
pub durable: DurableInMemoryCaches,
pub ephemeral: EphemeralInMemoryCaches,
}
/// Durable caches
#[derive(Clone)]
pub struct DurableInMemoryCaches {
/// Lightweight model token (for generic inference) metadata
pub model_token_info: ModelTokenToInfoCache,
}
/// In-memory caches with TTL-based eviction.
#[derive(Clone)]
pub struct EphemeralInMemoryCaches {
/// Contains a list of all TTS models.
pub tts_model_list: SingleItemTtlCache<Vec<TtsModelRecordForResponse>>,
@@ -179,7 +194,6 @@ pub struct StripeSettings {
pub client: stripe::Client,
}
/// Flags set at service startup
#[derive(Clone)]
pub struct StaticFeatureFlags {
@@ -1,2 +1,3 @@
pub mod db_health_checker_thread;
pub mod poll_ip_banlist_thread;
pub mod poll_ip_banlist_thread;
pub mod poll_model_token_info_thread;
@@ -0,0 +1,64 @@
use crate::memory_cache::model_token_to_info_cache::{ModelInfoLite, ModelTokenToInfoCache};
use enums::by_table::generic_inference_jobs::inference_category::InferenceCategory;
use enums::by_table::generic_inference_jobs::inference_model_type::InferenceModelType;
use enums::by_table::voice_conversion_models::voice_conversion_model_type::VoiceConversionModelType;
use log::{debug, error, info};
use mysql_queries::queries::voice_conversion::model_info_lite::list_voice_conversion_model_info_lite::list_voice_conversion_model_info_lite;
use sqlx::MySqlPool;
use std::time::Duration;
pub async fn poll_model_token_info_thread(
model_token_info_cache: ModelTokenToInfoCache,
mysql_pool: MySqlPool,
) {
let startup_wait = easyenv::get_env_duration_seconds_or_default(
"POLL_MODEL_TOKEN_STARTUP_WAIT_DURATION_SECS", Duration::from_secs(5));
let interval_wait = easyenv::get_env_duration_seconds_or_default(
"POLL_MODEL_TOKEN_INTERVAL_SECS", Duration::from_secs(10 * 60));
std::thread::sleep(startup_wait);
loop {
debug!("Job fetching token info...");
let token_infos=
match list_voice_conversion_model_info_lite(&mysql_pool).await {
Ok(infos) => infos,
Err(err) => {
error!("Error polling model token info: {:?}", err);
std::thread::sleep(interval_wait);
continue;
}
};
let mut token_info_items = Vec::with_capacity(token_infos.len());
for token_info in token_infos.into_iter() {
let model_type = match token_info.model_type {
VoiceConversionModelType::RvcV2 => InferenceModelType::RvcV2,
VoiceConversionModelType::SoVitsSvc => InferenceModelType::SoVitsSvc,
VoiceConversionModelType::SoftVc => {
continue // NB: SoftVC is not supported.
},
};
let info = ModelInfoLite {
inference_category: InferenceCategory::VoiceConversion,
model_type: InferenceModelType::RvcV2,
};
token_info_items.push((token_info.token.to_string(), info));
}
let database_count = token_info_items.len();
info!("Job found {} token info items from database.", database_count);
if let Err(err) = model_token_info_cache.insert_many(token_info_items) {
error!("Error inserting model token info: {:?}", err);
}
std::thread::sleep(interval_wait);
}
}