roundabout

public

Star0

Watch0

10 commits 1 branch 0 tags 0 releases158.8 KiB

use std::{
    env,
    fmt::Write as _,
    os::unix::{net::UnixStream, process::CommandExt},
    process::{Command, ExitStatus},
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicI32, Ordering},
    },
    thread,
    time::{Duration, Instant},
};

use crate::{
    artifacts, client,
    config::{Config, OverCap},
    history, paths,
    proto::{self, Acquire, Event, Reader, Request},
    sched::Need,
    sys,
    units::{Amount, Bytes, Dur},
};

pub const EXIT_TIMEOUT: i32 = 124;
pub const EXIT_GAVE_UP: i32 = 75;

#[derive(Default)]
pub struct Args {
    pub size: Option<String>,
    pub ram: Option<Bytes>,
    pub disk: Option<Bytes>,
    pub cores: Option<Amount<u32>>,
    pub time: Option<Dur>,
    pub deadline: Option<Dur>,
    pub max_wait: Option<Dur>,
    pub session: Option<String>,
    pub exact: bool,
    pub cmd: Vec<String>,
}

static PGID: AtomicI32 = AtomicI32::new(0);
#[cfg(target_os = "linux")]
static FORWARDED: AtomicI32 = AtomicI32::new(0);

extern "C" fn forward(sig: libc::c_int) {
    #[cfg(target_os = "linux")]
    FORWARDED.store(sig, Ordering::Relaxed);
    let pgid = PGID.load(Ordering::Relaxed);
    if pgid > 1 {
        unsafe { libc::kill(-pgid, sig) };
    }
}

fn say(msg: &str) {
    eprintln!("rabt: {msg}");
}

pub fn need(cfg: &Config, a: &Args) -> Result<Need, String> {
    let base = cfg.preset(a.size.as_deref().unwrap_or(&cfg.enforce.default_size))?;
    Ok(Need {
        ram: a.ram.unwrap_or(base.ram),
        disk: a.disk.unwrap_or(base.disk),
        cores: a.cores.unwrap_or(base.cores).resolve(sys::cores()),
        time: a.time.unwrap_or(base.time).0,
    })
}

pub fn base(cfg: &Config) -> Result<Need, String> {
    let p = cfg.preset(&cfg.enforce.default_size)?;
    Ok(Need {
        ram: p.ram,
        disk: p.disk,
        cores: p.cores.resolve(sys::cores()),
        time: p.time.0,
    })
}

struct Grant {
    stream: UnixStream,
    lease: u64,
    cores: u32,
    ram: Bytes,
}

fn acquire(req: &Acquire, max_wait: Option<Dur>) -> Result<Grant, i32> {
    let started = Instant::now();
    // a daemon on its way out can accept us then hang up, so go round again
    for _ in 0..3 {
        let mut stream = client::connect_or_spawn().map_err(|e| {
            say(&format!("can't reach the daemon: {e}"));
            1
        })?;
        if proto::send(&mut stream, &Request::Acquire(req.clone())).is_err() {
            continue;
        }
        let _ = stream.set_read_timeout(Some(Duration::from_secs(1)));
        let mut reader = Reader::new(&stream).map_err(|_| 1)?;
        loop {
            match reader.recv::<Event>() {
                Ok(Some(Event::Granted {
                    lease,
                    cores,
                    ram,
                    heavy,
                })) => {
                    let bonus = cores.saturating_sub(req.need.cores);
                    say(&format!(
                        "lease {lease} go, cores {cores}{}{}",
                        if bonus > 0 {
                            format!(" ({bonus} spare)")
                        } else {
                            String::new()
                        },
                        if heavy { ", heavy" } else { "" }
                    ));
                    let _ = stream.set_read_timeout(None);
                    return Ok(Grant {
                        stream,
                        lease,
                        cores,
                        ram,
                    });
                }
                Ok(Some(Event::Queued {
                    lease,
                    position,
                    reason,
                })) => {
                    let why = reason.map(|r| format!(", {r}")).unwrap_or_default();
                    say(&format!("lease {lease} queued at #{position}{why}"));
                }
                Ok(Some(Event::Kill)) => {
                    say("killed while queued");
                    return Err(143);
                }
                Ok(Some(_)) => {}
                Err(e)
                    if matches!(
                        e.kind(),
                        std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
                    ) =>
                {
                    if max_wait.is_some_and(|m| started.elapsed() > m.0) {
                        say("gave up waiting");
                        return Err(EXIT_GAVE_UP);
                    }
                }
                Ok(None) | Err(_) => break,
            }
        }
    }
    say("daemon keeps hanging up");
    Err(1)
}

