Merge branch 'master' into dan_stateless

# Conflicts (solved):
#	scripts/automation/trex_control_plane/client_utils/parsing_opts.py

17 files changed, 817 insertions, 41 deletions

diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index 8640e7db..5b298023 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -1668,5 +1668,180 @@ TEST_F(basic_stl, dp_stop_event) {
      
 }
 
+TEST_F(basic_stl, graph_generator1) {
+    std::vector<TrexStream *> streams;
+    TrexStreamsGraph graph;
+    TrexStream *stream;
+
+     /* stream 1 */
+     stream = new TrexStream(TrexStream::stSINGLE_BURST, 0, 1);
+     stream->m_enabled = true;
+     stream->m_self_start = true;
+
+     stream->m_isg_usec = 42;
+     stream->set_pps(10);
+     stream->set_single_burst(43281);
+     stream->m_pkt.len = 512;
+
+     stream->m_next_stream_id = 2;
+     
+
+     streams.push_back(stream);
+
+     /* stream 2 */
+     stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 2);
+     stream->m_enabled = true;
+     stream->m_self_start = false;
+
+     stream->set_pps(20);
+     stream->set_multi_burst(4918, 13, 7);
+     stream->m_next_stream_id = -1;
+     stream->m_pkt.len = 64;
+
+     streams.push_back(stream);
+
+     /* stream 3 */
+     stream = new TrexStream(TrexStream::stCONTINUOUS, 0, 3);
+     stream->m_enabled = true;
+     stream->m_self_start = true;
+
+     stream->m_isg_usec = 50;
+     stream->set_pps(30);
+     stream->m_next_stream_id = -1;
+     stream->m_pkt.len = 1512;
+
+     streams.push_back(stream);
+
+
+     const TrexStreamsGraphObj &obj = graph.generate(streams);
+     EXPECT_EQ(obj.get_max_bps(), 405120);
+     EXPECT_EQ(obj.get_max_pps(), 50);
+
+     for (auto stream : streams) {
+         delete stream;
+     }
+}   
+
+
+TEST_F(basic_stl, graph_generator2) {
+    std::vector<TrexStream *> streams;
+    TrexStreamsGraph graph;
+    TrexStream *stream;
+
+    /* add some multi burst streams */
+    stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 1);
+    stream->m_enabled = true;
+    stream->m_self_start = true;
+    
+
+    stream->set_pps(1000);
+
+    /* a burst of 2000 packets with a delay of 1 second */
+    stream->m_isg_usec = 0;
+    stream->set_multi_burst(1000, 500, 1000 * 1000);
+    stream->m_pkt.len = 64;
+
+    stream->m_next_stream_id = -1;
+
+    streams.push_back(stream);
+
+    /* another multi burst stream but with a shorter burst ( less 2 ms ) and
+       higher ibg (2 ms) , one milli for each side
+     */
+    stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 2);
+    stream->m_enabled = true;
+    stream->m_self_start = true;
+
+    stream->set_pps(1000);
+    stream->m_isg_usec = 1000 * 1000 + 1000;
+    stream->set_multi_burst(1000 - 2, 1000, 1000 * 1000 + 2000);
+    stream->m_pkt.len = 128;
+
+    stream->m_next_stream_id = -1;
+
+    streams.push_back(stream);
+
+    const TrexStreamsGraphObj &obj = graph.generate(streams);
+    EXPECT_EQ(obj.get_max_pps(), 1000.0);
+
+    EXPECT_EQ(obj.get_max_bps(), (1000 * (128 + 4) * 8));
+    
+
+    for (auto stream : streams) {
+        delete stream;
+    }
+}
+
+/* stress test */
+#if 0 
+TEST_F(basic_stl, graph_generator2) {
+    std::vector<TrexStream *> streams;
+    TrexStreamsGraph graph;
+    TrexStream *stream;
+
+     /* add some multi burst streams */
+     stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 1);
+     stream->m_enabled = true;
+     stream->m_self_start = true;
+     stream->m_isg_usec = 100;
+
+     stream->set_pps(20);
+     stream->set_multi_burst(4918, 321312, 15);
+     stream->m_next_stream_id = -1;
+     stream->m_pkt.len = 64;
+
+     streams.push_back(stream);
+
+     stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 2);
+     stream->m_enabled = true;
+     stream->m_self_start = true;
+     stream->m_isg_usec = 59281;
+
+     stream->set_pps(30);
+     stream->set_multi_burst(4918, 51040, 27);
+     stream->m_next_stream_id = -1;
+     stream->m_pkt.len = 64;
+
+     streams.push_back(stream);
+
+     stream = new TrexStream(TrexStream::stMULTI_BURST, 0, 3);
+     stream->m_enabled = true;
+     stream->m_self_start = true;
+     stream->m_isg_usec = 59281492;
+
+     stream->set_pps(40);
+     stream->set_multi_burst(4918, 412312, 2917);
+     stream->m_next_stream_id = -1;
+     stream->m_pkt.len = 64;
+
+     streams.push_back(stream);
+
+
+     /* stream 3 */
+     stream = new TrexStream(TrexStream::stCONTINUOUS, 0, 4);
+     stream->m_enabled = true;
+     stream->m_self_start = true;
+
+     stream->m_isg_usec = 50;
+     stream->set_pps(30);
+     stream->m_next_stream_id = -1;
+     stream->m_pkt.len = 1512;
+
+     streams.push_back(stream);
+
+
+     const TrexStreamsGraphObj &obj = graph.generate(streams);
+     printf("event_count is: %lu, max BPS: %f, max PPS: %f\n", obj.get_events().size(), obj.get_max_bps(), obj.get_max_pps());
+
+//     for (const TrexStreamsGraphObj::rate_event_st &ev : obj.get_events()) {
+//         printf("time: %f, diff bps: %f, diff pps: %f\n", ev.time, ev.diff_bps, ev.diff_pps);
+//     }
+
+     for (auto stream : streams) {
+         delete stream;
+     }
+}   
+
+#endif
 
 /********************************************* Itay Tests End *************************************/
diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h
index 5890a965..343b8004 100644
--- a/src/internal_api/trex_platform_api.h
+++ b/src/internal_api/trex_platform_api.h
@@ -24,6 +24,7 @@ limitations under the License.
 
 #include <stdint.h>
 #include <vector>
+#include <string>
 
 /**
  * Global stats
@@ -97,10 +98,20 @@ public:
 
 class TrexPlatformApi {
 public:
+
+    enum driver_speed_e {
+        SPEED_INVALID,
+        SPEED_1G,
+        SPEED_10G,
+        SPEED_40G,
+    };
+
     virtual void port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const = 0;
     virtual void get_global_stats(TrexPlatformGlobalStats &stats) const = 0;
     virtual void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const = 0;
+    virtual void get_interface_info(uint8_t interface_id, std::string &driver_name, driver_speed_e &speed) const = 0;
     virtual uint8_t get_dp_core_count() const = 0;
+    
     virtual ~TrexPlatformApi() {}
 };
 
@@ -115,7 +126,9 @@ public:
     void port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const;
     void get_global_stats(TrexPlatformGlobalStats &stats) const;
     void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const;
+    void get_interface_info(uint8_t interface_id, std::string &driver_name, driver_speed_e &speed) const;
     uint8_t get_dp_core_count() const;
+    
 };
 
 /**
@@ -128,6 +141,11 @@ public:
     void port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const {}
     void get_global_stats(TrexPlatformGlobalStats &stats) const;
     void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const;
+    void get_interface_info(uint8_t interface_id, std::string &driver_name, driver_speed_e &speed) const {
+        driver_name = "MOCK";
+        speed = SPEED_INVALID;
+    }
+
     uint8_t get_dp_core_count() const;
 };
 
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index b1c9ed12..3e5418b9 100755
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -127,6 +127,9 @@ class CPhyEthIFStats ;
 
 class CTRexExtendedDriverBase {
 public:
+
+    virtual TrexPlatformApi::driver_speed_e get_driver_speed() = 0;
+
     virtual int get_min_sample_rate(void)=0;
     virtual void update_configuration(port_cfg_t * cfg)=0;
     virtual void update_global_config_fdir(port_cfg_t * cfg)=0;
@@ -153,6 +156,10 @@ public:
     CTRexExtendedDriverBase1G(){
     }
 
+    TrexPlatformApi::driver_speed_e get_driver_speed() {
+        return TrexPlatformApi::SPEED_1G;
+    }
+
     static CTRexExtendedDriverBase * create(){
         return ( new CTRexExtendedDriverBase1G() );
     }
@@ -191,6 +198,10 @@ public:
         CGlobalInfo::m_options.preview.set_vm_one_queue_enable(true);
     }
 
+    TrexPlatformApi::driver_speed_e get_driver_speed() {
+        return TrexPlatformApi::SPEED_1G;
+    }
+
     static CTRexExtendedDriverBase * create(){
         return ( new CTRexExtendedDriverBase1GVm() );
     }
@@ -229,6 +240,11 @@ class CTRexExtendedDriverBase10G : public CTRexExtendedDriverBase {
 public:
     CTRexExtendedDriverBase10G(){
     }
+
+    TrexPlatformApi::driver_speed_e get_driver_speed() {
+        return TrexPlatformApi::SPEED_10G;
+    }
+
     static CTRexExtendedDriverBase * create(){
         return ( new CTRexExtendedDriverBase10G() );
     }
@@ -261,6 +277,10 @@ public:
     CTRexExtendedDriverBase40G(){
     }
 
+    TrexPlatformApi::driver_speed_e get_driver_speed() {
+        return TrexPlatformApi::SPEED_40G;
+    }
+
     static CTRexExtendedDriverBase * create(){
         return ( new CTRexExtendedDriverBase40G() );
     }
@@ -303,6 +323,11 @@ public:
 
 class CTRexExtendedDriverDb {
 public:
+
+   const std::string & get_driver_name() {
+       return m_driver_name;
+   }
+
    bool is_driver_exists(std::string name);
 
 
@@ -3249,6 +3274,7 @@ void CGlobalTRex::try_stop_all_dp(){
         delay(100);
     }
     if ( all_core_finished ){
+        m_zmq_publisher.publish_event(TrexPublisher::EVENT_SERVER_STOPPED);
         printf(" All cores stopped !! \n");
     }else{
         printf(" ERROR one of the DP core is stucked !\n");
@@ -5275,3 +5301,11 @@ TrexDpdkPlatformApi::port_id_to_cores(uint8_t port_id, std::vector<std::pair<uin
     }
 }
 
+void
+TrexDpdkPlatformApi::get_interface_info(uint8_t interface_id,
+                                        std::string &driver_name,
+                                        driver_speed_e &speed) const {
+
+    driver_name = CTRexExtendedDriverDb::Ins()->get_driver_name();
+    speed = CTRexExtendedDriverDb::Ins()->get_drv()->get_driver_speed();
+}
diff --git a/src/publisher/trex_publisher.h b/src/publisher/trex_publisher.h
index 82603fda..8d1be064 100644
--- a/src/publisher/trex_publisher.h
+++ b/src/publisher/trex_publisher.h
@@ -39,10 +39,13 @@ public:
     void publish_json(const std::string &s);
 
     enum event_type_e {
-        EVENT_PORT_STOPPED = 0
+        EVENT_PORT_STARTED       = 0,
+        EVENT_PORT_STOPPED       = 1,
+        EVENT_SERVER_STOPPED     = 2,
+        EVENT_PORT_FINISHED_TX   = 3,
     };
 
-    void publish_event(event_type_e type, const Json::Value &data);
+    void publish_event(event_type_e type, const Json::Value &data = Json::nullValue);
 
 private:
     void show_zmq_last_error(const std::string &err);
diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index 1a7132ff..9570aae7 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -25,6 +25,8 @@ limitations under the License.
 #include <trex_stateless_port.h>
 #include <trex_rpc_cmds_table.h>
 
+#include <internal_api/trex_platform_api.h>
+
 #include <fstream>
 #include <iostream>
 #include <unistd.h>
@@ -167,14 +169,34 @@ TrexRpcCmdGetSysInfo::_run(const Json::Value &params, Json::Value &result) {
 
     for (int i = 0; i < main->get_port_count(); i++) {
         string driver;
-        string speed;
+        TrexPlatformApi::driver_speed_e speed;
 
         TrexStatelessPort *port = main->get_port_by_id(i);
         port->get_properties(driver, speed);
 
         section["ports"][i]["index"]   = i;
+
         section["ports"][i]["driver"]  = driver;
-        section["ports"][i]["speed"]   = speed;
+
+        switch (speed) {
+        case TrexPlatformApi::SPEED_1G:
+            section["ports"][i]["speed"]   = 1;
+            break;
+
+        case TrexPlatformApi::SPEED_10G:
+            section["ports"][i]["speed"]   = 10;
+            break;
+
+        case TrexPlatformApi::SPEED_40G:
+            section["ports"][i]["speed"]   = 40;
+            break;
+
+        default:
+            /* unknown value */
+            section["ports"][i]["speed"]   = 0;
+            break;
+        }
+
 
         section["ports"][i]["owner"] = port->get_owner();
 
diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index cdd13ed6..96224d4e 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -463,7 +463,6 @@ trex_rpc_cmd_rc_e
 TrexRpcCmdStartTraffic::_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);
     double duration  = parse_double(params, "duration", result);
 
     if (port_id >= get_stateless_obj()->get_port_count()) {
@@ -474,8 +473,33 @@ TrexRpcCmdStartTraffic::_run(const Json::Value &params, Json::Value &result) {
 
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
+
+    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->start_traffic(mul, duration);
+        port->start_traffic(m, duration);
+
     } catch (const TrexRpcException &ex) {
         generate_execute_err(result, ex.what());
     }
@@ -617,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;
@@ -627,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 40392e68..8e18a5bf 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -56,10 +56,12 @@ TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api
     std::vector<std::pair<uint8_t, uint8_t>> core_pair_list;
 
     m_port_id = port_id;
-
     m_port_state = PORT_STATE_IDLE;
     clear_owner();
 
+    /* get the platform specific data */
+    api->get_interface_info(port_id, m_driver_name, m_speed);
+
     /* get the DP cores belonging to this port */
     api->port_id_to_cores(m_port_id, core_pair_list);
 
@@ -111,9 +113,12 @@ TrexStatelessPort::start_traffic(double mul, double duration) {
     vector<TrexStream *> streams;
     get_object_list(streams);
 
+    /* split it per core */
+    double per_core_mul = mul / m_cores_id_list.size();
+
     /* compiler it */
     TrexStreamsCompiler compiler;
-    TrexStreamsCompiledObj *compiled_obj = new TrexStreamsCompiledObj(m_port_id, mul);
+    TrexStreamsCompiledObj *compiled_obj = new TrexStreamsCompiledObj(m_port_id, per_core_mul);
 
     bool rc = compiler.compile(streams, *compiled_obj);
     if (!rc) {
@@ -134,6 +139,34 @@ TrexStatelessPort::start_traffic(double mul, double duration) {
 
     send_message_to_dp(start_msg);
     
+    Json::Value data;
+    data["port_id"] = m_port_id;
+    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STARTED, data);
+}
+
+
+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);
+
+    return (max_bps / obj.get_max_bps());
+}
+
+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);
+
+    return (max_pps / obj.get_max_pps());
 }
 
 /**
@@ -161,6 +194,10 @@ TrexStatelessPort::stop_traffic(void) {
 
     change_state(PORT_STATE_STREAMS);
     
+    Json::Value data;
+    data["port_id"] = m_port_id;
+    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STOPPED, data);
+
 }
 
 void
@@ -199,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 
@@ -235,11 +270,10 @@ TrexStatelessPort::get_state_as_string() const {
 }
 
 void
-TrexStatelessPort::get_properties(string &driver, string &speed) {
+TrexStatelessPort::get_properties(std::string &driver, TrexPlatformApi::driver_speed_e &speed) {
 
-    /* take this from DPDK */
-    driver = "e1000";
-    speed  = "1 Gbps";
+    driver = m_driver_name;
+    speed  = m_speed;
 }
 
 bool
