added cycles to port statistics

Author: rstade

framework/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "e2d2"
-version = "0.3.6"
+version = "0.4.0"
 authors = ["Aurojit Panda <[email protected]>", "rainer <[email protected]>"]
 build = "build.rs"
 

framework/src/interface/packet.rs

@@ -253,7 +253,6 @@ impl<T: EndOffset, M: Sized + Send> Packet<T, M> {
 
     /// clone has same mbuf as the original
     pub fn clone(&mut self) -> Packet<T, M> {
-        // unsafe { packet_from_mbuf(self.mbuf, self.offset) };
         reference_mbuf(self.mbuf);
 
         Packet::<T, M> {
@@ -266,6 +265,19 @@ impl<T: EndOffset, M: Sized + Send> Packet<T, M> {
         }
     }
 
+    /// same as clone, but without increment of mbuf ref count
+    pub fn clone_without_ref_counting(&mut self) -> Packet<T, M> {
+        Packet::<T, M> {
+            mbuf: self.mbuf,
+            _phantom_m: PhantomData,
+            offset: self.offset,
+            pre_pre_header: self.pre_pre_header,
+            pre_header: self.pre_header,
+            header: self.header,
+        }
+    }
+
+
     /// copy gets us a new mbuf
     pub unsafe fn copy(&self) -> Packet<T, M> {
         // unsafe { packet_from_mbuf(self.mbuf, self.offset) };

framework/src/interface/port/mod.rs

@@ -6,7 +6,7 @@ use allocators::*;
 use common::*;
 use interface::{PacketRx, PacketTx};
 use native::zcsi::MBuf;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicUsize, AtomicU64, Ordering};
 
 mod phy_port;
 mod virt_port;
@@ -17,6 +17,7 @@ pub struct PortStats {
     pub stats: AtomicUsize,
     pub q_len: AtomicUsize,
     pub max_q_len: AtomicUsize,
+    pub cycles: AtomicU64,
 }
 
 impl PortStats {
@@ -25,11 +26,13 @@ impl PortStats {
             stats: AtomicUsize::new(0),
             q_len: AtomicUsize::new(0),
             max_q_len: AtomicUsize::new(0),
+            cycles: AtomicU64::new(0),
         })
     }
 
     pub fn get_q_len(&self) -> usize { self.q_len.load(Ordering::Relaxed) }
     pub fn get_max_q_len(&self) -> usize { self.max_q_len.load(Ordering::Relaxed) }
+    pub fn cycles(&self) -> u64 { self.cycles.load(Ordering::Relaxed) }
 
     pub fn set_q_len(&self, len: usize) -> usize {
         let q_max= self.get_max_q_len();

framework/src/interface/port/phy_port.rs

@@ -15,7 +15,7 @@ use std::ptr;
 use std::ptr::Unique;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
-use utils::FiveTupleV4;
+use utils::{FiveTupleV4, rdtsc_unsafe};
 
 /// A DPDK based PMD port. Send and receive should not be called directly on this structure but on the port queue
 /// structure instead.
@@ -114,12 +114,13 @@ impl fmt::Display for PortQueue {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(
             f,
-            "port: {} ({}) rxq: {} txq: {}, max_rxq_len: {}",
+            "port: {} ({}) rxq: {} txq: {}, max_rxq_len: {}, recv_cycles: {}",
             self.port.mac_address(),
             self.port_id,
             self.rxq,
             self.txq,
             self.stats_rx.get_max_q_len(),
+            self.stats_rx.cycles(),
         )
     }
 }
@@ -147,6 +148,7 @@ impl PortQueue {
 
     #[inline]
     fn recv_queue(&self, pkts: &mut [*mut MBuf], to_recv: u16) -> errors::Result<(u32, i32)> {
+        let start= rdtsc_unsafe();
         unsafe {
             let (recv, q_count) = if self.port.is_kni() {
                 //debug!("calling rte_kni_rx_burst for {}.{}", self.port, self.rxq);
@@ -160,6 +162,10 @@ impl PortQueue {
             let update = self.stats_rx.stats.load(Ordering::Relaxed) + recv as usize;
             self.stats_rx.stats.store(update, Ordering::Relaxed);
             self.stats_rx.set_q_len(q_count as usize);
+            if recv > 0 ||q_count > 0 {
+                let update = self.stats_rx.cycles.load(Ordering::Relaxed) + (rdtsc_unsafe() - start);
+                self.stats_rx.cycles.store(update, Ordering::Relaxed);
+            }
             Ok((recv, q_count))
         }
     }
@@ -298,11 +304,6 @@ impl PmdPort {
         }
     }
 
-    /// Current port ID.
-    //    #[inline]
-    //    pub fn name(&self) -> i32 {
-    //        self.port
-    //    }
     /// Get stats for an RX/TX queue pair.
     pub fn stats(&self, queue: i32) -> (usize, usize, usize) {
         let idx = queue as usize;
@@ -313,6 +314,18 @@ impl PmdPort {
         )
     }
 
+    /// Get stats for an RX/TX queue pair.
+    fn stats_4(&self, queue: i32) -> (usize, usize, usize, u64) {
+        let idx = queue as usize;
+        (
+            self.stats_rx[idx].stats.load(Ordering::Relaxed),
+            self.stats_tx[idx].stats.load(Ordering::Relaxed),
+            self.stats_rx[idx].get_max_q_len(),
+            self.stats_rx[idx].cycles(),
+        )
+    }
+
+
     pub fn map_rx_flow_2_queue(&self, rxq: u16, flow: FiveTupleV4, flow_mask: FiveTupleV4) -> Option<&RteFlow> {
         unsafe {
             let mut error = RteFlowError {
@@ -353,17 +366,17 @@ impl PmdPort {
 
      pub fn print_soft_statistics(&self) {
         println!(
-            "{0:>3} | {1: >20} | {2: >20} | {3: >20} | {4: >20} | {5: >20} | {6: >20} | {7: >20}",
-            "q", "ipackets", "opackets", "ibytes", "obytes", "ierrors", "oerrors", "queue_len"
+            "{0:>3} | {1: >20} | {2: >20} | {3: >20} | {4: >20} | {5: >20} | {6: >20} | {7: >20} | {8: >20}",
+            "q", "ipackets", "opackets", "ibytes", "obytes", "ierrors", "oerrors", "queue_len", "rx_cycles"
         );
         let (mut sin_p, mut sout_p) = (0usize, 0usize);
         for q in 0..self.rxqs() {
-            let (in_p, out_p, rx_max_q_len) = self.stats(q);
+            let (in_p, out_p, rx_max_q_len, cycles) = self.stats_4(q);
             sin_p += in_p;
             sout_p += out_p;
             println!(
-                "{0:>3} | {1: >20} | {2: >20} | {3: >20} | {4: >20} | {5: >20} | {6: >20} | {7: >20}",
-                q, in_p, out_p, 0, 0, 0, 0, rx_max_q_len
+                "{0:>3} | {1: >20} | {2: >20} | {3: >20} | {4: >20} | {5: >20} | {6: >20} | {7: >20} | {8: >20}",
+                q, in_p, out_p, 0, 0, 0, 0, rx_max_q_len, cycles
             );
         }
         println!(

framework/src/operators/merge_batch_auto.rs

@@ -77,15 +77,6 @@ impl<T: Batch> MergeBatchAuto<T> {
 impl<T: Batch> Batch for MergeBatchAuto<T> {
     #[inline]
     fn queued(&self) -> usize {
-        /* let mut result = 0;
-        for b in &self.state {
-            if *b>0 {
-                result = *b;
-                break;
-            }
-        }
-        result
-        */
         self.queue_size
     }
 }