|
|
… |
46 unchanged lines |
| 47 |
47 |
|
error: Some(CANCELLED.into()), |
| 48 |
48 |
|
peak_disk_mib: None, |
| 49 |
49 |
|
peak_memory_mib: None, |
|
50 |
+ |
phases: Default::default(), |
| 50 |
51 |
|
} |
| 51 |
52 |
|
} |
| 52 |
53 |
|
|
|
|
… |
502 unchanged lines |
| 555 |
556 |
|
/// Fetch `claim`'s image from the hub when this home lacks it |
| 556 |
557 |
|
/// (oxygenflow#39), saying so in the job's log. A hub that does not hold it |
| 557 |
558 |
|
/// leaves it missing, and the preflight declines the job as before. |
| 558 |
|
- |
fn fetch_image(home: &Home, hub: &Hub, cfg: &RunnerConfig, swarm: Option<&crate::swarm::Swarm>, claim: &Claim) { |
|
559 |
+ |
fn fetch_image(home: &Home, hub: &Hub, cfg: &RunnerConfig, swarm: Option<&crate::swarm::Swarm>, claim: &Claim) -> Option<u64> { |
| 559 |
560 |
|
let hash = &claim.sandbox.rootfs; |
| 560 |
561 |
|
if hash.is_empty() || home.rootfs(hash).is_file() || !jet::images::is_hash(hash) { |
| 561 |
|
- |
return; |
|
562 |
+ |
return None; |
| 562 |
563 |
|
} |
| 563 |
564 |
|
let note = |text: String| { |
| 564 |
565 |
|
eprintln!("job {}: {text}", claim.job_id); |
|
|
… |
3 unchanged lines |
| 568 |
569 |
|
static FETCHING: std::sync::Mutex<()> = std::sync::Mutex::new(()); |
| 569 |
570 |
|
let _one = FETCHING.lock().unwrap_or_else(|p| p.into_inner()); |
| 570 |
571 |
|
if home.rootfs(hash).is_file() { |
| 571 |
|
- |
return; |
|
572 |
+ |
return None; |
| 572 |
573 |
|
} |
| 573 |
574 |
|
let client = match crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref()) { |
| 574 |
575 |
|
Ok(c) => c, |
| 575 |
|
- |
Err(e) => return note(format!("image {}: cannot reach the hub: {e:#}", &hash[..12])), |
|
576 |
+ |
Err(e) => { |
|
577 |
+ |
note(format!("image {}: cannot reach the hub: {e:#}", &hash[..12])); |
|
578 |
+ |
return None; |
|
579 |
+ |
} |
| 576 |
580 |
|
}; |
|
581 |
+ |
let fetching = Instant::now(); |
| 577 |
582 |
|
let fetched = match swarm { |
| 578 |
583 |
|
Some(sw) => { |
| 579 |
584 |
|
note(format!("image {} is not on this runner: fetching it from its peers and the hub", &hash[..12])); |
|
|
… |
4 unchanged lines |
| 584 |
589 |
|
client.fetch(home, hash) |
| 585 |
590 |
|
} |
| 586 |
591 |
|
}; |
|
592 |
+ |
let took = fetching.elapsed().as_millis() as u64; |
| 587 |
593 |
|
match fetched { |
| 588 |
594 |
|
Ok(Some(m)) => note(format!( |
| 589 |
595 |
|
"image {}: fetched {} chunks, {}{} in {:.1} s", |
|
|
… |
6 unchanged lines |
| 596 |
602 |
|
Ok(None) => note(format!("image {}: the hub does not hold it either", &hash[..12])), |
| 597 |
603 |
|
Err(e) => note(format!("image {}: fetch failed: {e:#}", &hash[..12])), |
| 598 |
604 |
|
} |
|
605 |
+ |
Some(took) |
| 599 |
606 |
|
} |
| 600 |
607 |
|
|
| 601 |
608 |
|
/// Why this runner cannot run `claim` now (spec §11.5), checked before |
|
|
… |
340 unchanged lines |
| 942 |
949 |
|
let (home, hub, cfg, used, key, held, swarm) = (home, &hub, &cfg, &used, &key, &held, swarm.as_ref()); |
| 943 |
950 |
|
scope.spawn(move || { |
| 944 |
951 |
|
let id = claim.job_id; |
| 945 |
|
- |
fetch_image(home, hub, cfg, swarm, &claim); |
|
952 |
+ |
let image_ms = fetch_image(home, hub, cfg, swarm, &claim); |
| 946 |
953 |
|
if let Some(why) = preflight(home, &claim) { |
| 947 |
954 |
|
eprintln!("job {id}: declined: {why}"); |
| 948 |
955 |
|
crate::metrics::DECLINED.fetch_add(1, std::sync::atomic::Ordering::Relaxed); |
|
|
… |
14 unchanged lines |
| 963 |
970 |
|
claim.heads.first().map(|h| &h[..12]).unwrap_or("-") |
| 964 |
971 |
|
); |
| 965 |
972 |
|
crate::metrics::RUNNING.fetch_add(1, std::sync::atomic::Ordering::Relaxed); |
| 966 |
|
- |
let finish = run_claim(home, hub, cfg, key, &claim).unwrap_or_else(|e| Finish { |
|
973 |
+ |
let finish = run_claim(home, hub, cfg, key, &claim, image_ms).unwrap_or_else(|e| Finish { |
| 967 |
974 |
|
exit_code: None, |
| 968 |
975 |
|
duration_ms: 0, |
| 969 |
976 |
|
receipt: None, |
|
|
… |
1 unchanged lines |
| 971 |
978 |
|
error: Some(format!("{e:#}")), |
| 972 |
979 |
|
peak_disk_mib: None, |
| 973 |
980 |
|
peak_memory_mib: None, |
|
981 |
+ |
phases: Default::default(), |
| 974 |
982 |
|
}); |
| 975 |
983 |
|
eprintln!( |
| 976 |
984 |
|
"job {id}: exit {:?}{}", |
|
|
… |
280 unchanged lines |
| 1257 |
1265 |
|
format!("{:.1} s", d.as_secs_f64()) |
| 1258 |
1266 |
|
} |
| 1259 |
1267 |
|
|
| 1260 |
|
- |
fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim) -> Result<Finish> { |
|
1268 |
+ |
/// `image_ms`: how long fetching its image took, when the runner lacked it. |
|
1269 |
+ |
fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, key: &RunnerKey, claim: &Claim, image_ms: Option<u64>) -> Result<Finish> { |
| 1261 |
1270 |
|
let started = Instant::now(); |
| 1262 |
1271 |
|
let mut sink = LogSink { |
| 1263 |
1272 |
|
hub, |
|
|
… |
173 unchanged lines |
| 1437 |
1446 |
|
); |
| 1438 |
1447 |
|
} |
| 1439 |
1448 |
|
let timeout = Duration::from_secs(claim.timeout_s); |
|
1449 |
+ |
// Where the time goes (oxygenflow#35): what came before the |
|
1450 |
+ |
// sandbox, the sandbox, and what comes after. |
|
1451 |
+ |
let prepare_ms = started.elapsed().as_millis() as u64; |
|
1452 |
+ |
let sandbox_started = Instant::now(); |
| 1440 |
1453 |
|
let mut outcome = if wasm_tier { |
| 1441 |
1454 |
|
crate::wasm::run(&claim.sandbox, &job, &scratch.join("tree"), &scratch, timeout, &mut sink)? |
| 1442 |
1455 |
|
} else if syscall_tier { |
|
|
… |
2 unchanged lines |
| 1445 |
1458 |
|
} else { |
| 1446 |
1459 |
|
microvm::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("workspace.ext4"), &scratch, timeout, &mut sink)? |
| 1447 |
1460 |
|
}; |
|
1461 |
+ |
let run_ms = sandbox_started.elapsed().as_millis() as u64; |
| 1448 |
1462 |
|
if outcome.cancelled { |
| 1449 |
1463 |
|
// Handed back as the runner leaves: its checkpoints first, for |
| 1450 |
1464 |
|
// the next attempt (oxygenflow#77). |
|
|
… |
117 unchanged lines |
| 1568 |
1582 |
|
Ok(None) => {} |
| 1569 |
1583 |
|
Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })), |
| 1570 |
1584 |
|
} |
|
1585 |
+ |
} |
|
1586 |
+ |
let total_ms = started.elapsed().as_millis() as u64; |
|
1587 |
+ |
let mut phases: std::collections::BTreeMap<String, u64> = |
|
1588 |
+ |
[("prepare", prepare_ms), ("run", run_ms), ("finish", total_ms.saturating_sub(prepare_ms + run_ms))] |
|
1589 |
+ |
.into_iter() |
|
1590 |
+ |
.map(|(k, v)| (k.to_string(), v)) |
|
1591 |
+ |
.collect(); |
|
1592 |
+ |
if let Some(ms) = image_ms { |
|
1593 |
+ |
phases.insert("image".into(), ms); |
| 1571 |
1594 |
|
} |
| 1572 |
1595 |
|
Ok(Finish { |
|
1596 |
+ |
phases, |
| 1573 |
1597 |
|
exit_code: outcome.exit_code, |
| 1574 |
|
- |
duration_ms: started.elapsed().as_millis() as u64, |
|
1598 |
+ |
duration_ms: total_ms, |
| 1575 |
1599 |
|
receipt, |
| 1576 |
1600 |
|
signed_receipt, |
| 1577 |
1601 |
|
error: output_error.or_else(|| outcome.timed_out.then(|| "stopped at the time limit".to_string())), |
|
|
… |
482 unchanged lines |
| 2060 |
2084 |
|
error: None, |
| 2061 |
2085 |
|
peak_disk_mib: None, |
| 2062 |
2086 |
|
peak_memory_mib: None, |
|
2087 |
+ |
phases: Default::default(), |
| 2063 |
2088 |
|
}; |
| 2064 |
2089 |
|
let quick = Duration::from_millis(10); |
| 2065 |
2090 |
|
let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]); |
|
|
… |
236 unchanged lines |