now support multiple interfaces / ports

11 files changed, 116 insertions, 25 deletions

diff --git a/src/bp_sim.cpp b/src/bp_sim.cpp
index 92beab91..842bc478 100755
--- a/src/bp_sim.cpp
+++ b/src/bp_sim.cpp
@@ -3421,8 +3421,6 @@ int CNodeGenerator::flush_file(dsec_t max_time,
             dsec_t dt ;
             thread->m_cpu_dp_u.commit();
 
-            thread->check_msgs();
-
             while ( true ) {
                 dt = now_sec() - n_time ;
 
@@ -3569,8 +3567,12 @@ void CNodeGenerator::handle_slow_messages(uint8_t type,
                 m_p_queue.push(node);
             }
 
-        } else if ( type == CGenNode::FLOW_SYNC ) { 
+        } else if ( type == CGenNode::FLOW_SYNC ) {
+
+            /* flow sync message is a sync point for time */
+            thread->m_cur_time_sec = node->m_time;
 
+            /* first pop the node */
             m_p_queue.pop();
 
             thread->check_msgs(); /* check messages */
@@ -3905,6 +3907,7 @@ const uint8_t test_udp_pkt[]={
 };
 
 void CFlowGenListPerThread::start_stateless_daemon(){
+    m_cur_time_sec = 0;
     m_stateless_dp_info->start();
 }
 
diff --git a/src/bp_sim.h b/src/bp_sim.h
index af084757..75958776 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -372,6 +372,15 @@ public:
      * @return 
      */
     virtual int update_mac_addr_from_global_cfg(pkt_dir_t       dir, rte_mbuf_t      *m)=0;
+
+    /**
+     * translate a port_id to the correct dir on the core
+     * 
+     */
+    virtual pkt_dir_t port_id_to_dir(uint8_t port_id) {
+        return (CS_INVALID);
+    }
+
 public:
 
 
diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h
index 5c2d42d2..5890a965 100644
--- a/src/internal_api/trex_platform_api.h
+++ b/src/internal_api/trex_platform_api.h
@@ -23,6 +23,7 @@ limitations under the License.
 #define __TREX_PLATFORM_API_H__
 
 #include <stdint.h>
+#include <vector>
 
 /**
  * Global stats
@@ -96,6 +97,7 @@ public:
 
 class TrexPlatformApi {
 public:
+    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 uint8_t get_dp_core_count() const = 0;
@@ -110,6 +112,7 @@ public:
  */
 class TrexDpdkPlatformApi : public TrexPlatformApi {
 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;
     uint8_t get_dp_core_count() const;
@@ -122,6 +125,7 @@ public:
  */
 class TrexMockPlatformApi : public TrexPlatformApi {
 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;
     uint8_t get_dp_core_count() const;
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index d4e07ef2..ed7f5e90 100755
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -1849,6 +1849,7 @@ public:
 
     virtual int update_mac_addr_from_global_cfg(pkt_dir_t       dir, rte_mbuf_t      *m);
 
+    virtual pkt_dir_t port_id_to_dir(uint8_t port_id);
 
 public:
     void GetCoreCounters(CVirtualIFPerSideStats *stats);
@@ -1861,6 +1862,10 @@ public:
         return ( CGlobalInfo::m_socket.port_to_socket( m_ports[0].m_port->get_port_id() ) );
     }
 
+    const CCorePerPort * get_ports() {
+        return m_ports;
+    }
+
 protected:
 
     int send_burst(CCorePerPort * lp_port,
@@ -2265,7 +2270,17 @@ int CCoreEthIF::update_mac_addr_from_global_cfg(pkt_dir_t  dir,
     return (0);
 }
 
+pkt_dir_t 
+CCoreEthIF::port_id_to_dir(uint8_t port_id) {
+
+    for (pkt_dir_t dir = 0; dir < CS_NUM; dir++) {
+        if (m_ports[dir].m_port->get_port_id() == port_id) {
+            return dir;
+        }
+    }
 
+    return (CS_INVALID);
+}
 
 class CLatencyHWPort : public CPortLatencyHWBase {
 public:
@@ -5172,3 +5187,21 @@ TrexDpdkPlatformApi::get_dp_core_count() const {
     return CGlobalInfo::m_options.preview.getCores();
 }
 
+
+void
+TrexDpdkPlatformApi::port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const {
+
+    cores_id_list.clear();
+
+    /* iterate over all DP cores */
+    for (uint8_t core_id = 0; core_id < g_trex.get_cores_tx(); core_id++) {
+
+        /* iterate over all the directions*/
+        for (uint8_t dir = 0 ; dir < CS_NUM; dir++) {
+            if (g_trex.m_cores_vif[core_id + 1]->get_ports()[dir].m_port->get_port_id() == port_id) {
+                cores_id_list.push_back(std::make_pair(core_id, dir));
+            }
+        }
+    }
+}
+
diff --git a/src/rx_check.h b/src/rx_check.h
index 6f9763a2..07f5684c 100755
--- a/src/rx_check.h
+++ b/src/rx_check.h
@@ -30,9 +30,10 @@ limitations under the License.
                                  
 
 typedef enum {
-    CLIENT_SIDE=0,    
-    SERVER_SIDE=1,    
-    CS_NUM=2
+    CLIENT_SIDE = 0,    
+    SERVER_SIDE = 1,    
+    CS_NUM = 2,
+    CS_INVALID = 255
 } pkt_dir_enum_t;
 
 typedef uint8_t   pkt_dir_t  ;
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;