chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
knotless-run: a batch built ahead goes first once the one under it lands
.changeset/queue-ahead-promoted.mdnew+7−0
| @@ -0,0 +1,7 @@ | ||
--- | ||
default: patch | ||
--- | ||
| ||
# A merge queue's next batch goes first once the batch before it lands | ||
| ||
A batch built ahead of one still being checked waits for a job behind pulls' runs. Once the batch under it lands, a runner now ranks what is left of it with pushes to the default branch, so it lands next instead of waiting behind every pre-flight queued meanwhile. | ||
crates/knotless-run/src/lib.rs+18−5
| @@ -893,9 +893,10 @@ impl Subject { | ||
| ||
/// Whether its runs go first, at the CPU share and before the jobs | ||
/// others wait for: a push to the default branch, or the merge | ||
/// queue's batch that lands next. One run ahead goes as a pull does. | ||
/// queue's batch that lands next. One run ahead goes as a pull does, | ||
/// until the batch it is on lands. | ||
fn first(&self) -> bool { | ||
self.main && !self.batch.as_ref().is_some_and(|b| b.ahead) | ||
self.main && !self.batch.as_ref().is_some_and(|b| b.ahead()) | ||
} | ||
| ||
/// The code of someone who may not push here, which gets no secrets. | ||
| @@ -1167,6 +1168,9 @@ impl Cancelled { | ||
/// had at once, or waited for. | ||
type OnRef = std::pin::Pin<Box<dyn Future<Output = turns::Turn> + Send>>; | ||
| ||
/// Ends once a run's batch is no longer ahead of another. | ||
type Promoted = turns::Promoted; | ||
| ||
/// Why a run is cancelled: a newer commit's run of its workflow on the same | ||
/// ref or pull. | ||
const NEWER: &str = "for a newer commit's run"; | ||
| @@ -2649,7 +2653,12 @@ impl Runner { | ||
Some(After::Not { status, line }) => return Waited::Not { status, line }, | ||
None => return Waited::Cancelled, | ||
}; | ||
let held = self.take_turns(rank, watch, on_ref, cancelled).await; | ||
// A batch's run ranks as a pull's only while it is ahead. | ||
let promoted = (subject.batch.as_ref()) | ||
.filter(|_| rank.class == Class::Later) | ||
.map(|b| Box::pin(b.promoted()) as Promoted); | ||
let held = self.take_turns(rank, watch, on_ref, promoted, cancelled); | ||
let held = held.await; | ||
held.map_or(Waited::Cancelled, |mut held| { | ||
held.said = waited + &held.said; | ||
Waited::Held(held) | ||
| @@ -2662,11 +2671,15 @@ impl Runner { | ||
/// then on; then every run waits for a job. Its caches it shares, each | ||
/// run through overlays of its own. The turn on a ref is waited for | ||
/// only by runs that hold nothing, so no two runs wait on each other. | ||
/// One of a batch built ahead waits as a pull's until `promoted`, the | ||
/// batch it was ahead of having landed, and then as main's, in the | ||
/// order it was started in. | ||
async fn take_turns( | ||
&self, | ||
rank: Rank, | ||
watch: &Watch, | ||
on_ref: Option<OnRef>, | ||
promoted: Option<Promoted>, | ||
cancelled: &mut Cancelled, | ||
) -> Option<Held> { | ||
let mut kept = None; | ||
| @@ -2676,8 +2689,8 @@ impl Runner { | ||
} | ||
let (whose, over) = self.whose(watch); | ||
let began = std::time::Instant::now(); | ||
let turn = until_cancelled(cancelled, self.turns.clone().take(rank, whose, Vec::new())); | ||
let turn = turn.await?; | ||
let turn = self.turns.clone().take_or_promoted(rank, whose, promoted); | ||
let turn = until_cancelled(cancelled, turn).await?; | ||
let mut said = waited(over, began.elapsed()); | ||
// Holding its job, so that no other run starts meanwhile. | ||
let room = room(&self.caches, self.opts.disk_floor, ROOM_EVERY); | ||
crates/knotless-run/src/queue.rs+31−5
| @@ -33,7 +33,9 @@ | ||
//! passed (`--depth` 2, the default; 1 runs one at a time). It runs ahead | ||
//! as a pull does: after the runs of pushes to the default branch and of | ||
//! the batch it is on, with a job left over and the CPU share a pull's run | ||
//! has, so it never holds up a deploy. The batch it is on failing, or | ||
//! has, so it never holds up a deploy; once that batch lands, it is the | ||
//! one that lands next, and what of it still waits goes as main's does. | ||
//! The batch it is on failing, or | ||
//! changing, drops it, cancelling its runs; it lands only on the commit | ||
//! that batch landed as, and nothing lands but on the very commit it was | ||
//! built on. | ||
| @@ -99,8 +101,8 @@ pub(crate) struct Batch { | ||
/// the pushes' caches. | ||
pub trusted: bool, | ||
/// Whether it was built on a batch still being checked: its runs go | ||
/// as a pull's do. | ||
pub ahead: bool, | ||
/// as a pull's do until that one lands ([`Batch::promote`]). | ||
ahead: Arc<tokio::sync::watch::Sender<bool>>, | ||
/// The accounts whose shared steps its lock adds that the repository | ||
/// does not vouch for, which make it untrusted. | ||
pub strangers: Vec<String>, | ||
| @@ -114,12 +116,32 @@ impl Batch { | ||
base_commit: base_commit.to_string(), | ||
pulls, | ||
trusted, | ||
ahead: false, | ||
ahead: Arc::new(tokio::sync::watch::Sender::new(false)), | ||
strangers: Vec::new(), | ||
runs: Arc::default(), | ||
} | ||
} | ||
| ||
/// Whether it is still ahead of a batch being checked. | ||
pub fn ahead(&self) -> bool { | ||
*self.ahead.borrow() | ||
} | ||
| ||
/// Done once the batch it is ahead of lands, and it lands next: its | ||
/// runs waiting for a job go first from then on, and those yet to | ||
/// start take the CPU share main's do. | ||
pub fn promote(&self) { | ||
self.ahead.send_replace(false); | ||
} | ||
| ||
/// Ends once it is no longer ahead, at once if it never was. | ||
pub fn promoted(&self) -> impl Future<Output = ()> + Send + 'static { | ||
let mut ahead = self.ahead.subscribe(); | ||
async move { | ||
let _ = ahead.wait_for(|a| !*a).await; | ||
} | ||
} | ||
| ||
/// Notes a run of `workflow` started. | ||
pub fn started(&self, workflow: &str) { | ||
self.runs.lock().unwrap().insert(workflow.to_string(), None); | ||
| @@ -455,6 +477,10 @@ impl Runner { | ||
} | ||
line.unlanded = None; | ||
tip = flight.stack.tip().to_string(); | ||
// The batch built ahead of it is the one that lands next. | ||
if let Some(next) = line.flights.first() { | ||
next.batch.promote(); | ||
} | ||
let landed = flight.stack.stacked.len(); | ||
telemetry::batch( | ||
"batch landed", | ||
| @@ -839,7 +865,7 @@ impl Runner { | ||
}); | ||
let refname = format!("refs/heads/{}", on.branch); | ||
let mut batch = Batch::new(&refname, &stack.base, pulls.collect(), trusted); | ||
batch.ahead = ahead; | ||
batch.ahead.send_replace(ahead); | ||
batch.strangers = strangers; | ||
let changed = changes(on.git, &stack.base, stack.tip()).await; | ||
let subject = Subject { | ||
crates/knotless-run/src/turns.rs+59−0
| @@ -44,6 +44,9 @@ pub enum Class { | ||
Asked, | ||
} | ||
| ||
/// Ends once a run waiting in a lower class goes as `Main` instead. | ||
pub type Promoted = std::pin::Pin<Box<dyn Future<Output = ()> + Send>>; | ||
| ||
/// A run's class, and the order runs were started in. | ||
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] | ||
pub struct Rank { | ||
| @@ -218,6 +221,33 @@ impl Turns { | ||
} | ||
} | ||
| ||
/// A job, as [`Turns::take`] gives one with no keys; or, if | ||
/// `promoted` ends while the run still waits, waiting again as `Main` | ||
/// from then on, in the order it was started in, as the merge queue's | ||
/// batch built ahead does once the one it is on lands. | ||
pub async fn take_or_promoted( | ||
self: Arc<Self>, | ||
rank: Rank, | ||
whose: Whose, | ||
promoted: Option<Promoted>, | ||
) -> Turn { | ||
let Some(promoted) = promoted else { | ||
return self.take(rank, whose, Vec::new()).await; | ||
}; | ||
let again = Whose { | ||
owner: whose.owner.clone(), | ||
most: whose.most, | ||
}; | ||
tokio::select! { | ||
biased; | ||
turn = self.clone().take(rank, whose, Vec::new()) => turn, | ||
() = promoted => { | ||
let main = Rank { class: Class::Main, ..rank }; | ||
self.take(main, again, Vec::new()).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> { | ||
| @@ -368,6 +398,35 @@ mod tests { | ||
assert!(ready(&mut unqueued).is_some()); | ||
} | ||
| ||
#[test] | ||
fn promoted_goes_as_main() { | ||
let turns = Turns::new(2, 1); | ||
let going = ready(&mut wait(&turns, rank(Class::Other, 1), &[])).unwrap(); | ||
// A batch built ahead waits as a pull's, with the pulls' share | ||
// taken, though a job is free. | ||
let (promote, promoted) = futures::channel::oneshot::channel::<()>(); | ||
let promoted: Promoted = Box::pin(async { | ||
let _ = promoted.await; | ||
}); | ||
let whose = Whose { | ||
owner: "o".into(), | ||
most: usize::MAX, | ||
}; | ||
let later = rank(Class::Later, 2); | ||
let mut ahead: Waiting = | ||
Box::pin(turns.clone().take_or_promoted(later, whose, Some(promoted))); | ||
let mut pull = wait(&turns, rank(Class::Other, 3), &[]); | ||
assert!(ready(&mut ahead).is_none()); | ||
assert!(ready(&mut pull).is_none()); | ||
// The batch it was on landing, it goes as main's, at once. | ||
promote.send(()).unwrap(); | ||
let ahead = ready(&mut ahead).unwrap(); | ||
assert!(ready(&mut pull).is_none()); | ||
drop(going); | ||
assert!(ready(&mut pull).is_some()); | ||
drop(ahead); | ||
} | ||
| ||
#[test] | ||
fn keys_one_at_a_time() { | ||
let turns = Turns::new(4, 4); | ||
crates/knotless-run/tests/run.rs+54−0
| @@ -4441,6 +4441,60 @@ fn a_merge_queue_runs_ahead_with_a_job_to_spare() { | ||
assert_eq!(second["status"], "success", "{second}"); | ||
} | ||
| ||
#[test] | ||
fn a_batch_ahead_goes_first_once_the_one_it_is_on_lands() { | ||
// A batch built ahead waits as a pull's run does, so a pre-flight | ||
// holding the pulls' share keeps it waiting. Once the batch it is on | ||
// lands, it is the batch that lands next, and goes as main's. | ||
let rt = tokio::runtime::Runtime::new().unwrap(); | ||
let q = Line::new(&rt); | ||
let fly = q.go.join("fly"); | ||
let first = format!( | ||
"queue: first\non: {{pull: {{base: [refs/heads/main], paths: [slow]}}}}\nsteps:\n - run: until [ -f {} ]; do sleep 0.1; done\n", | ||
fly.display() | ||
); | ||
q.branch("main", ".knotless/workflows/first.yml", &first, false); | ||
let l = q.lander(&rt, "state.json", |opts| { | ||
(opts.batch, opts.depth) = (1, 2); | ||
(opts.jobs, opts.pull_jobs) = (2, 1); | ||
}); | ||
q.hold(); | ||
// The slow one's pre-flight takes the pulls' share and keeps it. | ||
let s = q.pull("s", "slow", "s\n"); | ||
q.queue(&[&s]); | ||
l.step(&rt); | ||
let url = q.url.clone(); | ||
let preflight = || { | ||
let args = ["run", "list", &url, "--pull", &s, "--workflow", "first"]; | ||
q.ok(&args).starts_with("running") | ||
}; | ||
wait_until(preflight); | ||
// a1's batch takes main's job, and a2's, built on it, waits. Both are | ||
// queued before the runner sees them, so neither runs as a pull. | ||
let (a1, a2) = (q.pull("a1", "a1", "1\n"), q.pull("a2", "a2", "2\n")); | ||
q.queue(&[&a1, &a2]); | ||
wait_until(|| { | ||
l.step(&rt); | ||
q.ran().len() == 1 | ||
}); | ||
let start = q.main(); | ||
let on = q.batch_on(&start); | ||
let ahead = (q.batches().into_iter()) | ||
.find(|c| c["baseCommit"] == on["commit"] && c["workflow"] == "ci") | ||
.unwrap(); | ||
assert_eq!(ahead["status"], "queued", "{ahead}"); | ||
// a1 lands, and a2 goes past the pre-flight still holding the pulls' | ||
// share, and lands too. | ||
q.release(); | ||
wait_until(|| { | ||
l.step(&rt); | ||
q.log(2) == ["a2", "a1"] | ||
}); | ||
assert!(preflight()); | ||
std::fs::write(&fly, "").unwrap(); | ||
l.pass(&rt); | ||
} | ||
| ||
#[test] | ||
fn a_batch_passing_as_the_runner_stops_lands() { | ||
// A batch whose runs pass once the runner is stopping lands before it | ||
docs/design/merge-queue.md+4−0
| @@ -425,6 +425,10 @@ built on the branch itself, where a conflict is real. | ||
The batch that lands next runs in the runner's highest class, with pushes | ||
to the default branch. A batch ahead runs in a lower class with a lower CPU | ||
weight, so work that may be thrown away yields to the batch about to land. | ||
Once the batch under it lands, it is the batch that lands next: its runs | ||
still waiting for a job move up to the highest class, keeping the order | ||
they were started in, and those that start from then on get the higher CPU | ||
weight. Runs it already has going keep the lower weight until they end. | ||
| ||
## Noticing a queued pull | ||
| ||
docs/design/runner.md+1−1
| @@ -404,7 +404,7 @@ main's runs wait only for runs already going. | ||
| --- | --- | | ||
| `Main` | pushes to the default branch, and the merge-queue batch that lands next | | ||
| `Other` | pull runs, and pushes to other refs | | ||
| `Later` | pulls not in the queue, where this runner lands the queue or runs the pulls alone, and a batch built ahead of one still checking | | ||
| `Later` | pulls not in the queue, where this runner lands the queue or runs the pulls alone, and a batch built ahead of one still checking, until that one lands | | ||
| `Asked` | runs asked for by name, after every other waiting run, however old | | ||
| ||
The classes other than `Main` together hold at most `--pull-jobs` jobs, | ||