support for multiple streams

8 files changed, 120 insertions, 51 deletions

diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 92d0a7f8..d3828f53 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -58,32 +58,16 @@ TrexStatelessPort::start_traffic(void) {
         return (RC_ERR_BAD_STATE_FOR_OP);
     }
 
-    if (get_stream_table()->size() == 0) {
-        return (RC_ERR_NO_STREAMS);
-    }
-
-    m_port_state = PORT_STATE_TRANSMITTING;
-
-
-    /* ************* A HACK FOR NOW *************/
-    /* we support only one stream continious */
-    if (get_stream_table()->size() != 1)  {
-        return (RC_ERR_FAILED_TO_COMPILE_STREAMS);
-    }
-
-    TrexStream *stream;
-    for (auto it = get_stream_table()->begin(); it != get_stream_table()->end(); it++ ) {
-        stream = (*it).second;
-    }
+    TrexStreamsCompiledObj *compiled_obj = new TrexStreamsCompiledObj(); 
 
-    /* support only cont streams */
-    TrexStreamContinuous *cont_stream = dynamic_cast<TrexStreamContinuous *>(stream);
-    if (!cont_stream) {
+    /* compile the streams */
+    bool rc = get_stream_table()->compile(*compiled_obj);
+    if (!rc) {
         return (RC_ERR_FAILED_TO_COMPILE_STREAMS);
     }
 
     /* generate a message to all the relevant DP cores to start transmitting */
-    TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(cont_stream->m_pkt.binary, cont_stream->m_pkt.len, cont_stream->get_pps());
+    TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(compiled_obj);
 
     // FIXME (add the right core list)
     CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(0);
diff --git a/src/stateless/cp/trex_stream.cpp b/src/stateless/cp/trex_stream.cpp
index 182036f1..2fd91560 100644
--- a/src/stateless/cp/trex_stream.cpp
+++ b/src/stateless/cp/trex_stream.cpp
@@ -20,6 +20,7 @@ limitations under the License.
 */
 #include <trex_stream.h>
 #include <cstddef>
+#include <string.h>
 
 /**************************************
  * stream
@@ -57,6 +58,29 @@ TrexStream::get_stream_json() {
 }
 
 /**************************************
+ * stream compiled object
+ *************************************/
+TrexStreamsCompiledObj::~TrexStreamsCompiledObj() {
+    for (auto &obj : m_objs) {
+        delete obj.m_pkt;
+    }
+    m_objs.clear();
+}
+
+void 
+TrexStreamsCompiledObj::add_compiled_stream(double pps, uint8_t *pkt, uint16_t pkt_len) {
+    obj_st obj;
+
+    obj.m_pps = pps;
+    obj.m_pkt_len = pkt_len;
+
+    obj.m_pkt = new uint8_t[pkt_len];
+    memcpy(obj.m_pkt, pkt, pkt_len);
+
+    m_objs.push_back(obj);
+}
+
+/**************************************
  * stream table
  *************************************/
 TrexStreamTable::TrexStreamTable() {
@@ -114,3 +138,31 @@ void TrexStreamTable::get_stream_list(std::vector<uint32_t> &stream_list) {
 int TrexStreamTable::size() {
     return m_stream_table.size();
 }
+
+
+bool 
+TrexStreamTable::compile(TrexStreamsCompiledObj &obj) {
+
+    /* for now we do something trivial, */
+    for (auto it = m_stream_table.begin(); it != m_stream_table.end(); it++ ) {
+        TrexStream *stream = (*it).second;
+
+        if (!stream->m_enabled) {
+            continue;
+        }
+        if (!stream->m_self_start) {
+            continue;
+        }
+
+        /* support only continous for now ... */
+        TrexStreamContinuous *cont_stream = dynamic_cast<TrexStreamContinuous *>(stream);
+        if (!cont_stream) {
+            continue;
+        }
+
+        obj.add_compiled_stream(cont_stream->get_pps(), cont_stream->m_pkt.binary, cont_stream->m_pkt.len);
+    }
+
+    return true;
+}
+
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index 91e64d23..d422f9f4 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -150,6 +150,28 @@ protected:
 };
 
 /**
+ * compiled object for a table of streams
+ * 
+ * @author imarom (28-Oct-15)
+ */
+class TrexStreamsCompiledObj {
+public:
+
+    TrexStreamsCompiledObj() {}
+    ~TrexStreamsCompiledObj();
+
+    void add_compiled_stream(double pps, uint8_t *pkt, uint16_t pkt_len);
+
+    struct obj_st {
+        double   m_pps;
+        uint8_t *m_pkt;
+        uint16_t m_pkt_len;
+    };
+
+    std::vector<obj_st> m_objs;
+};
+
+/**
  * holds all the streams 
  *  
  */
@@ -193,6 +215,16 @@ public:
     void get_stream_list(std::vector<uint32_t> &stream_list);
 
     /**
+     * compiles all the streams in the table to a DP object that 
+     * can be passed to the DP cores 
+     * 
+     * @author imarom (28-Oct-15)
+     * 
+     * @return bool 
+     */
+    bool compile(TrexStreamsCompiledObj &obj);
+
+    /**
      * get the table size
      * 
      */
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 20eadfc5..0262a80a 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -20,6 +20,7 @@ limitations under the License.
 */
 #include <trex_stateless_dp_core.h>
 #include <trex_stateless_messaging.h>
+#include <trex_stream.h>
 
 #include <bp_sim.h>
 
@@ -74,10 +75,7 @@ TrexStatelessDpCore::handle_pkt_event(CGenNode *node) {
 }
 
 void
-TrexStatelessDpCore::start_const_traffic(const uint8_t *pkt,
-                                         uint16_t pkt_len,
-                                         double pps) {
-
+TrexStatelessDpCore::add_cont_stream(double pps, const uint8_t *pkt, uint16_t pkt_len) {
     CGenNodeStateless *node = m_core->create_node_sl();
 
     /* add periodic */
@@ -117,7 +115,13 @@ TrexStatelessDpCore::start_const_traffic(const uint8_t *pkt,
     m_state = TrexStatelessDpCore::STATE_TRANSMITTING;
 
     m_core->m_node_gen.add_node((CGenNode *)node);
+}
 
+void
+TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj) {
+    for (auto single_stream : obj->m_objs) {
+        add_cont_stream(single_stream.m_pps, single_stream.m_pkt, single_stream.m_pkt_len);
+    }
 }
 
 void
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index 65d894d2..da8484a6 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -28,6 +28,7 @@ class TrexStatelessCpToDpMsgBase;
 class TrexStatelessDpStart;
 class CFlowGenListPerThread;
 class CGenNode;
+class TrexStreamsCompiledObj;
 
 class TrexStatelessDpCore {
 
@@ -62,7 +63,7 @@ public:
      * @param pkt 
      * @param pkt_len 
      */
-    void start_const_traffic(const uint8_t *pkt, uint16_t pkt_len, double pps);
+    void start_traffic(TrexStreamsCompiledObj *obj);
 
     /**
      * stop all traffic for this core
@@ -107,6 +108,8 @@ private:
      */
     void handle_cp_msg(TrexStatelessCpToDpMsgBase *msg);
 
+    void add_cont_stream(double pps, const uint8_t *pkt, uint16_t pkt_len);
+
     uint8_t              m_thread_id;
     state_e              m_state;
     CNodeRing           *m_ring_from_cp;
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 4cd5b416..4ef508fc 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -20,32 +20,25 @@ limitations under the License.
 */
 #include <trex_stateless_messaging.h>
 #include <trex_stateless_dp_core.h>
+#include <trex_stream.h>
 #include <string.h>
 
 /*************************
   start traffic message
  ************************/ 
-TrexStatelessDpStart::TrexStatelessDpStart(const uint8_t *pkt, uint16_t pkt_len, double pps) {
-    assert(pkt);
-    assert(pkt_len > 0);
-
-    m_pkt = new uint8_t[pkt_len];
-    memcpy(m_pkt, pkt, pkt_len);
-    m_pkt_len = pkt_len;
-
-    m_pps = pps;
+TrexStatelessDpStart::TrexStatelessDpStart(TrexStreamsCompiledObj *obj) : m_obj(obj) {
 }
 
 TrexStatelessDpStart::~TrexStatelessDpStart() {
-    if (m_pkt) {
-        delete m_pkt;
+    if (m_obj) {
+        delete m_obj;
     }
 }
 
 bool
 TrexStatelessDpStart::handle(TrexStatelessDpCore *dp_core) {
 
-    dp_core->start_const_traffic(m_pkt, m_pkt_len, m_pps);
+    dp_core->start_traffic(m_obj);
     return true;
 }
 
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index af05aa4c..13f6c05a 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -24,6 +24,7 @@ limitations under the License.
 #include <msg_manager.h>
 
 class TrexStatelessDpCore;
+class TrexStreamsCompiledObj;
 
 /**
  * defines the base class for CP to DP messages
@@ -54,24 +55,14 @@ public:
 class TrexStatelessDpStart : public TrexStatelessCpToDpMsgBase {
 public:
 
-    TrexStatelessDpStart(const uint8_t *pkt, uint16_t pkt_len, double pps);
+    TrexStatelessDpStart(TrexStreamsCompiledObj *obj);
 
     ~TrexStatelessDpStart();
 
-    const uint8_t * get_pkt() {
-        return m_pkt;
-    }
-
-    uint16_t get_pkt_len() {
-        return m_pkt_len;
-    }
-
     virtual bool handle(TrexStatelessDpCore *dp_core);
 
 private:
-    uint8_t *m_pkt;
-    uint16_t m_pkt_len;
-    double   m_pps;
+    TrexStreamsCompiledObj *m_obj;
 };
 
 /**
diff --git a/src/stub/trex_stateless_stub.cpp b/src/stub/trex_stateless_stub.cpp
index 8e8d4fa8..06a9e189 100644
--- a/src/stub/trex_stateless_stub.cpp
+++ b/src/stub/trex_stateless_stub.cpp
@@ -4,7 +4,17 @@
 class CFlowGenListPerThread;
 class TrexStatelessCpToDpMsgBase;
 
-TrexStatelessDpCore::TrexStatelessDpCore(unsigned char, CFlowGenListPerThread*) {}
+TrexStatelessDpCore::TrexStatelessDpCore(unsigned char, CFlowGenListPerThread*) {
+    m_thread_id = 0;
+    m_core = NULL;
+
+    m_state = STATE_IDLE;
+
+    CMessagingManager * cp_dp = CMsgIns::Ins()->getCpDp();
+
+    m_ring_from_cp = cp_dp->getRingCpToDp(0);
+    m_ring_to_cp   = cp_dp->getRingDpToCp(0);
+}
 
 void TrexStatelessDpCore::start(){}