merge from master

5 files changed, 434 insertions, 192 deletions

diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 95bdca0b..9770c735 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -57,7 +57,6 @@ TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api
 
     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);
@@ -86,17 +85,35 @@ TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api
  */
 void 
 TrexStatelessPort::acquire(const std::string &user, bool force) {
-    if ( (!is_free_to_aquire()) && (get_owner() != user) && (!force)) {
-        throw TrexRpcException("port is already taken by '" + get_owner() + "'");
+
+    /* if port is free - just take it */
+    if (get_owner().is_free()) {
+        get_owner().own(user);
+        return;
+    }
+
+    if (force) {
+        get_owner().own(user);
+
+        /* inform the other client of the steal... */
+        Json::Value data;
+        data["port_id"] = m_port_id;
+        get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_FORCE_ACQUIRED, data);
+
+    } else {
+        /* not same user or session id and not force - report error */
+        if (get_owner().get_name() == user) {
+            throw TrexRpcException("port is already owned by another session of '" + user + "'");
+        } else {
+            throw TrexRpcException("port is already taken by '" + get_owner().get_name() + "'");
+        }
     }
 
-    set_owner(user);
 }
 
 void
 TrexStatelessPort::release(void) {
-    verify_state( ~(PORT_STATE_TX | PORT_STATE_PAUSE) );
-    clear_owner();
+    get_owner().release();
 }
 
 /**
@@ -115,7 +132,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,12 +140,18 @@ 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);
+    TrexStreamsCompiler compiler;
+    bool rc = compiler.compile(m_port_id,
+                               streams,
+                               compiled_objs,
+                               get_dp_core_count(),
+                               factor,
+                               &fail_msg);
     if (!rc) {
-        throw TrexRpcException("Failed to compile streams");
+        throw TrexRpcException(fail_msg);
     }
 
     /* generate a message to all the relevant DP cores to start transmitting */
@@ -137,21 +160,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;
+    
 }
 
 
@@ -179,7 +210,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);
     
@@ -202,11 +233,15 @@ 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);
+
+    Json::Value data;
+    data["port_id"] = m_port_id;
+    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_PAUSED, data);
 }
 
 void
@@ -215,11 +250,16 @@ 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);
+
+
+    Json::Value data;
+    data["port_id"] = m_port_id;
+    get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_RESUMED, data);
 }
 
 void
@@ -230,19 +270,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");
         }
@@ -255,9 +295,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;
 
 }
 
@@ -310,27 +350,6 @@ TrexStatelessPort::change_state(port_state_e new_state) {
     m_port_state = new_state;
 }
 
-/**
- * generate a random connection handler
- * 
- */
-std::string 
-TrexStatelessPort::generate_handler() {
-    std::stringstream ss;
-
-    static const char alphanum[] =
-        "0123456789"
-        "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
-        "abcdefghijklmnopqrstuvwxyz";
-
-    /* generate 8 bytes of random handler */
-    for (int i = 0; i < 8; ++i) {
-        ss << alphanum[rand() % (sizeof(alphanum) - 1)];
-    }
-
-    return (ss.str());
-}
-
 
 void
 TrexStatelessPort::encode_stats(Json::Value &port) {
@@ -356,15 +375,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);
 }
 
 /**
@@ -393,7 +419,7 @@ TrexStatelessPort::on_dp_event_occured(TrexDpPortEvent::event_e event_type) {
 }
 
 uint64_t
-TrexStatelessPort::get_port_speed_bps() {
+TrexStatelessPort::get_port_speed_bps() const {
     switch (m_speed) {
     case TrexPlatformApi::SPEED_1G:
         return (1LLU * 1000 * 1000 * 1000);
@@ -410,11 +436,11 @@ TrexStatelessPort::get_port_speed_bps() {
 }
 
 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) */
@@ -424,19 +450,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:
@@ -518,3 +544,74 @@ TrexPortMultiplier(const std::string &type_str, const std::string &op_str, doubl
 
 }
 
+const TrexStreamsGraphObj *
+TrexStatelessPort::validate(void) {
+
+    /* first compile the graph */
+
+    vector<TrexStream *> streams;
+    get_object_list(streams);
+
+    if (streams.size() == 0) {
+        throw TrexException("no streams attached to port");
+    }
+
+    TrexStreamsCompiler compiler;
+    std::vector<TrexStreamsCompiledObj *> compiled_objs;
+
+    std::string fail_msg;
+    bool rc = compiler.compile(m_port_id,
+                               streams,
+                               compiled_objs,
+                               get_dp_core_count(),
+                               1.0,
+                               &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();
+    }
+
+    return m_graph_obj;
+}
+
+
+/************* Trex Port Owner **************/
+
+TrexPortOwner::TrexPortOwner() {
+    m_is_free = true;
+
+    /* for handlers random generation */
+    srand(time(NULL));
+}
+
+/**
+ * generate a random connection handler
+ * 
+ */
+std::string 
+TrexPortOwner::generate_handler() {
+    std::stringstream ss;
+
+    static const char alphanum[] =
+        "0123456789"
+        "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+        "abcdefghijklmnopqrstuvwxyz";
+
+    /* generate 8 bytes of random handler */
+    for (int i = 0; i < 8; ++i) {
+        ss << alphanum[rand() % (sizeof(alphanum) - 1)];
+    }
+
+    return (ss.str());
+}
+
+const std::string TrexPortOwner::g_unowned_name = "<FREE>";
+const std::string TrexPortOwner::g_unowned_handler = "";
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 45eb16e8..4988b46a 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -29,6 +29,77 @@ class TrexStatelessCpToDpMsgBase;
 class TrexStreamsGraphObj;
 class TrexPortMultiplier;
 
