burst bug -

generate compiler object per core
send message per core instead of duplicating it

6 files changed, 213 insertions, 117 deletions

diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 25fae038..10e863c5 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -115,7 +115,7 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration)
     /* on start - we can only provide absolute values */
     assert(mul.m_op == TrexPortMultiplier::OP_ABS);
 
-    double per_core_mul = calculate_effective_mul(mul);
+    double factor = calculate_effective_factor(mul);
 
     /* fetch all the streams from the table */
     vector<TrexStream *> streams;
@@ -123,11 +123,16 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration)
 
 
     /* compiler it */
-    TrexStreamsCompiler compiler;
-    TrexStreamsCompiledObj *compiled_obj = new TrexStreamsCompiledObj(m_port_id, per_core_mul);
-
+    std::vector<TrexStreamsCompiledObj *> compiled_objs;
     std::string fail_msg;
-    bool rc = compiler.compile(streams, *compiled_obj, &fail_msg);
+
+    TrexStreamsCompiler compiler;
+    bool rc = compiler.compile(m_port_id,
+                               get_dp_core_count(),
+                               factor,
+                               streams,
+                               compiled_objs,
+                               &fail_msg);
     if (!rc) {
         throw TrexRpcException(fail_msg);
     }
@@ -138,21 +143,29 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration)
     /* mark that DP event of stoppped is possible */
     m_dp_events.wait_for_event(TrexDpPortEvent::EVENT_STOP, event_id);
 
-    TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_port_id, event_id, compiled_obj, duration);
-
-    m_last_all_streams_continues = compiled_obj->get_all_streams_continues();
-    m_last_duration =duration;
 
+    /* update object status */
+    m_factor = factor;
+    m_last_all_streams_continues = compiled_objs[0]->get_all_streams_continues();
+    m_last_duration = duration;
     change_state(PORT_STATE_TX);
 
-    send_message_to_dp(start_msg);
+
+    /* update the DP - messages will be freed by the DP */
+    int index = 0;
+    for (auto core_id : m_cores_id_list) {
+
+        TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_port_id, event_id, compiled_objs[index], duration);
+        send_message_to_dp(core_id, start_msg);
+
+        index++;
+    }
     
+    /* update subscribers */    
     Json::Value data;
     data["port_id"] = m_port_id;
     get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STARTED, data);
-
-    /* save the per core multiplier for update messages */
-    m_current_per_core_m = per_core_mul;
+    
 }
 
 
@@ -180,7 +193,7 @@ TrexStatelessPort::stop_traffic(void) {
     /* generate a message to all the relevant DP cores to start transmitting */
     TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpStop(m_port_id);
 
-    send_message_to_dp(stop_msg);
+    send_message_to_all_dp(stop_msg);
 
     change_state(PORT_STATE_STREAMS);
     
@@ -203,9 +216,9 @@ TrexStatelessPort::pause_traffic(void) {
         throw TrexRpcException(" pause is supported when duration is not enable is start command ");
     }
 
-    TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpPause(m_port_id);
+    TrexStatelessCpToDpMsgBase *pause_msg = new TrexStatelessDpPause(m_port_id);
 
-    send_message_to_dp(stop_msg);
+    send_message_to_all_dp(pause_msg);
 
     change_state(PORT_STATE_PAUSE);
 }
@@ -216,9 +229,9 @@ TrexStatelessPort::resume_traffic(void) {
     verify_state(PORT_STATE_PAUSE);
 
     /* generate a message to all the relevant DP cores to start transmitting */
-    TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpResume(m_port_id);
+    TrexStatelessCpToDpMsgBase *resume_msg = new TrexStatelessDpResume(m_port_id);
 
-    send_message_to_dp(stop_msg);
+    send_message_to_all_dp(resume_msg);
 
     change_state(PORT_STATE_TX);
 }
