Optimization with reducing copies and eliminating buffer grows (#77)

* Optimized build and disable kanal async

* adds packet store for optimized packet processing

* adapt packet_store for optimized packet processing

* Fix lint

* refactor: replace Mutex with RwLock for latest packet storage

* use THREAD_BUFFER in packet_store

* add branches as dependency

* fix ipv6 bug + adds unlikely branches

* use PacketStore directly inside egress and ingress for much better performance than any channel

* reuse string buffer instead of allocating new one each time

* adds reusable display buffer for packets

* optimize the packet_store even more (insomnia edition)

* fix fmt

* adds get and write_range_into optimized implementations

* add cacheguard dependency and refactor PacketStore to use CacheGuard for atomic operations to improve cpu cache hits

* optimize alert and synflood detection, add rustc-hash for FxHashmap

* optimize alert packet counting

* rollback link.rs change

* gracefully handle out-of-bounds access in packet_store methods

* add stack allocated ArrayVec optimization for packet_store

* fix: correct packet_index handling in Inspection::render

* clear alerts on next time frame

* strict range of write_range_into for packet display buffer to avoid over copy when data is being written at realtime

Author: fereidani

Cargo.lock

@@ -16,6 +16,9 @@ homepage = "https://github.com/pythops/oryx"
 network-types = { git = "https://github.com/vadorovsky/network-types", rev = "b78424c" }
 
 [profile.release]
-lto = "fat"
-strip = true
+opt-level = 3
+debug = false
+lto = 'fat'
+panic = 'abort'
 codegen-units = 1
+rpath = false

Cargo.toml

@@ -11,6 +11,7 @@ homepage = "https://github.com/pythops/oryx"
 aya-ebpf = "0.1.1"
 oryx-common = { path = "../oryx-common" }
 network-types = { git = "https://github.com/vadorovsky/network-types", rev = "b78424c" }
+branches = { version = "0.3", default-features = false }
 
 [[bin]]
 name = "oryx"

oryx-ebpf/Cargo.lock

@@ -8,6 +8,7 @@ use aya_ebpf::{
     maps::{Array, HashMap, RingBuf},
     programs::TcContext,
 };
+use branches::unlikely;
 use core::mem;
 use network_types::{
     arp::ArpHdr,
@@ -194,7 +195,7 @@ fn process(ctx: TcContext) -> Result<i32, ()> {
                         u16::from_be_bytes(unsafe { (*tcp_header).dest })
                     };
 
-                    if block_ipv4(addr, port) {
+                    if unlikely(block_ipv4(addr, port)) {
                         return Ok(TC_ACT_SHOT); //block packet
                     }
 
@@ -227,7 +228,7 @@ fn process(ctx: TcContext) -> Result<i32, ()> {
                         u16::from_be_bytes(unsafe { (*udp_header).dst })
                     };
 
-                    if block_ipv4(addr, port) {
+                    if unlikely(block_ipv4(addr, port)) {
                         return Ok(TC_ACT_SHOT); //block packet
                     }
 
@@ -260,7 +261,7 @@ fn process(ctx: TcContext) -> Result<i32, ()> {
                         u16::from_be_bytes(unsafe { (*sctp_header).dst })
                     };
 
-                    if block_ipv4(addr, port) {
+                    if unlikely(block_ipv4(addr, port)) {
                         return Ok(TC_ACT_SHOT); //block packet
                     }
 
@@ -389,7 +390,7 @@ fn process(ctx: TcContext) -> Result<i32, ()> {
                     }
                 }
                 IpProto::Sctp => {
-                    let sctp_header: *const SctpHdr = ptr_at(&ctx, EthHdr::LEN + Ipv4Hdr::LEN)?;
+                    let sctp_header: *const SctpHdr = ptr_at(&ctx, EthHdr::LEN + Ipv6Hdr::LEN)?;
 
                     let port = if is_ingress() {
                         u16::from_be_bytes(unsafe { (*sctp_header).src })
@@ -401,7 +402,7 @@ fn process(ctx: TcContext) -> Result<i32, ()> {
                         return Ok(TC_ACT_SHOT); //block packet
                     }
 
-                    if filter_packet(Protocol::Network(NetworkProtocol::Ipv4))
+                    if filter_packet(Protocol::Network(NetworkProtocol::Ipv6))
                         || filter_packet(Protocol::Transport(TransportProtocol::SCTP))
                         || filter_direction()
                     {

oryx-ebpf/Cargo.toml

@@ -19,7 +19,7 @@ oryx-common = { path = "../oryx-common" }
 mio = { version = "1", features = ["os-poll", "os-ext"] }
 itertools = "0.14"
 dirs = "6"
-kanal = "0.1"
+kanal = { version = "0.1", default-features = false }
 mimalloc = "0.1"
 clap = { version = "4", features = ["derive", "cargo"] }
 network-types = { workspace = true }
@@ -32,6 +32,10 @@ regex = "1"
 chrono = "0.4"
 strum = { version = "0.27", features = ["derive"] }
 anyhow = "1"
+branches = "0.3"
+cacheguard = "0.1"
+rustc-hash = "2.1.1"
+arrayvec = "0.7"
 
 [[bin]]
 name = "oryx"

oryx-ebpf/src/main.rs

@@ -1,29 +1,16 @@
-use chrono::Utc;
 use clap::ArgMatches;
 use itertools::Itertools;
-use oryx_common::{
-    RawData, RawFrame,
-    protocols::{LinkProtocol, NetworkProtocol, TransportProtocol},
-};
+use oryx_common::protocols::{LinkProtocol, NetworkProtocol, TransportProtocol};
 use ratatui::{
     Frame,
     layout::{Constraint, Direction, Layout},
 };
-use std::{
-    error,
-    str::FromStr,
-    sync::{Arc, RwLock},
-    thread,
-    time::Duration,
-};
+use std::{error, str::FromStr, thread, time::Duration};
 
 use crate::{
-    filter::Filter,
-    help::Help,
-    packet::{EthFrame, direction::TrafficDirection},
+    filter::Filter, filter::IoChannels, help::Help, notification::Notification,
+    packet::direction::TrafficDirection, packet_store::PacketStore, section::Section,
 };
-use crate::{filter::IoChannels, notification::Notification};
-use crate::{packet::AppPacket, section::Section};
 
 pub type AppResult<T> = std::result::Result<T, Box<dyn error::Error>>;
 
@@ -40,7 +27,6 @@ pub enum ActivePopup {
 
 #[derive(Debug)]
 pub struct DataEventHandler {
-    pub sender: kanal::Sender<[u8; RawFrame::LEN]>,
     pub handler: thread::JoinHandle<()>,
 }
 
@@ -50,46 +36,20 @@ pub struct App {
     pub help: Help,
     pub filter: Filter,
     pub start_sniffing: bool,
-    pub app_packets: Arc<RwLock<Vec<AppPacket>>>,
+    pub app_packets: PacketStore,
     pub notifications: Vec<Notification>,
     pub section: Section,
-    pub data_channel_sender: kanal::Sender<([u8; RawData::LEN], TrafficDirection)>,
     pub is_editing: bool,
     pub active_popup: Option<ActivePopup>,
     pub start_from_cli: bool,
 }
 
 impl App {
     pub fn new(cli_args: &ArgMatches) -> Self {
-        let app_packets = Arc::new(RwLock::new(Vec::with_capacity(
-            AppPacket::LEN * 1024 * 1024,
-        )));
-
-        let (sender, receiver) = kanal::unbounded();
+        let app_packets = PacketStore::new();
 
         let firewall_channels = IoChannels::new();
 
-        thread::spawn({
-            let app_packets = app_packets.clone();
-            move || loop {
-                if let Ok((raw_data, direction)) = receiver.recv() {
-                    let data = RawData::from(raw_data);
-                    let frame = EthFrame::from(data.frame);
-                    let pid = data.pid;
-
-                    let mut app_packets = app_packets.write().unwrap();
-
-                    let app_packet = AppPacket {
-                        frame,
-                        direction,
-                        pid,
-                        timestamp: Utc::now(),
-                    };
-                    app_packets.push(app_packet);
-                }
-            }
-        });
-
         let (interface_name, transport_protocols, network_protocols, link_protocols, direction) = {
             if let Some(interface) = cli_args.get_one::<String>("interface") {
                 let transport_protocols = {
@@ -193,7 +153,6 @@ impl App {
             app_packets: app_packets.clone(),
             notifications: Vec::new(),
             section: Section::new(app_packets.clone(), firewall_channels.clone()),
-            data_channel_sender: sender,
             is_editing: false,
             active_popup: None,
             start_from_cli: interface_name.is_some(),

oryx-tui/Cargo.toml

@@ -95,12 +95,12 @@ impl Bandwidth {
                 }
 
                 loop {
+                    buffer.clear();
                     thread::sleep(Duration::from_secs(1));
                     if fd.seek(std::io::SeekFrom::Start(0)).is_err() {
                         drop(fd);
                         fd = File::open("/proc/net/dev").unwrap();
                     }
-                    let mut buffer = String::new();
                     fd.read_to_string(&mut buffer).unwrap();
 
                     let mut lines = buffer.lines();