PCAP refinement

12 files changed, 300 insertions, 141 deletions

diff --git a/src/bp_sim.cpp b/src/bp_sim.cpp
index 2491d122..18b1053e 100755
--- a/src/bp_sim.cpp
+++ b/src/bp_sim.cpp
@@ -3713,17 +3713,14 @@ int CNodeGenerator::flush_file(dsec_t max_time,
             m_p_queue.pop();
 
             CGenNodePCAP *node_pcap = (CGenNodePCAP *)node;
-            node_pcap->handle(thread);
-            
-            if (node_pcap->has_next()) {
-                node_pcap->next();
-                node_pcap->m_time += node_pcap->get_ipg();
-                m_p_queue.push(node);              
-            } else {
-                thread->free_node(node);
-                thread->m_stateless_dp_info.stop_traffic(node_pcap->get_port_id(), false, 0);
 
+            /* might have been marked for free */
+            if ( unlikely( node_pcap->is_marked_for_free() ) ) {
+                thread->free_node(node);
+            } else {
+                node_pcap->handle(thread);
             }
+
                         
         } else {
             bool exit_sccheduler = handle_slow_messages(type,node,thread,always);
diff --git a/src/bp_sim.h b/src/bp_sim.h
index 7399137b..132824b3 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -3591,6 +3591,10 @@ public:
     bool  set_stateless_next_node( CGenNodeStateless * cur_node,
                                    CGenNodeStateless * next_node);
 
+    void stop_stateless_traffic(uint8_t port_id) { 
+        m_stateless_dp_info.stop_traffic(port_id, false, 0);
+    }
+
     void Dump(FILE *fd);
     void DumpCsv(FILE *fd);
     void DumpStats(FILE *fd);
diff --git a/src/common/pcap.cpp b/src/common/pcap.cpp
index 9b360a3e..a68fb620 100755
--- a/src/common/pcap.cpp
+++ b/src/common/pcap.cpp
@@ -156,8 +156,7 @@ bool LibPCapReader::ReadPacket(CCapPktRaw *lpPacket)
    }
    if (pkt_header.len > READER_MAX_PACKET_SIZE) {
        /* cannot read this packet */
-       printf("ERROR packet is too big, bigger than %d \n",READER_MAX_PACKET_SIZE);
-       exit(-1);
+       //printf("ERROR packet is too big, bigger than %d \n",READER_MAX_PACKET_SIZE);
        return false;
    }
 
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 3c345aa5..5a383bc6 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -2671,6 +2671,10 @@ public:
     }
     int run_in_rx_core();
     int run_in_master();
+
+    bool handle_fast_path();
+    bool handle_slow_path(bool &was_stopped);
+
     int stop_master();
     /* return the minimum number of dp cores needed to support the active ports
        this is for c==1 or  m_cores_mul==1
@@ -3715,125 +3719,150 @@ CGlobalTRex::publish_async_barrier(uint32_t key) {
     m_zmq_publisher.publish_barrier(key);
 }
 
-int CGlobalTRex::run_in_master() {
-    bool was_stopped=false;
 
-    if ( get_is_stateless() ) {
-        m_trex_stateless->launch_control_plane();
-    }
+bool
+CGlobalTRex::handle_slow_path(bool &was_stopped) {
+    m_stats_cnt+=1;
 
-    /* exception and scope safe */
-    std::unique_lock<std::mutex> cp_lock(m_cp_lock);
 
