added support for 'update'

8 files changed, 150 insertions, 38 deletions

diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index f148261c..96224d4e 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -479,18 +479,26 @@ TrexRpcCmdStartTraffic::_run(const Json::Value &params, Json::Value &result) {
     std::string mul_type = parse_string(mul, "type", result);
     double max_rate = parse_double(mul, "max", result);
 
+
+    double m = 0;
+
     /* dispatch according to type of multiplier  */
+    if (mul_type == "raw") {
+        m = max_rate;
 
-    try {
-        if (mul_type == "raw") {
-            port->start_traffic(max_rate, duration);
+    } else if (mul_type == "max_bps") {
+        m = port->calculate_m_from_bps(max_rate);
 
-        } else if (mul_type == "max_bps") {
-            port->start_traffic_max_bps(max_rate, duration);
+    } else if (mul_type == "max_pps") {
+        m = port->calculate_m_from_pps(max_rate);
 
-        } else if (mul_type == "max_pps") {
-            port->start_traffic_max_pps(max_rate, duration);
-        }
+    } else {
+        generate_parse_err(result, "multiplier type can be either 'raw', 'max_bps' or 'max_pps'");
+    }
+
+
+    try {
+        port->start_traffic(m, duration);
 
     } catch (const TrexRpcException &ex) {
         generate_execute_err(result, ex.what());
@@ -633,7 +641,6 @@ trex_rpc_cmd_rc_e
 TrexRpcCmdUpdateTraffic::_run(const Json::Value &params, Json::Value &result) {
 
     uint8_t port_id = parse_byte(params, "port_id", result);
-    double mul = parse_double(params, "mul", result);
 
     if (port_id >= get_stateless_obj()->get_port_count()) {
         std::stringstream ss;
@@ -643,8 +650,30 @@ TrexRpcCmdUpdateTraffic::_run(const Json::Value &params, Json::Value &result) {
 
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
+    /* multiplier */
+    const Json::Value &mul = parse_object(params, "mul", result);
+
+    std::string mul_type = parse_string(mul, "type", result);
+    double max_rate = parse_double(mul, "max", result);
+
+    double m = 0;
+
+    /* dispatch according to type of multiplier  */
+    if (mul_type == "raw") {
+        m = max_rate;
+
+    } else if (mul_type == "max_bps") {
+        m = port->calculate_m_from_bps(max_rate);
+
+    } else if (mul_type == "max_pps") {
+        m = port->calculate_m_from_pps(max_rate);
+
+    } else {
+        generate_parse_err(result, "multiplier type can be either 'raw', 'max_bps' or 'max_pps'");
+    }
+
     try {
-        port->update_traffic(mul);
+        port->update_traffic(m);
     } catch (const TrexRpcException &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 2fd948fd..8e18a5bf 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -144,32 +144,29 @@ TrexStatelessPort::start_traffic(double mul, double duration) {
     get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STARTED, data);
 }
 
-void
-TrexStatelessPort::start_traffic_max_bps(double max_bps, double duration) {
+
+double
+TrexStatelessPort::calculate_m_from_bps(double max_bps) {
     /* fetch all the streams from the table */
     vector<TrexStream *> streams;
     get_object_list(streams);
 
     TrexStreamsGraph graph;
     const TrexStreamsGraphObj &obj = graph.generate(streams);
-    double m = (max_bps / obj.get_max_bps());
 
-    /* call the main function */
-    start_traffic(m, duration);
+    return (max_bps / obj.get_max_bps());
 }
 
-void
-TrexStatelessPort::start_traffic_max_pps(double max_pps, double duration) {
+double
+TrexStatelessPort::calculate_m_from_pps(double max_pps) {
     /* fetch all the streams from the table */
     vector<TrexStream *> streams;
     get_object_list(streams);
 
     TrexStreamsGraph graph;
     const TrexStreamsGraphObj &obj = graph.generate(streams);
-    double m = (max_pps / obj.get_max_pps());
 
-    /* call the main function */
-    start_traffic(m, duration);
+    return (max_pps / obj.get_max_pps());
 }
 
 /**
@@ -239,16 +236,14 @@ TrexStatelessPort::resume_traffic(void) {
 void
 TrexStatelessPort::update_traffic(double mul) {
 
-    verify_state(PORT_STATE_STREAMS | PORT_STATE_TX | PORT_STATE_PAUSE);
+    verify_state(PORT_STATE_TX | PORT_STATE_PAUSE);
 
-    #if 0
     /* generate a message to all the relevant DP cores to start transmitting */
-    TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpStop(m_port_id);
+    double per_core_mul = mul / m_cores_id_list.size();
+    TrexStatelessCpToDpMsgBase *update_msg = new TrexStatelessDpUpdate(m_port_id, per_core_mul);
 
-    send_message_to_dp(stop_msg);
+    send_message_to_dp(update_msg);
 
-    m_port_state = PORT_STATE_UP_IDLE;
-    #endif
 }
 
 std::string 
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 20acd927..b061a414 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -77,8 +77,18 @@ public:
      * throws TrexException in case of an error
      */
     void start_traffic(double mul, double duration = -1);
-    void start_traffic_max_bps(double max_bps, double duration = -1);
-    void start_traffic_max_pps(double max_pps, double duration = -1);
+
+    /**
+     * given a BPS rate calculate ther correct M for this port
+     * 
+     */
+    double calculate_m_from_bps(double max_bps);
+
+    /**
+     * given a PPS rate calculate the correct M for this port
+     * 
+     */
+    double calculate_m_from_pps(double max_pps);
 
     /**
      * stop traffic
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 4e429c8f..dd4937cd 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -139,6 +139,20 @@ bool TrexStatelessDpPerPort::resume_traffic(uint8_t port_id){
     return (true);
 }
 
+bool TrexStatelessDpPerPort::update_traffic(uint8_t port_id, double mul) {
+
+    assert( (m_state == TrexStatelessDpPerPort::ppSTATE_TRANSMITTING || 
+            (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);
+
+        node->set_multiplier(mul);
+    }
+
+    return (true);
+}
 
 bool TrexStatelessDpPerPort::pause_traffic(uint8_t port_id){
 
@@ -402,9 +416,9 @@ TrexStatelessDpCore::add_port_duration(double duration,
 
 
 void
-TrexStatelessDpCore::add_cont_stream(TrexStatelessDpPerPort * lp_port,
-                                     TrexStream * stream,
-                                     TrexStreamsCompiledObj *comp) {
+TrexStatelessDpCore::add_stream(TrexStatelessDpPerPort * lp_port,
+                                TrexStream * stream,
+                                TrexStreamsCompiledObj *comp) {
 
     CGenNodeStateless *node = m_core->create_node_sl();
 
@@ -439,8 +453,8 @@ TrexStatelessDpCore::add_cont_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_base_pps = stream->get_pps();
+    node->set_multiplier(comp->get_multiplier());
 
     /* stateless specific fields */
     switch ( stream->m_type ) {
@@ -520,7 +534,7 @@ TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj,
     for (auto single_stream : obj->get_objects()) {
         /* all commands should be for the same port */
         assert(obj->get_port_id() == single_stream.m_stream->m_port_id);
-        add_cont_stream(lp_port,single_stream.m_stream,obj);
+        add_stream(lp_port,single_stream.m_stream,obj);
     }
 
     uint32_t nodes = lp_port->m_active_nodes.size();
@@ -582,6 +596,14 @@ TrexStatelessDpCore::pause_traffic(uint8_t port_id){
     lp_port->pause_traffic(port_id);
 }
 
+void 
+TrexStatelessDpCore::update_traffic(uint8_t port_id, double mul) {
+
+    TrexStatelessDpPerPort * lp_port = get_port_db(port_id);
+
+    lp_port->update_traffic(port_id, mul);
+}
+
 
 void
 TrexStatelessDpCore::stop_traffic(uint8_t port_id,
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index eda1ae59..563159b2 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -68,6 +68,8 @@ public:
 
     bool resume_traffic(uint8_t port_id);
 
+    bool update_traffic(uint8_t port_id, double mul);
+
     bool stop_traffic(uint8_t port_id,
                       bool stop_on_id, 
                       int event_id);
@@ -159,6 +161,14 @@ public:
 
 
     /**
+     * update current traffic rate
+     * 
+     * @author imarom (25-Nov-15)
+     * 
+     */
+    void update_traffic(uint8_t port_id, double mul);
+
+    /**
      * 
      * stop all traffic for this core
      * 
@@ -250,9 +260,9 @@ private:
 
     void add_global_duration(double duration);
 
-    void add_cont_stream(TrexStatelessDpPerPort * lp_port,
-                         TrexStream * stream,
-                         TrexStreamsCompiledObj *comp);
+    void add_stream(TrexStatelessDpPerPort * lp_port,
+                    TrexStream * stream,
+                    TrexStreamsCompiledObj *comp);
 
     uint8_t              m_thread_id;
     uint8_t              m_local_port_offset;
diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h
index ccf99eaa..5997376f 100644
--- a/src/stateless/dp/trex_stream_node.h
+++ b/src/stateless/dp/trex_stream_node.h
@@ -86,8 +86,9 @@ private:
     TrexStream *        m_ref_stream_info; /* the stream info */
     CGenNodeStateless  * m_next_stream;
 
+    double              m_base_pps;
     /* pad to match the size of CGenNode */
-    uint8_t             m_pad_end[56];
+    uint8_t             m_pad_end[48];
 
 
 
@@ -99,6 +100,15 @@ public:
     }
 
 
+    /**
+     * calculate the time offset based 
+     * on the PPS and multiplier 
+     * 
+     */
+    void set_multiplier(double mul) {
+        m_next_time_offset =  1.0 / (m_base_pps * mul) ;
+    }
+
     /* we restart the stream, schedule it using stream isg */
     inline void update_refresh_time(double cur_time){
         m_time = cur_time + usec_to_sec(m_ref_stream_info->m_isg_usec);
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index ec8b7839..3210f29a 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -163,6 +163,22 @@ TrexStatelessDpCanQuit::clone(){
     return new_msg;
 }
 
+/*************************
+  update traffic message
+ ************************/ 
+bool
+TrexStatelessDpUpdate::handle(TrexStatelessDpCore *dp_core) {
+    dp_core->update_traffic(m_port_id, m_mul);
+
+    return true;
+}
+
+TrexStatelessCpToDpMsgBase *
+TrexStatelessDpUpdate::clone() {
+    TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpUpdate(m_port_id, m_mul);
+
+    return new_msg;
+}
 
 /************************* messages from DP to CP **********************/
 bool
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index 6bd0dbe3..7390be60 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -225,6 +225,26 @@ public:
 };
 
 
+/**
+ * update message
+ */
+class TrexStatelessDpUpdate : public TrexStatelessCpToDpMsgBase {
+public:
+
+    TrexStatelessDpUpdate(uint8_t port_id, double mul)  {
+        m_port_id = port_id;
+        m_mul = mul;
+    }
+
+    virtual bool handle(TrexStatelessDpCore *dp_core);
+
+    virtual TrexStatelessCpToDpMsgBase * clone();
+
+private:
+    uint8_t  m_port_id;
+    double   m_mul;
+};
+
 
 /************************* messages from DP to CP **********************/