6 files changed, 187 insertions, 53 deletions

diff --git a/scripts/automation/trex_control_plane/client/trex_stateless_sim.py b/scripts/automation/trex_control_plane/client/trex_stateless_sim.py
index 7655b27c..5d06c6e5 100644
--- a/scripts/automation/trex_control_plane/client/trex_stateless_sim.py
+++ b/scripts/automation/trex_control_plane/client/trex_stateless_sim.py
@@ -34,11 +34,41 @@ import argparse
 import tempfile
 import subprocess
 import os
+from dpkt import pcap
+from operator import itemgetter
+
+
+def merge_cap_files (pcap_file_list, out_filename, delete_src = False):
+
+    out_pkts = []
+
+    # read all packets to a list
+    for src in pcap_file_list:
+        f = open(src, 'r')
+        reader = pcap.Reader(f)
+        pkts = reader.readpkts()
+        out_pkts += pkts
+        f.close()
+        if delete_src:
+            os.unlink(src)
+
+    # sort by the timestamp
+    out_pkts = sorted(out_pkts, key=itemgetter(0))
+
+
+    out = open(out_filename, 'w')
+    out_writer = pcap.Writer(out)
+
+    for ts, pkt in out_pkts:
+        out_writer.writepkt(pkt, ts)
+
+    out.close()
+
 
 
 
 class SimRun(object):
-    def __init__ (self, yaml_file, dp_core_count, core_index, packet_limit, output_filename, is_valgrind, is_gdb):
+    def __init__ (self, yaml_file, dp_core_count, core_index, packet_limit, output_filename, is_valgrind, is_gdb, limit):
 
         self.yaml_file = yaml_file
         self.output_filename = output_filename
@@ -47,6 +77,7 @@ class SimRun(object):
         self.packet_limit = packet_limit
         self.is_valgrind = is_valgrind
         self.is_gdb = is_gdb
+        self.limit = limit
 
         # dummies
         self.handler = 0
@@ -99,18 +130,47 @@ class SimRun(object):
         f.close()
 
         try:
-            cmd = ['bp-sim-64-debug', '--sl', '--cores', str(self.dp_core_count), '--core_index', str(self.core_index), '-f', f.name, '-o', self.output_filename]
+            cmd = ['bp-sim-64-debug',
+                   '--pcap',
+                   '--sl',
+                   '--cores',
+                   str(self.dp_core_count),
+                   '--limit',
+                   str(self.limit),
+                   '-f',
+                   f.name,
+                   '-o',
+                   self.output_filename]
+
+            if self.core_index != None:
+                cmd += ['--core_index', str(self.core_index)]
+
             if self.is_valgrind:
                 cmd = ['valgrind', '--leak-check=full'] + cmd
             elif self.is_gdb:
                 cmd = ['gdb', '--args'] + cmd
 
+            print "executing command: '{0}'".format(" ".join(cmd))
             subprocess.call(cmd)
 
+            # core index 
+            if (self.dp_core_count > 1) and (self.core_index == None):
+                self.merge_results()
+
         finally:
             os.unlink(f.name)
 
 
+    def merge_results (self):
+        if (self.core_index != None) or (self.dp_core_count == 1):
+            # nothing to do
+            return
+
+        inputs = ["{0}-{1}".format(self.output_filename, index) for index in xrange(0, self.dp_core_count)]
+        merge_cap_files(inputs, self.output_filename, delete_src = True)
+
+
+
 def is_valid_file(filename):
     if not os.path.isfile(filename):
         raise argparse.ArgumentTypeError("The file '%s' does not exist" % filename)