-    while ( true ) {
-        m_stats_cnt+=1;
-
-
-        if ( CGlobalInfo::m_options.preview.get_no_keyboard() ==false ){
-            if ( m_io_modes.handle_io_modes() ){
-                was_stopped=true;
-                break;
-            }
-        }
-
-        if ( sanity_check() ){
-            printf(" Test was stopped \n");
+    if ( CGlobalInfo::m_options.preview.get_no_keyboard() ==false ) {
+        if ( m_io_modes.handle_io_modes() ) {
             was_stopped=true;
-            break;
+            return false;
         }
-        if (m_io_modes.m_g_mode != CTrexGlobalIoMode::gDISABLE ) {
+    }
+
+    if ( sanity_check() ) {
+        printf(" Test was stopped \n");
+        was_stopped=true;
+        return false;
+    }
+    if (m_io_modes.m_g_mode != CTrexGlobalIoMode::gDISABLE ) {
+        fprintf(stdout,"\033[2J");
+        fprintf(stdout,"\033[2H");
+
+    } else {
+        if ( m_io_modes.m_g_disable_first  ) {
+            m_io_modes.m_g_disable_first=false;
             fprintf(stdout,"\033[2J");
             fprintf(stdout,"\033[2H");
-
-        }else{
-            if ( m_io_modes.m_g_disable_first  ){
-                m_io_modes.m_g_disable_first=false;
-                fprintf(stdout,"\033[2J");
-                fprintf(stdout,"\033[2H");
-                printf("clean !!!\n");
-                fflush(stdout);
-            }
+            printf("clean !!!\n");
+            fflush(stdout);
         }
+    }
 
 
-        if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gHELP ) {
-            m_io_modes.DumpHelp(stdout);
-        }
+    if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gHELP ) {
+        m_io_modes.DumpHelp(stdout);
+    }
 
-        dump_stats(stdout,CGlobalStats::dmpTABLE);
+    dump_stats(stdout,CGlobalStats::dmpTABLE);
 
-        if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gNORMAL ) {
-            fprintf (stdout," current time    : %.1f sec  \n",now_sec());
-            float d= CGlobalInfo::m_options.m_duration - now_sec();
-            if (d<0) {
-                d=0;
+    if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gNORMAL ) {
+        fprintf (stdout," current time    : %.1f sec  \n",now_sec());
+        float d= CGlobalInfo::m_options.m_duration - now_sec();
+        if (d<0) {
+            d=0;
 
-            }
-            fprintf (stdout," test duration   : %.1f sec  \n",d);
         }
+        fprintf (stdout," test duration   : %.1f sec  \n",d);
+    }
 
