parent
b75871e25d
commit
2816e72cbd
19 changed files with 301 additions and 265 deletions
@ -0,0 +1,14 @@ |
|||||||
|
use serde::Serialize; |
||||||
|
|
||||||
|
use crate::{messaging::AsMsg, scheduler::EntryStat, types::Id, ufs::IndexFileMeta}; |
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize)] |
||||||
|
pub struct Stats { |
||||||
|
pub agent_id: Id, |
||||||
|
pub running_jobs: Vec<String>, |
||||||
|
pub scheduled_jobs: Vec<EntryStat>, |
||||||
|
pub cached_jobs: Vec<Id>, |
||||||
|
pub cached_payloads: Vec<IndexFileMeta>, |
||||||
|
} |
||||||
|
|
||||||
|
impl AsMsg for Stats {} |
||||||
@ -1,85 +0,0 @@ |
|||||||
use std::time::Duration; |
|
||||||
|
|
||||||
use crate::jobs::AnonymousJobBatch; |
|
||||||
|
|
||||||
use super::entry::{Entry, EntryType}; |
|
||||||
use chrono::Utc; |
|
||||||
use tokio::time::timeout; |
|
||||||
|
|
||||||
#[async_trait] |
|
||||||
pub trait Spawner { |
|
||||||
async fn spawn(self); |
|
||||||
} |
|
||||||
|
|
||||||
/// Spawn jobs in scheduler thru `tokio::spawn`
|
|
||||||
pub struct CommonSpawner<'e> { |
|
||||||
pub entries: &'e mut [Entry], |
|
||||||
} |
|
||||||
|
|
||||||
#[async_trait] |
|
||||||
impl Spawner for CommonSpawner<'_> { |
|
||||||
async fn spawn(self) { |
|
||||||
for entry in self |
|
||||||
.entries |
|
||||||
.iter_mut() |
|
||||||
.filter(|e| matches!(e.runnable, EntryType::Common(_))) |
|
||||||
{ |
|
||||||
if let EntryType::Common(runnable) = &entry.runnable { |
|
||||||
match entry.next.as_ref() { |
|
||||||
Some(next) => { |
|
||||||
let cancel_timeout = |
|
||||||
next.timestamp_millis() - Utc::now().timestamp_millis(); |
|
||||||
let cancel_timeout = Duration::from_millis(cancel_timeout as u64); |
|
||||||
|
|
||||||
tokio::spawn(timeout(cancel_timeout, runnable.call())); |
|
||||||
|
|
||||||
entry.set_next_run_time(); |
|
||||||
} |
|
||||||
None => { |
|
||||||
error!("not implemented yet"); |
|
||||||
todo!(); |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
if entry.next.as_ref().unwrap().gt(&Utc::now()) { |
|
||||||
break; |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
/// Spawn jobs in scheduler thru `URunner`
|
|
||||||
pub struct URunnerSpawner<'e> { |
|
||||||
pub entries: &'e mut [Entry], |
|
||||||
} |
|
||||||
|
|
||||||
#[async_trait] |
|
||||||
impl Spawner for URunnerSpawner<'_> { |
|
||||||
async fn spawn(self) { |
|
||||||
let mut job_batch = vec![]; |
|
||||||
for entry in self |
|
||||||
.entries |
|
||||||
.iter_mut() |
|
||||||
.filter(|e| matches!(e.runnable, EntryType::URunner(_))) |
|
||||||
{ |
|
||||||
if let EntryType::URunner(runnable) = &entry.runnable { |
|
||||||
match entry.next.as_ref() { |
|
||||||
Some(_) => { |
|
||||||
job_batch.push(runnable.clone()); |
|
||||||
entry.set_next_run_time(); |
|
||||||
} |
|
||||||
None => { |
|
||||||
error!("not implemented yet"); |
|
||||||
todo!(); |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
if entry.next.as_ref().unwrap().gt(&Utc::now()) { |
|
||||||
break; |
|
||||||
} |
|
||||||
} |
|
||||||
AnonymousJobBatch::from_prepared_jobs(job_batch) |
|
||||||
.spawn() |
|
||||||
.await; |
|
||||||
} |
|
||||||
} |
|
||||||
Loading…
Reference in new issue