@@ -143,8 +203,8 @@ def setParserOptions():
                         choices = xrange(1, 9))
 
     parser.add_argument("-n", "--core_index",
-                        help = "DP core index to examine [default is 0]",
-                        default = 0,
+                        help = "Record only a specific core",
+                        default = None,
                         type = int)
 
     parser.add_argument("-j", "--join",
@@ -174,8 +234,10 @@ def setParserOptions():
 
 
 def validate_args (parser, options):
-    if options.core_index < 0 or options.core_index >= options.cores:
-        parser.error("DP core index valid range is 0 to {0}".format(options.cores - 1))
+
+    if options.core_index:
+        if not options.core_index in xrange(0, options.cores):
+            parser.error("DP core index valid range is 0 to {0}".format(options.cores - 1))
 
 
 
@@ -191,7 +253,8 @@ def main ():
                options.limit,
                options.output_file,
                options.valgrind,
-               options.gdb)
+               options.gdb,
+               options.limit)
 
     r.run()
 
diff --git a/src/bp_sim.cpp b/src/bp_sim.cpp
index 6342b7f3..d08e61c8 100755
--- a/src/bp_sim.cpp
+++ b/src/bp_sim.cpp
@@ -3146,7 +3146,8 @@ int CNodeGenerator::open_file(std::string file_name,
     /* ser preview mode */
     m_v_if->set_review_mode(preview_mode);
     m_v_if->open_file(file_name);
-    m_cnt = 0;
+    m_cnt   = 0;
+    m_limit = 0;
     return (0);
 }
 
@@ -3179,6 +3180,10 @@ int CNodeGenerator::update_stats(CGenNode * node){
     return (0);
 }
 
