Compare commits
7 Commits
8b7d76017f
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| e8e4e3636c | |||
| 2660d7d2f5 | |||
| 9ef27716c6 | |||
| 3bac1243a1 | |||
| d196662ac6 | |||
| 44d917c264 | |||
| 46e980fe76 |
25
Cargo.lock
generated
25
Cargo.lock
generated
@@ -154,12 +154,24 @@ version = "0.8.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
|
||||
|
||||
[[package]]
|
||||
name = "equivalent"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
|
||||
|
||||
[[package]]
|
||||
name = "find-msvc-tools"
|
||||
version = "0.1.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.17.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51"
|
||||
|
||||
[[package]]
|
||||
name = "heck"
|
||||
version = "0.5.0"
|
||||
@@ -190,6 +202,16 @@ dependencies = [
|
||||
"cc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "indexmap"
|
||||
version = "2.14.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
|
||||
dependencies = [
|
||||
"equivalent",
|
||||
"hashbrown",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "is_terminal_polyfill"
|
||||
version = "1.70.2"
|
||||
@@ -413,8 +435,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "zfsbackup"
|
||||
version = "0.2.3"
|
||||
version = "0.4.0"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"clap",
|
||||
"indexmap",
|
||||
]
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
[package]
|
||||
name = "zfsbackup"
|
||||
version = "0.2.3"
|
||||
version = "0.4.0"
|
||||
edition = "2024"
|
||||
publish = false
|
||||
|
||||
[dependencies]
|
||||
chrono = "0.4.44"
|
||||
clap = { version = "4.6.1", features = ["derive"] }
|
||||
indexmap = "2.14.0"
|
||||
|
||||
@@ -6,7 +6,12 @@ pub trait Filter {
|
||||
fn filter(&self, reader: Box<dyn Read>) -> Box<dyn Read>;
|
||||
}
|
||||
|
||||
pub fn exec_command(command: &Vec<&str>) -> Result<String, String> {
|
||||
// TODO: Make this support quoting for args with spaces (and escape nested quotes)
|
||||
fn display_command(command: &Vec<&str>) -> String {
|
||||
command.join(" ")
|
||||
}
|
||||
|
||||
pub fn exec(command: &Vec<&str>) -> Result<String, String> {
|
||||
if command.is_empty() {
|
||||
return Err("Command is empty".to_string());
|
||||
}
|
||||
@@ -14,11 +19,13 @@ pub fn exec_command(command: &Vec<&str>) -> Result<String, String> {
|
||||
if command.len() > 1 {
|
||||
cmd.args(&command[1..]);
|
||||
}
|
||||
let output = cmd.output().map_err(|e| e.to_string())?;
|
||||
let output = cmd
|
||||
.output()
|
||||
.map_err(|e| format!("Failed to run command {}: {}", command[0], e))?;
|
||||
if !output.status.success() {
|
||||
return Err(format!(
|
||||
"Command {:?} failed with status {}: {}",
|
||||
command,
|
||||
"Command '{}' failed with status {}: {}",
|
||||
display_command(command),
|
||||
output.status,
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
));
|
||||
@@ -27,7 +34,10 @@ pub fn exec_command(command: &Vec<&str>) -> Result<String, String> {
|
||||
Ok(output_str)
|
||||
}
|
||||
|
||||
pub fn exec_piped_commands(
|
||||
/// Executes a pipeline of commands, where the output of the source command is passed through a
|
||||
/// series of filters before being piped into the destination command. The filters are applied in
|
||||
/// the order they are provided.
|
||||
pub fn exec_pipe(
|
||||
source: &Vec<&str>,
|
||||
dest: &Vec<&str>,
|
||||
filters: Vec<Box<dyn Filter>>,
|
||||
|
||||
157
src/job.rs
157
src/job.rs
@@ -2,11 +2,13 @@ use crate::command;
|
||||
use crate::progress::BackupEvent;
|
||||
use crate::progress::filter;
|
||||
use chrono::{Local, NaiveDateTime};
|
||||
use indexmap::set::IndexSet;
|
||||
use std::collections::HashSet;
|
||||
use std::fmt;
|
||||
use std::sync::mpsc::Sender;
|
||||
|
||||
pub struct Job {
|
||||
datasets: HashSet<String>,
|
||||
datasets: IndexSet<String>,
|
||||
target: String,
|
||||
source_zfs_command: Vec<String>,
|
||||
target_zfs_command: Vec<String>,
|
||||
@@ -14,26 +16,25 @@ pub struct Job {
|
||||
sender: Option<Sender<BackupEvent>>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
#[derive(Copy, Clone)]
|
||||
enum JobSide {
|
||||
Source,
|
||||
Destination,
|
||||
Target,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Snapshot {
|
||||
snapshot: String,
|
||||
dataset: String,
|
||||
snapshot_id: String,
|
||||
side: JobSide,
|
||||
}
|
||||
|
||||
impl Job {
|
||||
pub fn dump(&self) {
|
||||
println!("Datasets: {:?}", self.datasets);
|
||||
println!("Target: {}", self.target);
|
||||
println!("Source ZFS Command: {:?}", self.source_zfs_command);
|
||||
println!("Target ZFS Command: {:?}", self.target_zfs_command);
|
||||
impl fmt::Display for Snapshot {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}@{}", self.dataset, self.snapshot_id)
|
||||
}
|
||||
}
|
||||
|
||||
impl Job {
|
||||
fn send_event(&self, event: BackupEvent) {
|
||||
if let Some(sender) = &self.sender {
|
||||
sender.send(event).ok();
|
||||
@@ -43,16 +44,21 @@ impl Job {
|
||||
fn get_side_command(&self, side: JobSide) -> Vec<&str> {
|
||||
match side {
|
||||
JobSide::Source => self.source_zfs_command.iter().map(|s| s.as_str()).collect(),
|
||||
JobSide::Destination => self.target_zfs_command.iter().map(|s| s.as_str()).collect(),
|
||||
JobSide::Target => self.target_zfs_command.iter().map(|s| s.as_str()).collect(),
|
||||
}
|
||||
}
|
||||
|
||||
fn create_snapshot(&self, dataset: &str, side: JobSide) -> Result<String, String> {
|
||||
let snapshot = format!("{}@{}", dataset, Local::now().format("%Y-%m-%dT%H:%M:%S"));
|
||||
fn create_snapshot(&self, dataset: &str, side: JobSide) -> Result<Snapshot, String> {
|
||||
let snapshot = Snapshot {
|
||||
dataset: dataset.to_string(),
|
||||
snapshot_id: Local::now().format("%Y-%m-%dT%H:%M:%S").to_string(),
|
||||
side,
|
||||
};
|
||||
let name = snapshot.to_string();
|
||||
let mut cmd = self.get_side_command(side);
|
||||
cmd.extend(["snapshot", &snapshot]);
|
||||
command::exec_command(&cmd)?;
|
||||
self.send_event(BackupEvent::SnapshotCreated(snapshot.clone()));
|
||||
cmd.extend(["snapshot", &name]);
|
||||
command::exec(&cmd)?;
|
||||
self.send_event(BackupEvent::SnapshotCreated(name.clone()));
|
||||
Ok(snapshot)
|
||||
}
|
||||
|
||||
@@ -64,7 +70,7 @@ impl Job {
|
||||
}
|
||||
cmd.extend(["-n", "-P"]);
|
||||
cmd.push(source);
|
||||
let output = command::exec_command(&cmd)?;
|
||||
let output = command::exec(&cmd)?;
|
||||
let size = output
|
||||
.lines()
|
||||
.last()
|
||||
@@ -77,6 +83,7 @@ impl Job {
|
||||
self.send_event(BackupEvent::Estimate(size));
|
||||
Ok(size)
|
||||
}
|
||||
|
||||
fn send_receive(
|
||||
&self,
|
||||
source: &str,
|
||||
@@ -91,21 +98,24 @@ impl Job {
|
||||
}
|
||||
send_cmd.push(source);
|
||||
|
||||
let mut receive_cmd = self.get_side_command(JobSide::Destination);
|
||||
let mut receive_cmd = self.get_side_command(JobSide::Target);
|
||||
receive_cmd.extend(["receive", "-F", dest]);
|
||||
let filters = match self.sender.as_ref() {
|
||||
Some(sender) => vec![filter::CountingReaderBuilder::build(sender.clone(), total)],
|
||||
let filters: Vec<Box<dyn command::Filter>> = match self.sender.as_ref() {
|
||||
Some(sender) => vec![Box::new(filter::ByteCountFilter::new(
|
||||
sender.clone(),
|
||||
total,
|
||||
))],
|
||||
None => vec![],
|
||||
};
|
||||
command::exec_piped_commands(&send_cmd, &receive_cmd, filters)?;
|
||||
command::exec_pipe(&send_cmd, &receive_cmd, filters)?;
|
||||
self.send_event(BackupEvent::DatasetCompleted(source.to_string()));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn list_snapshots(&self, source: &str, side: JobSide) -> Result<Vec<String>, String> {
|
||||
fn list_snapshot_ids(&self, source: &str, side: JobSide) -> Result<Vec<String>, String> {
|
||||
let mut cmd = self.get_side_command(side);
|
||||
cmd.extend(["list", "-H", "-o", "name", "-t", "snapshot", source]);
|
||||
let output = command::exec_command(&cmd)?;
|
||||
let output = command::exec(&cmd)?;
|
||||
let snapshots: Vec<&str> = output
|
||||
.split_whitespace()
|
||||
.map(|s| {
|
||||
@@ -125,16 +135,16 @@ impl Job {
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
fn find_latest_matching_snapshot(&self, source: &str, dest: &str) -> Result<String, String> {
|
||||
self.find_matching_snapshots(source, dest)?
|
||||
fn find_latest_matching_snapshot_id(&self, source: &str, dest: &str) -> Result<String, String> {
|
||||
self.find_matching_snapshot_ids(source, dest)?
|
||||
.into_iter()
|
||||
.next()
|
||||
.ok_or(String::from("no matching snapshots"))
|
||||
}
|
||||
|
||||
fn find_matching_snapshots(&self, source: &str, dest: &str) -> Result<Vec<String>, String> {
|
||||
let source_snapshots = self.list_snapshots(source, JobSide::Source)?;
|
||||
let mut dest_snapshots = self.list_snapshots(dest, JobSide::Destination)?;
|
||||
fn find_matching_snapshot_ids(&self, source: &str, dest: &str) -> Result<Vec<String>, String> {
|
||||
let source_snapshots = self.list_snapshot_ids(source, JobSide::Source)?;
|
||||
let mut dest_snapshots = self.list_snapshot_ids(dest, JobSide::Target)?;
|
||||
dest_snapshots.sort_by(|a, b| b.cmp(a));
|
||||
|
||||
let source_set: HashSet<String> = source_snapshots.into_iter().collect();
|
||||
@@ -152,9 +162,9 @@ impl Job {
|
||||
}
|
||||
|
||||
fn find_old_snapshots(&self, source: &str, dest: &str) -> Result<Vec<Snapshot>, String> {
|
||||
let source_snapshots = self.list_snapshots(source, JobSide::Source)?;
|
||||
let dest_snapshots = self.list_snapshots(dest, JobSide::Destination)?;
|
||||
let matching_snapshots = self.find_matching_snapshots(source, dest)?;
|
||||
let source_snapshots = self.list_snapshot_ids(source, JobSide::Source)?;
|
||||
let dest_snapshots = self.list_snapshot_ids(dest, JobSide::Target)?;
|
||||
let matching_snapshots = self.find_matching_snapshot_ids(source, dest)?;
|
||||
if matching_snapshots.is_empty() {
|
||||
return Err(String::from("no matching snapshots found"));
|
||||
}
|
||||
@@ -163,7 +173,8 @@ impl Job {
|
||||
.into_iter()
|
||||
.filter(|s| !retain.contains(s))
|
||||
.map(|s| Snapshot {
|
||||
snapshot: format!("{}@{}", source, s),
|
||||
dataset: source.to_string(),
|
||||
snapshot_id: s,
|
||||
side: JobSide::Source,
|
||||
})
|
||||
.chain(
|
||||
@@ -171,22 +182,22 @@ impl Job {
|
||||
.into_iter()
|
||||
.filter(|s| !retain.contains(s))
|
||||
.map(|s| Snapshot {
|
||||
snapshot: format!("{}@{}", dest, s),
|
||||
side: JobSide::Destination,
|
||||
dataset: dest.to_string(),
|
||||
snapshot_id: s,
|
||||
side: JobSide::Target,
|
||||
}),
|
||||
)
|
||||
.collect();
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
fn delete_snapshot(&self, snapshot: Snapshot) -> Result<String, String> {
|
||||
fn delete_snapshot(&self, snapshot: Snapshot) -> Result<(), String> {
|
||||
let mut cmd = self.get_side_command(snapshot.side);
|
||||
cmd.extend(["destroy", &snapshot.snapshot]);
|
||||
let res = command::exec_command(&cmd);
|
||||
if res.is_ok() {
|
||||
self.send_event(BackupEvent::SnapshotDeleted(snapshot.snapshot.clone()));
|
||||
}
|
||||
res
|
||||
let name = snapshot.to_string();
|
||||
cmd.extend(["destroy", &name]);
|
||||
command::exec(&cmd)?;
|
||||
self.send_event(BackupEvent::SnapshotDeleted(name.clone()));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn run(&self, execute: bool) -> Result<(), String> {
|
||||
@@ -194,15 +205,15 @@ impl Job {
|
||||
// Check the source exists
|
||||
let mut cmd = self.get_side_command(JobSide::Source);
|
||||
cmd.extend(["list", "-H", "-o", "name", source]);
|
||||
let _ = command::exec_command(&cmd)?;
|
||||
let _ = command::exec(&cmd)?;
|
||||
|
||||
// Check whether the destination exists
|
||||
// TODO: This will assume the destination doesn't exist if the
|
||||
// command fails for any reason.
|
||||
let dest = format!("{}/{}", self.target, source);
|
||||
cmd = self.get_side_command(JobSide::Destination);
|
||||
cmd = self.get_side_command(JobSide::Target);
|
||||
cmd.extend(["list", "-H", "-o", "name", &dest]);
|
||||
let dest_exists = command::exec_command(&cmd).is_ok();
|
||||
let dest_exists = command::exec(&cmd).is_ok();
|
||||
|
||||
// Run backup
|
||||
if dest_exists {
|
||||
@@ -215,20 +226,19 @@ impl Job {
|
||||
let inc_snapshot = format!(
|
||||
"{}@{}",
|
||||
source,
|
||||
self.find_latest_matching_snapshot(source, &dest)?
|
||||
self.find_latest_matching_snapshot_id(source, &dest)?
|
||||
);
|
||||
|
||||
let snapshot = self.create_snapshot(source, JobSide::Source)?;
|
||||
let total = self.estimate(&snapshot, Some(&inc_snapshot)).ok();
|
||||
let total = self
|
||||
.estimate(&snapshot.to_string(), Some(&inc_snapshot))
|
||||
.ok();
|
||||
if !execute {
|
||||
self.send_event(BackupEvent::DryrunCompleted(source.clone()));
|
||||
self.delete_snapshot(Snapshot {
|
||||
snapshot: snapshot.clone(),
|
||||
side: JobSide::Source,
|
||||
})?;
|
||||
return Ok(());
|
||||
self.delete_snapshot(snapshot)?;
|
||||
continue;
|
||||
}
|
||||
self.send_receive(&snapshot, &dest, Some(&inc_snapshot), total)?;
|
||||
self.send_receive(&snapshot.to_string(), &dest, Some(&inc_snapshot), total)?;
|
||||
} else {
|
||||
self.send_event(BackupEvent::StartingFullBackup {
|
||||
source: source.clone(),
|
||||
@@ -237,16 +247,13 @@ impl Job {
|
||||
total: self.datasets.len(),
|
||||
});
|
||||
let snapshot = self.create_snapshot(source, JobSide::Source)?;
|
||||
let total = self.estimate(&snapshot, None).ok();
|
||||
let total = self.estimate(&snapshot.to_string(), None).ok();
|
||||
if !execute {
|
||||
self.send_event(BackupEvent::DryrunCompleted(source.clone()));
|
||||
self.delete_snapshot(Snapshot {
|
||||
snapshot: snapshot.clone(),
|
||||
side: JobSide::Source,
|
||||
})?;
|
||||
return Ok(());
|
||||
self.delete_snapshot(snapshot)?;
|
||||
continue;
|
||||
}
|
||||
self.send_receive(&snapshot, &dest, None, total)?;
|
||||
self.send_receive(&snapshot.to_string(), &dest, None, total)?;
|
||||
}
|
||||
|
||||
// Clean up snapshots
|
||||
@@ -278,24 +285,23 @@ fn parse_command(commandstr: &str) -> Vec<String> {
|
||||
|
||||
for c in commandstr.chars() {
|
||||
match c {
|
||||
'\'' => {
|
||||
if !in_double_quote {
|
||||
'\'' if !in_double_quote => {
|
||||
in_single_quote = !in_single_quote;
|
||||
} else {
|
||||
}
|
||||
'\'' => {
|
||||
arg.push(c);
|
||||
}
|
||||
'"' if !in_single_quote => {
|
||||
in_double_quote = !in_double_quote;
|
||||
}
|
||||
'"' => {
|
||||
if !in_single_quote {
|
||||
in_double_quote = !in_double_quote;
|
||||
} else {
|
||||
arg.push(c);
|
||||
}
|
||||
' ' if in_single_quote || in_double_quote => {
|
||||
arg.push(c);
|
||||
}
|
||||
' ' => {
|
||||
if in_single_quote || in_double_quote {
|
||||
arg.push(c);
|
||||
} else if !arg.is_empty() {
|
||||
if !arg.is_empty() {
|
||||
command.push(arg.clone());
|
||||
arg.clear();
|
||||
}
|
||||
@@ -356,7 +362,7 @@ impl JobBuilder {
|
||||
}
|
||||
|
||||
pub fn build(self) -> Result<Job, String> {
|
||||
let mut datasets: HashSet<String> = HashSet::new();
|
||||
let mut datasets: IndexSet<String> = IndexSet::new();
|
||||
for source in &self.sources {
|
||||
let recurse = source.ends_with("/...");
|
||||
let source = source.trim_end_matches("/...");
|
||||
@@ -369,14 +375,14 @@ impl JobBuilder {
|
||||
|
||||
let mut cmd: Vec<&str> = self.source_zfs_command.iter().map(|s| s.as_str()).collect();
|
||||
cmd.extend(args);
|
||||
let output = command::exec_command(&cmd)?;
|
||||
let output = command::exec(&cmd)?;
|
||||
datasets.extend(output.lines().map(str::to_string));
|
||||
}
|
||||
if datasets.is_empty() {
|
||||
return Err(String::from("no matching source datasets found"));
|
||||
}
|
||||
Ok(Job {
|
||||
datasets: datasets,
|
||||
datasets,
|
||||
target: self.target,
|
||||
source_zfs_command: self.source_zfs_command,
|
||||
target_zfs_command: self.target_zfs_command,
|
||||
@@ -386,13 +392,6 @@ impl JobBuilder {
|
||||
}
|
||||
}
|
||||
|
||||
struct JobConfig {
|
||||
sources: Vec<String>,
|
||||
target: String,
|
||||
source_zfs_command: Option<String>,
|
||||
target_zfs_command: Option<String>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -4,7 +4,7 @@ use std::io::{self, Read};
|
||||
use std::sync::mpsc::Sender;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
struct CountingReader<R: Read> {
|
||||
struct ByteCountReader<R: Read> {
|
||||
inner: R,
|
||||
sender: Sender<BackupEvent>,
|
||||
bytes: u64,
|
||||
@@ -12,7 +12,7 @@ struct CountingReader<R: Read> {
|
||||
total: Option<u64>,
|
||||
}
|
||||
|
||||
impl<R: Read> Read for CountingReader<R> {
|
||||
impl<R: Read> Read for ByteCountReader<R> {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
let n = self.inner.read(buf)?;
|
||||
self.bytes += n as u64;
|
||||
@@ -29,31 +29,31 @@ impl<R: Read> Read for CountingReader<R> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<R: Read> CountingReader<R> {
|
||||
impl<R: Read> ByteCountReader<R> {
|
||||
fn new(inner: R, sender: Sender<BackupEvent>, total: Option<u64>) -> Self {
|
||||
Self {
|
||||
inner,
|
||||
sender,
|
||||
total,
|
||||
bytes: 0,
|
||||
last_send: Instant::now() - Duration::from_secs(1),
|
||||
total,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct CountingReaderBuilder {
|
||||
pub(crate) struct ByteCountFilter {
|
||||
sender: Sender<BackupEvent>,
|
||||
total: Option<u64>,
|
||||
}
|
||||
|
||||
impl Filter for CountingReaderBuilder {
|
||||
fn filter(&self, reader: Box<dyn Read>) -> Box<dyn Read> {
|
||||
Box::new(CountingReader::new(reader, self.sender.clone(), self.total))
|
||||
impl ByteCountFilter {
|
||||
pub(crate) fn new(sender: Sender<BackupEvent>, total: Option<u64>) -> Self {
|
||||
Self { sender, total }
|
||||
}
|
||||
}
|
||||
|
||||
impl CountingReaderBuilder {
|
||||
pub fn build(sender: Sender<BackupEvent>, total: Option<u64>) -> Box<dyn Filter> {
|
||||
Box::new(Self { sender, total })
|
||||
impl Filter for ByteCountFilter {
|
||||
fn filter(&self, inner: Box<dyn Read>) -> Box<dyn Read> {
|
||||
Box::new(ByteCountReader::new(inner, self.sender.clone(), self.total))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,21 +1,30 @@
|
||||
pub mod filter;
|
||||
pub mod log;
|
||||
mod rate;
|
||||
pub mod terminal;
|
||||
|
||||
pub trait Progressor: Send {
|
||||
fn run(&mut self);
|
||||
}
|
||||
|
||||
fn human_bytes(bytes: u64) -> String {
|
||||
fn human_bytes(bytes: f64) -> String {
|
||||
let units = ["B", "kB", "MB", "GB", "TB"];
|
||||
|
||||
let mut value = bytes as f64;
|
||||
let mut index = 0;
|
||||
while value >= 1024.0 && index < units.len() - 1 {
|
||||
value /= 1024.0;
|
||||
index += 1;
|
||||
let mut a = bytes;
|
||||
let mut i = 0;
|
||||
while a >= 1024.0 && i < units.len() - 1 {
|
||||
a /= 1024.0;
|
||||
i += 1;
|
||||
}
|
||||
format!("{:.2} {:2}", value, units[index])
|
||||
format!("{:.2} {:2}", a, units[i])
|
||||
}
|
||||
|
||||
fn format_eta(seconds: f64) -> String {
|
||||
let secs = seconds as u64;
|
||||
let h = secs / 3600;
|
||||
let m = (secs % 3600) / 60;
|
||||
let s = secs % 60;
|
||||
format!("{:02}:{:02}:{:02}", h, m, s)
|
||||
}
|
||||
|
||||
pub enum BackupEvent {
|
||||
|
||||
74
src/progress/rate.rs
Normal file
74
src/progress/rate.rs
Normal file
@@ -0,0 +1,74 @@
|
||||
use std::time::Instant;
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct Sample {
|
||||
at: Instant,
|
||||
bytes: u64,
|
||||
}
|
||||
|
||||
pub(crate) struct RateTracker {
|
||||
samples: Vec<Sample>,
|
||||
head: usize,
|
||||
count: usize,
|
||||
smoothed: f64,
|
||||
alpha: f64,
|
||||
}
|
||||
|
||||
impl RateTracker {
|
||||
pub(crate) fn new(capacity: usize) -> Self {
|
||||
RateTracker {
|
||||
samples: vec![
|
||||
Sample {
|
||||
at: Instant::now(),
|
||||
bytes: 0,
|
||||
};
|
||||
capacity
|
||||
],
|
||||
head: 0,
|
||||
count: 0,
|
||||
smoothed: 0.0,
|
||||
alpha: 0.05,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn push(&mut self, bytes: u64) {
|
||||
let sample = Sample {
|
||||
at: Instant::now(),
|
||||
bytes,
|
||||
};
|
||||
self.samples[self.head] = sample;
|
||||
self.head = (self.head + 1) % self.samples.capacity();
|
||||
self.count = (self.count + 1).min(self.samples.capacity())
|
||||
}
|
||||
|
||||
pub(crate) fn rate(&mut self) -> Option<f64> {
|
||||
if self.count < 2 {
|
||||
return None;
|
||||
}
|
||||
|
||||
let cap = self.samples.capacity();
|
||||
|
||||
let oldest_idx = (cap + self.head - self.count) % cap;
|
||||
let newest_idx = (cap + self.head - 1) % cap;
|
||||
|
||||
let oldest = self.samples[oldest_idx];
|
||||
let newest = self.samples[newest_idx];
|
||||
|
||||
let elapsed = newest.at.duration_since(oldest.at).as_secs_f64();
|
||||
if elapsed == 0.0 {
|
||||
return None;
|
||||
}
|
||||
|
||||
// If for any reason oldest < newest, the subtraction will overflow
|
||||
if newest.bytes < oldest.bytes {
|
||||
return None;
|
||||
}
|
||||
|
||||
let rate = (newest.bytes - oldest.bytes) as f64 / elapsed;
|
||||
|
||||
// exponentially weighted moving average to smooth those spikes
|
||||
self.smoothed = self.alpha * rate + (1.0 - self.alpha) * self.smoothed;
|
||||
|
||||
Some(self.smoothed)
|
||||
}
|
||||
}
|
||||
@@ -1,9 +1,13 @@
|
||||
use crate::progress::format_eta;
|
||||
|
||||
use super::rate::RateTracker;
|
||||
use super::{BackupEvent, human_bytes};
|
||||
use std::{io::Write, sync::mpsc::Receiver};
|
||||
|
||||
pub struct Progressor {
|
||||
receiver: Receiver<BackupEvent>,
|
||||
estimated_size: u64,
|
||||
rate_tracker: RateTracker,
|
||||
}
|
||||
|
||||
impl super::Progressor for Progressor {
|
||||
@@ -11,7 +15,7 @@ impl super::Progressor for Progressor {
|
||||
while let Ok(event) = self.receiver.recv() {
|
||||
match event {
|
||||
BackupEvent::Estimate(size) => {
|
||||
println!("Estimated total backup size: {}", human_bytes(size));
|
||||
println!("Estimated total backup size: {}", human_bytes(size as f64));
|
||||
self.estimated_size = size;
|
||||
}
|
||||
BackupEvent::StartingFullBackup {
|
||||
@@ -53,15 +57,34 @@ impl super::Progressor for Progressor {
|
||||
} else {
|
||||
0.0
|
||||
};
|
||||
self.rate_tracker.push(bytes);
|
||||
let rate = self.rate_tracker.rate();
|
||||
let rate_display = match rate {
|
||||
None => String::from("[unknown]"),
|
||||
Some(0.0) => {
|
||||
String::from("[stalled] ETA heat death of the universe")
|
||||
}
|
||||
Some(rate) => {
|
||||
let eta = if bytes > total {
|
||||
String::from("any moment now")
|
||||
} else {
|
||||
let eta = (total - bytes) as f64 / rate;
|
||||
format_eta(eta)
|
||||
};
|
||||
format!("{}/s ETA {}", human_bytes(rate), eta)
|
||||
}
|
||||
};
|
||||
|
||||
print!(
|
||||
"{:>3.0}% {}/{} transferred\r",
|
||||
"\x1b[2K{:>3.0}% {}/{} @ {}\r",
|
||||
percent * 100.0,
|
||||
human_bytes(bytes),
|
||||
human_bytes(total)
|
||||
human_bytes(bytes as f64),
|
||||
human_bytes(total as f64),
|
||||
rate_display,
|
||||
);
|
||||
}
|
||||
None => {
|
||||
print!("{} transferred\r", human_bytes(bytes));
|
||||
print!("\x1b[2K{} transferred\r", human_bytes(bytes as f64));
|
||||
}
|
||||
}
|
||||
std::io::stdout().flush().ok();
|
||||
@@ -82,6 +105,7 @@ impl Progressor {
|
||||
Self {
|
||||
receiver,
|
||||
estimated_size: 0,
|
||||
rate_tracker: RateTracker::new(100),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user