chadtmiller.com/knotless
git that lives in your Atmosphere account
7
//! Turns at what runs share: a job, of the `--jobs` at once, and keys
//! each held by one run at a time, such as a workflow's turn on a ref
//! whose older runs finish. Of the jobs, runs other than `Main` hold only
//! so many at once, so a burst of them, as every open pull run again when
//! the branch they are into moves, leaves the rest of the machine room;
//! and each owner's runs only as many as the run waiting may, so one
//! owner's burst leaves room for the rest.
//!
//! A run takes all it needs at once, once all of it is free, so it never
//! holds one thing while it waits for another, and two runs never wait on
//! each other. Runs waiting go by rank: one ranked before another has
//! first claim on the keys it waits for, and on a job when both could
//! start. One ranked lower goes ahead only with keys none before it waits
//! for and a job left over, so a run waiting for a volume in use does not
//! keep others from the jobs free. A run going is never stopped for one
//! ranked before it.
//!
//! Within a class, owners take turns: each of an owner's runs has the
//! turn after its last still waiting, or with none, the turn of the last
//! run to start, so a burst of one owner's runs goes one by one between
//! others' rather than all before them. One owner's runs go in the order
//! they were started.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::{Arc, Mutex};
use tokio::sync::Notify;
/// What a run is, for its place in line, first first.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum Class {
/// A push to its repository's default branch, which deploys wait on.
Main,
/// A pull's, or a push to any other ref.
Other,
/// A pull's not in the merge queue the runner lands, whose runs go
/// after the queue's.
Later,
/// One its owner asked for by name, as one too costly for every push:
/// none waits for it.
Asked,
}
/// A run's class, and the order runs were started in.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct Rank {
pub class: Class,
pub seq: u64,
}
/// A run's place in line: by its class, then its owner's turn, then
/// oldest first.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
struct Place {
class: Class,
turn: u64,
seq: u64,
}
pub struct Turns {
board: Mutex<Board>,
/// Rung whenever what is held, or who waits, changes.
changed: Notify,
}
/// The jobs free, those of them runs other than `Main` may still take,
/// the keys held and the jobs each owner does, what each run waiting needs, and the turn of the last run to
/// start in each class.
struct Board {
jobs: usize,
others: usize,
held: HashSet<String>,
owners: HashMap<String, usize>,
waiting: BTreeMap<Place, Want>,
last: HashMap<Class, u64>,
}
/// Whose a run is: its repository's owner, and the most jobs that
/// owner's runs may hold for it to start.
pub struct Whose {
pub owner: String,
pub most: usize,
}
/// Whose a run waiting is, and the keys it needs.
struct Want {
whose: Whose,
keys: Vec<String>,
}
impl Board {
/// Whether the run at `place` may start now: with those before it
/// that can start counted as started, and the keys those that cannot
/// wait for kept for them.
fn starts(&self, place: &Place, want: &Want) -> bool {
let mut jobs = (self.jobs, self.others);
let mut ahead = HashMap::new();
let mut claimed = HashSet::new();
for (theirs, w) in self.waiting.range(..place) {
if self.room(theirs.class, jobs, w, &ahead) && self.free(&w.keys, &claimed) {
jobs = taken(theirs.class, jobs);
*ahead.entry(w.whose.owner.as_str()).or_default() += 1;
}
claimed.extend(w.keys.iter().map(String::as_str));
}
self.room(place.class, jobs, want, &ahead) && self.free(&want.keys, &claimed)
}
/// Whether a run of `class` that `want` has a job among `jobs` free,
/// its owner holding fewer than `want` may, with those `ahead` of it.
fn room(
&self,
class: Class,
jobs: (usize, usize),
want: &Want,
ahead: &HashMap<&str, usize>,
) -> bool {
let owner = want.whose.owner.as_str();
let holds = self.owners.get(owner).copied().unwrap_or_default();
room(class, jobs) && holds + ahead.get(owner).copied().unwrap_or_default() < want.whose.most
}
fn free(&self, keys: &[String], claimed: &HashSet<&str>) -> bool {
keys.iter()
.all(|k| !self.held.contains(k) && !claimed.contains(k.as_str()))
}
/// The place in line of `owner`'s run ranked `rank`: the turn after
/// its last waiting in the class started before it, or the last run's
/// to start, if later. Of runs started together, the one first in
/// line is then still the one started first.
fn place(&self, rank: Rank, owner: &str) -> Place {
let last = self.last.get(&rank.class).copied().unwrap_or_default();
let mine = self
.waiting
.iter()
.filter(|(p, w)| p.class == rank.class && p.seq < rank.seq && w.whose.owner == owner);
let after = mine.map(|(p, _)| p.turn + 1).max().unwrap_or_default();
Place {
class: rank.class,
turn: after.max(last),
seq: rank.seq,
}
}
}
/// Whether a run of `class` has a job among `(jobs, others)` free.
fn room(class: Class, (jobs, others): (usize, usize)) -> bool {
jobs > 0 && (class == Class::Main || others > 0)
}
/// `(jobs, others)` once a run of `class` has taken one.
fn taken(class: Class, (jobs, others): (usize, usize)) -> (usize, usize) {
let other = usize::from(class != Class::Main);
(jobs - 1, others - other)
}
impl Turns {
/// `jobs` at once, `others` of them at most for runs other than
/// `Main`.
pub fn new(jobs: usize, others: usize) -> Arc<Turns> {
Arc::new(Turns {
board: Mutex::new(Board {
jobs,
others,
held: HashSet::new(),
owners: HashMap::new(),
waiting: BTreeMap::new(),
last: HashMap::new(),
}),
changed: Notify::new(),
})
}
/// A job and `keys`, once the run ranked `rank`, `whose`, may have
/// them. Dropped while it waits, it leaves the line.
pub async fn take(self: Arc<Self>, rank: Rank, whose: Whose, mut keys: Vec<String>) -> Turn {
keys.sort_unstable();
keys.dedup();
let place = {
let mut board = self.board.lock().unwrap();
let place = board.place(rank, &whose.owner);
let want = Want { whose, keys };
board.waiting.insert(place, want);
place
};
let _line = Line {
turns: &self,
place,
};
loop {
// Made before looking, so a change after the look still wakes it.
let changed = self.changed.notified();
if let Some(want) = self.start(&place) {
return Turn {
turns: self.clone(),
class: place.class,
owner: want.whose.owner,
keys: want.keys,
};
}
changed.await;
}
}
/// Takes what the run at `place` waits for, if it may start, and
/// gives back what that was.
fn start(&self, place: &Place) -> Option<Want> {
let mut board = self.board.lock().unwrap();
let want = &board.waiting[place];
if !board.starts(place, want) {
return None;
}
let want = board.waiting.remove(place)?;
(board.jobs, board.others) = taken(place.class, (board.jobs, board.others));
board.held.extend(want.keys.iter().cloned());
*board.owners.entry(want.whose.owner.clone()).or_default() += 1;
let last = board.last.entry(place.class).or_default();
*last = (*last).max(place.turn);
Some(want)
}
}
/// A run's place while it waits.
struct Line<'a> {
turns: &'a Turns,
place: Place,
}
impl Drop for Line<'_> {
fn drop(&mut self) {
let mut board = self.turns.board.lock().unwrap();
if board.waiting.remove(&self.place).is_some() {
drop(board);
self.turns.changed.notify_waiters();
}
}
}
/// A job and keys held, let go when dropped.
pub struct Turn {
turns: Arc<Turns>,
class: Class,
owner: String,
keys: Vec<String>,
}
impl Drop for Turn {
fn drop(&mut self) {
let mut board = self.turns.board.lock().unwrap();
board.jobs += 1;
board.others += usize::from(self.class != Class::Main);
for key in &self.keys {
board.held.remove(key);
}
if let Some(holds) = board.owners.get_mut(&self.owner) {
*holds -= 1;
}
drop(board);
self.turns.changed.notify_waiters();
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures::FutureExt;
use std::pin::Pin;
type Waiting = Pin<Box<dyn Future<Output = Turn>>>;
fn rank(class: Class, seq: u64) -> Rank {
Rank { class, seq }
}
fn wait(turns: &Arc<Turns>, rank: Rank, keys: &[&str]) -> Waiting {
owned(turns, "o", rank, keys)
}
/// `owner`'s run waiting, whose owner may hold every job.
fn owned(turns: &Arc<Turns>, owner: &str, rank: Rank, keys: &[&str]) -> Waiting {
capped(turns, owner, usize::MAX, rank, keys)
}
/// `owner`'s run waiting, which goes while its owner holds fewer than
/// `most` jobs.
fn capped(turns: &Arc<Turns>, owner: &str, most: usize, rank: Rank, keys: &[&str]) -> Waiting {
let keys = keys.iter().map(|k| k.to_string()).collect();
let owner = owner.to_string();
Box::pin(turns.clone().take(rank, Whose { owner, most }, keys))
}
/// The turn, if `waiting` has it now.
fn ready(waiting: &mut Waiting) -> Option<Turn> {
waiting.as_mut().now_or_never()
}
#[test]
fn main_first_then_oldest() {
let turns = Turns::new(1, 1);
let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
let mut pull = wait(&turns, rank(Class::Other, 2), &[]);
let mut later = wait(&turns, rank(Class::Other, 3), &[]);
let mut main = wait(&turns, rank(Class::Main, 4), &[]);
let mut asked = wait(&turns, rank(Class::Asked, 0), &[]);
for w in [&mut pull, &mut later, &mut main, &mut asked] {
assert!(ready(w).is_none());
}
// The run going is not stopped; once it ends, the push to main
// goes before the older pull.
drop(going);
assert!(ready(&mut pull).is_none());
let main = ready(&mut main).unwrap();
assert!(ready(&mut later).is_none());
drop(main);
assert!(ready(&mut later).is_none());
let pull = ready(&mut pull).unwrap();
drop(pull);
let later = ready(&mut later).unwrap();
// One asked for goes after the rest, however old.
assert!(ready(&mut asked).is_none());
drop(later);
assert!(ready(&mut asked).is_some());
}
#[test]
fn later_after_the_rest() {
let turns = Turns::new(1, 1);
let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
let mut unqueued = wait(&turns, rank(Class::Later, 2), &[]);
let mut pull = wait(&turns, rank(Class::Other, 3), &[]);
for w in [&mut unqueued, &mut pull] {
assert!(ready(w).is_none());
}
drop(going);
assert!(ready(&mut unqueued).is_none());
drop(ready(&mut pull).unwrap());
assert!(ready(&mut unqueued).is_some());
}
#[test]
fn keys_one_at_a_time() {
let turns = Turns::new(4, 4);
// Named twice, a key is taken once.
let a = ready(&mut wait(&turns, rank(Class::Other, 1), &["v", "v"])).unwrap();
let mut b = wait(&turns, rank(Class::Other, 2), &["v"]);
assert!(ready(&mut b).is_none());
// One with a key of its own goes beside it.
let side = ready(&mut wait(&turns, rank(Class::Other, 3), &["w"]));
assert!(side.is_some());
drop(a);
assert!(ready(&mut b).is_some());
}
#[test]
fn claims_and_leftovers() {
let turns = Turns::new(2, 2);
let going = ready(&mut wait(&turns, rank(Class::Other, 1), &["v"])).unwrap();
// Main waits for v, in use; w, which it needs too, is kept for it.
let mut main = wait(&turns, rank(Class::Main, 2), &["v", "w"]);
assert!(ready(&mut main).is_none());
let mut wants_w = wait(&turns, rank(Class::Other, 3), &["w"]);
assert!(ready(&mut wants_w).is_none());
// One needing neither has the job left over.
let mut free = wait(&turns, rank(Class::Other, 4), &["x"]);
let free = ready(&mut free).unwrap();
drop(free);
drop(going);
let main = ready(&mut main).unwrap();
assert!(ready(&mut wants_w).is_none());
drop(main);
assert!(ready(&mut wants_w).is_some());
}
#[test]
fn others_take_only_their_share() {
let turns = Turns::new(3, 1);
let pull = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
// A second pull waits, though jobs are free, and main goes past it.
let mut second = wait(&turns, rank(Class::Other, 2), &[]);
assert!(ready(&mut second).is_none());
let main = ready(&mut wait(&turns, rank(Class::Main, 3), &[]));
assert!(main.is_some());
drop(pull);
assert!(ready(&mut second).is_some());
}
#[test]
fn leaving_the_line() {
let turns = Turns::new(1, 1);
let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
let mut main = wait(&turns, rank(Class::Main, 2), &["v"]);
let mut pull = wait(&turns, rank(Class::Other, 3), &["v"]);
assert!(ready(&mut main).is_none());
assert!(ready(&mut pull).is_none());
drop(going);
// Cancelled while it waits, the one before leaves the line.
drop(main);
assert!(ready(&mut pull).is_some());
}
#[test]
fn started_first_goes_first() {
let turns = Turns::new(1, 1);
let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap();
// Started in this order, but in line the other way round.
let mut second = wait(&turns, rank(Class::Main, 3), &[]);
assert!(ready(&mut second).is_none());
let mut first = wait(&turns, rank(Class::Main, 2), &[]);
assert!(ready(&mut first).is_none());
drop(going);
assert!(ready(&mut second).is_none());
drop(ready(&mut first).unwrap());
assert!(ready(&mut second).is_some());
}
#[test]
fn owners_take_turns() {
let turns = Turns::new(1, 1);
let going = ready(&mut owned(&turns, "a", rank(Class::Other, 1), &[])).unwrap();
// A burst of a's, then one of b's: b's goes after a's first
// waiting, before the rest.
let mut a2 = owned(&turns, "a", rank(Class::Other, 2), &[]);
let mut a3 = owned(&turns, "a", rank(Class::Other, 3), &[]);
let mut b4 = owned(&turns, "b", rank(Class::Other, 4), &[]);
for w in [&mut a2, &mut a3, &mut b4] {
assert!(ready(w).is_none());
}
drop(going);
assert!(ready(&mut a3).is_none());
assert!(ready(&mut b4).is_none());
drop(ready(&mut a2).unwrap());
assert!(ready(&mut a3).is_none());
drop(ready(&mut b4).unwrap());
assert!(ready(&mut a3).is_some());
}
#[test]
fn an_owner_holds_only_its_share() {
let turns = Turns::new(3, 3);
let going = ready(&mut owned(&turns, "b", rank(Class::Other, 1), &[])).unwrap();
let busy = ready(&mut owned(&turns, "c", rank(Class::Other, 2), &[])).unwrap();
let more = ready(&mut owned(&turns, "c", rank(Class::Other, 3), &[])).unwrap();
let mut a4 = capped(&turns, "a", 1, rank(Class::Other, 4), &[]);
let mut a5 = capped(&turns, "a", 1, rank(Class::Other, 5), &[]);
assert!(ready(&mut a4).is_none());
assert!(ready(&mut a5).is_none());
drop((going, busy, more));
// Three jobs free, but a's first waiting takes a's one: its second
// waits, and another owner's goes past it.
assert!(ready(&mut a5).is_none());
let a4 = ready(&mut a4).unwrap();
assert!(ready(&mut a5).is_none());
let b6 = ready(&mut owned(&turns, "b", rank(Class::Other, 6), &[]));
assert!(b6.is_some());
// What a run may hold is its own: one of a's that may hold two goes.
let a7 = ready(&mut capped(&turns, "a", 2, rank(Class::Other, 7), &[]));
assert!(a7.is_some());
drop((a4, a7));
assert!(ready(&mut a5).is_some());
}
}