diff --git a/crates/service/twitch_pubsub_subscriber/src/main.rs b/crates/service/twitch_pubsub_subscriber/src/main.rs index eca1f18f65..f583c4c70f 100644 --- a/crates/service/twitch_pubsub_subscriber/src/main.rs +++ b/crates/service/twitch_pubsub_subscriber/src/main.rs @@ -104,17 +104,13 @@ pub async fn main() -> AnyhowResult<()> { .enable_all() .build()?; - let runtime = Arc::new(runtime); - // https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html - let runtime_2 = Builder::new_multi_thread() + let runtime_2 = Arc::new(Builder::new_multi_thread() .worker_threads(8) .thread_name("twitch-pubsub-") .thread_stack_size(3 * 1024 * 1024) .enable_all() - .build()?; - - let runtime_2 = Arc::new(runtime_2); + .build()?); info!("Thread pool created"); diff --git a/crates/service/websocket_gateway/src/main.rs b/crates/service/websocket_gateway/src/main.rs index 3a2b3e4582..dd998883ee 100644 --- a/crates/service/websocket_gateway/src/main.rs +++ b/crates/service/websocket_gateway/src/main.rs @@ -24,7 +24,7 @@ use config::shared_constants::DEFAULT_REDIS_DATABASE_1_CONNECTION_STRING; use config::shared_constants::DEFAULT_RUST_LOG; use container_common::anyhow_result::AnyhowResult; use crate::endpoints_ws::obs_gateway_websocket_handler::obs_gateway_websocket_handler; -use crate::server_state::{ObsGatewayServerState, EnvConfig, TwitchOauthSecrets, BackendsConfig}; +use crate::server_state::{ObsGatewayServerState, EnvConfig, TwitchOauthSecrets, BackendsConfig, MultithreadingConfig}; use futures::Future; use futures::executor::ThreadPool; use http_server_common::cors::build_common_cors_config; @@ -40,7 +40,7 @@ use sqlx::mysql::MySqlPoolOptions; use std::sync::Arc; use std::thread::sleep; use std::time::Duration; -use tokio::runtime::{Handle, Runtime}; +use tokio::runtime::{Builder, Handle, Runtime}; use twitch_api2::pubsub::Topic; use twitch_api2::pubsub; use twitch_common::twitch_secrets::TwitchSecrets; @@ -106,7 +106,15 @@ async fn main() -> AnyhowResult<()> { let redis_pool = r2d2::Pool::builder() .build(redis_manager)?; + let runtime = Arc::new(Builder::new_multi_thread() + .worker_threads(32) + .thread_name("redis-pubsub-event-consumer-") + .thread_stack_size(3 * 1024 * 1024) + .enable_all() + .build()?); + let server_state = ObsGatewayServerState { + hostname: server_hostname, env_config: EnvConfig { num_workers, bind_address, @@ -119,11 +127,13 @@ async fn main() -> AnyhowResult<()> { client_id: twitch_secrets.app_client_id.clone(), client_secret: twitch_secrets.app_client_secret.clone(), }, - hostname: server_hostname, backends: BackendsConfig { mysql_pool: pool, redis_pool, - } + }, + multithreading: MultithreadingConfig { + redis_pubsub_runtime: runtime, + }, }; info!("Starting server..."); diff --git a/crates/service/websocket_gateway/src/server_state.rs b/crates/service/websocket_gateway/src/server_state.rs index 230ebdcbec..3630ed8739 100644 --- a/crates/service/websocket_gateway/src/server_state.rs +++ b/crates/service/websocket_gateway/src/server_state.rs @@ -1,18 +1,23 @@ use r2d2_redis::{r2d2, RedisConnectionManager}; use sqlx::MySqlPool; +use std::sync::Arc; +use tokio::runtime::Runtime; use twitch_oauth2::{ClientId, ClientSecret}; /// State that is injected into every endpoint. #[derive(Clone)] pub struct ObsGatewayServerState { + pub hostname: String, + /// Configuration from ENV vars. /// Some of this might not be used. pub env_config: EnvConfig, - pub hostname: String, pub twitch_oauth_secrets: TwitchOauthSecrets, pub backends: BackendsConfig, + + pub multithreading: MultithreadingConfig, } #[derive(Clone)] @@ -39,3 +44,8 @@ pub struct TwitchOauthSecrets { pub client_id: String, pub client_secret: String, } + +#[derive(Clone)] +pub struct MultithreadingConfig { + pub redis_pubsub_runtime: Arc, +}