fix for http://trex-tgn.cisco.com/youtrack/issue/trex-174

15 files changed, 169 insertions, 35 deletions

diff --git a/scripts/automation/trex_control_plane/client/trex_stateless_client.py b/scripts/automation/trex_control_plane/client/trex_stateless_client.py
index 78adbcc2..506decfe 100755
--- a/scripts/automation/trex_control_plane/client/trex_stateless_client.py
+++ b/scripts/automation/trex_control_plane/client/trex_stateless_client.py
@@ -25,6 +25,7 @@ from trex_port import Port
 from common.trex_types import *
 from common.trex_stl_exceptions import *
 from trex_async_client import CTRexAsyncClient
+from yaml import YAMLError
 
 
 ############################     logger     #############################
@@ -436,9 +437,6 @@ class STLClient(object):
         self.stats_generator = trex_stats.CTRexInfoGenerator(self.global_stats,
                                                               self.ports)
 
-        # stream DB
-        self.streams_db = CStreamsDB()
-
  
  
     ############# private functions - used by the class itself ###########
@@ -1301,17 +1299,28 @@ class STLClient(object):
         if not rc:
             raise STLArgumentError('ports', ports, valid_values = self.get_all_ports())
 
-        # load the YAML
+
+        streams = None
+
+        # try YAML
         try:
-            streams_pack = self.streams_db.load_yaml_file(filename)
-        except Exception as e:
-            raise STLError(str(e))
-
-        # HACK - convert the stream pack to simple streams
-        streams = []
-        for stream in streams_pack.compiled:
-            s = HACKSTLStream(stream)
-            streams.append(s)
+            streams_db = CStreamsDB()
+            stream_list = streams_db.load_yaml_file(filename)
+
+            # convert to new style stream object
+            streams = [HACKSTLStream(stream) for stream in stream_list.compiled]
+        except YAMLError:
+            # try python
+            try:
+                basedir = os.path.dirname(filename)
+                sys.path.append(basedir)
+                file    = os.path.basename(filename).split('.')[0]
+                module = __import__(file, globals(), locals(), [], -1)
+
+                streams = module.register().get_streams()
+
+            except (AttributeError, ImportError):
+                raise STLError("bad format input file '{0}'".format(filename))
 
         self.add_streams(streams, ports)
 
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 7f65996d..d8f6ed92 100644
--- a/scripts/automation/trex_control_plane/client/trex_stateless_sim.py
+++ b/scripts/automation/trex_control_plane/client/trex_stateless_sim.py
@@ -25,7 +25,7 @@ except ImportError:
     import client.outer_packages
 
 from common.trex_stl_exceptions import STLError
-from yaml.scanner import ScannerError
+from yaml import YAMLError
 from common.trex_streams import *
 from client_utils import parsing_opts
 
@@ -103,7 +103,7 @@ class STLSim(object):
 
             # convert to new style stream object
             return [HACKSTLStream(stream) for stream in stream_list.compiled]
-        except ScannerError:
+        except YAMLError:
             pass
 
         # try python
@@ -111,12 +111,12 @@ class STLSim(object):
             basedir = os.path.dirname(input_file)
             sys.path.append(basedir)
 
-            file    = os.path.basename(input_file).split('.py')[0]
+            file    = os.path.basename(input_file).split('.')[0]
             module = __import__(file, globals(), locals(), [], -1)
 
             return module.register().get_streams()
 
-        except AttributeError:
+        except (AttributeError, ImportError):
             pass
 
         raise STLError("bad format input file '{0}'".format(input_file))