@@ -334,7 +368,7 @@ TrexStatelessPort::on_dp_event_occured(TrexDpPortEvent::event_e event_type) {
         /* send a ZMQ event */
 
         data["port_id"] = m_port_id;
-        get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STOPPED, data);
+        get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_FINISHED_TX, data);
         break;
 
     default:
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 006ec97c..b061a414 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -23,8 +23,8 @@ limitations under the License.
 
 #include <trex_stream.h>
 #include <trex_dp_port_events.h>
+#include <internal_api/trex_platform_api.h>
 
-class TrexPlatformApi;
 class TrexStatelessCpToDpMsgBase;
 
 /**
@@ -79,6 +79,18 @@ public:
     void start_traffic(double mul, 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
      * throws TrexException in case of an error
      */
@@ -124,7 +136,7 @@ public:
      * @param driver 
      * @param speed 
      */
-    void get_properties(std::string &driver, std::string &speed);
+    void get_properties(std::string &driver, TrexPlatformApi::driver_speed_e &speed);
 
 
     /**
@@ -258,6 +270,9 @@ private:
     port_state_e       m_port_state;
     std::string        m_owner;
     std::string        m_owner_handler;
+    std::string        m_driver_name;
+
+    TrexPlatformApi::driver_speed_e m_speed;
 
     /* holds the DP cores associated with this port */
     std::vector<int> m_cores_id_list;