-        if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gMem) {
+    if (m_io_modes.m_g_mode == CTrexGlobalIoMode::gMem) {
 
-            if ( m_stats_cnt%4==0){
-                fprintf (stdout," %s \n",CGlobalInfo::dump_pool_as_json().c_str());
-            }
+        if ( m_stats_cnt%4==0) {
+            fprintf (stdout," %s \n",CGlobalInfo::dump_pool_as_json().c_str());
         }
+    }
 
 
-        if ( CGlobalInfo::m_options.is_rx_enabled() ){
-            m_mg.update();
+    if ( CGlobalInfo::m_options.is_rx_enabled() ) {
+        m_mg.update();
 
-            if ( m_io_modes.m_g_mode ==  CTrexGlobalIoMode::gNORMAL ){
-                switch (m_io_modes.m_l_mode) {
-                case CTrexGlobalIoMode::lDISABLE:
-                    fprintf(stdout,"\n+Latency stats disabled \n");
+        if ( m_io_modes.m_g_mode ==  CTrexGlobalIoMode::gNORMAL ) {
+            switch (m_io_modes.m_l_mode) {
+            case CTrexGlobalIoMode::lDISABLE:
+                fprintf(stdout,"\n+Latency stats disabled \n");
+                break;
+            case CTrexGlobalIoMode::lENABLE:
+                fprintf(stdout,"\n-Latency stats enabled \n");
+                m_mg.DumpShort(stdout);
+                break;
+            case CTrexGlobalIoMode::lENABLE_Extended:
+                fprintf(stdout,"\n-Latency stats extended \n");
+                m_mg.Dump(stdout);
+                break;
+            }
+
+            if ( get_is_rx_check_mode() ) {
+
+                switch (m_io_modes.m_rc_mode) {
+                case CTrexGlobalIoMode::rcDISABLE:
+                    fprintf(stdout,"\n+Rx Check stats disabled \n");
                     break;
-                case CTrexGlobalIoMode::lENABLE:
-                    fprintf(stdout,"\n-Latency stats enabled \n");
-                    m_mg.DumpShort(stdout);
+                case CTrexGlobalIoMode::rcENABLE:
+                    fprintf(stdout,"\n-Rx Check stats enabled \n");
+                    m_mg.DumpShortRxCheck(stdout);
                     break;
-                case CTrexGlobalIoMode::lENABLE_Extended:
-                    fprintf(stdout,"\n-Latency stats extended \n");
-                    m_mg.Dump(stdout);
+                case CTrexGlobalIoMode::rcENABLE_Extended:
+                    fprintf(stdout,"\n-Rx Check stats enhanced \n");
+                    m_mg.DumpRxCheck(stdout);
                     break;
                 }
 
-                if ( get_is_rx_check_mode() ) {
-
-                    switch (m_io_modes.m_rc_mode) {
-                    case CTrexGlobalIoMode::rcDISABLE:
-                        fprintf(stdout,"\n+Rx Check stats disabled \n");
-                        break;
-                    case CTrexGlobalIoMode::rcENABLE:
-                        fprintf(stdout,"\n-Rx Check stats enabled \n");
-                        m_mg.DumpShortRxCheck(stdout);
-                        break;
-                    case CTrexGlobalIoMode::rcENABLE_Extended:
-                        fprintf(stdout,"\n-Rx Check stats enhanced \n");
-                        m_mg.DumpRxCheck(stdout);
-                        break;
-                    }
-
-                }
-
             }
+
         }
+    }
 
+    /* publish data */
+    publish_async_data(false);
 
+    return true;
+}
 
-        /* publish data */
-        publish_async_data(false);
 
-        /* check from messages from DP */
-        check_for_dp_messages();
+bool
+CGlobalTRex::handle_fast_path() {
+    /* check from messages from DP */
+    check_for_dp_messages();
 
-        cp_lock.unlock();
-        delay(500);
-        cp_lock.lock();
+    if ( is_all_cores_finished() ) {
+        return false;
+    }
+   
+    return true;
+}
 
-        if ( is_all_cores_finished() ) {
+int CGlobalTRex::run_in_master() {
+    bool was_stopped=false;
+
+    if ( get_is_stateless() ) {
+        m_trex_stateless->launch_control_plane();
+    }
+
+    /* exception and scope safe */
+    std::unique_lock<std::mutex> cp_lock(m_cp_lock);
+
+    uint32_t slow_path_ms = 0;
+
+    while ( true ) {
+
+        if (slow_path_ms >= 500) {
+            if (!handle_slow_path(was_stopped)) {
+                break;
+            }
+            slow_path_ms = 0;
+        }
+        
+        if (!handle_fast_path()) {
             break;
         }
+
+        cp_lock.unlock();
+        delay(20);
+        slow_path_ms += 20;
+        cp_lock.lock();
     }
 
     /* on exit release the lock */
diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index d48c770e..5fe707af 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -462,14 +462,14 @@ TrexRpcCmdPushRemote::_run(const Json::Value &params, Json::Value &result) {
 
     uint8_t port_id = parse_port(params, result);
     std::string pcap_filename = parse_string(params, "pcap_filename", result);
-    double ipg                = parse_double(params, "ipg", result);
+    double ipg_usec           = parse_double(params, "ipg_usec", result);
     double speedup            = parse_double(params, "speedup", result);
     uint32_t count            = parse_uint32(params, "count", result);
 
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
     try {
-        port->push_remote(pcap_filename, ipg, speedup, count);
+        port->push_remote(pcap_filename, ipg_usec, speedup, count);
     } catch (const TrexException &ex) {
         generate_execute_err(result, ex.what());
     }
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;
 };