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/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..d94004b5 --- /dev/null +++ b/src/rust/lqos_sys/src/bpf/common/kprobe_helper.h @@ -0,0 +1,171 @@ +#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 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). + __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); +} + +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( + 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/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/bpf/lqos_kern.c b/src/rust/lqos_sys/src/bpf/lqos_kern.c index b1199caa..f739de15 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,24 @@ // 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; + __u16 vlan_tci; + __u32 len; + unsigned char *data; +} __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 +536,41 @@ int flow_reader(struct bpf_iter__bpf_map_elem *ctx) return 0; } +/* 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 + struct sk_buff *skb = (struct sk_buff *)PT_REGS_PARM1(ctx); + + // If no SKB - bail out! + if (!skb) 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; + + __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; + + 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; +} + 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..4de922d9 100644 --- a/src/rust/lqos_sys/src/lqos_kernel.rs +++ b/src/rust/lqos_sys/src/lqos_kernel.rs @@ -282,12 +282,38 @@ 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, +} + 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 +329,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 +343,16 @@ 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 { + 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")); + } + link + } else { + std::ptr::null_mut() + }; // Configure CPU Maps let shaping_physical_cpus: Vec = match lqos_config::load_config() { @@ -477,7 +515,10 @@ pub fn attach_xdp_and_tc_to_interface( } } - Ok(skeleton) + Ok(AttachedPrograms { + skeleton, + kprobe_link, + }) } /// Safety: Direct calls to C functions 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); } } diff --git a/src/rust/lqosd/src/lts2_sys/control_channel.rs b/src/rust/lqosd/src/lts2_sys/control_channel.rs index f8c23869..0406c4de 100644 --- a/src/rust/lqosd/src/lts2_sys/control_channel.rs +++ b/src/rust/lqosd/src/lts2_sys/control_channel.rs @@ -1326,20 +1326,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/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..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 @@ -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: "Throughput", + 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: "Throughput", + 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..f193c2e8 100644 --- a/src/rust/lqosd/src/node_manager/static2/circuit.html +++ b/src/rust/lqosd/src/node_manager/static2/circuit.html @@ -135,6 +135,10 @@
+ Enqueued + + + Throughput @@ -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 5278170d..bcb149aa 100644 --- a/src/rust/lqosd/src/node_manager/static2/node_manager.css +++ b/src/rust/lqosd/src/node_manager/static2/node_manager.css @@ -556,6 +556,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); @@ -636,6 +640,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 5f91b2a4..ae17b5e8 100644 --- a/src/rust/lqosd/src/shaped_devices_tracker/mod.rs +++ b/src/rust/lqosd/src/shaped_devices_tracker/mod.rs @@ -561,6 +561,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 @@ -679,6 +680,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 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/stats_submission.rs b/src/rust/lqosd/src/throughput_tracker/stats_submission.rs index 0cd4cc48..8abbbf58 100644 --- a/src/rust/lqosd/src/throughput_tracker/stats_submission.rs +++ b/src/rust/lqosd/src/throughput_tracker/stats_submission.rs @@ -96,7 +96,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() { @@ -206,8 +206,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; @@ -284,11 +286,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; @@ -298,7 +300,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; @@ -307,7 +309,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/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..4b3415b3 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()), @@ -194,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 @@ -368,9 +383,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 +473,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; @@ -489,8 +508,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), @@ -567,10 +592,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 +1075,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 +1092,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 +1111,8 @@ impl ThroughputTracker { |( bytes_down, bytes_up, + actual_bytes_down, + actual_bytes_up, packets_down, packets_up, tcp_down, @@ -1089,6 +1125,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 +1138,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 +1188,25 @@ 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() + } + + #[allow(dead_code)] 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() }