Files
aria2-rust-pro/crates/aria2-rust-pro-tests/benches/rpc_pressure/rpc_runtime_pressure.rs
T

668 lines
23 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::{
BTreeMap, BenchmarkId, Criterion, InProcessRpcDispatcher, RpcMethod, RpcValue, RuntimeConfig,
Throughput, rpc_request,
};
#[derive(Clone, Copy, Debug)]
struct PressureScenario {
task_count: usize,
rounds: usize,
split: usize,
max_connections_per_server: usize,
}
impl PressureScenario {
fn runtime(self) -> RuntimeConfig {
RuntimeConfig {
split: self.split,
max_connections_per_server: self.max_connections_per_server,
max_connection_per_server: self.max_connections_per_server,
min_split_size: 1024,
piece_length: 1024,
..RuntimeConfig::default()
}
}
}
#[derive(Clone, Copy, Debug)]
struct ResourceLimitScenario {
task_count: usize,
rounds: usize,
global_download_limit: u64,
per_download_limit: u64,
global_upload_limit: u64,
per_upload_limit: u64,
disk_cache_bytes: u64,
}
#[derive(Clone, Copy, Debug)]
struct SharedRuntimeFairnessScenario {
label: &'static str,
task_count: usize,
rounds: usize,
global_download_limit: u64,
default_per_download_limit: u64,
constrained_per_download_limit: u64,
global_upload_limit: u64,
default_per_upload_limit: u64,
constrained_per_upload_limit: u64,
}
fn seed_pressure_dispatcher(scenario: PressureScenario) -> (InProcessRpcDispatcher, Vec<String>) {
let mut dispatcher = InProcessRpcDispatcher::with_runtime(scenario.runtime());
let mut gids = Vec::with_capacity(scenario.task_count);
for i in 0..scenario.task_count {
let magnet = format!(
"magnet:?xt=urn:btih:{:040x}&dn=bench-{i}&tr=http%3A%2F%2Ftracker.example.org%2Fannounce",
i + 40_001
);
let response = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2AddUri,
vec![RpcValue::String(magnet)],
));
match response.result {
Some(RpcValue::String(gid)) => gids.push(gid),
other => panic!("unexpected addUri result in bench setup: {other:?}"),
}
}
for (index, gid) in gids.iter().enumerate() {
dispatcher
.apply_bt_runtime_tick(
gid,
1024 + (index % 4) as u64 * 256,
256 + (index % 3) as u64 * 64,
300 + index as u64 % 80,
120 + index as u64 % 40,
1,
1,
false,
Some(4 + (index % 4) as u32),
)
.expect("bench setup runtime tick should succeed");
}
(dispatcher, gids)
}
fn seed_limited_dispatcher(
scenario: ResourceLimitScenario,
) -> (InProcessRpcDispatcher, Vec<String>) {
let base = PressureScenario {
task_count: scenario.task_count,
rounds: scenario.rounds,
split: 6,
max_connections_per_server: 6,
};
let (mut dispatcher, gids) = seed_pressure_dispatcher(base);
let change_global = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2ChangeGlobalOption,
vec![RpcValue::Object(BTreeMap::from([
(
"max-overall-download-limit".to_owned(),
RpcValue::String(scenario.global_download_limit.to_string()),
),
(
"max-overall-upload-limit".to_owned(),
RpcValue::String(scenario.global_upload_limit.to_string()),
),
(
"disk-cache".to_owned(),
RpcValue::String(scenario.disk_cache_bytes.to_string()),
),
]))],
));
assert!(
change_global.error.is_none(),
"changeGlobalOption should succeed in benchmark setup"
);
for gid in &gids {
let change_option = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2ChangeOption,
vec![
RpcValue::String(gid.clone()),
RpcValue::Object(BTreeMap::from([
(
"max-download-limit".to_owned(),
RpcValue::String(scenario.per_download_limit.to_string()),
),
(
"max-upload-limit".to_owned(),
RpcValue::String(scenario.per_upload_limit.to_string()),
),
])),
],
));
assert!(
change_option.error.is_none(),
"changeOption should succeed in benchmark setup"
);
}
(dispatcher, gids)
}
fn seed_shared_runtime_fairness_dispatcher(
scenario: SharedRuntimeFairnessScenario,
) -> (InProcessRpcDispatcher, Vec<String>) {
let base = PressureScenario {
task_count: scenario.task_count,
rounds: scenario.rounds,
split: 6,
max_connections_per_server: 6,
};
let (mut dispatcher, gids) = seed_pressure_dispatcher(base);
let change_global = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2ChangeGlobalOption,
vec![RpcValue::Object(BTreeMap::from([
(
"max-overall-download-limit".to_owned(),
RpcValue::String(scenario.global_download_limit.to_string()),
),
(
"max-overall-upload-limit".to_owned(),
RpcValue::String(scenario.global_upload_limit.to_string()),
),
]))],
));
assert!(
change_global.error.is_none(),
"changeGlobalOption should succeed in shared-runtime fairness setup"
);
for (index, gid) in gids.iter().enumerate() {
let (download_limit, upload_limit) = if index == 0 {
(
scenario.constrained_per_download_limit,
scenario.constrained_per_upload_limit,
)
} else {
(
scenario.default_per_download_limit,
scenario.default_per_upload_limit,
)
};
let change_option = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2ChangeOption,
vec![
RpcValue::String(gid.clone()),
RpcValue::Object(BTreeMap::from([
(
"max-download-limit".to_owned(),
RpcValue::String(download_limit.to_string()),
),
(
"max-upload-limit".to_owned(),
RpcValue::String(upload_limit.to_string()),
),
])),
],
));
assert!(
change_option.error.is_none(),
"changeOption should succeed in shared-runtime fairness setup"
);
}
(dispatcher, gids)
}
fn run_tell_status_pressure(
dispatcher: &mut InProcessRpcDispatcher,
gids: &[String],
rounds: usize,
) -> usize {
let mut calls = 0usize;
for round in 0..rounds {
let gid = &gids[round % gids.len()];
dispatcher
.apply_bt_runtime_tick(
gid,
0,
0,
400 + round as u64 * 10,
160 + round as u64 * 5,
0,
0,
false,
Some(8),
)
.expect("bench tellStatus churn should succeed");
for gid in gids {
let response = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2TellStatus,
vec![RpcValue::String(gid.clone())],
));
match response.result {
Some(RpcValue::Object(payload)) => {
assert_eq!(payload.get("isBt"), Some(&RpcValue::Bool(true)));
calls += 1;
}
other => panic!("unexpected tellStatus payload in bench: {other:?}"),
}
}
}
calls
}
fn run_resource_limit_pressure(
dispatcher: &mut InProcessRpcDispatcher,
gids: &[String],
scenario: ResourceLimitScenario,
) -> usize {
let mut calls = 0usize;
let expected_download_speed = (scenario.global_download_limit / scenario.task_count as u64)
.min(scenario.per_download_limit)
.max(1);
let expected_upload_speed = (scenario.global_upload_limit / scenario.task_count as u64)
.min(scenario.per_upload_limit)
.max(1);
for round in 0..scenario.rounds {
for gid in gids {
dispatcher
.apply_bt_runtime_tick(gid, 64, 32, 5_000, 2_000, 1, 1, false, Some(6))
.expect("bench limited runtime tick should succeed");
}
let gid = &gids[round % gids.len()];
let tell_status = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2TellStatus,
vec![RpcValue::String(gid.clone())],
));
match tell_status.result {
Some(RpcValue::Object(payload)) => {
assert_eq!(
payload.get("downloadSpeed"),
Some(&RpcValue::String(expected_download_speed.to_string()))
);
assert_eq!(
payload.get("uploadSpeed"),
Some(&RpcValue::String(expected_upload_speed.to_string()))
);
calls += 1;
}
other => panic!("unexpected tellStatus payload under resource caps: {other:?}"),
}
let tell_global =
dispatcher.dispatch_json(rpc_request(RpcMethod::Aria2TellGlobalStat, Vec::new()));
match tell_global.result {
Some(RpcValue::Object(payload)) => {
assert_eq!(
payload.get("downloadSpeed"),
Some(&RpcValue::String(
expected_download_speed
.saturating_mul(scenario.task_count as u64)
.to_string(),
))
);
assert_eq!(
payload.get("uploadSpeed"),
Some(&RpcValue::String(
expected_upload_speed
.saturating_mul(scenario.task_count as u64)
.to_string(),
))
);
calls += 1;
}
other => panic!("unexpected tellGlobalStat payload under resource caps: {other:?}"),
}
}
calls
}
fn run_mixed_rpc_pressure(
dispatcher: &mut InProcessRpcDispatcher,
gids: &[String],
rounds: usize,
) -> usize {
let mut calls = 0usize;
for round in 0..rounds {
let gid = &gids[round % gids.len()];
dispatcher
.apply_bt_runtime_tick(
gid,
128,
64,
500 + round as u64 * 15,
180 + round as u64 * 6,
1,
1,
round + 1 >= rounds,
Some(6 + (round % 4) as u32),
)
.expect("bench mixed churn should succeed");
let tell_status = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2TellStatus,
vec![RpcValue::String(gid.clone())],
));
assert!(matches!(tell_status.result, Some(RpcValue::Object(_))));
calls += 1;
let tell_active =
dispatcher.dispatch_json(rpc_request(RpcMethod::Aria2TellActive, Vec::new()));
assert!(matches!(tell_active.result, Some(RpcValue::Array(_))));
calls += 1;
let tell_global =
dispatcher.dispatch_json(rpc_request(RpcMethod::Aria2TellGlobalStat, Vec::new()));
assert!(matches!(tell_global.result, Some(RpcValue::Object(_))));
calls += 1;
}
calls
}
fn run_shared_runtime_fairness_pressure(
dispatcher: &mut InProcessRpcDispatcher,
gids: &[String],
scenario: SharedRuntimeFairnessScenario,
) -> usize {
let mut calls = 0usize;
let initial_shared_download_speed =
(scenario.global_download_limit / scenario.task_count as u64).max(1);
let initial_shared_upload_speed =
(scenario.global_upload_limit / scenario.task_count as u64).max(1);
let constrained_initial_download_speed = initial_shared_download_speed
.min(scenario.constrained_per_download_limit)
.max(1);
let constrained_initial_upload_speed = initial_shared_upload_speed
.min(scenario.constrained_per_upload_limit)
.max(1);
let rebalanced_active_count = scenario.task_count.saturating_sub(1).max(1);
let rebalanced_download_speed = (scenario.global_download_limit
/ rebalanced_active_count as u64)
.min(scenario.default_per_download_limit)
.max(1);
let rebalanced_upload_speed = (scenario.global_upload_limit / rebalanced_active_count as u64)
.min(scenario.default_per_upload_limit)
.max(1);
for round in 0..scenario.rounds {
let active_slice = if round == 0 { gids } else { &gids[1..] };
for gid in active_slice {
dispatcher
.apply_bt_runtime_tick(gid, 128, 64, 5_000, 2_000, 0, 0, false, Some(6))
.expect("shared-runtime fairness tick should succeed");
}
for (index, gid) in gids.iter().enumerate() {
let tell_status = dispatcher.dispatch_json(rpc_request(
RpcMethod::Aria2TellStatus,
vec![RpcValue::String(gid.clone())],
));
match tell_status.result {
Some(RpcValue::Object(payload)) => {
if round > 0 && index == 0 {
assert_eq!(
payload.get("status"),
Some(&RpcValue::String("complete".to_owned()))
);
} else {
let expected_download_speed = if index == 0 {
constrained_initial_download_speed
} else if round == 0 {
initial_shared_download_speed
.min(scenario.default_per_download_limit)
.max(1)
} else {
rebalanced_download_speed
};
let expected_upload_speed = if index == 0 {
constrained_initial_upload_speed
} else if round == 0 {
initial_shared_upload_speed
.min(scenario.default_per_upload_limit)
.max(1)
} else {
rebalanced_upload_speed
};
assert_eq!(
payload.get("downloadSpeed"),
Some(&RpcValue::String(expected_download_speed.to_string()))
);
assert_eq!(
payload.get("uploadSpeed"),
Some(&RpcValue::String(expected_upload_speed.to_string()))
);
}
calls += 1;
}
other => panic!(
"unexpected tellStatus payload in shared-runtime fairness bench: {other:?}"
),
}
}
let tell_global =
dispatcher.dispatch_json(rpc_request(RpcMethod::Aria2TellGlobalStat, Vec::new()));
match tell_global.result {
Some(RpcValue::Object(payload)) => {
let (expected_download_speed, expected_upload_speed) = if round == 0 {
(
constrained_initial_download_speed.saturating_add(
initial_shared_download_speed
.min(scenario.default_per_download_limit)
.max(1)
.saturating_mul(rebalanced_active_count as u64),
),
constrained_initial_upload_speed.saturating_add(
initial_shared_upload_speed
.min(scenario.default_per_upload_limit)
.max(1)
.saturating_mul(rebalanced_active_count as u64),
),
)
} else {
(
constrained_initial_download_speed
.saturating_add(
rebalanced_download_speed
.saturating_mul(rebalanced_active_count as u64),
)
.min(scenario.global_download_limit),
constrained_initial_upload_speed
.saturating_add(
rebalanced_upload_speed
.saturating_mul(rebalanced_active_count as u64),
)
.min(scenario.global_upload_limit),
)
};
assert_eq!(
payload.get("downloadSpeed"),
Some(&RpcValue::String(expected_download_speed.to_string()))
);
assert_eq!(
payload.get("uploadSpeed"),
Some(&RpcValue::String(expected_upload_speed.to_string()))
);
calls += 1;
}
other => panic!(
"unexpected tellGlobalStat payload in shared-runtime fairness bench: {other:?}"
),
}
if round == 0 {
dispatcher
.mark_complete(&gids[0])
.expect("shared-runtime fairness benchmark should complete one gid");
}
}
calls
}
pub(super) fn bench_tell_status_pressure(c: &mut Criterion) {
let mut group = c.benchmark_group("rpc_tell_status_pressure");
for scenario in [
PressureScenario {
task_count: 64,
rounds: 4,
split: 4,
max_connections_per_server: 4,
},
PressureScenario {
task_count: 128,
rounds: 4,
split: 8,
max_connections_per_server: 8,
},
] {
group.throughput(Throughput::Elements(
(scenario.task_count * scenario.rounds) as u64,
));
group.bench_with_input(
BenchmarkId::new("tell_status", scenario.task_count),
&scenario,
|b, &scenario| {
b.iter_batched(
|| seed_pressure_dispatcher(scenario),
|(mut dispatcher, gids)| {
let calls =
run_tell_status_pressure(&mut dispatcher, &gids, scenario.rounds);
assert_eq!(calls, scenario.task_count * scenario.rounds);
},
criterion::BatchSize::SmallInput,
);
},
);
}
group.finish();
}
pub(super) fn bench_mixed_rpc_pressure(c: &mut Criterion) {
let mut group = c.benchmark_group("rpc_mixed_pressure");
for scenario in [
PressureScenario {
task_count: 96,
rounds: 6,
split: 6,
max_connections_per_server: 6,
},
PressureScenario {
task_count: 192,
rounds: 6,
split: 8,
max_connections_per_server: 8,
},
] {
group.throughput(Throughput::Elements((scenario.rounds * 3) as u64));
group.bench_with_input(
BenchmarkId::new("mixed_rpc", scenario.task_count),
&scenario,
|b, &scenario| {
b.iter_batched(
|| seed_pressure_dispatcher(scenario),
|(mut dispatcher, gids)| {
let calls = run_mixed_rpc_pressure(&mut dispatcher, &gids, scenario.rounds);
assert_eq!(calls, scenario.rounds * 3);
},
criterion::BatchSize::SmallInput,
);
},
);
}
group.finish();
}
pub(super) fn bench_resource_limit_pressure(c: &mut Criterion) {
let mut group = c.benchmark_group("rpc_speed_limit_pressure");
for scenario in [
ResourceLimitScenario {
task_count: 32,
rounds: 6,
global_download_limit: 2_560,
per_download_limit: 160,
global_upload_limit: 1_280,
per_upload_limit: 80,
disk_cache_bytes: 8 * 1024 * 1024,
},
ResourceLimitScenario {
task_count: 64,
rounds: 6,
global_download_limit: 5_120,
per_download_limit: 120,
global_upload_limit: 2_560,
per_upload_limit: 60,
disk_cache_bytes: 16 * 1024 * 1024,
},
] {
group.throughput(Throughput::Elements((scenario.rounds * 2) as u64));
group.bench_with_input(
BenchmarkId::new("speed_limit", scenario.task_count),
&scenario,
|b, &scenario| {
b.iter_batched(
|| seed_limited_dispatcher(scenario),
|(mut dispatcher, gids)| {
let calls = run_resource_limit_pressure(&mut dispatcher, &gids, scenario);
assert_eq!(calls, scenario.rounds * 2);
},
criterion::BatchSize::SmallInput,
);
},
);
}
group.finish();
}
pub(super) fn bench_shared_runtime_fairness_pressure(c: &mut Criterion) {
let mut group = c.benchmark_group("rpc_shared_runtime_fairness_pressure");
for scenario in [
SharedRuntimeFairnessScenario {
label: "three_way_rebalance",
task_count: 3,
rounds: 4,
global_download_limit: 1_200,
default_per_download_limit: 900,
constrained_per_download_limit: 250,
global_upload_limit: 600,
default_per_upload_limit: 500,
constrained_per_upload_limit: 120,
},
SharedRuntimeFairnessScenario {
label: "four_way_rebalance",
task_count: 4,
rounds: 4,
global_download_limit: 1_600,
default_per_download_limit: 900,
constrained_per_download_limit: 220,
global_upload_limit: 800,
default_per_upload_limit: 500,
constrained_per_upload_limit: 100,
},
] {
group.throughput(Throughput::Elements(
(scenario.task_count * scenario.rounds) as u64,
));
group.bench_with_input(
BenchmarkId::new("shared_runtime_fairness", scenario.label),
&scenario,
|b, &scenario| {
b.iter_batched(
|| seed_shared_runtime_fairness_dispatcher(scenario),
|(mut dispatcher, gids)| {
let calls =
run_shared_runtime_fairness_pressure(&mut dispatcher, &gids, scenario);
assert_eq!(calls, scenario.rounds * (scenario.task_count + 1));
},
criterion::BatchSize::SmallInput,
);
},
);
}
group.finish();
}