diff --git a/Cargo.lock b/Cargo.lock index a3d6d18..65f456b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -243,7 +243,7 @@ dependencies = [ [[package]] name = "qex" -version = "0.7.1" +version = "0.8.1" dependencies = [ "anyhow", "clap", diff --git a/Cargo.toml b/Cargo.toml index 60dfe19..9bd5439 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "qex" -version = "0.7.1" +version = "0.8.1" edition = "2021" description = "Queued EXecutor — a resource-aware local job queue for long-running tasks" license = "Apache-2.0" diff --git a/README.md b/README.md index db0234c..6fe4337 100644 --- a/README.md +++ b/README.md @@ -149,6 +149,36 @@ coordinated. which limits the damage, but an accurate claim is better. qex measures each job and uses the measurement for the next job of the same command. +## When the kernel stops a job for memory + +A training run with `--mem guess` that the kernel stops at hour four gets the +state `oom`. That state says one thing: **the claim was too small**. + +qex corrects it. It multiplies the claim, starts the job again with the same id +and the same record, and says in the record what it did: + +``` +claim: 1 core(s), 16GB (qex raised it, because the earlier claim was too small) +note: the kernel stopped attempt 1 of this job, because the job used more + memory than its claim of 8GB. THE CLAIM WAS TOO SMALL. qex raised + the claim to 16GB and starts the job again. +``` + +The job goes through the queue again, so its new claim meets the budget in the +same way as a new job. qex also keeps the lesson: that kill says that the +command needs **more than** 8GB, so the next run of the same command starts +above that value and does not die in the same way. + +qex makes this correction when it applied the memory limit itself, with +`[enforce] mode`. The kernel then stopped the job at the claim, and the kill is +proof. With no limit — the default — qex can read the count of the login session +only, and that count also rises when the kernel stops a different program of the +same user. qex then reports the state `oom`, says what you can do, and starts no +new attempt: the machine can be full while your claim is correct. + +See [the reference](docs/reference.md#a-job-that-the-kernel-stops-for-memory) +for the limit on the raises. + ## The documentation The full documentation is at diff --git a/docs/design.md b/docs/design.md index c12b489..b946b72 100644 --- a/docs/design.md +++ b/docs/design.md @@ -177,9 +177,18 @@ This is deliberate, and the reason is the job that climbs for hours: a claim that is too small stops the job, and a claim that is a little large costs capacity only. -qex records a job that COMPLETED only. A job that somebody stopped, or that the -out-of-memory killer stopped, shows the memory that it reached and not the -memory that it needs, and that number would teach qex the wrong size. +qex records a job that COMPLETED, and a job that the KERNEL STOPPED AT THE LIMIT +THAT QEX APPLIED. The second kind is a lower bound and not a peak: the job did +not finish, so the true need is above the value. qex records that bound when it +made the cgroup of the job and read the counter of that cgroup; the counter of +the login session also counts a kill in a different program of the same user, +and the machine can be full while the claim of the job is correct. qex keeps the two kinds apart in `usage.json`, and +the next claim goes above the largest of both. A lower bound costs a whole run +to obtain, and an average with the smaller runs would lose it. + +qex records nothing else. A job that somebody stopped, or that reached its time +limit, shows the memory that it reached and not the memory that it needs, and +that number would teach qex the wrong size. ### Is there fairness between agents? diff --git a/docs/reference.md b/docs/reference.md index 9cfce25..c660ae5 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -75,11 +75,88 @@ qex submit -- cargo test # run 2: the claim comes from run 1 the name, because `cargo build` and `cargo test` need different sizes. qex uses the **largest** measurement it holds plus a margin, because a claim that is too small stops the job while a claim that is a little large costs only capacity. -A job that did not complete is never recorded: it shows the memory it reached, -not the memory it needs. + +qex records two kinds of measurement, and it keeps them apart: + +| The job | What the sample says | +| ------- | -------------------- | +| completed | The memory that the job needs. | +| the kernel stopped it at its own limit | A **lower bound**. The need is above this value. | + +qex keeps one lower bound for a command, because a ladder of attempts makes one +at each step and they would fill the store. The learned claim also never goes +above `[budget] mem`: qex makes that number itself, and it must not make a +number that it then refuses. + +A lower bound costs a whole run to obtain, so qex never averages it away. The +next claim is above it, and a smaller run that succeeds later does not remove +it. A job that you stopped, or that reached its time limit, is never recorded: +it shows the memory it reached, not the memory it needs. Turn it off with `[learn] enabled = false`. +### A job that the kernel stops for memory + +The kernel stops a job that uses more memory than its claim, and the job gets +the state `oom`. That state says one thing: **the claim was too small**. + +qex corrects it. It multiplies the claim, starts the job again with the same id +and the same record, and writes in the record what it did: + +``` +$ qex status $ID +state: completed +claim: 1 core(s), 16GB (qex raised it, because the earlier claim was too small) +note: the kernel stopped attempt 1 of this job, because the job used more + memory than its claim of 8GB. THE CLAIM WAS TOO SMALL. qex raised + the claim to 16GB and starts the job again. +attempts: 2 +``` + +| Rule | Value | +| ---- | ----- | +| How many raises | `[retry] on_oom`, 2 by default. | +| The multiplier | `[retry] growth`, 2.0 by default. | +| The limit | The claim never goes above `[budget] mem`. | +| The count | Separate from `--retries`, which stays for your own faults. | + +Each attempt costs the full time of the job, which is why the ladder has a +limit. A job that stops for memory at the limit keeps the state `oom`, and the +record tells you to give a larger `--mem` value or to use a larger machine. + +The job goes through the **queue** again. Its claim is now larger, and the queue +never admitted that claim, so qex tests it against the budget in the same way as +a new job. A raised job thus waits while other jobs hold the budget, and the sum +of the claims stays inside the budget. + +#### When qex acts, and when it only reports + +qex finds a kill for memory with the count in `memory.events`, which Linux keeps +for each cgroup. The counter of a cgroup counts the kills in each cgroup below +it, so **where qex reads it decides what qex may do**: + +| `[enforce] mode` | What qex reads | What qex does | +| ---------------- | -------------- | ------------- | +| `soft` or `hard` | The cgroup that qex made for this job. | Reports `oom`, raises the claim, runs the job again, and teaches the learner. | +| `off` (default) | The cgroup of your login session. | Reports `oom` and says what you can do. It starts no new attempt and teaches the learner nothing. | + +With no limit, the count also rises when the kernel stops a **different program +of the same user**. A machine that is short of memory is also the machine on +which a person uses `kill -9`, so the two events arrive together. That evidence +does not prove that the claim of this job was too small: the machine can be full +while the claim is correct. qex therefore reports the state and stops. + +To get the correction, set `[enforce] mode`. The kernel then stops the job at +the claim, and a kill is proof that the claim was too small. + +`qex kill` writes a mark before it sends the signal, and that mark always wins. +A job that you stopped never runs again with a larger claim, and it teaches the +learner nothing. + +On a machine with no cgroup, such as macOS, qex has no count. A `SIGKILL` that +no qex command sent then gives the state `killed`, which starts no new attempt, +and the record says that qex could not tell the cause. + Do not run a small test job to measure a task. Give `guess` and start the real task. qex measures each job, and you can read the true use later: @@ -234,6 +311,10 @@ max_pressure = 20 # maximum PSI memory pressure (Linux only) [queue] oversized = "run-when-idle" # run-when-idle, reject or queue +[retry] +on_oom = 2 # times to raise the claim after a kill for memory +growth = 2.0 # the multiplier for the claim at each raise + [defaults] cpu = 1 # the default is 1 core mem = "2GB" # the default is the machine memory / the core count diff --git a/src/commands.rs b/src/commands.rs index 3f9031b..5e18b0f 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -385,6 +385,9 @@ fn print_status(s: &JobStatus, show_env: bool) -> Result<()> { // calculated it from the earlier jobs, and that no agent chose it. "learned" => " (from the earlier jobs of this command)", "default" => " (the default; give --cpu and --mem to change it)", + // A claim that qex raised must say so. Without this text, a reader + // sees a number that no agent gave and no measurement produced. + "raised" => " (qex raised it, because the earlier claim was too small)", _ => "", } ); @@ -416,7 +419,15 @@ fn print_status(s: &JobStatus, show_env: bool) -> Result<()> { println!("waits for: {r}"); } if let Some(e) = &s.error { - println!("error: {e}"); + // A job that succeeded can also hold a text here. qex writes the story + // of an out-of-memory kill in this field, and that story stays in the + // record after a later attempt succeeded. The word `error` would then + // contradict the state, so this line uses the word `note`. + if s.state == JobState::Completed { + println!("note: {e}"); + } else { + println!("error: {e}"); + } } if s.attempts > 1 || s.retries_left > 0 { println!( @@ -1261,13 +1272,20 @@ fn describe_result(s: &JobStatus) -> String { }, JobState::Killed => "a command stopped the job".to_string(), JobState::Timeout => "the job reached its time limit".to_string(), - JobState::Oom => { + // Say that the CLAIM was too small, and not that the machine was full. + // + // The words "the machine ran out of memory" sent the reader to the + // machine, and the fault was in the claim. qex holds the full story in + // the error field, with the claim that it tried, so give that text when + // qex wrote it. + JobState::Oom => s.error.clone().unwrap_or_else(|| { format!( - "the machine ran out of memory. The job claimed {} and used {}.", + "the kernel stopped the job for memory. The claim of {} was too small, and the \ + job reached {}. Give a larger `--mem` value.", format_size(s.mem), format_size(s.usage.max_rss) ) - } + }), JobState::Cancelled => "the job left the queue".to_string(), // Give the cause here. A reader of the last job of a pipeline then // learns which job failed, with no other command. @@ -1963,6 +1981,21 @@ pub fn rerun(args: cli::RerunArgs) -> Result { let mut spec = crate::job::read_spec(&dir) .with_context(|| format!("reading the specification of the job {id}"))?; + // Use the claim IN FORCE, and not the claim of the submission. + // + // The record holds the claim that the job had at the end. That value is the + // value of the specification, except after a kill for memory: qex then + // raised the claim, and the job succeeded at the larger value. A rerun from + // the specification would repeat the claim that the kernel already stopped, + // and the correction that cost a whole run would go away. + if let Ok(status) = crate::job::read_status(&dir) { + if status.mem > spec.mem { + spec.mem = status.mem; + spec.cpu = status.cpu.max(spec.cpu); + spec.claim_source = status.claim_source.clone(); + } + } + // A new job needs a new id, and it must not keep the dependencies of the // first job: those jobs have stopped, and a dependency on a job that // succeeded is not correct. @@ -2404,6 +2437,7 @@ mod tests { locks: vec![], attempts: 1, retries_left: 0, + oom_raises: 0, caused_by: None, tags: vec![], } @@ -2465,6 +2499,19 @@ mod tests { // The text must give the claim and the true use. An agent then corrects // its claim from this line. assert!(text.contains("1GB") && text.contains("2GB"), "got: {text}"); + // It must also name the CLAIM as the fault. The words "the machine ran + // out of memory" sent the reader to the machine, and the fault was in + // the claim. + assert!(text.contains("too small"), "got: {text}"); + + // qex writes the full story in the error field: the claim that failed, + // the new claim, and the attempt. That text must win, because it says + // more than the line above. + s.error = Some("qex raised the claim to 2GB and starts the job again".into()); + assert!( + describe_result(&s).contains("raised the claim"), + "the record of qex must win" + ); } #[test] diff --git a/src/config.rs b/src/config.rs index ea2e22c..d442202 100644 --- a/src/config.rs +++ b/src/config.rs @@ -289,6 +289,39 @@ impl Default for LearnConfig { } } +/// Controls what qex does when the kernel stops a job for memory. +#[derive(Debug, Clone, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +pub struct RetryConfig { + /// The number of times that qex raises the claim and starts the job again. + /// + /// This count is SEPARATE from `--retries`, and the reason is important. + /// `--retries` is for a fault outside the task, such as a network that is + /// not ready, and the user chose that number for that fault. A kill for + /// memory is a fault of the CLAIM, and qex made the claim in the usual + /// case: `--mem guess` and the learned claim both come from qex. qex must + /// therefore correct its own fault, and it must not spend a budget that the + /// user gave for a different purpose. A job with no `--retries` value thus + /// still gets this correction. + /// + /// The count has a limit, and 2 raises give 4 times the first claim. Each + /// attempt costs the full time of the job: the job in the README runs for + /// four hours before the kernel stops it. A ladder with no limit can thus + /// use a day of the machine and give no result. + pub on_oom: u32, + /// The multiplier for the claim at each raise. + pub growth: f64, +} + +impl Default for RetryConfig { + fn default() -> Self { + Self { + on_oom: 2, + growth: 2.0, + } + } +} + #[derive(Debug, Clone, Default, Deserialize, Serialize)] #[serde(default, deny_unknown_fields)] pub struct Config { @@ -300,6 +333,7 @@ pub struct Config { pub submit: SubmitConfig, pub defaults: DefaultsConfig, pub learn: LearnConfig, + pub retry: RetryConfig, pub history: HistoryConfig, pub gc: GcConfig, } @@ -430,6 +464,13 @@ impl Config { self.learn.margin ); } + if self.retry.on_oom > 0 && self.retry.growth <= 1.0 { + anyhow::bail!( + "config [retry] growth is {}. Use a value above 1.0. A smaller value gives the \ + claim that the kernel already stopped, and the job would stop again.", + self.retry.growth + ); + } if self.enforce.mem_overcommit < 1.0 { anyhow::bail!( "config [enforce] mem_overcommit is {}. Use a value of 1.0 or more. \ @@ -519,6 +560,10 @@ stale_after = "30s" oversized = "run-when-idle" settle = "3s" +[retry] +on_oom = 2 +growth = 2.0 + [submit] env_capture = "minimal" minimal_env = ["PATH", "HOME"] @@ -531,6 +576,7 @@ timeout = "0" let c: Config = toml::from_str(text).unwrap(); c.validate().unwrap(); assert_eq!(c.enforce.mode, EnforceMode::Soft); + assert_eq!(c.retry.on_oom, 2); assert_eq!(c.submit.env_capture, EnvCapture::Minimal); assert_eq!(c.budget_mem().unwrap(), 20 << 30); assert_eq!(c.default_timeout().unwrap(), None); diff --git a/src/daemon.rs b/src/daemon.rs index d00c0c7..b629e8a 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -135,6 +135,31 @@ impl State { pub fn count_state(&self, f: impl Fn(JobState) -> bool) -> usize { self.jobs.values().filter(|j| f(j.status.state)).count() } + + /// Puts a job in the queue, after each job of the same priority or a higher + /// priority. The queue is thus stable. + /// + /// A job enters the queue two times in its life: at the submission, and + /// again after the kernel stopped it for memory and qex raised its claim. + /// The second case must use the same rule as the first, or a job that qex + /// corrects would go in front of the jobs that waited for it. + pub fn enqueue(&mut self, id: uuid::Uuid) { + if self.queue.contains(&id) { + return; + } + let priority = self.jobs.get(&id).map(|j| j.spec.priority).unwrap_or(0); + let pos = self + .queue + .iter() + .position(|other| { + self.jobs + .get(other) + .map(|j| j.spec.priority < priority) + .unwrap_or(false) + }) + .unwrap_or(self.queue.len()); + self.queue.insert(pos, id); + } } /// The coordinator. The threads share this value. @@ -541,7 +566,7 @@ fn handle_submit(coord: &Arc, spec: JobSpec) -> Response { // learns immediately. It does not wait for the job to start. let warning = { let state = coord.state.lock().unwrap(); - match crate::sched::size_check(&state.cfg, &spec) { + match crate::sched::size_check(&state.cfg, spec.cpu, spec.mem) { crate::sched::Size::Fits => None, crate::sched::Size::TooBig(reason) => { use crate::config::OversizedPolicy; @@ -591,7 +616,6 @@ fn handle_submit(coord: &Arc, spec: JobSpec) -> Response { { let mut state = coord.state.lock().unwrap(); - let priority = spec.priority; status.sequence = state.next_sequence; state.next_sequence += 1; state.jobs.insert( @@ -603,20 +627,7 @@ fn handle_submit(coord: &Arc, spec: JobSpec) -> Response { }, ); - // Put the job in the queue after each job of the same priority or a - // higher priority. The queue is thus stable. - let pos = state - .queue - .iter() - .position(|other| { - state - .jobs - .get(other) - .map(|j| j.spec.priority < priority) - .unwrap_or(false) - }) - .unwrap_or(state.queue.len()); - state.queue.insert(pos, id); + state.enqueue(id); } // Keep a short record of this job, so a reader can tell "the record was diff --git a/src/enforce.rs b/src/enforce.rs index 986143e..d98dfd3 100644 --- a/src/enforce.rs +++ b/src/enforce.rs @@ -313,18 +313,126 @@ pub fn oom_count(_cgroup: &Path) -> u64 { /// signal is the signal that `qex kill` sends, so this test separates the two /// causes. The states `oom` and `killed` need different corrections. pub fn was_oom_killed(job_dir: &Path) -> bool { - if job_dir.join("oom").exists() { - return true; + oom_evidence(job_dir).is_some() +} + +/// How well qex knows that the kernel stopped a job for memory. +/// +/// The two values need different answers, and the difference between them is +/// the difference between a report and an action. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OomScope { + /// qex made the cgroup of this job and read the counter of THAT cgroup. + /// + /// The count belongs to this job and to no other program, and the kernel + /// stopped the job at the limit that qex made from the claim. The claim was + /// therefore too small, and qex can act on that. + Job, + /// qex read the counter of the session, because it made no cgroup. + /// + /// The counter of a cgroup counts the kills in each cgroup below it, so it + /// also counts a kill in a different program of the same user. A machine + /// that is short of memory is also the machine on which a person uses + /// `kill -9`, so the two events arrive together. + /// + /// This evidence is sufficient to REPORT the state `oom`. It is not + /// sufficient to run the job again with a larger claim, and it is not + /// sufficient to teach the learner: the claim can be correct, and the + /// machine full. + Session, +} + +impl OomScope { + fn as_str(self) -> &'static str { + match self { + Self::Job => "job", + Self::Session => "session", + } } +} + +/// Gives the evidence that qex holds for a kill for memory. +pub fn oom_evidence(job_dir: &Path) -> Option { + if let Ok(text) = std::fs::read_to_string(job_dir.join("oom")) { + return Some(match text.trim() { + "job" => OomScope::Job, + // A record from an earlier version of qex holds `1` and names no + // scope. Read it as the weaker evidence: qex then reports the state + // and starts no new attempt, which is the safe answer. + _ => OomScope::Session, + }); + } + // qex made the cgroup of the job, so the counter of that cgroup counts the + // kills of this job only. match job_cgroup_path(job_dir) { - Some(cgroup) => cgroup_had_oom(&cgroup), - None => false, + Some(cgroup) if cgroup_had_oom(&cgroup) => Some(OomScope::Job), + _ => None, + } +} + +/// Records an out-of-memory event for a job, with the evidence for it. +pub fn mark_oom(job_dir: &Path, scope: OomScope) { + // Keep the stronger evidence. The supervisor makes two tests, and the + // second test can read the counter of the session. + if oom_evidence(job_dir) == Some(OomScope::Job) { + return; } + std::fs::write(job_dir.join("oom"), scope.as_str().as_bytes()).ok(); +} + +/// Deletes the out-of-memory record of a job. +/// +/// The record belongs to ONE attempt. qex starts the job again with a larger +/// claim after such a kill, and a record that stays would make the next attempt +/// an out-of-memory kill as well, whatever stopped it. +pub fn clear_oom(job_dir: &Path) { + std::fs::remove_file(job_dir.join("oom")).ok(); +} + +/// Records that a command stopped this job. +/// +/// `qex kill` writes this mark BEFORE it sends the signal. +/// +/// The kernel and `qex kill` both use `SIGKILL`, and the cgroup counter is not +/// exact when qex applies no limit: qex then reads the counter of the session, +/// which also counts a kill in a different program of the same user. A job that +/// a person stopped must NEVER look like an out-of-memory kill, because qex +/// answers an out-of-memory kill with a larger claim and a new attempt. It must +/// not repeat work that somebody stopped on purpose. +/// +/// This mark is thus the first evidence, and it wins against the counter. +pub fn mark_user_kill(job_dir: &Path) { + std::fs::write(job_dir.join("killed-by-user"), b"1").ok(); } -/// Records an out-of-memory event for a job. -pub fn mark_oom(job_dir: &Path) { - std::fs::write(job_dir.join("oom"), b"1").ok(); +/// Tests if a command stopped this job. +pub fn was_user_killed(job_dir: &Path) -> bool { + job_dir.join("killed-by-user").exists() +} + +/// Deletes that mark. +/// +/// The mark belongs to ONE attempt, in the same way as the out-of-memory +/// record. A job with `--retries` can stop with `qex kill` on the first attempt +/// and stop for memory on the second, and a mark that stayed would say that a +/// command stopped an attempt that no command touched. The record would then +/// lose the lesson of the kill for memory. +pub fn clear_user_kill(job_dir: &Path) { + std::fs::remove_file(job_dir.join("killed-by-user")).ok(); +} + +/// Tests if this machine can tell an out-of-memory kill from another kill. +/// +/// Linux counts the kills of the out-of-memory killer in `memory.events`, and +/// every process is in a cgroup, so the evidence is there whether qex applies a +/// limit or not. macOS has no cgroup and no equivalent counter. +/// +/// qex uses this test to say what it does not know. A kill with no evidence +/// gives the state `killed`, which is the safe answer: qex starts no new +/// attempt for it. A reader must learn that qex could not tell, and not believe +/// that qex knew. +pub fn oom_evidence_is_available() -> bool { + own_cgroup().is_some() } /// The name of the variable that stops a second start with systemd. @@ -442,14 +550,51 @@ mod tests { #[test] fn the_out_of_memory_record_is_read_back() { let dir = std::env::temp_dir().join(format!("qex-oom-{}", std::process::id())); + std::fs::remove_dir_all(&dir).ok(); std::fs::create_dir_all(&dir).unwrap(); assert!( !was_oom_killed(&dir), "a new job has no out-of-memory record" ); - mark_oom(&dir); + mark_oom(&dir, OomScope::Session); assert!(was_oom_killed(&dir)); + assert_eq!(oom_evidence(&dir), Some(OomScope::Session)); + + // The stronger evidence replaces the weaker evidence, and the weaker + // evidence never replaces the stronger. qex acts on `Job` only, so a + // second test that overwrote the first would decide the behaviour. + mark_oom(&dir, OomScope::Job); + assert_eq!(oom_evidence(&dir), Some(OomScope::Job)); + mark_oom(&dir, OomScope::Session); + assert_eq!(oom_evidence(&dir), Some(OomScope::Job)); + + clear_oom(&dir); + assert_eq!(oom_evidence(&dir), None); + + // A record from an earlier version of qex names no scope. Read it as + // the weaker evidence: qex then reports the state and starts no new + // attempt, which is the safe answer. + std::fs::write(dir.join("oom"), b"1").unwrap(); + assert_eq!(oom_evidence(&dir), Some(OomScope::Session)); + + std::fs::remove_dir_all(&dir).ok(); + } + + /// The mark of a kill by a command must go away with the attempt that it + /// belongs to. A mark that stayed said that a command stopped an attempt + /// that no command touched. + #[test] + fn the_mark_of_a_kill_by_a_command_can_be_cleared() { + let dir = std::env::temp_dir().join(format!("qex-userkill-{}", std::process::id())); + std::fs::remove_dir_all(&dir).ok(); + std::fs::create_dir_all(&dir).unwrap(); + + assert!(!was_user_killed(&dir)); + mark_user_kill(&dir); + assert!(was_user_killed(&dir)); + clear_user_kill(&dir); + assert!(!was_user_killed(&dir)); std::fs::remove_dir_all(&dir).ok(); } diff --git a/src/help.rs b/src/help.rs index 98d71f5..2d73f4d 100644 --- a/src/help.rs +++ b/src/help.rs @@ -320,8 +320,17 @@ qex uses the LARGEST measurement that it holds, and it adds a margin. A claim that is too small stops the job, and a claim that is a little too large costs some capacity only. -qex records a job that completed only. A job that the out-of-memory killer -stopped shows the memory that it reached, and not the memory that it needs. +qex records a job that completed, and a job that the kernel stopped for memory. +The two give different evidence, and qex keeps them apart: + + - a job that COMPLETED gives the memory that the job needs; + - a job that the KERNEL STOPPED AT ITS OWN LIMIT gives a lower bound. The true + need is above that value, so the next claim is above it as well. A smaller + run that succeeds later does not remove that lesson. qex records this + measurement when it applied the limit itself; see `qex help config`. + +qex records nothing else. A job that you stopped, or that reached its time +limit, shows the memory that it reached and not the memory that it needs. In short: give `guess`, start the task, and read the result. Add an exact claim later, and only if you repeat the task. After the first run of a command, qex @@ -396,6 +405,11 @@ Other useful options every attempt. Use it for a fault outside the task, such as a network that is not ready. + You do not need this option for a job that the kernel + stops for memory. qex raises the claim and starts such + a job again by itself, and that correction does not use + this count. See `qex help config`. + --lock NAME two jobs with one lock name never operate together. Use it for work that shares something that a claim cannot express: a build directory, a port, a database. @@ -636,6 +650,10 @@ that qex uses now. enabled = true # use the earlier jobs of a command as the claim margin = 1.5 # the multiplier for a measurement + [retry] + on_oom = 2 # times to raise the claim after a kill for memory + growth = 2.0 # the multiplier for the claim at each raise + [history] keep = \"1d\" # how long to keep the id of a job after its removal @@ -655,6 +673,33 @@ section gives no value, qex uses 1 core and an equal part of the machine memory. On a machine with 16 cores and 32GB, the default job is 1 core and 2GB. The default job size thus scales with the machine. +A kill for memory +----------------- + +The kernel stops a job that uses more memory than its claim. That kill says one +thing: the claim was too small. qex raises the claim and starts the job again, +up to `[retry] on_oom` times, and it multiplies the claim by `growth` at each +raise. The default values give 2 raises and 4 times the first claim. + +This count is separate from `--retries`. `--retries` is for a fault outside the +task, and you chose that number for that fault. The claim is usually the work of +qex, so qex corrects its own fault and does not spend your count. + +The claim never goes above `[budget] mem`. A job that already claims the whole +budget keeps the state `oom`, and the record says that you need a larger machine +or a larger budget. The job also goes through the QUEUE again, because the queue +never admitted the new claim. + +qex acts on the evidence of THIS JOB only. With `mode = \"soft\"` or +`mode = \"hard\"` above, qex makes a cgroup for each job and reads the count of +that cgroup: the kernel stopped the job at the claim, so the claim was too +small. With `mode = \"off\"`, which is the default, qex reads the count of your +login session, and that count also rises when the kernel stops a DIFFERENT +program of the same user. qex then reports the state `oom` and starts no new +attempt: the machine can be full while the claim of this job is correct. + +Set `[enforce] mode` to get the correction. Set `on_oom = 0` to stop it. + Enforcement ----------- @@ -747,8 +792,9 @@ these conditions are true: The `usage` field gives `max_rss` in bytes and `cpu_secs`. A task that always uses much less than its claim wastes capacity: put an exact claim in a job file, -and more jobs then operate together. A task that stops with an out-of-memory -error needs a larger claim. +and more jobs then operate together. A task that the kernel stops for memory +needs a larger claim, and qex gives it that claim itself: it multiplies the +claim and starts the job again. For one task, this step is not necessary. @@ -775,7 +821,8 @@ qex job states failed the job stopped with an exit code that is not 0. killed the command `qex kill` stopped the job. timeout the job used more time than its `--timeout` value. - oom the out-of-memory killer stopped the job. + oom the kernel stopped the job, because the job used more + memory than its claim. cancelled qex removed the job from the queue before it started. skipped a job that this job needed did not succeed, so this job did not start. The field `caused_by` names the job that @@ -784,8 +831,20 @@ qex job states The states `queued`, `starting` and `running` are not final. Each other state is final and does not change. -The state `oom` is different from `failed`. For `oom`, correct your memory claim -or use a larger machine. +The state `oom` is different from `failed`. It says that the kernel stopped the +job for memory. + +When qex applied the memory limit itself, that kill proves that the claim was +too small: qex raises the claim and starts the job again, up to 2 times. With no +limit, which is the default, qex reads the count of the login session, and that +count also rises for a different program of the same user. qex then reports the +state and starts no new attempt. The record of the job says which of the two +happened, and what you can do. + +The state `killed` also covers a kill that qex cannot explain. The kernel and +`qex kill` both use the signal KILL, and a machine with no cgroup keeps no count +of the kills for memory. qex then gives `killed`, which starts no new attempt, +and the record says that qex could not tell. Use `qex list --state running` to select the jobs in one state. "; diff --git a/src/job.rs b/src/job.rs index 5e0bcaa..b7877fd 100644 --- a/src/job.rs +++ b/src/job.rs @@ -22,13 +22,20 @@ pub enum JobState { /// The job stopped with an exit code that is not 0, or a signal stopped it. Failed, /// The command `qex kill` stopped the job. + /// + /// qex also gives this state to a kill that it cannot explain. The kernel + /// and `qex kill` both use SIGKILL, and a machine without a cgroup gives no + /// evidence to separate them. `killed` is then the safe answer, because qex + /// starts no new attempt for it. Killed, /// The job used more time than the `--timeout` value. Timeout, /// The out-of-memory killer or the cgroup limit stopped the job. /// /// This state is different from `failed`. The correction is also different: - /// the memory claim is too small, or the machine is too small. + /// the memory claim is too small, or the machine is too small. qex makes + /// the first correction itself: it raises the claim and starts the job + /// again. See `oom_raises`. Oom, /// qex removed the job from the queue before the job started. Cancelled, @@ -164,11 +171,19 @@ pub struct JobStatus { pub started_at: Option, pub finished_at: Option, pub cpu: u64, + /// The memory claim in force, in bytes. + /// + /// This value starts as the claim in `spec.json`, and it can become larger: + /// qex raises it after the kernel stops the job for memory. The + /// specification keeps the value that the user asked for, and this field + /// gives the value that the queue and the memory limit use now. pub mem: u64, - /// Where the claim came from: `explicit`, `learned` or `default`. + /// Where the claim came from: `explicit`, `learned`, `default` or `raised`. /// /// A reader can then see that a claim came from a measurement, and it does - /// not look like a value that the agent chose. + /// not look like a value that the agent chose. The value `raised` says that + /// the kernel stopped an earlier attempt for memory, and that qex made the + /// claim larger. #[serde(default)] pub claim_source: String, /// The pipeline that this job belongs to. @@ -219,6 +234,15 @@ pub struct JobStatus { /// The number of times that qex may still start this job again. #[serde(default)] pub retries_left: u32, + /// The number of times that qex raised the memory claim of this job. + /// + /// The kernel stops a job that uses more memory than its claim. That kill + /// says that the claim was too small, so qex multiplies the claim and + /// starts the job again. This count is separate from `retries_left`, + /// because the user gave that number for a different kind of fault. See + /// `[retry] on_oom` in the config file. + #[serde(default)] + pub oom_raises: u32, /// The job that caused this job to stop, for a job in the state `skipped`. /// /// This value names the first job that failed, and not the job before this @@ -262,6 +286,7 @@ impl JobStatus { locks: spec.locks.clone(), attempts: 0, retries_left: spec.retries, + oom_raises: 0, caused_by: None, tags: spec.tags.clone(), } diff --git a/src/lifecycle.rs b/src/lifecycle.rs index 2fa568d..2482d5a 100644 --- a/src/lifecycle.rs +++ b/src/lifecycle.rs @@ -81,6 +81,19 @@ pub fn kill(coord: &Arc, id: uuid::Uuid, signal: i32, grace_secs: u } }; + // Record the cause BEFORE the signal. + // + // The kernel uses SIGKILL for an out-of-memory kill, and this command uses + // the same signal. qex answers an out-of-memory kill with a larger claim + // and a new attempt, so a job that a person stopped must never receive that + // answer: it would repeat work that somebody stopped on purpose. + // + // The mark goes to the disk first, because the supervisor can read the + // record in the moment after the signal. + if let Ok(dir) = paths::job_dir(&id) { + crate::enforce::mark_user_kill(&dir); + } + // Signal the process group. The supervisor put the job in its own group, // so this call reaches each child of the job. let sent = unsafe { libc::killpg(pid, signal) }; diff --git a/src/sched.rs b/src/sched.rs index 2579370..e191388 100644 --- a/src/sched.rs +++ b/src/sched.rs @@ -32,21 +32,29 @@ pub enum Size { /// /// This test uses the budget, not the free capacity. A job that fails this test /// can never start by the normal rule. -pub fn size_check(cfg: &Config, spec: &JobSpec) -> Size { +/// +/// # Which claim the queue uses +/// +/// The caller gives the claim IN FORCE, and not the claim in the specification. +/// The two are the same until the kernel stops the job for memory: qex then +/// raises the claim in the record and gives the job back to the queue. The +/// queue must use the raised claim, because that is the claim that the job +/// holds and that the memory limit applies. A test against the first claim +/// would admit a job of 1GB into the space that qex kept for 600MB. +pub fn size_check(cfg: &Config, cpu: u64, mem: u64) -> Size { let cpu_budget = cfg.budget_cpu().unwrap_or(1); let mem_budget = cfg.budget_mem().unwrap_or(0); let mut reasons = Vec::new(); - if spec.cpu > cpu_budget { + if cpu > cpu_budget { reasons.push(format!( - "the job claims {} cores and the budget is {} cores", - spec.cpu, cpu_budget + "the job claims {cpu} cores and the budget is {cpu_budget} cores" )); } - if spec.mem > mem_budget { + if mem > mem_budget { reasons.push(format!( "the job claims {} of memory and the budget is {}", - format_size(spec.mem), + format_size(mem), format_size(mem_budget) )); } @@ -92,23 +100,24 @@ fn lock_conflict(state: &crate::daemon::State, spec: &JobSpec) -> Option } /// Tests if a job can start now. -fn admit(cfg: &Config, spec: &JobSpec, cpu_used: u64, mem_used: u64) -> Admit { +/// +/// `cpu` and `mem` are the claim IN FORCE. See [`size_check`]. +fn admit(cfg: &Config, cpu: u64, mem: u64, cpu_used: u64, mem_used: u64) -> Admit { let cpu_budget = cfg.budget_cpu().unwrap_or(1); let mem_budget = cfg.budget_mem().unwrap_or(0); // Test 1: the budget of this user. - if cpu_used + spec.cpu > cpu_budget { + if cpu_used + cpu > cpu_budget { return Admit::No(format!( - "waits for cores: {} of {} are in use and the job needs {}", - cpu_used, cpu_budget, spec.cpu + "waits for cores: {cpu_used} of {cpu_budget} are in use and the job needs {cpu}" )); } - if mem_used + spec.mem > mem_budget { + if mem_used + mem > mem_budget { return Admit::No(format!( "waits for memory: {} of {} is in use and the job needs {}", format_size(mem_used), format_size(mem_budget), - format_size(spec.mem) + format_size(mem) )); } @@ -117,13 +126,13 @@ fn admit(cfg: &Config, spec: &JobSpec, cpu_used: u64, mem_used: u64) -> Admit { if cfg.peers.enabled { let peers = crate::peers::claims(cfg); if peers.cpu > 0 || peers.mem > 0 { - if cpu_used + peers.cpu + spec.cpu > cpu_budget { + if cpu_used + peers.cpu + cpu > cpu_budget { return Admit::No(format!( "waits for cores: {} user(s) claim {} cores", peers.count, peers.cpu )); } - if mem_used + peers.mem + spec.mem > mem_budget { + if mem_used + peers.mem + mem > mem_budget { return Admit::No(format!( "waits for memory: {} user(s) claim {}", peers.count, @@ -137,7 +146,7 @@ fn admit(cfg: &Config, spec: &JobSpec, cpu_used: u64, mem_used: u64) -> Admit { // only. It is the test that a program outside qex cannot avoid. let reserve = cfg.reserve_mem().unwrap_or(0); let available = sys::available_memory(); - if available < reserve + spec.mem { + if available < reserve + mem { // Say what this number is, and what it is not. // // A machine can be healthy and still report a small number here: the @@ -153,7 +162,7 @@ fn admit(cfg: &Config, spec: &JobSpec, cpu_used: u64, mem_used: u64) -> Admit { "waits for memory: the machine reports {} that a new program can use, and the job \ needs {} with {} in reserve", format_size(available), - format_size(spec.mem), + format_size(mem), format_size(reserve) ); match sys::memory_pressure() { @@ -424,8 +433,13 @@ fn choose(state: &mut crate::daemon::State) -> Option { continue; } - match size_check(&cfg, &job.spec) { - Size::Fits => match admit(&cfg, &job.spec, cpu_used, mem_used) { + // The claim in force lives in the record, and not in the + // specification. qex raises it after the kernel stops the job for + // memory, and the queue must then use the raised claim. + let (claim_cpu, claim_mem) = (job.status.cpu, job.status.mem); + + match size_check(&cfg, claim_cpu, claim_mem) { + Size::Fits => match admit(&cfg, claim_cpu, claim_mem, cpu_used, mem_used) { Admit::Yes if chosen.is_none() => { chosen = Some(id); break; @@ -535,7 +549,7 @@ fn start_job(coord: &Arc, id: uuid::Uuid) -> anyhow::Result<()> { return Ok(()); } - let forced = match size_check(&state.cfg, &job.spec) { + let forced = match size_check(&state.cfg, job.status.cpu, job.status.mem) { Size::TooBig(reason) => Some(format!( "{reason}. qex started this job alone because no other job operated." )), @@ -661,42 +675,18 @@ mod tests { .unwrap() } - fn spec_with(cpu: u64, mem: u64) -> JobSpec { - JobSpec { - id: uuid::Uuid::new_v4(), - name: "t".into(), - cwd: "/".into(), - command: vec!["true".into()], - env: Default::default(), - cpu, - mem, - timeout: None, - tags: vec![], - priority: 0, - env_capture: crate::config::EnvCapture::None, - claim_source: "explicit".into(), - group: None, - group_name: None, - locks: vec![], - retries: 0, - needs: vec![], - after: vec![], - submitted_at: 0, - } - } - #[test] fn a_job_inside_the_budget_fits() { let cfg = cfg_with("4", "8GB"); - assert_eq!(size_check(&cfg, &spec_with(4, 8 << 30)), Size::Fits); - assert_eq!(size_check(&cfg, &spec_with(1, 1 << 30)), Size::Fits); + assert_eq!(size_check(&cfg, 4, 8 << 30), Size::Fits); + assert_eq!(size_check(&cfg, 1, 1 << 30), Size::Fits); } #[test] fn a_job_larger_than_the_budget_is_too_big() { let cfg = cfg_with("4", "8GB"); - let Size::TooBig(reason) = size_check(&cfg, &spec_with(64, 1 << 30)) else { + let Size::TooBig(reason) = size_check(&cfg, 64, 1 << 30) else { panic!("a job of 64 cores must not fit a budget of 4 cores"); }; assert!( @@ -704,7 +694,7 @@ mod tests { "the reason must name the cores: {reason}" ); - let Size::TooBig(reason) = size_check(&cfg, &spec_with(1, 64 << 30)) else { + let Size::TooBig(reason) = size_check(&cfg, 1, 64 << 30) else { panic!("a job of 64GB must not fit a budget of 8GB"); }; assert!( @@ -714,7 +704,7 @@ mod tests { // A job that is too large in both values must give both reasons. The // agent then corrects the claim one time only. - let Size::TooBig(reason) = size_check(&cfg, &spec_with(64, 64 << 30)) else { + let Size::TooBig(reason) = size_check(&cfg, 64, 64 << 30) else { panic!("this job must not fit"); }; assert!( @@ -740,19 +730,19 @@ mod tests { #[test] fn the_budget_limits_the_jobs_that_operate_together() { let cfg = cfg_with("4", "256MB"); - let job = spec_with(2, 64 << 20); + let (cpu, mem) = (2, 64 << 20); // Two cores are in use. A job of two cores fits. - assert!(matches!(admit(&cfg, &job, 2, 64 << 20), Admit::Yes)); + assert!(matches!(admit(&cfg, cpu, mem, 2, 64 << 20), Admit::Yes)); // Four cores are in use. The same job must wait. - let Admit::No(reason) = admit(&cfg, &job, 4, 64 << 20) else { + let Admit::No(reason) = admit(&cfg, cpu, mem, 4, 64 << 20) else { panic!("a job must not start when the cores are in use"); }; assert!(reason.contains("cores"), "got: {reason}"); // The memory is in use. The job must wait. - let Admit::No(reason) = admit(&cfg, &job, 0, 224 << 20) else { + let Admit::No(reason) = admit(&cfg, cpu, mem, 0, 224 << 20) else { panic!("a job must not start when the memory is in use"); }; assert!(reason.contains("memory"), "got: {reason}"); @@ -767,11 +757,8 @@ mod tests { #[test] fn a_job_that_fills_the_budget_exactly_starts() { let cfg = cfg_with("4", "256MB"); - assert!(matches!( - admit(&cfg, &spec_with(4, 256 << 20), 0, 0), - Admit::Yes - )); - assert_eq!(size_check(&cfg, &spec_with(4, 256 << 20)), Size::Fits); + assert!(matches!(admit(&cfg, 4, 256 << 20, 0, 0), Admit::Yes)); + assert_eq!(size_check(&cfg, 4, 256 << 20), Size::Fits); } /// The reserve keeps memory for the programs that qex does not control. @@ -780,19 +767,49 @@ mod tests { let mut cfg = cfg_with("4", "8GB"); // Ask for a reserve that is larger than the machine. Each job must wait. cfg.system.reserve_mem = "1000GB".into(); - let Admit::No(reason) = admit(&cfg, &spec_with(1, 1 << 20), 0, 0) else { + let Admit::No(reason) = admit(&cfg, 1, 1 << 20, 0, 0) else { panic!("the reserve must stop this job"); }; assert!(reason.contains("reserve"), "got: {reason}"); } + /// The queue must test the claim IN FORCE, and not the claim of the + /// submission. + /// + /// qex raises the claim in the record after the kernel stops a job for + /// memory, and it gives the job back to the queue. A test against the first + /// claim would admit a job of 1GB into the space that qex kept for 600MB, + /// and the sum of the claims would go above the budget. Stopping that is + /// the work of this module. + #[test] + fn the_queue_tests_the_claim_that_the_job_holds_now() { + let cfg = cfg_with("4", "1GB"); + + // A job of 400MB operates. The first claim of 600MB fits beside it. + assert!(matches!( + admit(&cfg, 1, 600 << 20, 1, 400 << 20), + Admit::Yes + )); + + // The raised claim of 1GB does not fit beside it, and it must wait. + let Admit::No(reason) = admit(&cfg, 1, 1 << 30, 1, 400 << 20) else { + panic!("a raised claim must wait for capacity"); + }; + assert!(reason.contains("memory"), "got: {reason}"); + + // The raised claim alone still fits the budget, so the job is not an + // oversized job and it starts when the other job stops. + assert_eq!(size_check(&cfg, 1, 1 << 30), Size::Fits); + assert!(matches!(admit(&cfg, 1, 1 << 30, 0, 0), Admit::Yes)); + } + /// The pressure limit stops a job while the machine reclaims memory. #[test] fn the_pressure_limit_stops_a_job() { let mut cfg = cfg_with("4", "8GB"); cfg.system.max_pressure = -1.0; if sys::memory_pressure().is_some() { - let Admit::No(reason) = admit(&cfg, &spec_with(1, 1 << 20), 0, 0) else { + let Admit::No(reason) = admit(&cfg, 1, 1 << 20, 0, 0) else { panic!("the pressure limit must stop this job"); }; assert!(reason.contains("pressure"), "got: {reason}"); diff --git a/src/schema.rs b/src/schema.rs index 8afe0e0..4a6ec75 100644 --- a/src/schema.rs +++ b/src/schema.rs @@ -218,7 +218,18 @@ pub const STATUS: &str = r##"{ "started_at": { "type": ["integer", "null"], "description": "The start time, in seconds after the Unix epoch." }, "finished_at": { "type": ["integer", "null"], "description": "The stop time, in seconds after the Unix epoch." }, "cpu": { "type": "integer", "description": "The number of cores that the job claimed." }, - "mem": { "type": "integer", "description": "The memory in bytes that the job claimed." }, + "mem": { "type": "integer", "description": "The memory in bytes that the job claims NOW. The value starts as the claim in the submission, and qex raises it when the kernel stops the job for memory." }, + "claim_source": { + "type": "string", + "enum": ["explicit", "learned", "default", "raised"], + "description": "Where the claim came from. `raised` means that the kernel stopped an earlier attempt for memory and qex made the claim larger." + }, + "attempts": { "type": "integer", "description": "The number of times that qex started this job." }, + "retries_left": { "type": "integer", "description": "The number of times that qex may still start this job again after a failure. See --retries." }, + "oom_raises": { + "type": "integer", + "description": "The number of times that qex raised the memory claim of this job, because the kernel stopped an attempt at the memory limit that qex applied. This count is separate from retries_left. qex raises the claim only when it applied the limit itself, with [enforce] mode; with no limit it reports the state oom and starts no new attempt. See [retry] on_oom in the config file." + }, "usage": { "type": "object", "description": "The resources that the job used. Compare these values with the claim, then correct your next claim.", @@ -241,7 +252,7 @@ pub const STATUS: &str = r##"{ }, "error": { "type": ["string", "null"], - "description": "The reason that the job failed, when qex gives the reason. A command that does not exist is the usual cause." + "description": "The reason that the job failed, when qex gives the reason. A command that does not exist is the usual cause. A job that the kernel stopped for memory holds here the claim that failed, the new claim and the attempt, and that text stays in the record when a later attempt succeeds." }, "needs": { "type": "array", diff --git a/src/spec.rs b/src/spec.rs index 3aa22b6..2a4f3ba 100644 --- a/src/spec.rs +++ b/src/spec.rs @@ -307,12 +307,30 @@ impl JobSpec { // This step is the reason that qex measures each job. `guess` is safe // and frequently far too large: a test suite that uses 165MB would hold // one half of the budget and stop other work for the length of the run. - let learned = if cfg.learn.enabled && (asked_cpu.is_none() || asked_mem.is_none()) { + let mut learned = if cfg.learn.enabled && (asked_cpu.is_none() || asked_mem.is_none()) { crate::usage::suggest(&crate::usage::load(), &cwd, &command, cfg.learn.margin) } else { None }; + // A learned claim never goes above the budget. + // + // qex makes this number itself, and it must not make a number that it + // then refuses. A job that the kernel stopped for memory at the budget + // leaves a lower bound AT the budget, and the margin above that bound + // gave a claim of 1.5 budgets. `[queue] oversized = "reject"` then + // refused the submission with "Decrease the claim", and the user had + // given no claim to decrease. + // + // The claim stops at the budget instead. The job then starts, and if it + // still needs more memory, the record says that qex has no larger claim + // and that the machine is too small. That answer names a step that the + // user can take. + if let (Some(s), Ok(budget)) = (learned.as_mut(), cfg.budget_mem()) { + s.mem = s.mem.min(budget); + } + let learned = learned; + let mut source = "default"; let cpu = match asked_cpu { Some(c) => { diff --git a/src/supervisor.rs b/src/supervisor.rs index c98acf3..45ee42c 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -131,6 +131,26 @@ pub fn reap(coord: Arc, id: uuid::Uuid, pid: i32) { Ok(status) if status.state.is_terminal() => { job.status = status; } + // The supervisor gave the job back to the queue. + // + // It does this after the kernel stopped the job for memory: the + // claim is now larger, and the queue never admitted that claim. The + // coordinator must test the new claim against the budget, in the + // same way as a new job. Without this branch the job would go to + // the state `failed` with "the supervisor stopped without a + // result", and the work would stop. + Ok(status) if status.state == JobState::Queued => { + job.status = status; + job.status.supervisor_pid = None; + let claim = job.status.mem; + // Use the rule of the submission, so a job that qex corrects + // does not go in front of the jobs that waited for it. + state.enqueue(id); + log(&format!( + "job {id} is in the queue again, and it waits for capacity for {}", + crate::units::format_size(claim) + )); + } other => { // The supervisor stopped before it wrote a result. Something // stopped it: a signal, or the out-of-memory killer. @@ -229,6 +249,19 @@ pub fn main(id: uuid::Uuid) -> Result { status.supervisor_pid = Some(std::process::id() as i32); job::write_status(&dir, &status).context("writing the job status")?; + // Delete the marks of the attempt before this one. + // + // Each mark says what stopped ONE attempt. This point is safe for both: the + // job has no process id yet, so `qex kill` refuses the job and can write no + // mark, and no job process operates that the kernel can stop. + // + // Without this step, a job with `--retries` that a command stopped on the + // first attempt kept that mark for ever. A second attempt that the kernel + // stopped for memory then said that a command stopped it, and the lesson of + // the kill for memory went away. + crate::enforce::clear_user_kill(&dir); + crate::enforce::clear_oom(&dir); + // The output of a job holds secrets as frequently as its environment, so // these files use the same mode as the job specification. // @@ -302,7 +335,13 @@ pub fn main(id: uuid::Uuid) -> Result { let mut cgroup_dir: Option = None; let mut enforce_warning: Option = None; if cfg.enforce.mode.is_on() { - match crate::enforce::create_job_cgroup(&cfg, &id, spec.mem) { + // Use the claim in the RECORD, and not the claim in the specification. + // + // qex raises the claim in the record after the kernel stops the job for + // memory. A limit from the specification would hold the first claim, + // and the kernel would stop each new attempt at the size that already + // failed. + match crate::enforce::create_job_cgroup(&cfg, &id, status.mem) { Ok(cgroup) => match crate::enforce::add_process(&cgroup, std::process::id() as i32) { Ok(()) => { crate::enforce::record_cgroup_path(&dir, &cgroup); @@ -467,16 +506,27 @@ pub fn main(id: uuid::Uuid) -> Result { // cgroup, so this method finds a process that changed its process group. if let Some(cgroup) = crate::enforce::job_cgroup_path(&dir) { if crate::enforce::cgroup_had_oom(&cgroup) { - crate::enforce::mark_oom(&dir); + // This cgroup belongs to this job, so the count belongs to this job. + crate::enforce::mark_oom(&dir, crate::enforce::OomScope::Job); } crate::enforce::kill_cgroup(&cgroup); } // Test the out-of-memory count again. An increase during this job, with a // SIGKILL that no qex command sent, is the out-of-memory killer. + // + // Say WHICH cgroup gave the count. With a cgroup of its own, the count is + // about this job. Without one, qex reads the cgroup of the session, and the + // counter of a cgroup counts the kills in each cgroup below it: a kill in a + // different program of this user raises the same number. if let Some(cgroup) = &watch_cgroup { if crate::enforce::oom_count(cgroup) > oom_before { - crate::enforce::mark_oom(&dir); + let scope = if cgroup_dir.is_some() { + crate::enforce::OomScope::Job + } else { + crate::enforce::OomScope::Session + }; + crate::enforce::mark_oom(&dir, scope); } } @@ -497,6 +547,15 @@ pub fn main(id: uuid::Uuid) -> Result { let timed_out = outcome.load(std::sync::atomic::Ordering::SeqCst) == RACE_TIMER; status.state = classify(&spec, code, signal, timed_out, &dir); + // Keep an earlier message. The two messages that this field can already + // hold say that the memory limit is NOT active, or that qex could not read + // the configuration. Each of those is the cause of the note below, and not + // a smaller fact than it. + if status.error.is_none() { + if let Some(note) = unexplained_kill_note(status.state, signal, &dir) { + status.error = Some(note); + } + } status.exit_code = code; status.signal = signal; status.finished_at = Some(sys::now_secs()); @@ -508,6 +567,121 @@ pub fn main(id: uuid::Uuid) -> Result { status.last_pid = Some(pid); job::write_status(&dir, &status)?; + // Answer an out-of-memory kill with a larger claim and a new attempt. + // + // This is the case that the README describes: a training run with + // `--mem guess` that the kernel stops at hour four. The claim was too + // small, the claim came from qex, and qex can correct it. + if status.state == JobState::Oom { + // Act on the evidence of THIS JOB only. + // + // qex reads a cgroup counter to find a kill for memory. With a cgroup + // of its own for the job, that counter counts this job and nothing + // else, and the kernel stopped the job at the limit that qex made from + // the claim: the claim was too small, and that is a fact. + // + // With no cgroup of its own, qex reads the counter of the session. That + // counter also counts a kill in a different program of this user, and a + // machine that is short of memory is the machine on which a person uses + // `kill -9`. The two arrive together, so the count alone does not say + // that THIS job was the victim, and it does not say that the claim was + // too small: the machine can be full while the claim is correct. + // + // qex therefore REPORTS the state `oom` on the weaker evidence, and it + // ACTS on the stronger evidence only. A new attempt with a larger claim + // repeats work, holds more of the machine, and teaches the learner a + // number that no measurement supports. + let scope = crate::enforce::oom_evidence(&dir); + if scope != Some(crate::enforce::OomScope::Job) { + let note = format!( + "the kernel stopped this job for memory, and its claim was {}. qex applies no \ + memory limit to a job in this configuration, so it counts the kills of the \ + whole login session and it cannot prove that the claim of this job was too \ + small: the machine can be full while the claim is correct. qex therefore did \ + NOT start the job again. Compare the `usage` field with the claim. Give a \ + larger `--mem` value, or set `[enforce] mode` in the config file, and qex then \ + corrects the claim itself.", + crate::units::format_size(status.mem) + ); + log(&format!("job {id}: {note}")); + status.error = Some(note); + job::write_status(&dir, &status)?; + return Ok(code.unwrap_or(1)); + } + + // Keep the lesson NOW, and not at the end of this function. + // + // The new attempt does not come back to this point: this process gives + // the job to the coordinator and stops. A record at the end would thus + // lose the measurement of every attempt except the last, and the + // measurement of an attempt that the kernel stopped is the most + // valuable measurement that qex holds. + crate::usage::record_lower_bound(&spec, &status); + + match raise_claim(&cfg, &status, spec.mem) { + Raise::To(next) => { + let message = format!( + "the kernel stopped attempt {} of this job, because the job used more \ + memory than its claim of {}. THE CLAIM WAS TOO SMALL. qex raised the \ + claim to {} and starts the job again.", + status.attempts, + crate::units::format_size(status.mem), + crate::units::format_size(next) + ); + log(&format!("job {id}: {message}")); + + status.oom_raises += 1; + status.mem = next; + // Say where this claim came from. A reader of `qex status` then + // sees that qex made the number, and not the agent. + status.claim_source = "raised".to_string(); + status.state = JobState::Queued; + status.error = Some(message); + status.pid = None; + status.finished_at = None; + // This process stops now, so it holds the record no longer. + status.supervisor_pid = None; + job::write_status(&dir, &status)?; + + // The out-of-memory record belongs to the attempt that stopped. + // A record that stays would make the next attempt an + // out-of-memory kill as well, whatever stopped it. + crate::enforce::clear_oom(&dir); + + // GIVE THE JOB BACK TO THE COORDINATOR, and do not start it here. + // + // A retry after a failure keeps the same claim, so this process + // can start the job again itself: the budget that the queue gave + // to this job is still the correct budget. + // + // A retry after a kill for memory has a LARGER claim, and the + // queue never saw that claim. A start from this process would + // put a job of 1GB in a budget that admitted 600MB, beside a + // job that holds the rest, and the sum would be above the + // budget. Stopping exactly that is the work of the queue. With + // `[enforce] mode = "hard"` the kernel would also receive the + // sum of the limits, and the machine would meet the load that + // the budget exists to prevent. + // + // The record says `queued` now. The coordinator reads that + // record when this process stops, puts the job in the queue + // again, and starts it when the machine has capacity for the + // NEW claim. + log(&format!( + "job {id} waits for the queue again, with the claim {}", + crate::units::format_size(next) + )); + return Ok(0); + } + Raise::Stop(reason) => { + log(&format!("job {id}: {reason}")); + status.error = Some(reason); + job::write_status(&dir, &status)?; + return Ok(code.unwrap_or(1)); + } + } + } + // Run the job again when it failed and a retry is left. // // The job keeps one id and one record, so `qex wait` gives the final result @@ -666,6 +840,16 @@ fn classify( return JobState::Timeout; } + // A kill from a command wins against every other test. + // + // `qex kill` writes a mark before it sends the signal. qex answers an + // out-of-memory kill with a larger claim and a NEW ATTEMPT, so a job that a + // person stopped must never look like one: qex would repeat work that + // somebody stopped on purpose, at a larger size. + if signal.is_some() && crate::enforce::was_user_killed(dir) { + return JobState::Killed; + } + // The kernel stops a process with SIGKILL for an out-of-memory event. Read // the cgroup record to separate that event from a `qex kill` command. if signal == Some(libc::SIGKILL) && crate::enforce::was_oom_killed(dir) { @@ -681,6 +865,120 @@ fn classify( } } +/// The decision about the claim for the next attempt after a kill for memory. +enum Raise { + /// qex raises the claim to this value and starts the job again. + To(u64), + /// qex does not start the job again. The text says why, and what to do. + Stop(String), +} + +/// Chooses the claim for the next attempt after the kernel stopped the job. +/// +/// # Why the ladder has a limit +/// +/// A claim that doubles for ever finishes with the whole machine, and each +/// attempt costs the full time of the job. Two rules stop the ladder: +/// +/// 1. A number of raises, from `[retry] on_oom`. Two raises give four times the +/// first claim, which corrects the usual error of an estimate. +/// 2. The memory budget of qex. qex must not claim memory that it does not +/// have, and a claim above the budget makes the job an oversized job, which +/// the queue then starts alone. A job that already claims the full budget +/// has no larger claim available, and the answer for the user is a different +/// machine, and not another attempt. +fn raise_claim(cfg: &crate::config::Config, status: &job::JobStatus, first_claim: u64) -> Raise { + let claim = crate::units::format_size(status.mem); + + if cfg.retry.on_oom == 0 { + return Raise::Stop(format!( + "the kernel stopped this job, because the job used more memory than its claim of \ + {claim}. THE CLAIM WAS TOO SMALL. The config file sets `[retry] on_oom = 0`, so qex \ + did not start the job again. Give a larger claim with `--mem`." + )); + } + + if status.oom_raises >= cfg.retry.on_oom { + return Raise::Stop(format!( + "the kernel stopped this job {} times, because the job used more memory than its \ + claim. THE CLAIM WAS TOO SMALL. qex raised the claim from {} to {claim}, and that \ + claim was also too small. Give a larger claim with `--mem`, or use a machine with \ + more memory.", + status.attempts, + crate::units::format_size(first_claim) + )); + } + + // A budget that qex cannot read must not stop the correction, so a fault + // here gives the claim of this attempt and the rules below then stop the + // ladder. + let budget = cfg.budget_mem().unwrap_or(status.mem); + + // A job that already claims the budget or more has no larger claim. + // + // Say that in its own words. An oversized job is a supported case: the + // queue starts such a job alone. Its claim is ABOVE the budget, so a + // sentence that calls that claim "the whole budget" contradicts itself and + // gives the reader a number that is not the number in the record. + if status.mem >= budget { + return Raise::Stop(format!( + "the kernel stopped this job, because the job used more memory than its claim of \ + {claim}. THE CLAIM WAS TOO SMALL. The memory budget of qex on this machine is {}, \ + so qex has no larger claim to give. Use a machine with more memory, or raise \ + `[budget] mem` in the config file.", + crate::units::format_size(budget) + )); + } + + // Never above the budget. qex must not claim memory that it does not have. + let next = ((status.mem as f64 * cfg.retry.growth) as u64).min(budget); + + // A multiplier a little above 1.0 can give the claim that already failed, + // because the calculation gives whole bytes. A new attempt at that claim + // would stop in the same way and cost a whole run. + if next <= status.mem { + return Raise::Stop(format!( + "the kernel stopped this job, because the job used more memory than its claim of \ + {claim}. THE CLAIM WAS TOO SMALL. The config file sets `[retry] growth = {}`, which \ + gives the same claim again, so qex did not start the job again. Give a larger claim \ + with `--mem`, or raise `growth` in the config file.", + cfg.retry.growth + )); + } + + Raise::To(next) +} + +/// Gives a note for a kill that qex cannot explain. +/// +/// The kernel uses `SIGKILL` for an out-of-memory kill, and `qex kill` uses the +/// same signal. Linux counts the out-of-memory kills in the cgroup, so qex has +/// evidence there. A machine with no cgroup gives none. +/// +/// qex then gives the state `killed`, which is the safe answer: qex starts no +/// new attempt for it. The reader must learn that qex GUESSED, and must not +/// believe that a command stopped the job. +fn unexplained_kill_note( + state: JobState, + signal: Option, + dir: &std::path::Path, +) -> Option { + if state != JobState::Killed || signal != Some(libc::SIGKILL) { + return None; + } + if crate::enforce::was_user_killed(dir) || crate::enforce::oom_evidence_is_available() { + return None; + } + Some( + "the signal KILL stopped this job, and no qex command sent it. This machine keeps no \ + count of the kills for memory, so qex cannot say if the kernel stopped the job for \ + memory or if a different program stopped it. qex gave the state `killed` and did not \ + raise the claim. Compare the `usage` field with the claim, and give a larger `--mem` \ + value if the two are near." + .to_string(), + ) +} + fn exit_signal(exit: &std::process::ExitStatus) -> Option { use std::os::unix::process::ExitStatusExt; exit.signal() @@ -819,6 +1117,227 @@ mod tests { ); } + /// Makes an empty job directory for a test of the classification. + fn job_dir(tag: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!( + "qex-sv-{tag}-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + std::fs::remove_dir_all(&dir).ok(); + std::fs::create_dir_all(&dir).unwrap(); + dir + } + + /// A kill for memory gives the state `oom`, and not the state `killed`. + /// + /// The kernel and `qex kill` both use SIGKILL. Without the record in the + /// job directory, the two causes look the same, and qex would tell the user + /// to correct a claim that was correct. + #[test] + fn a_kill_for_memory_gives_the_state_oom() { + let dir = job_dir("oom"); + crate::enforce::mark_oom(&dir, crate::enforce::OomScope::Job); + assert_eq!( + classify(&spec(), None, Some(libc::SIGKILL), false, &dir), + JobState::Oom + ); + std::fs::remove_dir_all(&dir).ok(); + } + + /// A job that a USER stopped must never look like a kill for memory. + /// + /// This test protects the feature from doing harm. qex answers a kill for + /// memory with a larger claim and a NEW ATTEMPT. A job that somebody + /// stopped on purpose must not run again, and it must not teach the learner + /// that the command needs more memory. + /// + /// The two marks can both exist: the out-of-memory count of a session also + /// counts a kill in a different program of the same user. The mark from the + /// command wins. + #[test] + fn a_job_that_a_user_stopped_is_never_an_out_of_memory_kill() { + let dir = job_dir("userkill"); + crate::enforce::mark_user_kill(&dir); + assert_eq!( + classify(&spec(), None, Some(libc::SIGKILL), false, &dir), + JobState::Killed + ); + + crate::enforce::mark_oom(&dir, crate::enforce::OomScope::Session); + assert_eq!( + classify(&spec(), None, Some(libc::SIGKILL), false, &dir), + JobState::Killed, + "a kill from a command must win against the count of the session" + ); + assert_eq!( + classify(&spec(), None, Some(libc::SIGTERM), false, &dir), + JobState::Killed + ); + std::fs::remove_dir_all(&dir).ok(); + } + + /// A kill that qex cannot explain gives the safe answer, and it says that + /// qex could not tell. A machine with no cgroup counts no kill for memory, + /// so a guess there would send the reader to the wrong correction. + #[test] + fn a_kill_that_qex_cannot_explain_says_so() { + let dir = job_dir("unexplained"); + + // A job that a command stopped needs no note: qex knows the cause. + crate::enforce::mark_user_kill(&dir); + assert_eq!( + unexplained_kill_note(JobState::Killed, Some(libc::SIGKILL), &dir), + None + ); + std::fs::remove_file(dir.join("killed-by-user")).unwrap(); + + let note = unexplained_kill_note(JobState::Killed, Some(libc::SIGKILL), &dir); + if crate::enforce::oom_evidence_is_available() { + // This machine counts the kills for memory, so qex knows that the + // kernel did not stop this job. + assert_eq!(note, None); + } else { + let note = note.expect("qex must say that it could not tell the cause"); + assert!(note.contains("cannot say"), "got: {note}"); + assert!( + note.contains("--mem"), + "the note must say what to do: {note}" + ); + } + + // No note for a state that qex did not guess. + assert_eq!( + unexplained_kill_note(JobState::Completed, Some(libc::SIGKILL), &dir), + None + ); + assert_eq!(unexplained_kill_note(JobState::Killed, None, &dir), None); + std::fs::remove_dir_all(&dir).ok(); + } + + /// Gives a configuration with a budget and a retry rule. + fn cfg_with(budget: &str, on_oom: u32) -> crate::config::Config { + toml::from_str(&format!( + "[budget]\nmem = \"{budget}\"\n[retry]\non_oom = {on_oom}\ngrowth = 2.0\n" + )) + .unwrap() + } + + fn oom_status(claim: u64, raises: u32, attempts: u32) -> job::JobStatus { + let mut s = job::JobStatus::new(&spec()); + s.state = JobState::Oom; + s.mem = claim; + s.oom_raises = raises; + s.attempts = attempts; + s + } + + /// The claim doubles after a kill for memory. The next attempt then has a + /// chance to succeed; an attempt with the same claim has none. + #[test] + fn the_claim_doubles_after_a_kill_for_memory() { + let cfg = cfg_with("8GB", 2); + let Raise::To(next) = raise_claim(&cfg, &oom_status(1 << 30, 0, 1), 1 << 30) else { + panic!("the first kill for memory must raise the claim"); + }; + assert_eq!(next, 2 << 30); + + let Raise::To(next) = raise_claim(&cfg, &oom_status(2 << 30, 1, 2), 1 << 30) else { + panic!("the second kill for memory must raise the claim"); + }; + assert_eq!(next, 4 << 30); + } + + /// The ladder has a limit. A claim that doubles for ever finishes with the + /// whole machine, and each attempt costs the full time of the job. + #[test] + fn the_claim_stops_growing_at_the_limit() { + let cfg = cfg_with("64GB", 2); + let Raise::Stop(reason) = raise_claim(&cfg, &oom_status(4 << 30, 2, 3), 1 << 30) else { + panic!("a job must not double for ever"); + }; + assert!(reason.contains("too small"), "got: {reason}"); + assert!( + reason.contains("1GB") && reason.contains("4GB"), + "the reason must give the first claim and the last one: {reason}" + ); + assert!( + reason.contains("--mem"), + "the reason must say what to do: {reason}" + ); + } + + /// qex must not claim memory that it does not have. A claim that is already + /// the whole budget has no larger claim available, and the answer for the + /// user is a different machine. + #[test] + fn the_claim_never_goes_above_the_memory_budget() { + let cfg = cfg_with("6GB", 3); + + // The double is above the budget, so the claim stops at the budget. + let Raise::To(next) = raise_claim(&cfg, &oom_status(4 << 30, 0, 1), 4 << 30) else { + panic!("a claim below the budget must still grow"); + }; + assert_eq!(next, 6 << 30, "the claim must stop at the budget"); + + // The claim is the whole budget. There is no larger claim. + let Raise::Stop(reason) = raise_claim(&cfg, &oom_status(6 << 30, 1, 2), 4 << 30) else { + panic!("qex must not claim more memory than its budget"); + }; + assert!(reason.contains("budget"), "got: {reason}"); + assert!( + reason.contains("machine with more memory"), + "the reason must say what to do: {reason}" + ); + + // A job that is larger than the budget already keeps its own claim. + // + // The message must not call that claim "the whole budget". An + // oversized job is a supported case, and the text would then give a + // number that is not the number in the record. + let Raise::Stop(reason) = raise_claim(&cfg, &oom_status(20 << 30, 0, 1), 20 << 30) else { + panic!("an oversized job has no larger claim"); + }; + assert!( + reason.contains("20GB") && reason.contains("6GB"), + "the reason must give the claim and the budget: {reason}" + ); + assert!( + !reason.contains("claim of 20GB. THE CLAIM WAS TOO SMALL. That claim is already"), + "the reason must not call a claim above the budget the whole budget: {reason}" + ); + } + + /// A multiplier a little above 1.0 can give the claim that already failed. + /// A new attempt at that claim costs a whole run and stops in the same way. + #[test] + fn a_multiplier_that_gives_the_same_claim_stops_the_ladder() { + let cfg: crate::config::Config = + toml::from_str("[budget]\nmem = \"64GB\"\n[retry]\non_oom = 3\ngrowth = 1.0000001\n") + .unwrap(); + cfg.validate().expect("the config file is valid"); + + let Raise::Stop(reason) = raise_claim(&cfg, &oom_status(1024, 0, 1), 1024) else { + panic!("a multiplier that gives the same claim must stop the ladder"); + }; + assert!(reason.contains("growth"), "got: {reason}"); + assert!( + reason.contains("--mem"), + "the reason must say what to do: {reason}" + ); + } + + /// The config file can turn the correction off. qex must then say why it + /// did not start the job again. + #[test] + fn the_config_file_can_stop_the_correction() { + let cfg = cfg_with("8GB", 0); + let Raise::Stop(reason) = raise_claim(&cfg, &oom_status(1 << 30, 0, 1), 1 << 30) else { + panic!("`on_oom = 0` must stop the correction"); + }; + assert!(reason.contains("on_oom"), "got: {reason}"); + } + /// A fault in the program gives the state `failed`. #[test] fn a_fault_signal_gives_the_state_failed() { diff --git a/src/top.rs b/src/top.rs index 9ee2354..977feef 100644 --- a/src/top.rs +++ b/src/top.rs @@ -378,7 +378,9 @@ fn note_for(job: &JobStatus) -> String { None => "failed".to_string(), }, JobState::Skipped => "a job that it needed did not succeed".to_string(), - JobState::Oom => "out of memory".to_string(), + // Name the cause, and not the symptom. This column holds one short + // line, and the reader must learn what to correct: the claim. + JobState::Oom => "the memory claim was too small".to_string(), JobState::Timeout => "reached its time limit".to_string(), JobState::Killed => "stopped by a command".to_string(), JobState::Cancelled => "left the queue".to_string(), @@ -422,6 +424,7 @@ mod tests { locks: vec![], attempts: 1, retries_left: 0, + oom_raises: 0, caused_by: None, tags: vec![], } diff --git a/src/usage.rs b/src/usage.rs index 7c47994..e82869c 100644 --- a/src/usage.rs +++ b/src/usage.rs @@ -27,11 +27,25 @@ //! //! # Which jobs qex records //! -//! A job that completed only. A job that the out-of-memory killer stopped, or -//! that reached its time limit, gives a measurement that is too small: it shows -//! the memory that the job reached before something stopped it, and not the -//! memory that the job needs. A record from such a job would make the next +//! A job that completed, and a job that the kernel stopped for memory. Each of +//! the two gives a different kind of evidence, and this module keeps them +//! apart: +//! +//! - A job that COMPLETED gives a peak. The job did all its work, so the peak +//! is the memory that the job needs. +//! - A job that the kernel STOPPED FOR MEMORY gives a lower bound. The job did +//! not finish, so the true need is ABOVE this value. This is the most +//! valuable sample that qex holds: it costs a whole run to obtain, and it is +//! the answer to the question that the next claim asks. +//! +//! qex records nothing else. A job that somebody stopped, or that reached its +//! time limit, gives a measurement that is too small and no bound: something +//! outside the memory stopped it, and the memory that it reached says nothing +//! about the memory that it needs. A sample from such a job would make the next //! claim too small, and the next job would stop in the same way. +//! +//! A lower bound is never averaged with a peak. `suggest` takes the largest +//! peak AND the largest lower bound, and the claim is above both. use crate::job::JobStatus; use crate::paths; @@ -48,8 +62,26 @@ const SAMPLES: usize = 5; /// day, with a larger input, so a very small claim is not useful. const MIN_MEMORY: u64 = 64 << 20; +/// What one measurement says about the memory that a command needs. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum Measurement { + /// The job completed, and this value is the memory that it used. + /// + /// This is the value of every sample that qex wrote before it learned from + /// an out-of-memory kill, so it is the value for a file with no `kind` + /// field. An old file thus keeps its meaning. + #[default] + Peak, + /// The kernel stopped the job for memory. The true need is ABOVE this value. + LowerBound, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Sample { + /// What this measurement says. See [`Measurement`]. + #[serde(default)] + pub kind: Measurement, /// The peak memory of the job, in bytes. pub max_rss: u64, /// The CPU time of the job, in seconds. @@ -124,18 +156,58 @@ pub fn load() -> Store { serde_json::from_str(&text).unwrap_or_default() } +/// Records the peak of a job that completed. +/// +/// A job that did not complete does not come here. It reached the memory that +/// something stopped it at, and that number is not the memory that the job +/// needs. +pub fn record(spec: &JobSpec, status: &JobStatus) { + if status.state != crate::job::JobState::Completed || status.usage.max_rss == 0 { + return; + } + add(spec, status, Measurement::Peak, status.usage.max_rss); +} + +/// Records a lower bound from a job that the kernel stopped for memory. +/// +/// # The caller must hold the evidence of THIS JOB +/// +/// Call this function only when qex made the cgroup of the job and read the +/// counter of that cgroup. The kernel then stopped the job at the limit that +/// qex made from the claim, so the claim was too small and the need is above +/// it. +/// +/// The counter of the session does not support this record. It also counts a +/// kill in a different program of the same user, and the machine can be full +/// while the claim of this job is correct. A bound from that evidence would +/// raise the claim of a command that needs no more memory, and the ladder of +/// attempts would raise it again at each run. +pub fn record_lower_bound(spec: &JobSpec, status: &JobStatus) { + if status.state != crate::job::JobState::Oom { + return; + } + // Take the LARGER of the claim and the measured peak. + // + // The kernel stopped the job at the claim, so the need is above the claim. + // The measurement can be larger when the job used memory that the limit + // does not count, and it can be zero when the kernel stopped the job before + // any child of the supervisor ended. The larger of the two is the value + // that does not repeat the failure. + add( + spec, + status, + Measurement::LowerBound, + status.usage.max_rss.max(status.mem), + ); +} + /// Adds one measurement for a command. /// /// Two supervisors can stop at the same time, so this function holds a lock on /// the file while it reads and writes. Without the lock, one measurement would /// replace the other. -pub fn record(spec: &JobSpec, status: &JobStatus) { - // A job that did not complete gives a measurement that is too small. - if status.state != crate::job::JobState::Completed { - return; - } - // A job with no measurement gives nothing. - if status.usage.max_rss == 0 { +fn add(spec: &JobSpec, status: &JobStatus, kind: Measurement, bytes: u64) { + if bytes == 0 { return; } @@ -165,8 +237,35 @@ pub fn record(spec: &JobSpec, status: &JobStatus) { .entry(key(&spec.cwd, &spec.command)) .or_default(); entry.name = spec.name.clone(); + + // Keep ONE lower bound for a command. + // + // A ladder of attempts makes a bound at each step: one job of three + // attempts made three bounds and filled three of the five places. The + // measurements of the jobs that completed then went away, and the store + // held one job only. The largest bound holds every fact that the smaller + // bounds of the same ladder hold, so qex keeps that one. + if kind == Measurement::LowerBound { + if let Some(pos) = entry + .samples + .iter() + .position(|s| s.kind == Measurement::LowerBound) + { + if entry.samples[pos].max_rss >= bytes { + // An earlier bound is larger, so this one adds nothing. Move it + // to the end, so that it stays the newest measurement. + let earlier = entry.samples.remove(pos); + entry.samples.push(earlier); + write_store(&path, &store, &lock); + return; + } + entry.samples.remove(pos); + } + } + entry.samples.push(Sample { - max_rss: status.usage.max_rss, + kind, + max_rss: bytes, cpu_secs: status.usage.cpu_secs, elapsed_secs: status.elapsed().map(|d| d.as_secs()).unwrap_or(0), at: crate::sys::now_secs(), @@ -177,11 +276,16 @@ pub fn record(spec: &JobSpec, status: &JobStatus) { let extra = entry.samples.len().saturating_sub(SAMPLES); entry.samples.drain(..extra); - if let Ok(bytes) = serde_json::to_vec_pretty(&store) { + write_store(&path, &store, &lock); +} + +/// Writes the store and releases the lock. +fn write_store(path: &std::path::Path, store: &Store, lock: &std::fs::File) { + use std::os::unix::io::AsRawFd; + if let Ok(bytes) = serde_json::to_vec_pretty(store) { // Mode 0600: this file names the jobs of this user. - crate::job::write_atomic(&path, &bytes, 0o600).ok(); + crate::job::write_atomic(path, &bytes, 0o600).ok(); } - unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_UN); } @@ -205,13 +309,45 @@ pub fn suggest( // // A claim that is too small stops the job, and a claim that is a little too // large costs some capacity only. The two faults are not equal. - let peak_mem = entry.samples.iter().map(|s| s.max_rss).max().unwrap_or(0); - let mem = ((peak_mem as f64 * margin) as u64).max(MIN_MEMORY); + let peak_mem = entry + .samples + .iter() + .filter(|s| s.kind == Measurement::Peak) + .map(|s| s.max_rss) + .max() + .unwrap_or(0); + let mut mem = (peak_mem as f64 * margin) as u64; + + // A lower bound comes from a job that the kernel stopped for memory. It is + // a claim that FAILED, so the next claim must be above it. An average with + // the peaks of the smaller runs would lose that lesson, and the next job + // would stop in the same way and cost a second whole run. + let bound = entry + .samples + .iter() + .filter(|s| s.kind == Measurement::LowerBound) + .map(|s| s.max_rss) + .max() + .unwrap_or(0); + if bound > 0 { + // The margin is the usual step above a measurement. A margin of exactly + // 1.0 is permitted in the config file, and it would give here the claim + // that the kernel already stopped, so the claim goes at least one tenth + // above the bound. + let above = ((bound as f64 * margin) as u64).max(bound + bound / 10 + 1); + mem = mem.max(above); + } + + let mem = mem.max(MIN_MEMORY); // Calculate the cores from the CPU time and the time that the job operated. // // The CPU time of a job that used two cores for 10 seconds is 20 seconds. // The division thus gives the number of cores that the job used together. + // + // Each sample counts here, and a lower bound also. The memory of a job that + // the kernel stopped is not the memory that the job needs, but the cores + // that it used in the time that it ran are a true measurement. let cores = entry .samples .iter() @@ -239,6 +375,7 @@ mod tests { fn sample(max_rss: u64, cpu_secs: f64, elapsed_secs: u64) -> Sample { Sample { + kind: Measurement::Peak, max_rss, cpu_secs, elapsed_secs, @@ -246,6 +383,18 @@ mod tests { } } + /// A measurement from a job that the kernel stopped for memory. The true + /// need is above this value. + fn lower_bound(max_rss: u64) -> Sample { + Sample { + kind: Measurement::LowerBound, + max_rss, + cpu_secs: 1.0, + elapsed_secs: 1, + at: 0, + } + } + fn dir() -> std::path::PathBuf { std::path::PathBuf::from("/project") } @@ -370,6 +519,156 @@ mod tests { ); } + /// A job that the kernel stopped for memory gives a LOWER BOUND, and the + /// next claim must be above it. + /// + /// This measurement costs a whole run to obtain. qex threw it away before, + /// so the same claim died in the same way on the next run. + #[test] + fn the_next_claim_is_above_a_lower_bound() { + let cmd: Vec = vec!["train".into()]; + let store = store_with(&["train"], vec![lower_bound(8 << 30)]); + let s = suggest(&store, &dir(), &cmd, 1.5).unwrap(); + assert!( + s.mem > (8 << 30), + "the claim must be above a claim that failed, and it was {}", + crate::units::format_size(s.mem) + ); + assert_eq!(s.mem, 12 << 30); + + // A margin of exactly 1.0 is permitted in the config file. It must + // still give a claim ABOVE the value that the kernel stopped. + let s = suggest(&store, &dir(), &cmd, 1.0).unwrap(); + assert!( + s.mem > (8 << 30), + "a margin of 1.0 gave the claim that already failed" + ); + } + + /// A small run that succeeds must not hide the lesson of a run that the + /// kernel stopped. + /// + /// The kill says that the command needs more than 8GB. Three later runs of + /// 1GB do not answer that: they had a smaller input. A claim from those + /// three would stop the next large run in the same way, and that costs a + /// whole run. + #[test] + fn a_lower_bound_is_not_averaged_away_by_the_smaller_runs() { + let cmd: Vec = vec!["train".into()]; + let store = store_with( + &["train"], + vec![ + sample(1 << 30, 1.0, 10), + lower_bound(8 << 30), + sample(1 << 30, 1.0, 10), + sample(1 << 30, 1.0, 10), + ], + ); + let s = suggest(&store, &dir(), &cmd, 1.5).unwrap(); + assert!( + s.mem > (8 << 30), + "the lower bound went away, and the claim is {}", + crate::units::format_size(s.mem) + ); + } + + /// A peak that is larger than a lower bound must win. The bound says "more + /// than 2GB", and a run that completed with 6GB says "6GB is sufficient". + #[test] + fn the_largest_evidence_wins_whatever_its_kind() { + let cmd: Vec = vec!["train".into()]; + let store = store_with( + &["train"], + vec![lower_bound(2 << 30), sample(6 << 30, 1.0, 10)], + ); + let s = suggest(&store, &dir(), &cmd, 1.5).unwrap(); + assert_eq!(s.mem, 9 << 30, "6GB and one half"); + } + + /// A file that qex wrote before this feature holds no `kind` field. Each of + /// those samples is a peak, so an old file must give the claim that it gave + /// before. Without this rule, every earlier measurement would become a + /// lower bound, and every claim would go up. + #[test] + fn an_earlier_file_keeps_its_meaning() { + let text = r#"{"commands":{"x":{"name":"t","samples":[ + {"max_rss":1073741824,"cpu_secs":1.0,"elapsed_secs":10,"at":0}]}}}"#; + let store: Store = serde_json::from_str(text).unwrap(); + let sample = &store.commands["x"].samples[0]; + assert_eq!(sample.kind, Measurement::Peak); + assert_eq!(sample.max_rss, 1 << 30); + } + + /// The record of a job that the kernel stopped must hold the CLAIM when the + /// claim is the larger number. With a memory limit, the kernel stops the + /// job at the claim, so the need is above the claim and not at the peak + /// that qex measured. + #[test] + fn a_kill_for_memory_records_the_claim_when_it_is_larger() { + use crate::testutil::{env_lock, EnvVar}; + let _guard = env_lock(); + let dir = std::env::temp_dir().join(format!("qex-usage-oom-{}", std::process::id())); + std::fs::remove_dir_all(&dir).ok(); + std::fs::create_dir_all(&dir).unwrap(); + let _env = EnvVar::set("XDG_STATE_HOME", dir.to_str().unwrap()); + + let mut spec = crate::spec::JobSpec { + id: uuid::Uuid::new_v4(), + name: "train".into(), + cwd: "/project".into(), + command: vec!["train".into()], + env: Default::default(), + cpu: 1, + mem: 4 << 30, + timeout: None, + tags: vec![], + priority: 0, + env_capture: crate::config::EnvCapture::None, + claim_source: "explicit".into(), + group: None, + group_name: None, + locks: vec![], + retries: 0, + needs: vec![], + after: vec![], + submitted_at: 0, + }; + + let mut status = crate::job::JobStatus::new(&spec); + status.state = crate::job::JobState::Oom; + status.usage.max_rss = 1 << 30; + record_lower_bound(&spec, &status); + + // `record` keeps the peak of a job that COMPLETED. A job that the + // kernel stopped needs the caller to hold the evidence of that job, so + // it has a function of its own and this one must do nothing. + record(&spec, &status); + + let store = load(); + let entry = &store.commands[&key(&spec.cwd, &spec.command)]; + assert_eq!(entry.samples.len(), 1); + assert_eq!(entry.samples[0].kind, Measurement::LowerBound); + assert_eq!( + entry.samples[0].max_rss, + 4 << 30, + "the claim is the larger evidence" + ); + + // A job that somebody stopped teaches nothing. The memory that it + // reached says nothing about the memory that it needs. + spec.command = vec!["other".into()]; + let mut killed = crate::job::JobStatus::new(&spec); + killed.state = crate::job::JobState::Killed; + killed.usage.max_rss = 3 << 30; + record(&spec, &killed); + assert!( + !load().commands.contains_key(&key(&spec.cwd, &spec.command)), + "a job that a command stopped must teach the learner nothing" + ); + + std::fs::remove_dir_all(&dir).ok(); + } + /// The key must not hold the command or the directory. A command line can /// hold a token, and a directory names the work of a user. #[test] diff --git a/tests/e2e.rs b/tests/e2e.rs index 38afaac..b92faf2 100644 --- a/tests/e2e.rs +++ b/tests/e2e.rs @@ -2572,3 +2572,559 @@ fn the_status_gives_the_measured_use_of_a_job() { } fn _unused(_: &Path) {} + +/// The directory that a test uses to control a job, and the job directory of a +/// job. +/// +/// # Why these tests simulate the kill and do not make one +/// +/// A genuine kill by the out-of-memory killer needs one of two things: a cgroup +/// memory limit, or a machine with no free memory. qex can apply a limit only +/// when the coordinator owns its cgroup, and a usual test machine gives the +/// login session to the root user, so the limit is not available. A test that +/// fills the machine is worse: other work operates on the same machine, and the +/// kernel chooses its victim itself. +/// +/// These tests therefore make the EVIDENCE that a true kill leaves — the +/// out-of-memory record in the job directory — and the job then stops itself +/// with the same signal that the kernel uses, `SIGKILL`. Every step after that +/// point is the true code: the classification, the new claim, the new attempt, +/// the words in the record, and the learner. The unit tests in `src/enforce.rs` +/// and `src/supervisor.rs` cover the reading of the cgroup counter itself. +/// +/// The record says `job`, which is the evidence of a cgroup that qex made for +/// this job. qex acts on that evidence only, so the test must produce it. On a +/// machine with a delegated cgroup, `[enforce] mode` gives it. These tests must +/// not use that mode: qex then starts the coordinator again in a systemd unit, +/// which does not receive the `XDG_*` values of the test, and the test would +/// use the state directory of the user. +struct OomJob { + control: PathBuf, +} + +impl OomJob { + fn new(h: &Harness) -> Self { + let control = h.root.join("control"); + std::fs::create_dir_all(&control).unwrap(); + Self { control } + } + + /// The command of a job that makes the evidence of a kill for memory, then + /// stops itself with the signal that the kernel uses. + /// + /// The job waits for the file that holds its own directory. The test knows + /// that directory after the submission only, because the name of the + /// directory is the job id. + /// + /// `kills` gives the number of attempts that stop in this way. A later + /// attempt stops with the code 0. + fn script(&self, kills: u32) -> String { + let c = self.control.display(); + format!( + "until [ -f {c}/dir ]; do sleep 0.1; done; \ + n=$(cat {c}/n 2>/dev/null || echo 0); n=$((n+1)); echo $n > {c}/n; \ + echo attempt $n; \ + if [ $n -le {kills} ]; then echo job > \"$(cat {c}/dir)/oom\"; kill -9 $$; fi; \ + exit 0" + ) + } + + /// Tells the job where its own directory is. The job then starts. + fn release(&self, h: &Harness, id: &str) { + let dir = h.root.join("state/qex/jobs").join(id); + assert!( + dir.is_dir(), + "the job directory {} is missing", + dir.display() + ); + std::fs::write(self.control.join("dir"), dir.to_string_lossy().as_bytes()).unwrap(); + } + + /// The number of attempts that the job made. + fn attempts(&self) -> u32 { + std::fs::read_to_string(self.control.join("n")) + .map(|s| s.trim().parse().unwrap_or(0)) + .unwrap_or(0) + } +} + +/// A job that the kernel stops for memory must run again with a LARGER claim, +/// and the record must say that the claim was too small. +/// +/// This is the case in the README: a long run with `--mem guess` that the +/// kernel stops. Before this feature the job stopped with the state `killed`, +/// `--retries` did not see it, and the same claim died in the same way on the +/// next run. +#[test] +fn a_job_that_the_kernel_stops_for_memory_runs_again_with_a_larger_claim() { + let h = Harness::new( + "oomretry", + "[peers]\nenabled = false\n\ + [system]\nreserve_mem = \"0\"\nmax_pressure = 100\n\ + [budget]\ncpu = \"2\"\nmem = \"1GB\"\n", + ); + let job = OomJob::new(&h); + + let id = h.submit(&[ + "submit", + "--mem", + "128MB", + "--cpu", + "1", + "--", + "bash", + "-c", + &job.script(1), + ]); + job.release(&h, &id); + + h.ok(&["wait", &id, "--timeout", "60s"]); + let s = h.status_json(&id); + + assert_eq!( + s["state"], "completed", + "the second attempt must succeed: {s}" + ); + assert_eq!(s["attempts"], 2, "qex must start the job again: {s}"); + assert_eq!(job.attempts(), 2, "the job itself must run two times"); + assert_eq!(s["oom_raises"], 1, "qex must count the raise: {s}"); + assert_eq!( + s["mem"].as_u64().unwrap(), + 256 * 1024 * 1024, + "the claim must double: {s}" + ); + assert_eq!(s["claim_source"], "raised", "the claim came from qex: {s}"); + + // The record must say what happened, in words that need no knowledge of a + // cgroup. A reader of `qex status` alone must understand it. + let note = s["error"].as_str().unwrap_or(""); + assert!( + note.contains("kernel stopped") && note.contains("TOO SMALL"), + "the record must say that the claim was too small: {note}" + ); + assert!( + note.contains("128MB") && note.contains("256MB"), + "the record must give both claims: {note}" + ); + + // The same words must reach a reader of the text output. + let text = h.ok(&["status", &id]); + assert!( + text.contains("qex raised it"), + "the claim line must say that qex raised the claim: {text}" + ); + // A job that succeeded must not have the word `error` on that line. The + // state and the label would contradict each other. + assert!( + text.contains("note:") && !text.contains("error:"), + "a job that succeeded gives a note and not an error: {text}" + ); + + // The log must hold every attempt, in the same way as `--retries`. + let logs = h.ok(&["logs", &id]); + assert!( + logs.contains("attempt 1") && logs.contains("attempt 2"), + "the log must hold every attempt: {logs}" + ); + + // `qex rerun` must repeat the claim that WORKED, and not the claim that the + // kernel stopped. The correction cost a whole run, so it must not go away. + let again = h.ok(&["rerun", &id]); + h.ok(&["wait", &again, "--timeout", "60s"]); + let s = h.status_json(&again); + assert_eq!( + s["mem"].as_u64().unwrap(), + 256 * 1024 * 1024, + "a rerun must use the claim in force, and not the first claim: {s}" + ); + + // The learner must use the lesson. The next job of the same command gets a + // claim above the claim that the kernel stopped, and the agent gives no + // `--mem` value at all. + let next = h.submit(&["submit", "--", "bash", "-c", &job.script(1)]); + h.ok(&["wait", &next, "--timeout", "60s"]); + let s = h.status_json(&next); + assert_eq!(s["claim_source"], "learned", "got: {s}"); + assert!( + s["mem"].as_u64().unwrap() > 128 * 1024 * 1024, + "the claim must be above the claim that the kernel stopped: {s}" + ); +} + +/// A job that a USER stopped must NOT run again, and it must teach the learner +/// nothing. +/// +/// This test protects the feature from doing harm. The kernel and `qex kill` +/// both use `SIGKILL`. A job that somebody stopped on purpose must never run +/// again with a larger claim: qex would repeat work that the user stopped, and +/// it would also record that the command needs more memory than it does. +#[test] +fn a_job_that_a_user_killed_is_not_retried_and_teaches_the_learner_nothing() { + let h = Harness::new( + "userkill", + "[peers]\nenabled = false\n\ + [system]\nreserve_mem = \"0\"\nmax_pressure = 100\n\ + [budget]\ncpu = \"2\"\nmem = \"1GB\"\n", + ); + + // Give the job the evidence of a kill for memory as well. + // + // This step is the point of the test. qex reads the out-of-memory count of + // the SESSION when it applies no limit, and that count also holds a kill in + // a different program of the same user. The mark from `qex kill` must win + // against that evidence. Without the mark, this job looks like a job that + // the kernel stopped for memory, and qex starts it again with a larger + // claim. + let job = OomJob::new(&h); + let control = h.root.join("control"); + let script = format!( + "until [ -f {c}/dir ]; do sleep 0.1; done; echo job > \"$(cat {c}/dir)/oom\"; \ + echo ready > {c}/ready; sleep 60", + c = control.display() + ); + let id = h.submit(&["submit", "--mem", "128MB", "--", "bash", "-c", &script]); + h.until("the job operates", Duration::from_secs(30), || { + h.state_of(&id) == "running" + }); + job.release(&h, &id); + h.until("the job made the record", Duration::from_secs(30), || { + control.join("ready").exists() + }); + + // Use KILL, which is the signal that the out-of-memory killer uses. With + // TERM the two causes are already separate. + h.ok(&["kill", &id, "--signal", "KILL", "--grace", "1s"]); + h.until("the job stops", Duration::from_secs(30), || { + h.status_json(&id)["state"] + .as_str() + .map(|s| s != "running" && s != "starting") + .unwrap_or(false) + }); + + let s = h.status_json(&id); + assert_eq!(s["state"], "killed", "a command stopped this job: {s}"); + assert_eq!(s["attempts"], 1, "qex must not start the job again: {s}"); + assert_eq!(s["oom_raises"], 0, "qex must not raise the claim: {s}"); + assert_eq!( + s["mem"].as_u64().unwrap(), + 128 * 1024 * 1024, + "the claim must not change: {s}" + ); + + // The learner must hold nothing. The memory that a job reached before + // somebody stopped it says nothing about the memory that it needs. + let store = h.root.join("state/qex/usage.json"); + assert!( + !store.exists(), + "a job that a command stopped must teach the learner nothing, and the store holds: {}", + std::fs::read_to_string(&store).unwrap_or_default() + ); +} + +/// The claim must not double for ever. A job that reaches the limit keeps the +/// state `oom`, and the record says what the user must do. +/// +/// Each attempt costs the full time of the job. A ladder with no limit can use +/// a day of the machine and give no result. +#[test] +fn a_claim_that_stays_too_small_stops_at_the_limit() { + let h = Harness::new( + "oomlimit", + "[peers]\nenabled = false\n\ + [system]\nreserve_mem = \"0\"\nmax_pressure = 100\n\ + [budget]\ncpu = \"2\"\nmem = \"1GB\"\n\ + [retry]\non_oom = 2\ngrowth = 2.0\n", + ); + let job = OomJob::new(&h); + + // Every attempt stops for memory. + let id = h.submit(&[ + "submit", + "--mem", + "128MB", + "--cpu", + "1", + "--", + "bash", + "-c", + &job.script(9), + ]); + job.release(&h, &id); + + // The code is 125: something stopped the job. + let out = h.qex(&["wait", &id, "--timeout", "90s"]); + assert_eq!(out.status.code(), Some(125), "got: {out:?}"); + + let s = h.status_json(&id); + assert_eq!(s["state"], "oom", "got: {s}"); + assert_eq!(s["attempts"], 3, "one attempt and two raises: {s}"); + assert_eq!(s["oom_raises"], 2, "got: {s}"); + assert_eq!( + s["mem"].as_u64().unwrap(), + 512 * 1024 * 1024, + "the claim doubles two times: {s}" + ); + + let note = s["error"].as_str().unwrap_or(""); + assert!( + note.contains("TOO SMALL") && note.contains("--mem"), + "the record must say what the user must do: {note}" + ); + assert!( + note.contains("128MB") && note.contains("512MB"), + "the record must give the first claim and the last one: {note}" + ); +} + +/// A kill for memory that qex cannot connect to THIS job must not start the job +/// again, and must teach the learner nothing. +/// +/// With no memory limit — the default — qex reads the out-of-memory count of the +/// whole login session. That count also rises when the kernel stops a DIFFERENT +/// program of the same user. A machine that is short of memory is also the +/// machine on which a person uses `kill -9`, so the two events arrive together. +/// +/// qex may report the state `oom` on that evidence. It must not act on it: a new +/// attempt repeats work, holds more of the machine, and teaches the learner a +/// number that no measurement supports. +#[test] +fn a_kill_for_memory_with_no_limit_reports_but_does_not_run_the_job_again() { + let h = Harness::new( + "oomsession", + "[peers]\nenabled = false\n\ + [system]\nreserve_mem = \"0\"\nmax_pressure = 100\n\ + [budget]\ncpu = \"2\"\nmem = \"1GB\"\n", + ); + let job = OomJob::new(&h); + let control = h.root.join("control"); + + // The record says `session`, which is the evidence of the counter of the + // login session. qex writes that value when it made no cgroup for the job. + let script = format!( + "until [ -f {c}/dir ]; do sleep 0.1; done; echo session > \"$(cat {c}/dir)/oom\"; \ + kill -9 $$", + c = control.display() + ); + let id = h.submit(&["submit", "--mem", "128MB", "--", "bash", "-c", &script]); + job.release(&h, &id); + + let out = h.qex(&["wait", &id, "--timeout", "60s"]); + assert_eq!(out.status.code(), Some(125), "got: {out:?}"); + + let s = h.status_json(&id); + // The report is correct: the kernel stopped the job for memory. + assert_eq!(s["state"], "oom", "got: {s}"); + // The action must not happen. + assert_eq!(s["attempts"], 1, "qex must not start the job again: {s}"); + assert_eq!(s["oom_raises"], 0, "qex must not raise the claim: {s}"); + assert_eq!( + s["mem"].as_u64().unwrap(), + 128 * 1024 * 1024, + "the claim must not change: {s}" + ); + + let note = s["error"].as_str().unwrap_or(""); + assert!( + note.contains("did NOT start the job again"), + "the record must say that qex did not act: {note}" + ); + assert!( + note.contains("enforce") && note.contains("--mem"), + "the record must say what the user can do: {note}" + ); + + // The learner must hold nothing. The machine can be full while the claim of + // this job is correct, so this kill is not a measurement of this command. + let store = h.root.join("state/qex/usage.json"); + assert!( + !store.exists(), + "a kill that qex cannot connect to this job must teach it nothing, and the store holds: {}", + std::fs::read_to_string(&store).unwrap_or_default() + ); +} + +/// A job with a RAISED claim must go through the queue again, and it must wait +/// for capacity for the new claim. +/// +/// The queue exists to keep the sum of the claims inside the budget. A raise +/// makes a claim that the queue never admitted, so a new attempt that started +/// in the supervisor would put that claim beside the jobs that already hold the +/// budget. With `[enforce] mode = "hard"` the kernel would receive the sum of +/// those limits, and the machine would meet the load that the budget exists to +/// prevent. +#[test] +fn a_raised_claim_waits_for_capacity_in_the_queue() { + let h = Harness::new( + "oomreadmit", + "[peers]\nenabled = false\n\ + [system]\nreserve_mem = \"0\"\nmax_pressure = 100\n\ + [budget]\ncpu = \"4\"\nmem = \"1GB\"\n", + ); + let job = OomJob::new(&h); + let control = h.root.join("control"); + + // This job holds 400MB of the budget until the test releases it. + let holder = h.submit(&[ + "submit", + "--name", + "holder", + "--mem", + "400MB", + "--cpu", + "1", + "--", + "bash", + "-c", + &format!( + "until [ -f {c}/stop ]; do sleep 0.2; done", + c = control.display() + ), + ]); + + // 400MB and 600MB fit the budget of 1GB together. The kernel stops this job + // one time, and the raise then asks for 1GB, which does not fit beside the + // job that holds 400MB. + let id = h.submit(&[ + "submit", + "--name", + "grower", + "--mem", + "600MB", + "--cpu", + "1", + "--", + "bash", + "-c", + &job.script(1), + ]); + + h.until("both jobs operate", Duration::from_secs(30), || { + h.state_of(&holder) == "running" && h.state_of(&id) == "running" + }); + job.release(&h, &id); + + h.until("qex raised the claim", Duration::from_secs(30), || { + h.status_json(&id)["oom_raises"].as_u64() == Some(1) + }); + + // The job must WAIT. The other job holds 400MB of the budget of 1GB, and + // this job now claims 1GB. + let s = h.status_json(&id); + assert_eq!( + s["mem"].as_u64().unwrap(), + 1024 * 1024 * 1024, + "the raise must stop at the budget: {s}" + ); + + let deadline = Instant::now() + Duration::from_secs(4); + let mut saw_reason = false; + while Instant::now() < deadline { + let s = h.status_json(&id); + let state = s["state"].as_str().unwrap_or(""); + assert_eq!( + state, "queued", + "the job must wait for capacity for its new claim: {s}" + ); + assert_eq!( + h.state_of(&holder), + "running", + "the job that holds the budget must still operate" + ); + if s["blocked_reason"] + .as_str() + .map(|r| r.contains("memory")) + .unwrap_or(false) + { + saw_reason = true; + } + std::thread::sleep(Duration::from_millis(200)); + } + assert!( + saw_reason, + "the job must say that it waits for memory: {}", + h.status_json(&id) + ); + + // Release the capacity. The job then starts and succeeds. + std::fs::write(control.join("stop"), b"1").unwrap(); + h.ok(&["wait", &holder, "--timeout", "60s"]); + h.ok(&["wait", &id, "--timeout", "60s"]); + + let s = h.status_json(&id); + assert_eq!(s["state"], "completed", "got: {s}"); + assert_eq!(s["attempts"], 2, "got: {s}"); +} + +/// A mark that says how an attempt stopped must not stay for the next attempt. +/// +/// A job with `--retries` can stop in two different ways. `qex kill` stopped +/// attempt 1 here, and the kernel stopped attempt 2 for memory. The mark of the +/// first attempt stayed before, so attempt 2 said that a command stopped it. The +/// record then named a cause that no command made, and the lesson of the kill +/// for memory went away. +#[test] +fn a_mark_from_one_attempt_does_not_decide_the_next_attempt() { + let h = Harness::new( + "oommarks", + "[peers]\nenabled = false\n\ + [system]\nreserve_mem = \"0\"\nmax_pressure = 100\n\ + [budget]\ncpu = \"2\"\nmem = \"1GB\"\n", + ); + let job = OomJob::new(&h); + let c = h.root.join("control"); + + let script = format!( + "until [ -f {c}/dir ]; do sleep 0.1; done; \ + n=$(cat {c}/n 2>/dev/null || echo 0); n=$((n+1)); echo $n > {c}/n; \ + if [ $n -eq 1 ]; then trap 'exit 3' TERM; touch {c}/ready; sleep 60 & wait; fi; \ + if [ $n -eq 2 ]; then echo job > \"$(cat {c}/dir)/oom\"; kill -9 $$; fi; \ + exit 0", + c = c.display() + ); + + let id = h.submit(&[ + "submit", + "--retries", + "2", + "--mem", + "128MB", + "--cpu", + "1", + "--", + "bash", + "-c", + &script, + ]); + job.release(&h, &id); + + // Stop attempt 1 with a command. The job answers with the exit code 3, so + // its state is `failed` and `--retries` starts it again. + h.until("attempt 1 operates", Duration::from_secs(30), || { + c.join("ready").exists() + }); + h.ok(&["kill", &id, "--signal", "TERM", "--grace", "30s"]); + + h.ok(&["wait", &id, "--timeout", "90s"]); + let s = h.status_json(&id); + + assert_eq!(s["state"], "completed", "got: {s}"); + assert_eq!( + s["attempts"], 3, + "one kill, one kill for memory, one run: {s}" + ); + assert_eq!( + s["oom_raises"], 1, + "the mark of attempt 1 must not hide the kill for memory of attempt 2: {s}" + ); + assert_eq!( + s["mem"].as_u64().unwrap(), + 256 * 1024 * 1024, + "the claim must double after the kill for memory: {s}" + ); + // `--retries` gave 2 and one attempt used one of them. A kill for memory + // must not use that count. + assert_eq!( + s["retries_left"], 1, + "a kill for memory must not spend a `--retries` credit: {s}" + ); +}