IntenseWebs/LibreQoS
3
0 Fork 0
mirror of https://github.com/LibreQoE/LibreQoS.git synced 2026-11-02 20:30:15 -06:00
Code Issues Packages Projects Releases Wiki Activity

WIP: Building tree output files are now identical. Now passes circuit_id and site_id hashes. Now builds the queue structure on the Rust side!

Browse Source
This commit is contained in:
Herbert Wolverson committed 2025-06-04 15:08:07 -05:00
1 parent 3421f0d3c0
commit 7a90d060a1
10 files changed
+915 -494
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+1
View File
@@ -135,3 +135,4 @@ src/bin/static2/flags
/src/rust/lqosd/lqosd-output.log
/src/bin/uisp_integration
/src/xdp_iphash_output.log
/src/linux_tc_rust.txt
src/LibreQoS.py
+18 -10
View File
@@ -25,7 +25,8 @@ from liblqos_python import is_lqosd_alive, clear_ip_mappings, delete_ip_mapping,
check_config, sqm, upstream_bandwidth_capacity_download_mbps, upstream_bandwidth_capacity_upload_mbps, \
interface_a, interface_b, enable_actual_shell_commands, use_bin_packing_to_balance_cpu, monitor_mode_only, \
run_shell_commands_as_sudo, generated_pn_download_mbps, generated_pn_upload_mbps, queues_available_override, \
on_a_stick, get_tree_weights, get_weights, is_network_flat, get_libreqos_directory, enable_insight_topology
on_a_stick, get_tree_weights, get_weights, is_network_flat, get_libreqos_directory, enable_insight_topology, \
Bakery
R2Q = 10
#MAX_R2Q = 200_000
@@ -870,7 +871,9 @@ def refreshShapers():
# Here is the actual call to the recursive traverseNetwork() function. finalMinor is not used.
minorByCPU = traverseNetwork(network, 0, major=1, minorByCPU=minorByCPUpreloaded, queue=1, parentClassID=None, upParentClassID=None, parentMaxDL=upstream_bandwidth_capacity_download_mbps(), parentMaxUL=upstream_bandwidth_capacity_upload_mbps(), parentMinDL=upstream_bandwidth_capacity_download_mbps(), parentMinUL=upstream_bandwidth_capacity_upload_mbps())
bakery = Bakery()
bakery.start_batch() # Initializes the bakery transaction
linuxTCcommands = []
devicesShaped = []
# Root HTB Setup
@@ -878,6 +881,7 @@ def refreshShapers():
thisInterface = interface_a()
logging.info("# MQ Setup for " + thisInterface)
command = 'qdisc replace dev ' + thisInterface + ' root handle 7FFF: mq'
bakery.setup_mq(queuesAvailable, stickOffset)
linuxTCcommands.append(command)
maxBandwidth = max(upstream_bandwidth_capacity_upload_mbps(), upstream_bandwidth_capacity_download_mbps())
calculateR2q(maxBandwidth)
@@ -944,6 +948,8 @@ def refreshShapers():
case _: return sqm
for node in data:
site_name = data[node]['name'] if 'name' in data[node] else "root"
bakery.add_site(site_name, data[node]['parentClassID'], data[node]['up_parentClassID'], data[node]['classMinor'], format_rate_for_tc(data[node]['downloadBandwidthMbpsMin']), format_rate_for_tc(data[node]['uploadBandwidthMbpsMin']), format_rate_for_tc(data[node]['downloadBandwidthMbps']), format_rate_for_tc(data[node]['uploadBandwidthMbps']), quantum(data[node]['downloadBandwidthMbps']), quantum(data[node]['uploadBandwidthMbps']))
command = 'class add dev ' + interface_a() + ' parent ' + data[node]['parentClassID'] + ' classid ' + data[node]['classMinor'] + ' htb rate '+ format_rate_for_tc(data[node]['downloadBandwidthMbpsMin']) + ' ceil '+ format_rate_for_tc(data[node]['downloadBandwidthMbps']) + ' prio 3' + quantum(data[node]['downloadBandwidthMbps'])
linuxTCcommands.append(command)
logging.info("Up ParentClassID: " + data[node]['up_parentClassID'])
@@ -969,6 +975,8 @@ def refreshShapers():
if 'comment' in circuit['devices'][0]:
tcComment = '' # tcComment + '| Comment: ' + circuit['devices'][0]['comment']
tcComment = tcComment.replace("\n", "")
circuit_name = circuit['circuitID'] if 'circuitID' in circuit else "unknown"
bakery.add_circuit(circuit_name, data[node]['classid'], data[node]['up_classid'], circuit['classMinor'], format_rate_for_tc(min_down), format_rate_for_tc(min_up), format_rate_for_tc(circuit['maxDownload']), format_rate_for_tc(circuit['maxUpload']), quantum(circuit['maxDownload']), quantum(circuit['maxUpload']), circuit['classMajor'], circuit['up_classMajor'], sqmFixupRate(circuit['maxDownload'], sqm()), sqmFixupRate(circuit['maxUpload'], sqm()), tcComment)
command = 'class add dev ' + interface_a() + ' parent ' + data[node]['classid'] + ' classid ' + circuit['classMinor'] + ' htb rate '+ format_rate_for_tc(min_down) + ' ceil '+ format_rate_for_tc(circuit['maxDownload']) + ' prio 3' + quantum(circuit['maxDownload']) + tcComment
linuxTCcommands.append(command)
# Only add CAKE / fq_codel qdisc if monitorOnlyMode is Off
@@ -1046,19 +1054,19 @@ def refreshShapers():
# Execute actual Linux TC commands
tcStartTime = datetime.now()
print("Executing linux TC class/qdisc commands")
# print("Executing linux TC class/qdisc commands")
with open('linux_tc.txt', 'w') as f:
for command in linuxTCcommands:
logging.info(command)
f.write(f"{command}\n")
if logging.DEBUG <= logging.root.level:
# Do not --force in debug mode, so we can see any errors
shell("/sbin/tc -b linux_tc.txt")
else:
shell("/sbin/tc -f -b linux_tc.txt")
# if logging.DEBUG <= logging.root.level:
# # Do not --force in debug mode, so we can see any errors
# shell("/sbin/tc -b linux_tc.txt")
# else:
# shell("/sbin/tc -f -b linux_tc.txt")
bakery.commit()
tcEndTime = datetime.now()
print("Executed " + str(len(linuxTCcommands)) + " linux TC class/qdisc commands")
# print("Executed " + str(len(linuxTCcommands)) + " linux TC class/qdisc commands")
# Execute actual XDP-CPUMAP-TC filter commands
xdpFilterStartTime = datetime.now()
src/rust/lqos_bakery/src/commands.rs
+440 -1
View File
@@ -1,4 +1,443 @@
use std::sync::Arc;
/// List of commands that the Bakery system can handle.
#[derive(Debug)]
#[derive(Debug, Clone)]
pub enum BakeryCommands {
StartBatch,
CommitBatch,
MqSetup { queues_available: usize, stick_offset: usize },
AddSite {
site_hash: i64,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
},
AddCircuit {
circuit_hash: i64,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
class_major: String,
up_class_major: String,
sqm_down: String,
sqm_up: String,
comment: String,
}
}
impl BakeryCommands {
pub fn to_commands(&self, config: &Arc<lqos_config::Config>) -> Option<Vec<Vec<String>>> {
match self {
BakeryCommands::MqSetup { queues_available, stick_offset } => Self::mq_setup(config, *queues_available, *stick_offset),
BakeryCommands::AddSite {
site_hash,
parent_class_id,
up_parent_class_id,
class_minor,
download_bandwidth_min,
upload_bandwidth_min,
download_bandwidth_max,
upload_bandwidth_max,
quantum_down,
quantum_up,
} => Self::add_site(
config, *site_hash, parent_class_id.clone(), up_parent_class_id.clone(),
class_minor.clone(), download_bandwidth_min.clone(),
upload_bandwidth_min.clone(), download_bandwidth_max.clone(),
upload_bandwidth_max.clone(), quantum_down.clone(),
quantum_up.clone(),
),
BakeryCommands::AddCircuit {
circuit_hash,
parent_class_id,
up_parent_class_id,
class_minor,
download_bandwidth_min,
upload_bandwidth_min,
download_bandwidth_max,
upload_bandwidth_max,
quantum_down,
quantum_up,
class_major,
up_class_major,
sqm_down,
sqm_up,
comment,
} => Self::add_circuit(
config, *circuit_hash, parent_class_id.clone(), up_parent_class_id.clone(),
class_minor.clone(), download_bandwidth_min.clone(),
upload_bandwidth_min.clone(), download_bandwidth_max.clone(),
upload_bandwidth_max.clone(), quantum_down.clone(),
quantum_up.clone(), class_major.clone(), up_class_major.clone(),
sqm_down.clone(), sqm_up.clone(), comment.clone(),
),
_ => None,
}
}
fn sqm_as_vec(config: &Arc<lqos_config::Config>) -> Vec<String> {
config.queues.default_sqm.split(" ").map(|s| s.to_string()).collect()
}
fn format_rate_for_tc(rate: u64) -> String {
// Format a rate in Mbps for TC commands with smart unit selection.
// - Rates >= 1000 Mbps use 'gbit'
// - Rates >= 1 Mbps use 'mbit'
// - Rates < 1 Mbps use 'kbit'
if rate >= 1000 {
format!("{:.1}gbit", rate as f64 / 1000.0)
} else if rate >= 1 {
format!("{:.1}mbit", rate as f64)
} else {
format!("{:.0}kbit", rate as f64 * 1000.0)
}
}
fn r2q(max_rate_in_mbps: u64) -> u64 {
// Constants from Python implementation
const MAX_R2Q: f64 = 60_000.0; // From LibreQoS.py
// Convert rate in Mbps to bytes per second
let max_rate_in_bytes_per_second = (max_rate_in_mbps * 125000) as f64;
// Start with a default r2q value of 10
let mut r2q = 10u64;
// Calculate initial quantum using floating point division to match Python
let mut quantum = max_rate_in_bytes_per_second / r2q as f64;
// Increment r2q until quantum is below MAX_R2Q
// This matches Python's behavior of comparing float values
while quantum > MAX_R2Q {
r2q += 1;
quantum = max_rate_in_bytes_per_second / r2q as f64;
}
r2q
}
fn quantum(rate: u64, r2q: u64) -> String {
// Constants from Python implementation
const MIN_QUANTUM: u64 = 1522; // From LibreQoS.py
// Convert rate in Mbps to bytes per second
let rate_in_bytes_per_second = rate * 125000;
// Calculate quantum value using the same logic as Python
let quantum = std::cmp::max(MIN_QUANTUM, rate_in_bytes_per_second / r2q);
// Format and return the quantum string
quantum.to_string()
}
fn mq_setup(
config: &Arc<lqos_config::Config>,
queues_available: usize,
stick_offset: usize,
) -> Option<Vec<Vec<String>>> {
// command = 'qdisc replace dev ' + thisInterface + ' root handle 7FFF: mq'
let mut result = Vec::new();
let sqm_strings = Self::sqm_as_vec(config);
let r2q = Self::r2q(u64::max(config.queues.uplink_bandwidth_mbps, config.queues.downlink_bandwidth_mbps));
// ISP-facing interface (interface_a in Python)
result.push(vec![
"qdisc".to_string(), "replace".to_string(), "dev".to_string(),
config.isp_interface(),
"root".to_string(), "handle".to_string(), "7FFF:".to_string(),
"mq".to_string(),
]);
/*
for queue in range(queuesAvailable):
command = 'qdisc add dev ' + thisInterface + ' parent 7FFF:' + hex(queue+1) + ' handle ' + hex(queue+1) + ': htb default 2'
linuxTCcommands.append(command)
command = 'class add dev ' + thisInterface + ' parent ' + hex(queue+1) + ': classid ' + hex(queue+1) + ':1 htb rate '+ format_rate_for_tc(upstream_bandwidth_capacity_download_mbps()) + ' ceil ' + format_rate_for_tc(upstream_bandwidth_capacity_download_mbps()) + quantum(upstream_bandwidth_capacity_download_mbps())
linuxTCcommands.append(command)
command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':1 ' + sqm()
linuxTCcommands.append(command)
# Default class - traffic gets passed through this limiter with lower priority if it enters the top HTB without a specific class.
# Technically, that should not even happen. So don't expect much if any traffic in this default class.
# Only 1/4 of defaultClassCapacity is guaranteed (to prevent hitting ceiling of upstream), for the most part it serves as an "up to" ceiling.
command = 'class add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':1 classid ' + hex(queue+1) + ':2 htb rate ' + format_rate_for_tc(round((upstream_bandwidth_capacity_download_mbps()-1)/4)) + ' ceil ' + format_rate_for_tc(upstream_bandwidth_capacity_download_mbps()-1) + ' prio 5' + quantum(upstream_bandwidth_capacity_download_mbps())
linuxTCcommands.append(command)
command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':2 ' + sqm()
linuxTCcommands.append(command)
*/
for queue in 0 .. queues_available {
// command = 'qdisc add dev ' + thisInterface + ' parent 7FFF:' + hex(queue+1) + ' handle ' + hex(queue+1) + ': htb default 2'
result.push(vec![
"qdisc".to_string(), "add".to_string(), "dev".to_string(),
config.isp_interface(),
"parent".to_string(), format!("7FFF:0x{:x}", queue + 1),
"handle".to_string(), format!("0x{:x}:", queue + 1), "htb".to_string(),
"default".to_string(), "2".to_string(),
]);
// command = 'class add dev ' + thisInterface + ' parent ' + hex(queue+1) + ': classid ' + hex(queue+1) + ':1 htb rate '+ format_rate_for_tc(upstream_bandwidth_capacity_download_mbps()) + ' ceil ' + format_rate_for_tc(upstream_bandwidth_capacity_download_mbps()) + quantum(upstream_bandwidth_capacity_download_mbps())
result.push(vec![
"class".to_string(), "add".to_string(), "dev".to_string(),
config.isp_interface(),
"parent".to_string(), format!("0x{:x}:", queue + 1),
"classid".to_string(), format!("0x{:x}:1", queue + 1), "htb".to_string(),
"rate".to_string(), Self::format_rate_for_tc(config.queues.uplink_bandwidth_mbps),
"ceil".to_string(), Self::format_rate_for_tc(config.queues.uplink_bandwidth_mbps),
"quantum".to_string(), Self::quantum(config.queues.uplink_bandwidth_mbps, r2q),
]);
// command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':1 ' + sqm()
let mut class = vec![
"qdisc".to_string(), "add".to_string(), "dev".to_string(),
config.isp_interface(),
"parent".to_string(), format!("0x{:x}:1", queue + 1),
];
class.extend(sqm_strings.clone());
result.push(class);
// Default class - traffic gets passed through this limiter with lower priority if it enters the top HTB without a specific class.
// command = 'class add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':1 classid ' + hex(queue+1) + ':2 htb rate ' + format_rate_for_tc(round((upstream_bandwidth_capacity_download_mbps()-1)/4)) + ' ceil ' + format_rate_for_tc(upstream_bandwidth_capacity_download_mbps()-1) + ' prio 5' + quantum(upstream_bandwidth_capacity_download_mbps())
let mbps = config.queues.uplink_bandwidth_mbps as f64;
let mbps_quarter = (mbps - 1.0) / 4.0;
let mbps_minus_one = mbps - 1.0;
result.push(vec![
"class".to_string(), "add".to_string(), "dev".to_string(),
config.isp_interface(),
"parent".to_string(), format!("0x{:x}:1", queue + 1),
"classid".to_string(), format!("0x{:x}:2", queue + 1), "htb".to_string(),
"rate".to_string(), Self::format_rate_for_tc(mbps_quarter as u64),
"ceil".to_string(), Self::format_rate_for_tc(mbps_minus_one as u64),
"prio".to_string(), "5".to_string(),
"quantum".to_string(), Self::quantum(config.queues.uplink_bandwidth_mbps, r2q),
]);
// command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':2 ' + sqm()
let mut default_class = vec![
"qdisc".to_string(), "add".to_string(), "dev".to_string(),
config.isp_interface(),
"parent".to_string(), format!("0x{:x}:2", queue + 1),
];
default_class.extend(sqm_strings.clone());
result.push(default_class);
}
// Internet-facing interface (interface_b in Python)
if !config.on_a_stick_mode() {
result.push(vec![
"qdisc".to_string(), "replace".to_string(), "dev".to_string(),
config.internet_interface(),
"root".to_string(), "handle".to_string(), "7FFF:".to_string(),
"mq".to_string(),
]);
}
/*
for queue in range(queuesAvailable):
command = 'qdisc add dev ' + thisInterface + ' parent 7FFF:' + hex(queue+stickOffset+1) + ' handle ' + hex(queue+stickOffset+1) + ': htb default 2'
linuxTCcommands.append(command)
command = 'class add dev ' + thisInterface + ' parent ' + hex(queue+stickOffset+1) + ': classid ' + hex(queue+stickOffset+1) + ':1 htb rate '+ format_rate_for_tc(upstream_bandwidth_capacity_upload_mbps()) + ' ceil ' + format_rate_for_tc(upstream_bandwidth_capacity_upload_mbps()) + quantum(upstream_bandwidth_capacity_upload_mbps())
linuxTCcommands.append(command)
command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+stickOffset+1) + ':1 ' + sqm()
linuxTCcommands.append(command)
# Default class - traffic gets passed through this limiter with lower priority if it enters the top HTB without a specific class.
# Technically, that should not even happen. So don't expect much if any traffic in this default class.
# Only 1/4 of defaultClassCapacity is guarenteed (to prevent hitting ceiling of upstream), for the most part it serves as an "up to" ceiling.
command = 'class add dev ' + thisInterface + ' parent ' + hex(queue+stickOffset+1) + ':1 classid ' + hex(queue+stickOffset+1) + ':2 htb rate ' + format_rate_for_tc(round((upstream_bandwidth_capacity_upload_mbps()-1)/4)) + ' ceil ' + format_rate_for_tc(upstream_bandwidth_capacity_upload_mbps()-1) + ' prio 5' + quantum(upstream_bandwidth_capacity_upload_mbps())
linuxTCcommands.append(command)
command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+stickOffset+1) + ':2 ' + sqm()
linuxTCcommands.append(command)
*/
for queue in 0 .. queues_available {
// command = 'qdisc add dev ' + thisInterface + ' parent 7FFF:' + hex(queue+stickOffset+1) + ' handle ' + hex(queue+stickOffset+1) + ': htb default 2'
result.push(vec![
"qdisc".to_string(), "add".to_string(), "dev".to_string(),
config.internet_interface(),
"parent".to_string(), format!("7FFF:0x{:x}", queue + stick_offset + 1),
"handle".to_string(), format!("0x{:x}:", queue + stick_offset + 1), "htb".to_string(),
"default".to_string(), "2".to_string(),
]);
// command = 'class add dev ' + thisInterface + ' parent ' + hex(queue+stickOffset+1) + ': classid ' + hex(queue+stickOffset+1) + ':1 htb rate '+ format_rate_for_tc(upstream_bandwidth_capacity_upload_mbps()) + ' ceil ' + format_rate_for_tc(upstream_bandwidth_capacity_upload_mbps()) + quantum(upstream_bandwidth_capacity_upload_mbps())
result.push(vec![
"class".to_string(), "add".to_string(), "dev".to_string(),
config.internet_interface(),
"parent".to_string(), format!("0x{:x}:", queue + stick_offset + 1),
"classid".to_string(), format!("0x{:x}:1", queue + stick_offset + 1), "htb".to_string(),
"rate".to_string(), Self::format_rate_for_tc(config.queues.downlink_bandwidth_mbps),
"ceil".to_string(), Self::format_rate_for_tc(config.queues.downlink_bandwidth_mbps),
"quantum".to_string(), Self::quantum(config.queues.downlink_bandwidth_mbps, r2q),
]);
// command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+stickOffset+1) + ':1 ' + sqm()
let mut class = vec![
"qdisc".to_string(), "add".to_string(), "dev".to_string(),
config.internet_interface(),
"parent".to_string(), format!("0x{:x}:1", queue + stick_offset + 1),
];
class.extend(sqm_strings.clone());
result.push(class);
// Default class - traffic gets passed through this limiter with lower priority if it enters the top HTB without a specific class.
// command = 'class add dev ' + thisInterface + ' parent ' + hex(queue+stickOffset+1) + ':1 classid ' + hex(queue+stickOffset+1) + ':2 htb rate ' + format_rate_for_tc(round((upstream_bandwidth_capacity_upload_mbps()-1)/4)) + ' ceil ' + format_rate_for_tc(upstream_bandwidth_capacity_upload_mbps()-1) + ' prio 5' + quantum(upstream_bandwidth_capacity_upload_mbps())
let mbps = config.queues.downlink_bandwidth_mbps as f64;
let mbps_quarter = (mbps - 1.0) / 4.0;
let mbps_minus_one = mbps - 1.0;
result.push(vec![
"class".to_string(), "add".to_string(), "dev".to_string(),
config.internet_interface(),
"parent".to_string(), format!("0x{:x}:1", queue + stick_offset + 1),
"classid".to_string(), format!("0x{:x}:2", queue + stick_offset + 1), "htb".to_string(),
"rate".to_string(), Self::format_rate_for_tc(mbps_quarter as u64),
"ceil".to_string(), Self::format_rate_for_tc(mbps_minus_one as u64),
"prio".to_string(), "5".to_string(),
"quantum".to_string(), Self::quantum(config.queues.downlink_bandwidth_mbps, r2q),
]);
// command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+stickOffset+1) + ':2 ' + sqm()
let mut default_class = vec![
"qdisc".to_string(), "add".to_string(), "dev".to_string(),
config.internet_interface(),
"parent".to_string(), format!("0x{:x}:2", queue + stick_offset + 1),
];
default_class.extend(sqm_strings.clone());
result.push(default_class);
}
Some(result)
}
fn add_site(
config: &Arc<lqos_config::Config>,
site_hash: i64,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
) -> Option<Vec<Vec<String>>> {
let mut result = Vec::new();
/*
bakery.add_site(data[node]['parentClassID'], data[node]['up_parentClassID'], data[node]['classMinor'], format_rate_for_tc(data[node]['downloadBandwidthMbpsMin']), format_rate_for_tc(data[node]['uploadBandwidthMbpsMin']), format_rate_for_tc(data[node]['downloadBandwidthMbps']), format_rate_for_tc(data[node]['uploadBandwidthMbps']), quantum(data[node]['downloadBandwidthMbps']), quantum(data[node]['uploadBandwidthMbps']))
command = 'class add dev ' + interface_a() + ' parent ' + data[node]['parentClassID'] + ' classid ' + data[node]['classMinor'] + ' htb rate '+ format_rate_for_tc(data[node]['downloadBandwidthMbpsMin']) + ' ceil '+ format_rate_for_tc(data[node]['downloadBandwidthMbps']) + ' prio 3' + quantum(data[node]['downloadBandwidthMbps'])
linuxTCcommands.append(command)
logging.info("Up ParentClassID: " + data[node]['up_parentClassID'])
logging.info("ClassMinor: " + data[node]['classMinor'])
command = 'class add dev ' + interface_b() + ' parent ' + data[node]['up_parentClassID'] + ' classid ' + data[node]['classMinor'] + ' htb rate '+ format_rate_for_tc(data[node]['uploadBandwidthMbpsMin']) + ' ceil '+ format_rate_for_tc(data[node]['uploadBandwidthMbps']) + ' prio 3' + quantum(data[node]['uploadBandwidthMbps'])
*/
result.push(vec![
"class".to_string(), "add".to_string(), "dev".to_string(), config.isp_interface(),
"parent".to_string(), parent_class_id.clone(),
"classid".to_string(), class_minor.clone(), "htb".to_string(),
"rate".to_string(), download_bandwidth_min.clone(),
"ceil".to_string(), download_bandwidth_max.clone(),
"prio".to_string(), "3".to_string(),
quantum_down.clone(),
]);
result.push(vec![
"class".to_string(), "add".to_string(), "dev".to_string(), config.internet_interface(),
"parent".to_string(), up_parent_class_id.clone(),
"classid".to_string(), class_minor.clone(),
"htb".to_string(), "rate".to_string(), upload_bandwidth_min.clone(),
"ceil".to_string(), upload_bandwidth_max.clone(),
"prio".to_string(), "3".to_string(),
quantum_up.clone(),
]);
Some(result)
}
fn add_circuit(
config: &Arc<lqos_config::Config>,
circuit_hash: i64,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
class_major: String,
up_class_major: String,
sqm_down: String,
sqm_up: String,
comment: String,
) -> Option<Vec<Vec<String>>> {
let mut result = Vec::new();
/*
bakery.add_circuit(data[node]['classid'], data[node]['up_classid'], circuit['classMinor'], format_rate_for_tc(min_down), format_rate_for_tc(min_up), format_rate_for_tc(circuit['maxDownload']), format_rate_for_tc(circuit['maxUpload']), quantum(circuit['maxDownload']), quantum(circuit['maxUpload']), circuit['classMajor'], circuit['up_classMajor'], sqmFixupRate(circuit['maxDownload'], sqm()), sqmFixupRate(circuit['maxUpload'], sqm()), tcComment)
command = 'class add dev ' + interface_a() + ' parent ' + data[node]['classid'] + ' classid ' + circuit['classMinor'] + ' htb rate '+ format_rate_for_tc(min_down) + ' ceil '+ format_rate_for_tc(circuit['maxDownload']) + ' prio 3' + quantum(circuit['maxDownload']) + tcComment
linuxTCcommands.append(command)
# Only add CAKE / fq_codel qdisc if monitorOnlyMode is Off
if monitor_mode_only() == False:
# SQM Fixup for lower rates
useSqm = sqmFixupRate(circuit['maxDownload'], sqm())
command = 'qdisc add dev ' + interface_a() + ' parent ' + circuit['classMajor'] + ':' + circuit['classMinor'] + ' ' + useSqm
linuxTCcommands.append(command)
command = 'class add dev ' + interface_b() + ' parent ' + data[node]['up_classid'] + ' classid ' + circuit['classMinor'] + ' htb rate '+ format_rate_for_tc(min_up) + ' ceil '+ format_rate_for_tc(circuit['maxUpload']) + ' prio 3' + quantum(circuit['maxUpload'])
linuxTCcommands.append(command)
# Only add CAKE / fq_codel qdisc if monitorOnlyMode is Off
if monitor_mode_only() == False:
# SQM Fixup for lower rates
useSqm = sqmFixupRate(circuit['maxUpload'], sqm())
command = 'qdisc add dev ' + interface_b() + ' parent ' + circuit['up_classMajor'] + ':' + circuit['classMinor'] + ' ' + useSqm
linuxTCcommands.append(command)
pass
*/
result.push(vec![
"class".to_string(), "add".to_string(), "dev".to_string(), config.isp_interface(),
"parent".to_string(), parent_class_id.clone(),
"classid".to_string(), class_minor.clone(), "htb".to_string(),
"rate".to_string(), download_bandwidth_min.clone(),
"ceil".to_string(), download_bandwidth_max.clone(),
"prio".to_string(), "3".to_string(),
quantum_down.clone(),
]);
if !config.queues.monitor_only {
let sqm = Self::sqm_as_vec(config);
let sqm_command = vec![
"qdisc".to_string(), "add".to_string(), "dev".to_string(),
config.isp_interface(),
"parent".to_string(), format!("{}:{}", class_major, class_minor),
sqm_down,
];
result.push(sqm_command);
}
result.push(vec![
"class".to_string(), "add".to_string(), "dev".to_string(), config.internet_interface(),
"parent".to_string(), up_parent_class_id.clone(),
"classid".to_string(), class_minor.clone(),
"htb".to_string(), "rate".to_string(), upload_bandwidth_min.clone(),
"ceil".to_string(), upload_bandwidth_max.clone(),
"prio".to_string(), "3".to_string(),
quantum_up.clone(),
]);
if !config.queues.monitor_only {
let sqm = Self::sqm_as_vec(config);
let sqm_command = vec![
"qdisc".to_string(), "add".to_string(), "dev".to_string(),
config.internet_interface(),
"parent".to_string(), format!("{}:{}", up_class_major, class_minor),
sqm_up,
];
result.push(sqm_command);
}
Some(result)
}
}
src/rust/lqos_bakery/src/lib.rs
+104 -9
View File
@@ -16,22 +16,28 @@
//! In phase 4, the Bakery will implement "live move" --- allowing queues to be moved losslessly. This will
//! complete the NLNet project goals.
mod tc_control;
mod utils;
mod commands;
use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::Path;
use std::sync::{Arc, Mutex};
use crossbeam_channel::Receiver;
use std::sync::{Arc, Mutex, OnceLock};
use crossbeam_channel::{Receiver, Sender};
use tracing::{debug, error, info, warn};
use utils::current_timestamp;
pub (crate) const CHANNEL_CAPACITY: usize = 65536; // 64k capacity for Bakery commands
pub use commands::BakeryCommands;
use lqos_config::Config;
pub static BAKERY_SENDER: OnceLock<Sender<BakeryCommands>> = OnceLock::new();
/// Starts the Bakery system, returning a channel sender for sending commands to the Bakery.
pub fn start_bakery() -> anyhow::Result<crossbeam_channel::Sender<BakeryCommands>> {
let (tx, rx) = crossbeam_channel::bounded(CHANNEL_CAPACITY);
if BAKERY_SENDER.set(tx.clone()).is_err() {
return Err(anyhow::anyhow!("Bakery sender is already initialized."));
}
std::thread::Builder::new()
.name("lqos_bakery".to_string())
.spawn(move || {
@@ -43,15 +49,104 @@ pub fn start_bakery() -> anyhow::Result<crossbeam_channel::Sender<BakeryCommands
fn bakery_main(rx: Receiver<BakeryCommands>) {
let Ok(config) = lqos_config::load_config() else {
error!("Failed to load configuration, exiting Bakery thread.");
return;
};
let lazy_queues_enabled = config.queues.lazy_queues.unwrap_or(false);
let queue_expiration_time_seconds = config.queues.lazy_expire_seconds.unwrap_or(600); // Default to 10 minutes
// Current operation batch
let mut batch = None;
while let Ok(command) = rx.recv() {
debug!("Bakery received command: {:?}", command);
match command {
BakeryCommands::StartBatch => {
batch = Some(Vec::new());
},
BakeryCommands::CommitBatch => {
let Ok(config) = lqos_config::load_config() else {
error!("Failed to load configuration, exiting Bakery thread.");
continue;
};
let new_batch = batch.take(); // Take the batch to avoid cloning
if let Some(new_batch) = new_batch {
process_batch(new_batch, &config);
}
batch = None; // Clear the batch after committing
},
BakeryCommands::MqSetup { .. } => {
if let Some(batch) = &mut batch {
batch.push(command.clone());
}
},
BakeryCommands::AddSite { .. } => {
if let Some(batch) = &mut batch {
batch.push(command.clone());
}
}
BakeryCommands::AddCircuit { .. } => {
if let Some(batch) = &mut batch {
batch.push(command.clone());
}
}
}
}
error!("Bakery thread exited unexpectedly.");
}
fn process_batch(batch: Vec<BakeryCommands>, config: &Arc<lqos_config::Config>) {
info!("Bakery: Processing batch of {} commands", batch.len());
let commands = batch
.into_iter()
.map(|b| b.to_commands(config))
.flatten()
.flatten()
.collect::<Vec<Vec<String>>>();
if write_command_file(&config, commands) {
// Something bad happened while writing the command file
return;
}
let path = Path::new(&config.lqos_directory)
.join("linux_tc_rust.txt");
let path_str = path.to_string_lossy().to_string();
info!("Bakery: Command file written successfully at {}", path_str);
// /sbin/tc -f -b linux_tc.txt
std::process::Command::new("/sbin/tc")
.arg("-f") // Force the command to run
.arg("-b") // Batch mode
.arg(path_str) // Path to the command file
.spawn()
.map_err(|e| error!("Failed to execute tc command: {}", e))
.ok(); // Ignore errors, as we just want to run the command
}
fn write_command_file(config: &&Arc<Config>, commands: Vec<Vec<String>>) -> bool {
// Output file
let path = Path::new(&config.lqos_directory)
.join("linux_tc_rust.txt");
let Ok(f) = File::create(&path) else {
error!("Failed to create output file: {}", path.display());
return true;
};
let mut f = BufWriter::new(f);
for line in commands {
for (idx, entry) in line.iter().enumerate() {
if let Err(e) = f.write_all(entry.as_bytes()) {
error!("Failed to write to output file: {}", e);
return true;
}
if idx < line.len() - 1 {
if let Err(e) = f.write_all(b" ") {
error!("Failed to write space to output file: {}", e);
return true;
}
}
}
let newline = "\n";
if let Err(e) = f.write_all(newline.as_bytes()) {
error!("Failed to write newline to output file: {}", e);
return true;
}
}
false
}
src/rust/lqos_bakery/src/tc_control.rs
-454
View File
@@ -1,454 +0,0 @@
//! Traffic Control (TC) command execution for LibreQoS
//!
//! This module mirrors LibreQoS.py's TC functionality with two distinct queue types:
//!
//! 1. **Structural Queues** (Sites/APs from network.json)
//! - Intermediate nodes in the HTB hierarchy
//! - Only have HTB classes, NO qdiscs
//! - Tracked by site_hash (i64 hash of site name)
//! - Created with `add_structural_htb_class()`
//!
//! 2. **Circuit Queues** (Customer circuits)
//! - Leaf nodes that shape actual traffic
//! - Have both HTB class AND CAKE/fq_codel qdisc
//! - Tracked by circuit_hash (i64 hash of circuit ID)
//! - Created with `add_circuit_htb_class()` + `add_circuit_qdisc()`
//!
//! The hashes are used for tracking in Phase 2+ but accepted/ignored in Phase 1.
const TC_COMMAND: &str = "/sbin/tc";
/// Execute a TC command or write it to a file for testing
///
/// # Arguments
/// * `args` - The TC command arguments (everything after /sbin/tc)
///
/// # Returns
/// * `Result<(), anyhow::Error>` - Returns Ok if successful, or an error if the command fails.
pub fn execute_tc_command(args: &[&str]) -> anyhow::Result<()> {
// Execute the actual command
let output = std::process::Command::new(TC_COMMAND)
.args(args)
.output()?;
if !output.status.success() {
return Err(anyhow::anyhow!(
"TC command failed: {}",
String::from_utf8_lossy(&output.stderr)
));
}
Ok(())
}
/// Clear all traffic control queues for a given network interface.
///
/// # Arguments
/// * `interface` - The name of the network interface to clear queues for.
///
/// # Returns
/// * `Result<(), anyhow::Error>` - Returns Ok if successful, or an error if the command fails.
pub fn clear_all_queues(interface: &str) -> anyhow::Result<()> {
execute_tc_command(&["qdisc", "delete", "dev", interface, "root"])
}
/// Check if the Multi-Queue (MQ) discipline is installed on a given network interface.
///
/// # Arguments
/// * `interface` - The name of the network interface to check for MQ installation.
///
/// # Returns
/// * `Result<bool, anyhow::Error>` - Returns Ok(true) if MQ is installed, Ok(false) if not, or an error if the command fails.
pub fn is_mq_installed(interface: &str) -> anyhow::Result<bool> {
let output = std::process::Command::new(TC_COMMAND)
.arg("qdisc")
.arg("show")
.arg("dev")
.arg(interface)
.arg("root")
.output()?;
if !output.status.success() {
return Err(anyhow::anyhow!(
"Failed to check MQ installation: {}",
String::from_utf8_lossy(&output.stderr)
));
}
let output_str = String::from_utf8_lossy(&output.stdout);
Ok(output_str.contains("mq"))
}
/// Replace the Multi-Queue (MQ) discipline on a given network interface.
///
/// # Arguments
/// * `interface` - The name of the network interface on which to replace the MQ discipline.
///
/// # Returns
/// * `Result<(), anyhow::Error>` - Returns Ok if the replacement is successful, or an error if the command fails.
pub fn replace_mq(interface: &str) -> anyhow::Result<()> {
// command = 'qdisc replace dev ' + thisInterface + ' root handle 7FFF: mq'
execute_tc_command(&["qdisc", "replace", "dev", interface, "root", "handle", "7FFF:", "mq"])
}
pub fn make_top_htb(interface: &str, queue: u32) -> anyhow::Result<()> {
// 'qdisc add dev ' + thisInterface + ' parent 7FFF:' + hex(queue+1) + ' handle ' + hex(queue+1) + ': htb default 2'
let queue_hex = format!("{:x}", queue + 1);
let parent = format!("7FFF:{}", queue_hex);
let handle = format!("{}:", queue_hex);
execute_tc_command(&[
"qdisc", "add", "dev", interface, "parent", &parent, "handle", &handle, "htb", "default", "2"
])
}
/// Format a rate in Mbps for TC commands with smart unit selection.
/// This matches Python's format_rate_for_tc function.
/// - Rates >= 1000 Mbps use 'gbit'
/// - Rates >= 1 Mbps use 'mbit'
/// - Rates < 1 Mbps use 'kbit'
pub fn format_rate_for_tc(rate_mbps: f64) -> String {
if rate_mbps >= 1000.0 {
format!("{:.1}gbit", rate_mbps / 1000.0)
} else if rate_mbps >= 1.0 {
format!("{:.1}mbit", rate_mbps)
} else {
format!("{:.0}kbit", rate_mbps * 1000.0)
}
}
/// Calculate quantum value based on bandwidth and r2q.
/// This matches Python's quantum function.
pub fn quantum(rate_mbps: f64, r2q: u64) -> u64 {
const MIN_QUANTUM: u64 = 1522;
let rate_in_bytes_per_second = (rate_mbps * 125_000.0) as u64;
let quantum = u64::max(MIN_QUANTUM, rate_in_bytes_per_second / r2q);
quantum
}
pub fn make_parent_class(interface: &str, queue: u32, mbps: f64, r2q: u64) -> anyhow::Result<()> {
// 'class add dev ' + thisInterface + ' parent ' + hex(queue+1) + ': classid ' + hex(queue+1) + ':1 htb rate '+ str(upstream_bandwidth_capacity_download_mbps()) + 'mbit ceil ' + str(upstream_bandwidth_capacity_download_mbps()) + 'mbit' + quantum(upstream_bandwidth_capacity_download_mbps())
let parent = format!("{:x}:", queue + 1);
let classid = format!("{:x}:1", queue + 1);
let rate_string = format_rate_for_tc(mbps);
let quantum_val = quantum(mbps, r2q);
execute_tc_command(&[
"class", "add", "dev", interface, "parent", &parent, "classid", &classid,
"htb", "rate", &rate_string, "ceil", &rate_string, "quantum", &quantum_val.to_string()
])
}
pub fn make_default_sqm_bucket(interface: &str, queue: u32, sqm:&[&str]) -> anyhow::Result<()> {
// command = 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':1 ' + sqm()
let parent = format!("{:x}:1", queue + 1);
let mut args = vec!["qdisc", "add", "dev", interface, "parent", &parent];
args.extend_from_slice(sqm);
execute_tc_command(&args)
}
pub fn make_default_class(interface: &str, queue: u32, mbps: f64, r2q: u64) -> anyhow::Result<()> {
// 'class add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':1 classid ' + hex(queue+1) + ':2 htb rate ' + str(round((upstream_bandwidth_capacity_download_mbps()-1)/4)) + 'mbit ceil ' + str(upstream_bandwidth_capacity_download_mbps()-1) + 'mbit prio 5' + quantum(upstream_bandwidth_capacity_download_mbps())
let parent = format!("{:x}:1", queue + 1);
let classid = format!("{:x}:2", queue + 1);
let mbps_quarter = (mbps - 1.0) / 4.0;
let mbps_minus_one = mbps - 1.0;
let quantum_val = quantum(mbps, r2q);
execute_tc_command(&[
"class", "add", "dev", interface, "parent", &parent,
"classid", &classid, "htb", "rate", &format_rate_for_tc(mbps_quarter),
"ceil", &format_rate_for_tc(mbps_minus_one), "prio", "5",
"quantum", &quantum_val.to_string()
])
}
pub fn make_default_class_sqm(interface: &str, queue: u32, sqm:&[&str]) -> anyhow::Result<()> {
// 'qdisc add dev ' + thisInterface + ' parent ' + hex(queue+1) + ':2 ' + sqm()
let parent = format!("{:x}:2", queue + 1);
let mut args = vec!["qdisc", "add", "dev", interface, "parent", &parent];
args.extend_from_slice(sqm);
execute_tc_command(&args)
}
// === Circuit Management Functions ===
/// Generic HTB class creation (used by specific functions below)
///
/// Note: You should typically use one of:
/// - `add_circuit_htb_class()` for circuits (leaf nodes with CAKE)
/// - `add_structural_htb_class()` for sites/APs (intermediate nodes)
/// - Direct calls only for special cases
pub fn add_htb_class(
interface: &str,
parent: &str,
classid: &str,
rate_mbps: f64,
ceil_mbps: f64,
prio: u32,
r2q: u64,
circuit_hash: Option<i64>, // Some(hash) for circuits, None for structural nodes
comment: Option<&str>
) -> anyhow::Result<()> {
// 'class add dev ' + interface + ' parent ' + parent + ' classid ' + classid +
// ' htb rate '+ rate + ' ceil '+ ceil + ' prio ' + prio + quantum + comment
let rate_str = format_rate_for_tc(rate_mbps);
let ceil_str = format_rate_for_tc(ceil_mbps);
let quantum_val = quantum(ceil_mbps, r2q); // Use ceil for quantum like Python
let prio_str = prio.to_string();
let quantum_str = quantum_val.to_string();
let args = vec![
"class", "add", "dev", interface, "parent", parent, "classid", classid,
"htb", "rate", &rate_str, "ceil", &ceil_str, "prio", &prio_str,
"quantum", &quantum_str
];
// Store circuit_hash for future use (Phase 2+)
if let Some(_hash) = circuit_hash {
// Will be used for tracking circuits in later phases
}
// Add comment if provided (though TC doesn't actually support comments)
if let Some(_comment) = comment {
// TC doesn't support comments in the command, but we include this for compatibility
}
execute_tc_command(&args)
}
/// Add a qdisc (CAKE/fq_codel) to a circuit class
///
/// This is ONLY for circuits (leaf nodes). Structural nodes (sites/APs)
/// do NOT get qdiscs - they only have HTB classes for hierarchy.
pub fn add_circuit_qdisc(
interface: &str,
parent_major: u32,
parent_minor: u32,
circuit_hash: i64, // Required for circuits
sqm_params: &[&str]
) -> anyhow::Result<()> {
// Validate that sqm_params is not empty to prevent incomplete TC commands
if sqm_params.is_empty() || (sqm_params.len() == 1 && sqm_params[0].trim().is_empty()) {
return Err(anyhow::anyhow!(
"Empty SQM parameters for circuit {}. Cannot create qdisc without specifying type (e.g., 'cake' or 'fq_codel')",
circuit_hash
));
}
// 'qdisc add dev ' + interface + ' parent ' + major + ':' + minor + ' ' + sqm
let parent = format!("{:x}:{:x}", parent_major, parent_minor);
let mut args = vec!["qdisc", "add", "dev", interface, "parent", &parent];
args.extend_from_slice(sqm_params);
// Store circuit_hash for future use (Phase 2+)
let _ = circuit_hash; // Will be used for tracking circuits in later phases
execute_tc_command(&args)
}
/// Delete all queues on an interface (used in clearPriorSettings)
pub fn delete_root_qdisc(interface: &str) -> anyhow::Result<()> {
// 'tc qdisc delete dev ' + interface + ' root'
execute_tc_command(&["qdisc", "delete", "dev", interface, "root"])
}
/// Check if MQ is installed on an interface (for clearPriorSettings logic)
pub fn has_mq_qdisc(interface: &str) -> anyhow::Result<bool> {
// This is already implemented as is_mq_installed
is_mq_installed(interface)
}
/// Create an HTB class specifically for a circuit (with CAKE shaper)
/// Circuits are leaf nodes that shape actual customer traffic
pub fn add_circuit_htb_class(
interface: &str,
parent: &str,
classid: &str,
rate_mbps: f64,
ceil_mbps: f64,
circuit_hash: i64,
comment: Option<&str>,
r2q: u64,
) -> anyhow::Result<()> {
add_htb_class(
interface,
parent,
classid,
rate_mbps,
ceil_mbps,
3, // Priority 3 for circuits (from Python)
r2q,
Some(circuit_hash),
comment,
)
}
/// Create an HTB class for a structural node (site/AP from network.json)
/// These are intermediate nodes in the hierarchy, NOT leaf nodes
pub fn add_structural_htb_class(
interface: &str,
parent: &str,
classid: &str,
rate_mbps: f64,
ceil_mbps: f64,
site_hash: i64, // Hash of site name from network.json
r2q: u64,
) -> anyhow::Result<()> {
// Store site_hash for future tracking (Phase 2+)
let _ = site_hash;
add_htb_class(
interface,
parent,
classid,
rate_mbps,
ceil_mbps,
3, // Priority 3 for structural nodes (from Python)
r2q,
None, // Not a circuit
None, // No comment
)
}
/// Apply CAKE RTT adjustments for low bandwidth circuits
/// This matches Python's sqmFixupRate function but handles fractional speeds
pub fn sqm_fixup_rate(rate_mbps: f64, sqm: &str) -> String {
// If we aren't using cake, just return the sqm string
if !sqm.starts_with("cake") || sqm.contains("rtt") {
return sqm.to_string();
}
// Based on: 1 MTU is 1500 bytes, or 12,000 bits.
// At 1 Mbps, (1,000 bits per ms) transmitting an MTU takes 12ms. Add 3ms for overhead, and we get 15ms.
// So 15ms divided by 5 (for 1%) multiplied by 100 yields 300ms.
// Use ranges for fractional speed support
let rtt_suffix = if rate_mbps <= 1.5 {
" rtt 300ms"
} else if rate_mbps <= 2.5 {
" rtt 180ms"
} else if rate_mbps <= 3.5 {
" rtt 140ms"
} else if rate_mbps <= 4.5 {
" rtt 120ms"
} else {
""
};
format!("{}{}", sqm, rtt_suffix)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_format_rate_for_tc() {
// Test exact matches with Python's format_rate_for_tc
// Rates >= 1000 Mbps use gbit
assert_eq!(format_rate_for_tc(1000.0), "1.0gbit");
assert_eq!(format_rate_for_tc(1500.0), "1.5gbit");
assert_eq!(format_rate_for_tc(10000.0), "10.0gbit");
// Rates >= 1 Mbps use mbit
assert_eq!(format_rate_for_tc(1.0), "1.0mbit");
assert_eq!(format_rate_for_tc(1.5), "1.5mbit");
assert_eq!(format_rate_for_tc(10.0), "10.0mbit");
assert_eq!(format_rate_for_tc(999.9), "999.9mbit");
// Rates < 1 Mbps use kbit
assert_eq!(format_rate_for_tc(0.5), "500kbit");
assert_eq!(format_rate_for_tc(0.1), "100kbit");
assert_eq!(format_rate_for_tc(0.768), "768kbit");
}
#[test]
fn test_quantum() {
// Test quantum calculation with various r2q values
// MIN_QUANTUM is 1522
// Python: quantum(1.0, R2Q=10) -> 12500
assert_eq!(quantum(1.0, 10), 12500);
// Python: quantum(10.0, R2Q=10) -> 125000
assert_eq!(quantum(10.0, 10), 125000);
// Python: quantum(100.0, R2Q=10) -> 1250000
assert_eq!(quantum(100.0, 10), 1250000);
// Python: quantum(1000.0, R2Q=21) -> 5952380
assert_eq!(quantum(1000.0, 21), 5952380);
// Test fractional rates
assert_eq!(quantum(1.5, 10), 18750);
assert_eq!(quantum(0.5, 10), 6250); // Python returns 6250, not MIN_QUANTUM
}
#[test]
fn test_sqm_fixup_rate() {
// Test SQM fixup for CAKE at low rates
let cake_base = "cake bandwidth 5mbit";
// Test exact and fractional rates
assert_eq!(sqm_fixup_rate(1.0, cake_base), "cake bandwidth 5mbit rtt 300ms");
assert_eq!(sqm_fixup_rate(1.5, cake_base), "cake bandwidth 5mbit rtt 300ms");
assert_eq!(sqm_fixup_rate(2.0, cake_base), "cake bandwidth 5mbit rtt 180ms");
assert_eq!(sqm_fixup_rate(2.5, cake_base), "cake bandwidth 5mbit rtt 180ms");
assert_eq!(sqm_fixup_rate(3.0, cake_base), "cake bandwidth 5mbit rtt 140ms");
assert_eq!(sqm_fixup_rate(3.5, cake_base), "cake bandwidth 5mbit rtt 140ms");
assert_eq!(sqm_fixup_rate(4.0, cake_base), "cake bandwidth 5mbit rtt 120ms");
assert_eq!(sqm_fixup_rate(4.5, cake_base), "cake bandwidth 5mbit rtt 120ms");
assert_eq!(sqm_fixup_rate(5.0, cake_base), "cake bandwidth 5mbit");
// Test non-CAKE qdiscs (should return unchanged)
let fq_codel = "fq_codel";
assert_eq!(sqm_fixup_rate(1.0, fq_codel), "fq_codel");
// Test CAKE with RTT already present (should return unchanged)
let cake_with_rtt = "cake bandwidth 5mbit rtt 100ms";
assert_eq!(sqm_fixup_rate(1.0, cake_with_rtt), "cake bandwidth 5mbit rtt 100ms");
}
#[test]
fn test_structural_vs_circuit_creation() {
// This test demonstrates the distinction between structural and circuit queues
let site_hash: i64 = 987654321; // Hash of "Site-A" or similar
let circuit_hash: i64 = 1234567890; // Hash of circuit ID
// 1. Structural node (site/AP) - only HTB class, no qdisc
let _ = add_structural_htb_class(
"eth0",
"1:1", // Parent is root or another structural node
"1:10", // This site's class
100.0, // Site's total bandwidth
100.0,
site_hash,
21,
);
// 2. Circuit (leaf) - HTB class + CAKE qdisc
let _ = add_circuit_htb_class(
"eth0",
"1:10", // Parent is the site class above
"1:100", // Circuit's class
10.5, // Circuit rate (fractional)
15.0, // Circuit ceil
circuit_hash,
Some("Customer ABC"),
21,
);
// Only circuits get qdiscs (CAKE shaper)
let _ = add_circuit_qdisc(
"eth0",
1,
100,
circuit_hash,
&["cake", "bandwidth", "15mbit"],
);
// Structural nodes do NOT get qdiscs - they're just HTB hierarchy
}
}
src/rust/lqos_bus/src/bus/request.rs
+40
View File
@@ -227,6 +227,46 @@ pub enum BusRequest {
/// Finish a blackboard session
BlackboardFinish,
// lqos_bakery requests
/// Start a bakery session
BakeryStart,
/// Request a bakery commit
BakeryCommit,
/// Setup the MQ top
BakeryMqSetup { queues_available: usize, stick_offset: usize },
/// Add a site to the bakery
BakeryAddSite {
site_hash: i64,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
},
/// Add a circuit to the bakery
BakeryAddCircuit {
circuit_hash: i64,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
class_major: String,
up_class_major: String,
sqm_down: String,
sqm_up: String,
comment: String,
},
}
/// Defines the parts of the blackboard
src/rust/lqos_python/src/lib.rs
+210
View File
@@ -105,6 +105,7 @@ fn liblqos_python(_py: Python, m: &PyModule) -> PyResult<()> {
m.add_wrapped(wrap_pyfunction!(client_bandwidth_multiplier))?;
m.add_wrapped(wrap_pyfunction!(calculate_hash))?;
m.add_class::<Bakery>()?;
Ok(())
}
@@ -825,3 +826,212 @@ fn calculate_hash() -> PyResult<i64> {
Ok(hash)
}
////////////////////////////// The Bakery class //////////////////////////////
#[derive(Clone)]
enum BakeryCommands {
StartBatch,
Commit,
MqSetup { queues_available: usize, stick_offset: usize },
AddSite {
site_hash: i64,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
},
AddCircuit {
circuit_hash: i64,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
class_major: String,
up_class_major: String,
sqm_down: String,
sqm_up: String,
comment: String,
}
}
#[pyclass]
pub struct Bakery {
queue: Vec<BakeryCommands>,
}
#[pymethods]
impl Bakery {
#[new]
pub fn new() -> PyResult<Self> {
Ok(Self {
queue: Vec::new(),
})
}
pub fn start_batch(&mut self) -> PyResult<()> {
self.queue.push(BakeryCommands::StartBatch);
Ok(())
}
pub fn commit(&mut self) -> PyResult<()> {
self.queue.push(BakeryCommands::Commit);
// Send the commands batched up to the bus
let queue = self.queue.clone();
let handle = std::thread::spawn(move || {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build().unwrap()
.block_on(async {
let mut bus = lqos_bus::LibreqosBusClient::new().await.unwrap();
let chunks = queue.chunks(512);
for chunk in chunks {
let mut requests = Vec::new();
for msg in chunk {
match msg {
BakeryCommands::StartBatch => requests.push(BusRequest::BakeryStart),
BakeryCommands::Commit => requests.push(BusRequest::BakeryCommit),
BakeryCommands::MqSetup { queues_available, stick_offset } => {
requests.push(BusRequest::BakeryMqSetup {
queues_available: *queues_available,
stick_offset: *stick_offset,
});
},
BakeryCommands::AddSite { site_hash, parent_class_id, up_parent_class_id, class_minor, download_bandwidth_min, upload_bandwidth_min, download_bandwidth_max, upload_bandwidth_max, quantum_down, quantum_up } => {
let command = BusRequest::BakeryAddSite {
site_hash: *site_hash,
parent_class_id: parent_class_id.clone(),
up_parent_class_id: up_parent_class_id.clone(),
class_minor: class_minor.clone(),
download_bandwidth_min: download_bandwidth_min.clone(),
upload_bandwidth_min: upload_bandwidth_min.clone(),
download_bandwidth_max: download_bandwidth_max.clone(),
upload_bandwidth_max: upload_bandwidth_max.clone(),
quantum_down: quantum_down.trim().to_string(),
quantum_up: quantum_up.trim().to_string(),
};
requests.push(command);
}
BakeryCommands::AddCircuit { circuit_hash, parent_class_id, up_parent_class_id, class_minor, download_bandwidth_min, upload_bandwidth_min, download_bandwidth_max, upload_bandwidth_max, quantum_down, quantum_up, class_major, up_class_major, sqm_down, sqm_up, comment } => {
let command = BusRequest::BakeryAddCircuit {
circuit_hash: *circuit_hash,
parent_class_id: parent_class_id.clone(),
up_parent_class_id: up_parent_class_id.clone(),
class_minor: class_minor.clone(),
download_bandwidth_min: download_bandwidth_min.clone(),
upload_bandwidth_min: upload_bandwidth_min.clone(),
download_bandwidth_max: download_bandwidth_max.clone(),
upload_bandwidth_max: upload_bandwidth_max.clone(),
quantum_down: quantum_down.trim().to_string(),
quantum_up: quantum_up.trim().to_string(),
class_major: class_major.trim().to_string(),
up_class_major: up_class_major.clone(),
sqm_down: sqm_down.clone(),
sqm_up: sqm_up.clone(),
comment: comment.clone(),
};
requests.push(command);
}
}
}
if let Err(e) = bus.request(requests).await {
eprintln!("Failed to send batch commands: {}", e);
} else {
println!("Sent a batch of commands to Bakery");
}
}
});
});
let _ = handle.join();
Ok(())
}
pub fn setup_mq(&mut self, queues_available: usize, stick_offset: usize) -> PyResult<()> {
self.queue.push(BakeryCommands::MqSetup { queues_available, stick_offset });
Ok(())
}
pub fn add_site(
&mut self,
site_name: String,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
) -> PyResult<()> {
let site_hash = lqos_utils::hash_to_i64(&site_name);
//println!("Name hash for site {site_name} is {site_hash}");
let command = BakeryCommands::AddSite {
site_hash,
parent_class_id,
up_parent_class_id,
class_minor,
download_bandwidth_min,
upload_bandwidth_min,
download_bandwidth_max,
upload_bandwidth_max,
quantum_down,
quantum_up,
};
self.queue.push(command);
Ok(())
}
pub fn add_circuit(
&mut self,
circuit_name: String,
parent_class_id: String,
up_parent_class_id: String,
class_minor: String,
download_bandwidth_min: String,
upload_bandwidth_min: String,
download_bandwidth_max: String,
upload_bandwidth_max: String,
quantum_down: String,
quantum_up: String,
class_major: String,
up_class_major: String,
sqm_down: String,
sqm_up: String,
comment: String,
) -> PyResult<()> {
let circuit_hash = lqos_utils::hash_to_i64(&circuit_name);
//println!("Name: {circuit_name}, hash: {circuit_hash}");
let command = BakeryCommands::AddCircuit {
circuit_hash,
parent_class_id,
up_parent_class_id,
class_minor,
download_bandwidth_min,
upload_bandwidth_min,
download_bandwidth_max,
upload_bandwidth_max,
quantum_down,
quantum_up,
class_major,
up_class_major,
sqm_down,
sqm_up,
comment,
};
self.queue.push(command);
Ok(())
}
}
src/rust/lqosd/src/main.rs
+95 -12
View File
@@ -54,7 +54,6 @@ use mimalloc::MiMalloc;
use crate::blackboard::{BLACKBOARD_SENDER, BlackboardCommand};
use std::sync::OnceLock;
static BAKERY_SENDER: OnceLock<crossbeam_channel::Sender<lqos_bakery::BakeryCommands>> = OnceLock::new();
use crate::remote_commands::start_remote_commands;
#[cfg(feature = "flamegraphs")]
use crate::shaped_devices_tracker::NETWORK_JSON;
@@ -149,14 +148,9 @@ fn main() -> Result<()> {
};
// Start the Bakery for TC command execution
match lqos_bakery::start_bakery() {
Ok(sender) => {
info!("Bakery started successfully");
let _ = BAKERY_SENDER.set(sender);
},
Err(e) => {
error!("Failed to start Bakery: {:?}", e);
}
let Ok(bakery_sender) = lqos_bakery::start_bakery() else {
error!("Failed to start Bakery, exiting.");
std::process::exit(1);
};
// Spawn tracking sub-systems
@@ -180,7 +174,7 @@ fn main() -> Result<()> {
long_term_stats_tx.clone(),
flow_tx,
system_usage_tx.clone(),
BAKERY_SENDER.get().cloned(),
bakery_sender.clone(),
)?;
spawn_queue_monitor()?;
lqos_sys::bpf_garbage_collector();
@@ -296,12 +290,15 @@ fn memory_debug() {
#[cfg(not(feature = "flamegraphs"))]
fn memory_debug() {}
fn handle_bus_requests(requests: &[BusRequest], responses: &mut Vec<BusResponse>) {
fn handle_bus_requests(
requests: &[BusRequest],
responses: &mut Vec<BusResponse>,
) {
for req in requests.iter() {
//println!("Request: {:?}", req);
BUS_REQUESTS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
responses.push(match req {
BusRequest::Ping => lqos_bus::BusResponse::Ack,
BusRequest::Ping => BusResponse::Ack,
BusRequest::GetCurrentThroughput => throughput_tracker::current_throughput(),
BusRequest::GetHostCounter => throughput_tracker::host_counters(),
BusRequest::GetTopNDownloaders { start, end } => {
@@ -429,6 +426,92 @@ fn handle_bus_requests(requests: &[BusRequest], responses: &mut Vec<BusResponse>
}
BusResponse::Ack
}
BusRequest::BakeryStart => {
if let Some(sender) = lqos_bakery::BAKERY_SENDER.get() {
let sender = sender.clone();
let _ = sender.send(lqos_bakery::BakeryCommands::StartBatch);
BusResponse::Ack
} else {
BusResponse::Fail("Bakery not initialized".to_string())
}
}
BusRequest::BakeryCommit => {
if let Some(sender) = lqos_bakery::BAKERY_SENDER.get() {
let sender = sender.clone();
let _ = sender.send(lqos_bakery::BakeryCommands::CommitBatch);
BusResponse::Ack
} else {
BusResponse::Fail("Bakery not initialized".to_string())
}
}
BusRequest::BakeryMqSetup { queues_available, stick_offset } => {
if let Some(sender) = lqos_bakery::BAKERY_SENDER.get() {
let sender = sender.clone();
let _ = sender.send(lqos_bakery::BakeryCommands::MqSetup {
queues_available: *queues_available,
stick_offset: *stick_offset
});
BusResponse::Ack
} else {
BusResponse::Fail("Bakery not initialized".to_string())
}
}
BusRequest::BakeryAddSite {
site_hash,
parent_class_id,
up_parent_class_id,
class_minor,
download_bandwidth_min,
upload_bandwidth_min,
download_bandwidth_max,
upload_bandwidth_max,
quantum_down,
quantum_up,
} => {
if let Some(sender) = lqos_bakery::BAKERY_SENDER.get() {
let sender = sender.clone();
let _ = sender.send(lqos_bakery::BakeryCommands::AddSite {
site_hash: *site_hash,
parent_class_id: parent_class_id.clone(),
up_parent_class_id: up_parent_class_id.clone(),
class_minor: class_minor.clone(),
download_bandwidth_min: download_bandwidth_min.clone(),
upload_bandwidth_min: upload_bandwidth_min.clone(),
download_bandwidth_max: download_bandwidth_max.clone(),
upload_bandwidth_max: upload_bandwidth_max.clone(),
quantum_down: quantum_down.clone(),
quantum_up: quantum_up.clone(),
});
BusResponse::Ack
} else {
BusResponse::Fail("Bakery not initialized".to_string())
}
}
BusRequest::BakeryAddCircuit { circuit_hash, parent_class_id, up_parent_class_id, class_minor, download_bandwidth_min, upload_bandwidth_min, download_bandwidth_max, upload_bandwidth_max, quantum_down, quantum_up, class_major, up_class_major, sqm_down, sqm_up, comment } => {
if let Some(sender) = lqos_bakery::BAKERY_SENDER.get() {
let sender = sender.clone();
let _ = sender.send(lqos_bakery::BakeryCommands::AddCircuit {
circuit_hash: *circuit_hash,
parent_class_id: parent_class_id.clone(),
up_parent_class_id: up_parent_class_id.clone(),
class_minor: class_minor.clone(),
download_bandwidth_min: download_bandwidth_min.clone(),
upload_bandwidth_min: upload_bandwidth_min.clone(),
download_bandwidth_max: download_bandwidth_max.clone(),
upload_bandwidth_max: upload_bandwidth_max.clone(),
quantum_down: quantum_down.clone(),
quantum_up: quantum_up.clone(),
class_major: class_major.clone(),
up_class_major: up_class_major.clone(),
sqm_down: sqm_down.clone(),
sqm_up: sqm_up.clone(),
comment: comment.clone(),
});
BusResponse::Ack
} else {
BusResponse::Fail("Bakery not initialized".to_string())
}
}
});
}
}
src/rust/lqosd/src/throughput_tracker/mod.rs
+3 -3
View File
@@ -43,7 +43,7 @@ pub fn spawn_throughput_monitor(
long_term_stats_tx: Sender<StatsUpdateMessage>,
netflow_sender: crossbeam_channel::Sender<(FlowbeeKey, (FlowbeeLocalData, FlowAnalysis))>,
system_usage_actor: crossbeam_channel::Sender<tokio::sync::oneshot::Sender<SystemStats>>,
bakery_sender: Option<crossbeam_channel::Sender<lqos_bakery::BakeryCommands>>,
bakery_sender: crossbeam_channel::Sender<lqos_bakery::BakeryCommands>,
) -> anyhow::Result<()> {
debug!("Starting the bandwidth monitor thread.");
std::thread::Builder::new()
@@ -106,7 +106,7 @@ fn throughput_task(
long_term_stats_tx: Sender<StatsUpdateMessage>,
netflow_sender: crossbeam_channel::Sender<(FlowbeeKey, (FlowbeeLocalData, FlowAnalysis))>,
system_usage_actor: crossbeam_channel::Sender<tokio::sync::oneshot::Sender<SystemStats>>,
bakery_sender: Option<crossbeam_channel::Sender<BakeryCommands>>,
bakery_sender: crossbeam_channel::Sender<BakeryCommands>,
) {
// Obtain the flow timeout from the config, default to 30 seconds
let timeout_seconds = if let Ok(config) = lqos_config::load_config() {
@@ -162,7 +162,7 @@ fn throughput_task(
timer_metrics.zero_throughput_and_rtt = timer_metrics.start.elapsed().as_secs_f64();
THROUGHPUT_TRACKER.copy_previous_and_reset_rtt();
timer_metrics.copy_previous_and_reset_rtt = timer_metrics.start.elapsed().as_secs_f64();
THROUGHPUT_TRACKER.apply_new_throughput_counters(&mut net_json_calc, bakery_sender.as_ref());
THROUGHPUT_TRACKER.apply_new_throughput_counters(&mut net_json_calc, bakery_sender.clone());
timer_metrics.apply_new_throughput_counters =
timer_metrics.start.elapsed().as_secs_f64();
THROUGHPUT_TRACKER.apply_flow_data(
src/rust/lqosd/src/throughput_tracker/tracking_data.rs
+4 -5
View File
@@ -131,7 +131,7 @@ impl ThroughputTracker {
pub(crate) fn apply_new_throughput_counters(
&self,
net_json_calc: &mut NetworkJson,
bakery_sender: Option<&crossbeam_channel::Sender<BakeryCommands>>,
bakery_sender: crossbeam_channel::Sender<BakeryCommands>,
) {
let self_cycle = self.cycle.load(std::sync::atomic::Ordering::Relaxed);
let mut raw_data = self.raw_data.lock().unwrap();
@@ -168,7 +168,7 @@ impl ThroughputTracker {
if entry.packets != entry.prev_packets {
entry.most_recent_cycle = self_cycle;
// Call to Bakery Update for existing traffic
if let (Some(sender), Some(circuit_hash)) = (bakery_sender, entry.circuit_hash) {
if let Some(circuit_hash) = entry.circuit_hash {
debug!("Sending UpdateCircuit for circuit_hash: {}", circuit_hash);
// TODO: Update goes here
} else if entry.circuit_hash.is_some() {
@@ -222,12 +222,11 @@ impl ThroughputTracker {
} else {
let (circuit_id, circuit_hash) = Self::lookup_circuit_id(xdp_ip);
// Call to Bakery Queue Creation for new circuits
if let (Some(sender), Some(circuit_hash)) = (bakery_sender, circuit_hash) {
if let Some(circuit_hash) = circuit_hash {
info!("NEW CIRCUIT DETECTED! Sending CreateCircuit for circuit_hash: {}", circuit_hash);
// TODO: Update goes here
} else {
debug!("No bakery sender or circuit_hash for new circuit (sender: {}, circuit_hash: {:?})",
bakery_sender.is_some(), circuit_hash);
debug!("No bakery sender or circuit_hash for new circuit (circuit_hash: {:?})", circuit_hash);
}
let mut entry = ThroughputEntry {
circuit_id: circuit_id.clone(),