diff --git a/src/stateless/cp/trex_stream.cpp b/src/stateless/cp/trex_stream.cpp
index 5203b2a2..cad603e2 100644
--- a/src/stateless/cp/trex_stream.cpp
+++ b/src/stateless/cp/trex_stream.cpp
@@ -21,6 +21,7 @@ limitations under the License.
 #include <trex_stream.h>
 #include <cstddef>
 #include <string.h>
+#include <assert.h>
 
 /**************************************
  * stream
@@ -129,6 +130,7 @@ TrexStream::get_stream_json() {
     return m_stream_json;
 }
 
+
 /**************************************
  * stream table
  *************************************/
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index 0634829e..3e48d7e4 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -90,7 +90,7 @@ public:
         m_next_stream_id = next_stream_id;
     }
 
-    double get_pps() {
+    double get_pps() const {
         return m_pps;
     }
 
@@ -150,6 +150,16 @@ public:
             return (dp);
     }
 
+
+    double get_burst_length_usec() const {
+        return ( (m_burst_total_pkts / m_pps) * 1000 * 1000);
+    }
+
+    double get_bps() const {
+        /* packet length + 4 CRC bytes to bits and multiplied by PPS */
+        return (m_pps * (m_pkt.len + 4) * 8);
+    }
+
     void Dump(FILE *fd);
 public:
     /* basic */
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index 302863ae..b28989be 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -34,19 +34,25 @@ limitations under the License.
  */
 class GraphNode {
 public:
-    GraphNode(TrexStream *stream, GraphNode *next) : m_stream(stream), m_next(next) {
-        marked   = false;
+    GraphNode(const TrexStream *stream, GraphNode *next) : m_stream(stream), m_next(next) {
+        m_marked   = false;
         m_compressed_stream_id=-1;
+
     }
 
     uint32_t get_stream_id() const {
         return m_stream->m_stream_id;
     }
 
+    uint32_t get_next_stream_id() const {
+        return m_stream->m_next_stream_id;
+
+    }
+
     const TrexStream *m_stream;
     GraphNode *m_next;
     std::vector<const GraphNode *> m_parents;
-    bool marked;
+    bool m_marked;
     int m_compressed_stream_id;
 };
 
@@ -97,13 +103,13 @@ public:
 
     void clear_marks() {
         for (auto node : m_nodes) {
-            node.second->marked = false;
+            node.second->m_marked = false;
         }
     }
 
     void get_unmarked(std::vector <GraphNode *> &unmarked) {
         for (auto node : m_nodes) {
-            if (!node.second->marked) {
+            if (!node.second->m_marked) {
                 unmarked.push_back(node.second);
             }
         }
@@ -132,6 +138,7 @@ private:
     GraphNode m_dead_end;
 };
 
+
 /**************************************
  * stream compiled object
  *************************************/
@@ -317,11 +324,11 @@ TrexStreamsCompiler::check_for_unreachable_streams(GraphNodeMap *nodes) {
         /* pull one */
         GraphNode *node = next_nodes.back();
         next_nodes.pop_back();
-        if (node->marked) {
+        if (node->m_marked) {
             continue;
         }
 
-        node->marked = true;
+        node->m_marked = true;
 
         if (node->m_next != NULL) {
             next_nodes.push_back(node->m_next);
@@ -427,9 +434,254 @@ TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams,
                                 my_stream_id,
                                 my_next_stream_id
                                 );
+        
     }
     obj.m_all_continues =all_continues;
     return true;
 }
 
+/**************************************
+ * streams graph
+ *************************************/
+
+/**
+ * for each stream we create the right rate events (up/down)
+ * 
+ * @author imarom (24-Nov-15)
+ * 
+ * @param offset_usec 
+ * @param stream 
+ */
+void
+TrexStreamsGraph::add_rate_events_for_stream(double &offset_usec, const TrexStream *stream) {
+
+    switch (stream->get_type()) {
+   
+    case TrexStream::stCONTINUOUS:
+        add_rate_events_for_stream_cont(offset_usec, stream);
+        return;
+        
+    case TrexStream::stSINGLE_BURST:
+        add_rate_events_for_stream_single_burst(offset_usec, stream);
+        return;
+
+    case TrexStream::stMULTI_BURST:
+        add_rate_events_for_stream_multi_burst(offset_usec, stream);
+        return;
+    }
+}
+
+/**
+ * continous stream
+ * 
+ */
+void
+TrexStreamsGraph::add_rate_events_for_stream_cont(double &offset_usec, const TrexStream *stream) {
+
+    TrexStreamsGraphObj::rate_event_st start_event;
+
+    /* for debug purposes */
+    start_event.stream_id = stream->m_stream_id;
+
+    start_event.time = offset_usec + stream->m_isg_usec;
+    start_event.diff_pps = stream->get_pps();
+    start_event.diff_bps = stream->get_bps();
+    m_graph_obj.add_rate_event(start_event);
+
+    /* no more events after this stream */
+    offset_usec = -1;
+}
+
+/**
+ * single burst stream
+ * 
+ */
+void
+TrexStreamsGraph::add_rate_events_for_stream_single_burst(double &offset_usec, const TrexStream *stream) {
+    TrexStreamsGraphObj::rate_event_st start_event;
+    TrexStreamsGraphObj::rate_event_st stop_event;
+
+
+    /* for debug purposes */
+    start_event.stream_id = stream->m_stream_id;
+    stop_event.stream_id   = stream->m_stream_id;
+
+     /* start event */
+    start_event.time = offset_usec + stream->m_isg_usec;
+    start_event.diff_pps = stream->get_pps();
+    start_event.diff_bps = stream->get_bps();
+    m_graph_obj.add_rate_event(start_event);
+
+    /* stop event */
+    stop_event.time = start_event.time + stream->get_burst_length_usec();
+    stop_event.diff_pps = -(start_event.diff_pps);
+    stop_event.diff_bps = -(start_event.diff_bps);
+    m_graph_obj.add_rate_event(stop_event);
+
+    /* next stream starts from here */
+    offset_usec = stop_event.time;
+
+}
+
+/**
+ * multi burst stream
+ * 
+ */
+void
+TrexStreamsGraph::add_rate_events_for_stream_multi_burst(double &offset_usec, const TrexStream *stream) {
+    TrexStreamsGraphObj::rate_event_st start_event;
+    TrexStreamsGraphObj::rate_event_st stop_event;
+
+    /* first the delay is the inter stream gap */
+    double delay = stream->m_isg_usec;
+
+    /* for debug purposes */
+    
+    start_event.diff_pps   = stream->get_pps();
+    start_event.diff_bps   = stream->get_bps();
+    start_event.stream_id  = stream->m_stream_id;
+
+    stop_event.diff_pps    = -(start_event.diff_pps);
+    stop_event.diff_bps    = -(start_event.diff_bps);
+    stop_event.stream_id   = stream->m_stream_id;
+
+    /* for each burst create up/down events */
+    for (int i = 0; i < stream->m_num_bursts; i++) {
+
+        start_event.time = offset_usec + delay;
+        m_graph_obj.add_rate_event(start_event);
+
+        stop_event.time = start_event.time + stream->get_burst_length_usec();
+        m_graph_obj.add_rate_event(stop_event);
+
+        /* after the first burst, the delay is inter burst gap */
+        delay = stream->m_ibg_usec;
+
+        offset_usec = stop_event.time;
+    }
+}
+
+/**
+ * for a single root we can until done or a loop detected
+ * 
+ * @author imarom (24-Nov-15)
+ * 
+ * @param root_stream_id 
+ */
+void
+TrexStreamsGraph::generate_graph_for_one_root(uint32_t root_stream_id) {
+
+    std::unordered_map<uint32_t, bool> loop_hash;
+    std::stringstream ss;
+    
+    uint32_t stream_id = root_stream_id;
+    double offset = 0;
+
+    while (true) {
+        const TrexStream *stream;
+        
+        /* fetch the stream from the hash - if it is not present, report an error */
+        try {
+            stream = m_streams_hash.at(stream_id);
+        } catch (const std::out_of_range &e) {
+            ss << "stream id " << stream_id << " does not exists";
+            throw TrexException(ss.str());
+        }
+
+        /* add the node to the hash for loop detection */
+        loop_hash[stream_id] = true;
+
+        /* create the right rate events for the stream */
+        add_rate_events_for_stream(offset, stream);
+
+        /* do we have a next stream ? */
+        if (stream->m_next_stream_id == -1) {
+            break;
+        }
+
+        /* loop detection */
+        auto search = loop_hash.find(stream->m_next_stream_id);
+        if (search != loop_hash.end()) {
+            break;
+        }
+
+        /* handle the next one */
+        stream_id = stream->m_next_stream_id;
+    }
+}
+
+/**
+ * for a vector of streams generate a graph of BW 
+ * see graph object for more details 
+ * 
+ */
+const TrexStreamsGraphObj &
+TrexStreamsGraph::generate(const std::vector<TrexStream *> &streams) {
+    std::vector <uint32_t> root_streams;
+
+    /* before anything we create a hash streams ID
+       and grab the root nodes
+     */
+    for (TrexStream *stream : streams) {
+
+        /* skip non enabled streams */
+        if (!stream->m_enabled) {
+            continue;
+        }
+
+        /* for fast search we populate all the streams in a hash */        
+        m_streams_hash[stream->m_stream_id] = stream;
+
+        /* hold all the self start nodes in a vector */
+        if (stream->m_self_start) {
+            root_streams.push_back(stream->m_stream_id);
+        }
+    }
+
+    /* for each node - scan until done or loop */
+    for (uint32_t root_id : root_streams) {
+        generate_graph_for_one_root(root_id);
+    }
+
+
+    m_graph_obj.generate();
+
+    return m_graph_obj;
+}
+
+/**************************************
+ * streams graph object
+ *************************************/
+void
+TrexStreamsGraphObj::find_max_rate() {
+    double max_rate_pps = 0;
+    double current_rate_pps = 0;
+
+    double max_rate_bps = 0;
+    double current_rate_bps = 0;
+
+    /* now we simply walk the list and hold the max */
+    for (auto &ev : m_rate_events) {
+
+        current_rate_pps += ev.diff_pps;
+        current_rate_bps += ev.diff_bps;
+
+        max_rate_pps = std::max(max_rate_pps, current_rate_pps);
+        max_rate_bps = std::max(max_rate_bps, current_rate_bps);
+    }
+
+    m_max_pps = max_rate_pps;
+    m_max_bps = max_rate_bps;
+}
+
+static 
+bool event_compare (const TrexStreamsGraphObj::rate_event_st &first, const TrexStreamsGraphObj::rate_event_st &second) {
+    return (first.time < second.time);
+}
+
+void
+TrexStreamsGraphObj::generate() {
+    m_rate_events.sort(event_compare);
+    find_max_rate();
+}
 
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index 17ca3c74..70a31c5e 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -23,7 +23,9 @@ limitations under the License.
 
 #include <stdint.h>
 #include <vector>
+#include <list>
 #include <string>
+#include <unordered_map>
 
 class TrexStreamsCompiler;
 class TrexStream;
@@ -114,6 +116,91 @@ private:
     void err(const std::string &err);
 
     std::vector<std::string> m_warnings;
+    
+};
+
+class TrexStreamsGraph;
+
+/**************************************
+ * streams graph object 
+ *  
+ * holds the step graph for bandwidth 
+ *************************************/
+class TrexStreamsGraphObj {
+    friend class TrexStreamsGraph;
+
+public:
+
+    /**
+     * rate event is defined by those: 
+     * time - the time of the event on the timeline 
+     * diff - what is the nature of the change ? 
+     * 
+     * @author imarom (23-Nov-15)
+     */
+    struct rate_event_st {
+        double time;
+        double diff_pps;
+        double diff_bps;
+        uint32_t stream_id;
+    };
+
+    double get_max_pps() const {
+        return m_max_pps;
+    }
+
+    double get_max_bps() const {
+        return m_max_bps;
+    }
+
+    const std::list<rate_event_st> & get_events() const {
+        return m_rate_events;
+    }
+
+private:
+
+    void add_rate_event(const rate_event_st &ev) {
+        m_rate_events.push_back(ev);
+    }
+
+    void generate();
+    void find_max_rate();
+
+    double m_max_pps;
+    double m_max_bps;
+
+    /* list of rate events */
+    std::list<rate_event_st> m_rate_events;
+};
+
+/**
+ * graph creator 
+ * 
+ * @author imarom (23-Nov-15)
+ */
+class TrexStreamsGraph {
+public:
+
+    /**
+     * generate a sequence graph for streams
+     * 
+     */
+    const TrexStreamsGraphObj & generate(const std::vector<TrexStream *> &streams);
+
+private:
+
+    void generate_graph_for_one_root(uint32_t root_stream_id);
+
+    void add_rate_events_for_stream(double &offset, const TrexStream *stream);
+    void add_rate_events_for_stream_cont(double &offset_usec, const TrexStream *stream);
+    void add_rate_events_for_stream_single_burst(double &offset_usec, const TrexStream *stream);
+    void add_rate_events_for_stream_multi_burst(double &offset_usec, const TrexStream *stream);
+
+    /* for fast processing of streams */
+    std::unordered_map<uint32_t, const TrexStream *> m_streams_hash;
+
+    /* main object to hold the graph - returned to the user */
+    TrexStreamsGraphObj m_graph_obj;
 };
 
 #endif /* __TREX_STREAMS_COMPILER_H__ */
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 9b4a6ad9..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){
 
@@ -309,6 +323,7 @@ TrexStatelessDpCore::start_scheduler() {
     /* bail out in case of terminate */
     if (m_state != TrexStatelessDpCore::STATE_TERMINATE) {
         m_core->m_node_gen.close_file(m_core);
+        m_state = STATE_IDLE; /* we exit from all ports and we have nothing to do, we move to IDLE state */
     }
 }
 
@@ -401,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();
 
@@ -438,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 ) {
@@ -519,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();
@@ -581,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 **********************/