Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 15 Packages Wiki Activity Actions 7

joule-moq-relay: reset a stream its lane was killed under; eoc-meter: wait on the child's CPU #531

Merged
dcharlot merged 1 commit from wai/flake-fixes-relay-energy into main 2026-10-29 06:56:17 -04:00
Conversation 0 Commits 1 Files changed 2 +19 -5
2 changed files with 19 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

15
eoc/eoc-rs/crates/eoc-meter/tests/process_energy.rs
View file

@ -190,8 +190,9 @@ fn a_sleeping_span_reads_under_0_1_j() {
}
/// A child process's work is not this process's energy, neither while it
/// runs nor once it is reaped. The child (`yes`, to /dev/null) runs for at
/// least 300 ms and until its own counter has read at least 0.5 J (or 5 s);
/// runs nor once it is reaped. The child (`yes`, to /dev/null) runs until it
/// has used at least 300 ms of CPU and its own counter has read at least 0.5 J (or
/// 10 s);
/// the parent's counter, read before the spawn and after the reap, moves by
/// under 0.1 J and by under a fifth of the child's own.
#[test]
@ -203,10 +204,14 @@ fn a_child_processs_work_is_not_attributed() {
let mut child = Command::new("yes").stdout(Stdio::null()).spawn().expect("spawn yes");
let started = Instant::now();
let mut child_nj = None;
while started.elapsed() < Duration::from_secs(5) {
// Wait on the child's own CPU time, not the wall clock: on a loaded host 300 ms of wall
// time gave it 130 ms of CPU. What is reaped below is at least what is read here.
while started.elapsed() < Duration::from_secs(10) {
std::thread::sleep(Duration::from_millis(50));
child_nj = energy_nj(child.id());
if child_nj.is_some_and(|nj| nj >= 500_000_000) && started.elapsed() >= Duration::from_millis(300) {
let w = rusage(child.id());
child_nj = w.map(|w| w[ENERGY_NJ]);
let cpu = w.map(|w| mach(w[USER_TIME] + w[SYSTEM_TIME]));
if child_nj.is_some_and(|nj| nj >= 500_000_000) && cpu.is_some_and(|c| c >= Duration::from_millis(300)) {
break;
}
}

9
joule-code/joule-code-rs/crates/joule-moq-relay/src/sched.rs
View file

@ -687,6 +687,14 @@ impl Sched {
if let Some(lane) = self.lanes.get_mut(&d.lane) {
lane.busy = false;
lane.send = d.send;
// Killed while this write was in flight (a rise, an unsubscribe): the stream
// was with the write then, not in the lane to reset, and a stream dropped
// unreset is finished whole. It is reset now.
if lane.dead
&& let Some(s) = lane.send.as_mut()
{
s.reset(lane.dead_code.unwrap_or(StreamResetCode::CANCELLED.0));
}
}
self.refresh(d.lane);
}
@ -723,6 +731,7 @@ impl Sched {
for k in keys {
if let Some(lane) = self.lanes.get_mut(&k) {
lane.dead = true;
lane.dead_code.get_or_insert(code);
if let Some(s) = lane.send.as_mut() {
s.reset(code);
}