dp support for pause/resume only continues is supported

5 files changed, 147 insertions, 3 deletions

diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index f8afb3bb..03b13d6c 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -124,6 +124,37 @@ bool TrexStatelessDpPerPort::update_number_of_active_streams(uint32_t d){
     return (false);
 }
 
+bool TrexStatelessDpPerPort::resume_traffic(uint8_t port_id){
+
+    /* we are working with continues streams so we must be in transmit mode */
+    assert(m_state == TrexStatelessDpPerPort::ppSTATE_PAUSE);
+
+    for (auto dp_stream : m_active_nodes) {
+        CGenNodeStateless * node =dp_stream.m_node; 
+        assert(node->get_port_id() == port_id);
+        assert(node->is_pause() == true);
+        node->set_pause(false);
+    }
+    m_state = TrexStatelessDpPerPort::ppSTATE_TRANSMITTING;
+    return (true);
+}
+
+
+bool TrexStatelessDpPerPort::pause_traffic(uint8_t port_id){
+
+    /* we are working with continues streams so we must be in transmit mode */
+    assert(m_state == TrexStatelessDpPerPort::ppSTATE_TRANSMITTING);
+
+    for (auto dp_stream : m_active_nodes) {
+        CGenNodeStateless * node =dp_stream.m_node; 
+        assert(node->get_port_id() == port_id);
+        assert(node->is_pause() == false);
+        node->set_pause(true);
+    }
+    m_state = TrexStatelessDpPerPort::ppSTATE_PAUSE;
+    return (true);
+}
+
 
 bool TrexStatelessDpPerPort::stop_traffic(uint8_t port_id,
                                           bool stop_on_id, 
@@ -294,6 +325,8 @@ TrexStatelessDpCore::run_once(){
 }
 
 
+
+
 void
 TrexStatelessDpCore::start() {
 
@@ -402,6 +435,7 @@ TrexStatelessDpCore::add_cont_stream(TrexStatelessDpPerPort * lp_port,
     uint16_t pkt_size = stream->m_pkt.len;
     const uint8_t *stream_pkt = stream->m_pkt.binary;
 
+    node->m_pause =0;
     node->m_stream_type = stream->m_type;
     node->m_next_time_offset =  1.0 / (stream->get_pps() * comp->get_multiplier()) ;
 
@@ -529,6 +563,24 @@ bool TrexStatelessDpCore::are_all_ports_idle(){
 }
 
 
+void 
+TrexStatelessDpCore::resume_traffic(uint8_t port_id){
+
+    TrexStatelessDpPerPort * lp_port = get_port_db(port_id);
+
+    lp_port->resume_traffic(port_id);
+}
+
+
+void 
+TrexStatelessDpCore::pause_traffic(uint8_t port_id){
+
+    TrexStatelessDpPerPort * lp_port = get_port_db(port_id);
+
+    lp_port->pause_traffic(port_id);
+}
+
+
 void
 TrexStatelessDpCore::stop_traffic(uint8_t port_id,
                                   bool stop_on_id, 
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index 187c40d8..eda1ae59 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -53,7 +53,9 @@ public:
         /* states */
     enum state_e {
         ppSTATE_IDLE,
-        ppSTATE_TRANSMITTING
+        ppSTATE_TRANSMITTING,
+        ppSTATE_PAUSE
+
     };
 
 public:
@@ -62,6 +64,10 @@ public:
 
     void create(CFlowGenListPerThread   *  core);
 
+    bool pause_traffic(uint8_t port_id);
+
+    bool resume_traffic(uint8_t port_id);
+
     bool stop_traffic(uint8_t port_id,
                       bool stop_on_id, 
                       int event_id);
@@ -143,7 +149,17 @@ public:
                        double duration,
                        int m_event_id);
 
+
+    /* pause the streams, work only if all are continues  */
+    void pause_traffic(uint8_t port_id);
+
+
+
+    void resume_traffic(uint8_t port_id);
+
+
     /**
+     * 
      * stop all traffic for this core
      * 
      */
diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h
index 20e32b78..ccf99eaa 100644
--- a/src/stateless/dp/trex_stream_node.h
+++ b/src/stateless/dp/trex_stream_node.h
@@ -48,6 +48,7 @@ public:
 
 static_assert(sizeof(CGenNodeCommand) == sizeof(CGenNode), "sizeof(CGenNodeCommand) != sizeof(CGenNode)" );
 
+
 /* this is a event for stateless */
 struct CGenNodeStateless : public CGenNodeBase  {
 friend class TrexStatelessDpCore;
@@ -74,7 +75,7 @@ private:
     stream_state_t      m_state;
     uint8_t             m_port_id;
     uint8_t             m_stream_type; /* see TrexStream::STREAM_TYPE ,stream_type_t */
-    uint8_t             m_pad;
+    uint8_t             m_pause;
 
     uint32_t            m_single_burst; /* the number of bursts in case of burst */
     uint32_t            m_single_burst_refill; 
@@ -112,7 +113,18 @@ public:
         /* only to be safe */
         m_ref_stream_info= NULL;
         m_next_stream= NULL;
+    }
 
+    bool is_pause(){
+        return (m_pause==1?true:false);
+    }
+
+    void set_pause(bool enable){
+        if ( enable ){
+            m_pause=1;
+        }else{
+            m_pause=0;
+        }
     }
 
     inline uint8_t  get_stream_type(){
@@ -143,7 +155,10 @@ public:
     void refresh();
 
     inline void handle_continues(CFlowGenListPerThread *thread) {
-        thread->m_node_gen.m_v_if->send_node( (CGenNode *)this);
+
+        if (unlikely (is_pause()==false)) {
+            thread->m_node_gen.m_v_if->send_node( (CGenNode *)this);
+        }
 
         /* in case of continues */
         m_time += m_next_time_offset;
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index bbd4b68c..ec8b7839 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -87,6 +87,31 @@ void TrexStatelessDpStop::on_node_remove(){
 }
 
 
+TrexStatelessCpToDpMsgBase * TrexStatelessDpPause::clone(){
+
+    TrexStatelessDpPause *new_msg = new TrexStatelessDpPause(m_port_id);
+    return new_msg;
+}
+
+
+bool TrexStatelessDpPause::handle(TrexStatelessDpCore *dp_core){
+    dp_core->pause_traffic(m_port_id);
+    return (true);
+}
+
+
+
+TrexStatelessCpToDpMsgBase * TrexStatelessDpResume::clone(){
+    TrexStatelessDpResume *new_msg = new TrexStatelessDpResume(m_port_id);
+    return new_msg;
+}
+
+bool TrexStatelessDpResume::handle(TrexStatelessDpCore *dp_core){
+    dp_core->resume_traffic(m_port_id);
+    return (true);
+}
+
+
 /**
  * clone for DP stop message
  * 
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index afa5953a..6bd0dbe3 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -99,6 +99,42 @@ private:
 
 };
 
+class TrexStatelessDpPause : public TrexStatelessCpToDpMsgBase {
+public:
+
+    TrexStatelessDpPause(uint8_t port_id) : m_port_id(port_id) {
+    }
+
+
+    virtual TrexStatelessCpToDpMsgBase * clone();
+
+
+    virtual bool handle(TrexStatelessDpCore *dp_core);
+
+
+private:
+    uint8_t m_port_id;
+};
+
+
+class TrexStatelessDpResume : public TrexStatelessCpToDpMsgBase {
+public:
+
+    TrexStatelessDpResume(uint8_t port_id) : m_port_id(port_id) {
+    }
+
+
+    virtual TrexStatelessCpToDpMsgBase * clone();
+
+
+    virtual bool handle(TrexStatelessDpCore *dp_core);
+
+
+private:
+    uint8_t m_port_id;
+};
+
+
 /**
  * a message to stop traffic
  *