now support multiple interfaces / ports

6 files changed, 60 insertions, 19 deletions

diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index a0b57b63..a64ee92f 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -36,6 +36,9 @@ limitations under the License.
 #include <rte_ethdev.h>
 #include <os_time.h>
 
+void
+port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list);
+
 using namespace std;
 
 /***************************
@@ -79,10 +82,7 @@ TrexStatelessPort::start_traffic(double mul) {
     /* generate a message to all the relevant DP cores to start transmitting */
     TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(compiled_obj);
 
-    // FIXME (add the right core list)
-    CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(0);
-
-    ring->Enqueue((CGenNode *)start_msg);
+    send_message_to_dp(start_msg);
 
     /* move the state to transmiting */
     m_port_state = PORT_STATE_TRANSMITTING;
@@ -101,10 +101,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);
 
-    // FIXME (add the right core list)
-    CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(0);
-
-    ring->Enqueue((CGenNode *)stop_msg);
+    send_message_to_dp(stop_msg);
 
     m_port_state = PORT_STATE_UP_IDLE;
 
@@ -191,3 +188,18 @@ TrexStatelessPort::encode_stats(Json::Value &port) {
     port["tx_rx_errors"]    = Json::Value::UInt64(stats.m_stats.m_tx_rx_errors);
 }
 
+void 
+TrexStatelessPort::send_message_to_dp(TrexStatelessCpToDpMsgBase *msg) {
+
+    std::vector<std::pair<uint8_t, uint8_t>> cores_id_list;
+
+    get_stateless_obj()->get_platform_api()->port_id_to_cores(m_port_id, cores_id_list);
+
+    for (auto core_pair : cores_id_list) {
+
+        /* send the message to the core */
+        CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_pair.first);
+        ring->Enqueue((CGenNode *)msg);
+    }
+
+}
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 3e071954..09183768 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -23,6 +23,8 @@ limitations under the License.
 
 #include <trex_stream.h>
 
+class TrexStatelessCpToDpMsgBase;
+
 /**
  * describes a stateless port
  * 
@@ -150,6 +152,8 @@ private:
 
     std::string generate_handler();
 
+    void send_message_to_dp(TrexStatelessCpToDpMsgBase *msg);
+
     TrexStreamTable  m_stream_table;
     uint8_t          m_port_id;
     port_state_e     m_port_state;
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index 5e2602ec..f394b971 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -37,12 +37,13 @@ TrexStreamsCompiledObj::~TrexStreamsCompiledObj() {
 }
 
 void 
-TrexStreamsCompiledObj::add_compiled_stream(double pps, uint8_t *pkt, uint16_t pkt_len) {
+TrexStreamsCompiledObj::add_compiled_stream(double isg_usec, double pps, uint8_t *pkt, uint16_t pkt_len) {
     obj_st obj;
 
-    obj.m_port_id = m_port_id;
-    obj.m_pps     = pps * m_mul;
-    obj.m_pkt_len = pkt_len;
+    obj.m_isg_usec  = isg_usec;
+    obj.m_port_id   = m_port_id;
+    obj.m_pps       = pps * m_mul;
+    obj.m_pkt_len   = pkt_len;
 
     obj.m_pkt = new uint8_t[pkt_len];
     memcpy(obj.m_pkt, pkt, pkt_len);
@@ -75,7 +76,8 @@ TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams, TrexStrea
         }
 
         /* add it */
-        obj.add_compiled_stream(cont_stream->get_pps(),
+        obj.add_compiled_stream(cont_stream->m_isg_usec,
+                                cont_stream->get_pps(),
                                 cont_stream->m_pkt.binary,
                                 cont_stream->m_pkt.len);
     }
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index 06f992ed..752f76b2 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -40,6 +40,7 @@ public:
     ~TrexStreamsCompiledObj();
 
     struct obj_st {
+        double   m_isg_usec;
         double   m_pps;
         uint8_t *m_pkt;
         uint16_t m_pkt_len;
@@ -51,7 +52,7 @@ public:
     }
 
 private:
-    void add_compiled_stream(double pps, uint8_t *pkt, uint16_t pkt_len);
+    void add_compiled_stream(double isg_usec, double pps, uint8_t *pkt, uint16_t pkt_len);
     std::vector<obj_st> m_objs;
 
     uint8_t m_port_id;
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 306b23d0..5f4e553a 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -25,6 +25,11 @@ limitations under the License.
 
 #include <bp_sim.h>
 
+static inline double
+usec_to_sec(double usec) {
+    return (usec / (1000 * 1000));
+}
+
 TrexStatelessDpCore::TrexStatelessDpCore(uint8_t thread_id, CFlowGenListPerThread *core) {
     m_thread_id = thread_id;
     m_core = core;
@@ -82,12 +87,20 @@ TrexStatelessDpCore::start() {
 }
 
 void
-TrexStatelessDpCore::add_cont_stream(double pps, const uint8_t *pkt, uint16_t pkt_len) {
+TrexStatelessDpCore::add_cont_stream(uint8_t port_id,
+                                     double isg_usec,
+                                     double pps,
+                                     const uint8_t *pkt,
+                                     uint16_t pkt_len) {
+
     CGenNodeStateless *node = m_core->create_node_sl();
 
     /* add periodic */
     node->m_type = CGenNode::STATELESS_PKT;
-    node->m_time = m_core->m_cur_time_sec + 0.0 /* STREAM ISG */;
+
+    node->m_time = m_core->m_cur_time_sec + usec_to_sec(isg_usec);
+
+    pkt_dir_t dir = m_core->m_node_gen.m_v_if->port_id_to_dir(port_id);
     node->m_flags = 0; 
 
     /* set socket id */
@@ -97,8 +110,10 @@ TrexStatelessDpCore::add_cont_stream(double pps, const uint8_t *pkt, uint16_t pk
     uint16_t pkt_size = pkt_len;
     const uint8_t *stream_pkt = pkt;
 
+    /* stateless specific fields */
     node->m_next_time_offset = 1.0 / pps;
     node->m_is_stream_active = 1;
+    node->m_port_id = port_id;
 
     /* allocate const mbuf */
     rte_mbuf_t *m = CGlobalInfo::pktmbuf_alloc(node->get_socket_id(), pkt_size);
@@ -110,7 +125,6 @@ TrexStatelessDpCore::add_cont_stream(double pps, const uint8_t *pkt, uint16_t pk
     memcpy(p,stream_pkt,pkt_size);
 
     /* set dir 0 or 1 client or server */
-    pkt_dir_t dir = 0;
     node->set_mbuf_cache_dir(dir);
 
     /* TBD repace the mac if req we should add flag  */
@@ -132,7 +146,11 @@ TrexStatelessDpCore::add_cont_stream(double pps, const uint8_t *pkt, uint16_t pk
 void
 TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj) {
     for (auto single_stream : obj->get_objects()) {
-        add_cont_stream(single_stream.m_pps, single_stream.m_pkt, single_stream.m_pkt_len);
+        add_cont_stream(single_stream.m_port_id,
+                        single_stream.m_isg_usec,
+                        single_stream.m_pps,
+                        single_stream.m_pkt,
+                        single_stream.m_pkt_len);
     }
 }
 
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index 698cac2f..f3b5ff62 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -115,7 +115,11 @@ private:
      */
     void handle_cp_msg(TrexStatelessCpToDpMsgBase *msg);
 
-    void add_cont_stream(double pps, const uint8_t *pkt, uint16_t pkt_len);
+    void add_cont_stream(uint8_t dir,
+                         double isg,
+                         double pps,
+                         const uint8_t *pkt,
+                         uint16_t pkt_len);
 
     uint8_t              m_thread_id;
     state_e              m_state;