draft: RX software feature

Signed-off-by: imarom <imarom@cisco.com>

15 files changed, 321 insertions, 6 deletions

diff --git a/linux_dpdk/ws_main.py b/linux_dpdk/ws_main.py
index 0a6ff7c1..91cd728a 100755
--- a/linux_dpdk/ws_main.py
+++ b/linux_dpdk/ws_main.py
@@ -202,7 +202,8 @@ stateless_src = SrcGroup(dir='src/stateless/',
                                     'cp/trex_dp_port_events.cpp',
                                     'dp/trex_stateless_dp_core.cpp',
                                     'messaging/trex_stateless_messaging.cpp',
-                                    'rx/trex_stateless_rx_core.cpp'
+                                    'rx/trex_stateless_rx_core.cpp',
+                                    'rx/trex_stateless_rx_port_mngr.cpp'
                                     ])
 # JSON package
 json_src = SrcGroup(dir='external_libs/json',
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
index 80a4c4dc..2e3f681c 100755
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py
@@ -1739,6 +1739,10 @@ class STLClient(object):
 
         """
 
+        rc = self.ports[0].set_rx_filter_mode("all")
+        if not rc:
+            raise STLError(rc)
+
         self.logger.pre_cmd("Pinging the server on '{0}' port '{1}': ".format(self.connection_info['server'],
                                                                               self.connection_info['sync_port']))
         rc = self._transmit("ping", api_class = None)
@@ -3267,12 +3271,13 @@ class STLClient(object):
             return self.set_port_attr(opts.ports, opts.prom, opts.link, opts.led, opts.flow_ctrl)
 
 
+                        
     @__console
     def show_profile_line (self, line):
         '''Shows profile information'''
 
         parser = parsing_opts.gen_parser(self,
-                                         "port",
+                                         "profile",
                                          self.show_profile_line.__doc__,
                                          parsing_opts.FILE_PATH)
 
diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py
index cec3761f..bf0251b8 100644
--- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py
+++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py
@@ -487,6 +487,21 @@ class Port(object):
 
         return self.ok()
 
+    @writeable
+    def set_rx_filter_mode (self, filter_mode):
+        assert(filter_mode in ["hw", "all"])
+
+        params = {"handler":      self.handler,
+                  "port_id":      self.port_id,
+                  "type":         "filter_mode",
+                  "filter_type":  filter_mode}
+
+        rc = self.transmit("set_rx_feature", params)
+        if rc.bad():
+            return self.err(rc.err())
+
+        return self.ok()
+
     @owned
     def pause (self):
 
@@ -583,6 +598,18 @@ class Port(object):
 
         return self.ok()
 
+    @owned
+    def get_rx_sw_pkts (self):
+        params = {"handler": self.handler,
+                  "port_id": self.port_id}
+
+        rc = self.transmit("get_rx_sw_pkts", params)
+        if rc.bad():
+            return self.err(rc.err())
+
+        return self.ok()
+
+
     @writeable
     def push_remote (self, pcap_filename, ipg_usec, speedup, count, duration, is_dual, slave_handler):
 
diff --git a/src/common/pcap.h b/src/common/pcap.h
index 3f8dfd21..c9139e4c 100755
--- a/src/common/pcap.h
+++ b/src/common/pcap.h
@@ -1,5 +1,5 @@
-#ifndef __LIBPCAP_H__
-#define __LIBPCAP_H__ 
+#ifndef __TREX_LIBPCAP_H__
+#define __TREX_LIBPCAP_H__ 
 
 /*
 Copyright (c) 2015-2015 Cisco Systems, Inc.
@@ -151,4 +151,5 @@ private:
     bool m_is_open;
     uint32_t m_pkt_count;
 };
+
 #endif
diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h
index 631f9a3e..4feb386f 100644
--- a/src/internal_api/trex_platform_api.h
+++ b/src/internal_api/trex_platform_api.h
@@ -28,6 +28,7 @@ limitations under the License.
 #include <string.h>
 #include "flow_stat_parser.h"
 #include "trex_defs.h"
+#include "trex_stateless_rx_defs.h"
 #include "trex_port_attr.h"
 #include <json/json.h>
 
@@ -160,6 +161,8 @@ public:
     virtual int get_xstats_values(uint8_t port_id, xstats_values_t &xstats_values) const = 0;
     virtual int get_xstats_names(uint8_t port_id, xstats_names_t &xstats_names) const = 0;
 
+    virtual void set_rx_filter_mode(uint8_t port_id, rx_filter_mode_e filter_mode) const = 0;
+
     virtual ~TrexPlatformApi() {}
 };
 
@@ -197,6 +200,7 @@ public:
     int get_mbuf_util(Json::Value &result) const;
     void mark_for_shutdown() const;
     CFlowStatParser *get_flow_stat_parser() const;
+    void set_rx_filter_mode(uint8_t port_id, rx_filter_mode_e filter_mode) const;
     TRexPortAttr *getPortAttrObj(uint8_t port_id) const;
 
     int get_xstats_values(uint8_t port_id, xstats_values_t &xstats_values) const;
@@ -277,6 +281,8 @@ public:
     int get_xstats_values(uint8_t port_id, xstats_values_t &xstats_values) const {return 0;};
     int get_xstats_names(uint8_t port_id, xstats_names_t &xstats_names) const {return 0;};
 
+    void set_rx_filter_mode(uint8_t port_id, rx_filter_mode_e filter_mode) const {}
+
 private:
     int m_dp_core_count;
     SimTRexPortAttr * m_port_attr;
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 9f2b62b2..dbc45889 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -6603,3 +6603,14 @@ TRexPortAttr *TrexDpdkPlatformApi::getPortAttrObj(uint8_t port_id) const {
 void TrexDpdkPlatformApi::mark_for_shutdown() const {
     g_trex.mark_for_shutdown(CGlobalTRex::SHUTDOWN_RPC_REQ);
 }
+
+
+void TrexDpdkPlatformApi::set_rx_filter_mode(uint8_t port_id, rx_filter_mode_e filter_mode) const {
+    
+    CPhyEthIF *_if = &g_trex.m_ports[port_id];
+
+    bool recv_all = (filter_mode == RX_FILTER_MODE_ALL);
+    CTRexExtendedDriverDb::Ins()->get_drv()->set_rcv_all(_if, recv_all);
+}
+
+
diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index 109cc1a4..8599b61b 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -27,6 +27,8 @@ limitations under the License.
 
 #include <internal_api/trex_platform_api.h>
 
+#include "trex_stateless_rx_core.h"
+
 #include <fstream>
 #include <iostream>
 #include <unistd.h>
@@ -640,3 +642,84 @@ TrexRpcCmdPushRemote::_run(const Json::Value &params, Json::Value &result) {
 
 }
 
+/**
+ * set on/off RX software receive mode
+ *
+ */
+trex_rpc_cmd_rc_e
+TrexRpcCmdSetRxFeature::_run(const Json::Value &params, Json::Value &result) {
+    
+    uint8_t port_id = parse_port(params, result);
+	TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    /* decide which feature is being set */
+    const std::string type = parse_choice(params, "type", {"filter_mode", "record", "queue", "server"}, result);
+
+    if (type == "filter_mode") {
+        parse_filter_mode_msg(params, port, result);
+    } else if (type == "record") {
+        parse_record_msg(params, port, result);
+    } else if (type == "queue") {
+        parse_queue_msg(params, port, result);
+    } else if (type == "server") {
+        parse_server_msg(params, port, result);
+    } else {
+        assert(0);
+    }
+
+	result["result"] = Json::objectValue;
+    return (TREX_RPC_CMD_OK);
+   
+}
+
+void 
+TrexRpcCmdSetRxFeature::parse_filter_mode_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result) {
+	const std::string type = parse_choice(msg, "filter_type", {"hw", "all"}, result);
+
+	rx_filter_mode_e filter_mode;
+	if (type == "hw") {
+		filter_mode = RX_FILTER_MODE_HW;
+	} else if (type == "all") {
+		filter_mode = RX_FILTER_MODE_ALL;
+	} else {
+		assert(0);
+	}
+
+	try {
+        port->set_rx_filter_mode(filter_mode);
+    } catch (const TrexException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+}
+
+void 
+TrexRpcCmdSetRxFeature::parse_record_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result) {
+}
+
+void 
+TrexRpcCmdSetRxFeature::parse_queue_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result) {
+}
+
+void 
+TrexRpcCmdSetRxFeature::parse_server_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result) {
+}
+
+
+trex_rpc_cmd_rc_e
+TrexRpcCmdGetRxSwPkts::_run(const Json::Value &params, Json::Value &result) {
+    
+    uint8_t port_id = parse_port(params, result);
+
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
+
+    try {
+        RxPacketBuffer *pkt_buffer = port->get_rx_sw_pkts();
+        result["result"]["pkts"] = pkt_buffer->to_json();
+
+    } catch (const TrexException &ex) {
+        generate_execute_err(result, ex.what());
+    }
+
+    
+    return (TREX_RPC_CMD_OK);
+}
diff --git a/src/rpc-server/commands/trex_rpc_cmds.h b/src/rpc-server/commands/trex_rpc_cmds.h
index 5fde1d0c..8d4aeb0c 100644
--- a/src/rpc-server/commands/trex_rpc_cmds.h
+++ b/src/rpc-server/commands/trex_rpc_cmds.h
@@ -27,6 +27,7 @@ limitations under the License.
 #include <memory>
 
 class TrexStream;
+class TrexStatelessPort;
 
 /* all the RPC commands decl. goes here */
 
@@ -144,5 +145,15 @@ TREX_RPC_CMD_DEFINE(TrexRpcCmdPushRemote, "push_remote", 6, true, APIClass::API_
 
 TREX_RPC_CMD_DEFINE(TrexRpcCmdShutdown, "shutdown", 2, false, APIClass::API_CLASS_TYPE_CORE);
 
+TREX_RPC_CMD_DEFINE_EXTENDED(TrexRpcCmdSetRxFeature, "set_rx_feature", 3, false, APIClass::API_CLASS_TYPE_CORE,
+    void parse_filter_mode_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result);
+	void parse_record_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result);
+	void parse_queue_msg(const Json::Value &msg,  TrexStatelessPort *port, Json::Value &result);
+	void parse_server_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result);
+
+);
+
+TREX_RPC_CMD_DEFINE(TrexRpcCmdGetRxSwPkts, "get_rx_sw_pkts", 2, false, APIClass::API_CLASS_TYPE_CORE);
+
 #endif /* __TREX_RPC_CMD_H__ */
 
diff --git a/src/rpc-server/trex_rpc_cmds_table.cpp b/src/rpc-server/trex_rpc_cmds_table.cpp
index cddf19b9..45e32d4a 100644
--- a/src/rpc-server/trex_rpc_cmds_table.cpp
+++ b/src/rpc-server/trex_rpc_cmds_table.cpp
@@ -71,6 +71,9 @@ TrexRpcCommandsTable::TrexRpcCommandsTable() {
     register_command(new TrexRpcCmdPushRemote());
 
     register_command(new TrexRpcCmdShutdown());
+
+    register_command(new TrexRpcCmdSetRxFeature());
+    register_command(new TrexRpcCmdGetRxSwPkts());
 }
 
 
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 9bb20990..5d31abbf 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -25,6 +25,7 @@ limitations under the License.
 #include <trex_streams_compiler.h>
 #include <common/basic_utils.h>
 #include <common/captureFile.h>
+#include "trex_stateless_rx_defs.h"
 
 #include <string>
 
@@ -944,6 +945,32 @@ TrexStatelessPort::remove_and_delete_all_streams() {
     }
 }
 
+/**
+ * set the filter type for a port
+ * 
+ * @author imarom (10/31/2016)
+ * 
+ * @param rx_sw_cfg 
+ */
+void
+TrexStatelessPort::set_rx_filter_mode(rx_filter_mode_e filter_mode) {
+    TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxSetFilterMode(m_port_id, filter_mode);
+    send_message_to_rx(msg);
+}
+
+RxPacketBuffer *
+TrexStatelessPort::get_rx_sw_pkts() {
+
+    /* ask RX core for the pkt queue */
+    TrexStatelessMsgReply<RxPacketBuffer *> msg_reply;
+
+    TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxSwGetPkts(m_port_id, msg_reply);
+    send_message_to_rx(msg);
+
+    RxPacketBuffer *pkt_buffer = msg_reply.wait_for_reply();
+    return pkt_buffer;
+}
+
 /************* Trex Port Owner **************/
 
 TrexPortOwner::TrexPortOwner() {
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index e2a2aeba..41453caa 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -24,12 +24,14 @@ limitations under the License.
 #include "common/basic_utils.h"
 #include "internal_api/trex_platform_api.h"
 #include "trex_dp_port_events.h"
+#include "trex_stateless_rx_defs.h"
 #include "trex_stream.h"
 
 class TrexStatelessCpToDpMsgBase;
 class TrexStatelessCpToRxMsgBase;
 class TrexStreamsGraphObj;
 class TrexPortMultiplier;
+class RxPacketBuffer;
 
 /**
  * TRex port owner can perform
@@ -369,6 +371,17 @@ public:
 
     void get_pci_info(std::string &pci_addr, int &numa_node);
 
+    /**
+	 *  set RX filter mode
+	 *  can be hardware or software
+     */
+    void set_rx_filter_mode(rx_filter_mode_e);
+
+    /**
+     * fetch the RX software packets from the queue
+     * 
+     */
+    RxPacketBuffer *get_rx_sw_pkts();
 
 private:
 
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 95613b41..10b4355b 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -255,3 +255,25 @@ bool TrexStatelessRxQuit::handle (CRxCoreStateless *rx_core) {
     rx_core->quit();
     return true;
 }
+
+
+TrexStatelessRxSwGetPkts::TrexStatelessRxSwGetPkts(uint8_t port_id, TrexStatelessMsgReply<RxPacketBuffer *> &reply) :  m_reply(reply) {
+    m_port_id = port_id;
+}
+
+
+bool
+TrexStatelessRxSetFilterMode::handle(CRxCoreStateless *rx_core) {
+    rx_core->set_rx_filter_mode(m_port_id, m_filter_mode);
+
+    return true;
+}
+
+
+bool TrexStatelessRxSwGetPkts::handle(CRxCoreStateless *rx_core) {
+    RxPacketBuffer *pkt_buffer = rx_core->get_rx_sw_pkt_buffer(m_port_id);
+    assert(pkt_buffer);
+    m_reply.set(pkt_buffer);
+
+    return true;
+}
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index fb2c27ab..42a422b0 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -24,11 +24,15 @@ limitations under the License.
 
 #include "msg_manager.h"
 #include "trex_dp_port_events.h"
+#include "trex_exception.h"
+#include "trex_stateless_rx_defs.h"
+#include "os_time.h"
 
 class TrexStatelessDpCore;
 class CRxCoreStateless;
 class TrexStreamsCompiledObj;
 class CFlowGenListPerThread;
+class RxPacketBuffer;
 
 /**
  * defines the base class for CP to DP messages
@@ -312,7 +316,7 @@ private:
 /************************* messages from DP to CP **********************/
 
 /**
- * defines the base class for CP to DP messages
+ * defines the base class for DP to CP messages
  *
  * @author imarom (27-Oct-15)
  */
@@ -416,4 +420,76 @@ class TrexStatelessRxQuit : public TrexStatelessCpToRxMsgBase {
     bool handle (CRxCoreStateless *rx_core);
 };
 
+
+class TrexStatelessRxSetFilterMode : public TrexStatelessCpToRxMsgBase {
+public:
+    TrexStatelessRxSetFilterMode(uint8_t port_id, rx_filter_mode_e filter_mode) {
+        m_port_id = port_id;
+        m_filter_mode = filter_mode;
+    }
+    virtual bool handle(CRxCoreStateless *rx_core);
+
+private:
+    uint8_t           m_port_id;
+    rx_filter_mode_e  m_filter_mode;
+};
+
+
+template<typename T> class TrexStatelessMsgReply {
+public:
+    TrexStatelessMsgReply() {
+        m_pending = true;
+    }
+
+    bool is_pending() const {
+        return m_pending;
+    }
+
+    void set(T reply) {
+        m_reply = reply;
+
+        /* before marking as done - memory fence */
+        asm volatile("mfence" ::: "memory");
+        m_pending = false;
+    }
+
+    T wait_for_reply(int timeout_ms = 100, int backoff_ms = 1) {
+        int guard = timeout_ms;
+
+        while (is_pending()) {
+            guard -= backoff_ms;
+            if (guard < 0) {
+                throw TrexException("timeout: RX core has failed to reply");
+            }
+
+            delay(backoff_ms);
+            
+        }
+        return m_reply;
+
+    }
+private:
+    bool  m_pending;
+    T     m_reply;
+};
+
+
+
+class TrexStatelessRxSwGetPkts : public TrexStatelessCpToRxMsgBase {
+public:
+
+    TrexStatelessRxSwGetPkts(uint8_t port_id, TrexStatelessMsgReply<RxPacketBuffer *> &reply);
+
+    /**
+     * virtual function to handle a message
+     *
+     */
+    virtual bool handle(CRxCoreStateless *rx_core);
+
+private:
+    uint8_t                                    m_port_id;
+    TrexStatelessMsgReply<RxPacketBuffer*>    &m_reply;
+};
+
+
 #endif /* __TREX_STATELESS_MESSAGING_H__ */
diff --git a/src/stateless/rx/trex_stateless_rx_core.cpp b/src/stateless/rx/trex_stateless_rx_core.cpp
index d162c5b3..e8d5857e 100644
--- a/src/stateless/rx/trex_stateless_rx_core.cpp
+++ b/src/stateless/rx/trex_stateless_rx_core.cpp
@@ -26,6 +26,8 @@
 #include "pal/linux/sanb_atomic.h"
 #include "trex_stateless_messaging.h"
 #include "trex_stateless_rx_core.h"
+#include "trex_stateless.h"
+
 
 void CRFC2544Info::create() {
     m_latency.Create();
@@ -128,6 +130,7 @@ bool CRxCoreStateless::periodic_check_for_cp_messages() {
 
 }
 
+
 void CRxCoreStateless::idle_state_loop() {
     const int SHORT_DELAY_MS    = 2;
     const int LONG_DELAY_MS     = 50;
@@ -371,6 +374,7 @@ void CRxCoreStateless::flush_rx() {
     }// all ports
 }
 
+
 int CRxCoreStateless::try_rx() {
     rte_mbuf_t * rx_pkts[64];
     int i, total_pkts = 0;
@@ -385,7 +389,8 @@ int CRxCoreStateless::try_rx() {
             int j;
             for (j = 0; j < cnt_p; j++) {
                 m = rx_pkts[j];
-                handle_rx_pkt(lp, m);
+                //handle_rx_pkt(lp, m);
+                m_rx_port_mngr[i].handle_pkt(m);
                 rte_pktmbuf_free(m);
             }
             /* commit only if there was work to do ! */
@@ -472,3 +477,11 @@ void CRxCoreStateless::update_cpu_util(){
 double CRxCoreStateless::get_cpu_util() {
     return m_cpu_cp_u.GetVal();
 }
+
+
+void
+CRxCoreStateless::set_rx_filter_mode(uint8_t port_id, rx_filter_mode_e filter_mode) {
+    const TrexPlatformApi *api = get_stateless_obj()->get_platform_api();
+    api->set_rx_filter_mode(port_id, filter_mode);
+}
+
diff --git a/src/stateless/rx/trex_stateless_rx_core.h b/src/stateless/rx/trex_stateless_rx_core.h
index 3f9fb6cc..294c7527 100644
--- a/src/stateless/rx/trex_stateless_rx_core.h
+++ b/src/stateless/rx/trex_stateless_rx_core.h
@@ -25,6 +25,7 @@
 #include "os_time.h"
 #include "pal/linux/sanb_atomic.h"
 #include "utl_cpuu.h"
+#include "trex_stateless_rx_port_mngr.h"
 
 class TrexStatelessCpToRxMsgBase;
 
@@ -140,6 +141,18 @@ class CRxCoreStateless {
     double get_cpu_util();
     void update_cpu_util();
 
+    RxPacketBuffer *get_rx_sw_pkt_buffer(uint8_t port_id) {
+        return m_rx_port_mngr[port_id].get_pkt_buffer();
+    }
+
+    /**
+     * sets the port filter mode
+     * 
+     * @author imarom (10/31/2016)
+     * 
+     * @param filter_mode 
+     */
+    void set_rx_filter_mode(uint8_t port_id, rx_filter_mode_e filter_mode);
 
  private:
     void handle_cp_msg(TrexStatelessCpToRxMsgBase *msg);
@@ -147,6 +160,7 @@ class CRxCoreStateless {
     void tickle();
     void idle_state_loop();
     void handle_rx_pkt(CLatencyManagerPerPortStl * lp, rte_mbuf_t * m);
+    void handle_rx_pkt_2(int port_id, rte_mbuf_t *m);
     void capture_pkt(rte_mbuf_t *m);
     void handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r);
     void flush_rx();
@@ -171,5 +185,7 @@ class CRxCoreStateless {
     volatile bool m_ack_start_work_msg __rte_cache_aligned;
     CRxCoreErrCntrs m_err_cntrs;
     CRFC2544Info m_rfc2544[MAX_FLOW_STATS_PAYLOAD];
+
+    RXPortManager m_rx_port_mngr[TREX_MAX_PORTS];
 };
 #endif