PCAP refinement

7 files changed, 174 insertions, 44 deletions

diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index b09393f9..aa2d43f3 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -255,6 +255,13 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration,
 }
 
 
+bool TrexStatelessPort::is_active() const {
+    return   (  (m_port_state == PORT_STATE_TX) 
+             || (m_port_state == PORT_STATE_PAUSE)
+             || (m_port_state == PORT_STATE_PCAP_TX)
+             );
+}
+
 /**
  * stop traffic on port
  * 
@@ -264,9 +271,7 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration,
  */
 void
 TrexStatelessPort::stop_traffic(void) {
-
-    if (!( (m_port_state == PORT_STATE_TX) 
-        || (m_port_state == PORT_STATE_PAUSE) )) {
+    if (!is_active()) {
         return;
     }
 
@@ -438,10 +443,13 @@ TrexStatelessPort::push_remote(const std::string &pcap_filename, double ipg_usec
     }
 
     /* send a message to core */
-    change_state(PORT_STATE_TX);
+    change_state(PORT_STATE_PCAP_TX);
     TrexStatelessCpToDpMsgBase *push_msg = new TrexStatelessDpPushPCAP(m_port_id,
                                                                        m_pending_async_stop_event,
-                                                                       pcap_filename);
+                                                                       pcap_filename,
+                                                                       ipg_usec,
+                                                                       speedup,
+                                                                       count);
     send_message_to_dp(tx_core, push_msg);
 
     /* update subscribers */    
@@ -468,6 +476,9 @@ TrexStatelessPort::get_state_as_string() const {
 
     case PORT_STATE_PAUSE:
         return "PAUSE";
+
+    case PORT_STATE_PCAP_TX:
+        return "PCAP_TX";
     }
 
     return "UNKNOWN";
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 502c066d..ccbfad0d 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -129,11 +129,12 @@ public:
      * port state
      */
     enum port_state_e {
-        PORT_STATE_DOWN = 0x1,
-        PORT_STATE_IDLE = 0x2,
-        PORT_STATE_STREAMS = 0x4,
-        PORT_STATE_TX = 0x8,
-        PORT_STATE_PAUSE = 0x10,
+        PORT_STATE_DOWN     = 0x1,
+        PORT_STATE_IDLE     = 0x2,
+        PORT_STATE_STREAMS  = 0x4,
+        PORT_STATE_TX       = 0x8,
+        PORT_STATE_PAUSE    = 0x10,
+        PORT_STATE_PCAP_TX  = 0x20,
     };
 
     /**
@@ -226,6 +227,12 @@ public:
     }
 
     /**
+     * return true if the port is active
+     * (paused is considered active)
+     */
+    bool is_active() const;
+
+    /**
      * port state as string
      *
      */
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 31c907fa..6450d0f9 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -262,7 +262,11 @@ bool TrexStatelessDpPerPort::pause_traffic(uint8_t port_id){
     return (true);
 }
 
-bool TrexStatelessDpPerPort::push_pcap(uint8_t port_id, const std::string &pcap_filename){
+bool TrexStatelessDpPerPort::push_pcap(uint8_t port_id,
+                                       const std::string &pcap_filename,
+                                       double ipg_usec,
+                                       double speedup,
+                                       uint32_t count) {
 
     /* push pcap can only happen on an idle port from the core prespective */
     assert(m_state == TrexStatelessDpPerPort::ppSTATE_IDLE);
@@ -277,7 +281,13 @@ bool TrexStatelessDpPerPort::push_pcap(uint8_t port_id, const std::string &pcap_
     uint8_t mac_addr[12];
     m_core->m_node_gen.m_v_if->update_mac_addr_from_global_cfg(dir, mac_addr);
 
-    bool rc = pcap_node->create(port_id, pcap_filename, dir, mac_addr);
+    bool rc = pcap_node->create(port_id,
+                                dir,
+                                mac_addr,
+                                pcap_filename,
+                                ipg_usec,
+                                speedup,
+                                count);
     if (!rc) {
         m_core->free_node((CGenNode *)pcap_node);
         return (false);
@@ -287,8 +297,11 @@ bool TrexStatelessDpPerPort::push_pcap(uint8_t port_id, const std::string &pcap_
     pcap_node->m_time = m_core->m_cur_time_sec;
     m_core->m_node_gen.add_node((CGenNode *)pcap_node);
 
-    m_state = TrexStatelessDpPerPort::ppSTATE_TRANSMITTING;
+    /* hold a pointer to the node */
+    assert(m_active_pcap_node == NULL);
+    m_active_pcap_node = pcap_node;
 
+    m_state = TrexStatelessDpPerPort::ppSTATE_TRANSMITTING;
     return (true);
 }
 
@@ -324,6 +337,19 @@ bool TrexStatelessDpPerPort::stop_traffic(uint8_t  port_id,
         }
     }
 
+    /* check for active PCAP node */
+    if (m_active_pcap_node) {
+        /* when got async stop from outside or duration */
+        if (m_active_pcap_node->is_active()) {
+            m_active_pcap_node->mark_for_free();
+        } else {
+            /* graceful stop - node was put out by the scheduler */
+            m_core->free_node( (CGenNode *)m_active_pcap_node);
+        }
+
+        m_active_pcap_node = NULL;
+    }
+
     /* active stream should be zero */
     assert(m_active_streams==0);
     m_active_nodes.clear();
@@ -337,6 +363,7 @@ void TrexStatelessDpPerPort::create(CFlowGenListPerThread   *  core){
     m_state=TrexStatelessDpPerPort::ppSTATE_IDLE;
     m_active_streams=0;
     m_active_nodes.clear();
+    m_active_pcap_node = NULL;
 }
 
 
@@ -866,14 +893,19 @@ TrexStatelessDpCore::pause_traffic(uint8_t port_id){
 
 
 void 
-TrexStatelessDpCore::push_pcap(uint8_t port_id, int event_id, const std::string &pcap_filename) {
+TrexStatelessDpCore::push_pcap(uint8_t port_id,
+                               int event_id,
+                               const std::string &pcap_filename,
+                               double ipg_usec,
+                               double speedup,
+                               uint32_t count) {
 
     TrexStatelessDpPerPort * lp_port = get_port_db(port_id);
 
     lp_port->set_event_id(event_id);
 
     /* delegate the command to the port */
-    bool rc = lp_port->push_pcap(port_id, pcap_filename);
+    bool rc = lp_port->push_pcap(port_id, pcap_filename, ipg_usec, speedup, count);
     if (!rc) {
         /* report back that we stopped */
         CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingDpToCp(m_core->m_thread_id);
@@ -913,7 +945,6 @@ TrexStatelessDpCore::stop_traffic(uint8_t  port_id,
        the scheduler invokes it, it will be free */
 
     TrexStatelessDpPerPort * lp_port = get_port_db(port_id);
-
     if ( lp_port->stop_traffic(port_id,stop_on_id,event_id) == false){
         /* nothing to do ! already stopped */
         //printf(" skip .. %f\n",m_core->m_cur_time_sec);
@@ -961,13 +992,30 @@ TrexStatelessDpCore::barrier(uint8_t port_id, int event_id) {
 /**
  * PCAP node
  */
-bool CGenNodePCAP::create(uint8_t port_id, const std::string &pcap_filename, pkt_dir_t dir, const uint8_t *mac_addr) {
+bool CGenNodePCAP::create(uint8_t port_id,
+                          pkt_dir_t dir,
+                          const uint8_t *mac_addr,
+                          const std::string &pcap_filename,
+                          double ipg_usec,
+                          double speedup,
+                          uint32_t count) {
     std::stringstream ss;
 
     m_type       = CGenNode::PCAP_PKT;
     m_flags      = 0;
     m_src_port   = 0;
     m_port_id    = port_id;
+    m_count      = count;
+    
+    if (ipg_usec != -1) {
+        /* fixed IPG */
+        m_ipg_sec = usec_to_sec(ipg_usec / speedup);
+        m_speedup = 0;
+    } else {
+        /* packet IPG */
+        m_ipg_sec = -1;
+        m_speedup  = speedup;
+    }
 
     /* copy MAC addr info */
     memcpy(m_mac_addr, mac_addr, 12);
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index 01033a7c..115f8873 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -33,7 +33,7 @@ class CFlowGenListPerThread;
 class CGenNodeStateless;
 class TrexStreamsCompiledObj;
 class TrexStream;
-
+class CGenNodePCAP;
 
 class CDpOneStream  {
 public:
@@ -70,7 +70,11 @@ public:
 
     bool update_traffic(uint8_t port_id, double factor);
 
-    bool push_pcap(uint8_t port_id, const std::string &pcap_filename);
+    bool push_pcap(uint8_t port_id,
+                   const std::string &pcap_filename,
+                   double ipg_usec,
+                   double speedup,
+                   uint32_t count);
 
     bool stop_traffic(uint8_t port_id,
                       bool stop_on_id, 
@@ -97,6 +101,7 @@ public:
     uint32_t                  m_active_streams; /* how many active streams on this port  */
                                                 
     std::vector<CDpOneStream> m_active_nodes;   /* holds the current active nodes */
+    CGenNodePCAP              *m_active_pcap_node;
     CFlowGenListPerThread   *  m_core ;
     int                        m_event_id;
 };
@@ -165,7 +170,12 @@ public:
      * push a PCAP file on port
      * 
      */
-    void push_pcap(uint8_t port_id, int event_id, const std::string &pcap_filename);
+    void push_pcap(uint8_t port_id,
+                   int event_id,
+                   const std::string &pcap_filename,
+                   double ipg_usec,
+                   double speedup,
+                   uint32_t count);
 
 
     /**
diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h
index 8ccb5286..a970c1f7 100644
--- a/src/stateless/dp/trex_stream_node.h
+++ b/src/stateless/dp/trex_stream_node.h
@@ -398,7 +398,14 @@ public:
     /**
      * creates a node from a PCAP file 
      */
-    bool create(uint8_t port_id, const std::string &pcap_filename, pkt_dir_t dir, const uint8_t *mac_addr);
+    bool create(uint8_t port_id,
+                pkt_dir_t dir,
+                const uint8_t *mac_addr,
+                const std::string &pcap_filename,
+                double ipg_usec,
+                double speedup,
+                uint32_t count);
+
     void destroy();
  
     /**
@@ -407,36 +414,44 @@ public:
      * @author imarom (03-May-16)
      */
     void next() {
-        assert(m_state == PCAP_ACTIVE);
+        assert(is_active());
 
         /* save the previous packet time */
         m_last_pkt_time = m_raw_packet->get_time();
 
         /* advance */
         if ( m_reader->ReadPacket(m_raw_packet) == false ){
-            m_state = PCAP_EOF;
-            return;
+            m_count--;
+
+            /* if its the end - go home... */
+            if (m_count == 0) {
+                m_state = PCAP_INACTIVE;
+                return;
+            } 
+
+            /* rewind and load the first packet */
+            m_reader->Rewind();
+            if (!m_reader->ReadPacket(m_raw_packet)) {
+                m_state = PCAP_INACTIVE;
+                return;
+            }
         }
 
     }
 
     /**
-     * return true if the PCAP has next packet
-     * 
-     */
-    bool has_next() {
-        assert(m_state != PCAP_INVALID);
-        return (m_state == PCAP_ACTIVE);
-    }
-
-    /**
      * return the time for the next scheduling for a packet
      * 
      */
     inline double get_ipg() {
         assert(m_state != PCAP_INVALID);
-        return m_raw_packet->get_time() - m_last_pkt_time;
-        //return 0.00001;
+
+        /* fixed IPG */
+        if (m_ipg_sec != -1) {
+            return m_ipg_sec;
+        } else {
+            return ((m_raw_packet->get_time() - m_last_pkt_time) / m_speedup);
+        }
     }
 
     /**
@@ -465,6 +480,17 @@ public:
     inline void handle(CFlowGenListPerThread *thread) {
         assert(m_state != PCAP_INVALID);
         thread->m_node_gen.m_v_if->send_node( (CGenNode *)this);
+
+        // read the next packet
+        next();
+
+        if (is_active()) {
+            m_time += get_ipg();
+            thread->m_node_gen.m_p_queue.push((CGenNode *)this);  
+                            
+        } else {
+            thread->stop_stateless_traffic(get_port_id());
+        }
     }
 
     void set_mbuf_dir(pkt_dir_t dir) {
@@ -483,12 +509,25 @@ public:
         return m_port_id;
     }
 
+    void mark_for_free() {
+        m_state = PCAP_MARKED_FOR_FREE;
+    }
+
+    bool is_active() {
+        return (m_state == PCAP_ACTIVE);
+    }
+
+    bool is_marked_for_free() {
+        return (m_state == PCAP_MARKED_FOR_FREE);
+    }
+
 private:
 
     enum {
         PCAP_INVALID = 0,
         PCAP_ACTIVE,
-        PCAP_EOF
+        PCAP_INACTIVE,
+        PCAP_MARKED_FOR_FREE
     };
 
     /* cache line 0 */
@@ -496,11 +535,10 @@ private:
     uint8_t             m_mac_addr[12];
     uint8_t             m_state;
 
-    //double              m_base_time;
-    //double              m_current_pkt_time;
     double              m_last_pkt_time;
-
-    void *              m_cache_mbuf;
+    double              m_speedup;
+    double              m_ipg_sec;
+    uint32_t            m_count;
 
     double              m_next_time_offset; /* in sec */
 
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 59e0a0a8..c0151c76 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -187,13 +187,18 @@ TrexStatelessDpUpdate::clone() {
  ************************/
 bool
 TrexStatelessDpPushPCAP::handle(TrexStatelessDpCore *dp_core) {
-    dp_core->push_pcap(m_port_id, m_event_id, m_pcap_filename);
+    dp_core->push_pcap(m_port_id, m_event_id, m_pcap_filename, m_ipg_usec, m_speedup, m_count);
     return true;
 }
 
 TrexStatelessCpToDpMsgBase *
 TrexStatelessDpPushPCAP::clone() {
-    TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpPushPCAP(m_port_id, m_event_id, m_pcap_filename);
+    TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpPushPCAP(m_port_id,
+                                                                      m_event_id,
+                                                                      m_pcap_filename,
+                                                                      m_ipg_usec,
+                                                                      m_speedup,
+                                                                      m_count);
 
     return new_msg;
 }
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index 8fb2a456..c3de82ee 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -253,9 +253,17 @@ private:
 class TrexStatelessDpPushPCAP : public TrexStatelessCpToDpMsgBase {
 public:
 
-    TrexStatelessDpPushPCAP(uint8_t port_id, int event_id, const std::string &pcap_filename) : m_pcap_filename(pcap_filename)  {
+    TrexStatelessDpPushPCAP(uint8_t port_id,
+                            int event_id,
+                            const std::string &pcap_filename,
+                            double ipg_usec,
+                            double speedup,
+                            uint32_t count) : m_pcap_filename(pcap_filename)  {
         m_port_id  = port_id;
         m_event_id = event_id;
+        m_ipg_usec = ipg_usec;
+        m_speedup  = speedup;
+        m_count    = count;
     }
 
     virtual bool handle(TrexStatelessDpCore *dp_core);
@@ -264,8 +272,11 @@ public:
 
 private:
     std::string  m_pcap_filename;
-    uint8_t      m_port_id;
     int          m_event_id;
+    double       m_ipg_usec;
+    double       m_speedup;
+    uint32_t     m_count;
+    uint8_t      m_port_id;
 };