5 files changed, 51 insertions, 18 deletions

diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 86892e05..cf4de739 100755
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -3444,7 +3444,7 @@ bool CGlobalTRex::Create(){
           cfg.m_port_count         = CGlobalInfo::m_options.m_expected_portd;
           cfg.m_rpc_req_resp_cfg   = &rpc_req_resp_cfg;
           cfg.m_rpc_async_cfg      = NULL;
-          cfg.m_rpc_server_verbose = true;
+          cfg.m_rpc_server_verbose = false;
           cfg.m_platform_api       = new TrexDpdkPlatformApi();
 
           m_trex_stateless = new TrexStateless(cfg);
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 3e6e256f..375d1f63 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -76,9 +76,6 @@ TrexStatelessPort::start_traffic(void) {
         return (RC_ERR_FAILED_TO_COMPILE_STREAMS);
     }
 
-    /* move the state to transmiting */
-    m_port_state = PORT_STATE_TRANSMITTING;
-
     /* generate a message to all the relevant DP cores to start transmitting */
     TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(compiled_obj);
 
@@ -87,15 +84,18 @@ TrexStatelessPort::start_traffic(void) {
 
     ring->Enqueue((CGenNode *)start_msg);
 
+    /* move the state to transmiting */
+    m_port_state = PORT_STATE_TRANSMITTING;
+
     return (RC_OK);
 }
 
-void 
+TrexStatelessPort::rc_e
 TrexStatelessPort::stop_traffic(void) {
 
     /* real code goes here */
-    if (m_port_state == PORT_STATE_TRANSMITTING) {
-        m_port_state = PORT_STATE_UP_IDLE;
+    if (m_port_state != PORT_STATE_TRANSMITTING) {
+        return (RC_ERR_BAD_STATE_FOR_OP);
     }
 
     /* generate a message to all the relevant DP cores to start transmitting */
@@ -105,6 +105,10 @@ TrexStatelessPort::stop_traffic(void) {
     CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(0);
 
     ring->Enqueue((CGenNode *)stop_msg);
+
+    m_port_state = PORT_STATE_UP_IDLE;
+
+    return (RC_OK);
 }
 
 /**
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 474fccf7..4851a4b5 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -62,7 +62,7 @@ public:
      * stop traffic
      * 
      */
-    void stop_traffic(void);
+    rc_e stop_traffic(void);
 
     /**
      * access the stream table
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index f64686d4..73387f0e 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -24,20 +24,27 @@ limitations under the License.
 
 #include <bp_sim.h>
 
+/**
+ * extended info for the stateless node 
+ * TODO: 
+ * static_assert(sizeof(dp_node_extended_info_st) <= sizeof(CGenNodeStateless::m_pad_end), "hello"); 
+ */
 typedef struct dp_node_extended_info_ {
-    double next_time_offset;
+    double   next_time_offset;
+    uint8_t  is_stream_active;
+
 } dp_node_extended_info_st;
 
 TrexStatelessDpCore::TrexStatelessDpCore(uint8_t thread_id, CFlowGenListPerThread *core) {
     m_thread_id = thread_id;
     m_core = core;
 
-    m_state = STATE_IDLE;
-
     CMessagingManager * cp_dp = CMsgIns::Ins()->getCpDp();
 
     m_ring_from_cp = cp_dp->getRingCpToDp(thread_id);
     m_ring_to_cp   = cp_dp->getRingDpToCp(thread_id);
+
+    m_state = STATE_IDLE;
 }
 
 void
@@ -57,17 +64,20 @@ TrexStatelessDpCore::start() {
 void
 TrexStatelessDpCore::handle_pkt_event(CGenNode *node) {
 
-    /* if port has stopped - no transmition */
-    if (m_state == STATE_IDLE) {
+    //TODO: optimize the fast path here...
+
+    CGenNodeStateless *node_sl = (CGenNodeStateless *)node;
+    dp_node_extended_info_st *opaque = (dp_node_extended_info_st *)node_sl->get_opaque_storage();
+
+    /* is this stream active ? */
+    if (!opaque->is_stream_active) {
+        m_core->free_node(node);
         return;
     }
 
     m_core->m_node_gen.m_v_if->send_node(node);
 
-    CGenNodeStateless *node_sl = (CGenNodeStateless *)node;
-
     /* in case of continues */
-    dp_node_extended_info_st *opaque = (dp_node_extended_info_st *)node_sl->get_opaque_storage();
     node->m_time += opaque->next_time_offset;
 
     /* insert a new event */
@@ -92,6 +102,7 @@ TrexStatelessDpCore::add_cont_stream(double pps, const uint8_t *pkt, uint16_t pk
 
     dp_node_extended_info_st *opaque = (dp_node_extended_info_st *)node->get_opaque_storage();
     opaque->next_time_offset = 1.0 / pps;
+    opaque->is_stream_active = 1;
 
     /* allocate const mbuf */
     rte_mbuf_t *m = CGlobalInfo::pktmbuf_alloc(node->get_socket_id(), pkt_size);
@@ -114,6 +125,10 @@ TrexStatelessDpCore::add_cont_stream(double pps, const uint8_t *pkt, uint16_t pk
 
     m_state = TrexStatelessDpCore::STATE_TRANSMITTING;
 
+    /* keep track */
+    m_active_nodes.push_back(node);
+
+    /* schedule */
     m_core->m_node_gen.add_node((CGenNode *)node);
 }
 
@@ -126,6 +141,15 @@ TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj) {
 
 void
 TrexStatelessDpCore::stop_traffic() {
+    /* 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) {
+        dp_node_extended_info_st *opaque = (dp_node_extended_info_st *)node->get_opaque_storage();
+        opaque->is_stream_active = 0;
+    }
+    m_active_nodes.clear();
+
     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 da8484a6..d95f7eeb 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -21,6 +21,8 @@ limitations under the License.
 #ifndef __TREX_STATELESS_DP_CORE_H__
 #define __TREX_STATELESS_DP_CORE_H__
 
+#include <vector>
+
 #include <msg_manager.h>
 #include <pal_utl.h>
 
@@ -29,6 +31,7 @@ class TrexStatelessDpStart;
 class CFlowGenListPerThread;
 class CGenNode;
 class TrexStreamsCompiledObj;
+class CGenNodeStateless;
 
 class TrexStatelessDpCore {
 
@@ -85,11 +88,10 @@ public:
         }
 
         while ( true ) {
-            CGenNode * node;
+            CGenNode * node = NULL;
             if (m_ring_from_cp->Dequeue(node) != 0) {
                 break;
             }
-
             assert(node);
 
             TrexStatelessCpToDpMsgBase * msg = (TrexStatelessCpToDpMsgBase *)node;
@@ -115,6 +117,9 @@ private:
     CNodeRing           *m_ring_from_cp;
     CNodeRing           *m_ring_to_cp;
 
+    /* holds the current active nodes */
+    std::vector<CGenNodeStateless *> m_active_nodes;
+
     /* pointer to the main object */
     CFlowGenListPerThread *m_core;
 };