Add PacketList Datatypes

core/src/conntrack/conn/conn_info.rs

@@ -12,7 +12,7 @@ use crate::protocols::stream::{
     ConnData, ParseResult, ParserRegistry, ParsingState, ProbeRegistryResult,
 };
 use crate::subscription::{Subscription, Trackable};
-use crate::FiveTuple;
+use crate::{FiveTuple, Mbuf};
 
 #[derive(Debug)]
 pub(crate) struct ConnInfo<T>
@@ -77,7 +77,9 @@ where
         }
         if self.actions.buffer_frame() {
             // Track frame for (potential) future delivery
-            self.sdata.track_packet(pdu.mbuf_own());
+            // Used when a filter has partially matched for a
+            // subscription that requests packets
+            self.sdata.track_packet(Mbuf::new_ref(pdu.mbuf_ref()));
         }
     }
 

core/src/conntrack/mod.rs

@@ -138,9 +138,10 @@ where
                     };
                     if let Ok(mut conn) = conn {
                         conn.info.filter_first_packet(&pdu, subscription);
-                        if !conn.info.actions.drop() {
-                            conn.info.consume_pdu(pdu, subscription, &self.registry);
+                        if conn.info.actions.update_pdu() {
+                            conn.info.sdata.update(&pdu, false);
                         }
+                        conn.info.consume_pdu(pdu, subscription, &self.registry);
                         if !conn.remove_from_table() {
                             self.timerwheel.insert(
                                 &conn_id,

core/src/dpdk/inlined.c

@@ -29,6 +29,10 @@ uint16_t rte_mbuf_refcnt_update_(struct rte_mbuf* m, int16_t value) {
     return rte_mbuf_refcnt_update(m, value);
 }
 
+void rte_mbuf_refcnt_set_(struct rte_mbuf* m, int16_t value) {
+    return rte_mbuf_refcnt_set(m, value);
+}
+
 char* rte_pktmbuf_adj_(struct rte_mbuf* m, uint16_t len) {
     return rte_pktmbuf_adj(m, len);
 }

core/src/dpdk/mod.rs

@@ -27,6 +27,7 @@ extern "C" {
     ) -> u16;
     fn rte_mbuf_refcnt_read_(m: *const rte_mbuf) -> u16;
     fn rte_mbuf_refcnt_update_(m: *mut rte_mbuf, value: i16) -> u16;
+    fn rte_mbuf_refcnt_set_(m: *mut rte_mbuf, value: i16);
     fn rte_pktmbuf_adj_(packet: *mut rte_mbuf, len: u16) -> *mut c_char;
     fn rte_pktmbuf_trim_(packet: *mut rte_mbuf, len: u16) -> c_int;
     fn rte_lcore_id_() -> u16;
@@ -105,6 +106,11 @@ pub unsafe fn rte_mbuf_refcnt_update(m: *mut rte_mbuf, value: i16) -> u16 {
     rte_mbuf_refcnt_update_(m, value)
 }
 
+#[inline]
+pub unsafe fn rte_mbuf_refcnt_set(m: *mut rte_mbuf, value: i16) {
+    rte_mbuf_refcnt_set_(m, value)
+}
+
 #[inline]
 pub unsafe fn rte_pktmbuf_adj(packet: *mut rte_mbuf, len: u16) -> *mut c_char {
     rte_pktmbuf_adj_(packet, len)

core/src/filter/datatypes.rs

@@ -57,8 +57,6 @@ pub struct DataType {
     pub needs_update: bool,
     /// True if the datatype requires invoking `update` method after reassembly
     pub needs_update_reassembled: bool,
-    /// True if the datatype requires the framework to buffer matched packets
-    pub track_packets: bool,
     /// A vector of the application-layer parsers required by this datatype
     /// Retina loads the union of parsers required by all datatypes and filters
     pub stream_protos: Vec<&'static str>,
@@ -76,7 +74,6 @@ impl DataType {
             track_sessions: false,
             needs_update: true,
             needs_update_reassembled: false,
-            track_packets: false,
             stream_protos: vec![],
             as_str,
         }
@@ -91,7 +88,6 @@ impl DataType {
             track_sessions: false,
             needs_update: false,
             needs_update_reassembled: false,
-            track_packets: false,
             stream_protos,
             as_str,
         }
@@ -106,7 +102,6 @@ impl DataType {
             track_sessions: false,
             needs_update: false,
             needs_update_reassembled: false,
-            track_packets: false,
             stream_protos: vec![],
             as_str,
         }
@@ -121,7 +116,18 @@ impl DataType {
             track_sessions: false,
             needs_update: false,
             needs_update_reassembled: false,
-            track_packets: false,
+            stream_protos: vec![],
+            as_str,
+        }
+    }
+
+    pub fn new_default_pktlist(as_str: &'static str, reassembly: bool) -> Self {
+        DataType {
+            level: Level::Connection,
+            needs_parse: false,
+            track_sessions: false,
+            needs_update: !reassembly,
+            needs_update_reassembled: reassembly,
             stream_protos: vec![],
             as_str,
         }
@@ -208,15 +214,6 @@ impl DataType {
         }
     }
 
-    // Helper
-    fn track_packets(&self, actions: &mut MatchingActions) {
-        if self.track_packets {
-            actions.if_matched.data |= ActionData::PacketTrack;
-            actions.if_matched.terminal_actions |= ActionData::PacketTrack;
-            actions.if_matching.data |= ActionData::PacketTrack;
-        }
-    }
-
     // Helper for proto_filter and session_filter
     fn track_sessions(&self, actions: &mut MatchingActions, sub_level: &Level) {
         // SessionTrack should only be terminal if matched at packet layer
@@ -251,7 +248,6 @@ impl DataType {
 
         // Connection- and session-level subscriptions depend on the actions required
         self.needs_update(&mut actions);
-        self.track_packets(&mut actions);
         self.conn_deliver(sub_level, &mut actions);
 
         if self.needs_parse {
@@ -292,7 +288,6 @@ impl DataType {
 
         // Connection- and session-level subscriptions depend on the actions required
         self.needs_update(&mut actions);
-        self.track_packets(&mut actions);
         self.track_sessions(&mut actions, sub_level);
         self.conn_deliver(sub_level, &mut actions);
 
@@ -316,7 +311,6 @@ impl DataType {
         }
 
         self.needs_update(&mut actions);
-        self.track_packets(&mut actions);
         self.track_sessions(&mut actions, sub_level);
         self.conn_deliver(sub_level, &mut actions);
 

core/src/filter/ptree.rs

@@ -397,7 +397,7 @@ impl PTree {
                 // is equivalent performance-wise to implementing similar functionality in the
                 // framework.
                 panic!("Cannot access per-packet fields (e.g., TCP flags, length) after packet filter.\n\
-                       Subscribe to `ZcFrame` or PacketList instead.");
+                       Subscribe to `ZcFrame` or list of mbufs instead.");
             }
 
             // Predicate is already present

core/src/memory/mbuf.rs

@@ -45,6 +45,13 @@ impl Mbuf {
         })
     }
 
+    pub fn new_ref(mbuf: &Mbuf) -> Mbuf {
+        unsafe {
+            dpdk::rte_mbuf_refcnt_update(mbuf.raw.as_ptr(), 1);
+        }
+        Mbuf::new_unchecked(mbuf.raw.as_ptr())
+    }
+
     /// Creates a new Mbuf from a byte slice.
     pub(crate) fn from_bytes(data: &[u8], mp: *mut dpdk::rte_mempool) -> Result<Mbuf> {
         let mut mbuf = unsafe { Mbuf::new(dpdk::rte_pktmbuf_alloc(mp))? };
@@ -181,6 +188,12 @@ impl<'a> Packet<'a> for Mbuf {
 impl Drop for Mbuf {
     fn drop(&mut self) {
         // log::debug!("Dropping a Mbuf, freeing mbuf@{:p}", self.raw().buf_addr);
+
+        // Reference counting allows Mbufs to be shared across data structures
+        // (e.g., store in pre-reassembly and post-reassembly order) and cores.
+        // Using DPDK built-ins to manage reference counting is more efficient
+        // than Rust standard library (Rc or Arc). Note `rte_pktmbuf_free` updates
+        // refcount internally and only releases the mbuf if refcount becomes 0.
         unsafe { dpdk::rte_pktmbuf_free(self.raw()) };
     }
 }

datatypes/src/lib.rs

@@ -30,6 +30,8 @@ pub mod packet;
 pub use packet::{Payload, ZcFrame};
 pub mod static_type;
 pub use static_type::*;
+pub mod packet_list;
+pub use packet_list::*;
 pub use typedefs::*;
 
 use retina_core::conntrack::pdu::L4Pdu;
@@ -92,3 +94,17 @@ pub trait FromSubscription {
     /// the constant value (e.g., matched filter string).
     fn from_subscription(spec: &SubscriptionSpec) -> proc_macro2::TokenStream;
 }
+
+/// Trait for a datatype that is built from a list of raw packets.
+pub trait PacketList {
+    /// Initialize internal data; called once per connection.
+    /// Note `first_pkt` will also be delivered to `update`.
+    fn new(first_pkt: &L4Pdu) -> Self;
+    /// New packet in connection received (or reassembled, if reassembled=true)
+    /// Note this may be invoked both pre- and post-reassembly; types
+    /// should check `reassembled` to avoid double-counting.
+    fn update(&mut self, pdu: &L4Pdu, reassembled: bool);
+    /// Clear internal data; called if connection no longer matches filter
+    /// that requires the Tracked type.
+    fn clear(&mut self);
+}