first integration with DPDK layer

7 files changed, 211 insertions, 52 deletions

diff --git a/linux_dpdk/ws_main.py b/linux_dpdk/ws_main.py
index 6aad508a..b1fd8761 100755
--- a/linux_dpdk/ws_main.py
+++ b/linux_dpdk/ws_main.py
@@ -141,6 +141,7 @@ rpc_server_src = SrcGroup(dir='src/rpc-server/',
                           src_list=[
                               'trex_rpc_server.cpp',
                               'trex_rpc_req_resp_server.cpp',
+                              'trex_rpc_async_server.cpp',
                               'trex_rpc_jsonrpc_v2_parser.cpp',
                               'trex_rpc_cmds_table.cpp',
                               'trex_rpc_cmd.cpp',
@@ -148,8 +149,16 @@ rpc_server_src = SrcGroup(dir='src/rpc-server/',
                               'commands/trex_rpc_cmd_test.cpp',
                               'commands/trex_rpc_cmd_general.cpp',
                               'commands/trex_rpc_cmd_stream.cpp',
+
                           ])
 
+# stateless code
+stateless_src = SrcGroup(dir='src/stateless/cp/',
+                          src_list=['trex_stream.cpp',
+                                    'trex_stream_vm.cpp',
+                                    'trex_stateless.cpp',
+                                    'trex_stateless_port.cpp'
+                                    ])
 # JSON package
 json_src = SrcGroup(dir='external_libs/json',
                     src_list=[
@@ -346,6 +355,9 @@ bp =SrcGroups([
                 cmn_src ,
                 net_src ,
                 yaml_src,
+                rpc_server_src,
+                json_src,
+                stateless_src,
                 version_src
                 ]);
 
@@ -400,7 +412,7 @@ includes_path =''' ../src/pal/linux_dpdk/
                    ../src/
                    
                    ../src/rpc-server/
-                   ../src/stateless/
+                   ../src/stateless/cp/
 
                    ../external_libs/yaml-cpp/include/
                    ../external_libs/zmq/include/
diff --git a/src/bp_sim.h b/src/bp_sim.h
index 29b9a724..2f445cb7 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -688,6 +688,15 @@ public:
         RUN_FLAGS_RXCHECK_CONST_TS =1,
     };
 
+    /**
+     * different running modes for Trex
+     */
+    enum trex_run_mode_e {
+        RUN_MODE_INVALID,
+        RUN_MODE_BATCH,
+        RUN_MODE_INTERACTIVE
+    };
+
 public:
     CParserOption(){
         m_factor=1.0;
@@ -707,6 +716,7 @@ public:
         m_run_flags=0;
         prefix="";
         m_mac_splitter=0;
+        m_run_mode = RUN_MODE_INVALID;
     }
 
     CPreviewMode    preview;
@@ -730,13 +740,14 @@ public:
     uint8_t         m_mac_splitter;
     uint8_t         m_pad;
 
+    trex_run_mode_e    m_run_mode;
 
-    std::string     cfg_file;
-    std::string     mac_file;
-    std::string     platform_cfg_file;
+    std::string        cfg_file;
+    std::string        mac_file;
+    std::string        platform_cfg_file;
 
-    std::string     out_file;
-    std::string     prefix;
+    std::string        out_file;
+    std::string        prefix;
 
                                  
     CMacAddrCfg     m_mac_addr[MAX_LATENCY_PORTS];
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 820fb3fa..e3d13f2b 100755
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -55,6 +55,7 @@ limitations under the License.
 #include <common/arg/SimpleGlob.h>
 #include <common/arg/SimpleOpt.h>
 #include <common/basic_utils.h>
+#include <stateless/cp/trex_stateless.h>
 #include <../linux_dpdk/version.h>
 
 extern "C" {
@@ -426,7 +427,8 @@ static char global_loglevel_str[20];
 // cores =0==1,1*2,2,3,4,5,6
 // An enum for all the option types
 enum { OPT_HELP, 
-    OPT_CFG, 
+    OPT_MODE_BATCH, 
+    OPT_MODE_INTERACTIVE,
     OPT_NODE_DUMP,  
     OPT_UT,
     OPT_FILE_OUT,
@@ -478,15 +480,16 @@ enum { OPT_HELP,
 */
 static CSimpleOpt::SOption parser_options[] =
 {
-    { OPT_HELP,     "-?",           SO_NONE   },
-    { OPT_HELP,     "-h",           SO_NONE   },
-    { OPT_HELP,     "--help",       SO_NONE   },
-    { OPT_UT,       "--ut",         SO_NONE   },
-    { OPT_CFG,       "-f",          SO_REQ_SEP},
-    { OPT_PLAT_CFG_FILE,"--cfg", SO_REQ_SEP},
-    { OPT_REAL_TIME , "-r",         SO_NONE  },
-    { OPT_SINGLE_CORE     , "-s",   SO_NONE  },
-    { OPT_FILE_OUT   , "-o" ,        SO_REQ_SEP},
+    { OPT_HELP,                   "-?",                SO_NONE   },
+    { OPT_HELP,                   "-h",                SO_NONE   },
+    { OPT_HELP,                   "--help",            SO_NONE   },
+    { OPT_UT,                     "--ut",              SO_NONE   },
+    { OPT_MODE_BATCH,             "-f",                SO_REQ_SEP},
+    { OPT_MODE_INTERACTIVE,       "-i",                SO_NONE   },
+    { OPT_PLAT_CFG_FILE,          "--cfg",             SO_REQ_SEP},
+    { OPT_REAL_TIME ,             "-r",                SO_NONE  },
+    { OPT_SINGLE_CORE,            "-s",                SO_NONE  },
+    { OPT_FILE_OUT,               "-o" ,               SO_REQ_SEP},
     { OPT_FLIP_CLIENT_SERVER,"--flip",SO_NONE  },
     { OPT_FLOW_FLIP_CLIENT_SERVER,"-p",SO_NONE  },
     { OPT_FLOW_FLIP_CLIENT_SERVER_SIDE,"-e",SO_NONE  },
@@ -533,13 +536,18 @@ static CSimpleOpt::SOption parser_options[] =
 
 static int usage(){
 
-    printf(" Usage: t-rex-64 [OPTION] -f cfg.yaml -c cores   \n");
+    printf(" Usage: t-rex-64 [MODE] [OPTION] -f cfg.yaml -c cores   \n");
     printf(" \n");
     printf(" \n");
-    printf(" options \n");
+    
+    printf(" mode \n\n");
     printf(" -f [file]                  : YAML file  with template configuration \n");
+    printf(" -i                         : launch TRex in interactive mode (RPC server)\n");
     printf(" \n\n");
-    printf(" --mac [file]                  : YAML file with <client ip, mac addr> configuration \n");
+
+    printf(" options \n\n");
+
+    printf(" --mac [file]               : YAML file with <client ip, mac addr> configuration \n");
     printf(" \n\n");
     printf(" -r                         : realtime enable \n");
     printf(" \n\n");
@@ -612,7 +620,8 @@ static int usage(){
     printf(" --mac-spread              : Spread the destination mac-order by this factor. e.g 2 will generate the traffic to 2 devices DEST-MAC ,DEST-MAC+1  \n");
     printf("                             maximum is up to 128 devices   \n");
     
-    printf(" simulation mode : \n");
+    
+    printf("\n simulation mode : \n");
     printf(" Using this mode you can generate the traffic into a pcap file and learn how trex works \n");
     printf(" With this version you must be SUDO to use this mode ( I know this is not normal )  \n");
     printf(" you can use the Linux CEL version of t-rex to do it without super user   \n");
@@ -653,6 +662,7 @@ static int usage(){
     printf(" Open Source Components / Libraries \n");
     printf(" DPDK       (BSD)         \n");
     printf(" YAML-CPP   (BSD)       \n");
+    printf(" JSONCPP    (MIT)       \n");
     printf(" \n");
     printf(" Open Source Binaries \n");
     printf(" ZMQ        (LGPL v3plus) \n");
@@ -667,6 +677,11 @@ static int usage(){
 
 int gtest_main(int argc, char **argv) ;
 
+static void parse_err(const std::string &msg) {
+    std::cout << "\nArgument Parsing Error: \n\n" << "*** "<< msg << "\n\n";
+    exit(-1);
+}
+
 static int parse_options(int argc, char *argv[], CParserOption* po, bool first_time ) {
      CSimpleOpt args(argc, argv, parser_options);
 
@@ -679,36 +694,55 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
      int res1;
      uint32_t tmp_data;
 
+     po->m_run_mode = CParserOption::RUN_MODE_INVALID;
 
      while ( args.Next() ){
         if (args.LastError() == SO_SUCCESS) {
             switch (args.OptionId()) {
+
             case OPT_UT :
-                printf(" Supported only in simulation \n");
-                res1=0;
-                exit(res1);
+                parse_err("Supported only in simulation");
                 break;
+
             case OPT_HELP: 
                 usage();
                 return -1;
-            case OPT_CFG:
+
+            case OPT_MODE_BATCH:
+                if (po->m_run_mode != CParserOption::RUN_MODE_INVALID) {
+                    parse_err("Please specify single run mode");
+                }
+                po->m_run_mode = CParserOption::RUN_MODE_BATCH;
                 po->cfg_file = args.OptionArg();
                 break;
+
+            case OPT_MODE_INTERACTIVE:
+                if (po->m_run_mode != CParserOption::RUN_MODE_INVALID) {
+                    parse_err("Please specify single run mode");
+                }
+                po->m_run_mode = CParserOption::RUN_MODE_INTERACTIVE;
+                break;
+
             case OPT_NO_KEYBOARD_INPUT  :
                 po->preview.set_no_keyboard(true);
                 break;
+
             case OPT_MAC_FILE :
                 po->mac_file = args.OptionArg();
                 break;
+
             case OPT_PLAT_CFG_FILE :
                 po->platform_cfg_file = args.OptionArg();
                 break;
+
             case OPT_SINGLE_CORE :
                 po->preview.setSingleCore(true);
                 break;
+
             case OPT_IPV6:
                 po->preview.set_ipv6_mode_enable(true);
                 break;
+
             case OPT_VLAN:
                 po->preview.set_vlan_mode_enable(true);
                 break;
@@ -726,6 +760,7 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
                 printf(" warning -r is deprecated, real time is not needed any more , it is the default \n");
                 po->preview.setRealTime(true);
                 break;
+
             case OPT_NO_FLOW_CONTROL:
                 po->preview.set_disable_flow_control_setting(true);
                 break;
@@ -832,21 +867,20 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t
      } // End of while
 
 
-    if ((po->cfg_file =="") ) {
-         printf("Invalid combination of parameters you must add -f with configuration file \n");
-         return -1;
+    if ((po->m_run_mode ==  CParserOption::RUN_MODE_INVALID) ) {
+        parse_err("Please provide single run mode (e.g. batch or interactive)");
      }
 
     if ( po->m_mac_splitter > 128 ){
-       printf("maximum mac spreading is 128 you set it to %d \n",po->m_mac_splitter);
-       return -1;
+        std::stringstream ss;
+        ss << "maximum mac spreading is 128 you set it to: " << po->m_mac_splitter;
+        parse_err(ss.str());
     }
 
     if ( po->preview.get_learn_mode_enable()  ){
         if  ( po->preview.get_ipv6_mode_enable() ){
-            printf("--learn mode is not supported with --ipv6, beacuse there is not such thing NAT66 ( ipv6-ipv6) \n");
-            printf("if you think it is important,open a defect \n");
-            return -1;
+            parse_err("--learn mode is not supported with --ipv6, beacuse there is not such thing NAT66 ( ipv6-ipv6) \n" \
+                      "if you think it is important,open a defect \n");
         }
         if ( po->is_latency_disabled() ){
             /* set latency thread */
@@ -3863,6 +3897,7 @@ int CGlobalPortCfg::run_in_master(){
 
     std::string json;
     bool was_stopped=false;
+
     while ( true ) {
 
         if ( CGlobalInfo::m_options.preview.get_no_keyboard() ==false ){
@@ -3976,6 +4011,7 @@ int CGlobalPortCfg::run_in_master(){
             break;
         }
     }
+
     m_mg.stop();
     delay(1000);
     if ( was_stopped ){
@@ -4172,7 +4208,18 @@ static int latency_one_lcore(__attribute__((unused)) void *dummy)
 }
 
 
+static int stateless_entry(__attribute__((unused)) void *dummy) {
+    CPlatformSocketInfo * lpsock=&CGlobalInfo::m_socket;
+    physical_thread_id_t phy_id = rte_lcore_id();
+
+    if (lpsock->thread_phy_is_master( phy_id )) {
+        TrexStateless::get_instance().launch_control_plane();
+    } else {
+        TrexStateless::get_instance().launch_on_dp_core();
+    }
 
+    return (0);
+}
 
 static int slave_one_lcore(__attribute__((unused)) void *dummy)
 {
@@ -4381,7 +4428,37 @@ int sim_load_list_of_cap_files(CParserOption * op){
 
 
 
+static int
+launch_stateless_trex() {
+    CPlatformSocketInfo *lpsock=&CGlobalInfo::m_socket;
+    CParserOption       *lpop= &CGlobalInfo::m_options;
+    CPlatformYamlInfo   *cg=&global_platform_cfg_info;
+
+    TrexStatelessCfg cfg;
+
+    TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_TCP, 5050);
+    TrexRpcServerConfig rpc_async_cfg(TrexRpcServerConfig::RPC_PROT_TCP, 5051);
 
+    cfg.m_dp_core_count      = lpop->preview.getCores();
+    cfg.m_port_count         = lpop->m_expected_portd;
+    cfg.m_rpc_req_resp_cfg   = &rpc_req_resp_cfg;
+    cfg.m_rpc_async_cfg      = &rpc_async_cfg;
+    cfg.m_rpc_server_verbose = true;
+
+    TrexStateless::configure(cfg);
+
+    printf("\nStarting T-Rex Stateless\n");
+    printf("Starting RPC Server...\n\n");
+
+    rte_eal_mp_remote_launch(stateless_entry, NULL, CALL_MASTER);
+
+    unsigned lcore_id;
+    RTE_LCORE_FOREACH_SLAVE(lcore_id) {
+        if (rte_eal_wait_lcore(lcore_id) < 0)
+            return -1;
+    }
+    return (0);
+}
 
 
 
@@ -4436,6 +4513,11 @@ int main_test(int argc , char * argv[]){
         rte_exit(EXIT_FAILURE, "Invalid EAL arguments\n");
     }
 
+    /* patch here */
+    if (CGlobalInfo::m_options.m_run_mode == CParserOption::RUN_MODE_INTERACTIVE) {
+        return launch_stateless_trex();
+    }
+
 
     time_init();
     
@@ -4518,6 +4600,7 @@ int main_test(int argc , char * argv[]){
         if (rte_eal_wait_lcore(lcore_id) < 0)
             return -1;
     }
+
     ports_cfg.stop_master();
     ports_cfg.Delete();
     utl_termio_reset();
diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
index b440fa81..ae87d749 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp
@@ -155,7 +155,7 @@ TrexRpcCmdGetSysInfo::_run(const Json::Value &params, Json::Value &result) {
     section["uptime"] = TrexRpcServer::get_server_uptime();
 
     /* FIXME: core count */
-    section["dp_core_count"] = 1;
+    section["dp_core_count"] = instance.get_dp_core_count();
     section["core_type"] = get_cpu_model();
 
     /* ports */
diff --git a/src/rpc-server/trex_rpc_server_mock.cpp b/src/rpc-server/trex_rpc_server_mock.cpp
index 8dae42d7..de43f92f 100644
--- a/src/rpc-server/trex_rpc_server_mock.cpp
+++ b/src/rpc-server/trex_rpc_server_mock.cpp
@@ -62,11 +62,14 @@ int main(int argc, char *argv[]) {
     TrexRpcServerConfig rpc_async_cfg(TrexRpcServerConfig::RPC_PROT_TCP, 5051);
 
     cfg.m_port_count         = 4;
+    cfg.m_dp_core_count      = 2;
     cfg.m_rpc_req_resp_cfg   = &rpc_req_resp_cfg;
     cfg.m_rpc_async_cfg      = &rpc_async_cfg;
     cfg.m_rpc_server_verbose = (is_gtest ? false : true);
 
-    TrexStateless::create(cfg);
+    TrexStateless::configure(cfg);
+
+    TrexStateless::get_instance().launch_control_plane();
 
     /* gtest handling */
     if (is_gtest) {
diff --git a/src/stateless/cp/trex_stateless.cpp b/src/stateless/cp/trex_stateless.cpp
index 1e7e2dfa..872fdd92 100644
--- a/src/stateless/cp/trex_stateless.cpp
+++ b/src/stateless/cp/trex_stateless.cpp
@@ -22,6 +22,8 @@ limitations under the License.
 #include <trex_stateless_port.h>
 
 #include <sched.h>
+#include <iostream>
+#include <unistd.h>
 
 using namespace std;
 
@@ -35,10 +37,10 @@ TrexStateless::TrexStateless() {
 
 
 /**
- * creates the singleton stateless object
+ * configure the singleton stateless object
  * 
  */
-void TrexStateless::create(const TrexStatelessCfg &cfg) {
+void TrexStateless::configure(const TrexStatelessCfg &cfg) {
 
     TrexStateless& instance = get_instance_internal();
 
@@ -47,19 +49,9 @@ void TrexStateless::create(const TrexStatelessCfg &cfg) {
         throw TrexException("re-configuration of stateless object is not allowed");
     }
 
-    /* pin this process to the current running CPU
-       any new thread will be called on the same CPU
-       (control plane restriction)
-     */
-    cpu_set_t mask;
-    CPU_ZERO(&mask);
-    CPU_SET(sched_getcpu(), &mask);
-    sched_setaffinity(0, sizeof(mask), &mask);
-
-    /* start RPC servers */
+    /* create RPC servers */
     instance.m_rpc_server = new TrexRpcServer(cfg.m_rpc_req_resp_cfg, cfg.m_rpc_async_cfg);
     instance.m_rpc_server->set_verbose(cfg.m_rpc_server_verbose);
-    instance.m_rpc_server->start();
 
     /* configure ports */
 
@@ -69,11 +61,44 @@ void TrexStateless::create(const TrexStatelessCfg &cfg) {
         instance.m_ports.push_back(new TrexStatelessPort(i));
     }
 
+    /* cores */
+    instance.m_dp_core_count = cfg.m_dp_core_count;
+
     /* done */
     instance.m_is_configured = true;
 }
 
 /**
+ * starts the control plane side
+ * 
+ */
+void
+TrexStateless::launch_control_plane() {
+    //std::cout << "\n on control/master core \n";
+
+    /* pin this process to the current running CPU
+       any new thread will be called on the same CPU
+       (control plane restriction)
+     */
+    cpu_set_t mask;
+    CPU_ZERO(&mask);
+    CPU_SET(sched_getcpu(), &mask);
+    sched_setaffinity(0, sizeof(mask), &mask);
+
+    /* start RPC server */
+    m_rpc_server->start();
+}
+
+void
+TrexStateless::launch_on_dp_core() {
+    //std::cout << "\n on DP core \n";
+
+    while (true) {
+        sleep(1);
+    }
+}
+
+/**
  * destroy the singleton and release all memory
  * 
  * @author imarom (08-Oct-15)
@@ -115,10 +140,16 @@ TrexStatelessPort * TrexStateless::get_port_by_id(uint8_t port_id) {
 
 }
 
-uint8_t TrexStateless::get_port_count() {
+uint8_t 
+TrexStateless::get_port_count() {
     return m_port_count;
 }
 
+uint8_t 
+TrexStateless::get_dp_core_count() {
+    return m_dp_core_count;
+}
+
 void 
 TrexStateless::update_stats() {
 
diff --git a/src/stateless/cp/trex_stateless.h b/src/stateless/cp/trex_stateless.h
index 02eda7e2..5c674fd6 100644
--- a/src/stateless/cp/trex_stateless.h
+++ b/src/stateless/cp/trex_stateless.h
@@ -83,9 +83,10 @@ class TrexStatelessCfg {
 public:
     /* default values */
     TrexStatelessCfg() {
-        m_port_count = 0;
-        m_rpc_req_resp_cfg = NULL;
-        m_rpc_async_cfg = NULL;
+        m_port_count          = 0;
+        m_dp_core_count       = 0;
+        m_rpc_req_resp_cfg    = NULL;
+        m_rpc_async_cfg       = NULL;
         m_rpc_server_verbose = false;
     }
 
@@ -93,6 +94,7 @@ public:
     const TrexRpcServerConfig  *m_rpc_async_cfg;
     bool                        m_rpc_server_verbose;
     uint8_t                     m_port_count;
+    uint8_t                     m_dp_core_count;
 };
 
 /**
@@ -107,7 +109,7 @@ public:
      * reconfiguration is not allowed
      * an exception will be thrown
      */
-    static void create(const TrexStatelessCfg &cfg);
+    static void configure(const TrexStatelessCfg &cfg);
 
     /**
      * destroy the instance
@@ -129,9 +131,23 @@ public:
         return instance;
     }
 
+    /**
+     * starts the control plane side
+     * 
+     */
+    void launch_control_plane();
+
+    /**
+     * launch on a single DP core
+     * 
+     */
+    void launch_on_dp_core();
+
     TrexStatelessPort * get_port_by_id(uint8_t port_id);
     uint8_t             get_port_count();
 
+    uint8_t             get_dp_core_count();
+
     /**
      * update all the stats (deep update)
      * (include all the ports and global stats)
@@ -168,6 +184,9 @@ protected:
     std::vector <TrexStatelessPort *>   m_ports;
     uint8_t                             m_port_count;
 
+    /* cores */
+    uint8_t m_dp_core_count;
+
     /* stats */
     TrexStatelessStats   m_stats;
 };