From bf9bc013df6e1b75fb40fbdbea88d8ed0aa72906 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Thu, 8 Sep 2022 14:56:35 -0400 Subject: [PATCH] New dynamic job progress reporter with no-op version --- .../job_progress_reporter.rs | 24 +++++++ .../src/job_progress_reporter/mod.rs | 3 + .../noop_job_progress_reporter.rs | 42 +++++++++++ .../redis_job_progress_reporter.rs | 70 +++++++++++++++++++ crates/lib/jobs_common/src/lib.rs | 1 + .../src/redis_job_status_logger.rs | 8 +++ 6 files changed, 148 insertions(+) create mode 100644 crates/lib/jobs_common/src/job_progress_reporter/job_progress_reporter.rs create mode 100644 crates/lib/jobs_common/src/job_progress_reporter/mod.rs create mode 100644 crates/lib/jobs_common/src/job_progress_reporter/noop_job_progress_reporter.rs create mode 100644 crates/lib/jobs_common/src/job_progress_reporter/redis_job_progress_reporter.rs diff --git a/crates/lib/jobs_common/src/job_progress_reporter/job_progress_reporter.rs b/crates/lib/jobs_common/src/job_progress_reporter/job_progress_reporter.rs new file mode 100644 index 0000000000..79edfea991 --- /dev/null +++ b/crates/lib/jobs_common/src/job_progress_reporter/job_progress_reporter.rs @@ -0,0 +1,24 @@ +use container_common::anyhow_result::AnyhowResult; + +pub trait JobProgressReporterBuilder { + // NB: Still learning "Box" vs "impl Trait"; + // The former is dynamic dispatch, dynamic size, vtable-dispatched. The latter is static dispatch, + // Sized, and known at compile time: + // https://users.rust-lang.org/t/difference-between-returning-dyn-box-trait-and-impl-trait/57640/3 + + fn new_generic_download(&self, job_token: &str) -> AnyhowResult>; + + fn new_tts_download(&self, tts_job_token: &str) -> AnyhowResult>; + + fn new_tts_inference(&self, tts_job_token: &str) -> AnyhowResult>; + + fn new_w2l_download(&self, w2l_job_token: &str) -> AnyhowResult>; + + fn new_w2l_inference(&self, w2l_job_token: &str) -> AnyhowResult>; +} + +pub trait JobProgressReporter { + /// Record the job progress "perhaps somewhere" so the frontend can see what's going on. + /// Occasionally you may want to replace this with a no-op. + fn log_status(&mut self, logging_details: &str) -> AnyhowResult<()>; +} \ No newline at end of file diff --git a/crates/lib/jobs_common/src/job_progress_reporter/mod.rs b/crates/lib/jobs_common/src/job_progress_reporter/mod.rs new file mode 100644 index 0000000000..79f5d158d4 --- /dev/null +++ b/crates/lib/jobs_common/src/job_progress_reporter/mod.rs @@ -0,0 +1,3 @@ +pub mod job_progress_reporter; +pub mod noop_job_progress_reporter; +pub mod redis_job_progress_reporter; diff --git a/crates/lib/jobs_common/src/job_progress_reporter/noop_job_progress_reporter.rs b/crates/lib/jobs_common/src/job_progress_reporter/noop_job_progress_reporter.rs new file mode 100644 index 0000000000..8dc3c6f3f0 --- /dev/null +++ b/crates/lib/jobs_common/src/job_progress_reporter/noop_job_progress_reporter.rs @@ -0,0 +1,42 @@ +use container_common::anyhow_result::AnyhowResult; +use crate::job_progress_reporter::job_progress_reporter::{JobProgressReporter, JobProgressReporterBuilder}; + +/// A job progress reporter that does no work and has zero database dependencies. +pub struct NoOpJobProgressReporterBuilder {} + +/// A job progress reporter that does no work and has zero database dependencies. +pub struct NoOpJobProgressReporter {} + +impl NoOpJobProgressReporterBuilder { + fn create_instance() -> AnyhowResult> { + Ok(Box::new(NoOpJobProgressReporter {})) + } +} + +impl JobProgressReporterBuilder for NoOpJobProgressReporterBuilder { + fn new_generic_download(&self, job_token: &str) -> AnyhowResult> { + NoOpJobProgressReporterBuilder::create_instance() + } + + fn new_tts_download(&self, tts_job_token: &str) -> AnyhowResult> { + NoOpJobProgressReporterBuilder::create_instance() + } + + fn new_tts_inference(&self, tts_job_token: &str) -> AnyhowResult> { + NoOpJobProgressReporterBuilder::create_instance() + } + + fn new_w2l_download(&self, w2l_job_token: &str) -> AnyhowResult> { + NoOpJobProgressReporterBuilder::create_instance() + } + + fn new_w2l_inference(&self, w2l_job_token: &str) -> AnyhowResult> { + NoOpJobProgressReporterBuilder::create_instance() + } +} + +impl JobProgressReporter for NoOpJobProgressReporter { + fn log_status(&mut self, logging_details: &str) -> AnyhowResult<()> { + Ok(()) // No-op + } +} diff --git a/crates/lib/jobs_common/src/job_progress_reporter/redis_job_progress_reporter.rs b/crates/lib/jobs_common/src/job_progress_reporter/redis_job_progress_reporter.rs new file mode 100644 index 0000000000..f1c312dfbb --- /dev/null +++ b/crates/lib/jobs_common/src/job_progress_reporter/redis_job_progress_reporter.rs @@ -0,0 +1,70 @@ +use container_common::anyhow_result::AnyhowResult; +use crate::job_progress_reporter::job_progress_reporter::{JobProgressReporter, JobProgressReporterBuilder}; +use r2d2_redis::r2d2::PooledConnection; +use r2d2_redis::redis::Commands; +use r2d2_redis::{r2d2, RedisConnectionManager}; +use redis_common::redis_keys::RedisKeys; + +/// A job progress reporter that reports to Redis +pub struct RedisJobProgressReporterBuilder { + redis_pool: r2d2::Pool, +} + +/// A job progress reporter that reports to Redis +pub struct RedisJobProgressReporter { + redis: PooledConnection, + redis_key: String, +} + +impl RedisJobProgressReporterBuilder { + /// Create a new instance. The backing Redis pool is Sync/Send behind an Arc. + fn create_instance(redis_pool: r2d2::Pool, redis_key: String) -> AnyhowResult> { + let redis = redis_pool.get()?; + + Ok(Box::new(RedisJobProgressReporter { + redis, + redis_key, + })) + } +} + +impl RedisJobProgressReporter { + pub const STATUS_KEY_TTL_SECONDS : usize = 60 * 60; +} + +impl JobProgressReporterBuilder for RedisJobProgressReporterBuilder { + fn new_generic_download(&self, job_token: &str) -> AnyhowResult> { + let redis_key = RedisKeys::generic_download_extra_status_info(job_token); + RedisJobProgressReporterBuilder::create_instance(self.redis_pool.clone(), redis_key) + } + + fn new_tts_download(&self, tts_job_token: &str) -> AnyhowResult> { + let redis_key = RedisKeys::tts_download_extra_status_info(tts_job_token); + RedisJobProgressReporterBuilder::create_instance(self.redis_pool.clone(), redis_key) + } + + fn new_tts_inference(&self, tts_job_token: &str) -> AnyhowResult> { + let redis_key = RedisKeys::tts_inference_extra_status_info(tts_job_token); + RedisJobProgressReporterBuilder::create_instance(self.redis_pool.clone(), redis_key) + } + + fn new_w2l_download(&self, w2l_job_token: &str) -> AnyhowResult> { + let redis_key = RedisKeys::w2l_download_extra_status_info(w2l_job_token); + RedisJobProgressReporterBuilder::create_instance(self.redis_pool.clone(), redis_key) + } + + fn new_w2l_inference(&self, w2l_job_token: &str) -> AnyhowResult> { + let redis_key = RedisKeys::w2l_inference_extra_status_info(w2l_job_token); + RedisJobProgressReporterBuilder::create_instance(self.redis_pool.clone(), redis_key) + } +} + +impl JobProgressReporter for RedisJobProgressReporter { + fn log_status(&mut self, logging_details: &str) -> AnyhowResult<()> { + let _r : String = self.redis // NB: Compiler can't figure out the throwaway result type + .set_ex(&self.redis_key, + logging_details, + Self::STATUS_KEY_TTL_SECONDS)?; + Ok(()) // No-op + } +} diff --git a/crates/lib/jobs_common/src/lib.rs b/crates/lib/jobs_common/src/lib.rs index 899482f22d..1e4864a7f5 100644 --- a/crates/lib/jobs_common/src/lib.rs +++ b/crates/lib/jobs_common/src/lib.rs @@ -1,5 +1,6 @@ //! This crate is simply meant to collect common shared legacy "jobs" components //! It's an artifact of refactoring and maybe should mostly disappear. +pub mod job_progress_reporter; pub mod noop_logger; pub mod redis_job_status_logger; pub mod semi_persistent_cache_dir; \ No newline at end of file diff --git a/crates/lib/jobs_common/src/redis_job_status_logger.rs b/crates/lib/jobs_common/src/redis_job_status_logger.rs index 89b9508824..6e58ea3385 100644 --- a/crates/lib/jobs_common/src/redis_job_status_logger.rs +++ b/crates/lib/jobs_common/src/redis_job_status_logger.rs @@ -4,8 +4,10 @@ use r2d2_redis::r2d2::PooledConnection; use r2d2_redis::redis::Commands; use redis_common::redis_keys::RedisKeys; +#[deprecated(note="Use JobProgressReporterBuilder")] pub type RedisPool = PooledConnection; +#[deprecated(note="Use JobProgressReporterBuilder")] pub struct RedisJobStatusLogger <'a> { redis: &'a mut RedisPool, status_key: String, @@ -14,32 +16,38 @@ pub struct RedisJobStatusLogger <'a> { impl <'a> RedisJobStatusLogger <'a> { pub const STATUS_KEY_TTL_SECONDS : usize = 60 * 60; + #[deprecated(note="Use JobProgressReporterBuilder")] pub fn new_tts_download(redis: &'a mut RedisPool, tts_job_token: &str) -> Self { let status_key = RedisKeys::tts_download_extra_status_info(tts_job_token); Self { redis, status_key } } + #[deprecated(note="Use JobProgressReporterBuilder")] pub fn new_tts_inference(redis: &'a mut RedisPool, tts_job_token: &str) -> Self { let status_key = RedisKeys::tts_inference_extra_status_info(tts_job_token); Self { redis, status_key } } + #[deprecated(note="Use JobProgressReporterBuilder")] pub fn new_generic_download(redis: &'a mut RedisPool, job_token: &str) -> Self { let status_key = RedisKeys::generic_download_extra_status_info(job_token); Self { redis, status_key } } + #[deprecated(note="Use JobProgressReporterBuilder")] pub fn new_w2l_download(redis: &'a mut RedisPool, w2l_job_token: &str) -> Self { let status_key = RedisKeys::w2l_download_extra_status_info(w2l_job_token); Self { redis, status_key } } + #[deprecated(note="Use JobProgressReporterBuilder")] pub fn new_w2l_inference(redis: &'a mut RedisPool, w2l_job_token: &str) -> Self { let status_key = RedisKeys::w2l_inference_extra_status_info(w2l_job_token); Self { redis, status_key } } /// Record the status to Redis so the frontend has the latest state. + #[deprecated(note="Use JobProgressReporterBuilder")] pub fn log_status(&mut self, logging_details: &str) -> AnyhowResult<()> { let _r : String = self.redis // NB: Compiler can't figure out the throwaway result type .set_ex(&self.status_key,