+bool CNodeGenerator::has_limit_reached() {
+    /* do we have a limit and has it passed ? */
+    return ( (m_limit > 0) && (m_cnt >= m_limit) );
+}
 
 bool CFlowGenListPerThread::Create(uint32_t           thread_id,
                                    uint32_t           core_id,
@@ -3524,16 +3529,21 @@ int CNodeGenerator::flush_file(dsec_t max_time,
              m_p_queue.pop();
              CGenNodeStateless *node_sl = (CGenNodeStateless *)node;
 
-             #ifdef _DEBUG
-             update_stl_stats(node_sl);
-             #endif
-
              /* if the stream has been deactivated - end */
              if ( unlikely( node_sl->is_mask_for_free() ) ) {
                  thread->free_node(node);
              } else {
                  node_sl->handle(thread);
+
+                 #ifdef _DEBUG
+                 update_stl_stats(node_sl);
+                 if (has_limit_reached()) {
+                     thread->m_stateless_dp_info.stop_traffic(node_sl->get_port_id(), false, 0);
+                 }
+                 #endif
+
              }
+         
             
         }else{
             if ( likely( type == CGenNode::FLOW_PKT ) ) {
@@ -3974,9 +3984,11 @@ void CFlowGenListPerThread::check_msgs(void) {
 
 
 void CFlowGenListPerThread::start_stateless_simulation_file(std::string erf_file_name,
-                                     CPreviewMode &preview){
+                                                            CPreviewMode &preview,
+                                                            uint64_t limit){
     m_preview_mode = preview;
     m_node_gen.open_file(erf_file_name,&m_preview_mode);
+    m_node_gen.set_packet_limit(limit);
 }
 
 void CFlowGenListPerThread::stop_stateless_simulation_file(){
@@ -3987,7 +3999,6 @@ void CFlowGenListPerThread::start_stateless_daemon_simulation(){
 
     m_cur_time_sec = 0;
     m_stateless_dp_info.run_once();
-
 }
 
 
diff --git a/src/bp_sim.h b/src/bp_sim.h
index da8e8780..e0792336 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -1929,6 +1929,12 @@ public:
         add_node(node);
     }
 
+    /**
+     * set packet limit for the generator
+     */
+    void set_packet_limit(uint64_t limit) {
+        m_limit = limit;
+    }
 
     void DumpHist(FILE *fd){
         fprintf(fd,"\n");
@@ -1947,7 +1953,7 @@ private:
     }
     int   update_stats(CGenNode * node);
     int   update_stl_stats(CGenNodeStateless *node_sl);
-
+    bool  has_limit_reached();
 
     FORCE_NO_INLINE bool handle_slow_messages(uint8_t type,
                                               CGenNode * node,
@@ -1963,6 +1969,7 @@ public:
     CFlowGenListPerThread  *  m_parent;
     CPreviewMode              m_preview_mode;
     uint64_t                  m_cnt;
+    uint64_t                  m_limit;
     CTimeHistogram            m_realtime_his;  
 };
 
@@ -3477,7 +3484,7 @@ public:
     void start_stateless_daemon_simulation();
 
     /* open a file for simulation */
-    void start_stateless_simulation_file(std::string erf_file_name,CPreviewMode &preview);
+    void start_stateless_simulation_file(std::string erf_file_name,CPreviewMode &preview, uint64_t limit = 0);
     /* close a file for simulation */
     void stop_stateless_simulation_file();
 
diff --git a/src/main.cpp b/src/main.cpp
index ba6e258a..ea8e1e44 100755
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -23,6 +23,7 @@ limitations under the License.
 #include "bp_sim.h"
 #include "os_time.h"
 
+#include <unordered_map>
 #include <string>
 
 #include <common/arg/SimpleGlob.h>
@@ -35,7 +36,7 @@ using namespace std;
 // An enum for all the option types
 enum { OPT_HELP, OPT_CFG, OPT_NODE_DUMP, OP_STATS,
        OPT_FILE_OUT, OPT_UT, OPT_PCAP, OPT_IPV6, OPT_MAC_FILE,
-       OPT_SL, OPT_DP_CORE_COUNT, OPT_DP_CORE_INDEX};
+       OPT_SL, OPT_DP_CORE_COUNT, OPT_DP_CORE_INDEX, OPT_LIMIT};
 
 
 
@@ -73,13 +74,16 @@ static CSimpleOpt::SOption parser_options[] =
     { OPT_SL,            "--sl",         SO_NONE    },
     { OPT_DP_CORE_COUNT, "--cores",      SO_REQ_SEP },
     { OPT_DP_CORE_INDEX, "--core_index", SO_REQ_SEP },
+    { OPT_LIMIT,         "--limit",      SO_REQ_SEP },
 
     
     SO_END_OF_OPTIONS
 };
 
 
-
+static bool in_range(int x, int low, int high) {
+    return ( (x >= low) && (x <= high) );
+}
 
 static int usage(){
 
@@ -119,9 +123,7 @@ static int usage(){
 static int parse_options(int argc,
                          char *argv[],
                          CParserOption* po,
-                         opt_type_e &type,
-                         int &dp_core_count,
-                         int &dp_core_index) {
+                         std::unordered_map<std::string, int> &params) {
 
      CSimpleOpt args(argc, argv, parser_options);
 
@@ -131,16 +133,13 @@ static int parse_options(int argc,
      po->preview.setFileWrite(true);
 
      /* by default - type is stateful */
-     type = OPT_TYPE_SF;
-
-     dp_core_count = 1;
-     dp_core_index = 0;
+     params["type"] = OPT_TYPE_SF;
 
      while ( args.Next() ){
         if (args.LastError() == SO_SUCCESS) {
             switch (args.OptionId()) {
             case OPT_UT :
-                type = OPT_TYPE_GTEST;
+                params["type"] = OPT_TYPE_GTEST;
                 return (0);
                 break;
 
@@ -149,7 +148,7 @@ static int parse_options(int argc,
                 return -1;
 
             case OPT_SL:
-                type = OPT_TYPE_SL;
+                params["type"] = OPT_TYPE_SL;
                 break;
 
             case OPT_CFG:
@@ -179,11 +178,15 @@ static int parse_options(int argc,
                 break;
 
             case OPT_DP_CORE_COUNT:
-                dp_core_count = atoi(args.OptionArg());
+                params["dp_core_count"] = atoi(args.OptionArg());
                 break;
 
             case OPT_DP_CORE_INDEX:
-                dp_core_index = atoi(args.OptionArg());
+                params["dp_core_index"] = atoi(args.OptionArg());
+                break;
+
+            case OPT_LIMIT:
+                params["limit"] = atoi(args.OptionArg());
                 break;
 
             default:
@@ -215,16 +218,18 @@ static int parse_options(int argc,
         }
     }
 
-    if (dp_core_count != -1) {
-        if ( (dp_core_count < 1) || (dp_core_count > 8) ) {
+    /* did the user configure dp core count or dp core index ? */
+
+    if (params.count("dp_core_count") > 0) {
+        if (!in_range(params["dp_core_count"], 1, 8)) {
             printf("dp core count must be a value between 1 and 8\n");
             return (-1);
         }
     }
 
-     if (dp_core_index != -1) {
-        if ( (dp_core_index < 0) || (dp_core_index >= dp_core_count) ) {
-            printf("dp core count must be a value between 0 and cores - 1\n");
+     if (params.count("dp_core_index") > 0) {
+        if (!in_range(params["dp_core_index"], 0, params["dp_core_count"] - 1)) {
+            printf("dp core index must be a value between 0 and cores - 1\n");
             return (-1);
         }
     }
@@ -235,14 +240,14 @@ static int parse_options(int argc,
 
 int main(int argc , char * argv[]){
 
-    opt_type_e type;
-    int dp_core_count;
-    int dp_core_index;
+    std::unordered_map<std::string, int> params;
 
-    if ( parse_options(argc, argv, &CGlobalInfo::m_options , type, dp_core_count, dp_core_index) != 0) {
+    if ( parse_options(argc, argv, &CGlobalInfo::m_options , params) != 0) {
         exit(-1);
     }
 
+    opt_type_e type = (opt_type_e) params["type"];
+
     switch (type) {
     case OPT_TYPE_GTEST:
         {
@@ -259,7 +264,25 @@ int main(int argc , char * argv[]){
     case OPT_TYPE_SL:
         {
             SimStateless &st = SimStateless::get_instance();
-            return st.run(CGlobalInfo::m_options.cfg_file, CGlobalInfo::m_options.out_file, 2, dp_core_count, dp_core_index);
+
+            if (params.count("dp_core_count") == 0) {
+                params["dp_core_count"] = 1;
+            }
+
+            if (params.count("dp_core_index") == 0) {
+                params["dp_core_index"] = -1;
+            }
+
+            if (params.count("limit") == 0) {
+                params["limit"] = 5000;
+            }
+
+            return st.run(CGlobalInfo::m_options.cfg_file,
+                          CGlobalInfo::m_options.out_file,
+                          2,
+                          params["dp_core_count"],
+                          params["dp_core_index"],
+                          params["limit"]);
         }
     }
 }
diff --git a/src/sim/trex_sim.h b/src/sim/trex_sim.h
index cc02fd75..a541ce01 100644
--- a/src/sim/trex_sim.h
+++ b/src/sim/trex_sim.h
@@ -102,7 +102,8 @@ public:
             const std::string &out_filename,
             int port_count,
             int dp_core_count,
-            int dp_core_index);
+            int dp_core_index,
+            int limit);
 
     TrexStateless * get_stateless_obj() {
         return m_trex_stateless;
@@ -121,7 +122,7 @@ private:
     void execute_json(const std::string &json_filename);
 
     void run_dp(const std::string &out_filename);
-    void run_dp_core(int core_index, const std::string &out_filename);
+    uint64_t run_dp_core(int core_index, const std::string &out_filename);
 
     void flush_dp_to_cp_messages_core(int core_index);
 
@@ -141,6 +142,7 @@ private:
     int              m_port_count;
     int              m_dp_core_count;
     int              m_dp_core_index;
+    uint64_t         m_limit;
 };
 
 #endif /* __TREX_SIM_H__ */
diff --git a/src/sim/trex_sim_stateless.cpp b/src/sim/trex_sim_stateless.cpp
index 2821644f..2b73f686 100644
--- a/src/sim/trex_sim_stateless.cpp
+++ b/src/sim/trex_sim_stateless.cpp
@@ -120,6 +120,7 @@ SimStateless::SimStateless() {
     m_dp_core_count     = -1;
     m_dp_core_index     = -1;
     m_port_count        = -1;
+    m_limit             = 0;
 
     /* override ownership checks */
     TrexRpcCommand::test_set_override_ownership(true);
@@ -131,15 +132,18 @@ SimStateless::run(const string &json_filename,
                   const string &out_filename,
                   int port_count,
                   int dp_core_count,
-                  int dp_core_index) {
+                  int dp_core_index,
+                  int limit) {
 
     assert(dp_core_count > 0);
-    assert(dp_core_index >= 0);
-    assert(dp_core_index < dp_core_count);
+
+    /* -1 means its not set or positive value between 0 and the dp core count - 1*/
+    assert( (dp_core_index == -1) || ( (dp_core_index >=0 ) && (dp_core_index < dp_core_count) ) );
 
     m_dp_core_count = dp_core_count;
     m_dp_core_index = dp_core_index;
     m_port_count    = port_count;
+    m_limit         = limit;
 
     prepare_dataplane();
     prepare_control_plane();
@@ -266,33 +270,57 @@ SimStateless::validate_response(const Json::Value &resp) {
 
 void
 SimStateless::run_dp(const std::string &out_filename) {
+    uint64_t pkt_cnt = 0;
 
-    for (int i = 0; i < m_dp_core_count; i++) {
-        if (i == m_dp_core_index) {
-            run_dp_core(i, out_filename);
+    if (m_dp_core_count == 1) {
+        pkt_cnt = run_dp_core(0, out_filename);
+    } else {
+
+        /* do we have a specific core index to capture ? */
+        if (m_dp_core_index != -1) {
+            for (int i = 0; i < m_dp_core_count; i++) {
+                if (i == m_dp_core_index) {
+                    pkt_cnt += run_dp_core(i, out_filename);
+                } else {
+                    run_dp_core(i, "/dev/null");
+                }
+            }
         } else {
-            run_dp_core(i, "/dev/null");
+            for (int i = 0; i < m_dp_core_count; i++) {
+                std::stringstream ss;
+                ss << out_filename << "-" << i;
+                pkt_cnt += run_dp_core(i, ss.str());
+            }
         }
+
     }
 
-    CFlowGenListPerThread *lpt = m_fl.m_threads_info[m_dp_core_index];
   
     std::cout << "\n";
     std::cout << "ports:        " << m_port_count << "\n";
     std::cout << "cores:        " << m_dp_core_count << "\n";
-    std::cout << "core index:   " << m_dp_core_index << "\n";
-    std::cout << "\nwritten " << lpt->m_node_gen.m_cnt << " packets " << "to '" << out_filename << "'\n\n";
+
+    if (m_dp_core_index != -1) {
+        std::cout << "core index:   " << m_dp_core_index << "\n";
+    } else {
+        std::cout << "core index:   merge all\n";
+    }
+
+    std::cout << "pkt limit:    " << m_limit << "\n";
+    std::cout << "\nwritten " << pkt_cnt << " packets " << "to '" << out_filename << "'\n\n";
 }
 
-void
+uint64_t
 SimStateless::run_dp_core(int core_index, const std::string &out_filename) {
 
     CFlowGenListPerThread *lpt = m_fl.m_threads_info[core_index];
 
-    lpt->start_stateless_simulation_file((std::string)out_filename, CGlobalInfo::m_options.preview);
+    lpt->start_stateless_simulation_file((std::string)out_filename, CGlobalInfo::m_options.preview, m_limit / m_dp_core_count);
     lpt->start_stateless_daemon_simulation();
 
     flush_dp_to_cp_messages_core(core_index);
+
+    return lpt->m_node_gen.m_cnt;
 }