daniel543x/artcraft
1
0 Fork 0
mirror of https://github.com/storytold/artcraft.git synced 2026-11-03 08:57:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

improve readability

Browse Source
This commit is contained in:
Brandon Thomas committed 2023-10-12 01:14:03 -04:00
1 parent f1fd99be15
commit 48bf168d99
1 file changed
+126 -121
Download Patch File Download Diff File Expand all files Collapse all files
crates/service/job/inference_job/src/job/job_loop/main_loop.rs
+126 -121
View File
@@ -1,8 +1,8 @@
use std::time::Duration;
use log::{error, info, warn};
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
use errors::AnyhowResult;
use filesys::file_exists::file_exists;
use jobs_common::noop_logger::NoOpLogger;
@@ -23,22 +23,6 @@ const INCREASE_TIMEOUT_MILLIS : u64 = 1000;
/// Pause file millis
const PAUSE_FILE_EXISTS_WAIT_MILLIS : u64 = 1000 * 30;
#[derive(Eq,PartialEq)]
enum JobFailureClass {
// Jobs that can be retried
TransientFailure,
// Jobs that cannot be retried and must be marked dead
PermanentFailure,
}
#[derive(Eq,PartialEq)]
enum ContainerHealth {
// No impact to container health
Ignore,
// Increment the container health failure counter
IncrementContainerFailCount,
}
pub async fn main_loop(job_dependencies: JobDependencies) {
let mut noop_logger = NoOpLogger::new(job_dependencies.no_op_logger_millis as i64);
@@ -140,113 +124,134 @@ async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec<Availab
warn!("Success stats: {:?}", stats);
}
},
Err(e) => {
warn!("Failure to process job: {:?}", e);
let (
job_failure_class,
container_health_report,
internal_failure_reason,
maybe_public_failure_reason, // TODO(bt,2023-10-11): Remove this column in favor of "frontend_failure_category".
maybe_frontend_failure_category
) = match e {
// Permanent failures
ProcessSingleJobError::KeepAliveElapsed =>
(
JobFailureClass::PermanentFailure,
ContainerHealth::Ignore,
"keepalive elapsed".to_string(),
Some("keepalive elapsed"),
None,
),
ProcessSingleJobError::InvalidJob(ref err) =>
(
JobFailureClass::PermanentFailure,
ContainerHealth::Ignore,
format!("InvalidJob: {:?}", err),
Some("invalid job"),
None,
),
ProcessSingleJobError::NotYetImplemented =>
(
JobFailureClass::PermanentFailure,
ContainerHealth::Ignore,
"not yet implemented".to_string(),
Some("not yet implemented"),
None,
),
ProcessSingleJobError::FaceDetectionFailure =>
(
JobFailureClass::PermanentFailure,
ContainerHealth::Ignore,
"face not detected".to_string(),
Some("face not detected"),
Some(FrontendFailureCategory::FaceNotDetected),
),
// Non-permanent failures
ProcessSingleJobError::FilesystemFull =>
(
JobFailureClass::TransientFailure,
ContainerHealth::IncrementContainerFailCount,
"worker filesystem full".to_string(),
Some("worker filesystem full"),
Some(FrontendFailureCategory::RetryableWorkerError),
),
ProcessSingleJobError::Other(ref err) =>
(
JobFailureClass::TransientFailure,
ContainerHealth::IncrementContainerFailCount,
format!("OtherErr: {:?}", err),
None,
Some(FrontendFailureCategory::RetryableWorkerError),
),
};
if container_health_report == ContainerHealth::IncrementContainerFailCount {
// NB: We only increment the fail count for events that may indicate the job server is stuck.
let stats = job_dependencies.job_stats.increment_failure_count().ok();
warn!("Failure stats: {:?}", stats);
}
match job_failure_class {
JobFailureClass::PermanentFailure => {
let _r = mark_generic_inference_job_completely_failed(
&job_dependencies.mysql_pool,
&job,
maybe_public_failure_reason,
Some(&internal_failure_reason),
maybe_frontend_failure_category,
).await;
}
JobFailureClass::TransientFailure => {
let _r = mark_generic_inference_job_failure(
&job_dependencies.mysql_pool,
&job,
maybe_public_failure_reason,
&internal_failure_reason,
maybe_frontend_failure_category,
job_dependencies.job_max_attempts
).await;
}
}
match e {
// Post failure handling
ProcessSingleJobError::FilesystemFull => {
warn!("Clearing full filesystem...");
clear_full_filesystem(&job_dependencies.fs.semi_persistent_cache)?;
}
// No-op
ProcessSingleJobError::Other(_) => {}
ProcessSingleJobError::InvalidJob(_) => {}
ProcessSingleJobError::KeepAliveElapsed => {}
ProcessSingleJobError::NotYetImplemented => {}
ProcessSingleJobError::FaceDetectionFailure => {}
}
Err(err) => {
warn!("Failure to process job: {:?}", err);
let _r = handle_error(&job_dependencies, &job, err).await?;
}
}
}
Ok(())
}
#[derive(Eq,PartialEq)]
enum JobFailureClass {
// Jobs that can be retried
TransientFailure,
// Jobs that cannot be retried and must be marked dead
PermanentFailure,
}
#[derive(Eq,PartialEq)]
enum ContainerHealth {
// No impact to container health
Ignore,
// Increment the container health failure counter
IncrementContainerFailCount,
}
async fn handle_error(job_dependencies: &&JobDependencies, job: &AvailableInferenceJob, error: ProcessSingleJobError) -> AnyhowResult<()> {
let (
job_failure_class,
container_health_report,
internal_failure_reason,
maybe_public_failure_reason, // TODO(bt,2023-10-11): Remove this column in favor of "frontend_failure_category".
maybe_frontend_failure_category
) = match error {
// Permanent failures
ProcessSingleJobError::KeepAliveElapsed =>
(
JobFailureClass::PermanentFailure,
ContainerHealth::Ignore,
"keepalive elapsed".to_string(),
Some("keepalive elapsed"),
None,
),
ProcessSingleJobError::InvalidJob(ref err) =>
(
JobFailureClass::PermanentFailure,
ContainerHealth::Ignore,
format!("InvalidJob: {:?}", err),
Some("invalid job"),
None,
),
ProcessSingleJobError::NotYetImplemented =>
(
JobFailureClass::PermanentFailure,
ContainerHealth::Ignore,
"not yet implemented".to_string(),
Some("not yet implemented"),
None,
),
ProcessSingleJobError::FaceDetectionFailure =>
(
JobFailureClass::PermanentFailure,
ContainerHealth::Ignore,
"face not detected".to_string(),
Some("face not detected"),
Some(FrontendFailureCategory::FaceNotDetected),
),
// Non-permanent failures
ProcessSingleJobError::FilesystemFull =>
(
JobFailureClass::TransientFailure,
ContainerHealth::IncrementContainerFailCount,
"worker filesystem full".to_string(),
Some("worker filesystem full"),
Some(FrontendFailureCategory::RetryableWorkerError),
),
ProcessSingleJobError::Other(ref err) =>
(
JobFailureClass::TransientFailure,
ContainerHealth::IncrementContainerFailCount,
format!("OtherErr: {:?}", err),
None,
Some(FrontendFailureCategory::RetryableWorkerError),
),
};
if container_health_report == ContainerHealth::IncrementContainerFailCount {
// NB: We only increment the fail count for events that may indicate the job server is stuck.
let stats = job_dependencies.job_stats.increment_failure_count().ok();
warn!("Failure stats: {:?}", stats);
}
match job_failure_class {
JobFailureClass::PermanentFailure => {
let _r = mark_generic_inference_job_completely_failed(
&job_dependencies.mysql_pool,
&job,
maybe_public_failure_reason,
Some(&internal_failure_reason),
maybe_frontend_failure_category,
).await;
}
JobFailureClass::TransientFailure => {
let _r = mark_generic_inference_job_failure(
&job_dependencies.mysql_pool,
&job,
maybe_public_failure_reason,
&internal_failure_reason,
maybe_frontend_failure_category,
job_dependencies.job_max_attempts
).await;
}
}
match error {
// Post failure handling
ProcessSingleJobError::FilesystemFull => {
warn!("Clearing full filesystem...");
clear_full_filesystem(&job_dependencies.fs.semi_persistent_cache)?;
}
// No-op
ProcessSingleJobError::Other(_) => {}
ProcessSingleJobError::InvalidJob(_) => {}
ProcessSingleJobError::KeepAliveElapsed => {}
ProcessSingleJobError::NotYetImplemented => {}
ProcessSingleJobError::FaceDetectionFailure => {}
}
Ok(())
}