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

Fix mapping hash contract and simplify throughput reduction

Browse Source
This commit is contained in:
Herbert Wolverson committed 2026-03-09 09:45:14 -05:00
1 parent dfc8100967
commit cefeac61c4
5 files changed
+185 -170
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/rust/lqos_bus/src/bus/request.rs
+4 -4
View File
@@ -116,14 +116,14 @@ pub enum BusRequest {
/// Hashed circuit identifier (from ShapedDevices.csv).
///
/// Defaults to `0` for backward compatibility.
#[serde(default)]
/// This is required because lqosd treats kernel-provided hashes as the
/// authoritative source for circuit attribution.
circuit_id: u64,
/// Hashed device identifier (from ShapedDevices.csv).
///
/// Defaults to `0` for backward compatibility.
#[serde(default)]
/// This is required because lqosd treats kernel-provided hashes as the
/// authoritative source for device attribution.
device_id: u64,
/// If true, this is a *second* flow for the same IP range on
src/rust/lqos_python/src/lib.rs
+77 -47
View File
@@ -775,48 +775,43 @@ fn parse_add_ip(
classid: &str,
cpu: &str,
upload: bool,
circuit_id: Option<&str>,
device_id: Option<&str>,
circuit_id: &str,
device_id: &str,
) -> Result<BusRequest> {
if !classid.contains(':') {
return Err(Error::msg(format!(
"Class id must be in the format (major):(minor), e.g. 1:12. Provided string: {classid}"
)));
}
let circuit_id = circuit_id
.and_then(|s| (!s.is_empty()).then(|| lqos_utils::hash_to_i64(s) as u64))
.unwrap_or(0);
let device_id = device_id
.and_then(|s| (!s.is_empty()).then(|| lqos_utils::hash_to_i64(s) as u64))
.unwrap_or(0);
let circuit_id = circuit_id.trim();
if circuit_id.is_empty() {
return Err(Error::msg("circuit_id is required"));
}
let device_id = device_id.trim();
if device_id.is_empty() {
return Err(Error::msg("device_id is required"));
}
Ok(BusRequest::MapIpToFlow {
ip_address: ip.to_string(),
tc_handle: TcHandle::from_string(classid)?,
cpu: read_hex_string(cpu)?, // Force HEX representation
circuit_id,
device_id,
circuit_id: lqos_utils::hash_to_i64(circuit_id) as u64,
device_id: lqos_utils::hash_to_i64(device_id) as u64,
upload,
})
}
/// Adds an IP address mapping
#[pyfunction(signature = (ip, classid, cpu, upload, circuit_id=None, device_id=None))]
#[pyfunction(signature = (ip, classid, cpu, upload, circuit_id, device_id))]
fn add_ip_mapping(
ip: String,
classid: String,
cpu: String, // In HEX
upload: bool,
circuit_id: Option<String>,
device_id: Option<String>,
circuit_id: String,
device_id: String,
) -> PyResult<()> {
let request = parse_add_ip(
&ip,
&classid,
&cpu,
upload,
circuit_id.as_deref(),
device_id.as_deref(),
);
let request = parse_add_ip(&ip, &classid, &cpu, upload, &circuit_id, &device_id);
if let Ok(request) = request {
run_query(vec![request]).unwrap();
Ok(())
@@ -837,24 +832,17 @@ impl BatchedCommands {
Ok(Self { batch: Vec::new() })
}
#[pyo3(signature = (ip, classid, cpu, upload, circuit_id=None, device_id=None))]
#[pyo3(signature = (ip, classid, cpu, upload, circuit_id, device_id))]
pub fn add_ip_mapping(
&mut self,
ip: String,
classid: String,
cpu: String,
upload: bool,
circuit_id: Option<String>,
device_id: Option<String>,
circuit_id: String,
device_id: String,
) -> PyResult<()> {
let request = parse_add_ip(
&ip,
&classid,
&cpu,
upload,
circuit_id.as_deref(),
device_id.as_deref(),
);
let request = parse_add_ip(&ip, &classid, &cpu, upload, &circuit_id, &device_id);
if let Ok(request) = request {
self.batch.push(request);
Ok(())
@@ -971,21 +959,49 @@ fn overrides_circuit_adjustments(py: Python<'_>) -> PyResult<Vec<PyObject>> {
for adj in overrides.circuit_adjustments().iter() {
let d = PyDict::new(py);
match adj {
lqos_overrides::CircuitAdjustment::CircuitAdjustSpeed { circuit_id, min_download_bandwidth, max_download_bandwidth, min_upload_bandwidth, max_upload_bandwidth } => {
lqos_overrides::CircuitAdjustment::CircuitAdjustSpeed {
circuit_id,
min_download_bandwidth,
max_download_bandwidth,
min_upload_bandwidth,
max_upload_bandwidth,
} => {
d.set_item("type", "circuit_adjust_speed")?;
d.set_item("circuit_id", circuit_id.clone())?;
if let Some(v) = min_download_bandwidth { d.set_item("min_download_bandwidth", *v)?; }
if let Some(v) = max_download_bandwidth { d.set_item("max_download_bandwidth", *v)?; }
if let Some(v) = min_upload_bandwidth { d.set_item("min_upload_bandwidth", *v)?; }
if let Some(v) = max_upload_bandwidth { d.set_item("max_upload_bandwidth", *v)?; }
if let Some(v) = min_download_bandwidth {
d.set_item("min_download_bandwidth", *v)?;
}
if let Some(v) = max_download_bandwidth {
d.set_item("max_download_bandwidth", *v)?;
}
if let Some(v) = min_upload_bandwidth {
d.set_item("min_upload_bandwidth", *v)?;
}
if let Some(v) = max_upload_bandwidth {
d.set_item("max_upload_bandwidth", *v)?;
}
}
lqos_overrides::CircuitAdjustment::DeviceAdjustSpeed { device_id, min_download_bandwidth, max_download_bandwidth, min_upload_bandwidth, max_upload_bandwidth } => {
lqos_overrides::CircuitAdjustment::DeviceAdjustSpeed {
device_id,
min_download_bandwidth,
max_download_bandwidth,
min_upload_bandwidth,
max_upload_bandwidth,
} => {
d.set_item("type", "device_adjust_speed")?;
d.set_item("device_id", device_id.clone())?;
if let Some(v) = min_download_bandwidth { d.set_item("min_download_bandwidth", *v)?; }
if let Some(v) = max_download_bandwidth { d.set_item("max_download_bandwidth", *v)?; }
if let Some(v) = min_upload_bandwidth { d.set_item("min_upload_bandwidth", *v)?; }
if let Some(v) = max_upload_bandwidth { d.set_item("max_upload_bandwidth", *v)?; }
if let Some(v) = min_download_bandwidth {
d.set_item("min_download_bandwidth", *v)?;
}
if let Some(v) = max_download_bandwidth {
d.set_item("max_download_bandwidth", *v)?;
}
if let Some(v) = min_upload_bandwidth {
d.set_item("min_upload_bandwidth", *v)?;
}
if let Some(v) = max_upload_bandwidth {
d.set_item("max_upload_bandwidth", *v)?;
}
}
lqos_overrides::CircuitAdjustment::RemoveCircuit { circuit_id } => {
d.set_item("type", "remove_circuit")?;
@@ -995,7 +1011,10 @@ fn overrides_circuit_adjustments(py: Python<'_>) -> PyResult<Vec<PyObject>> {
d.set_item("type", "remove_device")?;
d.set_item("device_id", device_id.clone())?;
}
lqos_overrides::CircuitAdjustment::ReparentCircuit { circuit_id, parent_node } => {
lqos_overrides::CircuitAdjustment::ReparentCircuit {
circuit_id,
parent_node,
} => {
d.set_item("type", "reparent_circuit")?;
d.set_item("circuit_id", circuit_id.clone())?;
d.set_item("parent_node", parent_node.clone())?;
@@ -1020,13 +1039,24 @@ fn overrides_network_adjustments(py: Python<'_>) -> PyResult<Vec<PyObject>> {
for adj in overrides.network_adjustments().iter() {
let d = PyDict::new(py);
match adj {
lqos_overrides::NetworkAdjustment::AdjustSiteSpeed { site_name, download_bandwidth_mbps, upload_bandwidth_mbps } => {
lqos_overrides::NetworkAdjustment::AdjustSiteSpeed {
site_name,
download_bandwidth_mbps,
upload_bandwidth_mbps,
} => {
d.set_item("type", "adjust_site_speed")?;
d.set_item("site_name", site_name.clone())?;
if let Some(v) = download_bandwidth_mbps { d.set_item("download_bandwidth_mbps", *v)?; }
if let Some(v) = upload_bandwidth_mbps { d.set_item("upload_bandwidth_mbps", *v)?; }
if let Some(v) = download_bandwidth_mbps {
d.set_item("download_bandwidth_mbps", *v)?;
}
if let Some(v) = upload_bandwidth_mbps {
d.set_item("upload_bandwidth_mbps", *v)?;
}
}
lqos_overrides::NetworkAdjustment::SetNodeVirtual { node_name, virtual_node } => {
lqos_overrides::NetworkAdjustment::SetNodeVirtual {
node_name,
virtual_node,
} => {
d.set_item("type", "set_node_virtual")?;
d.set_item("node_name", node_name.clone())?;
d.set_item("virtual", *virtual_node)?;
src/rust/lqosd/src/ip_mapping.rs
+5
View File
@@ -18,6 +18,11 @@ pub(crate) fn map_ip_to_flow(
circuit_id: u64,
device_id: u64,
) -> BusResponse {
if circuit_id == 0 || device_id == 0 {
return BusResponse::Fail(
"MapIpToFlow requires non-zero circuit_id and device_id hashes".to_string(),
);
}
expect_ack(lqos_sys::add_ip_to_tc(
ip_address, *tc_handle, cpu, upload, circuit_id, device_id,
))
src/rust/lqosd/src/throughput_tracker/tracking_data.rs
+76 -101
View File
@@ -61,6 +61,60 @@ struct CircuitHeatmapAggregate {
tcp_packets: DownUpOrder<u64>,
}
struct ReducedHostCounters {
bytes: DownUpOrder<u64>,
packets: DownUpOrder<u64>,
tcp_packets: DownUpOrder<u64>,
udp_packets: DownUpOrder<u64>,
icmp_packets: DownUpOrder<u64>,
last_seen: u64,
tc_handle: TcHandle,
circuit_hash: Option<i64>,
device_hash: Option<i64>,
}
impl ReducedHostCounters {
fn from_counters(counts: &[lqos_sys::HostCounter]) -> Self {
let mut bytes = DownUpOrder::zeroed();
let mut packets = DownUpOrder::zeroed();
let mut tcp_packets = DownUpOrder::zeroed();
let mut udp_packets = DownUpOrder::zeroed();
let mut icmp_packets = DownUpOrder::zeroed();
let mut last_seen = 0u64;
let mut meta_last_seen = 0u64;
let mut meta_tc_handle = 0u32;
let mut meta_circuit_id = 0u64;
let mut meta_device_id = 0u64;
for c in counts {
bytes.checked_add_direct(c.download_bytes, c.upload_bytes);
packets.checked_add_direct(c.download_packets, c.upload_packets);
tcp_packets.checked_add_direct(c.tcp_download_packets, c.tcp_upload_packets);
udp_packets.checked_add_direct(c.udp_download_packets, c.udp_upload_packets);
icmp_packets.checked_add_direct(c.icmp_download_packets, c.icmp_upload_packets);
last_seen = u64::max(last_seen, c.last_seen);
if c.last_seen > meta_last_seen {
meta_last_seen = c.last_seen;
meta_tc_handle = c.tc_handle;
meta_circuit_id = c.circuit_id;
meta_device_id = c.device_id;
}
}
Self {
bytes,
packets,
tcp_packets,
udp_packets,
icmp_packets,
last_seen,
tc_handle: TcHandle::from_u32(meta_tc_handle),
circuit_hash: (meta_circuit_id != 0).then_some(meta_circuit_id as i64),
device_hash: (meta_device_id != 0).then_some(meta_device_id as i64),
}
}
}
impl ThroughputTracker {
pub(crate) fn new() -> Self {
// The capacity used to be taken from MAX_TRACKED_IPs, but
@@ -402,63 +456,21 @@ impl ThroughputTracker {
let cache = SHAPED_DEVICE_HASH_CACHE.load();
let mut raw_data = self.raw_data.lock();
throughput_for_each(&mut |xdp_ip, counts| {
let reduced = ReducedHostCounters::from_counters(counts);
if let Some(entry) = raw_data.get_mut(xdp_ip) {
// Zero the counter, we have to do a per-CPU sum
entry.bytes = DownUpOrder::zeroed();
entry.packets = DownUpOrder::zeroed();
entry.tcp_packets = DownUpOrder::zeroed();
entry.udp_packets = DownUpOrder::zeroed();
entry.icmp_packets = DownUpOrder::zeroed();
let mut last_seen = 0u64;
// Choose the most recent metadata to reduce staleness when mappings change.
let mut meta_last_seen = 0u64;
let mut meta_tc_handle = 0u32;
let mut meta_circuit_id = 0u64;
let mut meta_device_id = 0u64;
// Sum the counts across CPUs (it's a per-CPU map)
for c in counts {
entry
.bytes
.checked_add_direct(c.download_bytes, c.upload_bytes);
entry
.packets
.checked_add_direct(c.download_packets, c.upload_packets);
entry
.tcp_packets
.checked_add_direct(c.tcp_download_packets, c.tcp_upload_packets);
entry
.udp_packets
.checked_add_direct(c.udp_download_packets, c.udp_upload_packets);
entry
.icmp_packets
.checked_add_direct(c.icmp_download_packets, c.icmp_upload_packets);
last_seen = u64::max(last_seen, c.last_seen);
if c.last_seen > meta_last_seen {
meta_last_seen = c.last_seen;
meta_tc_handle = c.tc_handle;
meta_circuit_id = c.circuit_id;
meta_device_id = c.device_id;
}
}
entry.last_seen = last_seen;
entry.bytes = reduced.bytes;
entry.packets = reduced.packets;
entry.tcp_packets = reduced.tcp_packets;
entry.udp_packets = reduced.udp_packets;
entry.icmp_packets = reduced.icmp_packets;
entry.last_seen = reduced.last_seen;
let new_tc_handle = TcHandle::from_u32(meta_tc_handle);
let new_circuit_hash = if meta_circuit_id != 0 {
Some(meta_circuit_id as i64)
} else {
None
};
let new_device_hash = if meta_device_id != 0 {
Some(meta_device_id as i64)
} else {
None
};
let hashes_changed =
entry.circuit_hash != new_circuit_hash || entry.device_hash != new_device_hash;
entry.tc_handle = new_tc_handle;
entry.circuit_hash = new_circuit_hash;
entry.device_hash = new_device_hash;
let hashes_changed = entry.circuit_hash != reduced.circuit_hash
|| entry.device_hash != reduced.device_hash;
entry.tc_handle = reduced.tc_handle;
entry.circuit_hash = reduced.circuit_hash;
entry.device_hash = reduced.device_hash;
if hashes_changed {
let shaped_device = Self::shaped_device_for_hashes(
&shaped,
@@ -523,45 +535,8 @@ impl ThroughputTracker {
}
}
} else {
let mut last_seen = 0u64;
let mut meta_last_seen = 0u64;
let mut meta_tc_handle = 0u32;
let mut meta_circuit_id = 0u64;
let mut meta_device_id = 0u64;
let mut total_bytes: DownUpOrder<u64> = DownUpOrder::zeroed();
let mut total_packets: DownUpOrder<u64> = DownUpOrder::zeroed();
let mut total_tcp_packets: DownUpOrder<u64> = DownUpOrder::zeroed();
let mut total_udp_packets: DownUpOrder<u64> = DownUpOrder::zeroed();
let mut total_icmp_packets: DownUpOrder<u64> = DownUpOrder::zeroed();
for c in counts {
total_bytes.checked_add_direct(c.download_bytes, c.upload_bytes);
total_packets.checked_add_direct(c.download_packets, c.upload_packets);
total_tcp_packets
.checked_add_direct(c.tcp_download_packets, c.tcp_upload_packets);
total_udp_packets
.checked_add_direct(c.udp_download_packets, c.udp_upload_packets);
total_icmp_packets
.checked_add_direct(c.icmp_download_packets, c.icmp_upload_packets);
last_seen = u64::max(last_seen, c.last_seen);
if c.last_seen > meta_last_seen {
meta_last_seen = c.last_seen;
meta_tc_handle = c.tc_handle;
meta_circuit_id = c.circuit_id;
meta_device_id = c.device_id;
}
}
let tc_handle = TcHandle::from_u32(meta_tc_handle);
let circuit_hash = if meta_circuit_id != 0 {
Some(meta_circuit_id as i64)
} else {
None
};
let device_hash = if meta_device_id != 0 {
Some(meta_device_id as i64)
} else {
None
};
let circuit_hash = reduced.circuit_hash;
let device_hash = reduced.device_hash;
let shaped_device =
Self::shaped_device_for_hashes(&shaped, &cache, device_hash, circuit_hash);
let circuit_id = shaped_device.map(|d| d.circuit_id.clone());
@@ -573,7 +548,7 @@ impl ThroughputTracker {
if config.queues.lazy_threshold_bytes.is_some() {
let threshold = config.queues.lazy_threshold_bytes.unwrap_or(0);
if total_bytes.down.saturating_add(total_bytes.up) < threshold {
if reduced.bytes.down.saturating_add(reduced.bytes.up) < threshold {
add = false;
}
}
@@ -597,23 +572,23 @@ impl ThroughputTracker {
),
first_cycle: self_cycle,
most_recent_cycle: 0,
bytes: total_bytes,
packets: total_packets,
bytes: reduced.bytes,
packets: reduced.packets,
prev_bytes: DownUpOrder::zeroed(),
prev_packets: DownUpOrder::zeroed(),
bytes_per_second: DownUpOrder::zeroed(),
packets_per_second: DownUpOrder::zeroed(),
tcp_packets: total_tcp_packets,
udp_packets: total_udp_packets,
icmp_packets: total_icmp_packets,
tcp_packets: reduced.tcp_packets,
udp_packets: reduced.udp_packets,
icmp_packets: reduced.icmp_packets,
prev_tcp_packets: DownUpOrder::zeroed(),
prev_udp_packets: DownUpOrder::zeroed(),
prev_icmp_packets: DownUpOrder::zeroed(),
tc_handle,
tc_handle: reduced.tc_handle,
rtt_buffer: RttBuffer::default(),
recent_rtt_data: [RttData::from_nanos(0); 60],
last_fresh_rtt_data_cycle: 0,
last_seen,
last_seen: reduced.last_seen,
tcp_retransmits: DownUpOrder::zeroed(),
prev_tcp_retransmits: DownUpOrder::zeroed(),
qoq: QoqScores::default(),
src/rust/xdp_iphash_to_cpu_cmdline/src/main.rs
+23 -18
View File
@@ -29,11 +29,11 @@ enum Commands {
/// Circuit ID (raw string). Hashed before being stored.
#[arg(long)]
circuit_id: Option<String>,
circuit_id: String,
/// Device ID (raw string). Hashed before being stored.
#[arg(long)]
device_id: Option<String>,
device_id: String,
/// Add "--upload 1" if you are using on-a-stick and need to map upload separately
#[arg(long)]
@@ -98,8 +98,8 @@ fn parse_add_ip(
classid: &str,
cpu: &str,
upload: &Option<String>,
circuit_id: &Option<String>,
device_id: &Option<String>,
circuit_id: &str,
device_id: &str,
) -> Result<BusRequest> {
//if ip.parse::<IpAddr>().is_err() {
// return Err(Error::msg(format!("Unable to parse IP address: {ip}")));
@@ -109,22 +109,20 @@ fn parse_add_ip(
"Class id must be in the format (major):(minor), e.g. 1:12. Provided string: {classid}"
)));
}
let circuit_id = circuit_id
.as_deref()
.filter(|s| !s.is_empty())
.map(|s| lqos_utils::hash_to_i64(s) as u64)
.unwrap_or(0);
let device_id = device_id
.as_deref()
.filter(|s| !s.is_empty())
.map(|s| lqos_utils::hash_to_i64(s) as u64)
.unwrap_or(0);
let circuit_id = circuit_id.trim();
if circuit_id.is_empty() {
return Err(Error::msg("--circuit_id is required"));
}
let device_id = device_id.trim();
if device_id.is_empty() {
return Err(Error::msg("--device_id is required"));
}
Ok(BusRequest::MapIpToFlow {
ip_address: ip.to_string(),
tc_handle: TcHandle::from_string(classid)?,
cpu: read_hex_string(cpu)?, // Force HEX representation
circuit_id,
device_id,
circuit_id: lqos_utils::hash_to_i64(circuit_id) as u64,
device_id: lqos_utils::hash_to_i64(device_id) as u64,
upload: upload.is_some(),
})
}
@@ -142,8 +140,15 @@ pub async fn main() -> Result<()> {
device_id,
upload,
}) => {
talk_to_server(parse_add_ip(&ip, &classid, &cpu, &upload, &circuit_id, &device_id)?)
.await?;
talk_to_server(parse_add_ip(
&ip,
&classid,
&cpu,
&upload,
&circuit_id,
&device_id,
)?)
.await?;
}
Some(Commands::Del { ip, upload }) => {
talk_to_server(BusRequest::DelIpFlow {