+/** 
+ * TRex port owner can perform
+ * write commands
+ * while port is owned - others can
+ * do read only commands
+ * 
+ */
+class TrexPortOwner {
+public:
+
+    TrexPortOwner();
+
+    /**
+     * is port free to acquire
+     */
+    bool is_free() {
+        return m_is_free;
+    }
+
+    void release() {
+        m_is_free = true;
+        m_owner_name = "";
+        m_handler = "";
+    }
+
+    bool is_owned_by(const std::string &user) {
+        return ( !m_is_free && (m_owner_name == user) );
+    }
+
+    void own(const std::string &owner_name) {
+
+        /* save user data */
+        m_owner_name = owner_name;
+
+        /* internal data */
+        m_handler = generate_handler();
+        m_is_free = false;
+    }
+
+    bool verify(const std::string &handler) {
+        return ( (!m_is_free) && (m_handler == handler) );
+    }
+
+    const std::string &get_name() {
+        return (!m_is_free ? m_owner_name : g_unowned_name);
+    }
+
+    const std::string &get_handler() {
+        return (!m_is_free ? m_handler : g_unowned_handler);
+    }
+
+
+private:
+    std::string  generate_handler();
+
+    /* is this port owned by someone ? */
+    bool         m_is_free;
+
+    /* user provided info */
+    std::string  m_owner_name;
+
+    /* handler genereated internally */
+    std::string  m_handler;
+    
+
+    /* just references defaults... */
+    static const std::string g_unowned_name;
+    static const std::string g_unowned_handler;
+};
+
+
 /**
  * describes a stateless port
  * 
@@ -76,10 +147,20 @@ public:
     void release(void);
 
     /**
+     * validate the state of the port before start 
+     * it will return a stream graph 
+     * containing information about the streams 
+     * configured on this port 
+     *  
+     * on error it throws TrexException
+     */
+    const TrexStreamsGraphObj *validate(void);
+
+    /**
      * start traffic
      * throws TrexException in case of an error
      */
