603 lines
22 KiB
Rust
603 lines
22 KiB
Rust
#![expect(
|
|
clippy::redundant_pub_crate,
|
|
reason = "criterion bench entry points are re-exported only to the private bench root module"
|
|
)]
|
|
|
|
use super::support::{
|
|
Arc, AtomicBool, BTreeMap, BenchmarkId, ConnectorBackedDownloader, Criterion, Duration,
|
|
Instant, Invocation, Mutex, Ordering, PathBuf, Read, ReqwestHttpConnector, SocketAddr,
|
|
SystemTime, TcpListener, TcpStream, Throughput, UNIX_EPOCH, Write,
|
|
execute_runtime_with_downloader, fs, thread,
|
|
};
|
|
|
|
const LIVE_TRANSFER_SPLIT: usize = 4;
|
|
const LIVE_TRANSFER_MAX_CONNECTIONS_PER_SERVER: usize = 4;
|
|
|
|
#[derive(Clone, Copy, Debug)]
|
|
struct LiveTransferScenario {
|
|
label: &'static str,
|
|
total_length: usize,
|
|
piece_length: usize,
|
|
overall_download_limit: Option<usize>,
|
|
disk_cache_bytes: Option<usize>,
|
|
response_delay_ms: u64,
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug)]
|
|
struct LiveTransferContentionScenario {
|
|
label: &'static str,
|
|
download_count: usize,
|
|
total_length: usize,
|
|
piece_length: usize,
|
|
overall_download_limit: Option<usize>,
|
|
disk_cache_bytes: Option<usize>,
|
|
response_delay_ms: u64,
|
|
}
|
|
|
|
#[derive(Clone, Debug, Default)]
|
|
struct SegmentRequestMetrics {
|
|
total_requests: usize,
|
|
requests_by_path: BTreeMap<String, usize>,
|
|
}
|
|
|
|
struct TempConfigFile {
|
|
path: PathBuf,
|
|
}
|
|
|
|
impl TempConfigFile {
|
|
fn new(contents: &str) -> Self {
|
|
let unique = SystemTime::now()
|
|
.duration_since(UNIX_EPOCH)
|
|
.expect("system time should be after unix epoch")
|
|
.as_nanos();
|
|
let path = std::env::temp_dir().join(format!("aria2-rust-pro-{unique}.conf"));
|
|
fs::write(&path, contents).expect("benchmark config should write");
|
|
Self { path }
|
|
}
|
|
|
|
const fn path(&self) -> &PathBuf {
|
|
&self.path
|
|
}
|
|
}
|
|
|
|
impl Drop for TempConfigFile {
|
|
fn drop(&mut self) {
|
|
let _ = fs::remove_file(&self.path);
|
|
}
|
|
}
|
|
|
|
struct LocalHttpSegmentServer {
|
|
base_url: String,
|
|
metrics: Arc<Mutex<SegmentRequestMetrics>>,
|
|
stop: Arc<AtomicBool>,
|
|
handle: Option<thread::JoinHandle<()>>,
|
|
listen_address: SocketAddr,
|
|
}
|
|
|
|
impl LocalHttpSegmentServer {
|
|
fn spawn(scenario: LiveTransferScenario) -> Self {
|
|
let expected_requests_per_download = scenario.total_length.div_ceil(scenario.piece_length);
|
|
Self::spawn_many(scenario, 1, expected_requests_per_download)
|
|
}
|
|
|
|
fn spawn_many(
|
|
scenario: LiveTransferScenario,
|
|
_download_count: usize,
|
|
_expected_requests_per_download: usize,
|
|
) -> Self {
|
|
let listener = TcpListener::bind("127.0.0.1:0").expect("loopback listener should bind");
|
|
let addr = listener
|
|
.local_addr()
|
|
.expect("loopback listener should report local addr");
|
|
let piece_length = scenario.piece_length;
|
|
let total_length = scenario.total_length;
|
|
let response_delay = Duration::from_millis(scenario.response_delay_ms);
|
|
let payload_bytes = Arc::<[u8]>::from(vec![b'x'; total_length]);
|
|
let metrics = Arc::new(Mutex::new(SegmentRequestMetrics::default()));
|
|
let metrics_for_thread = Arc::clone(&metrics);
|
|
let stop = Arc::new(AtomicBool::new(false));
|
|
let stop_for_thread = Arc::clone(&stop);
|
|
let handle = thread::spawn(move || {
|
|
loop {
|
|
let (stream, _) = listener
|
|
.accept()
|
|
.unwrap_or_else(|error| panic!("bench client should connect: {error}"));
|
|
if stop_for_thread.load(Ordering::Relaxed) {
|
|
break;
|
|
}
|
|
let payload_bytes = Arc::clone(&payload_bytes);
|
|
let metrics = Arc::clone(&metrics_for_thread);
|
|
thread::spawn(move || {
|
|
handle_loopback_segment_request(
|
|
stream,
|
|
&payload_bytes,
|
|
&metrics,
|
|
piece_length,
|
|
total_length,
|
|
response_delay,
|
|
);
|
|
});
|
|
}
|
|
});
|
|
Self {
|
|
base_url: format!("http://{addr}"),
|
|
metrics,
|
|
stop,
|
|
handle: Some(handle),
|
|
listen_address: addr,
|
|
}
|
|
}
|
|
|
|
fn url_for(&self, path: &str) -> String {
|
|
format!("{}/{}", self.base_url, path.trim_start_matches('/'))
|
|
}
|
|
|
|
fn snapshot_metrics(&self) -> SegmentRequestMetrics {
|
|
self.metrics
|
|
.lock()
|
|
.expect("segment request metrics mutex should not poison")
|
|
.clone()
|
|
}
|
|
}
|
|
|
|
impl Drop for LocalHttpSegmentServer {
|
|
fn drop(&mut self) {
|
|
self.stop.store(true, Ordering::Relaxed);
|
|
let _ = TcpStream::connect(self.listen_address);
|
|
if let Some(handle) = self.handle.take() {
|
|
handle.join().expect("loopback server thread should join");
|
|
}
|
|
}
|
|
}
|
|
|
|
fn handle_loopback_segment_request(
|
|
mut stream: TcpStream,
|
|
payload_bytes: &Arc<[u8]>,
|
|
metrics: &Arc<Mutex<SegmentRequestMetrics>>,
|
|
piece_length: usize,
|
|
total_length: usize,
|
|
response_delay: Duration,
|
|
) {
|
|
stream
|
|
.set_nonblocking(false)
|
|
.expect("accepted bench socket should switch back to blocking mode");
|
|
let mut request = [0_u8; 4096];
|
|
let read = stream.read(&mut request).expect("request should read");
|
|
let request_text = String::from_utf8_lossy(&request[..read]);
|
|
let request_path = parse_request_path(&request_text).to_owned();
|
|
let (start, end_inclusive) = parse_requested_range(&request_text, piece_length, total_length);
|
|
let len = end_inclusive.saturating_sub(start).saturating_add(1);
|
|
let body = payload_bytes.get(start..=end_inclusive).unwrap_or(&[]);
|
|
{
|
|
let mut metrics = metrics
|
|
.lock()
|
|
.expect("segment request metrics mutex should not poison");
|
|
metrics.total_requests += 1;
|
|
*metrics.requests_by_path.entry(request_path).or_default() += 1;
|
|
}
|
|
thread::sleep(response_delay);
|
|
let response_head = format!(
|
|
"HTTP/1.1 206 Partial Content\r\nContent-Length: {len}\r\nContent-Range: bytes {start}-{end_inclusive}/{total_length}\r\nConnection: close\r\n\r\n"
|
|
);
|
|
stream
|
|
.write_all(response_head.as_bytes())
|
|
.expect("response head should write");
|
|
stream.write_all(body).expect("response body should write");
|
|
}
|
|
|
|
fn parse_request_path(request_text: &str) -> &str {
|
|
request_text
|
|
.lines()
|
|
.next()
|
|
.and_then(|line| line.split_whitespace().nth(1))
|
|
.unwrap_or("/payload.bin")
|
|
}
|
|
|
|
fn parse_requested_range(
|
|
request_text: &str,
|
|
piece_length: usize,
|
|
total_length: usize,
|
|
) -> (usize, usize) {
|
|
let requested = request_text
|
|
.lines()
|
|
.find_map(|line| {
|
|
let (name, value) = line.split_once(':')?;
|
|
if !name.trim().eq_ignore_ascii_case("range") {
|
|
return None;
|
|
}
|
|
value.trim().strip_prefix("bytes=")
|
|
})
|
|
.and_then(|value| value.split_once('-'))
|
|
.map(|(start, end)| {
|
|
let start = start.parse::<usize>().expect("range start should parse");
|
|
let end = end.parse::<usize>().expect("range end should parse");
|
|
(start, end.min(total_length.saturating_sub(1)))
|
|
});
|
|
|
|
requested.unwrap_or_else(|| {
|
|
let start = 0usize;
|
|
let end = piece_length
|
|
.saturating_sub(1)
|
|
.min(total_length.saturating_sub(1));
|
|
(start, end)
|
|
})
|
|
}
|
|
|
|
fn build_live_transfer_config(scenario: LiveTransferScenario) -> TempConfigFile {
|
|
let mut lines = vec![
|
|
format!("split={LIVE_TRANSFER_SPLIT}"),
|
|
format!("max-connection-per-server={LIVE_TRANSFER_MAX_CONNECTIONS_PER_SERVER}"),
|
|
format!("min-split-size={}", scenario.piece_length),
|
|
format!("piece-length={}", scenario.piece_length),
|
|
];
|
|
if let Some(limit) = scenario.overall_download_limit {
|
|
lines.push(format!("max-overall-download-limit={limit}"));
|
|
}
|
|
if let Some(disk_cache_bytes) = scenario.disk_cache_bytes {
|
|
lines.push(format!("disk-cache={disk_cache_bytes}"));
|
|
}
|
|
TempConfigFile::new(&lines.join("\n"))
|
|
}
|
|
|
|
fn min_expected_live_requests_per_download(scenario: LiveTransferContentionScenario) -> usize {
|
|
let total_segments = scenario.total_length.div_ceil(scenario.piece_length);
|
|
if LIVE_TRANSFER_MAX_CONNECTIONS_PER_SERVER >= 4 {
|
|
let probe_segments = if scenario.piece_length <= 64 * 1024 {
|
|
LIVE_TRANSFER_MAX_CONNECTIONS_PER_SERVER.saturating_mul(2)
|
|
} else {
|
|
2usize
|
|
};
|
|
if total_segments <= probe_segments {
|
|
return 1;
|
|
}
|
|
|
|
let remaining_segments = total_segments.saturating_sub(probe_segments);
|
|
let followup_floor = if remaining_segments <= 2 {
|
|
1
|
|
} else if remaining_segments <= 3 {
|
|
2
|
|
} else {
|
|
remaining_segments.min(LIVE_TRANSFER_SPLIT.saturating_sub(1))
|
|
};
|
|
return 1usize.saturating_add(followup_floor);
|
|
}
|
|
|
|
let mut segmented_floor = 1usize.saturating_add(
|
|
total_segments
|
|
.saturating_sub(1)
|
|
.min(LIVE_TRANSFER_SPLIT.saturating_sub(1)),
|
|
);
|
|
if segmented_floor > 3 && total_segments <= LIVE_TRANSFER_SPLIT {
|
|
segmented_floor = segmented_floor.saturating_sub(1);
|
|
}
|
|
total_segments.min(segmented_floor).max(1)
|
|
}
|
|
|
|
fn request_budget_per_download(scenario: LiveTransferContentionScenario) -> usize {
|
|
min_expected_live_requests_per_download(scenario).saturating_mul(4)
|
|
}
|
|
|
|
fn run_live_transfer_contention_scenario(scenario: LiveTransferScenario) -> usize {
|
|
let config = build_live_transfer_config(scenario);
|
|
let server = LocalHttpSegmentServer::spawn(scenario);
|
|
let connector = ReqwestHttpConnector::new().expect("reqwest connector should build");
|
|
let downloader = ConnectorBackedDownloader::new(connector.clone(), connector);
|
|
let report = execute_runtime_with_downloader(
|
|
Invocation::Run {
|
|
config_path: Some(config.path().clone()),
|
|
uris: vec![server.url_for("payload.bin")],
|
|
},
|
|
&downloader,
|
|
)
|
|
.expect("live transfer benchmark should complete");
|
|
assert_eq!(report.completed_download_count, 1);
|
|
assert_eq!(report.first_status.as_deref(), Some("complete"));
|
|
assert_eq!(
|
|
report.first_completed_length,
|
|
Some(scenario.total_length as u64)
|
|
);
|
|
scenario.total_length
|
|
}
|
|
|
|
fn run_live_transfer_fairness_contention_scenario(
|
|
scenario: LiveTransferContentionScenario,
|
|
) -> usize {
|
|
let single_transfer = LiveTransferScenario {
|
|
label: scenario.label,
|
|
total_length: scenario.total_length,
|
|
piece_length: scenario.piece_length,
|
|
overall_download_limit: scenario.overall_download_limit,
|
|
disk_cache_bytes: scenario.disk_cache_bytes,
|
|
response_delay_ms: scenario.response_delay_ms,
|
|
};
|
|
let min_requests_per_download = min_expected_live_requests_per_download(scenario);
|
|
let request_budget_per_download = request_budget_per_download(scenario);
|
|
let server = LocalHttpSegmentServer::spawn_many(
|
|
single_transfer,
|
|
scenario.download_count,
|
|
request_budget_per_download,
|
|
);
|
|
let worker_handles = (0..scenario.download_count)
|
|
.map(|_| build_live_transfer_config(single_transfer))
|
|
.enumerate()
|
|
.map(|(index, config)| {
|
|
let config_path = config.path().clone();
|
|
let uri = server.url_for(&format!("payload-{index}.bin"));
|
|
thread::spawn(move || {
|
|
let _config_guard = config;
|
|
let connector =
|
|
ReqwestHttpConnector::new().expect("reqwest connector should build");
|
|
let downloader = ConnectorBackedDownloader::new(connector.clone(), connector);
|
|
let started = Instant::now();
|
|
let report = execute_runtime_with_downloader(
|
|
Invocation::Run {
|
|
config_path: Some(config_path),
|
|
uris: vec![uri],
|
|
},
|
|
&downloader,
|
|
)
|
|
.expect("parallel live transfer benchmark should complete");
|
|
(started.elapsed(), report)
|
|
})
|
|
})
|
|
.collect::<Vec<_>>();
|
|
|
|
let mut elapsed = Vec::with_capacity(scenario.download_count);
|
|
for worker in worker_handles {
|
|
let (duration, report) = worker
|
|
.join()
|
|
.expect("parallel live transfer worker thread should join");
|
|
assert_eq!(report.completed_download_count, 1);
|
|
assert_eq!(report.first_status.as_deref(), Some("complete"));
|
|
assert_eq!(
|
|
report.first_completed_length,
|
|
Some(scenario.total_length as u64)
|
|
);
|
|
elapsed.push(duration);
|
|
}
|
|
|
|
let metrics = server.snapshot_metrics();
|
|
assert_eq!(metrics.requests_by_path.len(), scenario.download_count);
|
|
assert!(
|
|
metrics.total_requests >= min_requests_per_download * scenario.download_count,
|
|
"parallel live transfer should keep every download progressing through segmented requests: observed total_requests={} floor_per_download={} download_count={}",
|
|
metrics.total_requests,
|
|
min_requests_per_download,
|
|
scenario.download_count
|
|
);
|
|
for index in 0..scenario.download_count {
|
|
let path = format!("/payload-{index}.bin");
|
|
let observed = metrics
|
|
.requests_by_path
|
|
.get(&path)
|
|
.copied()
|
|
.unwrap_or_default();
|
|
assert!(
|
|
observed >= min_requests_per_download,
|
|
"each download should exercise the segmented path under contention: path={path} observed={observed} floor={min_requests_per_download}"
|
|
);
|
|
}
|
|
|
|
let min_elapsed = elapsed
|
|
.iter()
|
|
.min()
|
|
.copied()
|
|
.expect("at least one live contention run should exist");
|
|
let max_elapsed = elapsed
|
|
.iter()
|
|
.max()
|
|
.copied()
|
|
.expect("at least one live contention run should exist");
|
|
assert!(max_elapsed >= min_elapsed);
|
|
scenario.total_length * scenario.download_count
|
|
}
|
|
|
|
fn run_live_shared_runtime_multi_download_scenario(
|
|
scenario: LiveTransferContentionScenario,
|
|
) -> usize {
|
|
let single_transfer = LiveTransferScenario {
|
|
label: scenario.label,
|
|
total_length: scenario.total_length,
|
|
piece_length: scenario.piece_length,
|
|
overall_download_limit: scenario.overall_download_limit,
|
|
disk_cache_bytes: scenario.disk_cache_bytes,
|
|
response_delay_ms: scenario.response_delay_ms,
|
|
};
|
|
let config = build_live_transfer_config(single_transfer);
|
|
let min_requests_per_download = min_expected_live_requests_per_download(scenario);
|
|
let request_budget_per_download = request_budget_per_download(scenario);
|
|
let server = LocalHttpSegmentServer::spawn_many(
|
|
single_transfer,
|
|
scenario.download_count,
|
|
request_budget_per_download,
|
|
);
|
|
let connector = ReqwestHttpConnector::new().expect("reqwest connector should build");
|
|
let downloader = ConnectorBackedDownloader::new(connector.clone(), connector);
|
|
let report = execute_runtime_with_downloader(
|
|
Invocation::Run {
|
|
config_path: Some(config.path().clone()),
|
|
uris: (0..scenario.download_count)
|
|
.map(|index| server.url_for(&format!("shared-runtime-{index}.bin")))
|
|
.collect(),
|
|
},
|
|
&downloader,
|
|
)
|
|
.expect("shared-runtime live transfer benchmark should complete");
|
|
|
|
assert_eq!(report.accepted_uri_count, scenario.download_count);
|
|
assert_eq!(report.tracked_download_count, scenario.download_count);
|
|
assert_eq!(report.completed_download_count, scenario.download_count);
|
|
assert_eq!(report.first_status.as_deref(), Some("complete"));
|
|
assert_eq!(
|
|
report.first_completed_length,
|
|
Some(scenario.total_length as u64)
|
|
);
|
|
|
|
let metrics = server.snapshot_metrics();
|
|
assert_eq!(metrics.requests_by_path.len(), scenario.download_count);
|
|
assert!(
|
|
metrics.total_requests >= min_requests_per_download * scenario.download_count,
|
|
"shared-runtime live transfer should keep every registered download progressing: observed total_requests={} floor_per_download={} download_count={}",
|
|
metrics.total_requests,
|
|
min_requests_per_download,
|
|
scenario.download_count
|
|
);
|
|
for index in 0..scenario.download_count {
|
|
let path = format!("/shared-runtime-{index}.bin");
|
|
let observed = metrics
|
|
.requests_by_path
|
|
.get(&path)
|
|
.copied()
|
|
.unwrap_or_default();
|
|
assert!(
|
|
observed >= min_requests_per_download,
|
|
"shared-runtime live transfer should fully progress each registered download: path={path} observed={observed} floor={min_requests_per_download}"
|
|
);
|
|
}
|
|
|
|
scenario.total_length * scenario.download_count
|
|
}
|
|
|
|
pub(super) fn bench_live_http_transfer_contention_pressure(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("live_http_transfer_contention_pressure");
|
|
for scenario in [
|
|
LiveTransferScenario {
|
|
label: "loose_cap",
|
|
total_length: 16 * 1024,
|
|
piece_length: 4 * 1024,
|
|
overall_download_limit: None,
|
|
disk_cache_bytes: None,
|
|
response_delay_ms: 8,
|
|
},
|
|
LiveTransferScenario {
|
|
label: "tight_cap",
|
|
total_length: 16 * 1024,
|
|
piece_length: 4 * 1024,
|
|
overall_download_limit: Some(4 * 1024),
|
|
disk_cache_bytes: None,
|
|
response_delay_ms: 8,
|
|
},
|
|
] {
|
|
group.throughput(Throughput::Bytes(scenario.total_length as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::new("live_http_transfer", scenario.label),
|
|
&scenario,
|
|
|b, &scenario| {
|
|
b.iter_batched(
|
|
|| scenario,
|
|
|scenario| {
|
|
let bytes = run_live_transfer_contention_scenario(scenario);
|
|
assert_eq!(bytes, scenario.total_length);
|
|
},
|
|
criterion::BatchSize::SmallInput,
|
|
);
|
|
},
|
|
);
|
|
}
|
|
group.finish();
|
|
}
|
|
|
|
pub(super) fn bench_live_http_multi_download_contention_pressure(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("live_http_multi_download_contention_pressure");
|
|
for scenario in [
|
|
LiveTransferContentionScenario {
|
|
label: "loose_cap",
|
|
download_count: 3,
|
|
total_length: 16 * 1024,
|
|
piece_length: 4 * 1024,
|
|
overall_download_limit: None,
|
|
disk_cache_bytes: None,
|
|
response_delay_ms: 8,
|
|
},
|
|
LiveTransferContentionScenario {
|
|
label: "tight_cap",
|
|
download_count: 3,
|
|
total_length: 16 * 1024,
|
|
piece_length: 4 * 1024,
|
|
overall_download_limit: Some(4 * 1024),
|
|
disk_cache_bytes: None,
|
|
response_delay_ms: 8,
|
|
},
|
|
] {
|
|
group.throughput(Throughput::Bytes(
|
|
(scenario.total_length * scenario.download_count) as u64,
|
|
));
|
|
group.bench_with_input(
|
|
BenchmarkId::new("multi_live_http_transfer", scenario.label),
|
|
&scenario,
|
|
|b, &scenario| {
|
|
b.iter_batched(
|
|
|| scenario,
|
|
|scenario| {
|
|
let bytes = run_live_transfer_fairness_contention_scenario(scenario);
|
|
assert_eq!(bytes, scenario.total_length * scenario.download_count);
|
|
},
|
|
criterion::BatchSize::SmallInput,
|
|
);
|
|
},
|
|
);
|
|
}
|
|
group.finish();
|
|
}
|
|
|
|
pub(super) fn bench_live_http_shared_runtime_multi_download_pressure(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("live_http_shared_runtime_multi_download_pressure");
|
|
for scenario in [
|
|
LiveTransferContentionScenario {
|
|
label: "loose_cap",
|
|
download_count: 3,
|
|
total_length: 16 * 1024,
|
|
piece_length: 4 * 1024,
|
|
overall_download_limit: None,
|
|
disk_cache_bytes: None,
|
|
response_delay_ms: 8,
|
|
},
|
|
LiveTransferContentionScenario {
|
|
label: "tight_cap",
|
|
download_count: 3,
|
|
total_length: 16 * 1024,
|
|
piece_length: 4 * 1024,
|
|
overall_download_limit: Some(4 * 1024),
|
|
disk_cache_bytes: None,
|
|
response_delay_ms: 8,
|
|
},
|
|
LiveTransferContentionScenario {
|
|
label: "tight_cap_6way",
|
|
download_count: 6,
|
|
total_length: 16 * 1024,
|
|
piece_length: 4 * 1024,
|
|
overall_download_limit: Some(4 * 1024),
|
|
disk_cache_bytes: None,
|
|
response_delay_ms: 8,
|
|
},
|
|
LiveTransferContentionScenario {
|
|
label: "cache_pressure_6way_256k",
|
|
download_count: 6,
|
|
total_length: 256 * 1024,
|
|
piece_length: 32 * 1024,
|
|
overall_download_limit: Some(64 * 1024),
|
|
disk_cache_bytes: Some(64 * 1024),
|
|
response_delay_ms: 2,
|
|
},
|
|
] {
|
|
group.throughput(Throughput::Bytes(
|
|
(scenario.total_length * scenario.download_count) as u64,
|
|
));
|
|
group.bench_with_input(
|
|
BenchmarkId::new("shared_runtime_live_http_transfer", scenario.label),
|
|
&scenario,
|
|
|b, &scenario| {
|
|
b.iter_batched(
|
|
|| scenario,
|
|
|scenario| {
|
|
let bytes = run_live_shared_runtime_multi_download_scenario(scenario);
|
|
assert_eq!(bytes, scenario.total_length * scenario.download_count);
|
|
},
|
|
criterion::BatchSize::SmallInput,
|
|
);
|
|
},
|
|
);
|
|
}
|
|
group.finish();
|
|
}
|