diff --git a/Cargo.lock b/Cargo.lock index 4f7deb3082..0a5ef3e12d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1046,6 +1046,7 @@ name = "container_common" version = "0.0.1" dependencies = [ "anyhow", + "log", "once_cell", "rand 0.8.4", "uuid", diff --git a/crates/lib/container_common/Cargo.toml b/crates/lib/container_common/Cargo.toml index 6976629abf..8cdd40dd2b 100644 --- a/crates/lib/container_common/Cargo.toml +++ b/crates/lib/container_common/Cargo.toml @@ -19,3 +19,4 @@ anyhow = "1.0.40" once_cell = "1.9.0" rand = "0.8.3" uuid = { version = "0.8.2", features = ["v4"] } +log = "0.4.14" diff --git a/crates/lib/container_common/src/thread/async_thread_kill_signal.rs b/crates/lib/container_common/src/thread/async_thread_kill_signal.rs new file mode 100644 index 0000000000..6d014c78ad --- /dev/null +++ b/crates/lib/container_common/src/thread/async_thread_kill_signal.rs @@ -0,0 +1,78 @@ +use anyhow::anyhow; +use crate::anyhow_result::AnyhowResult; +use std::sync::{Arc, RwLock, PoisonError, RwLockWriteGuard}; +use std::time::{Instant, Duration}; + +/// Sometimes I need to kill a thread from somewhere else. This is a mechanism +/// that enables it. This utility can be combined with logic that will kill a +/// thread's JoinHandle. +/// NB: This was designed with interior mutability and clone-ability in mind. + +#[derive(Clone)] +pub struct AsyncThreadKillSignal { + internal: Arc>, +} + +#[derive(Clone)] +struct AsyncThreadKillSignalInternal { + ttl: Option, + last_refreshed: Instant, + manually_killed: bool, +} + +impl AsyncThreadKillSignal { + + pub fn new_with_ttl(ttl: Duration) -> Self { + Self { + internal: Arc::new(RwLock::new(AsyncThreadKillSignalInternal { + last_refreshed: Instant::now(), + ttl: Some(ttl), + manually_killed: false, + })) + } + } + + /// Whether we should keep the thread alive. + pub fn is_alive(&self) -> AnyhowResult { + match self.internal.read() { + Err(e) => { + Err(anyhow!("lock issue: {:?}", e)) + }, + Ok(internal) => { + if let Some(ttl) = &internal.ttl { + let now = Instant::now(); + if now.duration_since(internal.last_refreshed).gt(ttl) { + return Ok(false); + } + } + Ok(!internal.manually_killed) + }, + } + } + + /// If the thread has a TTL, refresh it. + pub fn bump_ttl(&self) -> AnyhowResult<()> { + match self.internal.write() { + Err(e) => { + Err(anyhow!("lock issue: {:?}", e)) + }, + Ok(mut internal) => { + internal.last_refreshed = Instant::now(); + Ok(()) + }, + } + } + + /// Tell the thread it should die. + pub fn mark_thread_for_kill(&self) -> AnyhowResult<()> { + match self.internal.write() { + Err(e) => { + Err(anyhow!("lock issue: {:?}", e)) + }, + Ok(mut internal) => { + internal.manually_killed = true; + Ok(()) + }, + } + } +} diff --git a/crates/lib/container_common/src/thread/mod.rs b/crates/lib/container_common/src/thread/mod.rs index 4a3d1cc9ff..04a9b2ba41 100644 --- a/crates/lib/container_common/src/thread/mod.rs +++ b/crates/lib/container_common/src/thread/mod.rs @@ -1 +1,2 @@ -pub mod thread_id; \ No newline at end of file +pub mod thread_id; +pub mod async_thread_kill_signal; \ No newline at end of file diff --git a/crates/service/websocket_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs b/crates/service/websocket_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs index 249def5a5b..6863649a90 100644 --- a/crates/service/websocket_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs +++ b/crates/service/websocket_gateway/src/endpoints_ws/obs_gateway_websocket_handler.rs @@ -4,8 +4,12 @@ use actix_web::web::Path; use actix_web::{middleware, web, App, Error, HttpRequest, HttpResponse, HttpServer}; use actix_web_actors::ws; use container_common::anyhow_result::AnyhowResult; +use container_common::thread::async_thread_kill_signal::AsyncThreadKillSignal; use container_common::token::random_crockford_token::random_crockford_token; +use crate::endpoints_ws::helpers::publish_active_browser_info::publish_active_browser_info; use crate::endpoints_ws::obs_gateway_websocket_handler::ResponseType::TtsEvent; +use crate::endpoints_ws::threads::redis_pubsub_event_listener_thread::RedisPubsubEventListenerThread; +use crate::endpoints_ws::threads::tts_inference_job_token_queue::TtsInferenceJobTokenQueue; use crate::server_state::ObsGatewayServerState; use database_queries::queries::twitch::twitch_oauth::find::{TwitchOauthTokenFinder, TwitchOauthTokenRecord}; use futures_timer::Delay; @@ -25,30 +29,18 @@ use std::sync::Arc; use std::thread::sleep; use std::time::Duration; use tokio::runtime::Handle; +use tokio::task::JoinHandle; use twitch_api2::pubsub::Topic; use twitch_api2::pubsub::TopicData::ChannelPointsChannelV1; use twitch_api2::pubsub::channel_bits::ChannelBitsEventsV2Reply::BitsEvent; use twitch_api2::pubsub; use twitch_common::twitch_user_id::TwitchUserId; -use crate::endpoints_ws::helpers::publish_active_browser_info::publish_active_browser_info; -use crate::endpoints_ws::threads::redis_pubsub_event_listener_thread::RedisPubsubEventListenerThread; -use crate::endpoints_ws::threads::tts_inference_job_token_queue::TtsInferenceJobTokenQueue; -use tokio::task::JoinHandle; - -// TODO: Redis calls are synchronous (but fast), but is there any way to make them async? #[derive(Deserialize)] pub struct PathInfo { twitch_username: String, } -#[derive(Deserialize)] -pub struct QueryData { - voice_token_1: Option, - voice_token_2: Option, - // Other preferences... -} - /// Sent back to the frontend websocket. #[derive(Serialize, Copy, Clone)] pub enum ResponseType { @@ -114,8 +106,8 @@ pub async fn obs_gateway_websocket_handler( info!("Building user Redis PubSub thread..."); - // TODO: Pass a timestamp-based and/or exit-based killswitch flag to the thread. - // struct AsyncThreadKillSignal { Option, Option, bool: manualKill } + let async_thread_kill_signal + = AsyncThreadKillSignal::new_with_ttl(Duration::from_secs(30)); // TODO let tts_job_token_queue = TtsInferenceJobTokenQueue::new(); @@ -123,6 +115,7 @@ pub async fn obs_gateway_websocket_handler( &twitch_user_id, &server_state.backends.redis_pubsub_connection_string, tts_job_token_queue.clone(), + async_thread_kill_signal.clone(), ).map_err(|e| { error!("Error creating pubsub thread: {:?}", e); CommonServerError::ServerError @@ -130,15 +123,17 @@ pub async fn obs_gateway_websocket_handler( info!("Starting user Redis PubSub thread..."); - let mut jh = server_state.multithreading.redis_pubsub_runtime.spawn(thread.start_thread()); + let mut join_handle = + server_state.multithreading.redis_pubsub_runtime.spawn(thread.start_thread()); - - server_state.multithreading.redis_pubsub_runtime.spawn(kill(jh)); + server_state.multithreading.redis_pubsub_runtime.spawn( + eventually_kill(join_handle, async_thread_kill_signal.clone())); let websocket = ObsGatewayWebSocket::new( tts_job_token_queue, twitch_user_id.clone(), - server_state_arc + server_state_arc, + async_thread_kill_signal ); info!("Begin Javascript WebSocket..."); @@ -150,13 +145,22 @@ pub async fn obs_gateway_websocket_handler( }) } -// NB: Well, abort() doesn't stop the thread :( -async fn kill (handle: JoinHandle<()>) { - error!("Waiting to kill pubsub thread..."); - sleep(Duration::from_secs(120)); - error!("Killing pubsub thread..."); - handle.abort(); - error!("Killed pubsub thread."); +async fn eventually_kill( + handle: JoinHandle<()>, + async_thread_kill_signal: AsyncThreadKillSignal, +) { + loop { + error!("Waiting to kill pubsub thread..."); + + if !async_thread_kill_signal.is_alive().unwrap() { + error!("Killing pubsub thread..."); + handle.abort(); + error!("Killed pubsub thread."); + return; + } + + sleep(Duration::from_secs(10)); + } } /// Websocket behavior @@ -164,6 +168,7 @@ struct ObsGatewayWebSocket { twitch_user_id: TwitchUserId, server_state: Arc, tts_job_token_queue: TtsInferenceJobTokenQueue, + async_thread_kill_signal: AsyncThreadKillSignal, } impl Actor for ObsGatewayWebSocket { @@ -180,9 +185,10 @@ impl StreamHandler> for ObsGatewayWebSock msg: Result, ctx: &mut Self::Context, ) { - //debug!("Socket Handler::handle()"); - if let Ok(msg) = msg { + // NB: Keep the Redis PubSub thread alive. + self.async_thread_kill_signal.bump_ttl().unwrap(); + //debug!("Socket Handler::handle(): msg = {:?}", msg); // TODO: Only send this every 60 seconds. @@ -269,11 +275,13 @@ impl ObsGatewayWebSocket { tts_job_token_queue: TtsInferenceJobTokenQueue, twitch_user_id: TwitchUserId, server_state: Arc, + async_thread_kill_signal: AsyncThreadKillSignal, ) -> Self { Self { tts_job_token_queue, twitch_user_id, server_state, + async_thread_kill_signal, } } diff --git a/crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs b/crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs index b319a2f070..1e3e97dc8e 100644 --- a/crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs +++ b/crates/service/websocket_gateway/src/endpoints_ws/threads/redis_pubsub_event_listener_thread.rs @@ -16,12 +16,14 @@ use std::sync::Arc; use std::thread::sleep; use time::Duration; use twitch_common::twitch_user_id::TwitchUserId; +use container_common::thread::async_thread_kill_signal::AsyncThreadKillSignal; pub struct RedisPubsubEventListenerThread { twitch_user_id: TwitchUserId, /// We'll be maintaining a thread-specific async connection to Redis. redis_pubsub_connection_string: String, token_queue: TtsInferenceJobTokenQueue, + async_thread_kill_signal: AsyncThreadKillSignal, } impl RedisPubsubEventListenerThread { @@ -29,14 +31,14 @@ impl RedisPubsubEventListenerThread { pub fn new( twitch_user_id: &TwitchUserId, redis_pubsub_connection_string: &str, - token_queue: TtsInferenceJobTokenQueue + token_queue: TtsInferenceJobTokenQueue, + async_thread_kill_signal: AsyncThreadKillSignal, ) -> AnyhowResult { - - Ok(Self { twitch_user_id: twitch_user_id.clone(), redis_pubsub_connection_string: redis_pubsub_connection_string.to_string(), token_queue, + async_thread_kill_signal, }) } @@ -68,6 +70,11 @@ impl RedisPubsubEventListenerThread { self.token_queue.enqueue_token(&tts_job_token).unwrap(); } + if !self.async_thread_kill_signal.is_alive().unwrap() { + info!("Thread has been instructed to die."); + return; + } + // TODO: Proper error handling and avoid thundering herd error!("Twitch user Redis PubSub stream ended. Restarting shortly..."); sleep(std::time::Duration::from_secs(5));