DioCrafts
812 lines
26 KiB
Rust
812 lines
26 KiB
Rust
//! Independent A/B for `DedupService::verify_integrity` phase 1.
|
|
//!
|
|
//! This deliberately does not import the OxiCloud crate. It models the exact
|
|
//! manifest validation/messages and a backend whose `blob_size` operation has
|
|
//! either local-filesystem scheduling latency or remote request latency.
|
|
|
|
use foldhash::quality::RandomState;
|
|
use futures::stream::{self, StreamExt};
|
|
use std::collections::HashMap;
|
|
use std::hint::black_box;
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
|
use std::time::{Duration, Instant};
|
|
|
|
const CONCURRENCY: usize = 16;
|
|
const SERIAL_FAST_PATH_OCCURRENCES: usize = 4;
|
|
type Manifest = (String, Vec<String>, Vec<i64>, i64);
|
|
type SizeMap = HashMap<String, Option<u64>, RandomState>;
|
|
|
|
#[inline]
|
|
fn uses_serial_fast_path(manifests: &[Manifest]) -> bool {
|
|
if manifests.len() == 1 {
|
|
let (_, hashes, sizes, _) = &manifests[0];
|
|
return hashes.len() != sizes.len() || hashes.len() <= SERIAL_FAST_PATH_OCCURRENCES;
|
|
}
|
|
let mut occurrences = 0usize;
|
|
for (_, hashes, sizes, _) in manifests {
|
|
if hashes.len() == sizes.len() {
|
|
occurrences = occurrences.saturating_add(hashes.len());
|
|
if occurrences > SERIAL_FAST_PATH_OCCURRENCES {
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
true
|
|
}
|
|
|
|
#[derive(Clone, Copy)]
|
|
enum Latency {
|
|
/// No delay: used only by the separate-process peak-RSS probe.
|
|
Immediate,
|
|
/// Warm/cached local metadata latency as observed by an async caller.
|
|
Local { metadata: Duration, hash: Duration },
|
|
/// Object-store HEAD request: asynchronously wait for network latency.
|
|
Remote(Duration),
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct SimBackend {
|
|
latency: Latency,
|
|
calls: Arc<AtomicUsize>,
|
|
}
|
|
|
|
impl SimBackend {
|
|
fn new(latency: Latency) -> Self {
|
|
Self {
|
|
latency,
|
|
calls: Arc::new(AtomicUsize::new(0)),
|
|
}
|
|
}
|
|
|
|
async fn blob_size(&self, hash: &str) -> Option<u64> {
|
|
self.calls.fetch_add(1, Ordering::Relaxed);
|
|
match self.latency {
|
|
Latency::Immediate => {}
|
|
Latency::Local { metadata, .. } => tokio::time::sleep(metadata).await,
|
|
Latency::Remote(delay) => tokio::time::sleep(delay).await,
|
|
}
|
|
if hash.starts_with("missing-") {
|
|
None
|
|
} else if hash.starts_with("wrong-") {
|
|
Some(999)
|
|
} else {
|
|
Some(256)
|
|
}
|
|
}
|
|
|
|
async fn hash_local_blob(&self) {
|
|
if let Latency::Local { hash, .. } = self.latency {
|
|
// Equal phase-2 work: model mmap/BLAKE3 verification separately
|
|
// from metadata. The exact value only dilutes the phase-1 win; it
|
|
// does not differ between current and candidate.
|
|
tokio::time::sleep(hash).await;
|
|
}
|
|
}
|
|
|
|
fn calls(&self) -> usize {
|
|
self.calls.load(Ordering::Relaxed)
|
|
}
|
|
}
|
|
|
|
fn label(value: &str) -> &str {
|
|
&value[..value.len().min(12)]
|
|
}
|
|
|
|
async fn current(manifests: &[Manifest], backend: &SimBackend) -> Vec<String> {
|
|
let mut issues = Vec::new();
|
|
for (file_hash, chunk_hashes, chunk_sizes, total_size) in manifests {
|
|
let file_label = label(file_hash);
|
|
if chunk_hashes.len() != chunk_sizes.len() {
|
|
issues.push(format!(
|
|
"Manifest {file_label}: chunk_hashes/chunk_sizes length mismatch"
|
|
));
|
|
continue;
|
|
}
|
|
let sum: i64 = chunk_sizes.iter().sum();
|
|
if sum != *total_size {
|
|
issues.push(format!(
|
|
"Manifest {file_label}: total_size {total_size} != sum of chunk_sizes {sum}"
|
|
));
|
|
}
|
|
for (index, chunk_hash) in chunk_hashes.iter().enumerate() {
|
|
let chunk_label = label(chunk_hash);
|
|
match backend.blob_size(chunk_hash).await {
|
|
Some(actual_size) if actual_size != chunk_sizes[index] as u64 => {
|
|
issues.push(format!(
|
|
"Manifest {file_label} chunk {chunk_label}: size mismatch \
|
|
(expected {}, actual {actual_size})",
|
|
chunk_sizes[index]
|
|
));
|
|
}
|
|
None => issues.push(format!(
|
|
"Manifest {file_label} chunk {chunk_label}: missing in backend"
|
|
)),
|
|
Some(_) => {}
|
|
}
|
|
}
|
|
}
|
|
issues
|
|
}
|
|
|
|
fn replay(manifests: &[Manifest], sizes: &SizeMap) -> Vec<String> {
|
|
let mut issues = Vec::new();
|
|
for (file_hash, chunk_hashes, chunk_sizes, total_size) in manifests {
|
|
let file_label = label(file_hash);
|
|
if chunk_hashes.len() != chunk_sizes.len() {
|
|
issues.push(format!(
|
|
"Manifest {file_label}: chunk_hashes/chunk_sizes length mismatch"
|
|
));
|
|
continue;
|
|
}
|
|
let sum: i64 = chunk_sizes.iter().sum();
|
|
if sum != *total_size {
|
|
issues.push(format!(
|
|
"Manifest {file_label}: total_size {total_size} != sum of chunk_sizes {sum}"
|
|
));
|
|
}
|
|
for (index, chunk_hash) in chunk_hashes.iter().enumerate() {
|
|
let chunk_label = label(chunk_hash);
|
|
match sizes.get(chunk_hash.as_str()).copied().flatten() {
|
|
Some(actual_size) if actual_size != chunk_sizes[index] as u64 => {
|
|
issues.push(format!(
|
|
"Manifest {file_label} chunk {chunk_label}: size mismatch \
|
|
(expected {}, actual {actual_size})",
|
|
chunk_sizes[index]
|
|
));
|
|
}
|
|
None => issues.push(format!(
|
|
"Manifest {file_label} chunk {chunk_label}: missing in backend"
|
|
)),
|
|
Some(_) => {}
|
|
}
|
|
}
|
|
}
|
|
issues
|
|
}
|
|
|
|
const CANDIDATE_BATCH_OCCURRENCES: usize = 256;
|
|
|
|
async fn fill_sizes(size_by_hash: SizeMap, backend: &SimBackend) -> SizeMap {
|
|
let hashes: Vec<String> = size_by_hash.keys().cloned().collect();
|
|
stream::iter(hashes)
|
|
.map(|hash| async move {
|
|
let size = backend.blob_size(&hash).await;
|
|
(hash, size)
|
|
})
|
|
.buffer_unordered(CONCURRENCY)
|
|
.fold(size_by_hash, |mut sizes, (hash, size)| async move {
|
|
sizes.insert(hash, size);
|
|
sizes
|
|
})
|
|
.await
|
|
}
|
|
|
|
async fn candidate_batch(manifests: &[Manifest], backend: &SimBackend) -> Vec<String> {
|
|
// Invalid manifests are skipped by the current implementation, so their
|
|
// hashes must not become backend calls in the candidate either.
|
|
let mut size_by_hash = SizeMap::default();
|
|
for (_, hashes, chunk_sizes, _) in manifests {
|
|
if hashes.len() == chunk_sizes.len() {
|
|
for hash in hashes {
|
|
size_by_hash.entry(hash.clone()).or_insert(None);
|
|
}
|
|
}
|
|
}
|
|
let size_by_hash = fill_sizes(size_by_hash, backend).await;
|
|
replay(manifests, &size_by_hash)
|
|
}
|
|
|
|
async fn candidate_large_manifest(manifest: &Manifest, backend: &SimBackend) -> Vec<String> {
|
|
let (file_hash, hashes, chunk_sizes, total_size) = manifest;
|
|
let file_label = label(file_hash);
|
|
if hashes.len() != chunk_sizes.len() {
|
|
return vec![format!(
|
|
"Manifest {file_label}: chunk_hashes/chunk_sizes length mismatch"
|
|
)];
|
|
}
|
|
let mut issues = Vec::new();
|
|
let sum: i64 = chunk_sizes.iter().sum();
|
|
if sum != *total_size {
|
|
issues.push(format!(
|
|
"Manifest {file_label}: total_size {total_size} != sum of chunk_sizes {sum}"
|
|
));
|
|
}
|
|
for offset in (0..hashes.len()).step_by(CANDIDATE_BATCH_OCCURRENCES) {
|
|
let end = (offset + CANDIDATE_BATCH_OCCURRENCES).min(hashes.len());
|
|
let mut size_by_hash = SizeMap::default();
|
|
for hash in &hashes[offset..end] {
|
|
size_by_hash.entry(hash.clone()).or_insert(None);
|
|
}
|
|
let size_by_hash = fill_sizes(size_by_hash, backend).await;
|
|
for (index, hash) in hashes[offset..end].iter().enumerate() {
|
|
let expected = chunk_sizes[offset + index];
|
|
let chunk_label = label(hash);
|
|
match size_by_hash.get(hash.as_str()).copied().flatten() {
|
|
Some(actual) if actual != expected as u64 => issues.push(format!(
|
|
"Manifest {file_label} chunk {chunk_label}: size mismatch \
|
|
(expected {expected}, actual {actual})"
|
|
)),
|
|
None => issues.push(format!(
|
|
"Manifest {file_label} chunk {chunk_label}: missing in backend"
|
|
)),
|
|
Some(_) => {}
|
|
}
|
|
}
|
|
}
|
|
issues
|
|
}
|
|
|
|
async fn candidate(manifests: &[Manifest], backend: &SimBackend) -> Vec<String> {
|
|
let mut issues = Vec::new();
|
|
let mut start = 0;
|
|
while start < manifests.len() {
|
|
let (_, hashes, chunk_sizes, _) = &manifests[start];
|
|
if hashes.len() == chunk_sizes.len() && hashes.len() > CANDIDATE_BATCH_OCCURRENCES {
|
|
issues.extend(candidate_large_manifest(&manifests[start], backend).await);
|
|
start += 1;
|
|
continue;
|
|
}
|
|
|
|
let mut occurrences = 0;
|
|
let mut end = start;
|
|
while end < manifests.len() {
|
|
let (_, hashes, chunk_sizes, _) = &manifests[end];
|
|
let next = if hashes.len() == chunk_sizes.len() {
|
|
hashes.len()
|
|
} else {
|
|
0
|
|
};
|
|
if next > CANDIDATE_BATCH_OCCURRENCES
|
|
|| (occurrences > 0 && occurrences + next > CANDIDATE_BATCH_OCCURRENCES)
|
|
{
|
|
break;
|
|
}
|
|
occurrences += next;
|
|
end += 1;
|
|
}
|
|
debug_assert!(end > start);
|
|
issues.extend(candidate_batch(&manifests[start..end], backend).await);
|
|
start = end;
|
|
}
|
|
issues
|
|
}
|
|
|
|
fn storage_hashes(manifests: &[Manifest]) -> Vec<&str> {
|
|
let mut sizes: HashMap<&str, (), RandomState> = HashMap::default();
|
|
for (_, hashes, chunk_sizes, _) in manifests {
|
|
if hashes.len() == chunk_sizes.len() {
|
|
for hash in hashes {
|
|
// Keep phase 2 free of synthetic issues: both variants then
|
|
// append exactly the same empty vector regardless of task
|
|
// completion order. Phase-1 anomaly semantics remain gated.
|
|
if !hash.starts_with("missing-") && !hash.starts_with("wrong-") {
|
|
sizes.entry(hash.as_str()).or_insert(());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
let mut hashes: Vec<&str> = sizes.into_keys().collect();
|
|
hashes.sort_unstable();
|
|
hashes
|
|
}
|
|
|
|
async fn phase_two(blob_hashes: &[&str], backend: &SimBackend) -> Vec<String> {
|
|
stream::iter(blob_hashes.iter().copied())
|
|
.map(|hash| async move {
|
|
let mut issues = Vec::new();
|
|
match backend.blob_size(hash).await {
|
|
Some(actual) if actual != 256 => {
|
|
issues.push(format!(
|
|
"{hash}: size mismatch (expected: 256, actual: {actual})"
|
|
));
|
|
}
|
|
None => {
|
|
issues.push(format!("{hash}: blob missing in backend"));
|
|
return issues;
|
|
}
|
|
Some(_) => {}
|
|
}
|
|
backend.hash_local_blob().await;
|
|
issues
|
|
})
|
|
.buffer_unordered(CONCURRENCY)
|
|
.flat_map(stream::iter)
|
|
.collect()
|
|
.await
|
|
}
|
|
|
|
fn fixture(manifests: usize, chunks: usize, unique: usize, anomalies: bool) -> Vec<Manifest> {
|
|
let unique = unique.max(1);
|
|
let mut rows = Vec::with_capacity(manifests + 3);
|
|
for manifest in 0..manifests {
|
|
let hashes: Vec<String> = (0..chunks)
|
|
.map(|chunk| format!("chunk-{:058}", (manifest * chunks + chunk) % unique))
|
|
.collect();
|
|
rows.push((
|
|
format!("file-{manifest:059}"),
|
|
hashes,
|
|
vec![256; chunks],
|
|
(chunks * 256) as i64,
|
|
));
|
|
}
|
|
|
|
if !anomalies {
|
|
return rows;
|
|
}
|
|
|
|
// Semantic gates: total mismatch, repeated wrong-size hash, repeated
|
|
// missing hash, and a malformed manifest that must not trigger a call.
|
|
rows.push((
|
|
"sum-mismatch-file".into(),
|
|
vec!["wrong-shared".into(), "wrong-shared".into()],
|
|
vec![256, 257],
|
|
1,
|
|
));
|
|
rows.push((
|
|
"missing-file".into(),
|
|
vec!["missing-shared".into(), "missing-shared".into()],
|
|
vec![256, 256],
|
|
512,
|
|
));
|
|
rows.push((
|
|
"malformed-file".into(),
|
|
vec!["missing-must-not-be-queried".into()],
|
|
vec![],
|
|
0,
|
|
));
|
|
rows
|
|
}
|
|
|
|
fn median(mut values: Vec<Duration>) -> Duration {
|
|
values.sort_unstable();
|
|
values[values.len() / 2]
|
|
}
|
|
|
|
async fn timed_once(
|
|
candidate_mode: bool,
|
|
manifests: &[Manifest],
|
|
latency: Latency,
|
|
repetitions: usize,
|
|
full_method: bool,
|
|
blob_hashes: &[&str],
|
|
) -> (Duration, usize, Vec<String>) {
|
|
let backend = SimBackend::new(latency);
|
|
let start = Instant::now();
|
|
let mut issues = Vec::new();
|
|
for _ in 0..repetitions {
|
|
issues = if candidate_mode && !uses_serial_fast_path(manifests) {
|
|
candidate(manifests, &backend).await
|
|
} else {
|
|
current(manifests, &backend).await
|
|
};
|
|
if full_method {
|
|
issues.extend(phase_two(blob_hashes, &backend).await);
|
|
}
|
|
black_box(&issues);
|
|
}
|
|
(
|
|
start.elapsed() / repetitions as u32,
|
|
backend.calls() / repetitions,
|
|
issues,
|
|
)
|
|
}
|
|
|
|
async fn timed_pair(
|
|
manifests: &[Manifest],
|
|
latency: Latency,
|
|
samples: usize,
|
|
repetitions: usize,
|
|
full_method: bool,
|
|
blob_hashes: &[&str],
|
|
) -> (
|
|
(Duration, usize, Vec<String>),
|
|
(Duration, usize, Vec<String>),
|
|
) {
|
|
let mut current_times = Vec::with_capacity(samples);
|
|
let mut candidate_times = Vec::with_capacity(samples);
|
|
let mut current_observation = None;
|
|
let mut candidate_observation = None;
|
|
for sample in 0..samples + 1 {
|
|
let (current_run, candidate_run) = if sample % 2 == 0 {
|
|
(
|
|
timed_once(
|
|
false,
|
|
manifests,
|
|
latency,
|
|
repetitions,
|
|
full_method,
|
|
blob_hashes,
|
|
)
|
|
.await,
|
|
timed_once(
|
|
true,
|
|
manifests,
|
|
latency,
|
|
repetitions,
|
|
full_method,
|
|
blob_hashes,
|
|
)
|
|
.await,
|
|
)
|
|
} else {
|
|
let candidate = timed_once(
|
|
true,
|
|
manifests,
|
|
latency,
|
|
repetitions,
|
|
full_method,
|
|
blob_hashes,
|
|
)
|
|
.await;
|
|
let current = timed_once(
|
|
false,
|
|
manifests,
|
|
latency,
|
|
repetitions,
|
|
full_method,
|
|
blob_hashes,
|
|
)
|
|
.await;
|
|
(current, candidate)
|
|
};
|
|
if sample > 0 {
|
|
current_times.push(current_run.0);
|
|
candidate_times.push(candidate_run.0);
|
|
current_observation = Some((current_run.1, current_run.2));
|
|
candidate_observation = Some((candidate_run.1, candidate_run.2));
|
|
}
|
|
}
|
|
let (current_calls, current_issues) = current_observation.expect("measured current run");
|
|
let (candidate_calls, candidate_issues) =
|
|
candidate_observation.expect("measured candidate run");
|
|
(
|
|
(median(current_times), current_calls, current_issues),
|
|
(median(candidate_times), candidate_calls, candidate_issues),
|
|
)
|
|
}
|
|
|
|
async fn raw_pair(
|
|
manifests: &[Manifest],
|
|
latency: Latency,
|
|
samples: usize,
|
|
repetitions: usize,
|
|
full_method: bool,
|
|
blob_hashes: &[&str],
|
|
) -> (Vec<f64>, Vec<f64>) {
|
|
let mut current_times = Vec::with_capacity(samples);
|
|
let mut candidate_times = Vec::with_capacity(samples);
|
|
for sample in 0..samples + 1 {
|
|
let (current, candidate) = if sample % 2 == 0 {
|
|
(
|
|
timed_once(
|
|
false,
|
|
manifests,
|
|
latency,
|
|
repetitions,
|
|
full_method,
|
|
blob_hashes,
|
|
)
|
|
.await,
|
|
timed_once(
|
|
true,
|
|
manifests,
|
|
latency,
|
|
repetitions,
|
|
full_method,
|
|
blob_hashes,
|
|
)
|
|
.await,
|
|
)
|
|
} else {
|
|
let candidate = timed_once(
|
|
true,
|
|
manifests,
|
|
latency,
|
|
repetitions,
|
|
full_method,
|
|
blob_hashes,
|
|
)
|
|
.await;
|
|
let current = timed_once(
|
|
false,
|
|
manifests,
|
|
latency,
|
|
repetitions,
|
|
full_method,
|
|
blob_hashes,
|
|
)
|
|
.await;
|
|
(current, candidate)
|
|
};
|
|
if sample > 0 {
|
|
current_times.push(current.0.as_secs_f64() * 1e3);
|
|
candidate_times.push(candidate.0.as_secs_f64() * 1e3);
|
|
}
|
|
}
|
|
(current_times, candidate_times)
|
|
}
|
|
|
|
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn main() {
|
|
let args: Vec<String> = std::env::args().collect();
|
|
if args.get(1).is_some_and(|value| value == "--tiny-raw") {
|
|
let rows = fixture(1, 1, 1, false);
|
|
let blob_hashes = storage_hashes(&rows);
|
|
let latency = Latency::Remote(Duration::from_millis(4));
|
|
let (phase_current, phase_candidate) =
|
|
raw_pair(&rows, latency, 31, 8, false, &blob_hashes).await;
|
|
let (full_current, full_candidate) =
|
|
raw_pair(&rows, latency, 31, 8, true, &blob_hashes).await;
|
|
println!("phase_current_ms={phase_current:?}");
|
|
println!("phase_candidate_ms={phase_candidate:?}");
|
|
println!("full_current_ms={full_current:?}");
|
|
println!("full_candidate_ms={full_candidate:?}");
|
|
return;
|
|
}
|
|
if args.get(1).is_some_and(|value| value == "--memory") {
|
|
let mode = args.get(2).map(String::as_str).unwrap_or("candidate");
|
|
assert!(matches!(mode, "current" | "candidate"));
|
|
// 250k unique occurrences: large enough for process-level max RSS to
|
|
// rise above allocator noise while keeping the probe quick.
|
|
let rows = fixture(1_000, 250, 250_000, false);
|
|
let backend = SimBackend::new(Latency::Immediate);
|
|
let issues = if mode == "candidate" {
|
|
candidate(&rows, &backend).await
|
|
} else {
|
|
current(&rows, &backend).await
|
|
};
|
|
black_box(&issues);
|
|
println!(
|
|
"mode={mode} manifests={} occurrences={} calls={} issues={}",
|
|
rows.len(),
|
|
250_000,
|
|
backend.calls(),
|
|
issues.len()
|
|
);
|
|
return;
|
|
}
|
|
|
|
let scenarios = [
|
|
(
|
|
"immediate_one_manifest_two",
|
|
Latency::Immediate,
|
|
1,
|
|
2,
|
|
2,
|
|
false,
|
|
51,
|
|
10_000,
|
|
),
|
|
(
|
|
"immediate_two_manifests_one",
|
|
Latency::Immediate,
|
|
2,
|
|
1,
|
|
2,
|
|
false,
|
|
51,
|
|
10_000,
|
|
),
|
|
(
|
|
"immediate_one_manifest_four",
|
|
Latency::Immediate,
|
|
1,
|
|
4,
|
|
4,
|
|
false,
|
|
51,
|
|
10_000,
|
|
),
|
|
(
|
|
"local_tiny_empty",
|
|
Latency::Local {
|
|
metadata: Duration::from_micros(250),
|
|
hash: Duration::from_millis(1),
|
|
},
|
|
0,
|
|
0,
|
|
1,
|
|
false,
|
|
101,
|
|
10_000,
|
|
),
|
|
(
|
|
"local_tiny_single",
|
|
Latency::Local {
|
|
metadata: Duration::from_micros(250),
|
|
hash: Duration::from_millis(1),
|
|
},
|
|
1,
|
|
1,
|
|
1,
|
|
false,
|
|
51,
|
|
128,
|
|
),
|
|
(
|
|
"local_small_unique",
|
|
Latency::Local {
|
|
metadata: Duration::from_micros(250),
|
|
hash: Duration::from_millis(1),
|
|
},
|
|
1,
|
|
4,
|
|
4,
|
|
false,
|
|
31,
|
|
32,
|
|
),
|
|
(
|
|
"local_semantics",
|
|
Latency::Local {
|
|
metadata: Duration::from_micros(250),
|
|
hash: Duration::from_millis(1),
|
|
},
|
|
2,
|
|
4,
|
|
4,
|
|
true,
|
|
15,
|
|
1,
|
|
),
|
|
(
|
|
"local_shared",
|
|
Latency::Local {
|
|
metadata: Duration::from_micros(250),
|
|
hash: Duration::from_millis(1),
|
|
},
|
|
64,
|
|
8,
|
|
32,
|
|
false,
|
|
7,
|
|
1,
|
|
),
|
|
(
|
|
"local_unique",
|
|
Latency::Local {
|
|
metadata: Duration::from_micros(250),
|
|
hash: Duration::from_millis(1),
|
|
},
|
|
32,
|
|
8,
|
|
256,
|
|
false,
|
|
7,
|
|
1,
|
|
),
|
|
(
|
|
"local_hash_dominated",
|
|
Latency::Local {
|
|
metadata: Duration::from_micros(250),
|
|
// Models large legacy/local blobs in phase 2. Both variants
|
|
// pay exactly the same bounded-concurrency rehash cost.
|
|
hash: Duration::from_millis(10),
|
|
},
|
|
64,
|
|
8,
|
|
32,
|
|
false,
|
|
3,
|
|
1,
|
|
),
|
|
(
|
|
"remote_tiny_single",
|
|
Latency::Remote(Duration::from_millis(4)),
|
|
1,
|
|
1,
|
|
1,
|
|
false,
|
|
31,
|
|
8,
|
|
),
|
|
(
|
|
"remote_shared",
|
|
Latency::Remote(Duration::from_millis(4)),
|
|
24,
|
|
8,
|
|
24,
|
|
false,
|
|
7,
|
|
1,
|
|
),
|
|
(
|
|
"remote_unique",
|
|
Latency::Remote(Duration::from_millis(4)),
|
|
16,
|
|
8,
|
|
128,
|
|
false,
|
|
7,
|
|
1,
|
|
),
|
|
];
|
|
|
|
println!(
|
|
"scenario,phase1_current_ms,phase1_candidate_ms,phase1_speedup,full_current_ms,\
|
|
full_candidate_ms,full_speedup,phase1_current_calls,phase1_candidate_calls,\
|
|
full_current_calls,full_candidate_calls,issues_equal"
|
|
);
|
|
let mut all_pass = true;
|
|
for (name, latency, manifest_count, chunks, unique, anomalies, samples, repetitions) in
|
|
scenarios
|
|
{
|
|
let rows = fixture(manifest_count, chunks, unique, anomalies);
|
|
let extra_storage: Vec<String> = if name == "local_hash_dominated" {
|
|
(0..128)
|
|
.map(|index| format!("legacy-{index:057}"))
|
|
.collect()
|
|
} else {
|
|
Vec::new()
|
|
};
|
|
let mut blob_hashes = storage_hashes(&rows);
|
|
blob_hashes.extend(extra_storage.iter().map(String::as_str));
|
|
let tiny = name.contains("tiny");
|
|
let serial_fast_path = uses_serial_fast_path(&rows);
|
|
let (
|
|
(phase_current, phase_current_calls, phase_current_issues),
|
|
(phase_candidate, phase_candidate_calls, phase_candidate_issues),
|
|
) = timed_pair(&rows, latency, samples, repetitions, false, &blob_hashes).await;
|
|
let (
|
|
(full_current, full_current_calls, full_current_issues),
|
|
(full_candidate, full_candidate_calls, full_candidate_issues),
|
|
) = timed_pair(
|
|
&rows,
|
|
latency,
|
|
samples,
|
|
if tiny {
|
|
repetitions.min(16)
|
|
} else {
|
|
repetitions
|
|
},
|
|
true,
|
|
&blob_hashes,
|
|
)
|
|
.await;
|
|
let equal = phase_current_issues == phase_candidate_issues
|
|
&& full_current_issues == full_candidate_issues;
|
|
let phase_speedup = if phase_current.is_zero() && phase_candidate.is_zero() {
|
|
1.0
|
|
} else {
|
|
phase_current.as_secs_f64() / phase_candidate.as_secs_f64()
|
|
};
|
|
let full_speedup = if full_current.is_zero() && full_candidate.is_zero() {
|
|
1.0
|
|
} else {
|
|
full_current.as_secs_f64() / full_candidate.as_secs_f64()
|
|
};
|
|
// Tiny fast paths tolerate only timer noise; substantive cases must
|
|
// be a strict win. Semantics and call-count reduction are hard gates.
|
|
let phase_timing_pass = if phase_current < Duration::from_nanos(100) {
|
|
// An empty Vec return is below the clock's useful resolution;
|
|
// permit at most twenty nanoseconds of measurement noise.
|
|
phase_candidate <= phase_current + Duration::from_nanos(20)
|
|
} else if serial_fast_path {
|
|
phase_speedup >= 0.95
|
|
} else {
|
|
phase_speedup > 1.0
|
|
};
|
|
let full_timing_pass = if full_current < Duration::from_nanos(100) {
|
|
full_candidate <= full_current + Duration::from_nanos(20)
|
|
} else if serial_fast_path {
|
|
full_speedup >= 0.95
|
|
} else {
|
|
full_speedup > 1.0
|
|
};
|
|
let calls_pass = phase_candidate_calls <= phase_current_calls
|
|
&& full_candidate_calls <= full_current_calls;
|
|
all_pass &= equal && phase_timing_pass && full_timing_pass && calls_pass;
|
|
println!(
|
|
"{name},{:.6},{:.6},{phase_speedup:.3},{:.6},{:.6},{full_speedup:.3},\
|
|
{phase_current_calls},{phase_candidate_calls},{full_current_calls},\
|
|
{full_candidate_calls},{equal}",
|
|
phase_current.as_secs_f64() * 1e3,
|
|
phase_candidate.as_secs_f64() * 1e3,
|
|
full_current.as_secs_f64() * 1e3,
|
|
full_candidate.as_secs_f64() * 1e3,
|
|
);
|
|
}
|
|
assert!(all_pass, "candidate failed a correctness/performance gate");
|
|
}
|