use blake3::Hash;
use ffsync::chunking::{
Blake3Hasher, Chunk, ChunkEvent, FastCDCChunker, Hasher, PipelineConfig, cdc_params_for_size,
process_file,
};
use ffsync::index::FileIndex;
use ffsync::protocol::sync::SyncOutcome;
use ffsync::reconstruction::{ChunkSource, FileExt, ReconstructionTask, Reconstructor};
use rand::{Rng, RngExt, SeedableRng, rngs::StdRng};
use std::collections::{HashMap, HashSet};
use std::fs::{self, File, FileTimes, OpenOptions};
use std::io::{self, Seek, Write};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{Duration, Instant, SystemTime};
const WRITE_CHUNK_BYTES: usize = 4 * 1024 * 1024;
const DEFAULT_SIZE_MIB: u64 = 128;
const QUICK_SIZE_MIB: u64 = 32;
const DEFAULT_OUT_DIR: &str = "./bench_tmp";
const LINKS: [(f64, &str); 4] = [
(20.0, "20 Mbit/s"),
(100.0, "100 Mbit/s"),
(1000.0, "1 Gbit/s"),
(10000.0, "10 Gbit/s"),
];
const RATIOS: [f64; 3] = [1.0, 2.0, 4.0];
#[derive(Clone, Copy, PartialEq, Eq)]
enum Scenario {
Identical,
Modified,
Appended,
Fresh,
}
impl Scenario {
const ALL: [Scenario; 4] = [
Scenario::Identical,
Scenario::Modified,
Scenario::Appended,
Scenario::Fresh,
];
fn name(self) -> &'static str {
match self {
Scenario::Identical => "identical",
Scenario::Modified => "modified",
Scenario::Appended => "appended",
Scenario::Fresh => "fresh",
}
}
}
struct BenchConfig {
size_mib: u64,
out_dir: PathBuf,
quick: bool,
rsync: bool,
json: bool,
size_explicit: bool,
seed: Option<u64>,
cdc: Option<(usize, usize, usize)>,
sweep: bool,
disconnect: bool,
warm: bool,
eager: bool,
tree: bool,
latency: bool,
gen_tree: bool,
rtt_ms: u64,
parallel_counts: Vec<u16>,
}
impl BenchConfig {
fn size_bytes(&self) -> u64 {
self.size_mib * 1024 * 1024
}
}
struct ScanStats {
total: Duration,
first_chunk: Option<Duration>,
chunks: usize,
hashed_bytes: u64,
}
struct DeltaStats {
target_bytes: u64,
missing_bytes: u64,
}
impl DeltaStats {
fn pct(&self) -> f64 {
if self.target_bytes == 0 {
0.0
} else {
self.missing_bytes as f64 / self.target_bytes as f64 * 100.0
}
}
}
struct ScenarioResult {
scenario: Scenario,
variant_size: u64,
source_scan: ScanStats,
target_scan: ScanStats,
delta: DeltaStats,
reconstruct: Duration,
}
struct RsyncRow {
mode: &'static str,
wall: Duration,
transferred: Option<u64>,
matched: Option<u64>,
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum RsyncMode {
Native,
Wsl,
}
impl RsyncMode {
fn label(self) -> &'static str {
match self {
RsyncMode::Native => "native",
RsyncMode::Wsl => "wsl",
}
}
}
enum RsyncProbe {
Available(RsyncMode),
Unavailable(String),
}
struct RsyncResults {
mode: Option<RsyncMode>,
rows: Vec<RsyncRow>,
}
fn print_usage() {
println!(
"ffsync benchmark harness
Usage: cargo run --bin bench -- [options]
Options:
--size-mib <N> fixture size in MiB (default: {})
--quick use a {} MiB fixture for a fast run
--seed <N> deterministic fixture seed (comparable runs across code changes)
--cdc <m/a/M> FastCDC sizes, e.g. 64k/256k/1m (default: derived from
the fixture size, D19)
--sweep run a chunk-size sweep (6 CDC configs) and compare deltas
--disconnect simulate a mid-transfer disconnect and compare resume costs
--warm measure the warm-start index: cold scan + index write, then
reload (stat-only) with content verification
--tree simulate a directory-tree session (D17): generate a
multi-file tree, run both sides' scan + triage, measure
per-file delta, session-level fast path, and warm starts
--eager eager-diff simulation (D8/§9.2): paired lazy-vs-eager
single-file sessions over RTT-modeled duplex, reporting
wall time and over-send bytes (use with --rtt-ms)
--latency run real in-process duplex tree sessions (the send/
receive state machines over in-memory links) and time
them per --parallel level
--rtt-ms <N> simulated round-trip time for --latency runs (default 30)
--parallel <K,...> comma list of connection counts for --latency,
e.g. 1,4,8 (each 1..=16)
--gen-tree emit the 100-file tree fixture into <out-dir>/lat_sender
and exit (used by the wan-measure workflow)
--out-dir <path> directory for fixtures and scratch output (default: {})
--rsync additionally compare against rsync (native, or via WSL)
--json print a machine-readable JSON summary as the last line
-h, --help show this help",
DEFAULT_SIZE_MIB, QUICK_SIZE_MIB, DEFAULT_OUT_DIR
);
}
fn die(msg: &str) -> ! {
eprintln!("bench: error: {}", msg);
print_usage();
std::process::exit(2);
}
fn parse_cdc(value: &str) -> (usize, usize, usize) {
let parse_unit = |s: &str| -> usize {
let s = s.trim().to_ascii_lowercase();
let (digits, mult) = match s.as_bytes().last() {
Some(b'k') => (&s[..s.len() - 1], 1024usize),
Some(b'm') => (&s[..s.len() - 1], 1024 * 1024),
Some(b'g') => (&s[..s.len() - 1], 1024 * 1024 * 1024),
_ => (s.as_str(), 1usize),
};
let n: usize = digits
.parse()
.unwrap_or_else(|_| die(&format!("invalid cdc size: {}", s)));
n * mult
};
let parts: Vec<&str> = value.split('/').collect();
if parts.len() != 3 {
die("--cdc expects min/avg/max, e.g. --cdc 64k/256k/1m");
}
let triple = (
parse_unit(parts[0]),
parse_unit(parts[1]),
parse_unit(parts[2]),
);
if triple.0 == 0 || triple.1 < triple.0 || triple.2 < triple.1 {
die("--cdc requires 0 < min <= avg <= max");
}
if triple.0 % 2 != 0 || triple.1 % 2 != 0 || triple.2 % 2 != 0 {
die("--cdc values must be even (fastcdc 5.0 requirement)");
}
triple
}
const SWEEP_POINTS: [(usize, usize, usize); 6] = [
(16 * 1024, 64 * 1024, 256 * 1024),
(32 * 1024, 128 * 1024, 512 * 1024),
(64 * 1024, 256 * 1024, 1024 * 1024),
(128 * 1024, 512 * 1024, 2 * 1024 * 1024),
(256 * 1024, 1024 * 1024, 4 * 1024 * 1024),
(512 * 1024, 2 * 1024 * 1024, 8 * 1024 * 1024),
];
fn parse_args() -> BenchConfig {
let mut cfg = BenchConfig {
size_mib: DEFAULT_SIZE_MIB,
out_dir: PathBuf::from(DEFAULT_OUT_DIR),
quick: false,
rsync: false,
json: false,
size_explicit: false,
seed: None,
cdc: None,
sweep: false,
disconnect: false,
warm: false,
tree: false,
latency: false,
gen_tree: false,
rtt_ms: 30,
parallel_counts: Vec::new(),
eager: false,
};
let mut args = std::env::args().skip(1);
while let Some(arg) = args.next() {
match arg.as_str() {
"--size-mib" => {
let value = args
.next()
.unwrap_or_else(|| die("--size-mib requires a value"));
let parsed: u64 = value
.parse()
.unwrap_or_else(|_| die(&format!("invalid --size-mib value: {}", value)));
if parsed == 0 {
die("--size-mib must be at least 1");
}
cfg.size_mib = parsed;
cfg.size_explicit = true;
}
"--quick" => cfg.quick = true,
"--out-dir" => {
let value = args
.next()
.unwrap_or_else(|| die("--out-dir requires a value"));
cfg.out_dir = PathBuf::from(value);
}
"--rsync" => cfg.rsync = true,
"--json" => cfg.json = true,
"--seed" => {
let value = args
.next()
.unwrap_or_else(|| die("--seed requires a value"));
let parsed: u64 = value
.parse()
.unwrap_or_else(|_| die(&format!("invalid --seed value: {}", value)));
cfg.seed = Some(parsed);
}
"--cdc" => {
let value = args
.next()
.unwrap_or_else(|| die("--cdc requires min/avg/max"));
let parsed = parse_cdc(&value);
cfg.cdc = Some(parsed);
}
"--sweep" => cfg.sweep = true,
"--disconnect" => cfg.disconnect = true,
"--warm" => cfg.warm = true,
"--tree" => cfg.tree = true,
"--latency" => cfg.latency = true,
"--eager" => cfg.eager = true,
"--gen-tree" => cfg.gen_tree = true,
"--rtt-ms" => {
let value = args
.next()
.unwrap_or_else(|| die("--rtt-ms requires a number"));
cfg.rtt_ms = value
.parse()
.unwrap_or_else(|_| die(&format!("invalid --rtt-ms value: {}", value)));
}
"--parallel" => {
let value = args
.next()
.unwrap_or_else(|| die("--parallel requires a comma list, e.g. 1,4,8"));
let counts: Vec<u16> = value
.split(',')
.map(|p| {
p.trim()
.parse()
.unwrap_or_else(|_| die(&format!("invalid --parallel value: {}", p)))
})
.collect();
if counts.iter().any(|c| *c < 1 || *c > 16) {
die("--parallel counts must be 1..=16");
}
cfg.parallel_counts = counts;
}
"-h" | "--help" => {
print_usage();
std::process::exit(0);
}
other => die(&format!("unknown argument: {}", other)),
}
}
if cfg.quick && !cfg.size_explicit {
cfg.size_mib = QUICK_SIZE_MIB;
}
cfg
}
fn write_random_file(path: &Path, size: u64, rng: &mut impl Rng) -> io::Result<()> {
let mut file = File::create(path)?;
let mut buf = vec![0u8; WRITE_CHUNK_BYTES];
let mut remaining = size;
while remaining > 0 {
let n = remaining.min(WRITE_CHUNK_BYTES as u64) as usize;
rng.fill_bytes(&mut buf[..n]);
file.write_all(&buf[..n])?;
remaining -= n as u64;
}
Ok(())
}
fn flip_middle_regions(path: &Path, size: u64, rng: &mut impl Rng) -> io::Result<()> {
let region_len = (size / 100).max(4096).min(size / 4);
let mut buf = vec![0u8; region_len as usize];
let file = OpenOptions::new().write(true).open(path)?;
for pos in [size * 45 / 100, size * 55 / 100] {
rng.fill_bytes(&mut buf);
file.write_at(&buf, pos)?;
}
Ok(())
}
fn append_random_tail(path: &Path, size: u64, rng: &mut impl Rng) -> io::Result<()> {
let tail_len = (size / 10).max(1);
let mut buf = vec![0u8; tail_len.min(WRITE_CHUNK_BYTES as u64) as usize];
let mut file = OpenOptions::new().append(true).open(path)?;
let mut remaining = tail_len;
while remaining > 0 {
let n = remaining.min(buf.len() as u64) as usize;
rng.fill_bytes(&mut buf[..n]);
file.write_all(&buf[..n])?;
remaining -= n as u64;
}
Ok(())
}
fn generate_fixtures(cfg: &BenchConfig) -> io::Result<(PathBuf, Vec<(Scenario, PathBuf)>)> {
fs::create_dir_all(&cfg.out_dir)?;
let size = cfg.size_bytes();
let mut rng = match cfg.seed {
Some(seed) => StdRng::seed_from_u64(seed),
None => StdRng::from_rng(&mut rand::rng()),
};
let source = cfg.out_dir.join("source_A.bin");
write_random_file(&source, size, &mut rng)?;
println!(
"fixture: source A {:>9} {}",
fmt_bytes(size),
source.display()
);
let mut variants = Vec::new();
for scenario in Scenario::ALL {
let path = cfg.out_dir.join(format!("variant_{}.bin", scenario.name()));
match scenario {
Scenario::Identical => {
fs::copy(&source, &path)?;
}
Scenario::Modified => {
fs::copy(&source, &path)?;
flip_middle_regions(&path, size, &mut rng)?;
}
Scenario::Appended => {
fs::copy(&source, &path)?;
append_random_tail(&path, size, &mut rng)?;
}
Scenario::Fresh => {
write_random_file(&path, size, &mut rng)?;
}
}
let variant_size = fs::metadata(&path)?.len();
println!(
"fixture: variant {:<9} {:>9} {}",
scenario.name(),
fmt_bytes(variant_size),
path.display()
);
variants.push((scenario, path));
}
Ok((source, variants))
}
fn scan_file(
path: &Path,
pipeline: &PipelineConfig,
cdc: (usize, usize, usize),
) -> io::Result<(ScanStats, Vec<Chunk>)> {
let start = Instant::now();
let (min, avg, max) = cdc;
let rx = process_file(
path,
move || FastCDCChunker::new(min, avg, max),
Blake3Hasher,
pipeline,
)?;
let mut first_chunk = None;
let mut chunks = Vec::new();
let mut hashed_bytes = 0u64;
loop {
match rx.recv() {
Ok(ChunkEvent::Chunk(chunk)) => {
if first_chunk.is_none() {
first_chunk = Some(start.elapsed());
}
hashed_bytes += chunk.length as u64;
chunks.push(chunk);
}
Ok(ChunkEvent::Done) => {}
Err(_) => break, }
}
chunks.sort_by_key(|c| c.offset);
let stats = ScanStats {
total: start.elapsed(),
first_chunk,
chunks: chunks.len(),
hashed_bytes,
};
Ok((stats, chunks))
}
fn compute_delta(source_chunks: &[Chunk], target_chunks: &[Chunk]) -> DeltaStats {
let source_hashes: HashSet<Hash> = source_chunks.iter().map(|c| c.hash).collect();
let mut missing_bytes = 0u64;
for chunk in target_chunks {
if !source_hashes.contains(&chunk.hash) {
missing_bytes += chunk.length as u64;
}
}
DeltaStats {
target_bytes: target_chunks.iter().map(|c| c.length as u64).sum(),
missing_bytes,
}
}
fn file_root_hash(path: &Path) -> io::Result<Hash> {
Blake3Hasher.hash_reader(&mut File::open(path)?)
}
struct DisconnectStats {
kill_pct: u32,
cold_restart_bytes: u64,
temp_only_resume_bytes: u64,
union_resume_bytes: u64,
temp_written_bytes: u64,
completion_verified: bool,
}
#[allow(clippy::too_many_arguments)]
fn simulate_disconnect(
source: &Path,
variant: &Path,
source_chunks: &[Chunk],
variant_chunks: &[Chunk],
variant_size: u64,
variant_hash: Hash,
cdc: (usize, usize, usize),
kill_pct: u32,
) -> io::Result<DisconnectStats> {
let rec_path = variant.with_file_name("disconnect_sim.bin");
let temp_path = rec_path.with_extension("ffsync-tmp");
let _ = fs::remove_file(&temp_path);
let source_hashes: HashSet<Hash> = source_chunks.iter().map(|c| c.hash).collect();
let remote_total = variant_chunks
.iter()
.filter(|c| !source_hashes.contains(&c.hash))
.count();
let kill_count = ((remote_total as u64 * kill_pct as u64) / 100) as usize;
let reconstructor = Reconstructor::new(&rec_path, source, variant_size)?;
let source_file = File::open(source)?;
{
let mut local_scratch: Vec<u8> = Vec::new();
for chunk in variant_chunks {
if let Some(&src_off) = source_index_lookup(source_chunks, chunk.hash) {
local_scratch.clear();
local_scratch.resize(chunk.length, 0);
let read = source_file.read_at(&mut local_scratch[..chunk.length], src_off)?;
if read != chunk.length {
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "short read"));
}
let task = ReconstructionTask {
target_offset: chunk.offset,
source: ChunkSource {
offset: src_off,
length: chunk.length,
},
};
reconstructor.write_chunk(&task, &local_scratch[..chunk.length])?;
}
}
}
let variant_file = File::open(variant)?;
let mut scratch: Vec<u8> = Vec::new();
let mut remote_seen = 0usize;
let mut temp_written_bytes = 0u64;
for chunk in variant_chunks {
if source_hashes.contains(&chunk.hash) {
continue; }
if remote_seen >= kill_count {
break; }
remote_seen += 1;
scratch.clear();
scratch.resize(chunk.length, 0);
let read = variant_file.read_at(&mut scratch[..chunk.length], chunk.offset)?;
if read != chunk.length {
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "short read"));
}
let task = ReconstructionTask {
target_offset: chunk.offset,
source: ChunkSource {
offset: 0,
length: chunk.length,
},
};
reconstructor.write_chunk(&task, &scratch)?;
temp_written_bytes += chunk.length as u64;
}
drop(reconstructor);
let (_, temp_chunks) = scan_file(&temp_path, &PipelineConfig::default(), cdc)?;
let temp_hashes: HashSet<Hash> = temp_chunks.iter().map(|c| c.hash).collect();
let temp_matching: u64 = variant_chunks
.iter()
.filter(|c| temp_hashes.contains(&c.hash))
.map(|c| c.length as u64)
.sum();
eprintln!(
" [dbg] kill={} wrote={} variant_chunks={} temp_chunks={} temp_bytes_matching_variant={}",
kill_pct,
temp_written_bytes,
variant_chunks.len(),
temp_chunks.len(),
temp_matching
);
let mut cold_restart_bytes = 0u64;
let mut temp_only_resume_bytes = 0u64;
let mut union_resume_bytes = 0u64;
for chunk in variant_chunks {
let len = chunk.length as u64;
if !source_hashes.contains(&chunk.hash) {
cold_restart_bytes += len;
}
if !temp_hashes.contains(&chunk.hash) {
temp_only_resume_bytes += len;
}
if !temp_hashes.contains(&chunk.hash) && !source_hashes.contains(&chunk.hash) {
union_resume_bytes += len;
}
}
let completion_verified = {
let temp = OpenOptions::new().read(true).write(true).open(&temp_path)?;
let mut local_scratch: Vec<u8> = Vec::new();
let mut remote_seen = 0usize;
for chunk in variant_chunks {
if source_hashes.contains(&chunk.hash) {
continue;
}
let already = remote_seen < kill_count;
remote_seen += 1;
if already {
continue;
}
scratch.clear();
scratch.resize(chunk.length, 0);
let read = variant_file.read_at(&mut scratch[..chunk.length], chunk.offset)?;
if read != chunk.length {
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "short read"));
}
temp.write_at(&scratch[..chunk.length], chunk.offset)?;
}
let source_file = File::open(source)?;
for chunk in variant_chunks {
if let Some(&src_off) = source_index_lookup(source_chunks, chunk.hash) {
local_scratch.clear();
local_scratch.resize(chunk.length, 0);
let read = source_file.read_at(&mut local_scratch[..chunk.length], src_off)?;
if read != chunk.length {
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "short read"));
}
temp.write_at(&local_scratch[..chunk.length], chunk.offset)?;
}
}
temp.sync_all()?;
let actual = file_root_hash(&temp_path)?;
actual == variant_hash
};
let _ = fs::remove_file(&temp_path);
Ok(DisconnectStats {
kill_pct,
cold_restart_bytes,
temp_only_resume_bytes,
union_resume_bytes,
temp_written_bytes,
completion_verified,
})
}
fn source_index_lookup(source_chunks: &[Chunk], hash: Hash) -> Option<&u64> {
source_chunks
.iter()
.find(|c| c.hash == hash)
.map(|c| &c.offset)
}
fn run_disconnect_simulation(
source: &Path,
variant: &Path,
source_chunks: &[Chunk],
variant_chunks: &[Chunk],
cdc: (usize, usize, usize),
) -> io::Result<()> {
let variant_size = fs::metadata(variant)?.len();
let variant_hash = file_root_hash(variant)?;
println!();
println!("==== disconnect simulation (scenario: modified) ====");
println!("kill % written cold restart temp-only resume union resume completed");
let mut all_verified = true;
for kill_pct in [25u32, 50, 75] {
let stats = simulate_disconnect(
source,
variant,
source_chunks,
variant_chunks,
variant_size,
variant_hash,
cdc,
kill_pct,
)?;
all_verified &= stats.completion_verified;
println!(
"{:>5}% {:>9} {:>13} {:>17} {:>13} {:>9}",
stats.kill_pct,
fmt_bytes(stats.temp_written_bytes),
fmt_bytes(stats.cold_restart_bytes),
fmt_bytes(stats.temp_only_resume_bytes),
fmt_bytes(stats.union_resume_bytes),
if stats.completion_verified {
"yes"
} else {
"NO"
},
);
}
println!();
println!(
"Reading: 'cold restart' is what v0.1 does today (re-send the full delta);\
\n'temp-only' resumes from the temp file re-scan; 'union' also matches the old\
\nfile's chunks (temp-first). Completion 'yes' proves the partial temp state\
\nis recoverable end-to-end."
);
if !all_verified {
return Err(io::Error::other(
"disconnect simulation failed verification",
));
}
Ok(())
}
fn run_warm_simulation(
file: &Path,
pipeline: &PipelineConfig,
cdc: (usize, usize, usize),
) -> io::Result<()> {
println!();
println!("==== warm-start index simulation (D15) ====");
let size = fs::metadata(file)?.len();
let (scan_stats, chunks) = scan_file(file, pipeline, cdc)?;
let index = FileIndex::from_scan(
chunks.clone(),
size,
ffsync::index::stat_mtime(file).unwrap_or((0, 0)),
(cdc.0 as u32, cdc.1 as u32, cdc.2 as u32),
);
let t_write = Instant::now();
ffsync::index::write_index(file, &index)?;
let write_us = t_write.elapsed().as_secs_f64() * 1e6;
let t_load = Instant::now();
let warm = ffsync::index::load_index(file);
let load_us = t_load.elapsed().as_secs_f64() * 1e6;
let Some(warm_idx) = warm else {
return Err(io::Error::other("warm index failed to load/validate"));
};
let content_ok = warm_idx == index;
let encoded_len = index.encode().len();
println!(
"file: {} ({} chunks, index {})",
fmt_bytes(size),
scan_stats.chunks,
fmt_bytes(encoded_len as u64)
);
println!(
"cold scan {:>10.2} ms ({} hashed, what every run pays without the index)",
scan_stats.total.as_secs_f64() * 1e3,
fmt_bytes(scan_stats.hashed_bytes)
);
println!(
"index write {:>10.2} µs (once per successful sync, at commit)",
write_us
);
println!(
"warm load {:>10.2} µs (validate + decode, stat-only — no file contents read)",
load_us
);
println!(
"speedup {:>10.1}x (cold scan / warm load)",
scan_stats.total.as_secs_f64() * 1e6 / load_us.max(0.01)
);
if !content_ok {
return Err(io::Error::other(
"warm index does not match the cold scan (root or chunk mismatch)",
));
}
println!("content verification: index matches cold scan exactly (root + all chunks)");
Ok(())
}
const TREE_FILE_COUNT: usize = 40;
const TREE_DIR_COUNT: usize = 6;
const TREE_MODIFIED_FRAC: usize = 4; const TREE_NEW_FRAC: usize = 10; const TREE_STALE_FRAC: usize = 20;
struct TreeStats {
files: usize,
total_bytes: u64,
conversations: usize,
transferred_bytes: u64,
deleted: usize,
deleted_bytes: u64,
scan_both: Duration,
delta_sim: Duration,
warm_session: Duration,
}
fn generate_tree_fixture(
root: &Path,
rng: &mut impl Rng,
dir_count: usize,
file_count: usize,
) -> io::Result<u64> {
fs::create_dir_all(root)?;
let mut total = 0u64;
let mut remaining = file_count;
for d in 0..dir_count {
let dir = root.join(format!("dir_{d:02}"));
fs::create_dir_all(&dir)?;
let per_dir = if d + 1 == dir_count {
0
} else if d + 2 == dir_count {
remaining } else {
remaining / (dir_count - 1 - d)
};
for f in 0..per_dir {
remaining -= 1;
let size = match rng.random_range(0u8..100) {
0..=5 => 0, 6..=15 => rng.random_range(20..64) * 1024 * 1024, _ => rng.random_range(1..4) * 1024 * 1024, };
let path = dir.join(format!("file_{f:03}.bin"));
write_random_file(&path, size, rng)?;
total += size;
}
}
Ok(total)
}
fn diverge_tree(
sender: &Path,
receiver: &Path,
rng: &mut impl Rng,
modified_every: usize,
new_every: usize,
stale_every: usize,
) -> io::Result<u64> {
let mut written = 0u64;
for d in 0..TREE_DIR_COUNT {
let dir = sender.join(format!("dir_{d:02}"));
let Ok(entries) = fs::read_dir(&dir) else {
continue;
};
for (i, e) in entries.flatten().enumerate() {
let name = e.file_name().to_string_lossy().into_owned();
let path = dir.join(&name);
if i % modified_every == 0 {
let size = fs::metadata(&path)?.len();
if size > 0 {
let mut f = fs::OpenOptions::new().write(true).open(&path)?;
let region = (size / 20).max(4096).min(size);
let start = size / 2;
let mut buf = vec![0u8; region as usize];
rng.fill_bytes(&mut buf);
f.seek(io::SeekFrom::Start(start))?;
f.write_all(&buf)?;
written += region;
}
}
if i % new_every == 0 {
let new_path = dir.join(format!("new_{i:03}.bin"));
let size = rng.random_range(1..4) * 1024 * 1024;
write_random_file(&new_path, size, rng)?;
written += size;
}
if i % stale_every == 0 {
let stale_path = receiver
.join(format!("dir_{d:02}"))
.join(format!("stale_{i:03}.bin"));
let size = 512 * 1024;
write_random_file(&stale_path, size, rng)?;
written += size;
}
}
}
Ok(written)
}
fn run_tree_simulation(cfg: &BenchConfig, pipeline: &PipelineConfig) -> io::Result<()> {
println!();
println!("==== tree sync simulation (D17) ====");
let cdc = cfg
.cdc
.unwrap_or_else(|| cdc_params_for_size(cfg.size_mib * 1024 * 1024));
let mut rng = match cfg.seed {
Some(seed) => StdRng::seed_from_u64(seed ^ 0x7EE5_1EE7),
None => StdRng::from_rng(&mut rand::rng()),
};
let sender = cfg.out_dir.join("tree_sender");
let receiver = cfg.out_dir.join("tree_receiver");
let _ = fs::remove_dir_all(&sender);
let _ = fs::remove_dir_all(&receiver);
let t = Instant::now();
let total = generate_tree_fixture(&sender, &mut rng, TREE_DIR_COUNT, TREE_FILE_COUNT)?;
let fixture_s = t.elapsed();
println!(
"fixture: {} files across {} dirs ({}), generated in {}",
TREE_FILE_COUNT,
TREE_DIR_COUNT,
fmt_bytes(total),
fmt_ms(fixture_s)
);
fs::create_dir_all(&receiver)?; let s1 = simulate_tree_session(&sender, &receiver, pipeline, cdc, false)?;
print_tree_stats("session 1 (receiver empty, everything transfers)", &s1);
let t = Instant::now();
copy_tree(&sender, &receiver)?;
let copy_s = t.elapsed();
println!(
"(session 1 commit simulated by copying the tree, {})",
fmt_ms(copy_s)
);
let t = Instant::now();
let diverged_bytes = diverge_tree(
&sender,
&receiver,
&mut rng,
TREE_MODIFIED_FRAC,
TREE_NEW_FRAC,
TREE_STALE_FRAC,
)?;
let diverge_s = t.elapsed();
println!(
"(divergence script wrote {} in {})",
fmt_bytes(diverged_bytes),
fmt_ms(diverge_s)
);
let s2 = simulate_tree_session(&sender, &receiver, pipeline, cdc, false)?;
print_tree_stats("session 2 (cold scan, diverged)", &s2);
copy_tree(&sender, &receiver)?;
let s3 = simulate_tree_session(&sender, &receiver, pipeline, cdc, true)?;
print_tree_stats("session 3 (warm, unchanged)", &s3);
let renamed_from = {
let dir = sender.join("dir_00");
let mut entries: Vec<_> = fs::read_dir(&dir)?
.flatten()
.map(|e| e.path())
.filter(|p| p.is_file())
.collect();
entries.sort();
entries[0].clone()
};
let renamed_size = fs::metadata(&renamed_from)?.len();
{
use ffsync::chunking::scan_file_ordered;
use ffsync::index::FileIndex;
let recv_old = receiver.join(renamed_from.strip_prefix(&sender).unwrap());
let (_stats, chunks) = scan_file_ordered(&recv_old, pipeline, cdc)?;
let idx = FileIndex::from_scan(
chunks,
renamed_size,
ffsync::index::stat_mtime(&recv_old).unwrap_or((0, 0)),
(cdc.0 as u32, cdc.1 as u32, cdc.2 as u32),
);
ffsync::index::write_index(&recv_old, &idx)?;
}
let renamed_to = renamed_from.parent().unwrap().join("renamed_d22.bin");
fs::rename(&renamed_from, &renamed_to)?;
let s4 = simulate_tree_session(&sender, &receiver, pipeline, cdc, false)?;
print_tree_stats("session 4 (one file renamed, D22 census)", &s4);
let rename_bytes = s4.transferred_bytes;
println!(
"rename cell: renamed {} ({} bytes); wire bytes for the rename: {} ({} without the census)",
renamed_from.file_name().unwrap().to_string_lossy(),
fmt_bytes(renamed_size),
fmt_bytes(rename_bytes),
fmt_bytes(renamed_size)
);
let cold_second = s2.scan_both + s2.delta_sim;
println!();
println!(
"conclusion: warm unchanged session is {:.1}x cheaper than a cold re-scan \
({} vs {}); diverged session transfers {} of {} ({:.1}%)",
cold_second.as_secs_f64() / s3.warm_session.as_secs_f64().max(1e-9),
fmt_ms(s3.warm_session),
fmt_ms(cold_second),
fmt_bytes(s2.transferred_bytes),
fmt_bytes(s2.total_bytes),
if s2.total_bytes == 0 {
0.0
} else {
s2.transferred_bytes as f64 / s2.total_bytes as f64 * 100.0
}
);
Ok(())
}
fn copy_tree(src: &Path, dst: &Path) -> io::Result<()> {
fs::create_dir_all(dst)?;
for e in fs::read_dir(src)? {
let e = e?;
let from = e.path();
let to = dst.join(e.file_name());
if from.is_dir() {
copy_tree(&from, &to)?;
} else {
fs::copy(&from, &to)?;
}
}
Ok(())
}
fn simulate_tree_session(
sender: &Path,
receiver: &Path,
pipeline: &PipelineConfig,
cdc: (usize, usize, usize),
warm: bool,
) -> io::Result<TreeStats> {
let t_scan = Instant::now();
let sender_manifest = ffsync::tree::TreeManifest::scan(sender)?;
let receiver_manifest = ffsync::tree::TreeManifest::scan(receiver)?;
let scan_both = t_scan.elapsed();
let mut all_paths: std::collections::BTreeSet<String> = Default::default();
for e in sender_manifest.entries() {
all_paths.insert(e.rel_path.clone());
}
for e in receiver_manifest.entries() {
all_paths.insert(e.rel_path.clone());
}
let mut conversations: Vec<String> = Vec::new();
let mut deletions: Vec<(String, u64)> = Vec::new();
for rel in &all_paths {
match ffsync::tree::triage(
rel,
sender_manifest.get(rel),
receiver_manifest.get(rel),
false,
) {
ffsync::tree::Triage::Skip => {}
ffsync::tree::Triage::Sync | ffsync::tree::Triage::Transfer => {
conversations.push(rel.clone());
}
ffsync::tree::Triage::CreateDir | ffsync::tree::Triage::Conflict { .. } => {}
ffsync::tree::Triage::Delete => {
if let Some(e) = receiver_manifest.get(rel) {
deletions.push((rel.clone(), e.size));
}
}
}
}
let t_delta = Instant::now();
let mut transferred = 0u64;
let mut total_bytes = 0u64;
let receiver_sizes: Vec<(std::path::PathBuf, u64)> = receiver_manifest
.entries()
.filter(|e| e.kind == ffsync::tree::EntryKind::File)
.map(|e| {
(
receiver.join(e.rel_path.replace('/', std::path::MAIN_SEPARATOR_STR)),
e.size,
)
})
.collect();
for rel in &conversations {
let src = sender.join(rel.replace('/', std::path::MAIN_SEPARATOR_STR));
let dst = receiver.join(rel.replace('/', std::path::MAIN_SEPARATOR_STR));
let (_, src_chunks) = scan_file(&src, pipeline, cdc)?;
if dst.exists() {
let (_, dst_chunks) = scan_file(&dst, pipeline, cdc)?;
let delta = compute_delta(&src_chunks, &dst_chunks);
transferred += delta.missing_bytes;
total_bytes += delta.target_bytes;
} else {
let src_size = fs::metadata(&src)?.len();
let mut census: std::collections::HashSet<blake3::Hash> =
std::collections::HashSet::new();
for (cand, size) in &receiver_sizes {
if *size != src_size || *cand == dst {
continue;
}
if let Some(idx) = ffsync::index::load_index(cand)
&& idx
.validate(cand, (cdc.0 as u32, cdc.1 as u32, cdc.2 as u32))
.is_some()
{
census.extend(idx.chunks.iter().map(|c| c.hash));
}
}
let (mut miss, mut tot) = (0u64, 0u64);
for chunk in &src_chunks {
tot += chunk.length as u64;
if !census.contains(&chunk.hash) {
miss += chunk.length as u64;
}
}
transferred += miss;
total_bytes += tot;
}
}
let delta_sim = t_delta.elapsed();
let warm_load_us = if warm {
let probe = sender_manifest
.entries()
.find(|e| e.kind == ffsync::tree::EntryKind::File)
.map(|e| sender.join(e.rel_path.replace('/', std::path::MAIN_SEPARATOR_STR)))
.unwrap_or_else(|| sender.join("dir_00/file_000.bin"));
if ffsync::index::load_index(&probe).is_none() {
let size = fs::metadata(&probe)?.len();
let (_, chunks) = scan_file(&probe, pipeline, cdc)?;
let index = ffsync::index::FileIndex::from_scan(
chunks,
size,
ffsync::index::stat_mtime(&probe).unwrap_or((0, 0)),
(cdc.0 as u32, cdc.1 as u32, cdc.2 as u32),
);
ffsync::index::write_index(&probe, &index)?;
}
let t = Instant::now();
let loaded = ffsync::index::load_index(&probe);
let us = t.elapsed().as_secs_f64() * 1e6;
if loaded.is_none() {
return Err(io::Error::other("warm index probe failed to load"));
}
us
} else {
0.0
};
let warm_session =
scan_both + Duration::from_secs_f64(warm_load_us * 1e-6 * conversations.len() as f64);
Ok(TreeStats {
files: sender_manifest.len(),
total_bytes,
conversations: conversations.len(),
transferred_bytes: if warm { 0 } else { transferred },
deleted: if warm { 0 } else { deletions.len() },
deleted_bytes: if warm {
0
} else {
deletions.iter().map(|(_, b)| b).sum()
},
scan_both,
delta_sim: if warm { Duration::ZERO } else { delta_sim },
warm_session,
})
}
struct LatencyEnd {
one_way: Duration,
rx: std::sync::mpsc::Receiver<Vec<u8>>,
tx: std::sync::mpsc::SyncSender<Vec<u8>>,
pending: Vec<u8>,
pending_pos: usize,
}
impl io::Read for LatencyEnd {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
if self.pending_pos >= self.pending.len() {
self.pending.clear();
self.pending_pos = 0;
match self.rx.recv() {
Ok(chunk) => {
self.pending = chunk;
while let Ok(more) = self.rx.try_recv() {
self.pending.extend_from_slice(&more);
}
if self.one_way > Duration::ZERO {
std::thread::sleep(self.one_way);
}
}
Err(_) => return Ok(0), }
}
let available = self.pending.len() - self.pending_pos;
let n = available.min(buf.len());
buf[..n].copy_from_slice(&self.pending[self.pending_pos..self.pending_pos + n]);
self.pending_pos += n;
Ok(n)
}
}
impl io::Write for LatencyEnd {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
match self.tx.send(buf.to_vec()) {
Ok(()) => Ok(buf.len()),
Err(_) => Err(io::Error::new(io::ErrorKind::BrokenPipe, "peer is gone")),
}
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
fn latency_pair(one_way: Duration) -> (LatencyEnd, LatencyEnd) {
let (a_tx, b_rx) = std::sync::mpsc::sync_channel::<Vec<u8>>(1024);
let (b_tx, a_rx) = std::sync::mpsc::sync_channel::<Vec<u8>>(1024);
(
LatencyEnd {
one_way,
rx: a_rx,
tx: a_tx,
pending: Vec::new(),
pending_pos: 0,
},
LatencyEnd {
one_way,
rx: b_rx,
tx: b_tx,
pending: Vec::new(),
pending_pos: 0,
},
)
}
fn run_one_latency_session(
sender_root: &Path,
receiver_root: &Path,
rtt: Duration,
parallel: u16,
) -> io::Result<Duration> {
use ffsync::protocol::tree_sync::{
ReadWrite, TreeReceiveRequest, TreeSendRequest, tree_receive, tree_send,
};
let mut pairs: Vec<(LatencyEnd, LatencyEnd)> = Vec::new();
for _ in 0..parallel {
pairs.push(latency_pair(rtt));
}
let send_root = sender_root.to_path_buf();
let recv_root = receiver_root.to_path_buf();
let mut sender_ends: Vec<LatencyEnd> = Vec::new();
let mut receiver_ends: Vec<LatencyEnd> = Vec::new();
for (s, r) in pairs {
sender_ends.push(s);
receiver_ends.push(r);
}
let receiver_ends = std::sync::Mutex::new(receiver_ends);
let receiver_ends_for_closure = std::sync::Arc::new(receiver_ends);
let recv_root2 = recv_root;
let ends2 = std::sync::Arc::clone(&receiver_ends_for_closure);
let recv_thread = std::thread::spawn(move || -> Result<
ffsync::protocol::tree_sync::TreeOutcome,
ffsync::protocol::sync::SyncError,
> {
let receiver_ends = ends2;
let mut control_end = receiver_ends.lock().unwrap().remove(0);
tree_receive(
&mut control_end,
&TreeReceiveRequest {
root: recv_root2,
force_checksum: false,
no_delete: false,
excludes: Vec::new(),
accept_worker: {
let ends = std::sync::Arc::clone(&receiver_ends);
std::sync::Arc::new(move || {
Ok(Box::new(ends.lock().unwrap().remove(0)) as Box<dyn ReadWrite>)
})
},
dry_run: false,
eager: false,
},
&PipelineConfig::default(),
&mut |_p| {},
)
});
let t = Instant::now();
let send_res = tree_send(
&mut sender_ends.remove(0),
&TreeSendRequest {
root: send_root,
force_checksum: false,
no_delete: false,
excludes: Vec::new(),
parallel,
dry_run: false,
eager: false,
connect_worker: {
let ends = std::sync::Mutex::new(sender_ends);
std::sync::Arc::new(move |_| {
Ok(Box::new(ends.lock().unwrap().remove(0)) as Box<dyn ReadWrite>)
})
},
},
&PipelineConfig::default(),
&mut |_p| {},
);
let elapsed = t.elapsed();
let recv_res = recv_thread.join().unwrap();
if let Err(e) = send_res {
return Err(io::Error::other(format!("sender failed: {e}")));
}
if let Err(e) = recv_res {
return Err(io::Error::other(format!("receiver failed: {e}")));
}
Ok(elapsed)
}
fn run_latency_simulation(cfg: &BenchConfig, _pipeline: &PipelineConfig) -> io::Result<()> {
println!();
println!("==== latency simulation (§16.8): sequential vs parallel conversations ====");
let mut rng = match cfg.seed {
Some(seed) => StdRng::seed_from_u64(seed ^ 0x1CE_5EED),
None => StdRng::from_rng(&mut rand::rng()),
};
let sender = cfg.out_dir.join("lat_sender");
let receiver = cfg.out_dir.join("lat_receiver");
let _ = fs::remove_dir_all(&sender);
let _ = fs::remove_dir_all(&receiver);
let total = generate_tree_fixture(&sender, &mut rng, 8, 100)?;
fs::create_dir_all(&receiver)?;
println!(
"fixture: 100 files ({}), receiver starts empty, simulated RTT = {} ms",
fmt_bytes(total),
cfg.rtt_ms
);
println!(" (model: each direction delayed by RTT/2, reads coalesce queued data;");
println!(" session wall time ≈ scan+I/O + conversations × round trips × RTT)");
let counts: Vec<u16> = if cfg.parallel_counts.is_empty() {
vec![1, 4, 8]
} else {
cfg.parallel_counts.clone()
};
println!();
println!(
" {:>10} {:>14} {:>16}",
"parallel", "wall time", "vs sequential"
);
let mut baseline: Option<Duration> = None;
for k in counts {
let _ = fs::remove_dir_all(&receiver);
fs::create_dir_all(&receiver)?;
let elapsed =
run_one_latency_session(&sender, &receiver, Duration::from_millis(cfg.rtt_ms), k)?;
let rel = match baseline {
None => {
baseline = Some(elapsed);
"baseline".to_string()
}
Some(b) => format!("{:.2}x", b.as_secs_f64() / elapsed.as_secs_f64()),
};
println!(" {:>10} {:>14} {:>16}", k, fmt_ms(elapsed), rel);
}
Ok(())
}
fn run_eager_simulation(cfg: &BenchConfig, pipeline: &PipelineConfig) -> io::Result<()> {
println!();
println!("==== eager diff simulation (D8/§9.2): lazy vs eager ====");
let source = cfg.out_dir.join("source_A.bin");
let variant = cfg.out_dir.join("variant_modified.bin");
if !source.exists() || !variant.exists() {
return Err(io::Error::other(
"eager simulation needs the fixtures (run without --eager once, or check --out-dir)",
));
}
let cdc = cdc_params_for_size(fs::metadata(&variant)?.len());
let (_, source_chunks) = scan_file(&source, pipeline, cdc)?;
let (_, variant_chunks) = scan_file(&variant, pipeline, cdc)?;
let delta = compute_delta(&source_chunks, &variant_chunks);
let minimum = delta.missing_bytes;
println!(
"fixture: modified variant, {} planned minimum wire bytes, RTT = {} ms",
fmt_bytes(minimum),
cfg.rtt_ms
);
println!();
println!(
" {:>8} {:>14} {:>12} {:>14}",
"mode", "wire bytes", "over-send", "wall time"
);
for (label, eager) in [("lazy", false), ("eager", true)] {
let (elapsed, wire) = run_one_eager_session(&source, &variant, cfg.rtt_ms, eager)?;
let over = wire.saturating_sub(minimum);
println!(
" {:>8} {:>14} {:>12} {:>14}",
label,
fmt_bytes(wire),
fmt_bytes(over),
fmt_ms(elapsed)
);
}
Ok(())
}
fn run_one_eager_session(
source: &Path,
variant: &Path,
rtt_ms: u64,
eager: bool,
) -> io::Result<(Duration, u64)> {
use ffsync::protocol::sync::{ReceiveRequest, SendRequest, receive, send};
let target = variant.with_file_name(format!("eager_target_{}.bin", usize::from(eager)));
let _ = fs::remove_file(&target);
fs::copy(source, &target)?;
let (mut s_end, mut r_end) = latency_pair(Duration::from_millis(rtt_ms) / 2);
let variant_p = variant.to_path_buf();
let target_r = target.clone();
let target_cleanup = target.clone();
let recv_thread = std::thread::spawn(
move || -> Result<SyncOutcome, ffsync::protocol::sync::SyncError> {
let request = ReceiveRequest {
path: target_r.clone(),
local_source: target_r.clone(),
force_checksum: false,
tree: false,
eager,
apply_mode: None,
apply_mtime: None,
extra_sources: Vec::new(),
};
receive(
&mut r_end,
&request,
&PipelineConfig::default(),
&mut |_p| {},
)
},
);
let t = Instant::now();
let send_req = SendRequest {
path: variant_p,
display_path: "eager_sim.bin".into(),
mtime: None,
force_checksum: false,
tree: false,
eager,
dry_run: false,
};
let send_res = send(
&mut s_end,
&send_req,
&PipelineConfig::default(),
&mut |_p| {},
)
.map_err(|e| io::Error::other(e.to_string()))?;
let elapsed = t.elapsed();
let bytes = match send_res {
SyncOutcome::Transferred { bytes, .. } => bytes,
SyncOutcome::InSync => 0,
SyncOutcome::WouldChange => 0,
};
recv_thread
.join()
.unwrap()
.map_err(|e| io::Error::other(e.to_string()))?;
let _ = fs::remove_file(&target_cleanup);
Ok((elapsed, bytes))
}
fn print_tree_stats(label: &str, s: &TreeStats) {
println!();
println!("---- {} ----", label);
println!(
" {:<22} {:>9} ({} files, {} total)",
"tree scan (both):",
fmt_ms(s.scan_both),
s.files,
fmt_bytes(s.total_bytes)
);
println!(
" {:<22} {:>9} ({} conversations)",
"delta sim:",
fmt_ms(s.delta_sim),
s.conversations
);
println!(
" {:<22} {:>9} ({} deleted, {})",
"warm session model:",
fmt_ms(s.warm_session),
s.deleted,
fmt_bytes(s.deleted_bytes)
);
}
fn reconstruct(
variant: &Path,
source: &Path,
rec_path: &Path,
source_index: &HashMap<Hash, u64>,
target_chunks: &[Chunk],
variant_hash: Hash,
) -> io::Result<Duration> {
let _ = fs::remove_file(rec_path);
let variant_size = fs::metadata(variant)?.len();
let start = Instant::now();
let reconstructor = Reconstructor::new(rec_path, source, variant_size)?;
let variant_file = File::open(variant)?;
let mut local_scratch: Vec<u8> = Vec::new();
let mut net_scratch: Vec<u8> = Vec::new();
for chunk in target_chunks {
if let Some(&source_offset) = source_index.get(&chunk.hash) {
let task = ReconstructionTask {
target_offset: chunk.offset,
source: ChunkSource {
offset: source_offset,
length: chunk.length,
},
};
reconstructor.copy_local_chunk(&task, &mut local_scratch)?;
} else {
net_scratch.resize(chunk.length, 0);
let read = variant_file.read_at(&mut net_scratch[..chunk.length], chunk.offset)?;
if read != chunk.length {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
format!("short read from variant at offset {}", chunk.offset),
));
}
let task = ReconstructionTask {
target_offset: chunk.offset,
source: ChunkSource {
offset: chunk.offset,
length: chunk.length,
},
};
reconstructor.write_chunk(&task, &net_scratch[..chunk.length])?;
}
}
reconstructor.finalize(&Blake3Hasher, variant_hash, rec_path)?;
Ok(start.elapsed())
}
fn run_scenario(
scenario: Scenario,
source: &Path,
variant: &Path,
pipeline: &PipelineConfig,
cdc: (usize, usize, usize),
) -> io::Result<ScenarioResult> {
let (source_scan, source_chunks) = scan_file(source, pipeline, cdc)?;
let (target_scan, target_chunks) = scan_file(variant, pipeline, cdc)?;
let delta = compute_delta(&source_chunks, &target_chunks);
let source_index: HashMap<Hash, u64> =
source_chunks.iter().map(|c| (c.hash, c.offset)).collect();
let variant_hash = file_root_hash(variant)?;
let rec_path = variant.with_file_name(format!("reconstructed_{}.bin", scenario.name()));
let reconstruct = reconstruct(
variant,
source,
&rec_path,
&source_index,
&target_chunks,
variant_hash,
)?;
fs::remove_file(&rec_path).ok();
Ok(ScenarioResult {
scenario,
variant_size: fs::metadata(variant)?.len(),
source_scan,
target_scan,
delta,
reconstruct,
})
}
fn print_scenario_report(result: &ScenarioResult, target: &Path) {
let first = |d: &Option<Duration>| match d {
Some(d) => fmt_ms(*d),
None => "n/a".to_string(),
};
let src_label = "scan source (A):";
let tgt_label = format!("scan target ({}):", result.scenario.name());
println!();
println!(
"---- scenario: {} (target: {}) ----",
result.scenario.name(),
target.display()
);
println!(
" {:<24} {:>9} (first chunk {:>9}, {:>4} chunks, {:>9} hashed)",
src_label,
fmt_ms(result.source_scan.total),
first(&result.source_scan.first_chunk),
result.source_scan.chunks,
fmt_bytes(result.source_scan.hashed_bytes),
);
println!(
" {:<24} {:>9} (first chunk {:>9}, {:>4} chunks, {:>9} hashed)",
tgt_label,
fmt_ms(result.target_scan.total),
first(&result.target_scan.first_chunk),
result.target_scan.chunks,
fmt_bytes(result.target_scan.hashed_bytes),
);
let dedup = if result.delta.missing_bytes == 0 {
"inf".to_string()
} else {
format!(
"{:.1}x",
result.delta.target_bytes as f64 / result.delta.missing_bytes as f64
)
};
println!(
" delta: {:>9} missing of {:>9} ({:.2}%), dedup {}",
fmt_bytes(result.delta.missing_bytes),
fmt_bytes(result.delta.target_bytes),
result.delta.pct(),
dedup,
);
println!(
" reconstruction: {:>9} (final hash verified)",
fmt_ms(result.reconstruct),
);
print_projection_table(&result.delta, result.scan_max());
}
impl ScenarioResult {
fn scan_max(&self) -> Duration {
self.source_scan.total.max(self.target_scan.total)
}
}
fn print_projection_table(delta: &DeltaStats, scan_max: Duration) {
println!(
" projection (transfer of {} missing; e2e = max(scans) + transfer):",
fmt_bytes(delta.missing_bytes)
);
println!(
" {:<10} {:>9} {:>9} {:>9} {:>9} {:>9} {:>9}",
"link", "1x xfer", "1x e2e", "2x xfer", "2x e2e", "4x xfer", "4x e2e"
);
for &(mbits, label) in &LINKS {
let bytes_per_sec = mbits * 1_000_000.0 / 8.0;
let mut cells = String::new();
for &ratio in &RATIOS {
let transfer = delta.missing_bytes as f64 / (bytes_per_sec * ratio);
let e2e = transfer + scan_max.as_secs_f64();
cells.push_str(&format!(
" {:>9} {:>9}",
fmt_secs(transfer),
fmt_secs(e2e)
));
}
println!(" {:<10}{}", label, cells);
}
}
fn print_sweep_conclusion(all_results: &[Vec<ScenarioResult>]) {
println!();
println!("==== sweep conclusion (delta % per scenario) ====");
println!(
"{:<22} {:>10} {:>10} {:>10} {:>10} {:>10}",
"cdc avg/min/max", "identical", "modified", "appended", "fresh", "src scan"
);
for (point, results) in SWEEP_POINTS.iter().zip(all_results) {
let get = |name: &str| -> &ScenarioResult {
results
.iter()
.find(|r| r.scenario.name() == name)
.expect("scenario in sweep results")
};
let scan = get("modified").source_scan.total;
println!(
"{:<22} {:>10.2} {:>10.2} {:>10.2} {:>10.2} {:>10}",
format!(
"{}/{}/{}",
fmt_bytes(point.0 as u64),
fmt_bytes(point.1 as u64),
fmt_bytes(point.2 as u64)
),
get("identical").delta.pct(),
get("modified").delta.pct(),
get("appended").delta.pct(),
get("fresh").delta.pct(),
fmt_ms(scan),
);
}
println!();
println!(
"Reading: lower delta % on modified/appended = fewer bytes on the wire;\
\nsmaller chunks cost scan throughput and produce more chunks (plan frames).\
\nPick the smallest avg size whose scan cost is acceptable for the target link."
);
}
fn print_summary(results: &[ScenarioResult]) {
println!();
println!("==== summary ====");
println!(
"{:<10} {:>9} {:>9} {:>9} {:>9} {:>8} {:>9} {:>9} {:>8}",
"scenario",
"src scan",
"tgt scan",
"missing",
"delta %",
"dedup",
"recon",
"e2e 1x",
"verified"
);
for r in results {
let dedup = if r.delta.missing_bytes == 0 {
"inf".to_string()
} else {
format!(
"{:.1}x",
r.delta.target_bytes as f64 / r.delta.missing_bytes as f64
)
};
let e2e = 1_000_000_000.0 / 8.0;
let e2e_1x = r.scan_max().as_secs_f64() + r.delta.missing_bytes as f64 / e2e;
println!(
"{:<10} {:>9} {:>9} {:>9} {:>8.2}% {:>8} {:>9} {:>9} {:>8}",
r.scenario.name(),
fmt_ms(r.source_scan.total),
fmt_ms(r.target_scan.total),
fmt_bytes(r.delta.missing_bytes),
r.delta.pct(),
dedup,
fmt_ms(r.reconstruct),
fmt_secs(e2e_1x),
"yes",
);
}
}
fn probe_rsync() -> RsyncProbe {
let native = Command::new("rsync").arg("--version").output();
match native {
Ok(out) if out.status.success() => return RsyncProbe::Available(RsyncMode::Native),
_ => {}
}
match wsl_command().arg("/bin/true").output() {
Err(_) => {
return RsyncProbe::Unavailable(
"rsync not found on PATH and WSL (wsl.exe) is unavailable \
(this comparison can be run on Linux, or on Windows inside WSL)"
.to_string(),
);
}
Ok(out) if !out.status.success() => {
return RsyncProbe::Unavailable(
"rsync not found on PATH and WSL has no usable distro \
(install one with `wsl.exe --install`, then install rsync inside it)"
.to_string(),
);
}
Ok(_) => {}
}
if let Ok(out) = wsl_command().args(["rsync", "--version"]).output() {
if out.status.success() {
return RsyncProbe::Available(RsyncMode::Wsl);
}
return RsyncProbe::Unavailable(
"rsync is not installed inside the WSL distro \
(install it there, e.g. `wsl.exe sudo apt install rsync`)"
.to_string(),
);
}
RsyncProbe::Unavailable(
"rsync not found on PATH and it could not be probed inside WSL \
(this comparison can be run on Linux, or on Windows inside WSL)"
.to_string(),
)
}
fn wsl_command() -> Command {
let mut cmd = Command::new("wsl.exe");
cmd.arg("--exec");
cmd
}
fn to_wsl_path(path: &Path) -> io::Result<String> {
let canonical = match fs::canonicalize(path) {
Ok(p) => p,
Err(e) if e.kind() == io::ErrorKind::NotFound => {
let parent = path.parent().ok_or_else(|| {
io::Error::other(format!("{} has no parent directory", path.display()))
})?;
let file_name = path
.file_name()
.ok_or_else(|| io::Error::other(format!("{} has no file name", path.display())))?;
fs::canonicalize(parent)?.join(file_name)
}
Err(e) => {
return Err(io::Error::new(
e.kind(),
format!("cannot canonicalize {} for WSL: {}", path.display(), e),
));
}
};
let windows = canonical.to_string_lossy().replace('\\', "/");
let windows = windows
.strip_prefix("//?/")
.map(|rest| rest.to_string())
.unwrap_or(windows);
let mut cmd = wsl_command();
cmd.args(["wslpath", "-a", "-u", &windows]);
if let Ok(out) = cmd.output()
&& out.status.success()
{
let converted = String::from_utf8_lossy(&out.stdout).trim().to_string();
if converted.starts_with('/') {
return Ok(converted);
}
}
let bytes = windows.as_bytes();
if bytes.len() >= 2 && bytes[1] == b':' {
let mut manual = String::from("/mnt/");
manual.push(bytes[0].to_ascii_lowercase() as char);
manual.push_str(&windows[2..]);
return Ok(manual);
}
Err(io::Error::other(format!(
"wslpath failed and {:?} is not a drive-letter path",
windows
)))
}
fn run_rsync_once(
mode: RsyncMode,
mode_name: &str,
extra_args: &[&str],
src: &str,
dst: &str,
) -> io::Result<(String, Duration)> {
let mut cmd = match mode {
RsyncMode::Native => {
let mut cmd = Command::new("rsync");
cmd.args(["-a", "--stats"]).args(extra_args);
cmd
}
RsyncMode::Wsl => {
let mut cmd = wsl_command();
cmd.args(["rsync", "-a", "--stats"]).args(extra_args);
cmd
}
};
cmd.arg(src).arg(dst);
let start = Instant::now();
let out = cmd.output()?;
let wall = start.elapsed();
if !out.status.success() {
return Err(io::Error::other(format!(
"rsync ({}) failed: {}",
mode_name,
String::from_utf8_lossy(&out.stderr).trim()
)));
}
Ok((String::from_utf8_lossy(&out.stdout).to_string(), wall))
}
fn run_rsync_comparison(
cfg: &BenchConfig,
source: &Path,
variants: &[(Scenario, PathBuf)],
) -> io::Result<RsyncResults> {
let probe = probe_rsync();
let mode = match probe {
RsyncProbe::Available(mode) => mode,
RsyncProbe::Unavailable(reason) => {
println!();
println!("rsync comparison skipped: {}.", reason);
return Ok(RsyncResults {
mode: None,
rows: Vec::new(),
});
}
};
let variant = &variants
.iter()
.find(|(s, _)| *s == Scenario::Modified)
.expect("modified variant fixture")
.1;
let src_dir = cfg.out_dir.join("rsync_src");
let dst_dir = cfg.out_dir.join("rsync_dst");
fs::create_dir_all(&src_dir)?;
fs::create_dir_all(&dst_dir)?;
let src_file = src_dir.join("A.bin");
let dst_file = dst_dir.join("A.bin");
fs::copy(source, &src_file)?;
let (src_arg, dst_arg) = match mode {
RsyncMode::Native => (
src_file.to_string_lossy().to_string(),
dst_file.to_string_lossy().to_string(),
),
RsyncMode::Wsl => (to_wsl_path(&src_file)?, to_wsl_path(&dst_file)?),
};
let modes: [(&str, &[&str]); 4] = [
("default", &[]),
("whole-file", &["--whole-file"]),
("checksum", &["--checksum"]),
("delta (--no-W)", &["--no-whole-file", "--no-implied-dirs"]),
];
let mut rows = Vec::new();
for (mode_name, extra_args) in modes {
fs::copy(variant, &dst_file)?;
let src_mtime = fs::metadata(&src_file)?.modified()?;
let older = src_mtime
.checked_sub(Duration::from_secs(2))
.unwrap_or(SystemTime::UNIX_EPOCH);
let times = FileTimes::new().set_accessed(older).set_modified(older);
let dst = OpenOptions::new().write(true).open(&dst_file)?;
dst.set_times(times).map_err(|e| {
io::Error::new(
e.kind(),
format!("cannot backdate mtime of {}: {}", dst_file.display(), e),
)
})?;
let (stdout, wall) = run_rsync_once(mode, mode_name, extra_args, &src_arg, &dst_arg)?;
let mut transferred = None;
let mut matched = None;
for line in stdout.lines() {
let parse = |prefix: &str| -> Option<u64> {
let rest = line.strip_prefix(prefix)?;
rest.trim()
.trim_end_matches("bytes")
.trim()
.replace(',', "")
.parse::<u64>()
.ok()
};
if transferred.is_none() {
transferred = parse("Literal data:");
}
if matched.is_none() {
matched = parse("Matched data:");
}
}
rows.push(RsyncRow {
mode: mode_name,
wall,
transferred,
matched,
});
}
Ok(RsyncResults {
mode: Some(mode),
rows,
})
}
fn print_rsync_report(rsync: &RsyncResults) {
let mode = match rsync.mode {
Some(mode) => mode,
None => return, };
println!();
println!(
"==== rsync comparison (scenario: modified, via {}) ====",
mode.label()
);
println!(
"{:<16} {:>10} {:>14} {:>14}",
"mode", "wall", "transferred", "matched"
);
for row in &rsync.rows {
let transferred = match row.transferred {
Some(bytes) => fmt_bytes(bytes),
None => "n/a".to_string(),
};
let matched = match row.matched {
Some(bytes) => fmt_bytes(bytes),
None => "n/a".to_string(),
};
println!(
"{:<16} {:>10} {:>14} {:>14}",
row.mode,
fmt_ms(row.wall),
transferred,
matched
);
}
}
fn emit_json(cfg: &BenchConfig, results: &[ScenarioResult], rsync: Option<&RsyncResults>) {
let mut out = String::from("{\"bench\":\"ffsync\"");
out.push_str(&format!(",\"size_mib\":{}", cfg.size_mib));
out.push_str(",\"scenarios\":[");
for (i, r) in results.iter().enumerate() {
if i > 0 {
out.push(',');
}
let dedup = if r.delta.missing_bytes == 0 {
"null".to_string()
} else {
format!(
"{:.3}",
r.delta.target_bytes as f64 / r.delta.missing_bytes as f64
)
};
out.push_str(&format!(
"{{\"variant\":\"{name}\",\"target_bytes\":{tb},\
\"source_scan_ms\":{ss:.3},\"target_scan_ms\":{ts:.3},\
\"source_first_chunk_ms\":{sf},\"target_first_chunk_ms\":{tf},\
\"source_chunks\":{sc},\"target_chunks\":{tc},\
\"hashed_bytes\":{hb},\"missing_bytes\":{mb},\
\"delta_pct\":{dp:.4},\"dedup_ratio\":{dd},\
\"reconstruct_ms\":{rc:.3},\"verified\":true}}",
name = r.scenario.name(),
tb = r.variant_size,
ss = ms(r.source_scan.total),
ts = ms(r.target_scan.total),
sf = json_ms(r.source_scan.first_chunk),
tf = json_ms(r.target_scan.first_chunk),
sc = r.source_scan.chunks,
tc = r.target_scan.chunks,
hb = r.source_scan.hashed_bytes,
mb = r.delta.missing_bytes,
dp = r.delta.pct(),
dd = dedup,
rc = ms(r.reconstruct),
));
}
out.push_str("],\"rsync\":");
match rsync {
None => out.push_str("null"),
Some(r) if r.mode.is_none() => out.push_str("{\"available\":false}"),
Some(r) => {
out.push_str(&format!(
"{{\"available\":true,\"via\":\"{}\",\"rows\":[",
r.mode.unwrap().label()
));
for (i, row) in r.rows.iter().enumerate() {
if i > 0 {
out.push(',');
}
let transferred = match row.transferred {
Some(bytes) => bytes.to_string(),
None => "null".to_string(),
};
out.push_str(&format!(
"{{\"mode\":\"{}\",\"wall_ms\":{:.3},\"transferred_bytes\":{}}}",
row.mode,
ms(row.wall),
transferred
));
}
out.push_str("]}");
}
}
out.push('}');
println!("{}", out);
}
fn ms(d: Duration) -> f64 {
d.as_secs_f64() * 1000.0
}
fn json_ms(d: Option<Duration>) -> String {
match d {
Some(d) => format!("{:.3}", ms(d)),
None => "null".to_string(),
}
}
fn fmt_bytes(bytes: u64) -> String {
let b = bytes as f64;
const KIB: f64 = 1024.0;
const MIB: f64 = 1024.0 * 1024.0;
const GIB: f64 = 1024.0 * 1024.0 * 1024.0;
if b >= GIB {
format!("{:.2} GiB", b / GIB)
} else if b >= MIB {
format!("{:.2} MiB", b / MIB)
} else if b >= KIB {
format!("{:.2} KiB", b / KIB)
} else {
format!("{} B", bytes)
}
}
fn fmt_ms(d: Duration) -> String {
let s = d.as_secs_f64();
if s < 1.0 {
format!("{:.1} ms", s * 1000.0)
} else {
format!("{:.2} s", s)
}
}
fn fmt_secs(secs: f64) -> String {
if secs < 1.0 {
format!("{:.1} ms", secs * 1000.0)
} else {
format!("{:.2} s", secs)
}
}
fn main() -> io::Result<()> {
let cfg = parse_args();
println!(
"ffsync bench: size = {} MiB, out-dir = {}, rsync = {}, json = {}",
cfg.size_mib,
cfg.out_dir.display(),
cfg.rsync,
cfg.json
);
if cfg.gen_tree {
let mut rng = match cfg.seed {
Some(seed) => StdRng::seed_from_u64(seed ^ 0x1CE_5EED),
None => StdRng::from_rng(&mut rand::rng()),
};
let sender = cfg.out_dir.join("lat_sender");
let _ = fs::remove_dir_all(&sender);
let total = generate_tree_fixture(&sender, &mut rng, 8, 100)?;
println!(
"fixture: 100 files ({}) in {} + empty dir, seed {}",
fmt_bytes(total),
sender.display(),
cfg.seed.unwrap_or(0)
);
return Ok(());
}
let (source, variants) = generate_fixtures(&cfg)?;
let pipeline = PipelineConfig::default();
let cdc = cfg
.cdc
.unwrap_or_else(|| cdc_params_for_size(cfg.size_mib * 1024 * 1024));
if cfg.sweep {
let mut all_results: Vec<Vec<ScenarioResult>> = Vec::new();
for cdc_point in SWEEP_POINTS {
println!(
"\n==== cdc min/avg/max = {}/{}/{} ====",
fmt_bytes(cdc_point.0 as u64),
fmt_bytes(cdc_point.1 as u64),
fmt_bytes(cdc_point.2 as u64)
);
let mut results = Vec::new();
for (scenario, path) in &variants {
let result = run_scenario(*scenario, &source, path, &pipeline, cdc_point)?;
print_scenario_report(&result, path);
results.push(result);
}
print_summary(&results);
all_results.push(results);
}
print_sweep_conclusion(&all_results);
return Ok(());
}
let mut results = Vec::new();
for (scenario, path) in &variants {
let result = run_scenario(*scenario, &source, path, &pipeline, cdc)?;
print_scenario_report(&result, path);
results.push(result);
}
print_summary(&results);
if cfg.disconnect {
let modified = &variants
.iter()
.find(|(s, _)| *s == Scenario::Modified)
.expect("modified variant")
.1;
let (_, variant_chunks) = scan_file(modified, &pipeline, cdc)?;
let (_, source_chunks) = scan_file(&source, &pipeline, cdc)?;
run_disconnect_simulation(&source, modified, &source_chunks, &variant_chunks, cdc)?;
}
if cfg.warm {
run_warm_simulation(&source, &pipeline, cdc)?;
}
if cfg.tree {
run_tree_simulation(&cfg, &pipeline)?;
}
if cfg.latency {
run_latency_simulation(&cfg, &pipeline)?;
}
if cfg.eager {
run_eager_simulation(&cfg, &pipeline)?;
return Ok(());
}
let rsync = if cfg.rsync {
Some(run_rsync_comparison(&cfg, &source, &variants)?)
} else {
None
};
if let Some(r) = &rsync {
print_rsync_report(r);
}
if cfg.json {
emit_json(&cfg, &results, rsync.as_ref());
}
Ok(())
}