use std::io::Write;
use std::time::Instant;
use super::cli::{Plan, Target};
use super::io::{Run, Stats};
use super::messages::{self as m, printf};
pub(crate) const REFRESH_MS: u64 = 200;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct Display {
pub level: i32,
last_ms: Option<u64>,
}
impl Display {
pub(crate) fn new(level: i32) -> Display {
Display {
level,
last_ms: None,
}
}
pub(crate) fn ready(&self, now_ms: u64) -> bool {
self.level >= 2
&& (self.level >= 4
|| self
.last_ms
.is_none_or(|t| now_ms.saturating_sub(t) >= REFRESH_MS))
}
pub(crate) fn update(
&mut self,
now_ms: u64,
err: &mut dyn Write,
render: impl FnOnce(&mut dyn Write),
) {
if self.ready(now_ms) {
self.last_ms = Some(now_ms);
render(err);
}
}
}
#[derive(Debug, Clone, Copy)]
pub(crate) struct Timer {
started: Option<(Instant, f64)>,
}
impl Timer {
pub(crate) fn start(real: bool) -> Timer {
Timer {
started: real.then(|| (Instant::now(), cpu_seconds())),
}
}
pub(crate) fn now_ms(&self) -> u64 {
self.started.map_or(0, |(t, _)| {
u64::try_from(t.elapsed().as_millis()).unwrap_or(u64::MAX)
})
}
pub(crate) fn elapsed(&self) -> (f64, f64) {
self.started.map_or((0.0, 0.0), |(t, cpu)| {
(t.elapsed().as_secs_f64(), cpu_seconds() - cpu)
})
}
}
fn cpu_seconds() -> f64 {
#[cfg(unix)]
{
let ru = unsafe {
let mut ru: libc::rusage = std::mem::zeroed();
libc::getrusage(libc::RUSAGE_SELF, &mut ru);
ru
};
let t = |tv: libc::timeval| tv.tv_sec as f64 + tv.tv_usec as f64 / 1e6;
t(ru.ru_utime) + t(ru.ru_stime)
}
#[cfg(not(unix))]
{
0.0
}
}
pub(crate) fn percent(part: u64, whole: u64) -> String {
format!("{:.2}", part as f64 / whole.max(1) as f64 * 100.0)
}
pub(crate) fn read_progress(w: &mut dyn Write, read: u64, written: u64) {
let mib = (read >> 20).to_string();
let ratio = percent(written, read);
printf(w, m::READ_PROGRESS, &[mib.as_bytes(), ratio.as_bytes()]);
}
pub(crate) fn decode_progress(w: &mut dyn Write, written: u64) {
printf(
w,
m::DECODE_PROGRESS,
&[(written >> 20).to_string().as_bytes()],
);
}
pub(crate) fn compress_summary(w: &mut dyn Write, legacy: bool, read: u64, written: u64) {
let fmt = if legacy {
m::COMPRESSED_LEGACY_SUMMARY
} else {
m::COMPRESSED_SUMMARY
};
let (read_s, written_s) = (read.to_string(), written.to_string());
let ratio = percent(written, read);
printf(w, m::CLEAR_LINE, &[]);
printf(
w,
fmt,
&[read_s.as_bytes(), written_s.as_bytes(), ratio.as_bytes()],
);
}
pub(crate) fn decode_summary(w: &mut dyn Write, name: &[u8], written: u64) {
let mut shown = name[..name.len().min(30)].to_vec();
shown.resize(30, b' ');
printf(w, m::CLEAR_LINE, &[]);
printf(
w,
m::DECODED_SUMMARY,
&[&shown, written.to_string().as_bytes()],
);
}
pub(crate) fn threads_line(w: &mut dyn Write, threads: u32, reduced_from: Option<u32>) {
if let Some(asked) = reduced_from {
printf(
w,
m::THREADS_REDUCED,
&[asked.to_string().as_bytes(), threads.to_string().as_bytes()],
);
} else if threads > 1 {
printf(w, m::USING_THREADS, &[threads.to_string().as_bytes()]);
}
}
pub(crate) fn done_line(w: &mut dyn Write, secs: f64, cpu: f64, bytes: u64) {
let (speed, load) = if secs > 0.0 {
(
bytes as f64 / (1u64 << 20) as f64 / secs,
cpu / secs * 100.0,
)
} else {
(0.0, 100.0)
};
let (secs, speed, load) = (
format!("{secs:.2}"),
format!("{speed:.2}"),
format!("{load:.0}"),
);
printf(
w,
m::DONE_IN,
&[secs.as_bytes(), speed.as_bytes(), load.as_bytes()],
);
}
impl Run<'_> {
pub(super) fn show_read_progress(&mut self, s: &Stats) {
let now = self.timer.now_ms();
self.display
.update(now, &mut *self.err, |w| read_progress(w, s.read, s.written));
}
pub(super) fn show_compress_summary(&mut self, s: &Stats) {
if self.level >= 2 {
compress_summary(&mut *self.err, self.cmd.opts.legacy, s.read, s.written);
}
}
pub(super) fn show_decode_summary(&mut self, name: &[u8], s: &Stats) {
if self.level >= 2 {
decode_summary(&mut *self.err, name, s.written);
}
}
pub(super) fn show_done(&mut self) {
if self.level >= 3 {
let (secs, cpu) = self.timer.elapsed();
done_line(&mut *self.err, secs, cpu, self.total);
}
}
pub(super) fn announce_run(&mut self, plan: &Plan, names: &[Vec<u8>]) {
if self.cmd.recursive {
for (i, name) in names.iter().enumerate() {
self.say(4, m::LIST_ENTRY, &[i.to_string().as_bytes(), name]);
}
}
if plan.op == super::cli::Op::Compress {
if self.level >= 3 {
threads_line(
&mut *self.err,
self.cmd.threads,
self.cmd.threads_reduced_from,
);
}
if self.cmd.opts.legacy {
self.say(3, m::GENERATING_LEGACY, &[]);
}
} else if plan.target == Target::Stdout || self.cmd.from_list {
self.stdout_note();
}
}
}
pub(super) fn show_decode_progress(
run: &mut Run,
written: u64,
frame: bool,
shown: &mut Option<u64>,
) {
let mib = written >> 20;
if (written == 0 && !frame) || *shown == Some(mib) {
return;
}
let now = run.timer.now_ms();
if run.display.ready(now) {
*shown = Some(mib);
}
run.display
.update(now, &mut *run.err, |w| decode_progress(w, written));
}
#[cfg(test)]
mod tests {
use super::*;
fn text(f: impl FnOnce(&mut dyn Write)) -> String {
let mut v = Vec::new();
f(&mut v);
String::from_utf8(v).unwrap()
}
fn rec(raw: &str) -> Vec<(u64, Vec<u8>)> {
raw.lines()
.map(|l| {
let (t, b) = l.split_once('\t').unwrap();
let lit = b.strip_prefix("b'").unwrap().strip_suffix('\'').unwrap();
let mut out = Vec::new();
let mut it = lit.bytes();
while let Some(c) = it.next() {
if c != b'\\' {
out.push(c);
continue;
}
match it.next().unwrap() {
b'r' => out.push(b'\r'),
b'n' => out.push(b'\n'),
b't' => out.push(b'\t'),
b'\\' => out.push(b'\\'),
b'\'' => out.push(b'\''),
b'x' => {
let h: Vec<u8> = it.by_ref().take(2).collect();
out.push(
u8::from_str_radix(std::str::from_utf8(&h).unwrap(), 16).unwrap(),
);
}
other => panic!("escape {other}"),
}
}
(t.parse().unwrap(), out)
})
.collect()
}
const C_T1_12: &str = include_str!("../../../tests/fixtures/lz4_progress/compress_t1_12.rec");
const C_T1_9: &str = include_str!("../../../tests/fixtures/lz4_progress/compress_t1_9.rec");
const C_LEGACY: &str =
include_str!("../../../tests/fixtures/lz4_progress/compress_legacy_9.rec");
const C_VV: &str = include_str!("../../../tests/fixtures/lz4_progress/compress_vv_9.rec");
const D_SLOW: &str = include_str!("../../../tests/fixtures/lz4_progress/decode_slow.rec");
const SMALL_V: &str = include_str!("../../../tests/fixtures/lz4_progress/small_v.rec");
const SMALL_M: &str = include_str!("../../../tests/fixtures/lz4_progress/small_m.rec");
const SMALL_D: &str = include_str!("../../../tests/fixtures/lz4_progress/small_decode_m.rec");
fn refresh_times(r: &[(u64, Vec<u8>)], marker: &str) -> Vec<u64> {
r.iter()
.filter(|(_, b)| String::from_utf8_lossy(b).contains(marker))
.map(|(t, _)| *t)
.collect()
}
#[test]
fn recorded_refreshes_are_never_closer_than_the_interval() {
let mut shortest = u64::MAX;
for (name, raw, marker) in [
("compress -12", C_T1_12, "Read :"),
("compress -9", C_T1_9, "Read :"),
("legacy -9", C_LEGACY, "Read :"),
("decode", D_SLOW, "Decompressed :"),
] {
let times = refresh_times(&rec(raw), marker);
assert!(times.len() > 8, "{name}");
for w in times.windows(2) {
shortest = shortest.min(w[1] - w[0]);
assert!(w[1] - w[0] >= REFRESH_MS, "{name}: {w:?}");
}
}
assert!(shortest < REFRESH_MS + 30, "{shortest}");
}
#[test]
fn level_four_draws_at_every_chance() {
let r = rec(C_VV);
let n = refresh_times(&r, "Read :").len();
assert_eq!(n, 48, "the recording: 200 MB in 4 MiB chunks");
let mut d = Display::new(4);
let mut out = Vec::new();
for i in 0..48 {
d.update(i, &mut out, |w| w.write_all(b".").unwrap());
}
assert_eq!(out.len(), 48);
}
#[test]
fn the_gate_draws_the_first_then_one_per_interval() {
for level in [2, 3] {
let mut d = Display::new(level);
let mut out = Vec::new();
let mut at = |d: &mut Display, t: u64| {
let n = out.len();
d.update(t, &mut out, |w| w.write_all(b"x").unwrap());
out.len() > n
};
assert!(at(&mut d, 5000), "the first, wherever the clock stands");
assert!(!at(&mut d, 5000));
assert!(!at(&mut d, 5000 + REFRESH_MS - 1));
assert!(at(&mut d, 5000 + REFRESH_MS));
assert!(!at(&mut d, 5000 + REFRESH_MS + 199));
assert!(at(&mut d, 5000 + 2 * REFRESH_MS));
assert!(!at(&mut d, 0));
}
for level in [0, 1] {
let mut d = Display::new(level);
let mut out = Vec::new();
d.update(0, &mut out, |w| w.write_all(b"x").unwrap());
d.update(10_000, &mut out, |w| w.write_all(b"x").unwrap());
assert!(out.is_empty(), "level {level}");
}
}
#[test]
fn the_gate_replays_a_recorded_run() {
for (raw, marker) in [(C_T1_12, "Read :"), (D_SLOW, "Decompressed :")] {
let times = refresh_times(&rec(raw), marker);
let mut d = Display::new(3);
let mut sink = Vec::new();
let mut at = |d: &mut Display, now: u64| {
let n = sink.len();
d.update(now, &mut sink, |w| w.write_all(b"x").unwrap());
sink.len() > n
};
for &t in × {
if t != times[0] {
let skipped = d.last_ms.unwrap() + 100;
assert!(!at(&mut d, skipped), "{raw:.20}");
}
assert!(at(&mut d, t), "drawn at {t}");
}
}
}
#[test]
fn rendered_lines_are_the_recorded_ones() {
let small = rec(SMALL_V);
let all: Vec<u8> = small.iter().flat_map(|(_, b)| b.clone()).collect();
let all = String::from_utf8(all).unwrap().replace("\r\n", "\n");
let want_threads = text(|w| threads_line(w, 3, None));
assert!(all.contains(&want_threads), "{all:?}");
let progress = text(|w| read_progress(w, 0, 1));
let blanks = " ".repeat(3);
assert!(
progress.starts_with(&format!("\rRead : 0 MiB{blanks}==> "))
&& progress.ends_with(&format!("%{blanks}"))
);
let summary = text(|w| compress_summary(w, false, 1760, 82));
assert!(all.contains(&summary), "{summary:?} in {all:?}");
let long = String::from_utf8(rec(C_T1_12).pop().unwrap().1).unwrap();
assert_eq!(
long.trim_end(),
"Done in 9.28 s ==> 20.56 MiB/s (cpu load : 99%)"
);
assert_eq!(
text(|w| done_line(w, 9.277, 9.277 * 0.99, 200_000_000)),
"Done in 9.28 s ==> 20.56 MiB/s (cpu load : 99%)\n"
);
let done = all.lines().find(|l| l.starts_with("Done in ")).unwrap();
assert!(
done.ends_with("%)")
&& done.contains(" s ==> ")
&& done.contains(" MiB/s (cpu load : ")
);
}
#[test]
fn several_files_have_one_thread_line_and_one_done() {
let all: Vec<u8> = rec(SMALL_M).into_iter().flat_map(|(_, b)| b).collect();
let all = String::from_utf8(all).unwrap().replace("\r\n", "\n");
let stripped: String = all
.split('\r')
.filter(|c| !c.starts_with("Read :"))
.collect::<Vec<_>>()
.join("\r");
assert_eq!(
stripped
.matches("Compressed 1760 bytes into 82 bytes ==> 4.66%\n")
.count(),
2,
"{stripped:?}"
);
assert_eq!(stripped.matches("Using 1 threads").count(), 0);
assert_eq!(all.matches("Done in ").count(), 1);
assert!(
all.trim_end()
.rsplit('\n')
.next()
.unwrap()
.starts_with("Done in ")
);
}
#[test]
fn decoding_summaries_are_the_recorded_ones() {
let all: Vec<u8> = rec(SMALL_D).into_iter().flat_map(|(_, b)| b).collect();
let all = String::from_utf8(all).unwrap().replace("\r\n", "\n");
assert!(all.contains(&text(|w| decode_progress(w, 1760))), "{all:?}");
let summary = text(|w| decode_summary(w, b"a.lz4", 1760));
assert!(all.contains(&summary), "{summary:?} in {all:?}");
assert!(all.contains("\nDone in "));
let long = text(|w| decode_summary(w, &[b'x'; 40], 1));
assert_eq!(
long,
format!("{}{} : decoded 1 bytes \n", m::CLEAR_LINE, "x".repeat(30))
);
let short = text(|w| decode_summary(w, b"s", 0));
assert!(short.ends_with(&format!("s{} : decoded 0 bytes \n", " ".repeat(29))));
}
#[test]
fn the_thread_line() {
assert_eq!(text(|w| threads_line(w, 1, None)), "");
assert_eq!(
text(|w| threads_line(w, 2, None)),
"Using 2 threads for compression \n"
);
assert_eq!(
text(|w| threads_line(w, 200, None)),
"Using 200 threads for compression \n"
);
assert_eq!(
text(|w| threads_line(w, 200, Some(201))),
"Requested 201 threads too large => automatically reduced to 200 \n"
);
assert_eq!(
text(|w| threads_line(w, 200, Some(300))),
"Requested 300 threads too large => automatically reduced to 200 \n"
);
}
#[test]
fn done_in_figures() {
assert_eq!(
text(|w| done_line(w, 2.0, 1.0, 8 << 20)),
"Done in 2.00 s ==> 4.00 MiB/s (cpu load : 50%)\n"
);
assert_eq!(
text(|w| done_line(w, 0.5, 1.2, 0)),
"Done in 0.50 s ==> 0.00 MiB/s (cpu load : 240%)\n"
);
assert_eq!(
text(|w| done_line(w, 0.0, 0.0, 99)),
"Done in 0.00 s ==> 0.00 MiB/s (cpu load : 100%)\n"
);
}
#[test]
fn timers() {
let t = Timer::start(false);
assert_eq!((t.now_ms(), t.elapsed()), (0, (0.0, 0.0)));
let t = Timer::start(true);
std::thread::sleep(std::time::Duration::from_millis(20));
assert!(t.now_ms() >= 20);
assert!(t.elapsed().0 >= 0.02);
}
#[test]
fn percent_of_nothing() {
assert_eq!(percent(15, 0), "1500.00");
assert_eq!(percent(1, 3), "33.33");
assert_eq!(percent(0, 0), "0.00");
}
}