fn command(cfg: &Config, a: &Args, grant: &Grant) -> Command {
    let n = grant.cores.to_string();
    let mut cmd = Command::new(&a.cmd[0]);
    cmd.args(&a.cmd[1..])
        .process_group(0)
        .env("ROUNDABOUT_LEASE", grant.lease.to_string())
        .env("CARGO_BUILD_JOBS", &n)
        .env("CMAKE_BUILD_PARALLEL_LEVEL", &n);

    let make = env::var("MAKEFLAGS").unwrap_or_default();
    if !make.contains("-j") {
        cmd.env("MAKEFLAGS", format!("-j{n} {make}").trim_end());
    }

    if cfg.nix.inject {
        let per = cfg.nix.cores_per_job.max(1);
        let jobs = (grant.cores / per).max(1);
        let cores = grant.cores.min(per);
        let mut nix = env::var("NIX_CONFIG").unwrap_or_default();
        let _ = write!(nix, "\ncores = {cores}\nmax-jobs = {jobs}\n");
        cmd.env("NIX_CONFIG", nix);
    }

    cmd
}

fn cpu_time() -> Duration {
    let mut ru = std::mem::MaybeUninit::<libc::rusage>::uninit();
    unsafe {
        if libc::getrusage(libc::RUSAGE_CHILDREN, ru.as_mut_ptr()) != 0 {
            return Duration::ZERO;
        }
        let ru = ru.assume_init();
        let tv = |t: libc::timeval| {
            Duration::from_secs(t.tv_sec as u64) + Duration::from_micros(t.tv_usec as u64)
        };
        tv(ru.ru_utime) + tv(ru.ru_stime)
    }
}

fn code(status: ExitStatus) -> i32 {
    use std::os::unix::process::ExitStatusExt;
    status
        .code()
        .or_else(|| status.signal().map(|s| 128 + s))
        .unwrap_or(1)
}