-    void start_traffic(const TrexPortMultiplier &mul, double duration = -1);
+    void start_traffic(const TrexPortMultiplier &mul, double duration);
 
     /**
      * stop traffic
@@ -130,29 +211,6 @@ public:
     void get_properties(std::string &driver, TrexPlatformApi::driver_speed_e &speed);
 
 
-    /**
-    * query for ownership
-    * 
-    */
-    const std::string &get_owner() {
-        return m_owner;
-    }
-
-    /**
-     * owner handler 
-     * for the connection 
-     * 
-     */
-    const std::string &get_owner_handler() {
-        return m_owner_handler;
-    }
-
-
-    bool verify_owner_handler(const std::string &handler) {
-
-        return ( (m_owner != "none") && (m_owner_handler == handler) );
-
-    }
 
     /**
      * encode stats as JSON
@@ -172,6 +230,7 @@ public:
         verify_state(PORT_STATE_IDLE | PORT_STATE_STREAMS);
 
         m_stream_table.add_stream(stream);
+        delete_streams_graph();
 
         change_state(PORT_STATE_STREAMS);
     }
@@ -180,6 +239,7 @@ public:
         verify_state(PORT_STATE_STREAMS);
 
         m_stream_table.remove_stream(stream);
+        delete_streams_graph();
 
         if (m_stream_table.size() == 0) {
             change_state(PORT_STATE_IDLE);
@@ -190,6 +250,7 @@ public:
         verify_state(PORT_STATE_IDLE | PORT_STATE_STREAMS);
 
         m_stream_table.remove_and_delete_all_streams();
+        delete_streams_graph();
 
         change_state(PORT_STATE_IDLE);
     }
@@ -211,30 +272,34 @@ public:
     }
 
 
-private:
-
-  
-
     /**
-    * take ownership of the server array 
-    * this is static 
-    * ownership is total 
-    * 
-    */
-    void set_owner(const std::string &owner) {
-        m_owner = owner;
-        m_owner_handler = generate_handler();
+     * returns the number of DP cores linked to this port
+     * 
+     */
+    uint8_t get_dp_core_count() {
+        return m_cores_id_list.size();
     }
 
-    void clear_owner() {
-        m_owner = "none";
-        m_owner_handler = "";
+    /**
+     * returns the traffic multiplier currently being used by the DP
+     * 
+     */
+    double get_multiplier() {
+        return (m_factor);
     }
 
-    bool is_free_to_aquire() {
-        return (m_owner == "none");
+    /**
+     * get port speed in bits per second
+     * 
+     */
+    uint64_t get_port_speed_bps() const;
+
+    TrexPortOwner & get_owner() {
+        return m_owner;
     }
 
+private:
+
 
     const std::vector<int> get_core_id_list () {
         return m_cores_id_list;
@@ -246,7 +311,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
@@ -259,13 +334,9 @@ private:
      * calculate effective M per core
      * 
      */
-    double calculate_effective_mul(const TrexPortMultiplier &mul);
+    double calculate_effective_factor(const TrexPortMultiplier &mul);
 
-    /**
-     * get port speed in bits per second
-     * 
-     */
-    uint64_t get_port_speed_bps();
+  
 
     /**
      * generates a graph of streams graph
@@ -284,25 +355,25 @@ private:
     TrexStreamTable    m_stream_table;
     uint8_t            m_port_id;
     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;
+    std::vector<int>   m_cores_id_list;
 
     bool               m_last_all_streams_continues;
     double             m_last_duration;
-    double             m_current_per_core_m;
+    double             m_factor;
 
     TrexDpPortEvents   m_dp_events;
 
     /* holds a graph of streams rate*/
     const TrexStreamsGraphObj  *m_graph_obj;
-};
 
+    /* owner information */
+    TrexPortOwner       m_owner;
+};
 
 
 /**
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index c42d0985..529dcbe4 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -129,8 +129,9 @@ 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_has_vm = m_has_vm;
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index c8aa1e40..478e09f8 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);
@@ -219,7 +207,7 @@ TrexStreamsCompiler::check_stream(const TrexStream *stream) {
     /* cont. stream can point only on itself */
     if (stream->get_type() == TrexStream::stCONTINUOUS) {
         if (stream->m_next_stream_id != -1) {
-            ss << "continous stream '" << stream->m_stream_id << "' cannot point on another stream";
+            ss << "continous stream '" << stream->m_stream_id << "' cannot point to another stream";
             err(ss.str());
         }
     }
@@ -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,
+                             const std::vector<TrexStream *>        &streams,
+                             std::vector<TrexStreamsCompiledObj *>  &objs,
+                             uint8_t                                dp_core_count,
+                             double                                 factor,
+                             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;
+        }
+    }
 
-    bool all_continues=true;
-    /* for now we do something trivial, */
+    /* 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);
+    }
+
+    /* 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
  *************************************/
@@ -490,6 +536,9 @@ TrexStreamsGraph::add_rate_events_for_stream_cont(double &offset_usec, const Tre
 
     /* no more events after this stream */
     offset_usec = -1;
+
+    /* also mark we have an inifite time */
+    m_graph_obj->m_expected_duration = -1;
 }
 
 /**
@@ -602,6 +651,7 @@ TrexStreamsGraph::generate_graph_for_one_root(uint32_t root_stream_id) {
         /* loop detection */
         auto search = loop_hash.find(stream->m_next_stream_id);
         if (search != loop_hash.end()) {
+            m_graph_obj->on_loop_detection();
             break;
         }
 
@@ -674,6 +724,11 @@ TrexStreamsGraphObj::find_max_rate() {
         max_rate_bps = std::max(max_rate_bps, current_rate_bps);
     }
 
+    /* if not mark as inifite - get the last event time */
+    if (m_expected_duration != -1) {
+        m_expected_duration = m_rate_events.back().time;
+    }
+
     m_max_pps = max_rate_pps;
     m_max_bps = max_rate_bps;
 }
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index a4c12f8d..7fe2dbf2 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,
+                 const std::vector<TrexStream *>        &streams,
+                 std::vector<TrexStreamsCompiledObj *>  &objs,
+                 uint8_t                                dp_core_count = 1,
+                 double                                 factor = 1.0,
+                 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;
@@ -131,6 +133,12 @@ class TrexStreamsGraphObj {
 
 public:
 
+    TrexStreamsGraphObj() {
+        m_max_pps           = 0;
+        m_max_bps           = 0;
+        m_expected_duration = 0;
+    }
+
     /**
      * rate event is defined by those: 
      * time - the time of the event on the timeline 
@@ -153,12 +161,21 @@ public:
         return m_max_bps;
     }
 
+    int get_duration() const {
+        return m_expected_duration;
+    }
+
     const std::list<rate_event_st> & get_events() const {
         return m_rate_events;
     }
 
+
 private:
 
+    void on_loop_detection() {
+        m_expected_duration = -1;
+    }
+
     void add_rate_event(const rate_event_st &ev) {
         m_rate_events.push_back(ev);
     }
@@ -166,8 +183,9 @@ private:
     void generate();
     void find_max_rate();
 
-    double m_max_pps;
-    double m_max_bps;
+    double  m_max_pps;
+    double  m_max_bps;
+    int     m_expected_duration;
 
     /* list of rate events */
     std::list<rate_event_st> m_rate_events;