@@ -231,19 +244,19 @@ TrexStatelessPort::update_traffic(const TrexPortMultiplier &mul) {
     verify_state(PORT_STATE_TX | PORT_STATE_PAUSE);
 
     /* generate a message to all the relevant DP cores to start transmitting */
-    double new_per_core_m = calculate_effective_mul(mul);
+    double new_factor = calculate_effective_factor(mul);
 
     switch (mul.m_op) {
     case TrexPortMultiplier::OP_ABS:
-        factor = new_per_core_m / m_current_per_core_m;
+        factor = new_factor / m_factor;
         break;
 
     case TrexPortMultiplier::OP_ADD:
-        factor = (m_current_per_core_m + new_per_core_m) / m_current_per_core_m;
+        factor = (m_factor + new_factor) / m_factor;
         break;
 
     case TrexPortMultiplier::OP_SUB:
-        factor = (m_current_per_core_m - new_per_core_m) / m_current_per_core_m;
+        factor = (m_factor - new_factor) / m_factor;
         if (factor <= 0) {
             throw TrexRpcException("Update request will lower traffic to less than zero");
         }
@@ -256,9 +269,9 @@ TrexStatelessPort::update_traffic(const TrexPortMultiplier &mul) {
 
     TrexStatelessCpToDpMsgBase *update_msg = new TrexStatelessDpUpdate(m_port_id, factor);
 
-    send_message_to_dp(update_msg);
+    send_message_to_all_dp(update_msg);
 
-    m_current_per_core_m *= factor;
+    m_factor *= factor;
 
 }
 
@@ -357,15 +370,22 @@ TrexStatelessPort::encode_stats(Json::Value &port) {
 }
 
 void 
-TrexStatelessPort::send_message_to_dp(TrexStatelessCpToDpMsgBase *msg) {
+TrexStatelessPort::send_message_to_all_dp(TrexStatelessCpToDpMsgBase *msg) {
 
     for (auto core_id : m_cores_id_list) {
-
-        /* send the message to the core */
-        CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_id);
-        ring->Enqueue((CGenNode *)msg->clone());
+        send_message_to_dp(core_id, msg->clone());
     }
 
+    /* original was not sent - delete it */
+    delete msg;
+}
+
+void 
+TrexStatelessPort::send_message_to_dp(uint8_t core_id, TrexStatelessCpToDpMsgBase *msg) {
+
+    /* send the message to the core */
+    CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_id);
+    ring->Enqueue((CGenNode *)msg);
 }
 
 /**
@@ -411,11 +431,11 @@ TrexStatelessPort::get_port_speed_bps() const {
 }
 
 double
-TrexStatelessPort::calculate_effective_mul(const TrexPortMultiplier &mul) {
+TrexStatelessPort::calculate_effective_factor(const TrexPortMultiplier &mul) {
 
-    /* for a simple factor request - calculate the multiplier per core */
+    /* for a simple factor request */
     if (mul.m_type == TrexPortMultiplier::MUL_FACTOR) {
-        return (mul.m_value /  m_cores_id_list.size());
+        return (mul.m_value);
     }
 
     /* we now need the graph - generate it if we don't have it (happens once) */
@@ -425,19 +445,19 @@ TrexStatelessPort::calculate_effective_mul(const TrexPortMultiplier &mul) {
 
     switch (mul.m_type) {
     case TrexPortMultiplier::MUL_BPS:
-        return ( (mul.m_value / m_graph_obj->get_max_bps()) / m_cores_id_list.size());
+        return (mul.m_value / m_graph_obj->get_max_bps());
 
     case TrexPortMultiplier::MUL_PPS:
-         return ( (mul.m_value / m_graph_obj->get_max_pps()) / m_cores_id_list.size());
+         return (mul.m_value / m_graph_obj->get_max_pps());
 
     case TrexPortMultiplier::MUL_PERCENTAGE:
         /* if abs percentage is from the line speed - otherwise its from the current speed */
 
         if (mul.m_op == TrexPortMultiplier::OP_ABS) {
             double required = (mul.m_value / 100.0) * get_port_speed_bps();
-            return ( (required / m_graph_obj->get_max_bps()) / m_cores_id_list.size());
+            return (required / m_graph_obj->get_max_bps());
         } else {
-            return (m_current_per_core_m * (mul.m_value / 100.0));
+            return (m_factor * (mul.m_value / 100.0));
         }
 
     default:
@@ -532,14 +552,23 @@ TrexStatelessPort::validate(void) {
     }
 
     TrexStreamsCompiler compiler;
-    TrexStreamsCompiledObj compiled_obj(m_port_id, 1);
+    std::vector<TrexStreamsCompiledObj *> compiled_objs;
 
     std::string fail_msg;
-    bool rc = compiler.compile(streams, compiled_obj, &fail_msg);
+    bool rc = compiler.compile(m_port_id,
+                               get_dp_core_count(),
+                               1.0,
+                               streams,
+                               compiled_objs,
+                               &fail_msg);
     if (!rc) {
         throw TrexException(fail_msg);
     }
 
+    for (auto obj : compiled_objs) {
+        delete obj;
+    }
+
     /* now create a stream graph */
     if (!m_graph_obj) {
         generate_streams_graph();
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index dbaac21d..28e42a17 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -225,11 +225,19 @@ public:
 
 
     /**
+     * returns the number of DP cores linked to this port
+     * 
+     */
+    uint8_t get_dp_core_count() {
+        return m_cores_id_list.size();
+    }
+
+    /**
      * returns the traffic multiplier currently being used by the DP
      * 
      */
     double get_multiplier() {
-        return (m_current_per_core_m * m_cores_id_list.size()); 
+        return (m_factor);
     }
 
     /**
@@ -273,7 +281,17 @@ private:
 
     std::string generate_handler();
 
-    void send_message_to_dp(TrexStatelessCpToDpMsgBase *msg);
+    /**
+     * send message to all cores using duplicate
+     * 
+     */
+    void send_message_to_all_dp(TrexStatelessCpToDpMsgBase *msg);
+
+    /**
+     * send message to specific DP core
+     * 
+     */
+    void send_message_to_dp(uint8_t core_id, TrexStatelessCpToDpMsgBase *msg);
 
     /**
      * triggered when event occurs
@@ -286,7 +304,7 @@ private:
      * calculate effective M per core
      * 
      */
-    double calculate_effective_mul(const TrexPortMultiplier &mul);
+    double calculate_effective_factor(const TrexPortMultiplier &mul);
 
   
 
@@ -318,7 +336,7 @@ private:
 
     bool               m_last_all_streams_continues;
     double             m_last_duration;
-    double             m_current_per_core_m;
+    double             m_factor;
 
     TrexDpPortEvents   m_dp_events;
 
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index 3e48d7e4..b991b05f 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -129,12 +129,13 @@ public:
     }
 
     /* create new stream */
-    TrexStream * clone_as_dp(){
-            TrexStream * dp=new TrexStream(m_type,m_port_id,m_stream_id);
+    TrexStream * clone_as_dp() const {
+
+            TrexStream *dp = new TrexStream(m_type,m_port_id,m_stream_id);
         
 
-            dp->m_isg_usec      = m_isg_usec;
-            dp->m_next_stream_id = m_next_stream_id;
+            dp->m_isg_usec        = m_isg_usec;
+            dp->m_next_stream_id  = m_next_stream_id;
 
             dp->m_enabled    = m_enabled;
             dp->m_self_start = m_self_start;
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index 1c601f85..36b165d1 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -142,8 +142,9 @@ private:
 /**************************************
  * stream compiled object
  *************************************/
-TrexStreamsCompiledObj::TrexStreamsCompiledObj(uint8_t port_id, double mul) : m_port_id(port_id), m_mul(mul) {
-      m_all_continues=false;
+TrexStreamsCompiledObj::TrexStreamsCompiledObj(uint8_t port_id) {
+    m_port_id = port_id;
+    m_all_continues = false;
 }
 
 TrexStreamsCompiledObj::~TrexStreamsCompiledObj() {
@@ -155,53 +156,40 @@ TrexStreamsCompiledObj::~TrexStreamsCompiledObj() {
 
 
 void 
-TrexStreamsCompiledObj::add_compiled_stream(TrexStream * stream){
+TrexStreamsCompiledObj::add_compiled_stream(TrexStream *stream){
 
     obj_st obj;
 
-    obj.m_stream = stream->clone_as_dp();
+    obj.m_stream = stream;
 
     m_objs.push_back(obj);
 }
 
-void 
-TrexStreamsCompiledObj::add_compiled_stream(TrexStream * stream,
-                                            uint32_t my_dp_id, int next_dp_id) {
-    obj_st obj;
-
-    obj.m_stream = stream->clone_as_dp();
-    /* compress the id's*/
-    obj.m_stream->fix_dp_stream_id(my_dp_id,next_dp_id);
-
-    m_objs.push_back(obj);
-}
-
-void TrexStreamsCompiledObj::Dump(FILE *fd){
-    for (auto obj : m_objs) {
-        obj.m_stream->Dump(fd);
-    }
-}
-
-
 
 TrexStreamsCompiledObj *
 TrexStreamsCompiledObj::clone() {
 
-    /* use multiplier of 1 to avoid double mult */
-    TrexStreamsCompiledObj *new_compiled_obj = new TrexStreamsCompiledObj(m_port_id, 1);
+    TrexStreamsCompiledObj *new_compiled_obj = new TrexStreamsCompiledObj(m_port_id);
 
     /**
      * clone each element
      */
     for (auto obj : m_objs) {
-        new_compiled_obj->add_compiled_stream(obj.m_stream);
+        TrexStream *new_stream = obj.m_stream->clone_as_dp();
+        new_compiled_obj->add_compiled_stream(new_stream);
     }
 
-    new_compiled_obj->m_mul = m_mul;
-
     return new_compiled_obj;
+
+}
+
+void TrexStreamsCompiledObj::Dump(FILE *fd){
+    for (auto obj : m_objs) {
+        obj.m_stream->Dump(fd);
+    }
 }
 
+
 void
 TrexStreamsCompiler::add_warning(const std::string &warning) {
     m_warnings.push_back("*** warning: " + warning);
@@ -381,12 +369,14 @@ TrexStreamsCompiler::pre_compile_check(const std::vector<TrexStream *> &streams,
  * stream compiler
  *************************************/
 bool 
-TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams,
-                             TrexStreamsCompiledObj &obj,
-                             std::string *fail_msg) {
+TrexStreamsCompiler::compile(uint8_t                                port_id,
+                             uint8_t                                dp_core_count,
+                             double                                 factor,
+                             const std::vector<TrexStream *>        &streams,
+                             std::vector<TrexStreamsCompiledObj *>  &objs,
+                             std::string                            *fail_msg) {
 
 #if 0
-    fprintf(stdout,"------------pre compile \n");
     for (auto stream : streams) {
         stream->Dump(stdout);
     }
@@ -398,7 +388,7 @@ TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams,
 
     /* compile checks */
     try {
-        pre_compile_check(streams,nodes);
+        pre_compile_check(streams, nodes);
     } catch (const TrexException &ex) {
         if (fail_msg) {
             *fail_msg = ex.what();
@@ -408,38 +398,94 @@ TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams,
         return false;
     }
 
+    /* check if all are cont. streams */
+    bool all_continues = true;
+    for (const auto stream : streams) {
+        if (stream->get_type() != TrexStream::stCONTINUOUS) {
+            all_continues = false;
+            break;
+        }
+    }
+
+    /* allocate objects for all DP cores */
+    for (uint8_t i = 0; i < dp_core_count; i++) {
+        TrexStreamsCompiledObj *obj = new TrexStreamsCompiledObj(port_id);
+        obj->m_all_continues = all_continues;
+        objs.push_back(obj);
+    }
 
-    bool all_continues=true;
-    /* for now we do something trivial, */
+    /* compile all the streams */
     for (auto stream : streams) {
 
         /* skip non-enabled streams */
         if (!stream->m_enabled) {
             continue;
         }
-        if (stream->get_type() != TrexStream::stCONTINUOUS ) {
-            all_continues=false;
-        }
-
-        int new_id= nodes.get(stream->m_stream_id)->m_compressed_stream_id;
-        assert(new_id>=0);
-        uint32_t my_stream_id = (uint32_t)new_id;
-        int my_next_stream_id=-1;
-        if (stream->m_next_stream_id>=0) {
-            my_next_stream_id=nodes.get(stream->m_next_stream_id)->m_compressed_stream_id;
-        }
-
-        /* add it */
-        obj.add_compiled_stream(stream,
-                                my_stream_id,
-                                my_next_stream_id
-                                );
+     
+        /* compile a single stream to all cores */
+        compile_stream(stream, factor, dp_core_count, objs, nodes);
         
     }
-    obj.m_all_continues =all_continues;
+
     return true;
 }
 
+/**
+ * compiles a single stream to DP objects
+ * 
+ * @author imarom (03-Dec-15)
+ * 
+ */
+void
+TrexStreamsCompiler::compile_stream(const TrexStream *stream,
+                                    double factor,
+                                    uint8_t dp_core_count,
+                                    std::vector<TrexStreamsCompiledObj *> &objs,
+                                    GraphNodeMap &nodes) {
+
+
+    /* fix the stream ids */
+    int new_id = nodes.get(stream->m_stream_id)->m_compressed_stream_id;
+    assert(new_id >= 0);
+
+    int new_next_id = -1;
+    if (stream->m_next_stream_id >= 0) {
+        new_next_id = nodes.get(stream->m_next_stream_id)->m_compressed_stream_id;
+    }
+
+    /* calculate rate */
+    double per_core_rate          = (stream->m_pps * (factor / dp_core_count));
+    int per_core_burst_total_pkts = (stream->m_burst_total_pkts / dp_core_count);
+
+    std::vector<TrexStream *> per_core_streams(dp_core_count);
+
+    /* for each core - creates its own version of the stream */
+    for (uint8_t i = 0; i < dp_core_count; i++) {
+        TrexStream *dp_stream = stream->clone_as_dp();
+
+        /* fix stream ID */
+        dp_stream->fix_dp_stream_id(new_id, new_next_id);
+
+
+        /* adjust rate and packets count */
+        dp_stream->m_pps               = per_core_rate;
+        dp_stream->m_burst_total_pkts  = per_core_burst_total_pkts;
+
+        per_core_streams[i] = dp_stream;
+    }
+
+    /* take care of remainder from a burst */
+    int burst_remainder = stream->m_burst_total_pkts - (per_core_burst_total_pkts * dp_core_count);
+    per_core_streams[0]->m_burst_total_pkts += burst_remainder;
+
+    /* attach the compiled stream of every core to its object */
+    for (uint8_t i = 0; i < dp_core_count; i++) {
+        objs[i]->add_compiled_stream(per_core_streams[i]);
+    }
+
+
+}
+
 /**************************************
  * streams graph
  *************************************/
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index c1cf3811..93a63061 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -38,9 +38,10 @@ class GraphNodeMap;
  */
 class TrexStreamsCompiledObj {
     friend class TrexStreamsCompiler;
+
 public:
 
-    TrexStreamsCompiledObj(uint8_t port_id, double m_mul);
+    TrexStreamsCompiledObj(uint8_t port_id);
     ~TrexStreamsCompiledObj();
 
     struct obj_st {
@@ -56,32 +57,22 @@ public:
         return (m_port_id);
     }
 
-    /**
-     * clone the compiled object
-     * 
-     */
-    TrexStreamsCompiledObj * clone();
-
-    double get_multiplier(){
-        return (m_mul);
-    }
-
     bool get_all_streams_continues(){
         return (m_all_continues);
     }
 
     void Dump(FILE *fd);
 
+    TrexStreamsCompiledObj* clone();
+
 private:
-    void add_compiled_stream(TrexStream * stream,
-                             uint32_t my_dp_id, int next_dp_id);
-    void add_compiled_stream(TrexStream * stream);
+    void add_compiled_stream(TrexStream *stream);
+
 
     std::vector<obj_st> m_objs;
 
     bool    m_all_continues;
     uint8_t m_port_id;
-    double  m_mul;
 };
 
 class TrexStreamsCompiler {
@@ -93,7 +84,13 @@ public:
      * @author imarom (28-Oct-15)
      * 
      */
-    bool compile(const std::vector<TrexStream *> &streams, TrexStreamsCompiledObj &obj, std::string *fail_msg = NULL);
+    bool compile(uint8_t                                port_id,
+                 uint8_t                                dp_core_count,
+                 double                                 factor,
+                 const std::vector<TrexStream *>        &streams,
+                 std::vector<TrexStreamsCompiledObj *>  &objs,
+                 std::string                            *fail_msg = NULL);
+
 
     /**
      * 
@@ -115,8 +112,13 @@ private:
     void add_warning(const std::string &warning);
     void err(const std::string &err);
 
+    void compile_stream(const TrexStream *stream,
+                        double factor,
+                        uint8_t dp_core_count,
+                        std::vector<TrexStreamsCompiledObj *> &objs,
+                        GraphNodeMap &nodes);
+
     std::vector<std::string> m_warnings;
-    
 };
 
 class TrexStreamsGraph;
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 9b62fabd..22ca922d 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -453,7 +453,7 @@ TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
 
     node->m_pause =0;
     node->m_stream_type = stream->m_type;
-    node->m_next_time_offset = 1.0 / (stream->get_pps() * comp->get_multiplier());
+    node->m_next_time_offset = 1.0 / stream->get_pps();
 
     /* stateless specific fields */
     switch ( stream->m_type ) {