pub fn run(cfg: &Config, a: &Args) -> i32 {
    if a.cmd.is_empty() {
        say("nothing to run, put the command after --");
        return 2;
    }
    if a.size.is_none() && a.ram.is_none() {
        let sizes: Vec<_> = cfg.sizes.keys().map(String::as_str).collect();
        say(&format!(
            "how big is it? pass --size {} or --ram and friends. `rabt suggest -- <cmd>` has a guess",
            sizes.join("|")
        ));
        return 2;
    }
    let mut need = match need(cfg, a) {
        Ok(n) => n,
        Err(e) => {
            say(&e);
            return 2;
        }
    };
    let cwd = env::current_dir().unwrap_or_else(|_| "/".into());
    let root = paths::project_root(&cwd);

    let report = artifacts::check(cfg, Some(&root), false);
    if report.over() {
        let msg = report.describe(cfg);
        if cfg.artifacts.over_cap == OverCap::Block {
            say(&format!("{msg}. clean up first, `rabt artifacts gc --cap`"));
            return EXIT_GAVE_UP;
        }
        say(&format!(
            "{msg}. `rabt artifacts gc --cap` when you get a sec"
        ));
    }

    let req = Acquire {
        need,
        cwd,
        cmd: a.cmd.clone(),
        session: a.session.clone(),
        deadline: a.deadline,
        exact: a.exact,
    };

    let grant = match acquire(&req, a.max_wait.or(cfg.queue.max_wait)) {
        Ok(g) => g,
        Err(code) => return code,
    };
    need.ram = need.ram.max(grant.ram);

    let command = command(cfg, a, &grant);
    #[cfg(target_os = "linux")]
    let (mut child, scope) =
        match crate::systemd::spawn(&command, grant.lease, grant.cores, grant.ram, &cfg.overrun) {
            Ok(spawned) => spawned,
            Err(error) => {
                say(&format!("couldn't start {}: {error}", a.cmd[0]));
                return 127;
            }
        };
    #[cfg(not(target_os = "linux"))]
    let mut command = command;
    #[cfg(not(target_os = "linux"))]
    let mut child = match command.spawn() {
        Ok(c) => c,
        Err(e) => {
            say(&format!("couldn't start {}: {e}", a.cmd[0]));
            return 127;
        }
    };
    let pgid = i32::try_from(child.id()).expect("pid fits in pid_t");
    PGID.store(pgid, Ordering::Relaxed);
    for sig in [libc::SIGINT, libc::SIGTERM, libc::SIGHUP] {
        unsafe { libc::signal(sig, forward as *const () as libc::sighandler_t) };
    }

    let mut out = grant.stream.try_clone().ok();
    if let Some(s) = &mut out {
        let _ = proto::send(s, &Request::Started { pid: pgid });
    }

    let killed = Arc::new(AtomicBool::new(false));
    if let Ok(mut reader) = Reader::new(&grant.stream) {
        let killed = killed.clone();
        thread::spawn(move || {
            while let Ok(Some(ev)) = reader.recv::<Event>() {
                if matches!(ev, Event::Kill) {
                    killed.store(true, Ordering::Relaxed);
                }
            }
        });
    }

    let started = Instant::now();
    let grace = need.time.mul_f64(cfg.overrun.time_grace);
    let loud_ram = Bytes((need.ram.0 as f64 * cfg.overrun.ram_slack) as u64);
    let mut peak = Bytes::ZERO;
    let mut last_sample: Option<Instant> = None;
    let mut term_sent: Option<Instant> = None;
    let mut timed_out = false;
    let mut warned_ram = false;
    let mut warned_time = false;

    let status = loop {
        #[cfg(target_os = "linux")]
        if let signal @ 1.. = FORWARDED.swap(0, Ordering::Relaxed) {
            let _ = scope.signal(signal);
        }
        match child.try_wait() {
            Ok(Some(status)) => break status,
            Ok(None) => {}
            Err(e) => {
                say(&format!("lost track of the child: {e}"));
                return 1;
            }
        }
        let elapsed = started.elapsed();

        if term_sent.is_none() {
            if killed.load(Ordering::Relaxed) {
                say("killed on request");
                term_sent = Some(Instant::now());
            } else if a.deadline.is_some_and(|d| elapsed > d.0) {
                say("deadline hit, killing it");
                timed_out = true;
                term_sent = Some(Instant::now());
            }
            if term_sent.is_some() {
                #[cfg(target_os = "linux")]
                let _ = scope.signal(libc::SIGTERM);
                unsafe { libc::kill(-pgid, libc::SIGTERM) };
            }
        } else if term_sent.is_some_and(|t| t.elapsed() > Duration::from_secs(10)) {
            #[cfg(target_os = "linux")]
            let _ = scope.signal(libc::SIGKILL);
            unsafe { libc::kill(-pgid, libc::SIGKILL) };
        }

        if last_sample.is_none_or(|t| t.elapsed() >= Duration::from_secs(2)) {
            last_sample = Some(Instant::now());
            let rss = sys::group_rss(pgid);
            peak = peak.max(rss);
            if let Some(s) = &mut out {
                let _ = proto::send(s, &Request::Usage { rss });
            }
            if !warned_ram && rss > loud_ram {
                warned_ram = true;
                say(&format!("using {rss} against a {} estimate", need.ram));
            }
        }
        if !warned_time && elapsed > grace {
            warned_time = true;
            say(&format!(
                "{} past its {} estimate and still going",
                Dur(elapsed),
                Dur(need.time)
            ));
        }
        thread::sleep(Duration::from_millis(250));
    };

    PGID.store(0, Ordering::Relaxed);
    #[cfg(target_os = "linux")]
    drop(scope);
    let wall = started.elapsed();
    if let Some(s) = &mut out {
        let _ = proto::send(s, &Request::Done);
    }
    history::record(
        history::key(&root, &a.cmd),
        peak,
        wall,
        cpu_time(),
        grant.cores,
    );
    artifacts::register(cfg, &root, true);
    if peak > loud_ram || wall > grace {
        say(&format!(
            "took {} and peaked at {}, asked for {} and {}. ask for more next time",
            Dur(wall),
            peak,
            Dur(need.time),
            need.ram
        ));
    }
    if timed_out {
        EXIT_TIMEOUT
    } else {
        code(status)
    }
}

