Flow: key jobs held for a better runner by hub as well as job c95446a17fdd
The times jobs were first held for a better runner lived in one process-wide map keyed by job id; test hubs running side by side number their jobs alike, so one's claim could clear another's hold and the ranking test failed now and then. Keyed by hub and job. refs andymccall/oxygenflow#42
- Author
- Andy McCall @andymccall <[email protected]>
- When
- · 2026-10-12 00:53 UTC
- Signature
- ✔ signed by Andy McCall
- Hash
c95446a17fddd892eaaa5513071c20ea4d154b748a7e0df5116d8e067ef834b2- Depends on
- 42fa95540468 501d8b268dd4
- Depended on by
- 951aed50c2f5 4fa0da126b70
- In bookmarks
- api-unknown-fields cancel-landed-runs clear-waiting close-proposal-140 diffusion-pin-bookmark draft-ready image-check main oh-comment-body-file proposal-list-fast
- In tags
- v0.5.0 v0.5.1 v0.5.2 v0.6.0 v0.6.1 v0.7.0 v0.8.0 v0.9.0 v0.9.1 v0.9.11 v0.9.15 v0.9.16 v0.9.19 v0.9.2 v0.9.20 v0.9.21 v0.9.22 v0.9.23 v0.9.24 v0.9.3 v0.9.4 v0.9.5 v0.9.6 v0.9.7 v0.9.8 v0.9.9
- Edits
- 6 atoms, 0 tree operations, 1 file
Browse the tree at this change
modified crates/hub/src/flow.rs +7 −4
| … | 525 unchanged lines |
||
| 526 | 526 | let (archs, held) = (fdb::image_archs(&app.db).await?, crate::images::held(&app.db).await?); |
|
| 527 | 527 | let now = db::now_ms(); |
|
| 528 | 528 | let fleet = Fleet { image_archs: &archs, hub_images: &held, now }; |
|
| 529 | + | let hub_key = app.repos.root().display().to_string(); |
|
| 529 | 530 | let better = |j: &fdb::Job| -> bool { |
|
| 530 | 531 | let Some((owner, repo_id)) = owners.get(&j.id) else { return false }; |
|
| 531 | 532 | let rootfs = rootfs_of(j); |
|
| 532 | 533 | let mut holds = HELD.lock().unwrap_or_else(|p| p.into_inner()); |
|
| 533 | 534 | let holds = holds.get_or_insert_with(Default::default); |
|
| 534 | 535 | holds.retain(|_, t| now - *t < 600_000); |
|
| 535 | - | if holds.get(&j.id).is_some_and(|t| now - *t >= BETTER_HOLD_MS) { |
|
| 536 | + | if holds.get(&(hub_key.clone(), j.id)).is_some_and(|t| now - *t >= BETTER_HOLD_MS) { |
|
| 536 | 537 | return false; |
|
| 537 | 538 | } |
|
| 538 | 539 | let mine = rank(&me_now, &rootfs); |
|
| … | 6 unchanged lines |
||
| 545 | 546 | && runner_lacks(j, owner, *repo_id, r, &fleet, &[]).is_none() |
|
| 546 | 547 | }); |
|
| 547 | 548 | if other { |
|
| 548 | - | holds.entry(j.id).or_insert(now); |
|
| 549 | + | holds.entry((hub_key.clone(), j.id)).or_insert(now); |
|
| 549 | 550 | } |
|
| 550 | 551 | other |
|
| 551 | 552 | }; |
|
| … | 3 unchanged lines |
||
| 555 | 556 | return Ok(StatusCode::NO_CONTENT.into_response()); |
|
| 556 | 557 | }; |
|
| 557 | 558 | // Why here (oxygenflow#42). |
|
| 558 | - | let waited = HELD.lock().unwrap_or_else(|p| p.into_inner()).as_mut().and_then(|h| h.remove(&job.id)).map(|t| now - t); |
|
| 559 | + | let waited = |
|
| 560 | + | HELD.lock().unwrap_or_else(|p| p.into_inner()).as_mut().and_then(|h| h.remove(&(hub_key.clone(), job.id))).map(|t| now - t); |
|
| 559 | 561 | let rootfs = rootfs_of(&job); |
|
| 560 | 562 | let mut why = vec![if me_now.image_list().is_some_and(|have| have.contains(&rootfs)) { |
|
| 561 | 563 | "holds the image".to_string() |
|
| … | 288 unchanged lines |
||
| 850 | 852 | const POLLING_MS: i64 = 10_000; |
|
| 851 | 853 | ||
| 852 | 854 | /// When each job was first passed over for a better runner. |
|
| 853 | - | static HELD: std::sync::Mutex<Option<std::collections::HashMap<i64, i64>>> = std::sync::Mutex::new(None); |
|
| 855 | + | /// Keyed by hub (tests run several in one process) and job. |
|
| 856 | + | static HELD: std::sync::Mutex<Option<std::collections::HashMap<(String, i64), i64>>> = std::sync::Mutex::new(None); |
|
| 854 | 857 | ||
| 855 | 858 | /// How good runner `r` is for a job on image `rootfs`: holding the image |
|
| 856 | 859 | /// counts most, then the speed class. |
|
| … | 998 unchanged lines |