fix: bound run projection work #43
+20
-6
@@ -1,39 +1,53 @@
|
||||
use std::collections::HashMap;
|
||||
use std::time::Duration;
|
||||
|
||||
use syncode_control_node::ProjectionClient;
|
||||
use syncode_control_runs::{ProjectedRun, RunId, RunLog, Runs, RunsError};
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum ProjectionError {
|
||||
#[error("cannot read the run projection: {0}")]
|
||||
Runs(#[from] RunsError),
|
||||
}
|
||||
|
||||
pub async fn serve<L: RunLog>(
|
||||
runs: Runs<L>,
|
||||
client: ProjectionClient,
|
||||
min_run_number: u64,
|
||||
) -> Result<(), ProjectionError> {
|
||||
let mut delivered = HashMap::<RunId, ProjectedRun>::new();
|
||||
loop {
|
||||
for projection in runs.projections().await? {
|
||||
if projection.number.get() < min_run_number {
|
||||
continue;
|
||||
}
|
||||
if delivered.get(&projection.run) == Some(&projection) {
|
||||
continue;
|
||||
}
|
||||
match client.send(&projection).await {
|
||||
Ok(()) => {
|
||||
delivered.insert(projection.run, projection);
|
||||
}
|
||||
Err(error) => {
|
||||
eprintln!("cannot project run state: {error}");
|
||||
}
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
}
|
||||
use std::collections::HashMap;
|
||||
use std::time::Duration;
|
||||
|
||||
use syncode_control_node::ProjectionClient;
|
||||
use syncode_control_runs::{ProjectedRun, RunId, RunLog, Runs, RunsError};
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum ProjectionError {
|
||||
#[error("cannot read the run projection: {0}")]
|
||||
Runs(#[from] RunsError),
|
||||
#[error("pending projection for run {0} is missing")]
|
||||
MissingPending(RunId),
|
||||
}
|
||||
|
||||
pub async fn serve<L: RunLog>(
|
||||
runs: Runs<L>,
|
||||
client: ProjectionClient,
|
||||
min_run_number: u64,
|
||||
) -> Result<(), ProjectionError> {
|
||||
let mut delivered = HashMap::<RunId, (u64, u64)>::new();
|
||||
let mut pending = HashMap::<RunId, ((u64, u64), ProjectedRun)>::new();
|
||||
loop {
|
||||
for (run, number, sequence, output_version) in runs.projection_versions().await {
|
||||
if number.get() < min_run_number {
|
||||
continue;
|
||||
}
|
||||
let version = (sequence, output_version);
|
||||
if delivered.get(&run) == Some(&version) {
|
||||
continue;
|
||||
}
|
||||
if pending.get(&run).map(|(version, _)| *version) != Some(version) {
|
||||
let Some(projection) = runs.projection(run).await? else {
|
||||
continue;
|
||||
};
|
||||
pending.insert(run, (version, projection));
|
||||
}
|
||||
let (_, projection) = pending
|
||||
.get(&run)
|
||||
.ok_or(ProjectionError::MissingPending(run))?;
|
||||
match client.send(projection).await {
|
||||
Ok(()) => {
|
||||
delivered.insert(run, version);
|
||||
pending.remove(&run);
|
||||
}
|
||||
Err(error) => {
|
||||
eprintln!("cannot project run state: {error}");
|
||||
}
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
}
|
||||
@@ -1,67 +1,110 @@
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use syncode_control_runs::{Forgotten, JobId, NodeId, Origin, ProjectedJobState, RunId, Runs};
|
||||
|
||||
type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
|
||||
|
||||
#[tokio::test]
|
||||
async fn projection_carries_the_current_runtime_authority() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let job = JobId::fresh();
|
||||
let origin = Origin::new(
|
||||
"syncode/meta".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned()));
|
||||
runs.queue(job, origin, b"plan".to_vec()).await?;
|
||||
let node = NodeId::fresh();
|
||||
let assignment = runs.take_next(node).await?.expect("assignment");
|
||||
|
||||
let projection = runs.projections().await?.pop().expect("projection");
|
||||
let projected = projection.jobs.first().expect("projected job");
|
||||
assert_eq!(projected.state, ProjectedJobState::Assigned);
|
||||
assert_eq!(projected.node, Some(node));
|
||||
assert_eq!(projected.fence, Some(assignment.fence()));
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn one_run_can_be_projected_without_sweeping_every_run() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let origin = Origin::new(
|
||||
"syncode/meta".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned()));
|
||||
let run = runs.queue(JobId::fresh(), origin, b"plan".to_vec()).await?;
|
||||
|
||||
let alone = runs.projection(run).await?.expect("projection");
|
||||
let swept = runs.projections().await?;
|
||||
assert_eq!(
|
||||
swept.len(),
|
||||
1,
|
||||
"the run added here is the only one to sweep"
|
||||
);
|
||||
assert_eq!(
|
||||
alone, swept[0],
|
||||
"one run's projection must match the sweep's"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_run_nobody_queued_has_nothing_to_project() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
assert_eq!(runs.projection(RunId::fresh()).await?, None);
|
||||
Ok(())
|
||||
}
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use syncode_control_runs::{Forgotten, JobId, NodeId, Origin, ProjectedJobState, RunId, Runs};
|
||||
|
||||
type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
|
||||
|
||||
#[tokio::test]
|
||||
async fn projection_carries_the_current_runtime_authority() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let job = JobId::fresh();
|
||||
let origin = Origin::new(
|
||||
"syncode/meta".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned()));
|
||||
runs.queue(job, origin, b"plan".to_vec()).await?;
|
||||
let node = NodeId::fresh();
|
||||
let assignment = runs.take_next(node).await?.expect("assignment");
|
||||
|
||||
let projection = runs.projections().await?.pop().expect("projection");
|
||||
let projected = projection.jobs.first().expect("projected job");
|
||||
assert_eq!(projected.state, ProjectedJobState::Assigned);
|
||||
assert_eq!(projected.node, Some(node));
|
||||
assert_eq!(projected.fence, Some(assignment.fence()));
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn one_run_can_be_projected_without_sweeping_every_run() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let origin = Origin::new(
|
||||
"syncode/meta".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned()));
|
||||
let run = runs.queue(JobId::fresh(), origin, b"plan".to_vec()).await?;
|
||||
|
||||
let alone = runs.projection(run).await?.expect("projection");
|
||||
let swept = runs.projections().await?;
|
||||
assert_eq!(
|
||||
swept.len(),
|
||||
1,
|
||||
"the run added here is the only one to sweep"
|
||||
);
|
||||
assert_eq!(
|
||||
alone, swept[0],
|
||||
"one run's projection must match the sweep's"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_run_nobody_queued_has_nothing_to_project() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
assert_eq!(runs.projection(RunId::fresh()).await?, None);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn projection_versions_change_without_building_every_projection() -> TestResult {
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let origin = Origin::new(
|
||||
"syncode/meta".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned()));
|
||||
let run = runs.queue(JobId::fresh(), origin, b"plan".to_vec()).await?;
|
||||
|
||||
let initial = runs.projection_versions().await;
|
||||
let initial_sequence = initial[0].2;
|
||||
let initial_output = initial[0].3;
|
||||
assert_eq!(initial[0].0, run);
|
||||
|
||||
let node = NodeId::fresh();
|
||||
let assignment = runs.take_next(node).await?.expect("assignment");
|
||||
|
||||
let assigned = runs.projection_versions().await;
|
||||
assert_eq!(assigned[0].0, run);
|
||||
assert!(assigned[0].2 > initial_sequence);
|
||||
assert_eq!(assigned[0].3, initial_output);
|
||||
|
||||
runs.logged(
|
||||
run,
|
||||
assignment.job(),
|
||||
node,
|
||||
assignment.fence(),
|
||||
0,
|
||||
&["line".to_owned()],
|
||||
)
|
||||
.await?;
|
||||
|
||||
let logged = runs.projection_versions().await;
|
||||
assert_eq!(logged[0].2, assigned[0].2);
|
||||
assert!(logged[0].3 > assigned[0].3);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,35 +1,37 @@
|
||||
//! What a job printed.
|
||||
//!
|
||||
//! Output is not a state change: a job that prints nothing is not less valid,
|
||||
//! and a line arriving late does not move the run anywhere. So logs travel
|
||||
//! beside the aggregate rather than through it.
|
||||
|
||||
use crate::{Fence, JobId, NodeId, RunId, RunLog, Runs, RunsError};
|
||||
|
||||
impl<L: RunLog> Runs<L> {
|
||||
/// Keep what a node said a job printed. The offset makes a repeated batch
|
||||
/// the same fact twice rather than new output.
|
||||
pub async fn logged(
|
||||
&self,
|
||||
run: RunId,
|
||||
job: JobId,
|
||||
node: NodeId,
|
||||
fence: Fence,
|
||||
offset: u64,
|
||||
lines: &[String],
|
||||
) -> Result<(), RunsError> {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.run(run)?
|
||||
.validate_holder(job, node, fence)?;
|
||||
self.store
|
||||
.logged(run, job, offset, lines)
|
||||
.await
|
||||
.map_err(RunsError::log)
|
||||
}
|
||||
|
||||
pub async fn lines(&self, run: RunId, job: JobId) -> Result<Vec<String>, RunsError> {
|
||||
self.store.lines(run, job).await.map_err(RunsError::log)
|
||||
}
|
||||
}
|
||||
//! What a job printed.
|
||||
//!
|
||||
//! Output is not a state change: a job that prints nothing is not less valid,
|
||||
//! and a line arriving late does not move the run anywhere. So logs travel
|
||||
//! beside the aggregate rather than through it.
|
||||
|
||||
use crate::{Fence, JobId, NodeId, RunId, RunLog, Runs, RunsError};
|
||||
|
||||
impl<L: RunLog> Runs<L> {
|
||||
/// Keep what a node said a job printed. The offset makes a repeated batch
|
||||
/// the same fact twice rather than new output.
|
||||
pub async fn logged(
|
||||
&self,
|
||||
run: RunId,
|
||||
job: JobId,
|
||||
node: NodeId,
|
||||
fence: Fence,
|
||||
offset: u64,
|
||||
lines: &[String],
|
||||
) -> Result<(), RunsError> {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.run(run)?
|
||||
.validate_holder(job, node, fence)?;
|
||||
self.store
|
||||
.logged(run, job, offset, lines)
|
||||
.await
|
||||
.map_err(RunsError::log)?;
|
||||
self.state.lock().await.record_output(run);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn lines(&self, run: RunId, job: JobId) -> Result<Vec<String>, RunsError> {
|
||||
self.store.lines(run, job).await.map_err(RunsError::log)
|
||||
}
|
||||
}
|
||||
@@ -1,53 +1,57 @@
|
||||
use crate::{ProjectedJob, ProjectedRun, RunId, RunLog};
|
||||
|
||||
use super::state::ProjectionSeed;
|
||||
use super::{Runs, RunsError};
|
||||
|
||||
impl<L: RunLog> Runs<L> {
|
||||
pub async fn projections(&self) -> Result<Vec<ProjectedRun>, RunsError> {
|
||||
let seeds = self.state.lock().await.projection_seeds();
|
||||
let mut projections = Vec::with_capacity(seeds.len());
|
||||
for seed in seeds {
|
||||
projections.push(self.project(seed).await?);
|
||||
}
|
||||
Ok(projections)
|
||||
}
|
||||
|
||||
/// The current projection of one run, if it has anything to project. Lets
|
||||
/// a caller push a single run's state out-of-band from the periodic sweep
|
||||
/// — namely dispatch, which needs the projection to have landed before it
|
||||
/// hands a node a token that is only valid once it has.
|
||||
pub async fn projection(&self, run: RunId) -> Result<Option<ProjectedRun>, RunsError> {
|
||||
let Some(seed) = self.state.lock().await.projection_seed(run) else {
|
||||
return Ok(None);
|
||||
};
|
||||
Ok(Some(self.project(seed).await?))
|
||||
}
|
||||
|
||||
async fn project(&self, seed: ProjectionSeed) -> Result<ProjectedRun, RunsError> {
|
||||
let mut jobs = Vec::with_capacity(seed.jobs.len());
|
||||
for source in seed.jobs {
|
||||
let logs = self
|
||||
.store
|
||||
.lines(seed.run, source.job)
|
||||
.await
|
||||
.map_err(RunsError::log)?;
|
||||
jobs.push(ProjectedJob {
|
||||
job: source.job,
|
||||
key: source.key,
|
||||
needs: source.needs,
|
||||
state: source.state,
|
||||
node: source.node,
|
||||
fence: source.fence,
|
||||
logs,
|
||||
});
|
||||
}
|
||||
Ok(ProjectedRun {
|
||||
run: seed.run,
|
||||
number: seed.number,
|
||||
origin: seed.origin,
|
||||
sequence: seed.sequence,
|
||||
jobs,
|
||||
})
|
||||
}
|
||||
}
|
||||
use crate::{ProjectedJob, ProjectedRun, RunId, RunLog};
|
||||
|
||||
use super::state::ProjectionSeed;
|
||||
use super::{Runs, RunsError};
|
||||
|
||||
impl<L: RunLog> Runs<L> {
|
||||
pub async fn projection_versions(&self) -> Vec<(RunId, crate::RunNumber, u64, u64)> {
|
||||
self.state.lock().await.projection_versions()
|
||||
}
|
||||
|
||||
pub async fn projections(&self) -> Result<Vec<ProjectedRun>, RunsError> {
|
||||
let seeds = self.state.lock().await.projection_seeds();
|
||||
let mut projections = Vec::with_capacity(seeds.len());
|
||||
for seed in seeds {
|
||||
projections.push(self.project(seed).await?);
|
||||
}
|
||||
Ok(projections)
|
||||
}
|
||||
|
||||
/// The current projection of one run, if it has anything to project. Lets
|
||||
/// a caller push a single run's state out-of-band from the periodic sweep
|
||||
/// — namely dispatch, which needs the projection to have landed before it
|
||||
/// hands a node a token that is only valid once it has.
|
||||
pub async fn projection(&self, run: RunId) -> Result<Option<ProjectedRun>, RunsError> {
|
||||
let Some(seed) = self.state.lock().await.projection_seed(run) else {
|
||||
return Ok(None);
|
||||
};
|
||||
Ok(Some(self.project(seed).await?))
|
||||
}
|
||||
|
||||
async fn project(&self, seed: ProjectionSeed) -> Result<ProjectedRun, RunsError> {
|
||||
let mut jobs = Vec::with_capacity(seed.jobs.len());
|
||||
for source in seed.jobs {
|
||||
let logs = self
|
||||
.store
|
||||
.lines(seed.run, source.job)
|
||||
.await
|
||||
.map_err(RunsError::log)?;
|
||||
jobs.push(ProjectedJob {
|
||||
job: source.job,
|
||||
key: source.key,
|
||||
needs: source.needs,
|
||||
state: source.state,
|
||||
node: source.node,
|
||||
fence: source.fence,
|
||||
logs,
|
||||
});
|
||||
}
|
||||
Ok(ProjectedRun {
|
||||
run: seed.run,
|
||||
number: seed.number,
|
||||
origin: seed.origin,
|
||||
sequence: seed.sequence,
|
||||
jobs,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,239 +1,190 @@
|
||||
mod diagnosis;
|
||||
mod scheduling;
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use crate::Fence;
|
||||
use crate::{
|
||||
Assignment, JobId, JobState, Origin, ProjectedJobState, Run, RunId, RunNumber, RunsError,
|
||||
SchedulerPolicy,
|
||||
};
|
||||
use crate::{Lease, NodeId};
|
||||
|
||||
pub struct Entry {
|
||||
run: Run,
|
||||
number: RunNumber,
|
||||
origin: Origin,
|
||||
}
|
||||
|
||||
pub struct State {
|
||||
entries: Vec<Entry>,
|
||||
organization_turns: BTreeMap<String, u64>,
|
||||
next_turn: u64,
|
||||
}
|
||||
|
||||
pub(super) struct ProjectionSeed {
|
||||
pub run: RunId,
|
||||
pub number: RunNumber,
|
||||
pub origin: Origin,
|
||||
pub sequence: u64,
|
||||
pub jobs: Vec<ProjectionJobSeed>,
|
||||
}
|
||||
|
||||
pub(super) struct ProjectionJobSeed {
|
||||
pub job: JobId,
|
||||
pub key: String,
|
||||
pub needs: Vec<String>,
|
||||
pub state: ProjectedJobState,
|
||||
pub node: Option<NodeId>,
|
||||
pub fence: Option<Fence>,
|
||||
}
|
||||
|
||||
impl Default for State {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
entries: Vec::new(),
|
||||
organization_turns: BTreeMap::new(),
|
||||
next_turn: 1,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl State {
|
||||
pub fn admit(&mut self, run: Run, number: RunNumber, origin: Origin) {
|
||||
self.entries.push(Entry {
|
||||
run,
|
||||
number,
|
||||
origin,
|
||||
});
|
||||
}
|
||||
|
||||
pub fn next_number(&self) -> RunNumber {
|
||||
self.entries
|
||||
.iter()
|
||||
.map(|entry| entry.number)
|
||||
.max()
|
||||
.map_or_else(RunNumber::first, RunNumber::after)
|
||||
}
|
||||
|
||||
pub fn run_by_origin(&self, origin: &Origin) -> Option<RunId> {
|
||||
self.entries
|
||||
.iter()
|
||||
.find(|entry| &entry.origin == origin)
|
||||
.map(|entry| entry.run.id())
|
||||
}
|
||||
|
||||
pub fn next_queued(&self) -> Option<(RunId, JobId)> {
|
||||
self.entries.iter().find_map(|entry| {
|
||||
entry
|
||||
.run
|
||||
.queued_jobs()
|
||||
.next()
|
||||
.map(|job| (entry.run.id(), job.id()))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn admits_queue(
|
||||
&self,
|
||||
origin: &Origin,
|
||||
jobs: usize,
|
||||
policy: SchedulerPolicy,
|
||||
) -> Result<(), RunsError> {
|
||||
let organization_open: usize = self
|
||||
.entries
|
||||
.iter()
|
||||
.filter(|entry| entry.origin.organization() == origin.organization())
|
||||
.map(|entry| entry.run.open_jobs())
|
||||
.sum();
|
||||
if organization_open.saturating_add(jobs) > policy.organization_queue_quota() as usize {
|
||||
return Err(RunsError::OrganizationQueueQuota {
|
||||
organization: origin.organization().to_owned(),
|
||||
limit: policy.organization_queue_quota(),
|
||||
});
|
||||
}
|
||||
if let Some(principal) = origin.principal() {
|
||||
let principal_open: usize = self
|
||||
.entries
|
||||
.iter()
|
||||
.filter(|entry| entry.origin.principal() == Some(principal))
|
||||
.map(|entry| entry.run.open_jobs())
|
||||
.sum();
|
||||
if principal_open.saturating_add(jobs) > policy.principal_queue_quota() as usize {
|
||||
return Err(RunsError::PrincipalQueueQuota {
|
||||
principal: principal.to_owned(),
|
||||
limit: policy.principal_queue_quota(),
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn assignment(&self, run: RunId, job: JobId) -> Result<Assignment, RunsError> {
|
||||
let entry = self.entry(run)?;
|
||||
let definition = entry.run.job_definition(job)?;
|
||||
let needs = entry.run.dependencies(job)?;
|
||||
Ok(Assignment::new(
|
||||
run,
|
||||
job,
|
||||
entry.number,
|
||||
entry.origin.clone(),
|
||||
definition,
|
||||
needs,
|
||||
))
|
||||
}
|
||||
|
||||
pub fn record_dispatch(&mut self, run: RunId) -> Result<(), RunsError> {
|
||||
let organization = self.entry(run)?.origin.organization().to_owned();
|
||||
self.organization_turns.insert(organization, self.next_turn);
|
||||
self.next_turn = self.next_turn.saturating_add(1);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn runs(&self) -> impl Iterator<Item = &Run> {
|
||||
self.entries.iter().map(|entry| &entry.run)
|
||||
}
|
||||
|
||||
pub fn cancellations(&self, node: NodeId) -> Vec<(RunId, JobId, Lease)> {
|
||||
self.entries
|
||||
.iter()
|
||||
.flat_map(|entry| {
|
||||
entry
|
||||
.run
|
||||
.cancellations(node)
|
||||
.map(move |(job, lease)| (entry.run.id(), job, lease))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub fn leases_held_by(&self, node: NodeId) -> usize {
|
||||
self.entries
|
||||
.iter()
|
||||
.flat_map(|entry| entry.run.leased_jobs())
|
||||
.filter(|(_, lease)| lease.node() == node)
|
||||
.count()
|
||||
}
|
||||
|
||||
pub fn run(&self, id: RunId) -> Result<&Run, RunsError> {
|
||||
self.entry(id).map(|entry| &entry.run)
|
||||
}
|
||||
|
||||
pub fn run_mut(&mut self, id: RunId) -> Result<&mut Run, RunsError> {
|
||||
self.entries
|
||||
.iter_mut()
|
||||
.find(|entry| entry.run.id() == id)
|
||||
.map(|entry| &mut entry.run)
|
||||
.ok_or(RunsError::UnknownRun(id))
|
||||
}
|
||||
|
||||
pub fn origin(&self, id: RunId) -> Result<&Origin, RunsError> {
|
||||
self.entry(id).map(|entry| &entry.origin)
|
||||
}
|
||||
|
||||
pub(super) fn projection_seeds(&self) -> Vec<ProjectionSeed> {
|
||||
self.entries
|
||||
.iter()
|
||||
.filter(|entry| entry.origin.delivery().is_some())
|
||||
.map(Self::projection_seed_of)
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The seed for one run, if it exists and is eligible for projection. Used
|
||||
/// to push a single run's state synchronously (at dispatch, so the token a
|
||||
/// node just received is already valid where it will be spent) alongside
|
||||
/// `projection_seeds`, which the periodic sweep uses for everything else.
|
||||
pub(super) fn projection_seed(&self, run: RunId) -> Option<ProjectionSeed> {
|
||||
self.entry(run)
|
||||
.ok()
|
||||
.filter(|entry| entry.origin.delivery().is_some())
|
||||
.map(Self::projection_seed_of)
|
||||
}
|
||||
|
||||
fn projection_seed_of(entry: &Entry) -> ProjectionSeed {
|
||||
ProjectionSeed {
|
||||
run: entry.run.id(),
|
||||
number: entry.number,
|
||||
origin: entry.origin.clone(),
|
||||
sequence: entry.run.sequence().get(),
|
||||
jobs: entry
|
||||
.run
|
||||
.projected_jobs()
|
||||
.map(|(job, state)| {
|
||||
let (state, authority) = match state {
|
||||
JobState::Waiting | JobState::Queued => (ProjectedJobState::Waiting, None),
|
||||
JobState::Assigned(lease) => (ProjectedJobState::Assigned, Some(lease)),
|
||||
JobState::Running(lease) => (ProjectedJobState::Running, Some(lease)),
|
||||
JobState::Finished(conclusion) => {
|
||||
(ProjectedJobState::Finished(conclusion), None)
|
||||
}
|
||||
JobState::Skipped => (ProjectedJobState::Skipped, None),
|
||||
};
|
||||
ProjectionJobSeed {
|
||||
job: job.id(),
|
||||
key: job.key().to_owned(),
|
||||
needs: job.needs().to_vec(),
|
||||
state,
|
||||
node: authority.map(|lease| lease.node()),
|
||||
fence: authority.map(|lease| lease.fence()),
|
||||
}
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
fn entry(&self, id: RunId) -> Result<&Entry, RunsError> {
|
||||
self.entries
|
||||
.iter()
|
||||
.find(|entry| entry.run.id() == id)
|
||||
.ok_or(RunsError::UnknownRun(id))
|
||||
}
|
||||
}
|
||||
mod diagnosis;
|
||||
mod projection;
|
||||
mod scheduling;
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use crate::Fence;
|
||||
use crate::{
|
||||
Assignment, JobId, Origin, ProjectedJobState, Run, RunId, RunNumber, RunsError, SchedulerPolicy,
|
||||
};
|
||||
use crate::{Lease, NodeId};
|
||||
|
||||
pub struct Entry {
|
||||
run: Run,
|
||||
number: RunNumber,
|
||||
origin: Origin,
|
||||
}
|
||||
|
||||
pub struct State {
|
||||
entries: Vec<Entry>,
|
||||
output_versions: BTreeMap<RunId, u64>,
|
||||
organization_turns: BTreeMap<String, u64>,
|
||||
next_turn: u64,
|
||||
}
|
||||
|
||||
pub(super) struct ProjectionSeed {
|
||||
pub run: RunId,
|
||||
pub number: RunNumber,
|
||||
pub origin: Origin,
|
||||
pub sequence: u64,
|
||||
pub jobs: Vec<ProjectionJobSeed>,
|
||||
}
|
||||
|
||||
pub(super) struct ProjectionJobSeed {
|
||||
pub job: JobId,
|
||||
pub key: String,
|
||||
pub needs: Vec<String>,
|
||||
pub state: ProjectedJobState,
|
||||
pub node: Option<NodeId>,
|
||||
pub fence: Option<Fence>,
|
||||
}
|
||||
|
||||
impl Default for State {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
entries: Vec::new(),
|
||||
output_versions: BTreeMap::new(),
|
||||
organization_turns: BTreeMap::new(),
|
||||
next_turn: 1,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl State {
|
||||
pub fn admit(&mut self, run: Run, number: RunNumber, origin: Origin) {
|
||||
self.entries.push(Entry {
|
||||
run,
|
||||
number,
|
||||
origin,
|
||||
});
|
||||
}
|
||||
|
||||
pub fn next_number(&self) -> RunNumber {
|
||||
self.entries
|
||||
.iter()
|
||||
.map(|entry| entry.number)
|
||||
.max()
|
||||
.map_or_else(RunNumber::first, RunNumber::after)
|
||||
}
|
||||
|
||||
pub fn run_by_origin(&self, origin: &Origin) -> Option<RunId> {
|
||||
self.entries
|
||||
.iter()
|
||||
.find(|entry| &entry.origin == origin)
|
||||
.map(|entry| entry.run.id())
|
||||
}
|
||||
|
||||
pub fn next_queued(&self) -> Option<(RunId, JobId)> {
|
||||
self.entries.iter().find_map(|entry| {
|
||||
entry
|
||||
.run
|
||||
.queued_jobs()
|
||||
.next()
|
||||
.map(|job| (entry.run.id(), job.id()))
|
||||
})
|
||||
}
|
||||
|
||||
pub fn admits_queue(
|
||||
&self,
|
||||
origin: &Origin,
|
||||
jobs: usize,
|
||||
policy: SchedulerPolicy,
|
||||
) -> Result<(), RunsError> {
|
||||
let organization_open: usize = self
|
||||
.entries
|
||||
.iter()
|
||||
.filter(|entry| entry.origin.organization() == origin.organization())
|
||||
.map(|entry| entry.run.open_jobs())
|
||||
.sum();
|
||||
if organization_open.saturating_add(jobs) > policy.organization_queue_quota() as usize {
|
||||
return Err(RunsError::OrganizationQueueQuota {
|
||||
organization: origin.organization().to_owned(),
|
||||
limit: policy.organization_queue_quota(),
|
||||
});
|
||||
}
|
||||
if let Some(principal) = origin.principal() {
|
||||
let principal_open: usize = self
|
||||
.entries
|
||||
.iter()
|
||||
.filter(|entry| entry.origin.principal() == Some(principal))
|
||||
.map(|entry| entry.run.open_jobs())
|
||||
.sum();
|
||||
if principal_open.saturating_add(jobs) > policy.principal_queue_quota() as usize {
|
||||
return Err(RunsError::PrincipalQueueQuota {
|
||||
principal: principal.to_owned(),
|
||||
limit: policy.principal_queue_quota(),
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn assignment(&self, run: RunId, job: JobId) -> Result<Assignment, RunsError> {
|
||||
let entry = self.entry(run)?;
|
||||
let definition = entry.run.job_definition(job)?;
|
||||
let needs = entry.run.dependencies(job)?;
|
||||
Ok(Assignment::new(
|
||||
run,
|
||||
job,
|
||||
entry.number,
|
||||
entry.origin.clone(),
|
||||
definition,
|
||||
needs,
|
||||
))
|
||||
}
|
||||
|
||||
pub fn record_dispatch(&mut self, run: RunId) -> Result<(), RunsError> {
|
||||
let organization = self.entry(run)?.origin.organization().to_owned();
|
||||
self.organization_turns.insert(organization, self.next_turn);
|
||||
self.next_turn = self.next_turn.saturating_add(1);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn runs(&self) -> impl Iterator<Item = &Run> {
|
||||
self.entries.iter().map(|entry| &entry.run)
|
||||
}
|
||||
|
||||
pub fn cancellations(&self, node: NodeId) -> Vec<(RunId, JobId, Lease)> {
|
||||
self.entries
|
||||
.iter()
|
||||
.flat_map(|entry| {
|
||||
entry
|
||||
.run
|
||||
.cancellations(node)
|
||||
.map(move |(job, lease)| (entry.run.id(), job, lease))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub fn leases_held_by(&self, node: NodeId) -> usize {
|
||||
self.entries
|
||||
.iter()
|
||||
.flat_map(|entry| entry.run.leased_jobs())
|
||||
.filter(|(_, lease)| lease.node() == node)
|
||||
.count()
|
||||
}
|
||||
|
||||
pub fn run(&self, id: RunId) -> Result<&Run, RunsError> {
|
||||
self.entry(id).map(|entry| &entry.run)
|
||||
}
|
||||
|
||||
pub fn run_mut(&mut self, id: RunId) -> Result<&mut Run, RunsError> {
|
||||
self.entries
|
||||
.iter_mut()
|
||||
.find(|entry| entry.run.id() == id)
|
||||
.map(|entry| &mut entry.run)
|
||||
.ok_or(RunsError::UnknownRun(id))
|
||||
}
|
||||
|
||||
pub fn origin(&self, id: RunId) -> Result<&Origin, RunsError> {
|
||||
self.entry(id).map(|entry| &entry.origin)
|
||||
}
|
||||
|
||||
fn entry(&self, id: RunId) -> Result<&Entry, RunsError> {
|
||||
self.entries
|
||||
.iter()
|
||||
.find(|entry| entry.run.id() == id)
|
||||
.ok_or(RunsError::UnknownRun(id))
|
||||
}
|
||||
}
|
||||
@@ -1,0 +1,78 @@
|
||||
use super::{Entry, ProjectionJobSeed, ProjectionSeed, State};
|
||||
use crate::{JobState, ProjectedJobState, RunId, RunNumber};
|
||||
|
||||
impl State {
|
||||
pub(in crate::registry) fn projection_seeds(&self) -> Vec<ProjectionSeed> {
|
||||
self.entries
|
||||
.iter()
|
||||
.filter(|entry| entry.origin.delivery().is_some())
|
||||
.map(Self::projection_seed_of)
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub(in crate::registry) fn projection_versions(&self) -> Vec<(RunId, RunNumber, u64, u64)> {
|
||||
self.entries
|
||||
.iter()
|
||||
.filter(|entry| entry.origin.delivery().is_some())
|
||||
.map(|entry| {
|
||||
(
|
||||
entry.run.id(),
|
||||
entry.number,
|
||||
entry.run.sequence().get(),
|
||||
self.output_versions
|
||||
.get(&entry.run.id())
|
||||
.copied()
|
||||
.unwrap_or(0),
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub(crate) fn record_output(&mut self, run: RunId) {
|
||||
let version = self.output_versions.entry(run).or_default();
|
||||
*version = version.saturating_add(1);
|
||||
}
|
||||
|
||||
/// The seed for one run, if it exists and is eligible for projection. Used
|
||||
/// to push a single run's state synchronously (at dispatch, so the token a
|
||||
/// node just received is already valid where it will be spent) alongside
|
||||
/// `projection_seeds`, which the periodic sweep uses for everything else.
|
||||
pub(in crate::registry) fn projection_seed(&self, run: RunId) -> Option<ProjectionSeed> {
|
||||
self.entry(run)
|
||||
.ok()
|
||||
.filter(|entry| entry.origin.delivery().is_some())
|
||||
.map(Self::projection_seed_of)
|
||||
}
|
||||
|
||||
fn projection_seed_of(entry: &Entry) -> ProjectionSeed {
|
||||
ProjectionSeed {
|
||||
run: entry.run.id(),
|
||||
number: entry.number,
|
||||
origin: entry.origin.clone(),
|
||||
sequence: entry.run.sequence().get(),
|
||||
jobs: entry
|
||||
.run
|
||||
.projected_jobs()
|
||||
.map(|(job, state)| {
|
||||
let (state, authority) = match state {
|
||||
JobState::Waiting | JobState::Queued => (ProjectedJobState::Waiting, None),
|
||||
JobState::Assigned(lease) => (ProjectedJobState::Assigned, Some(lease)),
|
||||
JobState::Running(lease) => (ProjectedJobState::Running, Some(lease)),
|
||||
JobState::Finished(conclusion) => {
|
||||
(ProjectedJobState::Finished(conclusion), None)
|
||||
}
|
||||
JobState::Skipped => (ProjectedJobState::Skipped, None),
|
||||
};
|
||||
ProjectionJobSeed {
|
||||
job: job.id(),
|
||||
key: job.key().to_owned(),
|
||||
needs: job.needs().to_vec(),
|
||||
state,
|
||||
node: authority.map(|lease| lease.node()),
|
||||
fence: authority.map(|lease| lease.fence()),
|
||||
}
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user