use anyhow::{bail, Result};
use std::path::Path;
pub const PLACEHOLDER: &str = "{}";
pub const DEFAULT_MAX_JOBS: usize = 1000;
const MAX_NAME: usize = 48;
const MAX_SLUG: usize = 24;
const MAX_BASE: usize = 16;
#[derive(Debug, Default, PartialEq, Eq)]
pub struct Input {
pub lines: Vec<String>,
pub blank: usize,
pub comments: usize,
}
impl Input {
pub fn skipped(&self) -> usize {
self.blank + self.comments
}
}
pub fn parse(text: &str) -> Input {
let mut input = Input::default();
for line in text.lines() {
let line = line.trim();
if line.is_empty() {
input.blank += 1;
} else if line.starts_with('#') {
input.comments += 1;
} else {
input.lines.push(line.to_string());
}
}
input
}
pub const MAX_INPUT_BYTES: u64 = 64 * 1024 * 1024;
pub fn read(path: &Path) -> Result<Input> {
use std::io::Read;
let mut buffer = Vec::new();
if path == Path::new("-") {
std::io::stdin()
.take(MAX_INPUT_BYTES + 1)
.read_to_end(&mut buffer)
.map_err(|e| {
anyhow::anyhow!(
"qex cannot read the lines from standard input: {e}\n\n\
`--each-line -` needs those lines, and one line makes one job, so qex \
submits nothing.\n\n\
Send the lines through a pipe, or give the path of a file:\n\
\x20 ls *.csv | qex submit --each-line - -- ./process {PLACEHOLDER}"
)
})?;
} else {
std::fs::File::open(path)
.and_then(|f| f.take(MAX_INPUT_BYTES + 1).read_to_end(&mut buffer))
.map_err(|e| {
anyhow::anyhow!(
"qex cannot read the input file {}: {e}\n\n\
`--each-line` needs that file, and one line makes one job, so qex \
submits nothing.\n\n\
Give the path of a file that exists, or use `-` to read the lines from \
another program.",
path.display()
)
})?;
}
if buffer.len() as u64 > MAX_INPUT_BYTES {
let name = if path == Path::new("-") {
"the input from standard input".to_string()
} else {
format!("the input file {}", path.display())
};
bail!(
"{name} is larger than {} MiB, and qex submits no job at all.\n\n\
qex holds the whole input in memory, because it makes every job before it \
submits the first one. An input of this size is almost always the wrong \
file.\n\n\
Give a file that holds one line for each job, or divide the work into \
several fan-outs.",
MAX_INPUT_BYTES / (1024 * 1024)
);
}
let bytes = buffer;
let text = match String::from_utf8(bytes) {
Ok(text) => text,
Err(e) => {
let at = e.utf8_error().valid_up_to();
let line = e.as_bytes()[..at].iter().filter(|b| **b == b'\n').count() + 1;
let name = if path == Path::new("-") {
"the input from standard input".to_string()
} else {
format!("the input file {}", path.display())
};
bail!(
"{name} is not UTF-8 text. The first incorrect byte is on line {line}.\n\n\
The command of a job is text, so qex cannot run that line, and qex \
submits no job at all.\n\n\
Correct line {line}, or write the file again with UTF-8."
);
}
};
Ok(parse(&text))
}
pub fn count_placeholders(command: &[String]) -> usize {
command.iter().map(|arg| scan(arg, None).1).sum()
}
pub fn substitute(command: &[String], line: &str) -> Vec<String> {
command.iter().map(|arg| scan(arg, Some(line)).0).collect()
}
fn scan(arg: &str, line: Option<&str>) -> (String, usize) {
let mut out = String::with_capacity(arg.len());
let mut count = 0usize;
let mut rest = arg;
loop {
let Some(at) = rest.find('{') else {
out.push_str(rest);
return (out, count);
};
out.push_str(&rest[..at]);
let tail = &rest[at..];
if let Some(next) = tail.strip_prefix("{{}}") {
out.push_str(PLACEHOLDER);
rest = next;
} else if let Some(next) = tail.strip_prefix(PLACEHOLDER) {
count += 1;
if let Some(line) = line {
out.push_str(line);
}
rest = next;
} else {
out.push('{');
rest = &tail[1..];
}
}
}
pub fn check_command(command: &[String]) -> Result<()> {
if command.is_empty() {
bail!(
"no command.\n\n\
Write the command after `--`, and put `{PLACEHOLDER}` where each line goes:\n\
\x20 qex submit --each-line inputs.txt -- ./process {PLACEHOLDER}"
);
}
if count_placeholders(command) == 0 {
bail!(
"the command holds no `{PLACEHOLDER}`, so every job would be the same command.\n\n\
`--each-line` runs one job for each line, and `{PLACEHOLDER}` says where the \
line goes. With no `{PLACEHOLDER}` the lines have no effect, and qex submits \
nothing.\n\n\
Put `{PLACEHOLDER}` in the command:\n\
\x20 qex submit --each-line inputs.txt -- ./process {PLACEHOLDER}\n\n\
`{PLACEHOLDER}` goes in any argument, in the program name, or inside an \
argument such as `--out={PLACEHOLDER}.log`. Write `{{{{}}}}` for a literal \
`{PLACEHOLDER}`."
);
}
Ok(())
}
pub fn check_count(count: usize, max: usize) -> Result<()> {
if count == 0 {
bail!(
"the input holds no line that gives a job, so qex submits nothing.\n\n\
qex passes over an empty line and a line that starts with `#`. Add one \
line with the input for a job."
);
}
if count > max {
bail!(
"the input holds {}, and the limit is {}.\n\n\
Each job takes a directory in the state of qex, so a very large fan-out \
fills the disk and makes `qex list` hard to read.\n\n\
Give fewer lines, or raise the limit with `--max-jobs {count}`.",
crate::units::count_of(count, "line"),
crate::units::count_of(max, "job")
);
}
Ok(())
}
pub fn job_name(base: &str, index: usize, count: usize, line: &str) -> String {
let width = count.to_string().len();
let base = cut(&slug(base), MAX_BASE);
let base = if base.is_empty() {
"job"
} else {
base.as_str()
};
let mut name = format!("{base}-{index:0width$}");
let slug = slug(line);
if !slug.is_empty() {
name.push('-');
name.push_str(&cut(&slug, MAX_SLUG));
}
cut(&name, MAX_NAME)
}
fn slug(line: &str) -> String {
let mut out = String::new();
let mut dash = false;
for c in line.chars() {
if c.is_ascii_alphanumeric() || c == '.' || c == '_' {
out.push(c);
dash = false;
} else if !dash && !out.is_empty() {
out.push('-');
dash = true;
}
}
out.trim_matches('-').to_string()
}
fn cut(text: &str, max: usize) -> String {
match text.char_indices().nth(max) {
Some((at, _)) => text[..at].trim_end_matches('-').to_string(),
None => text.to_string(),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn args(list: &[&str]) -> Vec<String> {
list.iter().map(|s| s.to_string()).collect()
}
#[test]
fn a_line_with_shell_characters_stays_one_argument() {
let out = substitute(&args(&["./p", "{}"]), "a b\"; rm -rf ~; echo $HOME");
assert_eq!(out.len(), 2);
assert_eq!(out[1], "a b\"; rm -rf ~; echo $HOME");
}
#[test]
fn a_line_with_a_newline_stays_one_argument() {
let out = substitute(&args(&["./p", "{}"]), "one\ntwo");
assert_eq!(out, vec!["./p".to_string(), "one\ntwo".to_string()]);
}
#[test]
fn the_place_can_be_the_program_or_a_part_of_an_argument() {
let out = substitute(&args(&["./{}.sh", "--out={}.log", "-v"]), "run");
assert_eq!(out, args(&["./run.sh", "--out=run.log", "-v"]));
}
#[test]
fn every_place_takes_the_line() {
let out = substitute(&args(&["cp", "{}", "{}.bak"]), "a.txt");
assert_eq!(out, args(&["cp", "a.txt", "a.txt.bak"]));
assert_eq!(count_placeholders(&args(&["cp", "{}", "{}.bak"])), 2);
}
#[test]
fn the_escape_gives_a_literal_placeholder() {
let command = args(&["jq", "{{}}", "{}"]);
assert_eq!(count_placeholders(&command), 1);
assert_eq!(
substitute(&command, "a.json"),
args(&["jq", "{}", "a.json"])
);
}
#[test]
fn a_brace_that_is_not_a_place_does_not_change() {
let command = args(&["fmt", "--style={indent: 2}", "{}"]);
assert_eq!(count_placeholders(&command), 1);
assert_eq!(
substitute(&command, "x"),
args(&["fmt", "--style={indent: 2}", "x"])
);
}
#[test]
fn a_command_with_no_place_is_refused() {
let err = check_command(&args(&["./process", "input"]))
.unwrap_err()
.to_string();
assert!(err.contains("holds no `{}`"), "got: {err}");
assert!(err.contains("--each-line"), "the error must say what to do");
assert!(check_command(&args(&["echo", "{{}}"])).is_err());
assert!(check_command(&args(&["echo", "{}"])).is_ok());
assert!(check_command(&[]).is_err());
}
#[test]
fn an_empty_line_and_a_comment_line_give_no_job() {
let input = parse("one\n\n# a note\ntwo\n \n#\nthree\n");
assert_eq!(input.lines, vec!["one", "two", "three"]);
assert_eq!(input.blank, 2);
assert_eq!(input.comments, 2);
assert_eq!(input.skipped(), 4);
}
#[test]
fn crlf_endings_and_a_missing_final_newline_give_the_same_lines() {
assert_eq!(parse("a\r\nb\r\n").lines, vec!["a", "b"]);
assert_eq!(parse("a\nb").lines, vec!["a", "b"]);
assert_eq!(parse("a").lines, vec!["a"]);
assert_eq!(parse("").lines, Vec::<String>::new());
}
#[test]
fn the_space_at_each_end_of_a_line_goes_away() {
let input = parse(" a b \n\tc\t\n");
assert_eq!(input.lines, vec!["a b", "c"]);
}
#[test]
fn a_count_above_the_limit_is_refused() {
assert!(check_count(10, 10).is_ok());
let err = check_count(11, 10).unwrap_err().to_string();
assert!(err.contains("--max-jobs 11"), "got: {err}");
let err = check_count(0, 10).unwrap_err().to_string();
assert!(err.contains("no line that gives a job"), "got: {err}");
}
#[test]
fn a_job_name_holds_the_position_and_the_line() {
assert_eq!(job_name("process", 1, 9, "a.txt"), "process-1-a.txt");
assert_eq!(job_name("process", 7, 120, "a.txt"), "process-007-a.txt");
let long = "/data/very/long/path/that/goes/on/and/on/file-000.parquet";
let one = job_name("p", 1, 100, long);
let two = job_name("p", 2, 100, long);
assert_ne!(one, two);
assert!(one.len() <= MAX_NAME, "{one} is too long");
assert!(one.starts_with("p-001-"), "got: {one}");
}
#[test]
fn a_job_name_holds_no_control_character_from_the_file() {
let name = job_name(
"\u{1b}[31mBOOM\u{1b}[0m",
1,
9,
"\u{1b}[2J\u{1b}]0;title\u{7}x",
);
assert!(
name.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '_' || c == '-'),
"the name holds a character that must not reach a terminal: {name:?}"
);
assert!(!name.contains('\u{1b}'), "got: {name:?}");
let name = job_name("a\nb", 2, 9, "c\r\nd");
assert!(
!name.contains('\n') && !name.contains('\r'),
"got: {name:?}"
);
}
#[test]
fn a_line_with_no_ordinary_character_still_gives_a_name() {
let name = job_name("p", 3, 10, "!!! ???");
assert_eq!(name, "p-03");
assert!(name.parse::<uuid::Uuid>().is_err());
}
#[test]
fn an_input_above_the_size_limit_is_refused() {
let dir = std::env::temp_dir().join(format!(
"qex-fanout-{}-{}",
std::process::id(),
MAX_INPUT_BYTES
));
std::fs::create_dir_all(&dir).unwrap();
let path = dir.join("big.txt");
let line = b"aaaaaaaaaaaaaaaa\n";
let mut text = Vec::with_capacity(MAX_INPUT_BYTES as usize + 1);
while (text.len() as u64) < MAX_INPUT_BYTES + 1 {
text.extend_from_slice(line);
}
std::fs::write(&path, &text).unwrap();
let err = read(&path).unwrap_err().to_string();
std::fs::remove_dir_all(&dir).ok();
assert!(err.contains("larger than 64 MiB"), "got: {err}");
assert!(err.contains("no job at all"), "got: {err}");
}
#[test]
fn a_directory_in_the_place_of_the_input_is_refused() {
let dir = std::env::temp_dir().join(format!("qex-fanout-dir-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let err = read(&dir).unwrap_err().to_string();
std::fs::remove_dir_all(&dir).ok();
assert!(err.contains("cannot read the input file"), "got: {err}");
assert!(err.contains("submits nothing"), "got: {err}");
}
#[test]
fn a_job_name_never_has_the_form_of_an_id() {
let name = job_name("x", 1, 1, "550e8400-e29b-41d4-a716-446655440000");
assert!(name.parse::<uuid::Uuid>().is_err(), "got: {name}");
}
}