1	use std::{
2	env,
3	fmt::Write as _,
4	os::unix::{net::UnixStream, process::CommandExt},
5	process::{Command, ExitStatus},
6	sync::{
7	Arc,
8	atomic::{AtomicBool, AtomicI32, Ordering},
9	},
10	thread,
11	time::{Duration, Instant},
12	};
13
14	use crate::{
15	artifacts, client,
16	config::{Config, OverCap},
17	history, paths,
18	proto::{self, Acquire, Event, Reader, Request},
19	sched::Need,
20	sys,
21	units::{Amount, Bytes, Dur},
22	};
23
24	pub const EXIT_TIMEOUT: i32 = 124;
25	pub const EXIT_GAVE_UP: i32 = 75;
26
27	#[derive(Default)]
28	pub struct Args {
29	pub size: Option<String>,
30	pub ram: Option<Bytes>,
31	pub disk: Option<Bytes>,
32	pub cores: Option<Amount<u32>>,
33	pub time: Option<Dur>,
34	pub deadline: Option<Dur>,
35	pub max_wait: Option<Dur>,
36	pub session: Option<String>,
37	pub exact: bool,
38	pub cmd: Vec<String>,
39	}
40
41	static PGID: AtomicI32 = AtomicI32::new(0);
42	#[cfg(target_os = "linux")]
43	static FORWARDED: AtomicI32 = AtomicI32::new(0);
44
45	extern "C" fn forward(sig: libc::c_int) {
46	#[cfg(target_os = "linux")]
47	FORWARDED.store(sig, Ordering::Relaxed);
48	let pgid = PGID.load(Ordering::Relaxed);
49	if pgid > 1 {
50	unsafe { libc::kill(-pgid, sig) };
51	}
52	}
53
54	fn say(msg: &str) {
55	eprintln!("rabt: {msg}");
56	}
57
58	pub fn need(cfg: &Config, a: &Args) -> Result<Need, String> {
59	let base = cfg.preset(a.size.as_deref().unwrap_or(&cfg.enforce.default_size))?;
60	Ok(Need {
61	ram: a.ram.unwrap_or(base.ram),
62	disk: a.disk.unwrap_or(base.disk),
63	cores: a.cores.unwrap_or(base.cores).resolve(sys::cores()),
64	time: a.time.unwrap_or(base.time).0,
65	})
66	}
67
68	pub fn base(cfg: &Config) -> Result<Need, String> {
69	let p = cfg.preset(&cfg.enforce.default_size)?;
70	Ok(Need {
71	ram: p.ram,
72	disk: p.disk,
73	cores: p.cores.resolve(sys::cores()),
74	time: p.time.0,
75	})
76	}
77
78	struct Grant {
79	stream: UnixStream,
80	lease: u64,
81	cores: u32,
82	ram: Bytes,
83	}
84
85	fn acquire(req: &Acquire, max_wait: Option<Dur>) -> Result<Grant, i32> {
86	let started = Instant::now();
87	// a daemon on its way out can accept us then hang up, so go round again
88	for _ in 0..3 {
89	let mut stream = client::connect_or_spawn().map_err(\|e\| {
90	say(&format!("can't reach the daemon: {e}"));
91	1
92	})?;
93	if proto::send(&mut stream, &Request::Acquire(req.clone())).is_err() {
94	continue;
95	}
96	let _ = stream.set_read_timeout(Some(Duration::from_secs(1)));
97	let mut reader = Reader::new(&stream).map_err(\|_\| 1)?;
98	loop {
99	match reader.recv::<Event>() {
100	Ok(Some(Event::Granted {
101	lease,
102	cores,
103	ram,
104	heavy,
105	})) => {
106	let bonus = cores.saturating_sub(req.need.cores);
107	say(&format!(
108	"lease {lease} go, cores {cores}{}{}",
109	if bonus > 0 {
110	format!(" ({bonus} spare)")
111	} else {
112	String::new()
113	},
114	if heavy { ", heavy" } else { "" }
115	));
116	let _ = stream.set_read_timeout(None);
117	return Ok(Grant {
118	stream,
119	lease,
120	cores,
121	ram,
122	});
123	}
124	Ok(Some(Event::Queued {
125	lease,
126	position,
127	reason,
128	})) => {
129	let why = reason.map(\|r\| format!(", {r}")).unwrap_or_default();
130	say(&format!("lease {lease} queued at #{position}{why}"));
131	}
132	Ok(Some(Event::Kill)) => {
133	say("killed while queued");
134	return Err(143);
135	}
136	Ok(Some(_)) => {}
137	Err(e)
138	if matches!(
139	e.kind(),
140	std::io::ErrorKind::WouldBlock \| std::io::ErrorKind::TimedOut
141	) =>
142	{
143	if max_wait.is_some_and(\|m\| started.elapsed() > m.0) {
144	say("gave up waiting");
145	return Err(EXIT_GAVE_UP);
146	}
147	}
148	Ok(None) \| Err(_) => break,
149	}
150	}
151	}
152	say("daemon keeps hanging up");
153	Err(1)
154	}
155
156	fn command(cfg: &Config, a: &Args, grant: &Grant) -> Command {
157	let n = grant.cores.to_string();
158	let mut cmd = Command::new(&a.cmd[0]);
159	cmd.args(&a.cmd[1..])
160	.process_group(0)
161	.env("ROUNDABOUT_LEASE", grant.lease.to_string())
162	.env("CARGO_BUILD_JOBS", &n)
163	.env("CMAKE_BUILD_PARALLEL_LEVEL", &n);
164
165	let make = env::var("MAKEFLAGS").unwrap_or_default();
166	if !make.contains("-j") {
167	cmd.env("MAKEFLAGS", format!("-j{n} {make}").trim_end());
168	}
169
170	if cfg.nix.inject {
171	let per = cfg.nix.cores_per_job.max(1);
172	let jobs = (grant.cores / per).max(1);
173	let cores = grant.cores.min(per);
174	let mut nix = env::var("NIX_CONFIG").unwrap_or_default();
175	let _ = write!(nix, "\ncores = {cores}\nmax-jobs = {jobs}\n");
176	cmd.env("NIX_CONFIG", nix);
177	}
178
179	cmd
180	}
181
182	fn cpu_time() -> Duration {
183	let mut ru = std::mem::MaybeUninit::<libc::rusage>::uninit();
184	unsafe {
185	if libc::getrusage(libc::RUSAGE_CHILDREN, ru.as_mut_ptr()) != 0 {
186	return Duration::ZERO;
187	}
188	let ru = ru.assume_init();
189	let tv = \|t: libc::timeval\| {
190	Duration::from_secs(t.tv_sec as u64) + Duration::from_micros(t.tv_usec as u64)
191	};
192	tv(ru.ru_utime) + tv(ru.ru_stime)
193	}
194	}
195
196	fn code(status: ExitStatus) -> i32 {
197	use std::os::unix::process::ExitStatusExt;
198	status
199	.code()
200	.or_else(\|\| status.signal().map(\|s\| 128 + s))
201	.unwrap_or(1)
202	}
203
204	pub fn run(cfg: &Config, a: &Args) -> i32 {
205	if a.cmd.is_empty() {
206	say("nothing to run, put the command after --");
207	return 2;
208	}
209	if a.size.is_none() && a.ram.is_none() {
210	let sizes: Vec<_> = cfg.sizes.keys().map(String::as_str).collect();
211	say(&format!(
212	"how big is it? pass --size {} or --ram and friends. `rabt suggest -- <cmd>` has a guess",
213	sizes.join("\|")
214	));
215	return 2;
216	}
217	let mut need = match need(cfg, a) {
218	Ok(n) => n,
219	Err(e) => {
220	say(&e);
221	return 2;
222	}
223	};
224	let cwd = env::current_dir().unwrap_or_else(\|_\| "/".into());
225	let root = paths::project_root(&cwd);
226
227	let report = artifacts::check(cfg, Some(&root), false);
228	if report.over() {
229	let msg = report.describe(cfg);
230	if cfg.artifacts.over_cap == OverCap::Block {
231	say(&format!("{msg}. clean up first, `rabt artifacts gc --cap`"));
232	return EXIT_GAVE_UP;
233	}
234	say(&format!(
235	"{msg}. `rabt artifacts gc --cap` when you get a sec"
236	));
237	}
238
239	let req = Acquire {
240	need,
241	cwd,
242	cmd: a.cmd.clone(),
243	session: a.session.clone(),
244	deadline: a.deadline,
245	exact: a.exact,
246	};
247
248	let grant = match acquire(&req, a.max_wait.or(cfg.queue.max_wait)) {
249	Ok(g) => g,
250	Err(code) => return code,
251	};
252	need.ram = need.ram.max(grant.ram);
253
254	let command = command(cfg, a, &grant);
255	#[cfg(target_os = "linux")]
256	let (mut child, scope) =
257	match crate::systemd::spawn(&command, grant.lease, grant.cores, grant.ram, &cfg.overrun) {
258	Ok(spawned) => spawned,
259	Err(error) => {
260	say(&format!("couldn't start {}: {error}", a.cmd[0]));
261	return 127;
262	}
263	};
264	#[cfg(not(target_os = "linux"))]
265	let mut command = command;
266	#[cfg(not(target_os = "linux"))]
267	let mut child = match command.spawn() {
268	Ok(c) => c,
269	Err(e) => {
270	say(&format!("couldn't start {}: {e}", a.cmd[0]));
271	return 127;
272	}
273	};
274	let pgid = i32::try_from(child.id()).expect("pid fits in pid_t");
275	PGID.store(pgid, Ordering::Relaxed);
276	for sig in [libc::SIGINT, libc::SIGTERM, libc::SIGHUP] {
277	unsafe { libc::signal(sig, forward as *const () as libc::sighandler_t) };
278	}
279
280	let mut out = grant.stream.try_clone().ok();
281	if let Some(s) = &mut out {
282	let _ = proto::send(s, &Request::Started { pid: pgid });
283	}
284
285	let killed = Arc::new(AtomicBool::new(false));
286	if let Ok(mut reader) = Reader::new(&grant.stream) {
287	let killed = killed.clone();
288	thread::spawn(move \|\| {
289	while let Ok(Some(ev)) = reader.recv::<Event>() {
290	if matches!(ev, Event::Kill) {
291	killed.store(true, Ordering::Relaxed);
292	}
293	}
294	});
295	}
296
297	let started = Instant::now();
298	let grace = need.time.mul_f64(cfg.overrun.time_grace);
299	let loud_ram = Bytes((need.ram.0 as f64 * cfg.overrun.ram_slack) as u64);
300	let mut peak = Bytes::ZERO;
301	let mut last_sample: Option<Instant> = None;
302	let mut term_sent: Option<Instant> = None;
303	let mut timed_out = false;
304	let mut warned_ram = false;
305	let mut warned_time = false;
306
307	let status = loop {
308	#[cfg(target_os = "linux")]
309	if let signal @ 1.. = FORWARDED.swap(0, Ordering::Relaxed) {
310	let _ = scope.signal(signal);
311	}
312	match child.try_wait() {
313	Ok(Some(status)) => break status,
314	Ok(None) => {}
315	Err(e) => {
316	say(&format!("lost track of the child: {e}"));
317	return 1;
318	}
319	}
320	let elapsed = started.elapsed();
321
322	if term_sent.is_none() {
323	if killed.load(Ordering::Relaxed) {
324	say("killed on request");
325	term_sent = Some(Instant::now());
326	} else if a.deadline.is_some_and(\|d\| elapsed > d.0) {
327	say("deadline hit, killing it");
328	timed_out = true;
329	term_sent = Some(Instant::now());
330	}
331	if term_sent.is_some() {
332	#[cfg(target_os = "linux")]
333	let _ = scope.signal(libc::SIGTERM);
334	unsafe { libc::kill(-pgid, libc::SIGTERM) };
335	}
336	} else if term_sent.is_some_and(\|t\| t.elapsed() > Duration::from_secs(10)) {
337	#[cfg(target_os = "linux")]
338	let _ = scope.signal(libc::SIGKILL);
339	unsafe { libc::kill(-pgid, libc::SIGKILL) };
340	}
341
342	if last_sample.is_none_or(\|t\| t.elapsed() >= Duration::from_secs(2)) {
343	last_sample = Some(Instant::now());
344	let rss = sys::group_rss(pgid);
345	peak = peak.max(rss);
346	if let Some(s) = &mut out {
347	let _ = proto::send(s, &Request::Usage { rss });
348	}
349	if !warned_ram && rss > loud_ram {
350	warned_ram = true;
351	say(&format!("using {rss} against a {} estimate", need.ram));
352	}
353	}
354	if !warned_time && elapsed > grace {
355	warned_time = true;
356	say(&format!(
357	"{} past its {} estimate and still going",
358	Dur(elapsed),
359	Dur(need.time)
360	));
361	}
362	thread::sleep(Duration::from_millis(250));
363	};
364
365	PGID.store(0, Ordering::Relaxed);
366	#[cfg(target_os = "linux")]
367	drop(scope);
368	let wall = started.elapsed();
369	if let Some(s) = &mut out {
370	let _ = proto::send(s, &Request::Done);
371	}
372	history::record(
373	history::key(&root, &a.cmd),
374	peak,
375	wall,
376	cpu_time(),
377	grant.cores,
378	);
379	artifacts::register(cfg, &root, true);
380	if peak > loud_ram \|\| wall > grace {
381	say(&format!(
382	"took {} and peaked at {}, asked for {} and {}. ask for more next time",
383	Dur(wall),
384	peak,
385	Dur(need.time),
386	need.ram
387	));
388	}
389	if timed_out {
390	EXIT_TIMEOUT
391	} else {
392	code(status)
393	}
394	}