From 435d87b61ea0f09a55a544f72bdda3120799da4c Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Mon, 6 Apr 2026 10:51:05 -0500 Subject: [PATCH 1/9] WIP: Minimal kprobe loader with interface identification. --- src/rust/lqos_sys/src/bpf/lqos_kern.c | 40 +++++++++++++++++++ src/rust/lqos_sys/src/bpf/wrapper.c | 26 +++++++++++++ src/rust/lqos_sys/src/bpf/wrapper.h | 2 + src/rust/lqos_sys/src/kernel_wrapper.rs | 51 ++++++++++++++++++++----- src/rust/lqos_sys/src/lqos_kernel.rs | 26 ++++++++++++- 5 files changed, 134 insertions(+), 11 deletions(-) diff --git a/src/rust/lqos_sys/src/bpf/lqos_kern.c b/src/rust/lqos_sys/src/bpf/lqos_kern.c index b1199caa..95502646 100644 --- a/src/rust/lqos_sys/src/bpf/lqos_kern.c +++ b/src/rust/lqos_sys/src/bpf/lqos_kern.c @@ -1,8 +1,15 @@ /* SPDX-License-Identifier: GPL-2.0 */ // Minimal XDP program that passes all packets. // Used to verify XDP functionality. +#ifndef __TARGET_ARCH_x86 +#define __TARGET_ARCH_x86 1 +#endif + #include +#include #include +#include +#include #include #include #include @@ -51,6 +58,21 @@ // If it stays at 255, we have a configuration error. int direction = 255; +// Constant passed in during to load, containing +// either -1 (none) or the ifindex of the interface +// in each direction. Required because the kbprobe +// is global. +int to_internet_ifindex = -1; +int to_isp_ifindex = -1; + +struct net_device { + int ifindex; +} __attribute__((preserve_access_index)); + +struct sk_buff { + struct net_device *dev; +} __attribute__((preserve_access_index)); + // Also configured during loading. For "on a stick" support, // these are mapped to the respective VLAN facing directions. __be16 internet_vlan = 0; // Note: turn these into big-endian @@ -511,4 +533,22 @@ int flow_reader(struct bpf_iter__bpf_map_elem *ctx) return 0; } +/* Runs at packet transmit */ +SEC("kprobe/dev_hard_start_xmit") +int kprobe_xmit(struct pt_regs *ctx) { + struct sk_buff *skb = (struct sk_buff *)PT_REGS_PARM1(ctx); + int ifindex; + int tracked_if = to_internet_ifindex; + + if (!skb) return 0; + ifindex = BPF_CORE_READ(skb, dev, ifindex); + + if (ifindex != tracked_if) { + tracked_if = to_isp_ifindex; + if (tracked_if < 0 || ifindex != tracked_if) return 0; + } + + return 0; +} + char _license[] SEC("license") = "GPL"; diff --git a/src/rust/lqos_sys/src/bpf/wrapper.c b/src/rust/lqos_sys/src/bpf/wrapper.c index afdc2eac..f6df4ac0 100644 --- a/src/rust/lqos_sys/src/bpf/wrapper.c +++ b/src/rust/lqos_sys/src/bpf/wrapper.c @@ -247,6 +247,32 @@ out: return err; } +struct bpf_link *attach_xmit_kprobe(struct lqos_kern *obj) +{ + struct bpf_link *link; + long err; + + link = bpf_program__attach_kprobe(obj->progs.kprobe_xmit, false, "dev_hard_start_xmit"); + err = libbpf_get_error(link); + if (err) { + const char *msg = "unknown"; + if (err < 0) { + msg = strerror((int)-err); + } + fprintf(stderr, "bpf_program__attach_kprobe() fails (%ld: %s)\n", err, msg); + return NULL; + } + + return link; +} + +void destroy_bpf_link(struct bpf_link *link) +{ + if (link != NULL) { + bpf_link__destroy(link); + } +} + // Iterator code #include #include diff --git a/src/rust/lqos_sys/src/bpf/wrapper.h b/src/rust/lqos_sys/src/bpf/wrapper.h index e24a3dc0..4a6735c7 100644 --- a/src/rust/lqos_sys/src/bpf/wrapper.h +++ b/src/rust/lqos_sys/src/bpf/wrapper.h @@ -10,6 +10,8 @@ extern int tc_attach_egress(int ifindex, bool verbose, struct lqos_kern *obj); extern int tc_detach_egress(int ifindex, bool verbose, bool flush_hook, const char * ifname); extern int tc_attach_ingress(int ifindex, bool verbose, struct lqos_kern *obj); extern int tc_detach_ingress(int ifindex, bool verbose, bool flush_hook, const char * ifname); +extern struct bpf_link * attach_xmit_kprobe(struct lqos_kern *obj); +extern void destroy_bpf_link(struct bpf_link *link); extern __u64 max_tracker_ips(); extern void do_not_print(); int read_tp_buffer(struct bpf_program *prog, struct bpf_map *map); diff --git a/src/rust/lqos_sys/src/kernel_wrapper.rs b/src/rust/lqos_sys/src/kernel_wrapper.rs index 4a1217c4..935b55f3 100644 --- a/src/rust/lqos_sys/src/kernel_wrapper.rs +++ b/src/rust/lqos_sys/src/kernel_wrapper.rs @@ -1,5 +1,5 @@ use crate::lqos_kernel::{ - InterfaceDirection, attach_xdp_and_tc_to_interface, + AttachedPrograms, InterfaceDirection, attach_xdp_and_tc_to_interface, bpf::{self, ring_buffer_sample_fn}, unload_xdp_from_interface, }; @@ -11,6 +11,7 @@ use parking_lot::Mutex; /// world insists on it. pub(crate) struct LqosKernBpfWrapper { ptr: *mut bpf::lqos_kern, + kprobe_link: *mut bpf::bpf_link, } impl LqosKernBpfWrapper { @@ -22,6 +23,14 @@ impl LqosKernBpfWrapper { unsafe impl Sync for LqosKernBpfWrapper {} unsafe impl Send for LqosKernBpfWrapper {} +impl Drop for LqosKernBpfWrapper { + fn drop(&mut self) { + unsafe { + bpf::destroy_bpf_link(self.kprobe_link); + } + } +} + pub(crate) static BPF_SKELETON: Lazy>> = Lazy::new(|| Mutex::new(None)); @@ -58,21 +67,34 @@ impl LibreQoSKernels { to_isp: to_isp.to_string(), on_a_stick: false, }; - let skeleton = attach_xdp_and_tc_to_interface( + let to_internet_ifindex = + crate::lqos_kernel::interface_name_to_index(&kernel.to_internet)? as i32; + let to_isp_ifindex = crate::lqos_kernel::interface_name_to_index(&kernel.to_isp)? as i32; + let AttachedPrograms { + skeleton, + kprobe_link, + } = attach_xdp_and_tc_to_interface( &kernel.to_internet, + to_internet_ifindex, + to_isp_ifindex, + true, InterfaceDirection::Internet, heimdall_event_handler, flowbee_event_handler, )?; attach_xdp_and_tc_to_interface( &kernel.to_isp, + to_internet_ifindex, + to_isp_ifindex, + false, InterfaceDirection::IspNetwork, heimdall_event_handler, flowbee_event_handler, )?; - BPF_SKELETON - .lock() - .replace(LqosKernBpfWrapper { ptr: skeleton }); + BPF_SKELETON.lock().replace(LqosKernBpfWrapper { + ptr: skeleton, + kprobe_link, + }); Ok(kernel) } @@ -100,21 +122,32 @@ impl LibreQoSKernels { to_isp: String::new(), on_a_stick: true, }; - let skeleton = attach_xdp_and_tc_to_interface( + let stick_ifindex = + crate::lqos_kernel::interface_name_to_index(&kernel.to_internet)? as i32; + let AttachedPrograms { + skeleton, + kprobe_link, + } = attach_xdp_and_tc_to_interface( &kernel.to_internet, + stick_ifindex, + -1, + true, InterfaceDirection::OnAStick(internet_vlan, isp_vlan, stick_offset), heimdall_event_handler, flowbee_event_handler, )?; - BPF_SKELETON - .lock() - .replace(LqosKernBpfWrapper { ptr: skeleton }); + BPF_SKELETON.lock().replace(LqosKernBpfWrapper { + ptr: skeleton, + kprobe_link, + }); Ok(kernel) } } impl Drop for LibreQoSKernels { fn drop(&mut self) { + let skeleton = BPF_SKELETON.lock().take(); + drop(skeleton); if !self.on_a_stick { let _ = unload_xdp_from_interface(&self.to_internet); let _ = unload_xdp_from_interface(&self.to_isp); diff --git a/src/rust/lqos_sys/src/lqos_kernel.rs b/src/rust/lqos_sys/src/lqos_kernel.rs index 7094d44c..2bba94da 100644 --- a/src/rust/lqos_sys/src/lqos_kernel.rs +++ b/src/rust/lqos_sys/src/lqos_kernel.rs @@ -282,12 +282,20 @@ pub enum InterfaceDirection { OnAStick(u16, u16, u32), } +pub(crate) struct AttachedPrograms { + pub(crate) skeleton: *mut lqos_kern, + pub(crate) kprobe_link: *mut bpf::bpf_link, +} + pub fn attach_xdp_and_tc_to_interface( interface_name: &str, + to_internet_ifindex: i32, + to_isp_ifindex: i32, + attach_xmit_kprobe: bool, direction: InterfaceDirection, heimdall_event_handler: bpf::ring_buffer_sample_fn, flowbee_event_handler: bpf::ring_buffer_sample_fn, -) -> Result<*mut lqos_kern> { +) -> Result { check_root()?; // If ABI changes were made to pinned maps, ensure we do not silently reuse // incompatible versions that truncate struct values. @@ -303,6 +311,8 @@ pub fn attach_xdp_and_tc_to_interface( InterfaceDirection::IspNetwork => 2, InterfaceDirection::OnAStick(..) => 3, }; + (*(*skeleton).data).to_internet_ifindex = to_internet_ifindex; + (*(*skeleton).data).to_isp_ifindex = to_isp_ifindex; if let InterfaceDirection::OnAStick(internet, isp, stick_offset) = direction { (*(*skeleton).bss).internet_vlan = internet.to_be(); (*(*skeleton).bss).isp_vlan = isp.to_be(); @@ -315,6 +325,15 @@ pub fn attach_xdp_and_tc_to_interface( attach_xdp_best_available(interface_index, prog_fd, interface_name)?; skeleton }; + let kprobe_link = if attach_xmit_kprobe { + let link = unsafe { bpf::attach_xmit_kprobe(skeleton) }; + if link.is_null() { + return Err(Error::msg("Unable to attach kprobe to dev_hard_start_xmit")); + } + link + } else { + std::ptr::null_mut() + }; // Configure CPU Maps let shaping_physical_cpus: Vec = match lqos_config::load_config() { @@ -477,7 +496,10 @@ pub fn attach_xdp_and_tc_to_interface( } } - Ok(skeleton) + Ok(AttachedPrograms { + skeleton, + kprobe_link, + }) } /// Safety: Direct calls to C functions From 94c86cd8b332caf074c1cc8d1ede6c4c97e84f9f Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Mon, 6 Apr 2026 11:13:56 -0500 Subject: [PATCH 2/9] WIP: Bail in invalid interface, and warn the user if they have changed offload settings and measurements are incorrect because of GSO/TSO. --- src/rust/lqos_sys/src/bpf/lqos_kern.c | 13 ++++++------- src/rust/lqos_sys/src/lqos_kernel.rs | 19 +++++++++++++++++++ 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/src/rust/lqos_sys/src/bpf/lqos_kern.c b/src/rust/lqos_sys/src/bpf/lqos_kern.c index 95502646..267ce8d6 100644 --- a/src/rust/lqos_sys/src/bpf/lqos_kern.c +++ b/src/rust/lqos_sys/src/bpf/lqos_kern.c @@ -536,17 +536,16 @@ int flow_reader(struct bpf_iter__bpf_map_elem *ctx) /* Runs at packet transmit */ SEC("kprobe/dev_hard_start_xmit") int kprobe_xmit(struct pt_regs *ctx) { + // Locate the skbuff struct sk_buff *skb = (struct sk_buff *)PT_REGS_PARM1(ctx); - int ifindex; - int tracked_if = to_internet_ifindex; + // If no SKB - bail out! if (!skb) return 0; - ifindex = BPF_CORE_READ(skb, dev, ifindex); - if (ifindex != tracked_if) { - tracked_if = to_isp_ifindex; - if (tracked_if < 0 || ifindex != tracked_if) return 0; - } + // Which interface called us? + int ifindex = BPF_CORE_READ(skb, dev, ifindex); + // If ifindex is not equal to EITHER to_internet or to_isp - not relevant, exit. + if (ifindex != to_internet_ifindex && ifindex != to_isp_ifindex) return 0; return 0; } diff --git a/src/rust/lqos_sys/src/lqos_kernel.rs b/src/rust/lqos_sys/src/lqos_kernel.rs index 2bba94da..4de922d9 100644 --- a/src/rust/lqos_sys/src/lqos_kernel.rs +++ b/src/rust/lqos_sys/src/lqos_kernel.rs @@ -282,6 +282,24 @@ pub enum InterfaceDirection { OnAStick(u16, u16, u32), } +fn warn_if_tx_segmentation_offloads_enabled() { + let Ok(config) = lqos_config::load_config() else { + warn!( + "Unable to load config before attaching xmit kprobe; TX rates may be measured incorrectly if GSO/TSO offloads remain enabled." + ); + return; + }; + let disabled_offloads = &config.tuning.disable_offload; + let gso_disabled = disabled_offloads.iter().any(|feature| feature == "gso"); + let tso_disabled = disabled_offloads.iter().any(|feature| feature == "tso"); + if !gso_disabled || !tso_disabled { + warn!( + "Attaching xmit kprobe while TX segmentation offloads are not fully disabled (gso_disabled={}, tso_disabled={}); actual transmission rates may be measured incorrectly.", + gso_disabled, tso_disabled + ); + } +} + pub(crate) struct AttachedPrograms { pub(crate) skeleton: *mut lqos_kern, pub(crate) kprobe_link: *mut bpf::bpf_link, @@ -326,6 +344,7 @@ pub fn attach_xdp_and_tc_to_interface( skeleton }; let kprobe_link = if attach_xmit_kprobe { + warn_if_tx_segmentation_offloads_enabled(); let link = unsafe { bpf::attach_xmit_kprobe(skeleton) }; if link.is_null() { return Err(Error::msg("Unable to attach kprobe to dev_hard_start_xmit")); From 2352e93059e55bc76ff65a0732bd45213e58b2f8 Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Tue, 7 Apr 2026 10:03:18 -0500 Subject: [PATCH 3/9] WIP: Add JUST the upload/download bytes fields to the throughput map. --- src/rust/lqos_sys/src/bpf/common/throughput.h | 2 ++ src/rust/lqos_sys/src/throughput.rs | 8 +++++++- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/src/rust/lqos_sys/src/bpf/common/throughput.h b/src/rust/lqos_sys/src/bpf/common/throughput.h index 3312d44b..cd7bb395 100644 --- a/src/rust/lqos_sys/src/bpf/common/throughput.h +++ b/src/rust/lqos_sys/src/bpf/common/throughput.h @@ -12,6 +12,8 @@ struct host_counter { __u64 download_bytes; __u64 upload_bytes; + __u64 actual_download_bytes; + __u64 actual_upload_bytes; __u64 download_packets; __u64 upload_packets; __u64 tcp_download_packets; diff --git a/src/rust/lqos_sys/src/throughput.rs b/src/rust/lqos_sys/src/throughput.rs index 878eb5d5..347f66ba 100644 --- a/src/rust/lqos_sys/src/throughput.rs +++ b/src/rust/lqos_sys/src/throughput.rs @@ -11,6 +11,12 @@ pub struct HostCounter { /// Upload bytes counter (keeps incrementing) pub upload_bytes: u64, + /// Actually transmitted download bytes counter (keeps incrementing) + pub actual_download_bytes: u64, + + /// Actually transmitted upload bytes counter (keeps incrementing) + pub actual_upload_bytes: u64, + /// Download packets counter (keeps incrementing) pub download_packets: u64, @@ -62,6 +68,6 @@ mod test { #[test] fn host_counter_size() { - assert_eq!(std::mem::size_of::(), 112); + assert_eq!(std::mem::size_of::(), 128); } } From d0093e0dcaf6c562e1162175310b178528325c56 Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Tue, 7 Apr 2026 10:34:50 -0500 Subject: [PATCH 4/9] WIP: Minimal effective direction detection for kprobe. --- .../lqos_sys/src/bpf/common/kprobe_helper.h | 77 +++++++++++++++++++ src/rust/lqos_sys/src/bpf/lqos_kern.c | 13 ++++ 2 files changed, 90 insertions(+) create mode 100644 src/rust/lqos_sys/src/bpf/common/kprobe_helper.h diff --git a/src/rust/lqos_sys/src/bpf/common/kprobe_helper.h b/src/rust/lqos_sys/src/bpf/common/kprobe_helper.h new file mode 100644 index 00000000..42367c42 --- /dev/null +++ b/src/rust/lqos_sys/src/bpf/common/kprobe_helper.h @@ -0,0 +1,77 @@ +#pragma once + +// Kprobe-only helpers. +// IMPORTANT: Do not use these from XDP/TC paths; keep XDP verifier balance intact. + +// Minimal VLAN header representation for scanning 802.1Q/AD tags. +struct vlan_hdr_kprobe { + __be16 h_vlan_TCI; + __be16 h_vlan_encapsulated_proto; +}; + +static __always_inline __be16 kprobe_scan_current_vlan(struct sk_buff *skb) +{ + unsigned char *data = (unsigned char *)BPF_CORE_READ(skb, data); + if (!data) { + return 0; + } + + struct ethhdr eth = {0}; + if (bpf_probe_read_kernel(ð, sizeof(eth), data) < 0) { + return 0; + } + + __u16 eth_type = bpf_ntohs(eth.h_proto); + __u32 offset = sizeof(struct ethhdr); + __be16 current_vlan = 0; + + // Keep this very small and bounded. +#pragma unroll + for (int i = 0; i < 2; i++) { + if (eth_type != ETH_P_8021AD && eth_type != ETH_P_8021Q) { + break; + } + struct vlan_hdr_kprobe vlan = {0}; + if (bpf_probe_read_kernel(&vlan, sizeof(vlan), data + offset) < 0) { + break; + } + current_vlan = vlan.h_vlan_TCI; + eth_type = bpf_ntohs(vlan.h_vlan_encapsulated_proto); + offset += sizeof(struct vlan_hdr_kprobe); + } + + return current_vlan; +} + +static __always_inline __be16 kprobe_current_vlan(struct sk_buff *skb) +{ + // Prefer the skb metadata tag (matches TC egress behavior). + __u16 vlan_tci = BPF_CORE_READ(skb, vlan_tci); + if (vlan_tci) { + return bpf_htons(vlan_tci); + } + // Fall back to scanning the L2 header for non-accelerated VLAN tags. + return kprobe_scan_current_vlan(skb); +} + +// Determine effective direction for *egress* (dev_hard_start_xmit) packets. +// Returns 1 for download (to LAN), 2 for upload (to Internet). +static __always_inline __u8 kprobe_determine_effective_direction( + struct sk_buff *skb, + int ifindex, + int to_internet_ifindex, + int to_isp_ifindex, + __be16 internet_vlan +) +{ + // On-a-stick: both directions share one ifindex, so use VLAN to infer egress. + if (to_isp_ifindex < 0) { + __be16 vlan = kprobe_current_vlan(skb); + // Out to Internet => UPLOAD (2). Out to LAN/core => DOWNLOAD (1). + return (vlan == internet_vlan) ? 2 : 1; + } + + // Two-interface: infer direction from the egress interface. + // to_internet_ifindex carries uploads; to_isp_ifindex carries downloads. + return (ifindex == to_internet_ifindex) ? 2 : 1; +} diff --git a/src/rust/lqos_sys/src/bpf/lqos_kern.c b/src/rust/lqos_sys/src/bpf/lqos_kern.c index 267ce8d6..494c5e34 100644 --- a/src/rust/lqos_sys/src/bpf/lqos_kern.c +++ b/src/rust/lqos_sys/src/bpf/lqos_kern.c @@ -71,6 +71,8 @@ struct net_device { struct sk_buff { struct net_device *dev; + __u16 vlan_tci; + unsigned char *data; } __attribute__((preserve_access_index)); // Also configured during loading. For "on a stick" support, @@ -534,6 +536,7 @@ int flow_reader(struct bpf_iter__bpf_map_elem *ctx) } /* Runs at packet transmit */ +#include "common/kprobe_helper.h" SEC("kprobe/dev_hard_start_xmit") int kprobe_xmit(struct pt_regs *ctx) { // Locate the skbuff @@ -547,6 +550,16 @@ int kprobe_xmit(struct pt_regs *ctx) { // If ifindex is not equal to EITHER to_internet or to_isp - not relevant, exit. if (ifindex != to_internet_ifindex && ifindex != to_isp_ifindex) return 0; + __u8 effective_direction = kprobe_determine_effective_direction( + skb, + ifindex, + to_internet_ifindex, + to_isp_ifindex, + internet_vlan + ); + // Keep the computation (used by subsequent xmit accounting work). + if (effective_direction == 0) return 0; + return 0; } From 9947950d5ffcef5bb49458be518530640f9c81de Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Tue, 7 Apr 2026 10:44:10 -0500 Subject: [PATCH 5/9] WIP: Actual byte counting --- .../lqos_sys/src/bpf/common/kprobe_helper.h | 94 +++++++++++++++++++ src/rust/lqos_sys/src/bpf/lqos_kern.c | 10 ++ 2 files changed, 104 insertions(+) diff --git a/src/rust/lqos_sys/src/bpf/common/kprobe_helper.h b/src/rust/lqos_sys/src/bpf/common/kprobe_helper.h index 42367c42..d94004b5 100644 --- a/src/rust/lqos_sys/src/bpf/common/kprobe_helper.h +++ b/src/rust/lqos_sys/src/bpf/common/kprobe_helper.h @@ -43,6 +43,19 @@ static __always_inline __be16 kprobe_scan_current_vlan(struct sk_buff *skb) return current_vlan; } +static __always_inline void kprobe_encode_ipv4(__be32 addr, struct in6_addr *out_address) +{ + out_address->in6_u.u6_addr32[0] = (__u32)0xFFFFFFFF; + out_address->in6_u.u6_addr32[1] = (__u32)0xFFFFFFFF; + out_address->in6_u.u6_addr32[2] = (__u32)0xFFFFFFFF; + out_address->in6_u.u6_addr32[3] = addr; +} + +static __always_inline void kprobe_encode_ipv6(struct in6_addr *ipv6_address, struct in6_addr *out_address) +{ + __builtin_memcpy(&out_address->in6_u.u6_addr8, &ipv6_address->in6_u.u6_addr8, 16); +} + static __always_inline __be16 kprobe_current_vlan(struct sk_buff *skb) { // Prefer the skb metadata tag (matches TC egress behavior). @@ -54,6 +67,87 @@ static __always_inline __be16 kprobe_current_vlan(struct sk_buff *skb) return kprobe_scan_current_vlan(skb); } +static __always_inline bool kprobe_build_host_key( + struct sk_buff *skb, + __u8 effective_direction, + struct in6_addr *host_key +) +{ + unsigned char *data = (unsigned char *)BPF_CORE_READ(skb, data); + if (!data) { + return false; + } + + struct ethhdr eth = {0}; + if (bpf_probe_read_kernel(ð, sizeof(eth), data) < 0) { + return false; + } + + __u16 eth_type = bpf_ntohs(eth.h_proto); + __u32 offset = sizeof(struct ethhdr); + + // Walk up to two VLAN tags. +#pragma unroll + for (int i = 0; i < 2; i++) { + if (eth_type != ETH_P_8021AD && eth_type != ETH_P_8021Q) { + break; + } + struct vlan_hdr_kprobe vlan = {0}; + if (bpf_probe_read_kernel(&vlan, sizeof(vlan), data + offset) < 0) { + return false; + } + eth_type = bpf_ntohs(vlan.h_vlan_encapsulated_proto); + offset += sizeof(struct vlan_hdr_kprobe); + } + + struct in6_addr src_ip = {0}; + struct in6_addr dst_ip = {0}; + + if (eth_type == ETH_P_IP) { + struct iphdr iph = {0}; + if (bpf_probe_read_kernel(&iph, sizeof(iph), data + offset) < 0) { + return false; + } + kprobe_encode_ipv4(iph.saddr, &src_ip); + kprobe_encode_ipv4(iph.daddr, &dst_ip); + } else if (eth_type == ETH_P_IPV6) { + struct ipv6hdr ip6h = {0}; + if (bpf_probe_read_kernel(&ip6h, sizeof(ip6h), data + offset) < 0) { + return false; + } + kprobe_encode_ipv6(&ip6h.saddr, &src_ip); + kprobe_encode_ipv6(&ip6h.daddr, &dst_ip); + } else { + return false; + } + + // Match XDP semantics: host_key is always the customer-side IP. + if (effective_direction == 1) { + __builtin_memcpy(host_key, &dst_ip, sizeof(*host_key)); + } else { + __builtin_memcpy(host_key, &src_ip, sizeof(*host_key)); + } + return true; +} + +static __always_inline void kprobe_track_actual_bytes( + __u8 effective_direction, + struct in6_addr *host_key, + __u32 bytes +) +{ + struct host_counter *counter = (struct host_counter *)bpf_map_lookup_elem(&map_traffic, host_key); + if (!counter) { + return; + } + + if (effective_direction == 1) { + counter->actual_download_bytes += bytes; + } else { + counter->actual_upload_bytes += bytes; + } +} + // Determine effective direction for *egress* (dev_hard_start_xmit) packets. // Returns 1 for download (to LAN), 2 for upload (to Internet). static __always_inline __u8 kprobe_determine_effective_direction( diff --git a/src/rust/lqos_sys/src/bpf/lqos_kern.c b/src/rust/lqos_sys/src/bpf/lqos_kern.c index 494c5e34..f739de15 100644 --- a/src/rust/lqos_sys/src/bpf/lqos_kern.c +++ b/src/rust/lqos_sys/src/bpf/lqos_kern.c @@ -72,6 +72,7 @@ struct net_device { struct sk_buff { struct net_device *dev; __u16 vlan_tci; + __u32 len; unsigned char *data; } __attribute__((preserve_access_index)); @@ -560,6 +561,15 @@ int kprobe_xmit(struct pt_regs *ctx) { // Keep the computation (used by subsequent xmit accounting work). if (effective_direction == 0) return 0; + struct in6_addr host_key = {0}; + if (!kprobe_build_host_key(skb, effective_direction, &host_key)) return 0; + + __u32 bytes = BPF_CORE_READ(skb, len); + if (!bytes) return 0; + + // Only update existing host entries; the XDP/TC path owns insertion. + kprobe_track_actual_bytes(effective_direction, &host_key, bytes); + return 0; } From 4fd8f4471148309ab724e5f6a50d341376707e1b Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Tue, 7 Apr 2026 11:05:12 -0500 Subject: [PATCH 6/9] WIP: Sum actual transmission bytes. --- .../throughput_tracker/throughput_entry.rs | 3 ++ .../src/throughput_tracker/tracking_data.rs | 40 +++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/src/rust/lqosd/src/throughput_tracker/throughput_entry.rs b/src/rust/lqosd/src/throughput_tracker/throughput_entry.rs index 2cb0902f..ed984fc0 100644 --- a/src/rust/lqosd/src/throughput_tracker/throughput_entry.rs +++ b/src/rust/lqosd/src/throughput_tracker/throughput_entry.rs @@ -13,16 +13,19 @@ pub(crate) struct ThroughputEntry { pub(crate) first_cycle: u64, pub(crate) most_recent_cycle: u64, pub(crate) bytes: DownUpOrder, // 0 DL, 1 UL + pub(crate) actual_bytes: DownUpOrder, // 0 DL, 1 UL pub(crate) packets: DownUpOrder, // 0 DL, 1 UL pub(crate) tcp_packets: DownUpOrder, // 0 DL, 1 UL pub(crate) udp_packets: DownUpOrder, // 0 DL, 1 UL pub(crate) icmp_packets: DownUpOrder, // 0 DL, 1 UL pub(crate) prev_bytes: DownUpOrder, // Has to mirror + pub(crate) prev_actual_bytes: DownUpOrder, pub(crate) prev_packets: DownUpOrder, pub(crate) prev_tcp_packets: DownUpOrder, pub(crate) prev_udp_packets: DownUpOrder, pub(crate) prev_icmp_packets: DownUpOrder, pub(crate) bytes_per_second: DownUpOrder, + pub(crate) actual_bytes_per_second: DownUpOrder, pub(crate) packets_per_second: DownUpOrder, pub(crate) tc_handle: TcHandle, pub(crate) rtt_buffer: RttBuffer, diff --git a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs index f19db921..769731b4 100644 --- a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs +++ b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs @@ -53,11 +53,13 @@ pub struct ThroughputTracker { pub(crate) cycle: AtomicU64, pub(crate) raw_data: Mutex>, pub(crate) bytes_per_second: AtomicDownUp, + pub(crate) actual_bytes_per_second: AtomicDownUp, pub(crate) packets_per_second: AtomicDownUp, pub(crate) tcp_packets_per_second: AtomicDownUp, pub(crate) udp_packets_per_second: AtomicDownUp, pub(crate) icmp_packets_per_second: AtomicDownUp, pub(crate) shaped_bytes_per_second: AtomicDownUp, + pub(crate) shaped_actual_bytes_per_second: AtomicDownUp, pub(crate) circuit_heatmaps: Mutex>, pub(crate) circuit_qoq_heatmaps: Mutex>, pub(crate) global_heatmap: Mutex, @@ -74,6 +76,7 @@ struct CircuitHeatmapAggregate { struct ReducedHostCounters { bytes: DownUpOrder, + actual_bytes: DownUpOrder, packets: DownUpOrder, tcp_packets: DownUpOrder, udp_packets: DownUpOrder, @@ -87,6 +90,7 @@ struct ReducedHostCounters { impl ReducedHostCounters { fn from_counters(counts: &[lqos_sys::HostCounter]) -> Self { let mut bytes = DownUpOrder::zeroed(); + let mut actual_bytes = DownUpOrder::zeroed(); let mut packets = DownUpOrder::zeroed(); let mut tcp_packets = DownUpOrder::zeroed(); let mut udp_packets = DownUpOrder::zeroed(); @@ -99,6 +103,8 @@ impl ReducedHostCounters { for c in counts { bytes.checked_add_direct(c.download_bytes, c.upload_bytes); + actual_bytes + .checked_add_direct(c.actual_download_bytes, c.actual_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); @@ -114,6 +120,7 @@ impl ReducedHostCounters { Self { bytes, + actual_bytes, packets, tcp_packets, udp_packets, @@ -136,11 +143,13 @@ impl ThroughputTracker { cycle: AtomicU64::new(RETIRE_AFTER_SECONDS), raw_data: Mutex::default(), bytes_per_second: AtomicDownUp::zeroed(), + actual_bytes_per_second: AtomicDownUp::zeroed(), packets_per_second: AtomicDownUp::zeroed(), tcp_packets_per_second: AtomicDownUp::zeroed(), udp_packets_per_second: AtomicDownUp::zeroed(), icmp_packets_per_second: AtomicDownUp::zeroed(), shaped_bytes_per_second: AtomicDownUp::zeroed(), + shaped_actual_bytes_per_second: AtomicDownUp::zeroed(), circuit_heatmaps: Mutex::default(), circuit_qoq_heatmaps: Mutex::default(), global_heatmap: Mutex::new(TemporalHeatmap::new()), @@ -368,9 +377,12 @@ impl ThroughputTracker { raw_data.iter_mut().for_each(|(_k, v)| { if v.first_cycle < self_cycle { v.bytes_per_second = v.bytes.checked_sub_or_zero(v.prev_bytes); + v.actual_bytes_per_second = + v.actual_bytes.checked_sub_or_zero(v.prev_actual_bytes); v.packets_per_second = v.packets.checked_sub_or_zero(v.prev_packets); } v.prev_bytes = v.bytes; + v.prev_actual_bytes = v.actual_bytes; v.prev_packets = v.packets; v.prev_tcp_packets = v.tcp_packets; v.prev_udp_packets = v.udp_packets; @@ -455,6 +467,7 @@ impl ThroughputTracker { if let Some(entry) = raw_data.get_mut(xdp_ip) { // Zero the counter, we have to do a per-CPU sum entry.bytes = reduced.bytes; + entry.actual_bytes = reduced.actual_bytes; entry.packets = reduced.packets; entry.tcp_packets = reduced.tcp_packets; entry.udp_packets = reduced.udp_packets; @@ -567,10 +580,13 @@ impl ThroughputTracker { first_cycle: self_cycle, most_recent_cycle: 0, bytes: reduced.bytes, + actual_bytes: reduced.actual_bytes, packets: reduced.packets, prev_bytes: DownUpOrder::zeroed(), + prev_actual_bytes: DownUpOrder::zeroed(), prev_packets: DownUpOrder::zeroed(), bytes_per_second: DownUpOrder::zeroed(), + actual_bytes_per_second: DownUpOrder::zeroed(), packets_per_second: DownUpOrder::zeroed(), tcp_packets: reduced.tcp_packets, udp_packets: reduced.udp_packets, @@ -1047,11 +1063,13 @@ impl ThroughputTracker { pub(crate) fn update_totals(&self) { let current_cycle = self.cycle.load(std::sync::atomic::Ordering::Relaxed); self.bytes_per_second.set_to_zero(); + self.actual_bytes_per_second.set_to_zero(); self.packets_per_second.set_to_zero(); self.tcp_packets_per_second.set_to_zero(); self.udp_packets_per_second.set_to_zero(); self.icmp_packets_per_second.set_to_zero(); self.shaped_bytes_per_second.set_to_zero(); + self.shaped_actual_bytes_per_second.set_to_zero(); let raw_data = self.raw_data.lock(); raw_data .iter() @@ -1062,6 +1080,10 @@ impl ThroughputTracker { ( v.bytes.down.saturating_sub(v.prev_bytes.down), v.bytes.up.saturating_sub(v.prev_bytes.up), + v.actual_bytes + .down + .saturating_sub(v.prev_actual_bytes.down), + v.actual_bytes.up.saturating_sub(v.prev_actual_bytes.up), v.packets.down.saturating_sub(v.prev_packets.down), v.packets.up.saturating_sub(v.prev_packets.up), v.tcp_packets.down.saturating_sub(v.prev_tcp_packets.down), @@ -1077,6 +1099,8 @@ impl ThroughputTracker { |( bytes_down, bytes_up, + actual_bytes_down, + actual_bytes_up, packets_down, packets_up, tcp_down, @@ -1089,6 +1113,8 @@ impl ThroughputTracker { )| { self.bytes_per_second .checked_add_tuple((bytes_down, bytes_up)); + self.actual_bytes_per_second + .checked_add_tuple((actual_bytes_down, actual_bytes_up)); self.packets_per_second .checked_add_tuple((packets_down, packets_up)); self.tcp_packets_per_second @@ -1100,6 +1126,8 @@ impl ThroughputTracker { if shaped { self.shaped_bytes_per_second .checked_add_tuple((bytes_down, bytes_up)); + self.shaped_actual_bytes_per_second + .checked_add_tuple((actual_bytes_down, actual_bytes_up)); } }, ); @@ -1148,12 +1176,24 @@ impl ThroughputTracker { self.bytes_per_second.as_down_up().to_bits_from_bytes() } + #[allow(dead_code)] + pub(crate) fn actual_bits_per_second(&self) -> DownUpOrder { + self.actual_bytes_per_second.as_down_up().to_bits_from_bytes() + } + pub(crate) fn shaped_bits_per_second(&self) -> DownUpOrder { self.shaped_bytes_per_second .as_down_up() .to_bits_from_bytes() } + #[allow(dead_code)] + pub(crate) fn shaped_actual_bits_per_second(&self) -> DownUpOrder { + self.shaped_actual_bytes_per_second + .as_down_up() + .to_bits_from_bytes() + } + pub(crate) fn packets_per_second(&self) -> DownUpOrder { self.packets_per_second.as_down_up() } From 50c296eaa7331bd86fa6e4d7b0f076ecbaa7914f Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Tue, 7 Apr 2026 11:15:38 -0500 Subject: [PATCH 7/9] WIP: use actual bytes for tree and insight --- .../lqosd/src/lts2_sys/control_channel.rs | 20 +++++++++---------- .../throughput_tracker/stats_submission.rs | 16 ++++++++------- .../src/throughput_tracker/tracking_data.rs | 10 ++++++++-- 3 files changed, 26 insertions(+), 20 deletions(-) diff --git a/src/rust/lqosd/src/lts2_sys/control_channel.rs b/src/rust/lqosd/src/lts2_sys/control_channel.rs index 774b42ad..ba765f8d 100644 --- a/src/rust/lqosd/src/lts2_sys/control_channel.rs +++ b/src/rust/lqosd/src/lts2_sys/control_channel.rs @@ -1325,20 +1325,18 @@ async fn shaper_snapshot_streaming( request_id: u64, reply: tokio::sync::mpsc::Sender, ) -> anyhow::Result<()> { - // Mirror node_manager ticker throughput: fetch current throughput and send compact tuple data - use lqos_bus::BusResponse; - - let resp = crate::throughput_tracker::current_throughput(); - if let BusResponse::CurrentThroughput { - bits_per_second, - packets_per_second, - shaped_bits_per_second, - .. - } = resp + // Mirror node_manager ticker throughput: send compact tuple data. + // Note: field names are historical; values are bits per second. { + let bits_per_second = crate::throughput_tracker::THROUGHPUT_TRACKER.actual_bits_per_second(); + let shaped_bits_per_second = crate::throughput_tracker::THROUGHPUT_TRACKER + .shaped_actual_bits_per_second(); + let packets_per_second = crate::throughput_tracker::THROUGHPUT_TRACKER + .packets_per_second + .as_down_up(); + let message = messages::WsMessage::StreamingShaper { request_id, - // Note: field names are historical; values are bits per second bytes_down: bits_per_second.down, bytes_up: bits_per_second.up, shaped_bytes_down: shaped_bits_per_second.down, diff --git a/src/rust/lqosd/src/throughput_tracker/stats_submission.rs b/src/rust/lqosd/src/throughput_tracker/stats_submission.rs index 1c2b64ba..b8934351 100644 --- a/src/rust/lqosd/src/throughput_tracker/stats_submission.rs +++ b/src/rust/lqosd/src/throughput_tracker/stats_submission.rs @@ -90,7 +90,7 @@ pub(crate) fn submit_throughput_stats( THROUGHPUT_TRACKER.icmp_packets_per_second.get_down(), THROUGHPUT_TRACKER.icmp_packets_per_second.get_up(), ); - let bits_per_second = THROUGHPUT_TRACKER.bits_per_second(); + let bits_per_second = THROUGHPUT_TRACKER.actual_bits_per_second(); // Check that the stats haven't gone wonky and don't submit obviously bad data. if let Ok(config) = load_config() { @@ -202,8 +202,10 @@ pub(crate) fn submit_throughput_stats( } // Send top-level throughput stats to LTS2 - let bytes = THROUGHPUT_TRACKER.bytes_per_second.as_down_up(); - let shaped_bytes = THROUGHPUT_TRACKER.shaped_bytes_per_second.as_down_up(); + let bytes = THROUGHPUT_TRACKER.actual_bytes_per_second.as_down_up(); + let shaped_bytes = THROUGHPUT_TRACKER + .shaped_actual_bytes_per_second + .as_down_up(); let mut min_rtt = None; let mut max_rtt = None; let mut median_rtt = None; @@ -280,11 +282,11 @@ pub(crate) fn submit_throughput_stats( .raw_data .lock() .iter() - .filter(|(_k, h)| h.circuit_id.is_some() && h.bytes_per_second.not_zero()) + .filter(|(_k, h)| h.circuit_id.is_some() && h.actual_bytes_per_second.not_zero()) .for_each(|(_k, h)| { let mut crazy = false; if let Some((dl, ul)) = plan_lookup.get(&h.circuit_hash.unwrap_or(0)) - && (h.bytes_per_second.down > *dl || h.bytes_per_second.up > *ul) + && (h.actual_bytes_per_second.down > *dl || h.actual_bytes_per_second.up > *ul) { crazy_values.insert(h.circuit_hash.unwrap_or(0)); crazy = true; @@ -294,7 +296,7 @@ pub(crate) fn submit_throughput_stats( return; } if let Some(c) = circuit_throughput.get_mut(&h.circuit_hash.unwrap_or(0)) { - c.bytes += h.bytes_per_second; + c.bytes += h.actual_bytes_per_second; c.packets += h.packets_per_second; c.tcp_packets += h.tcp_packets; c.udp_packets += h.udp_packets; @@ -303,7 +305,7 @@ pub(crate) fn submit_throughput_stats( circuit_throughput.insert( h.circuit_hash.unwrap_or(0), CircuitThroughputTemp { - bytes: h.bytes_per_second, + bytes: h.actual_bytes_per_second, packets: h.packets_per_second, tcp_packets: h.tcp_packets, udp_packets: h.udp_packets, diff --git a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs index 769731b4..dd7c3030 100644 --- a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs +++ b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs @@ -502,8 +502,14 @@ impl ThroughputTracker { net_json_calc.add_throughput_cycle( parents, ( - entry.bytes.down.saturating_sub(entry.prev_bytes.down), - entry.bytes.up.saturating_sub(entry.prev_bytes.up), + entry + .actual_bytes + .down + .saturating_sub(entry.prev_actual_bytes.down), + entry + .actual_bytes + .up + .saturating_sub(entry.prev_actual_bytes.up), ), ( entry.packets.down.saturating_sub(entry.prev_packets.down), From 54395a55bf3780b62075aea785fe9e8528a30fbc Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Tue, 7 Apr 2026 11:50:59 -0500 Subject: [PATCH 8/9] WIP: Circuit page uses transmitted throughput --- src/rust/lqos_bus/src/ip_stats.rs | 3 + .../src/node_manager/js_build/src/circuit.js | 20 +++--- .../src/graphs/queuing_activity_waveform.js | 71 +++++++++++++++---- .../local_api/circuit_activity.rs | 2 + .../src/node_manager/static2/circuit.html | 8 ++- .../src/node_manager/static2/node_manager.css | 8 +++ .../ws/single_user_channels/circuit.rs | 9 +++ .../lqosd/src/shaped_devices_tracker/mod.rs | 2 + 8 files changed, 101 insertions(+), 22 deletions(-) diff --git a/src/rust/lqos_bus/src/ip_stats.rs b/src/rust/lqos_bus/src/ip_stats.rs index bf68742a..b4569b65 100644 --- a/src/rust/lqos_bus/src/ip_stats.rs +++ b/src/rust/lqos_bus/src/ip_stats.rs @@ -207,6 +207,9 @@ pub struct Circuit { pub ip: IpAddr, /// Current bytes-per-second passing through this host. pub bytes_per_second: DownUpOrder, + /// Current bytes-per-second actually transmitted for this host. + #[serde(default)] + pub actual_bytes_per_second: DownUpOrder, /// Median latency for this host at the current time. pub median_latency: Option, /// Current RTT p50 (nanoseconds), per direction. diff --git a/src/rust/lqosd/src/node_manager/js_build/src/circuit.js b/src/rust/lqosd/src/node_manager/js_build/src/circuit.js index d154837e..a5885d6e 100644 --- a/src/rust/lqosd/src/node_manager/js_build/src/circuit.js +++ b/src/rust/lqosd/src/node_manager/js_build/src/circuit.js @@ -371,6 +371,10 @@ function pushQueuingActivitySample() { down: currentDirectionValue(latestCircuitSummary?.bytes_per_second, "down", 0) * 8, up: currentDirectionValue(latestCircuitSummary?.bytes_per_second, "up", 0) * 8, }, + actualThroughputBps: { + down: currentDirectionValue(latestCircuitSummary?.actual_bytes_per_second, "down", 0) * 8, + up: currentDirectionValue(latestCircuitSummary?.actual_bytes_per_second, "up", 0) * 8, + }, ceilingBps: { down: currentDirectionValue(plan, "down", 0) * 1_000_000.0, up: currentDirectionValue(plan, "up", 0) * 1_000_000.0, @@ -1072,16 +1076,16 @@ function applyCircuitSummary(summary) { } if (speedometer) { speedometer.update( - currentDirectionValue(summary?.bytes_per_second, "down", 0) * 8, - currentDirectionValue(summary?.bytes_per_second, "up", 0) * 8, + currentDirectionValue(summary?.actual_bytes_per_second, "down", 0) * 8, + currentDirectionValue(summary?.actual_bytes_per_second, "up", 0) * 8, currentDirectionValue(plan, "down", 0), currentDirectionValue(plan, "up", 0) ); } if (totalThroughput) { totalThroughput.update( - currentDirectionValue(summary?.bytes_per_second, "down", 0) * 8, - currentDirectionValue(summary?.bytes_per_second, "up", 0) * 8 + currentDirectionValue(summary?.actual_bytes_per_second, "down", 0) * 8, + currentDirectionValue(summary?.actual_bytes_per_second, "up", 0) * 8 ); } if (totalRetransmits) { @@ -1106,8 +1110,8 @@ function applyDeviceLiveData(devices) { const throughputGraph = deviceGraphs["throughputGraph_" + device.device_id]; if (throughputGraph !== undefined) { throughputGraph.update( - toNumber(device.bytes_per_second?.down, 0) * 8, - toNumber(device.bytes_per_second?.up, 0) * 8 + toNumber(device.actual_bytes_per_second?.down, 0) * 8, + toNumber(device.actual_bytes_per_second?.up, 0) * 8 ); } @@ -1812,14 +1816,14 @@ function fillLiveDevices(devices) { if (throughputDown !== null) { throughputDown.innerHTML = formatThroughput( - toNumber(device.bytes_per_second?.down, 0) * 8, + toNumber(device.actual_bytes_per_second?.down, 0) * 8, toNumber(device.plan?.down, 0) ); } if (throughputUp !== null) { throughputUp.innerHTML = formatThroughput( - toNumber(device.bytes_per_second?.up, 0) * 8, + toNumber(device.actual_bytes_per_second?.up, 0) * 8, toNumber(device.plan?.up, 0) ); } diff --git a/src/rust/lqosd/src/node_manager/js_build/src/graphs/queuing_activity_waveform.js b/src/rust/lqosd/src/node_manager/js_build/src/graphs/queuing_activity_waveform.js index 812927bc..38150814 100644 --- a/src/rust/lqosd/src/node_manager/js_build/src/graphs/queuing_activity_waveform.js +++ b/src/rust/lqosd/src/node_manager/js_build/src/graphs/queuing_activity_waveform.js @@ -45,6 +45,7 @@ function throughputPaletteColor(direction, fallback) { function getWaveformTheme(direction = "down") { const isDark = document.documentElement.getAttribute("data-bs-theme") !== "light"; const throughputColor = throughputPaletteColor(direction, isDark ? "#4992ff" : "#d87c7c"); + const actualThroughputColor = window.graphPalette?.[2] || (isDark ? "#fddd60" : "#d7ab82"); if (isDark) { return { @@ -58,6 +59,8 @@ function getWaveformTheme(direction = "down") { throughputAreaTop: hexToRgba(throughputColor, 0.48), throughputAreaMid: hexToRgba(throughputColor, 0.28), throughputAreaBottom: "rgba(9, 20, 31, 0.03)", + actualThroughputLine: actualThroughputColor, + actualThroughputGlow: hexToRgba(actualThroughputColor, 0.26), rttLine: "#b7a5ff", rttGlow: "rgba(183, 165, 255, 0.32)", rttAreaTop: "rgba(183, 165, 255, 0.18)", @@ -83,6 +86,8 @@ function getWaveformTheme(direction = "down") { throughputAreaTop: hexToRgba(throughputColor, 0.24), throughputAreaMid: hexToRgba(throughputColor, 0.13), throughputAreaBottom: "rgba(255, 255, 255, 0.02)", + actualThroughputLine: actualThroughputColor, + actualThroughputGlow: hexToRgba(actualThroughputColor, 0.16), rttLine: "#6f63cf", rttGlow: "rgba(111, 99, 207, 0.2)", rttAreaTop: "rgba(111, 99, 207, 0.1)", @@ -426,7 +431,7 @@ export class QueuingActivityWaveform extends DashboardGraph { ], series: [ { - name: "Throughput", + name: "Enqueued", type: "line", xAxisIndex: 0, yAxisIndex: 0, @@ -448,6 +453,27 @@ export class QueuingActivityWaveform extends DashboardGraph { }, data: [], }, + { + name: "Transmitted", + type: "line", + xAxisIndex: 0, + yAxisIndex: 0, + showSymbol: false, + smooth: false, + step: "start", + lineStyle: { + width: 2.0, + type: "dashed", + color: this.colors.actualThroughputLine, + shadowBlur: 10, + shadowColor: this.colors.actualThroughputGlow, + }, + itemStyle: { + opacity: 1, + }, + data: [], + z: 5, + }, { name: "Ceiling Base", type: "line", @@ -561,19 +587,21 @@ export class QueuingActivityWaveform extends DashboardGraph { { offset: 0.45, color: this.colors.throughputAreaMid }, { offset: 1, color: this.colors.throughputAreaBottom }, ]); - this.option.series[1].lineStyle.color = this.colors.ceilingInactive; - this.option.series[1].lineStyle.shadowColor = this.colors.ceilingInactiveGlow; - this.option.series[2].lineStyle.color = this.colors.ceilingActive; - this.option.series[2].lineStyle.shadowColor = this.colors.ceilingActiveGlow; + this.option.series[1].lineStyle.color = this.colors.actualThroughputLine; + this.option.series[1].lineStyle.shadowColor = this.colors.actualThroughputGlow; + this.option.series[2].lineStyle.color = this.colors.ceilingInactive; + this.option.series[2].lineStyle.shadowColor = this.colors.ceilingInactiveGlow; this.option.series[3].lineStyle.color = this.colors.ceilingActive; this.option.series[3].lineStyle.shadowColor = this.colors.ceilingActiveGlow; - this.option.series[4].lineStyle.color = this.colors.rttLine; - this.option.series[4].lineStyle.shadowColor = this.colors.rttGlow; - this.option.series[4].areaStyle.color = new echarts.graphic.LinearGradient(0, 0, 0, 1, [ + this.option.series[4].lineStyle.color = this.colors.ceilingActive; + this.option.series[4].lineStyle.shadowColor = this.colors.ceilingActiveGlow; + this.option.series[5].lineStyle.color = this.colors.rttLine; + this.option.series[5].lineStyle.shadowColor = this.colors.rttGlow; + this.option.series[5].areaStyle.color = new echarts.graphic.LinearGradient(0, 0, 0, 1, [ { offset: 0, color: this.colors.rttAreaTop }, { offset: 1, color: this.colors.rttAreaBottom }, ]); - this.option.series[4].markArea.data = rttMarkAreas( + this.option.series[5].markArea.data = rttMarkAreas( this.colors, this.rttThresholds, this.option.yAxis[1].max || this.rttThresholds.red_ms, @@ -625,10 +653,10 @@ export class QueuingActivityWaveform extends DashboardGraph { setRttThresholds(rawThresholds) { this.rttThresholds = normalizeRttThresholds(rawThresholds); this.option.yAxis[1].max = this.rttThresholds.red_ms; - this.option.series[4].markArea.data = rttMarkAreas(this.colors, this.rttThresholds, this.rttThresholds.red_ms); + this.option.series[5].markArea.data = rttMarkAreas(this.colors, this.rttThresholds, this.rttThresholds.red_ms); this.chart.setOption({ yAxis: [{}, { max: this.rttThresholds.red_ms }], - series: [{}, {}, {}, {}, { markArea: { data: this.option.series[4].markArea.data } }], + series: [{}, {}, {}, {}, {}, { markArea: { data: this.option.series[5].markArea.data } }], }); this.render(); } @@ -640,6 +668,10 @@ export class QueuingActivityWaveform extends DashboardGraph { down: toNumber(sample.throughputBps?.down, 0), up: toNumber(sample.throughputBps?.up, 0), }, + actualThroughputBps: { + down: toNumber(sample.actualThroughputBps?.down, 0), + up: toNumber(sample.actualThroughputBps?.up, 0), + }, ceilingBps: { down: toNumber(sample.ceilingBps?.down, 0), up: toNumber(sample.ceilingBps?.up, 0), @@ -715,6 +747,13 @@ export class QueuingActivityWaveform extends DashboardGraph { windowStart, displayNow, ); + const actualThroughputData = directionalSeriesData( + this.samples, + this.direction, + "actualThroughputBps", + windowStart, + displayNow, + ); const rttData = directionalSeriesData( this.samples, this.direction, @@ -761,9 +800,17 @@ export class QueuingActivityWaveform extends DashboardGraph { ], series: [ { - name: "Throughput", + name: "Enqueued", data: throughputData, }, + { + name: "Transmitted", + data: actualThroughputData, + lineStyle: { + color: this.colors.actualThroughputLine, + shadowColor: this.colors.actualThroughputGlow, + }, + }, { name: "Ceiling Base", data: ceilingSeries.base, diff --git a/src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs b/src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs index 050116b2..0abc0e61 100644 --- a/src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs +++ b/src/rust/lqosd/src/node_manager/local_api/circuit_activity.rs @@ -20,6 +20,8 @@ const TRAFFIC_FLOW_HIDE_THRESHOLD_BPS: u32 = 1_048_576; pub struct CircuitSummaryData { pub circuit_id: String, pub bytes_per_second: DownUpOrder, + #[serde(default)] + pub actual_bytes_per_second: DownUpOrder, pub rtt_current_p50_nanos: DownUpOrder>, pub tcp_retransmit_sample: DownUpOrder, pub qoo_score: Option, diff --git a/src/rust/lqosd/src/node_manager/static2/circuit.html b/src/rust/lqosd/src/node_manager/static2/circuit.html index 13abb7e0..9b611428 100644 --- a/src/rust/lqosd/src/node_manager/static2/circuit.html +++ b/src/rust/lqosd/src/node_manager/static2/circuit.html @@ -135,7 +135,11 @@
- Throughput + Enqueued + + + + Transmitted @@ -164,7 +168,7 @@
-
Throughput
+
Enqueued throughput
-
diff --git a/src/rust/lqosd/src/node_manager/static2/node_manager.css b/src/rust/lqosd/src/node_manager/static2/node_manager.css index ccc6ff3e..b680b1cd 100644 --- a/src/rust/lqosd/src/node_manager/static2/node_manager.css +++ b/src/rust/lqosd/src/node_manager/static2/node_manager.css @@ -504,6 +504,10 @@ html[data-bs-theme="light"] .sidebar .navbar-toggler-icon { background: #32d3bd; box-shadow: 0 0 10px rgba(50, 211, 189, 0.24); } +.lqos-circuit-activity-legend-dot.is-throughput-actual { + background: #fddd60; + box-shadow: 0 0 10px rgba(253, 221, 96, 0.22); +} .lqos-circuit-activity-legend-dot.is-qoo { background: #8fb0ff; box-shadow: 0 0 10px rgba(143, 176, 255, 0.22); @@ -584,6 +588,10 @@ html[data-bs-theme="light"] .sidebar .navbar-toggler-icon { background: #129a87; box-shadow: 0 0 10px rgba(18, 154, 135, 0.16); } +[data-bs-theme="light"] .lqos-circuit-activity-legend-dot.is-throughput-actual { + background: #d7ab82; + box-shadow: 0 0 10px rgba(215, 171, 130, 0.14); +} [data-bs-theme="light"] .lqos-circuit-activity-legend-dot.is-qoo { background: #4f79d9; box-shadow: 0 0 10px rgba(79, 121, 217, 0.12); diff --git a/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs b/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs index ac8acc40..9a7a9a6b 100644 --- a/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs +++ b/src/rust/lqosd/src/node_manager/ws/single_user_channels/circuit.rs @@ -90,6 +90,14 @@ fn summarize_circuit_devices(circuit: &str, devices: &[Circuit]) -> CircuitSumma acc }); + let actual_bytes_per_second = devices + .iter() + .fold(DownUpOrder::default(), |mut acc, device| { + acc.down += device.actual_bytes_per_second.down; + acc.up += device.actual_bytes_per_second.up; + acc + }); + let tcp_retransmit_sample = down_up_retransmit_sample( DownUpOrder { down: devices @@ -133,6 +141,7 @@ fn summarize_circuit_devices(circuit: &str, devices: &[Circuit]) -> CircuitSumma CircuitSummaryData { circuit_id: circuit.to_string(), bytes_per_second, + actual_bytes_per_second, rtt_current_p50_nanos, tcp_retransmit_sample, qoo_score: qoo_score_for_circuit(circuit), diff --git a/src/rust/lqosd/src/shaped_devices_tracker/mod.rs b/src/rust/lqosd/src/shaped_devices_tracker/mod.rs index 3ee3ee15..f041efba 100644 --- a/src/rust/lqosd/src/shaped_devices_tracker/mod.rs +++ b/src/rust/lqosd/src/shaped_devices_tracker/mod.rs @@ -560,6 +560,7 @@ pub fn get_all_circuits() -> BusResponse { Circuit { ip: k.as_ip(), bytes_per_second: v.bytes_per_second, + actual_bytes_per_second: v.actual_bytes_per_second, median_latency: v.median_latency(), rtt_current_p50_nanos: DownUpOrder { down: v @@ -678,6 +679,7 @@ pub fn get_circuit_by_id(desired_circuit_id: String) -> BusResponse { Some(Circuit { ip: k.as_ip(), bytes_per_second: v.bytes_per_second, + actual_bytes_per_second: v.actual_bytes_per_second, median_latency: v.median_latency(), rtt_current_p50_nanos: DownUpOrder { down: v From 687c6b41f2ce082cf97638b8b03b4feda84f7413 Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Tue, 7 Apr 2026 12:20:47 -0500 Subject: [PATCH 9/9] WIP: Dashboard uses actual throughput --- .../src/graphs/queuing_activity_waveform.js | 4 ++-- .../lqosd/src/node_manager/static2/circuit.html | 2 +- src/rust/lqosd/src/throughput_tracker/mod.rs | 14 +++++++------- .../lqosd/src/throughput_tracker/tracking_data.rs | 11 +++++++++-- 4 files changed, 19 insertions(+), 12 deletions(-) diff --git a/src/rust/lqosd/src/node_manager/js_build/src/graphs/queuing_activity_waveform.js b/src/rust/lqosd/src/node_manager/js_build/src/graphs/queuing_activity_waveform.js index 38150814..8539949b 100644 --- a/src/rust/lqosd/src/node_manager/js_build/src/graphs/queuing_activity_waveform.js +++ b/src/rust/lqosd/src/node_manager/js_build/src/graphs/queuing_activity_waveform.js @@ -454,7 +454,7 @@ export class QueuingActivityWaveform extends DashboardGraph { data: [], }, { - name: "Transmitted", + name: "Throughput", type: "line", xAxisIndex: 0, yAxisIndex: 0, @@ -804,7 +804,7 @@ export class QueuingActivityWaveform extends DashboardGraph { data: throughputData, }, { - name: "Transmitted", + name: "Throughput", data: actualThroughputData, lineStyle: { color: this.colors.actualThroughputLine, diff --git a/src/rust/lqosd/src/node_manager/static2/circuit.html b/src/rust/lqosd/src/node_manager/static2/circuit.html index 9b611428..f193c2e8 100644 --- a/src/rust/lqosd/src/node_manager/static2/circuit.html +++ b/src/rust/lqosd/src/node_manager/static2/circuit.html @@ -139,7 +139,7 @@ - Transmitted + Throughput diff --git a/src/rust/lqosd/src/throughput_tracker/mod.rs b/src/rust/lqosd/src/throughput_tracker/mod.rs index fb4bd04f..d83898f0 100644 --- a/src/rust/lqosd/src/throughput_tracker/mod.rs +++ b/src/rust/lqosd/src/throughput_tracker/mod.rs @@ -310,9 +310,9 @@ fn throughput_task( pub fn current_throughput() -> BusResponse { let (bits_per_second, packets_per_second, shaped_bits_per_second, tcp_pps, udp_pps, icmp_pps) = { ( - THROUGHPUT_TRACKER.bits_per_second(), + THROUGHPUT_TRACKER.actual_bits_per_second(), THROUGHPUT_TRACKER.packets_per_second(), - THROUGHPUT_TRACKER.shaped_bits_per_second(), + THROUGHPUT_TRACKER.shaped_actual_bits_per_second(), THROUGHPUT_TRACKER.tcp_packets_per_second(), THROUGHPUT_TRACKER.udp_packets_per_second(), THROUGHPUT_TRACKER.icmp_packets_per_second(), @@ -370,7 +370,7 @@ pub fn top_n(start: u32, end: u32) -> BusResponse { .map(|(k, te)| { ( *k, - te.bytes_per_second, + te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, @@ -414,7 +414,7 @@ pub fn top_n_up(start: u32, end: u32) -> BusResponse { .map(|(k, te)| { ( *k, - te.bytes_per_second, + te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, @@ -567,7 +567,7 @@ pub fn worst_n(start: u32, end: u32) -> BusResponse { .map(|(k, te)| { ( *k, - te.bytes_per_second, + te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, @@ -612,7 +612,7 @@ pub fn worst_n_retransmits(start: u32, end: u32) -> BusResponse { .map(|(k, te)| { ( *k, - te.bytes_per_second, + te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, @@ -670,7 +670,7 @@ pub fn best_n(start: u32, end: u32) -> BusResponse { .map(|(k, te)| { ( *k, - te.bytes_per_second, + te.actual_bytes_per_second, te.packets_per_second, te.median_latency().unwrap_or(0.0), te.tc_handle, diff --git a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs index dd7c3030..4b3415b3 100644 --- a/src/rust/lqosd/src/throughput_tracker/tracking_data.rs +++ b/src/rust/lqosd/src/throughput_tracker/tracking_data.rs @@ -203,8 +203,14 @@ impl ThroughputTracker { continue; }; - let download_delta = entry.bytes.down.saturating_sub(entry.prev_bytes.down); - let upload_delta = entry.bytes.up.saturating_sub(entry.prev_bytes.up); + let download_delta = entry + .actual_bytes + .down + .saturating_sub(entry.prev_actual_bytes.down); + let upload_delta = entry + .actual_bytes + .up + .saturating_sub(entry.prev_actual_bytes.up); total_download_bytes = total_download_bytes.saturating_add(download_delta); total_upload_bytes = total_upload_bytes.saturating_add(upload_delta); total_tcp_packets.down = total_tcp_packets @@ -1187,6 +1193,7 @@ impl ThroughputTracker { self.actual_bytes_per_second.as_down_up().to_bits_from_bytes() } + #[allow(dead_code)] pub(crate) fn shaped_bits_per_second(&self) -> DownUpOrder { self.shaped_bytes_per_second .as_down_up()