Stabilize flat-network binpacking assignments

This commit is contained in:
Herbert Wolverson committed 2026-07-21 15:12:10 -05:00
1 parent a920962b9d
commit d4fc3d255d
10 files changed
+1108 -80

No files matched your search

+180 -17
View File
@@ -736,6 +736,133 @@ def planner_circuit_identity_key(circuit):
return circuit_id
def flat_binpacking_enabled(network, binpacking_enabled):
if not binpacking_enabled:
return False
if loaded_network_is_flat(network):
return True
if not isinstance(network, dict) or not network:
return False
expected_bucket_names = {
"Generated_PN_" + str(index + 1) for index in range(len(network))
}
if set(network) != expected_bucket_names:
return False
for index in range(len(network)):
bucket_name = "Generated_PN_" + str(index + 1)
bucket = network.get(bucket_name)
expected_bucket_id = "libreqos:generated:flat:bucket:" + str(index)
if (
not isinstance(bucket, dict)
or bucket.get("id") != expected_bucket_id
or bucket.get("children", {}) != {}
):
return False
return True
def planner_top_level_item_key(circuit, flat_network):
if flat_network:
return "flat_circuit:" + planner_circuit_identity_key(circuit)
if 'idForCircuitsWithoutParentNodes' not in circuit:
raise ValueError("Missing transient planner ID for unparented circuit")
return str(circuit['idForCircuitsWithoutParentNodes'])
def reusable_identity_state(state, reuse_saved_identities):
if not reuse_saved_identities or not isinstance(state, dict):
return {}, {}
sites = state.get('sites', {})
circuits = state.get('circuits', {})
return (
sites if isinstance(sites, dict) else {},
circuits if isinstance(circuits, dict) else {},
)
def generated_parent_node_from_queue_key(queue_key, queues_available):
if not isinstance(queue_key, str) or queues_available <= 0:
return None
queue_index = _parse_int_token(queue_key.removeprefix("CpueQueue"))
if not queue_key.startswith("CpueQueue") or queue_index is None:
return None
if queue_index < 0 or queue_index >= queues_available:
return None
return "Generated_PN_" + str(queue_index + 1)
def plan_flat_generated_parent_assignments(
items,
generated_parent_nodes,
previous_assignments,
last_change_ts,
now_ts,
move_budget_per_run,
):
queues_available = len(generated_parent_nodes)
item_ids = {str(item["id"]) for item in items}
rust_previous_assignments = {
item_id: generated_parent_node_queue_key(parent_name, queues_available)
for item_id, parent_name in previous_assignments.items()
}
rust_previous_assignments = {
item_id: queue_key
for item_id, queue_key in rust_previous_assignments.items()
if queue_key is not None
}
try:
plan_result = plan_top_level_cpu_bins(
items,
queues_available,
prev_assign=rust_previous_assignments,
last_change_ts=last_change_ts,
now_ts=now_ts,
mode="stable_greedy",
move_budget_per_run=move_budget_per_run,
cooldown_seconds=3600.0,
hysteresis_threshold=0.03,
)
changed = list(plan_result.get("changed", []) or [])
assignments = {
item_id: parent_name
for item_id, queue_key in (plan_result.get("assignment", {}) or {}).items()
if (
parent_name := generated_parent_node_from_queue_key(
queue_key,
queues_available,
)
) is not None
}
if set(assignments) != item_ids:
raise ValueError("Shared Rust flat planner returned incomplete assignments")
return assignments, changed
except Exception as e:
warnings.warn(
f"Shared Rust flat planner failed ({e}); preserving prior assignments.",
stacklevel=2,
)
assignments = dict(previous_assignments)
weights_by_item = {str(item["id"]): float(item["weight"]) for item in items}
bin_loads = {parent_name: 0.0 for parent_name in generated_parent_nodes}
for item_id, parent_name in assignments.items():
bin_loads[parent_name] += weights_by_item.get(item_id, 1.0)
unassigned = [
(item_id, weight)
for item_id, weight in weights_by_item.items()
if item_id not in assignments
]
unassigned.sort(key=lambda item: (-item[1], item[0]))
for item_id, weight in unassigned:
target_parent = min(
bin_loads.items(),
key=lambda entry: (entry[1], entry[0]),
)[0]
assignments[item_id] = target_parent
bin_loads[target_parent] += weight
return assignments, []
def calculateR2q(maxRateInMbps):
# So we've learned that r2q defaults to 10, and is used to calculate quantum. Quantum is rateInBytes/r2q by
# default. This default gives errors at high rates, and tc clamps the quantum to 200000. Setting a high quantum
@@ -1404,6 +1531,7 @@ def refreshShapers():
# Flat networks ({}) don't require ParentNode entries. Treat every circuit as
# unparented so they can be distributed across generated parent nodes / CPUs.
flat_network = loaded_network_is_flat(network)
binpacking_enabled = use_bin_packing_to_balance_cpu()
# Virtual Nodes (logical-only): build a physical shaping topology that skips them,
# while leaving ShapedDevices.csv (and monitoring) unchanged.
@@ -1426,6 +1554,11 @@ def refreshShapers():
# Avoid bloating queuingStructure.json when there are no virtual nodes.
logical_to_physical_node = {}
reuse_flat_binpacking_identities = flat_binpacking_enabled(
network,
binpacking_enabled,
)
# Re-map circuits that are directly parented to a virtual node to the nearest real ancestor (milestone c).
if not flat_network and len(virtual_nodes) > 0 and isinstance(logical_to_physical_node, dict):
next_id = max(dictForCircuitsWithoutParentNodes.keys(), default=-1) + 1
@@ -1531,7 +1664,7 @@ def refreshShapers():
# Planner/device weights (fetched only when planner/binpacking is enabled).
# When disabled, we keep this empty and fall back to rate-based weights later.
weight_by_circuit_id = {}
if use_bin_packing_to_balance_cpu():
if binpacking_enabled:
print("Using internal planner to sort circuits by CPU core")
# Build item list with weights for circuits lacking a ParentNode
items = []
@@ -1549,9 +1682,10 @@ def refreshShapers():
pass
for circuit in subscriberCircuits:
if circuit.get('ParentNode') == 'none' and 'idForCircuitsWithoutParentNodes' in circuit:
item_id = circuit['idForCircuitsWithoutParentNodes']
transient_item_id = circuit['idForCircuitsWithoutParentNodes']
item_id = planner_top_level_item_key(circuit, flat_network)
# Prefer provided weights; default to 1.0
w = dictForCircuitsWithoutParentNodes.get(item_id, 1.0)
w = dictForCircuitsWithoutParentNodes.get(transient_item_id, 1.0)
# If a specific circuit weight exists, prefer it
if 'circuitID' in circuit and str(circuit['circuitID']) in weight_by_circuit_id:
w = weight_by_circuit_id[str(circuit['circuitID'])]
@@ -1570,14 +1704,18 @@ def refreshShapers():
capacities = {pn: 1.0 for pn in generatedPNs}
# Load planner state
try:
import bin_planner
except ImportError:
bin_planner = None
bin_planner = None
if not flat_network:
try:
import bin_planner
except ImportError:
pass
# Store planner state directly in lqos_directory (no hidden subdirs)
state_path = get_planner_state_path()
state = {}
if bin_planner is not None:
if flat_network:
state = load_planner_state(state_path, None)
elif bin_planner is not None:
state = load_planner_state(state_path, bin_planner)
now_ts = time.time()
prev_assign = {}
@@ -1598,14 +1736,27 @@ def refreshShapers():
"alpha": 0.1,
"hysteresis_threshold": 0.03,
"cooldown_seconds": 3600,
"move_budget_per_run": max(1, min(32, int(0.01 * max(1, len(items))))),
"move_budget_per_run": (
1
if flat_network
else max(1, min(32, int(0.01 * max(1, len(items)))))
),
"salt": state.get("salt", "default_salt") if isinstance(state, dict) else "default_salt",
"last_change_ts_by_item": last_change_ts,
}
if observe_mode:
params["move_budget_per_run"] = 0
if bin_planner is not None:
if flat_network:
assignments, changed = plan_flat_generated_parent_assignments(
items,
generatedPNs,
prev_assign,
last_change_ts,
now_ts,
params["move_budget_per_run"],
)
elif bin_planner is not None:
assignments, changed = bin_planner.plan_assignments(
items, bins_list, capacities, prev_assign, now_ts, params
)
@@ -1624,21 +1775,24 @@ def refreshShapers():
# Apply assignments to circuits
for circuit in subscriberCircuits:
if circuit.get('ParentNode') == 'none' and 'idForCircuitsWithoutParentNodes' in circuit:
item_id = circuit['idForCircuitsWithoutParentNodes']
item_key = str(item_id)
item_key = planner_top_level_item_key(circuit, flat_network)
if item_key in assignments:
circuit['ParentNode'] = assignments[item_key]
circuit['ParentNodeID'] = ''
circuit['effectiveParentNodeID'] = ''
# Update and save state
if bin_planner is not None and isinstance(state, dict):
if (flat_network or bin_planner is not None) and isinstance(state, dict):
if state.get("salt") is None:
state["salt"] = "default_salt"
if "assignments" not in state or not isinstance(state["assignments"], dict):
state["assignments"] = {}
if "last_change_ts" not in state or not isinstance(state["last_change_ts"], dict):
state["last_change_ts"] = {}
if flat_network:
for stale_item_id in set(state["assignments"]) - item_ids:
state["assignments"].pop(stale_item_id, None)
state["last_change_ts"].pop(stale_item_id, None)
for iid, b in assignments.items():
# record last change time if changed
if iid in changed:
@@ -1646,7 +1800,11 @@ def refreshShapers():
state["assignments"][iid] = b
try:
print(f"Saving planner state to {state_path} (generated PNs)")
save_planner_state(state, state_path, bin_planner)
save_planner_state(
state,
state_path,
None if flat_network else bin_planner,
)
except Exception as e:
warnings.warn(f"Failed to save planner state at {state_path}: {e}", stacklevel=2)
@@ -1875,7 +2033,7 @@ def refreshShapers():
return keys
# If we're in binpacking mode, we need to sort the network structure a bit
if use_bin_packing_to_balance_cpu() and not flat_network:
if binpacking_enabled and not flat_network:
# Binpacking is an Insight feature; if Insight is not enabled/licensed, fall back to
# deterministic round-robin placement so "virtual node promotion" can still spread
# the physical tree across CPUs.
@@ -2114,11 +2272,16 @@ def refreshShapers():
collect_identity_planner_inputs(network, 0, 1)
previous_site_state, previous_circuit_state = reusable_identity_state(
state,
reuse_flat_binpacking_identities,
)
identity_plan = plan_class_identities(
planner_site_inputs,
planner_circuit_groups,
site_state={},
circuit_state={},
site_state=previous_site_state,
circuit_state=previous_circuit_state,
stick_offset=stickOffset,
circuit_padding=CIRCUIT_PADDING,
)
+504 -56
View File
@@ -201,6 +201,103 @@ fn next_free_minor(start_minor: u32, reserved: &BTreeSet<u32>) -> u32 {
candidate
}
fn valid_class_minor(class_minor: u16) -> Option<u32> {
let class_minor = u32::from(class_minor);
(3..u32::from(u16::MAX))
.contains(&class_minor)
.then_some(class_minor)
}
fn expected_class_majors(queue: u32, stick_offset: u16) -> Option<(u16, u16)> {
let class_major = u16::try_from(queue).ok()?;
(class_major > 0).then_some((class_major, class_major.saturating_add(stick_offset)))
}
fn reserve_reusable_site_minors(
sites: &[SiteIdentityInput],
previous_sites: &BTreeMap<String, PlannerSiteIdentityState>,
stick_offset: u16,
reserved: &mut PlannerMinorReservations,
) -> BTreeMap<String, u32> {
let current_sites = sites
.iter()
.map(|site| (site.site_key.as_str(), site))
.collect::<BTreeMap<_, _>>();
let mut reusable = BTreeMap::new();
for (site_key, site) in current_sites {
let Some(stored) = previous_sites.get(site_key) else {
continue;
};
let Some(class_minor) = valid_class_minor(stored.class_minor) else {
continue;
};
let Some((class_major, up_class_major)) = expected_class_majors(site.queue, stick_offset)
else {
continue;
};
if stored.queue != site.queue
|| stored.parent_path != site.parent_path
|| stored.class_major != class_major
|| stored.up_class_major != up_class_major
{
continue;
}
let occupied = reserved.entry(site.queue).or_default();
if occupied.insert(class_minor) {
reusable.insert(site_key.to_string(), class_minor);
}
}
reusable
}
fn reserve_reusable_circuit_minors(
circuit_groups: &[CircuitIdentityGroupInput],
previous_circuits: &BTreeMap<String, PlannerCircuitIdentityState>,
stick_offset: u16,
reserved: &mut PlannerMinorReservations,
) -> BTreeMap<String, u32> {
let current_circuits = circuit_groups
.iter()
.flat_map(|group| {
group
.circuit_ids
.iter()
.map(move |circuit_id| (circuit_id.as_str(), group))
})
.collect::<BTreeMap<_, _>>();
let mut reusable = BTreeMap::new();
for (circuit_id, group) in current_circuits {
let Some(stored) = previous_circuits.get(circuit_id) else {
continue;
};
let Some(class_minor) = valid_class_minor(stored.class_minor) else {
continue;
};
let Some((class_major, up_class_major)) = expected_class_majors(group.queue, stick_offset)
else {
continue;
};
if stored.queue != group.queue
|| stored.parent_node != group.parent_node
|| stored.class_major != class_major
|| stored.up_class_major != up_class_major
{
continue;
}
let occupied = reserved.entry(group.queue).or_default();
if occupied.insert(class_minor) {
reusable.insert(circuit_id.to_string(), class_minor);
}
}
reusable
}
/// Builds reservation maps for identities that are not part of the current planning scope.
pub fn build_class_identity_reservations(
sites: &[SiteIdentityInput],
@@ -220,10 +317,12 @@ pub fn build_class_identity_reservations(
if planned_site_keys.contains(site_key.as_str()) {
continue;
}
reserved_site_minors
.entry(stored.queue)
.or_default()
.insert(stored.class_minor as u32);
if let Some(class_minor) = valid_class_minor(stored.class_minor) {
reserved_site_minors
.entry(stored.queue)
.or_default()
.insert(class_minor);
}
}
let mut reserved_circuit_minors: PlannerMinorReservations = BTreeMap::new();
@@ -231,10 +330,12 @@ pub fn build_class_identity_reservations(
if planned_circuit_ids.contains(circuit_id.as_str()) {
continue;
}
reserved_circuit_minors
.entry(stored.queue)
.or_default()
.insert(stored.class_minor as u32);
if let Some(class_minor) = valid_class_minor(stored.class_minor) {
reserved_circuit_minors
.entry(stored.queue)
.or_default()
.insert(class_minor);
}
}
(reserved_site_minors, reserved_circuit_minors)
@@ -334,7 +435,7 @@ pub fn plan_top_level_assignments(
let rr_assignment = round_robin_assign(items, bins);
let greedy_assignment = greedy_assign(items, bins);
let (mut assignment, planner_used) = match params.mode {
let (assignment, planner_used) = match params.mode {
TopLevelPlannerMode::RoundRobin => (rr_assignment.clone(), false),
TopLevelPlannerMode::Greedy => (greedy_assignment.clone(), false),
TopLevelPlannerMode::StableGreedy => {
@@ -361,11 +462,11 @@ pub fn plan_top_level_assignments(
});
let mut moves_used = 0usize;
for id in weighted_ids {
let Some(current_bin) = assignment.get(&id).cloned() else {
for id in &weighted_ids {
let Some(current_bin) = assignment.get(id).cloned() else {
continue;
};
let Some(target_bin) = greedy_assignment.get(&id).cloned() else {
let Some(target_bin) = greedy_assignment.get(id).cloned() else {
continue;
};
if current_bin == target_bin {
@@ -374,36 +475,53 @@ pub fn plan_top_level_assignments(
if moves_used >= params.move_budget_per_run {
continue;
}
let last_changed = last_change_ts.get(&id).copied().unwrap_or(0.0);
let last_changed = last_change_ts.get(id).copied().unwrap_or(0.0);
if now_ts - last_changed < params.cooldown_seconds {
continue;
}
let current_load = loads.get(&current_bin).copied().unwrap_or(0.0);
let target_load = loads.get(&target_bin).copied().unwrap_or(0.0);
if current_load - target_load <= min_improvement {
let weight = item_weights.get(id).copied().unwrap_or(1.0);
let imbalance_before = current_load - target_load;
let imbalance_after = ((current_load - weight) - (target_load + weight)).abs();
if imbalance_before - imbalance_after <= min_improvement {
continue;
}
let weight = item_weights.get(&id).copied().unwrap_or(1.0);
if let Some(load) = loads.get_mut(&current_bin) {
*load -= weight;
}
if let Some(load) = loads.get_mut(&target_bin) {
*load += weight;
}
assignment.insert(id, target_bin);
assignment.insert(id.clone(), target_bin);
moves_used += 1;
}
let used_bins = assignment.values().collect::<BTreeSet<_>>();
if bins.len() > 1
&& assignment.len() > 1
&& used_bins.len() <= 1
&& moves_used < params.move_budget_per_run
{
for id in weighted_ids {
let Some(current_bin) = assignment.get(&id) else {
continue;
};
let Some(target_bin) = greedy_assignment.get(&id) else {
continue;
};
if current_bin != target_bin {
assignment.insert(id, target_bin.clone());
break;
}
}
}
(assignment, true)
}
};
let used_bins: std::collections::BTreeSet<&String> = assignment.values().collect();
if bins.len() > 1 && assignment.len() > 1 && used_bins.len() <= 1 {
assignment = greedy_assignment.clone();
}
let mut changed: Vec<String> = assignment
.iter()
.filter_map(|(id, bin)| {
@@ -433,39 +551,49 @@ pub fn plan_class_identities_with_constraints(
constraints: &ClassIdentityPlannerConstraints,
) -> ClassIdentityPlannerOutput {
let mut reserved_site_minors = constraints.reserved_site_minors.clone();
for (queue, minors) in &constraints.reserved_circuit_minors {
reserved_site_minors
.entry(*queue)
.or_default()
.extend(minors.iter().copied());
}
let reusable_site_minors = reserve_reusable_site_minors(
sites,
previous_sites,
constraints.stick_offset,
&mut reserved_site_minors,
);
let reusable_circuit_minors = reserve_reusable_circuit_minors(
circuit_groups,
previous_circuits,
constraints.stick_offset,
&mut reserved_site_minors,
);
let mut next_site_minor_by_queue: BTreeMap<u32, u32> = BTreeMap::new();
let mut site_assignments = Vec::with_capacity(sites.len());
let mut site_state = BTreeMap::new();
for (queue, reserved) in &reserved_site_minors {
let next_minor = next_site_minor_by_queue
.entry(*queue)
.or_insert(constraints.site_minor_start.max(3));
if let Some(last_reserved) = reserved.iter().next_back().copied() {
*next_minor = (*next_minor).max(last_reserved.saturating_add(1));
}
}
for site in sites {
let reserved = reserved_site_minors.entry(site.queue).or_default();
let next_minor = next_site_minor_by_queue
.entry(site.queue)
.or_insert(constraints.site_minor_start.max(3));
let reuse_minor = previous_sites.get(&site.site_key).and_then(|stored| {
(stored.queue == site.queue
&& stored.parent_path == site.parent_path
&& !reserved.contains(&(stored.class_minor as u32)))
.then_some(stored.class_minor as u32)
});
let reuse_minor = reusable_site_minors.get(&site.site_key).copied();
let class_minor_u32 = reuse_minor.unwrap_or_else(|| next_free_minor(*next_minor, reserved));
let class_minor_u32 = if let Some(reused_minor) = reuse_minor {
reused_minor
} else {
let allocated_minor = next_free_minor(*next_minor, reserved);
*next_minor = allocated_minor.saturating_add(1);
if site.has_children {
*next_minor = (*next_minor).saturating_add(1);
}
allocated_minor
};
let class_minor = u16::try_from(class_minor_u32).unwrap_or(u16::MAX);
let class_major = u16::try_from(site.queue).unwrap_or(u16::MAX);
let up_class_major = class_major.saturating_add(constraints.stick_offset);
reserved.insert(class_minor_u32);
*next_minor = (*next_minor).max(class_minor_u32).saturating_add(1);
if site.has_children {
*next_minor = (*next_minor).saturating_add(1);
}
let assigned = SiteIdentityAssignment {
site_key: site.site_key.clone(),
@@ -489,12 +617,6 @@ pub fn plan_class_identities_with_constraints(
}
let mut reserved_circuit_minors = reserved_site_minors.clone();
for (queue, extras) in &constraints.reserved_circuit_minors {
reserved_circuit_minors
.entry(*queue)
.or_default()
.extend(extras.iter().copied());
}
let mut next_circuit_minor_by_queue = BTreeMap::new();
for (queue, reserved) in &reserved_circuit_minors {
let start = next_site_minor_by_queue
@@ -515,19 +637,18 @@ pub fn plan_class_identities_with_constraints(
.entry(group.queue)
.or_insert_with(|| next_free_minor(constraints.circuit_minor_start.max(3), reserved));
for circuit_id in &group.circuit_ids {
let reuse_minor = previous_circuits.get(circuit_id).and_then(|stored| {
(stored.queue == group.queue
&& stored.parent_node == group.parent_node
&& !reserved.contains(&(stored.class_minor as u32)))
.then_some(stored.class_minor as u32)
});
let class_minor_u32 =
reuse_minor.unwrap_or_else(|| next_free_minor(*next_minor, reserved));
let reuse_minor = reusable_circuit_minors.get(circuit_id).copied();
let class_minor_u32 = if let Some(reused_minor) = reuse_minor {
reused_minor
} else {
let allocated_minor = next_free_minor(*next_minor, reserved);
*next_minor = allocated_minor.saturating_add(1);
allocated_minor
};
let class_minor = u16::try_from(class_minor_u32).unwrap_or(u16::MAX);
let class_major = u16::try_from(group.queue).unwrap_or(u16::MAX);
let up_class_major = class_major.saturating_add(constraints.stick_offset);
reserved.insert(class_minor_u32);
*next_minor = (*next_minor).max(class_minor_u32).saturating_add(1);
circuit_assignments.push(CircuitIdentityAssignment {
circuit_id: circuit_id.clone(),
@@ -563,7 +684,15 @@ pub fn plan_class_identities_with_constraints(
.get(&queue)
.copied()
.unwrap_or(3);
last_used_minor_by_queue.insert(queue, site_next.max(circuit_next));
let highest_reserved_next = reserved_circuit_minors
.get(&queue)
.and_then(|reserved| reserved.iter().next_back().copied())
.unwrap_or(2)
.saturating_add(1);
last_used_minor_by_queue.insert(
queue,
site_next.max(circuit_next).max(highest_reserved_next),
);
}
ClassIdentityPlannerOutput {
@@ -726,6 +855,99 @@ mod tests {
assert!(result.planner_used);
}
#[test]
fn stable_greedy_does_not_make_balance_worse() {
let items = vec![
TopLevelPlannerItem {
id: "a".to_string(),
weight: 2.0,
},
TopLevelPlannerItem {
id: "b".to_string(),
weight: 2.0,
},
TopLevelPlannerItem {
id: "c".to_string(),
weight: 3.0,
},
TopLevelPlannerItem {
id: "d".to_string(),
weight: 5.0,
},
TopLevelPlannerItem {
id: "e".to_string(),
weight: 3.0,
},
];
let prev_assign = BTreeMap::from([
("a".to_string(), "CpueQueue1".to_string()),
("b".to_string(), "CpueQueue0".to_string()),
("c".to_string(), "CpueQueue1".to_string()),
("d".to_string(), "CpueQueue0".to_string()),
]);
let result = plan_top_level_assignments(
&items,
&bins(),
&prev_assign,
&BTreeMap::new(),
10_000.0,
&TopLevelPlannerParams {
mode: TopLevelPlannerMode::StableGreedy,
cooldown_seconds: 0.0,
move_budget_per_run: 1,
hysteresis_threshold: 0.03,
},
);
assert!(result.changed.is_empty());
for (circuit_id, previous_bin) in prev_assign {
assert_eq!(result.assignment[&circuit_id], previous_bin);
}
}
#[test]
fn stable_greedy_single_bin_recovery_respects_move_budget() {
let items = (0..6)
.map(|index| TopLevelPlannerItem {
id: format!("circuit-{index}"),
weight: 1.0,
})
.collect::<Vec<_>>();
let prev_assign = items
.iter()
.map(|item| (item.id.clone(), "CpueQueue0".to_string()))
.collect::<BTreeMap<_, _>>();
let last_change_ts = items
.iter()
.map(|item| (item.id.clone(), 999.0))
.collect::<BTreeMap<_, _>>();
let result = plan_top_level_assignments(
&items,
&bins(),
&prev_assign,
&last_change_ts,
1_000.0,
&TopLevelPlannerParams {
mode: TopLevelPlannerMode::StableGreedy,
cooldown_seconds: 3_600.0,
move_budget_per_run: 1,
hysteresis_threshold: 0.0,
},
);
assert_eq!(result.changed.len(), 1);
assert_eq!(
result
.assignment
.values()
.collect::<std::collections::BTreeSet<_>>()
.len(),
2
);
}
#[test]
fn class_identity_planner_reuses_site_and_circuit_minors() {
let site_inputs = vec![
@@ -965,4 +1187,230 @@ mod tests {
assert_eq!(result.sites.len(), 1);
assert!(result.sites[0].class_minor >= 0x1000);
}
#[test]
fn class_identity_planner_protects_existing_circuits_from_sorted_insertions() {
let circuit_groups = vec![CircuitIdentityGroupInput {
parent_node: "flat-bucket".to_string(),
queue: 1,
circuit_ids: vec![
"a-new".to_string(),
"m-existing".to_string(),
"z-existing".to_string(),
],
}];
let previous_circuits = BTreeMap::from([
(
"m-existing".to_string(),
PlannerCircuitIdentityState {
class_minor: 10,
queue: 1,
parent_node: "flat-bucket".to_string(),
class_major: 1,
up_class_major: 1,
},
),
(
"z-existing".to_string(),
PlannerCircuitIdentityState {
class_minor: 11,
queue: 1,
parent_node: "flat-bucket".to_string(),
class_major: 1,
up_class_major: 1,
},
),
]);
let result = plan_class_identities(
&[],
&circuit_groups,
&BTreeMap::new(),
&previous_circuits,
0,
0,
);
assert_eq!(result.circuit_state["m-existing"].class_minor, 10);
assert_eq!(result.circuit_state["z-existing"].class_minor, 11);
assert_eq!(result.circuit_state["a-new"].class_minor, 3);
let repeated = plan_class_identities(
&[],
&circuit_groups,
&BTreeMap::new(),
&result.circuit_state,
0,
0,
);
assert_eq!(repeated.circuit_state, result.circuit_state);
assert_eq!(
repeated.last_used_minor_by_queue,
result.last_used_minor_by_queue
);
}
#[test]
fn class_identity_planner_reassigns_only_the_circuit_that_moved() {
let circuit_groups = vec![
CircuitIdentityGroupInput {
parent_node: "bucket-a".to_string(),
queue: 1,
circuit_ids: vec!["stable".to_string()],
},
CircuitIdentityGroupInput {
parent_node: "bucket-b".to_string(),
queue: 2,
circuit_ids: vec!["moved".to_string()],
},
];
let previous_circuits = BTreeMap::from([
(
"stable".to_string(),
PlannerCircuitIdentityState {
class_minor: 20,
queue: 1,
parent_node: "bucket-a".to_string(),
class_major: 1,
up_class_major: 1,
},
),
(
"moved".to_string(),
PlannerCircuitIdentityState {
class_minor: 21,
queue: 1,
parent_node: "bucket-a".to_string(),
class_major: 1,
up_class_major: 1,
},
),
]);
let result = plan_class_identities(
&[],
&circuit_groups,
&BTreeMap::new(),
&previous_circuits,
0,
0,
);
assert_eq!(result.circuit_state["stable"], previous_circuits["stable"]);
assert_ne!(result.circuit_state["moved"], previous_circuits["moved"]);
assert_eq!(result.circuit_state["moved"].queue, 2);
assert_eq!(result.circuit_state["moved"].parent_node, "bucket-b");
}
#[test]
fn class_identity_planner_rejects_duplicate_stale_and_invalid_saved_identities() {
let circuit_groups = vec![CircuitIdentityGroupInput {
parent_node: "flat-bucket".to_string(),
queue: 1,
circuit_ids: vec![
"duplicate-a".to_string(),
"duplicate-b".to_string(),
"stale-parent".to_string(),
"reserved-minor".to_string(),
],
}];
let previous_circuits = BTreeMap::from([
(
"duplicate-a".to_string(),
PlannerCircuitIdentityState {
class_minor: 10,
queue: 1,
parent_node: "flat-bucket".to_string(),
class_major: 1,
up_class_major: 1,
},
),
(
"duplicate-b".to_string(),
PlannerCircuitIdentityState {
class_minor: 10,
queue: 1,
parent_node: "flat-bucket".to_string(),
class_major: 1,
up_class_major: 1,
},
),
(
"stale-parent".to_string(),
PlannerCircuitIdentityState {
class_minor: 11,
queue: 1,
parent_node: "old-bucket".to_string(),
class_major: 1,
up_class_major: 1,
},
),
(
"reserved-minor".to_string(),
PlannerCircuitIdentityState {
class_minor: u16::MAX,
queue: 1,
parent_node: "flat-bucket".to_string(),
class_major: 1,
up_class_major: 1,
},
),
]);
let result = plan_class_identities(
&[],
&circuit_groups,
&BTreeMap::new(),
&previous_circuits,
0,
0,
);
let assigned_minors = result
.circuits
.iter()
.map(|circuit| circuit.class_minor)
.collect::<std::collections::BTreeSet<_>>();
assert_eq!(result.circuits.len(), assigned_minors.len());
assert_eq!(result.circuit_state["duplicate-a"].class_minor, 10);
assert_ne!(result.circuit_state["duplicate-b"].class_minor, 10);
assert_ne!(result.circuit_state["stale-parent"].class_minor, 11);
assert_ne!(result.circuit_state["reserved-minor"].class_minor, u16::MAX);
assert!(
assigned_minors
.iter()
.all(|minor| (3..u16::MAX).contains(minor))
);
}
#[test]
fn class_identity_planner_fills_holes_below_high_saved_minors() {
let circuit_groups = vec![CircuitIdentityGroupInput {
parent_node: "flat-bucket".to_string(),
queue: 1,
circuit_ids: vec!["existing".to_string(), "new".to_string()],
}];
let previous_circuits = BTreeMap::from([(
"existing".to_string(),
PlannerCircuitIdentityState {
class_minor: u16::MAX - 1,
queue: 1,
parent_node: "flat-bucket".to_string(),
class_major: 1,
up_class_major: 1,
},
)]);
let result = plan_class_identities(
&[],
&circuit_groups,
&BTreeMap::new(),
&previous_circuits,
0,
0,
);
assert_eq!(result.circuit_state["existing"].class_minor, u16::MAX - 1);
assert_eq!(result.circuit_state["new"].class_minor, 3);
}
}
+71 -4
View File
@@ -74,6 +74,68 @@ fn runtime_flat_bucket_id(index: usize) -> String {
format!("libreqos:generated:flat:bucket:{index}")
}
fn current_flat_bucket_names(config: &Config) -> Option<BTreeSet<String>> {
let Value::Object(current_network) = read_json_value(&topology_effective_network_path(config))?
else {
return None;
};
if current_network.is_empty() {
return None;
}
let mut bucket_names = BTreeSet::new();
for index in 0..current_network.len() {
let bucket_name = runtime_flat_bucket_name(index);
let expected_id = runtime_flat_bucket_id(index);
let node = current_network.get(&bucket_name)?.as_object()?;
if node.get("id").and_then(Value::as_str) != Some(expected_id.as_str()) {
return None;
}
if !node
.get("children")
.and_then(Value::as_object)
.is_some_and(|children| children.is_empty())
{
return None;
}
bucket_names.insert(bucket_name);
}
(bucket_names.len() == current_network.len()).then_some(bucket_names)
}
fn previous_flat_bucket_assignments(
config: &Config,
bucket_names: &[String],
) -> BTreeMap<String, String> {
let expected_bucket_names = bucket_names.iter().cloned().collect::<BTreeSet<_>>();
if current_flat_bucket_names(config).as_ref() != Some(&expected_bucket_names) {
return BTreeMap::new();
}
let bucket_ids_by_name = bucket_names
.iter()
.enumerate()
.map(|(index, name)| (name.as_str(), runtime_flat_bucket_id(index)))
.collect::<HashMap<_, _>>();
let Ok(previous_inputs) = TopologyShapingInputsFile::load(config) else {
return BTreeMap::new();
};
previous_inputs
.circuits
.into_iter()
.filter_map(|circuit| {
if circuit.resolution_source != TopologyShapingResolutionSource::FlatBucket {
return None;
}
let expected_bucket_id =
bucket_ids_by_name.get(circuit.effective_parent_node_name.as_str())?;
(circuit.effective_parent_node_id == *expected_bucket_id)
.then_some((circuit.circuit_id, circuit.effective_parent_node_name))
})
.collect()
}
fn runtime_flat_bucket_network(config: &Config) -> Value {
let mut root = Map::new();
for index in 0..runtime_flat_bucket_count(config) {
@@ -141,24 +203,29 @@ fn build_flat_bucket_assignments(
.into_iter()
.map(|(id, weight)| TopLevelPlannerItem { id, weight })
.collect::<Vec<_>>();
let previous_assignments = if config.queues.use_binpacking {
previous_flat_bucket_assignments(config, &bucket_names)
} else {
BTreeMap::new()
};
let planner_mode = if config.queues.use_binpacking {
TopLevelPlannerMode::Greedy
TopLevelPlannerMode::StableGreedy
} else {
TopLevelPlannerMode::RoundRobin
};
let planner = plan_top_level_assignments(
&items,
&bucket_names,
&BTreeMap::new(),
&previous_assignments,
&BTreeMap::new(),
lqos_utils::unix_time::unix_now()
.map(|timestamp| timestamp as f64)
.unwrap_or(0.0),
&TopLevelPlannerParams {
mode: planner_mode,
hysteresis_threshold: 0.0,
hysteresis_threshold: 0.03,
cooldown_seconds: 0.0,
move_budget_per_run: usize::MAX,
move_budget_per_run: 1,
},
);
+1 -1
View File
@@ -26,7 +26,7 @@ use lqos_overrides::{
use lqos_topology_compile::{TopologyCompiledShapingFile, TopologyImportFile};
use lqos_utils::process_lock::{ProcessFileLock, ProcessLockConfig};
use serde_json::{Map, Value};
use std::collections::{BTreeMap, HashMap, HashSet, hash_map::Entry};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, hash_map::Entry};
use std::fs::File;
use std::io::Write;
use std::net::IpAddr;
@@ -0,0 +1,236 @@
fn flat_test_config(prefix: &str, queue_count: u32) -> Config {
let lqos_directory = unique_temp_dir(prefix);
let mut config = Config {
lqos_directory: lqos_directory.to_string_lossy().to_string(),
state_directory: None,
..Config::default()
};
config.topology.compile_mode = "flat".to_string();
config.queues.override_available_queues = Some(queue_count);
config.queues.use_binpacking = true;
config
}
fn write_flat_test_devices(config: &Config, circuits: &[(&str, f32)]) {
let mut csv = String::from(
"Circuit ID,Circuit Name,Device ID,Device Name,Parent Node,Parent Node ID,Anchor Node ID,MAC,IPv4,IPv6,Download Min Mbps,Upload Min Mbps,Download Max Mbps,Upload Max Mbps,Comment\n",
);
for (index, (circuit_id, rate)) in circuits.iter().enumerate() {
csv.push_str(&format!(
"\"{circuit_id}\",\"{circuit_id}\",\"device-{index}\",\"Device {index}\",\"\",\"\",\"\",\"02:00:00:00:00:{index:02x}\",\"192.0.2.{}/32\",\"\",\"10\",\"10\",\"{rate}\",\"{rate}\",\"\"\n",
index + 1,
));
}
fs::write(
PathBuf::from(&config.lqos_directory).join("ShapedDevices.csv"),
csv,
)
.expect("flat ShapedDevices.csv should write");
}
fn flat_test_artifacts(config: &Config) -> EffectiveTopologyArtifacts {
let canonical = TopologyCanonicalStateFile::from_legacy_network_json(&json!({}));
build_effective_topology_artifacts_from_canonical(
config,
&canonical,
&TopologyOverridesFile::default(),
&TopologyAttachmentHealthStateFile::default(),
)
.expect("flat mode effective artifacts should build")
}
fn flat_assignment(inputs: &TopologyShapingInputsFile) -> BTreeMap<String, String> {
inputs
.circuits
.iter()
.map(|circuit| {
(
circuit.circuit_id.clone(),
circuit.effective_parent_node_name.clone(),
)
})
.collect()
}
fn changed_existing_assignments(
before: &BTreeMap<String, String>,
after: &BTreeMap<String, String>,
) -> usize {
before
.iter()
.filter(|(circuit_id, bucket)| {
after
.get(*circuit_id)
.is_some_and(|next_bucket| next_bucket != *bucket)
})
.count()
}
#[test]
fn flat_binpacking_reuses_assignments_for_identical_input() {
let config = flat_test_config("lqos-topology-flat-stable-identical", 2);
write_flat_test_devices(
&config,
&[
("circuit-a", 100.0),
("circuit-b", 80.0),
("circuit-c", 60.0),
("circuit-d", 40.0),
],
);
let artifacts = flat_test_artifacts(&config);
publish_effective_topology_artifacts(&config, &artifacts, "flat-identical")
.expect("initial flat topology should publish");
let before =
TopologyShapingInputsFile::load(&config).expect("published shaping inputs should load");
let after = build_shaping_inputs(&config, &artifacts)
.expect("repeated flat shaping inputs should build")
.expect("repeated flat shaping inputs should be present");
assert_eq!(flat_assignment(&before), flat_assignment(&after));
}
#[test]
fn flat_binpacking_bounds_existing_moves_when_circuits_change() {
let config = flat_test_config("lqos-topology-flat-stable-membership", 2);
let initial_circuits = [
("circuit-b", 100.0),
("circuit-c", 80.0),
("circuit-d", 60.0),
("circuit-e", 40.0),
];
write_flat_test_devices(&config, &initial_circuits);
let artifacts = flat_test_artifacts(&config);
publish_effective_topology_artifacts(&config, &artifacts, "flat-before-add")
.expect("initial flat topology should publish");
let before_add = flat_assignment(
&TopologyShapingInputsFile::load(&config).expect("initial shaping inputs should load"),
);
let with_added_circuit = [
("circuit-a", 120.0),
("circuit-b", 100.0),
("circuit-c", 80.0),
("circuit-d", 60.0),
("circuit-e", 40.0),
];
write_flat_test_devices(&config, &with_added_circuit);
let after_add_inputs = build_shaping_inputs(&config, &artifacts)
.expect("flat shaping inputs with an added circuit should build")
.expect("flat shaping inputs with an added circuit should be present");
let after_add = flat_assignment(&after_add_inputs);
assert!(changed_existing_assignments(&before_add, &after_add) <= 1);
publish_effective_topology_artifacts(&config, &artifacts, "flat-before-remove")
.expect("flat topology with the added circuit should publish");
let after_add_published = flat_assignment(
&TopologyShapingInputsFile::load(&config)
.expect("added-circuit shaping inputs should load"),
);
write_flat_test_devices(
&config,
&[
("circuit-a", 120.0),
("circuit-b", 100.0),
("circuit-d", 60.0),
("circuit-e", 40.0),
],
);
let after_remove = flat_assignment(
&build_shaping_inputs(&config, &artifacts)
.expect("flat shaping inputs with a removed circuit should build")
.expect("flat shaping inputs with a removed circuit should be present"),
);
assert!(changed_existing_assignments(&after_add_published, &after_remove) <= 1);
}
#[test]
fn flat_binpacking_discards_invalid_or_wrong_queue_count_history() {
let mut config = flat_test_config("lqos-topology-flat-invalid-history", 2);
let circuits = [
("circuit-a", 100.0),
("circuit-b", 80.0),
("circuit-c", 60.0),
("circuit-d", 40.0),
];
write_flat_test_devices(&config, &circuits);
let artifacts = flat_test_artifacts(&config);
let cold_assignment = flat_assignment(
&build_shaping_inputs(&config, &artifacts)
.expect("cold flat shaping inputs should build")
.expect("cold flat shaping inputs should be present"),
);
publish_effective_topology_artifacts(&config, &artifacts, "flat-invalid-history")
.expect("initial flat topology should publish");
let mut invalid_history =
TopologyShapingInputsFile::load(&config).expect("published shaping inputs should load");
for circuit in &mut invalid_history.circuits {
let (bucket_name, bucket_id) = if circuit.effective_parent_node_name == "Generated_PN_1" {
("Generated_PN_2", "libreqos:generated:flat:bucket:1")
} else {
("Generated_PN_1", "libreqos:generated:flat:bucket:0")
};
circuit.effective_parent_node_name = bucket_name.to_string();
circuit.effective_parent_node_id = bucket_id.to_string();
}
invalid_history
.circuits
.iter_mut()
.find(|circuit| circuit.circuit_id == "circuit-a")
.expect("circuit-a should exist")
.effective_parent_node_id = "wrong-bucket-id".to_string();
invalid_history
.circuits
.iter_mut()
.find(|circuit| circuit.circuit_id == "circuit-b")
.expect("circuit-b should exist")
.resolution_source = TopologyShapingResolutionSource::RuntimeFallback;
let deliberately_retained_assignment = flat_assignment(&invalid_history);
invalid_history
.save(&config)
.expect("invalid shaping history should save");
let rebuilt_assignment = flat_assignment(
&build_shaping_inputs(&config, &artifacts)
.expect("flat shaping inputs should rebuild from invalid history")
.expect("rebuilt flat shaping inputs should be present"),
);
assert_eq!(
rebuilt_assignment["circuit-a"],
cold_assignment["circuit-a"]
);
assert_eq!(
rebuilt_assignment["circuit-b"],
cold_assignment["circuit-b"]
);
assert_eq!(
rebuilt_assignment["circuit-c"],
deliberately_retained_assignment["circuit-c"]
);
assert_eq!(
rebuilt_assignment["circuit-d"],
deliberately_retained_assignment["circuit-d"]
);
config.queues.override_available_queues = Some(3);
let three_bucket_artifacts = flat_test_artifacts(&config);
let after_queue_change = flat_assignment(
&build_shaping_inputs(&config, &three_bucket_artifacts)
.expect("flat shaping inputs should rebuild after queue-count change")
.expect("queue-count rebuild should produce shaping inputs"),
);
let cold_three_bucket_config = flat_test_config("lqos-topology-flat-cold-three-buckets", 3);
write_flat_test_devices(&cold_three_bucket_config, &circuits);
let cold_three_bucket_artifacts = flat_test_artifacts(&cold_three_bucket_config);
let cold_three_bucket_assignment = flat_assignment(
&build_shaping_inputs(&cold_three_bucket_config, &cold_three_bucket_artifacts)
.expect("cold three-bucket shaping inputs should build")
.expect("cold three-bucket shaping inputs should be present"),
);
assert_eq!(after_queue_change, cold_three_bucket_assignment);
}
+3 -2
View File
@@ -18,13 +18,14 @@
TopologyCanonicalRateInputSource, TopologyCanonicalStateFile, TopologyEditorNode,
TopologyEditorStateFile, TopologyEffectiveAttachmentState, TopologyEffectiveNodeState,
TopologyEffectiveStateFile, TopologyQueueVisibilityPolicy, TopologyRuntimeStatusFile,
TopologyShapingResolutionSource, compute_effective_network_file_generation,
TopologyShapingInputsFile, TopologyShapingResolutionSource,
compute_effective_network_file_generation,
topology_auto_attachment_option as auto_attachment_option, topology_effective_network_path,
topology_effective_state_path, topology_runtime_status_path, topology_shaping_inputs_path,
};
use lqos_overrides::{TopologyAttachmentMode, TopologyOverridesFile};
use serde_json::{Value, json};
use std::collections::{HashMap, HashSet};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fs;
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
+1
View File
@@ -5,3 +5,4 @@ include!("group_02.rs");
include!("group_03.rs");
include!("group_04.rs");
include!("group_05.rs");
include!("group_06.rs");
+108
View File
@@ -98,6 +98,114 @@ def tearDownModule():
class TestLibreQoSShapingInputs(unittest.TestCase):
def test_flat_planner_item_key_uses_stable_circuit_id(self):
circuit = {
"circuitID": "circuit-stable",
"idForCircuitsWithoutParentNodes": 17,
}
self.assertEqual(
LibreQoS.planner_top_level_item_key(circuit, flat_network=True),
"flat_circuit:circuit-stable",
)
self.assertEqual(
LibreQoS.planner_top_level_item_key(circuit, flat_network=False),
"17",
)
self.assertEqual(
LibreQoS.planner_top_level_item_key(
{"circuitID": "0", "idForCircuitsWithoutParentNodes": 17},
flat_network=True,
),
"flat_circuit:0",
)
def test_flat_binpacking_detection_covers_legacy_and_runtime_topologies(self):
runtime_flat_network = {
"Generated_PN_1": {
"id": "libreqos:generated:flat:bucket:0",
"children": {},
},
"Generated_PN_2": {
"id": "libreqos:generated:flat:bucket:1",
"children": {},
},
}
self.assertTrue(LibreQoS.flat_binpacking_enabled({}, True))
self.assertTrue(
LibreQoS.flat_binpacking_enabled(runtime_flat_network, True)
)
self.assertFalse(
LibreQoS.flat_binpacking_enabled({"Tower": {}}, True)
)
self.assertFalse(
LibreQoS.flat_binpacking_enabled({"Generated_PN_bad": {}}, True)
)
self.assertFalse(
LibreQoS.flat_binpacking_enabled({"Generated_PN_1": {}}, True)
)
self.assertFalse(
LibreQoS.flat_binpacking_enabled(
{"Generated_PN_1": {}, "Generated_PN_3": {}},
True,
)
)
self.assertFalse(
LibreQoS.flat_binpacking_enabled(runtime_flat_network, False)
)
malformed_runtime_network = dict(runtime_flat_network)
malformed_runtime_network["Generated_PN_1"] = {
"id": "libreqos:generated:flat:bucket:0",
"children": {"real-child": {}},
}
self.assertFalse(
LibreQoS.flat_binpacking_enabled(malformed_runtime_network, True)
)
def test_flat_planner_failure_preserves_existing_assignments(self):
items = [
{"id": "flat_circuit:existing", "weight": 10.0},
{"id": "flat_circuit:new", "weight": 5.0},
]
previous_assignments = {
"flat_circuit:existing": "Generated_PN_2",
}
with patch.object(
LibreQoS,
"plan_top_level_cpu_bins",
side_effect=RuntimeError("planner unavailable"),
):
with self.assertWarnsRegex(UserWarning, "preserving prior assignments"):
assignments, changed = LibreQoS.plan_flat_generated_parent_assignments(
items,
["Generated_PN_1", "Generated_PN_2"],
previous_assignments,
{},
1000.0,
1,
)
self.assertEqual(assignments["flat_circuit:existing"], "Generated_PN_2")
self.assertEqual(assignments["flat_circuit:new"], "Generated_PN_1")
self.assertEqual(changed, [])
def test_saved_identity_state_is_reused_only_for_flat_binpacking(self):
state = {
"sites": {"Generated_PN_1": {"class_minor": 3}},
"circuits": {"circuit-1": {"class_minor": 4}},
}
self.assertEqual(
LibreQoS.reusable_identity_state(state, True),
(state["sites"], state["circuits"]),
)
self.assertEqual(
LibreQoS.reusable_identity_state(state, False),
({}, {}),
)
def test_runtime_resolved_circuit_attachment_candidates_ignore_logical_parent(self):
candidates = LibreQoS._circuit_attachment_name_candidates(
{