mirrors/background-jobs
1
0
Fork 0
mirror of https://git.asonix.dog/asonix/background-jobs.git synced 2025-02-17 13:55:18 +00:00
Code Issues Projects Releases Packages Wiki Activity

actix: time running jobs

Browse source
This commit is contained in:
Aode (lion) 2021-11-17 12:33:15 -06:00
parent aa9b89e33a
commit a3485a1e3e
2 changed files with 31 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
jobs-actix/Cargo.toml
View file

@ -21,5 +21,5 @@ num_cpus = "1.10.0"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
thiserror = "1.0"
tokio = { version = "1", default-features = false, features = ["sync"] }
tokio = { version = "1", default-features = false, features = ["macros", "sync"] }
uuid = { version ="0.8.1", features = ["v4", "serde"] }

31
jobs-actix/src/worker.rs
View file

@ -2,6 +2,7 @@ use crate::Server;
use actix_rt::{spawn, Arbiter};
use background_jobs_core::{CachedProcessorMap, JobInfo};
use log::{debug, error, info, warn};
use std::future::Future;
use tokio::sync::mpsc::{channel, Sender};
use uuid::Uuid;
@ -76,6 +77,33 @@ impl Drop for WarnOnDrop {
}
}
async fn time_job<F: Future + Unpin>(mut future: F, job_id: Uuid) -> <F as Future>::Output {
let mut interval = actix_rt::time::interval(std::time::Duration::from_secs(5));
interval.tick().await;
let mut count = 0;
loop {
tokio::select! {
output = &mut future => { break output },
_ = interval.tick() => {
count += 5;
if count > 60 * 60 {
if count % (60 * 20) == 0 {
warn!("Job {} is taking a long time: {} hours", job_id, count / 60 / 60);
}
} else if count >= 60 {
if count % 20 == 0 {
info!("Job {} is taking a long time: {} minutes", job_id, count / 60);
}
} else {
info!("Job {} is taking a long time: {} seconds", job_id, count);
}
},
};
}
}
pub(crate) fn local_worker<State>(
queue: String,
processors: CachedProcessorMap<State>,
@ -103,7 +131,8 @@ pub(crate) fn local_worker<State>(
return;
}
while let Some(job) = rx.recv().await {
let return_job = processors.process(job).await;
let id = job.id();
let return_job = time_job(Box::pin(processors.process(job)), id).await;
if let Err(e) = server.return_job(return_job).await {
error!("Error returning job, {}", e);