diff --git a/scripts/exp/stl_vm_split_client_var.erf-0.erf b/scripts/exp/stl_vm_split_client_var.erf-0.erf
index 12e68c09..a9f330fe 100644
--- a/scripts/exp/stl_vm_split_client_var.erf-0.erf
+++ b/scripts/exp/stl_vm_split_client_var.erf-0.erf
diff --git a/scripts/exp/stl_vm_split_flow_var_big_range.erf-0.erf b/scripts/exp/stl_vm_split_flow_var_big_range.erf-0.erf
index 8f4fe297..8e62aca6 100644
--- a/scripts/exp/stl_vm_split_flow_var_big_range.erf-0.erf
+++ b/scripts/exp/stl_vm_split_flow_var_big_range.erf-0.erf
diff --git a/scripts/exp/stl_vm_split_flow_var_inc.erf-0.erf b/scripts/exp/stl_vm_split_flow_var_inc.erf-0.erf
index 82025972..acd31a64 100644
--- a/scripts/exp/stl_vm_split_flow_var_inc.erf-0.erf
+++ b/scripts/exp/stl_vm_split_flow_var_inc.erf-0.erf
diff --git a/scripts/exp/stl_vm_split_flow_var_small_range.erf-0.erf b/scripts/exp/stl_vm_split_flow_var_small_range.erf-0.erf
index 5dac9893..4cf58e67 100644
--- a/scripts/exp/stl_vm_split_flow_var_small_range.erf-0.erf
+++ b/scripts/exp/stl_vm_split_flow_var_small_range.erf-0.erf
diff --git a/src/bp_sim.cpp b/src/bp_sim.cpp
index a7fadf93..0acb991a 100755
--- a/src/bp_sim.cpp
+++ b/src/bp_sim.cpp
@@ -790,13 +790,18 @@ int CErfIF::write_pkt(CCapPktRaw *pkt_raw){
 
 int CErfIF::close_file(void){
 
-    BP_ASSERT(m_raw);
-    delete m_raw;
+    if (m_raw) {
+        delete m_raw;
+        m_raw = NULL;
+    }
+
     if ( m_preview_mode->getFileWrite() ){
-        BP_ASSERT(m_writer);
-        delete m_writer;
-        m_writer=0;
+        if (m_writer) {
+            delete m_writer;
+            m_writer = NULL;
+        }
     }
+
     return (0);
 }
 
@@ -4042,7 +4047,11 @@ void CFlowGenListPerThread::stop_stateless_simulation_file(){
 void CFlowGenListPerThread::start_stateless_daemon_simulation(){
 
     m_cur_time_sec = 0;
-    m_stateless_dp_info.run_once();
+
+    /* if no pending CP messages - the core will simply be stuck forever */
+    if (m_stateless_dp_info.are_any_pending_cp_messages()) {
+        m_stateless_dp_info.run_once();
+    }
 }
 
 
diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index 704a24b7..f01346b0 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -1432,7 +1432,7 @@ public:
         }
 
         lpt->start_stateless_daemon_simulation();
-
+        lpt->stop_stateless_simulation_file();
 
         //lpt->m_node_gen.DumpHist(stdout);
 
@@ -3054,6 +3054,8 @@ public:
         assert(compile.compile(0, streams, objs, dp_core_count));
 
         /* choose one DP object */
+        assert(objs[dp_core_to_check]);
+
         TrexStatelessDpStart *lpStartCmd = new TrexStatelessDpStart(0, 0, objs[dp_core_to_check], 1 /*sec */ );
         objs[dp_core_to_check] = NULL;
         /* free all the non used DP objects */
@@ -3089,7 +3091,8 @@ TEST_F(basic_stl, vm_split_flow_var_inc) {
     VmSplitTest split("exp/stl_vm_split_flow_var_inc.erf");
 
     TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
-    stream.set_pps(1000);
+    stream.set_single_burst(1000);
+    stream.set_pps(100000);
 
     split.set_stream(&stream);
     split.run(8, 4);
@@ -3101,7 +3104,8 @@ TEST_F(basic_stl, vm_split_flow_var_small_range) {
     VmSplitTest split("exp/stl_vm_split_flow_var_small_range.erf");
 
     TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
-    stream.set_pps(1000);
+    stream.set_single_burst(1000);
+    stream.set_pps(100000);
 
     split.set_stream(&stream);
     split.set_flow_var_as_split(StreamVmInstructionFlowMan::FLOW_VAR_OP_INC, 0, 1, 0);
@@ -3114,7 +3118,8 @@ TEST_F(basic_stl, vm_split_flow_var_big_range) {
     VmSplitTest split("exp/stl_vm_split_flow_var_big_range.erf");
 
     TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
-    stream.set_pps(1000);
+    stream.set_single_burst(1000);
+    stream.set_pps(100000);
 
     split.set_stream(&stream);
     split.set_flow_var_as_split(StreamVmInstructionFlowMan::FLOW_VAR_OP_DEC, 1, 1000, 1000);
@@ -3128,7 +3133,8 @@ TEST_F(basic_stl, vm_split_client_var) {
      VmSplitTest split("exp/stl_vm_split_client_var.erf");
 
     TrexStream stream(TrexStream::stSINGLE_BURST, 0, 0);
-    stream.set_pps(1000);
+    stream.set_single_burst(1000);
+    stream.set_pps(100000);
 
     split.set_stream(&stream);
     split.set_client_var_as_split(0x10000001, 0x100000fe, 5000, 5050);
diff --git a/src/sim/trex_sim_stateless.cpp b/src/sim/trex_sim_stateless.cpp
index 46b9916f..897d1fec 100644
--- a/src/sim/trex_sim_stateless.cpp
+++ b/src/sim/trex_sim_stateless.cpp
@@ -422,6 +422,7 @@ SimStateless::run_dp_core(int core_index,
 
     lpt->start_stateless_simulation_file((std::string)out_filename, CGlobalInfo::m_options.preview, get_limit_per_core(core_index));
     lpt->start_stateless_daemon_simulation();
+    lpt->stop_stateless_simulation_file();
 
     flush_dp_to_cp_messages_core(core_index);
 
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 2286ef76..a281e428 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -184,8 +184,15 @@ TrexStatelessPort::start_traffic(const TrexPortMultiplier &mul, double duration,
     int index = 0;
     for (auto core_id : m_cores_id_list) {
 
-        TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_port_id, event_id, compiled_objs[index], duration);
-        send_message_to_dp(core_id, start_msg);
+        /* was the core assigned a compiled object ? */
+        if (compiled_objs[index]) {
+            TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_port_id, event_id, compiled_objs[index], duration);
+            send_message_to_dp(core_id, start_msg);
+        } else {
+
+            /* mimic an end event */
+            m_dp_events.handle_event(TrexDpPortEvent::EVENT_STOP, core_id, event_id);
+        }
 
         index++;
     }
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index e854cdf7..525bf594 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -165,6 +165,19 @@ public:
     }
 
 
+    /* can this stream be split ? */
+    bool is_splitable(uint8_t dp_core_count) const {
+
+        /* cont stream is always splitable */
+        if (m_type == stCONTINUOUS) {
+            return true;
+        }
+
+        int per_core_burst_total_pkts = (m_burst_total_pkts / dp_core_count);
+
+        return (per_core_burst_total_pkts > 0);
+
+    }
 
     void set_multi_burst(uint32_t   burst_total_pkts, 
                          uint32_t   num_bursts,
diff --git a/src/stateless/cp/trex_stream_vm.h b/src/stateless/cp/trex_stream_vm.h
index ca7f71c1..dabc502c 100644
--- a/src/stateless/cp/trex_stream_vm.h
+++ b/src/stateless/cp/trex_stream_vm.h
@@ -1196,7 +1196,7 @@ public:
     void copy_instructions(StreamVm &other) const;
 
     
-    bool is_vm_empty() {
+    bool is_vm_empty() const {
         return (m_inst_list.size() == 0);
     }
 
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index d9f979e8..9d048dbd 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -391,6 +391,7 @@ TrexStreamsCompiler::compile(uint8_t                                port_id,
     }
 
 }
+
 bool 
 TrexStreamsCompiler::compile_internal(uint8_t                                port_id,
                                       const std::vector<TrexStream *>        &streams,
@@ -440,6 +441,14 @@ TrexStreamsCompiler::compile_internal(uint8_t                                por
         compile_stream(stream, factor, dp_core_count, objs, nodes);
     }
 
+    /* some objects might be empty - no streams were assigned */
+    for (uint8_t i = 0; i < dp_core_count; i++) {
+        if (objs[i]->is_empty()) {
+            delete objs[i];
+            objs[i] = NULL;
+        }
+    }
+
     return true;
 }
 
@@ -466,13 +475,43 @@ TrexStreamsCompiler::compile_stream(const TrexStream *stream,
         new_next_id = nodes.get(stream->m_next_stream_id)->m_compressed_stream_id;
     }
 
+
+    /* can this stream be split to many cores ? */
+    if (!stream->is_splitable(dp_core_count)) {
+        compile_stream_on_single_core(stream,
+                                      factor,
+                                      objs[0],
+                                      new_id,
+                                      new_next_id);
+    } else {
+        compile_stream_on_all_cores(stream,
+                                    factor,
+                                    dp_core_count,
+                                    objs,
+                                    new_id,
+                                    new_next_id);
+    }
+
+
+}
+
+/**
+ * compile the stream on all the cores available
+ * 
+ */
+void
+TrexStreamsCompiler::compile_stream_on_all_cores(const TrexStream *stream,
+                                                 double factor,
+                                                 uint8_t dp_core_count,
+                                                 std::vector<TrexStreamsCompiledObj *> &objs,
+                                                 int new_id,
+                                                 int new_next_id) {
+
     std::vector<TrexStream *> core_streams(dp_core_count);
 
-    /* calculate rate */
     double per_core_rate          = (stream->m_pps * (factor / dp_core_count));
     int per_core_burst_total_pkts = (stream->m_burst_total_pkts / dp_core_count);
 
-
     /* for each core - creates its own version of the stream */
     for (uint8_t i = 0; i < dp_core_count; i++) {
         TrexStream *dp_stream = stream->clone();
@@ -488,7 +527,7 @@ TrexStreamsCompiler::compile_stream(const TrexStream *stream,
         core_streams[i] = dp_stream;
     }
 
-    /* take care of remainder from a burst */
+    /* take care of remainder from a burst on core 0 */
     int burst_remainder = stream->m_burst_total_pkts - (per_core_burst_total_pkts * dp_core_count);
     core_streams[0]->m_burst_total_pkts += burst_remainder;
 
@@ -501,9 +540,33 @@ TrexStreamsCompiler::compile_stream(const TrexStream *stream,
         objs[i]->add_compiled_stream(core_streams[i]);
     }
 
-
 }
 
+/**
+ * compile the stream on core 0
+ * 
+ */
+void
+TrexStreamsCompiler::compile_stream_on_single_core(const TrexStream *stream,
+                                                   double factor,
+                                                   TrexStreamsCompiledObj *obj,
+                                                   int new_id,
+                                                   int new_next_id) {
+
+    TrexStream *dp_stream = stream->clone();
+
+    /* fix stream ID */
+    dp_stream->fix_dp_stream_id(new_id, new_next_id);
+
+    /* compile the VM if exists */
+    if (!stream->m_vm.is_vm_empty()) {
+        ((TrexStream *)stream)->vm_compile();
+        dp_stream->m_vm_dp = stream->m_vm_dp->clone();
+    }
+
+    /* update the core */
+    obj->add_compiled_stream(dp_stream);
+}
 
 /**************************************
  * streams graph
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index aaef78be..a3a1f8f7 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -65,6 +65,10 @@ public:
 
     TrexStreamsCompiledObj* clone();
 
+    bool is_empty() {
+        return (m_objs.size() == 0);
+    }
+
 private:
     void add_compiled_stream(TrexStream *stream);
 
@@ -125,6 +129,19 @@ private:
                         std::vector<TrexStreamsCompiledObj *> &objs,
                         GraphNodeMap &nodes);
 
+    void compile_stream_on_single_core(const TrexStream *stream,
+                                       double factor,
+                                       TrexStreamsCompiledObj *obj,
+                                       int new_id,
+                                       int new_next_id);
+
+    void compile_stream_on_all_cores(const TrexStream *stream,
+                                     double factor,
+                                     uint8_t dp_core_count,
+                                     std::vector<TrexStreamsCompiledObj *> &objs,
+                                     int new_id,
+                                     int new_next_id);
+
     std::vector<std::string> m_warnings;
 };
 
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index efdb364c..c8a5eff5 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -181,6 +181,15 @@ public:
 
 
     /**
+     * return true if core has any pending messages from CP
+     * 
+     */
+    bool are_any_pending_cp_messages() {
+        return (!m_ring_from_cp->isEmpty());
+    }
+
+
+    /**
      * check for and handle messages from CP
      * 
      * @author imarom (27-Oct-15)