Skip to content

Commit

Permalink
Streamlining output code, by delegating more work to the barkeeper
Browse files Browse the repository at this point in the history
  • Loading branch information
@jzbor
jzbor committed Jul 18, 2024
1 parent cab8f28 commit adb32cc
Show file tree
Hide file tree
Showing 4 changed files with 57 additions and 140 deletions.
140 changes: 43 additions & 97 deletions src/barkeeper.rs
View file
Original file line number Diff line number Diff line change
@@ -1,12 +1,4 @@
use std::fmt::Write;


#[cfg(feature = "progress")]
struct BarMessageWriter(String, indicatif::ProgressBar);
#[cfg(feature = "progress")]
struct BarPrintWriter(String, indicatif::ProgressBar);
struct DummyOutWriter();
struct DummyStatusWriter(String, String);
use crate::{queue::JobState, JobRealization};


pub trait StateTracker {
Expand All @@ -17,12 +9,13 @@ pub trait StateTracker {
}

pub trait ThreadStateTracker: Send {
fn out(&mut self) -> &mut impl Write;
fn status(&mut self) -> &mut impl Write;
fn job_completed(&self);
fn job_completed(&self, job: JobRealization, state: JobState);
fn start(&self);
fn set_prefix(&mut self, prefix: String);
fn clear_status(&mut self);
fn cmd_output(&mut self, job: &str, out: &str, verbose: bool);
fn flush_cmd_output(&mut self, job: &str, verbose: bool);
fn trace(&mut self, cmd: &str);
}

#[cfg(feature = "progress")]
Expand All @@ -33,18 +26,14 @@ pub struct Barkeeper {

#[cfg(feature = "progress")]
pub struct ThreadBarkeeper {
message_writer: BarMessageWriter,
print_writer: BarPrintWriter,
mp: indicatif::MultiProgress,
bar: indicatif::ProgressBar,
main_bar: indicatif::ProgressBar,
last_line: Option<String>,
}

pub struct DummyBarkeeper {}
pub struct DummyThreadBarkeeper {
out_writer: DummyOutWriter,
status_writer: DummyStatusWriter,
}
pub struct DummyThreadBarkeeper {}


#[cfg(feature = "progress")]
Expand Down Expand Up @@ -85,11 +74,10 @@ impl StateTracker for Barkeeper {
bar.set_style(indicatif::ProgressStyle::with_template("{spinner} {prefix:.cyan} {wide_msg}").unwrap());

ThreadBarkeeper {
message_writer: BarMessageWriter(String::new(), bar.clone()),
print_writer: BarPrintWriter(String::new(), bar.clone()),
mp: self.mp.clone(),
main_bar: self.bar.clone(),
bar,
last_line: None,
}

}).collect()
Expand All @@ -105,50 +93,43 @@ impl StateTracker for DummyBarkeeper {

fn for_threads(&self, nthreads: usize) -> Vec<DummyThreadBarkeeper> {
(0..nthreads).map(|_| {
let out_writer = DummyOutWriter();
let status_writer = DummyStatusWriter(String::new(), String::new());
DummyThreadBarkeeper { out_writer, status_writer }
DummyThreadBarkeeper {}
}).collect()
}
}

impl ThreadStateTracker for DummyThreadBarkeeper {
fn out(&mut self) -> &mut impl Write {
&mut self.out_writer
fn job_completed(&self, job: JobRealization, state: JobState) {
println!("{}", job_finished_msg(job, state));
}

fn status(&mut self) -> &mut impl Write {
&mut self.status_writer
}

fn job_completed(&self) {}

fn start(&self) {}

fn set_prefix(&mut self, prefix: String) {
self.status_writer.0 = prefix;
}
fn set_prefix(&mut self, _prefix: String) {}

fn clear_status(&mut self) {}
}

#[cfg(feature = "progress")]
impl ThreadStateTracker for ThreadBarkeeper {
fn out(&mut self) -> &mut impl Write {
&mut self.print_writer
fn cmd_output(&mut self, job: &str, out: &str, _verbose: bool) {
println!("{}: {}", job, out);
}

fn status(&mut self) -> &mut impl Write {
&mut self.message_writer
fn flush_cmd_output(&mut self, _job: &str, _verbose: bool) {}

fn trace(&mut self, cmd: &str) {
println!("{}", cmd);
}
}

#[cfg(feature = "progress")]
impl ThreadStateTracker for ThreadBarkeeper {
fn start(&self) {
self.mp.add(self.bar.clone());
self.bar.tick();
self.bar.enable_steady_tick(std::time::Duration::from_millis(75));
}

fn job_completed(&self) {
fn job_completed(&self, job: JobRealization, state: JobState) {
self.bar.println(format!("{}", job_finished_msg(job, state)));
self.main_bar.inc(1)
}

Expand All @@ -159,72 +140,37 @@ impl ThreadStateTracker for ThreadBarkeeper {
fn clear_status(&mut self) {
self.bar.set_message("");
}
}


#[cfg(feature = "progress")]
impl Write for BarMessageWriter {
fn write_str(&mut self, s: &str) -> std::fmt::Result {
for c in s.chars() {
self.write_char(c)?
}
Ok(())
}

fn write_char(&mut self, c: char) -> std::fmt::Result {
if c == '\n' {
let msg = self.0.clone();
self.1.set_message(msg);
self.0 = String::new();
} else {
self.0.push(c);
}
Ok(())
}
}
fn cmd_output(&mut self, job: &str, out: &str, verbose: bool) {
self.bar.set_message(out.to_owned());

#[cfg(feature = "progress")]
impl Write for BarPrintWriter {
fn write_str(&mut self, s: &str) -> std::fmt::Result {
for c in s.chars() {
self.write_char(c)?
if verbose {
if let Some(line) = self.last_line.take() {
self.bar.println(format!("{}: {}", job, line));
}
self.last_line = Some(out.to_owned());
}
Ok(())
}

fn write_char(&mut self, c: char) -> std::fmt::Result {
if c == '\n' {
self.1.println(&self.0);
self.0.clear();
} else {
self.0.push(c)
fn flush_cmd_output(&mut self, job: &str, verbose: bool) {
if verbose {
if let Some(line) = self.last_line.take() {
self.bar.println(format!("{}: {}", job, line));
}
}
Ok(())
}
}

impl Write for DummyOutWriter {
fn write_str(&mut self, s: &str) -> std::fmt::Result {
print!("{}", s);
Ok(())
fn trace(&mut self, cmd: &str) {
self.bar.println(cmd);
}
}

impl Write for DummyStatusWriter {
fn write_str(&mut self, s: &str) -> std::fmt::Result {
for c in s.chars() {
self.write_char(c)?
}
Ok(())
}

fn write_char(&mut self, c: char) -> std::fmt::Result {
if c == '\n' {
println!("{} {}", self.0, self.1);
self.1.clear();
} else {
self.1.push(c);
}
Ok(())
fn job_finished_msg(job: JobRealization, state: JobState ) -> String {
match state {
JobState::Finished => console::style(format!("=> DONE {}", job)).green().to_string(),
JobState::Skipped => console::style(format!("=> SKIPPED {}", job)).yellow().to_string(),
JobState::Failed => console::style(format!("=> FAILED {}", job)).red().to_string(),
_ => panic!("Invalid job state after run: {:?}", state),
}
}
22 changes: 5 additions & 17 deletions src/job.rs
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
use std::collections::HashMap;
use std::fmt::{self, Write};
use std::fmt;
use std::fs;
use std::path::Path;
use std::process::Command;
Expand Down Expand Up @@ -191,7 +191,7 @@ impl InnerJobRealization {
// skip if dry run
if options.dry_run {
if options.trace {
let _ = writeln!(tracker.out(), "{}", self.cmd());
tracker.trace(&self.cmd().to_string());
}
return Ok(JobState::Finished);
}
Expand All @@ -206,7 +206,7 @@ impl InnerJobRealization {

// print out trace
if options.trace {
let _ = writeln!(tracker.out(), "{}", self.cmd());
tracker.trace(&self.cmd().to_string());
}

let cmd_with_exit_setting = format!("set -e; {}", self.run);
Expand All @@ -226,22 +226,10 @@ impl InnerJobRealization {
.stderr(io_writer)
.spawn()?;

let mut last_line: Option<String> = None;
for line in BufReader::new(io_reader).lines().map_while(Result::ok) {
let _ = writeln!(tracker.status(), "{}", line);

if options.verbose {
if let Some(line) = last_line.take() {
let _ = writeln!(tracker.out(), "{}: {}", self, line);
}
last_line = Some(line);
}
}
if options.verbose {
if let Some(line) = last_line.take() {
let _ = writeln!(tracker.out(), "{}: {}", self, line);
}
tracker.cmd_output(&self.to_string(), &line, options.verbose);
}
tracker.flush_cmd_output(&self.to_string(), options.verbose);

process
};
Expand Down
14 changes: 5 additions & 9 deletions src/queue.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,16 +62,12 @@ impl Queue {
}
}

pub fn finished(&self, finished_job: JobRealization) {
pub fn finished(&self, job: JobRealization, state: JobState) {
let mut inner = self.inner.lock().unwrap();
inner.states.insert(finished_job, JobState::Finished);
self.cond_fetch_job.notify_all();
}

pub fn failed(&self, failed_job: JobRealization) {
let mut inner = self.inner.lock().unwrap();
inner.states.insert(failed_job, JobState::Failed);
inner.failed = true;
inner.states.insert(job, state);
if state == JobState::Failed {
inner.failed = true;
}
self.cond_fetch_job.notify_all();
}

Expand Down
21 changes: 4 additions & 17 deletions src/worker.rs
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@
use std::fmt::Write;

use crate::barkeeper::ThreadStateTracker;
use crate::queue::{JobState, Queue};
use crate::Options;
Expand All @@ -11,21 +9,10 @@ pub fn run_worker(queue: Queue, mut tracker: impl ThreadStateTracker, options: O

if let Some(job) = queue.fetch() {
tracker.set_prefix(job.to_string());
if let Ok(state) = job.run(&mut tracker, &options) {
let msg = match state {
JobState::Finished => console::style(format!("=> DONE {}", job)).green().to_string(),
JobState::Skipped => console::style(format!("=> SKIPPED {}", job)).yellow().to_string(),
JobState::Failed => console::style(format!("=> FAILED {}", job)).red().to_string(),
_ => panic!("Invalid job state after run: {:?}", state),
};
let _ = writeln!(tracker.out(), "{}", msg);
queue.finished(job);
} else if let Err(e) = job.run(&mut tracker, &options) {
let msg = console::style(format!("=> FAILED {}: {}", job, e)).red().to_string();
let _ = writeln!(tracker.out(), "{}", msg);
queue.failed(job);
}
tracker.job_completed();
let state = job.run(&mut tracker, &options)
.unwrap_or(JobState::Failed);
tracker.job_completed(job.clone(), state);
queue.finished(job, state);
} else {
break;
}
Expand Down

0 comments on commit adb32cc

Please sign in to comment.