DP stop message now disables only port related nodes

and not all of them

9 files changed, 40 insertions, 14 deletions

diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 375d1f63..feea5ed5 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -69,7 +69,7 @@ TrexStatelessPort::start_traffic(void) {
 
     /* compiler it */
     TrexStreamsCompiler compiler;
-    TrexStreamsCompiledObj *compiled_obj = new TrexStreamsCompiledObj();
+    TrexStreamsCompiledObj *compiled_obj = new TrexStreamsCompiledObj(m_port_id);
 
     bool rc = compiler.compile(streams, *compiled_obj);
     if (!rc) {
@@ -99,7 +99,7 @@ TrexStatelessPort::stop_traffic(void) {
     }
 
     /* generate a message to all the relevant DP cores to start transmitting */
-    TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpStop();
+    TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpStop(m_port_id);
 
     // FIXME (add the right core list)
     CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(0);
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index 6c77ad0f..2e544995 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -26,6 +26,9 @@ limitations under the License.
 /**************************************
  * stream compiled object
  *************************************/
+TrexStreamsCompiledObj::TrexStreamsCompiledObj(uint8_t port_id) : m_port_id(port_id) {
+}
+
 TrexStreamsCompiledObj::~TrexStreamsCompiledObj() {
     for (auto &obj : m_objs) {
         delete obj.m_pkt;
@@ -37,6 +40,7 @@ void
 TrexStreamsCompiledObj::add_compiled_stream(double pps, uint8_t *pkt, uint16_t pkt_len) {
     obj_st obj;
 
+    obj.m_port_id = m_port_id;
     obj.m_pps = pps;
     obj.m_pkt_len = pkt_len;
 
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index 90253cdf..82318dec 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -36,13 +36,14 @@ class TrexStreamsCompiledObj {
     friend class TrexStreamsCompiler;
 public:
 
-    TrexStreamsCompiledObj() {}
+    TrexStreamsCompiledObj(uint8_t port_id);
     ~TrexStreamsCompiledObj();
 
     struct obj_st {
         double   m_pps;
         uint8_t *m_pkt;
         uint16_t m_pkt_len;
+        uint8_t  m_port_id;
     };
 
     const std::vector<obj_st> & get_objects() {
@@ -52,6 +53,8 @@ public:
 private:
     void add_compiled_stream(double pps, uint8_t *pkt, uint16_t pkt_len);
     std::vector<obj_st> m_objs;
+
+    uint8_t m_port_id;
 };
 
 class TrexStreamsCompiler {
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 9c10504e..fd54256d 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -89,13 +89,14 @@ TrexStatelessDpCore::add_cont_stream(double pps, const uint8_t *pkt, uint16_t pk
     /* set the packet as a readonly */
     node->set_cache_mbuf(m);
 
-    m_state = TrexStatelessDpCore::STATE_TRANSMITTING;
-
     /* keep track */
     m_active_nodes.push_back(node);
 
     /* schedule */
     m_core->m_node_gen.add_node((CGenNode *)node);
+
+    m_state = TrexStatelessDpCore::STATE_TRANSMITTING;
+
 }
 
 void
@@ -106,16 +107,27 @@ TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj) {
 }
 
 void
-TrexStatelessDpCore::stop_traffic() {
+TrexStatelessDpCore::stop_traffic(uint8_t port_id) {
     /* we cannot remove nodes not from the top of the queue so
        for every active node - make sure next time
        the scheduler invokes it, it will be free */
     for (auto node : m_active_nodes) {
-        node->m_is_stream_active = 0;
+        if (node->m_port_id == port_id) {
+            node->m_is_stream_active = 0;
+        }
     }
-    m_active_nodes.clear();
 
-    m_state = STATE_IDLE;
+    /* remove all the non active nodes */
+    auto pred = std::remove_if(m_active_nodes.begin(),
+                               m_active_nodes.end(), 
+                               [](CGenNodeStateless *node) { return (!node->m_is_stream_active); });
+
+    m_active_nodes.erase(pred, m_active_nodes.end());
+
+    if (m_active_nodes.size() == 0) {
+        m_state = STATE_IDLE;
+    }
+    
 }
 
 /**
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index fa3c5b22..b71431ad 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -64,7 +64,7 @@ public:
      * stop all traffic for this core
      * 
      */
-    void stop_traffic();
+    void stop_traffic(uint8_t port_id);
 
     /**
      * check for and handle messages from CP
diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h
index 49481a5c..92b428ab 100644
--- a/src/stateless/dp/trex_stream_node.h
+++ b/src/stateless/dp/trex_stream_node.h
@@ -34,9 +34,10 @@ private:
 
     double              m_next_time_offset;
     uint8_t             m_is_stream_active;
+    uint8_t             m_port_id;
 
     /* pad to match the size of CGenNode */
-    uint8_t             m_pad_end[39];
+    uint8_t             m_pad_end[87];
 
 
 public:
@@ -97,7 +98,7 @@ public:
     }
 
 
-} __rte_cache_aligned; ;
+} __rte_cache_aligned;
 
 static_assert(sizeof(CGenNodeStateless) == sizeof(CGenNode), "sizeof(CGenNodeStateless) != sizeof(CGenNode)");
 
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 3c6a5933..3e754649 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -47,7 +47,7 @@ TrexStatelessDpStart::handle(TrexStatelessDpCore *dp_core) {
  ************************/ 
 bool
 TrexStatelessDpStop::handle(TrexStatelessDpCore *dp_core) {
-    dp_core->stop_traffic();
+    dp_core->stop_traffic(m_port_id);
     return true;
 }
 
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index 13f6c05a..381e146d 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -72,7 +72,14 @@ private:
  */
 class TrexStatelessDpStop : public TrexStatelessCpToDpMsgBase {
 public:
+
+    TrexStatelessDpStop(uint8_t port_id) : m_port_id(port_id) {
+    }
+
     virtual bool handle(TrexStatelessDpCore *dp_core);
+
+private:
+    uint8_t m_port_id;
 };
 
 
diff --git a/src/stub/trex_stateless_stub.cpp b/src/stub/trex_stateless_stub.cpp
index 06a9e189..de56e57a 100644
--- a/src/stub/trex_stateless_stub.cpp
+++ b/src/stub/trex_stateless_stub.cpp
@@ -20,4 +20,3 @@ void TrexStatelessDpCore::start(){}
 
 void TrexStatelessDpCore::handle_cp_msg(TrexStatelessCpToDpMsgBase*) {}
 
-void TrexStatelessDpCore::handle_pkt_event(CGenNode*) {}