Merge from master

16 files changed, 591 insertions, 91 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 0df2ac5d..4478ed3f 100755
--- a/scripts/automation/trex_control_plane/client/trex_stateless_client.py
+++ b/scripts/automation/trex_control_plane/client/trex_stateless_client.py
@@ -280,7 +280,7 @@ class Port:
         return self.ok()
 
     # start traffic
-    def start (self, mul):
+    def start (self, mul, duration):
         if self.state == self.STATE_DOWN:
             return self.err("Unable to start traffic - port is down")
 
@@ -292,7 +292,8 @@ class Port:
 
         params = {"handler": self.handler,
                   "port_id": self.port_id,
-                  "mul": mul}
+                  "mul": mul,
+                  "duration": duration}
 
         rc, data = self.transmit("start_traffic", params)
         if not rc:
@@ -419,7 +420,7 @@ class CTRexStatelessClient(object):
         return RC_OK()
 
     def is_connected (self):
-        return self.connected
+        return self.connected and self.comm_link.is_connected
 
 
     def disconnect(self):
@@ -580,14 +581,14 @@ class CTRexStatelessClient(object):
         return self.ports[port_id].get_stream_id_list()
 
 
-    def start_traffic (self, multiplier, port_id_list = None):
+    def start_traffic (self, multiplier, duration, port_id_list = None):
 
         port_id_list = self.__ports(port_id_list)
 
         rc = RC()
 
         for port_id in port_id_list:
-            rc.add(self.ports[port_id].start(multiplier))
+            rc.add(self.ports[port_id].start(multiplier, duration))
         
         return rc
 
@@ -685,7 +686,7 @@ class CTRexStatelessClient(object):
         return RC_OK()
 
     # start cmd
-    def cmd_start (self, port_id_list, stream_list, mult, force):
+    def cmd_start (self, port_id_list, stream_list, mult, force, duration):
 
         active_ports = list(set(self.get_active_ports()).intersection(port_id_list))
 
@@ -713,7 +714,7 @@ class CTRexStatelessClient(object):
 
 
         # finally, start the traffic
-        rc = self.start_traffic(mult, port_id_list)
+        rc = self.start_traffic(mult, duration, port_id_list)
         rc.annotate("Starting traffic on port(s) {0}:".format(port_id_list))
         if rc.bad():
             return rc
@@ -754,7 +755,7 @@ class CTRexStatelessClient(object):
                 return RC_ERR("Failed to load stream pack")
 
 
-        return self.cmd_start(opts.ports, stream_list, opts.mult, opts.force)
+        return self.cmd_start(opts.ports, stream_list, opts.mult, opts.force, opts.duration)
 
     def cmd_stop_line (self, line):
         '''Stop active traffic in specified ports on TRex\n'''
@@ -820,7 +821,7 @@ class CTRexStatelessClient(object):
         cmd_table['wait']  = self.cmd_wait_line
         cmd_table['exit']  = self.cmd_exit_line
 
-        for index, line in enumerate(script_lines):
+        for index, line in enumerate(script_lines, start = 1):
             line = line.strip()
             if line == "":
                 continue
diff --git a/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py b/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py
index 077c82ad..b826f02f 100755
--- a/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py
+++ b/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py
@@ -110,45 +110,45 @@ class JsonRpcClient(object):
             return id, msg
 
 
-    def invoke_rpc_method (self, method_name, params = {}, block = True):
+    def invoke_rpc_method (self, method_name, params = {}):
         if not self.connected:
             return False, "Not connected to server"
 
         id, msg = self.create_jsonrpc_v2(method_name, params)
 
-        return self.send_raw_msg(msg, block)
+        return self.send_raw_msg(msg)
 
 
     # low level send of string message
-    def send_raw_msg (self, msg, block = True):
+    def send_raw_msg (self, msg):
+
         self.verbose_msg("Sending Request To Server:\n\n" + self.pretty_json(msg) + "\n")
 
-        if block:
-            self.socket.send(msg)
-        else:
+        tries = 0
+        while True:
             try:
-                self.socket.send(msg, flags = zmq.NOBLOCK)
-            except zmq.error.ZMQError as e:
-                self.disconnect()
-                return CmdResponse(False, "Failed To Get Send Message")
-
-        got_response = False
+                self.socket.send(msg)
+                break
+            except zmq.Again:
+                sleep(0.1)
+                tries += 1
+                if tries > 10:
+                    self.disconnect()
+                    return CmdResponse(False, "Failed to send message to server")
+
+
+        tries = 0
+        while True:
+            try:
+                response = self.socket.recv()
+                break
+            except zmq.Again:
+                sleep(0.1)
+                tries += 1
+                if tries > 10:
+                    self.disconnect()
+                    return CmdResponse(False, "Failed to get server response")
 
-        if block:
-            response = self.socket.recv()
-            got_response = True
-        else:
-            for i in xrange(0 ,10):
-                try:
-                    response = self.socket.recv(flags = zmq.NOBLOCK)
-                    got_response = True
-                    break
-                except zmq.Again:
-                    sleep(0.2)
-
-        if not got_response:
-            self.disconnect()
-            return CmdResponse(False, "Failed To Get Server Response")
 
         self.verbose_msg("Server Response:\n\n" + self.pretty_json(response) + "\n")
 
@@ -223,6 +223,8 @@ class JsonRpcClient(object):
         except zmq.error.ZMQError as e:
             return False, "ZMQ Error: Bad server or port name: " + str(e)
 
+        self.socket.setsockopt(zmq.SNDTIMEO, 5)
+        self.socket.setsockopt(zmq.RCVTIMEO, 5)
 
         self.connected = True
 
diff --git a/scripts/automation/trex_control_plane/console/parsing_opts.py b/scripts/automation/trex_control_plane/console/parsing_opts.py
index c154ce24..d5c21af0 100755
--- a/scripts/automation/trex_control_plane/console/parsing_opts.py
+++ b/scripts/automation/trex_control_plane/console/parsing_opts.py
@@ -89,10 +89,13 @@ OPTIONS_DB = {MULTIPLIER: ArgumentPack(['-m', '--multiplier'],
                                          "dest": "all_ports",
                                          'help': "Set this flag to apply the command on all available ports"}),
               DURATION: ArgumentPack(['-d'],
-                                        {"action": "store",
+                                        {'action': "store",
                                          'metavar': 'TIME',
-                                         "type": match_time_unit,
+                                         'dest': 'duration',
+                                         'type': match_time_unit,
+                                         'default': -1.0,
                                          'help': "Set duration time for TRex."}),
+
               FORCE: ArgumentPack(['--force'],
                                         {"action": "store_true",
                                          'default': False,
diff --git a/scripts/automation/trex_control_plane/console/trex_console.py b/scripts/automation/trex_control_plane/console/trex_console.py
index 88e8dede..7cb65fa6 100755
--- a/scripts/automation/trex_control_plane/console/trex_console.py
+++ b/scripts/automation/trex_control_plane/console/trex_console.py
@@ -33,6 +33,7 @@ from client.trex_stateless_client import CTRexStatelessClient
 from common.text_opts import *
 from client_utils.general_utils import user_input, get_current_user
 import trex_status
+import parsing_opts
 
 
 __version__ = "1.0"
diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index 8b96ef88..9148d5ae 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -29,7 +29,7 @@ limitations under the License.
 #include <trex_stream.h>
 #include <trex_stateless_port.h>
 #include <trex_rpc_server_api.h>
-
+#include <iostream>
 
 #define EXPECT_EQ_UINT32(a,b) EXPECT_EQ((uint32_t)(a),(uint32_t)(b))
 
@@ -239,7 +239,7 @@ TEST_F(basic_stl, single_pkt_burst1) {
 
      TrexStream * stream1 = new TrexStream(TrexStream::stSINGLE_BURST, 0,0);
      stream1->set_pps(1.0);
-     stream1->set_signle_burtst(5);
+     stream1->set_single_burst(5);
      stream1->m_enabled = true;
      stream1->m_self_start = true;
 
@@ -252,10 +252,9 @@ TEST_F(basic_stl, single_pkt_burst1) {
 
      TrexStreamsCompiledObj comp_obj(0,1.0);
 
-     comp_obj.set_simulation_duration( 10.0);
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone() );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10.0 );
 
 
      t1.m_msg = lpstart;
@@ -303,10 +302,9 @@ TEST_F(basic_stl, single_pkt) {
 
      TrexStreamsCompiledObj comp_obj(0,1.0);
 
-     comp_obj.set_simulation_duration( 10.0);
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone() );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10 );
 
 
      t1.m_msg = lpstart;
@@ -346,7 +344,7 @@ TEST_F(basic_stl, multi_pkt1) {
 
      streams.push_back(stream1);
 
-     TrexStream * stream2 = new TrexStream(TrexStream::stCONTINUOUS,0,0);
+     TrexStream * stream2 = new TrexStream(TrexStream::stCONTINUOUS,0,1);
      stream2->set_pps(2.0);
 
      stream2->m_enabled = true;
@@ -361,10 +359,9 @@ TEST_F(basic_stl, multi_pkt1) {
      // stream - clean 
      TrexStreamsCompiledObj comp_obj(0,1.0);
 
-     comp_obj.set_simulation_duration( 10.0);
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone() );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10 );
 
      t1.m_msg = lpstart;
 
@@ -425,10 +422,9 @@ TEST_F(basic_stl, multi_pkt2) {
      // stream - clean 
      TrexStreamsCompiledObj comp_obj(0,5.0);
 
-     comp_obj.set_simulation_duration( 10.0);
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone() );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10 );
 
      t1.m_msg = lpstart;
 
@@ -472,10 +468,9 @@ TEST_F(basic_stl, multi_burst1) {
 
      TrexStreamsCompiledObj comp_obj(0,1.0);
 
-     comp_obj.set_simulation_duration( 40.0);
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone() );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 40 );
 
 
      t1.m_msg = lpstart;
@@ -487,6 +482,200 @@ TEST_F(basic_stl, multi_burst1) {
      EXPECT_EQ_UINT32(1, res?1:0)<< "pass";
 }
 
+/********************************************* Itay Tests Start *************************************/
+
+/**
+ * check that continous stream does not point to another stream 
+ * (makes no sense) 
+ */
+TEST_F(basic_stl, compile_bad_1) {
+
+     TrexStreamsCompiler compile;
+     std::vector<TrexStream *> streams;
+
+     TrexStream * stream1 = new TrexStream(TrexStream::stCONTINUOUS,0,2);
+     stream1->m_enabled = true;
+     stream1->set_pps(52.0);
+     stream1->m_next_stream_id = 3;
+
+     streams.push_back(stream1);
+
+     TrexStreamsCompiledObj comp_obj(0,1.0);
+
+     std::string err_msg;
+     EXPECT_FALSE(compile.compile(streams, comp_obj, &err_msg));
+
+     delete stream1;
+
+}
+
+/**
+ * check for streams pointing to non exsistant streams
+ * 
+ * @author imarom (16-Nov-15)
+ */
+TEST_F(basic_stl, compile_bad_2) {
+
+     TrexStreamsCompiler compile;
+     std::vector<TrexStream *> streams;
+
+     TrexStream * stream1 = new TrexStream(TrexStream::stSINGLE_BURST,0,1);
+     stream1->m_enabled = true;
+     stream1->set_pps(1.0);
+     stream1->set_single_burst(200);
+
+     /* non existant next stream */
+     stream1->m_next_stream_id = 5;
+
+     TrexStream * stream2 = new TrexStream(TrexStream::stCONTINUOUS,0,2);
+     stream1->set_pps(52.0);
+
+     streams.push_back(stream1);
+     streams.push_back(stream2);
+
+     TrexStreamsCompiledObj comp_obj(0,1.0);
+
+     std::string err_msg;
+     EXPECT_FALSE(compile.compile(streams, comp_obj, &err_msg));
+
+     delete stream1;
+     delete stream2;
+
+}
+
+/**
+ * check for "dead streams" in the mesh 
+ * a streams that cannot be reached 
+ * 
+ * @author imarom (16-Nov-15)
+ */
+TEST_F(basic_stl, compile_bad_3) {
+
+     TrexStreamsCompiler compile;
+     std::vector<TrexStream *> streams;
+     TrexStream *stream;
+
+     /* stream 1 */
+     stream = new TrexStream(TrexStream::stSINGLE_BURST, 0, 231);
+     stream->m_enabled = true;
+     stream->set_pps(1.0);
+     stream->set_single_burst(200);
+
+     stream->m_next_stream_id = 5481;
+     stream->m_self_start = true;
+
+     streams.push_back(stream);
 
+     /* stream 2 */
+     stream = new TrexStream(TrexStream::stCONTINUOUS, 0, 5481);
+     stream->m_enabled = true;
+     stream->m_next_stream_id = -1;
+     stream->m_self_start = false;
+     stream->set_pps(52.0);
+     
+     streams.push_back(stream);
+
+     /* stream 3 */
+
+     stream = new TrexStream(TrexStream::stSINGLE_BURST, 0, 1928);
+     stream->m_enabled = true;
+     stream->set_pps(1.0);
+     stream->set_single_burst(200);
+
+     stream->m_next_stream_id = -1;
+     stream->m_self_start = true;
+
+     streams.push_back(stream);
+
+      /* stream 4 */
+
+     stream = new TrexStream(TrexStream::stSINGLE_BURST, 0, 41231);
+     stream->m_enabled = true;
+     stream->set_pps(1.0);
+     stream->set_single_burst(200);
+
+     stream->m_next_stream_id = 3928;
+     stream->m_self_start = false;
+
+     streams.push_back(stream);
+
+     /* stream 5 */
+
+     stream = new TrexStream(TrexStream::stSINGLE_BURST, 0, 3928);
+     stream->m_enabled = true;
+     stream->set_pps(1.0);
+     stream->set_single_burst(200);
+
+     stream->m_next_stream_id = 41231;
+     stream->m_self_start = false;
+
+     streams.push_back(stream);
+
+     /* compile */
+     TrexStreamsCompiledObj comp_obj(0,1.0);
+
+     std::string err_msg;
+     EXPECT_FALSE(compile.compile(streams, comp_obj, &err_msg));
+
+     for (auto stream : streams) {
+         delete stream;
+     }
+
+}
+
+TEST_F(basic_stl, compile_with_warnings) {
+
+    TrexStreamsCompiler compile;
+     std::vector<TrexStream *> streams;
+     TrexStream *stream;
+
+     /* stream 1 */
+     stream = new TrexStream(TrexStream::stSINGLE_BURST, 0, 231);
+     stream->m_enabled = true;
+     stream->set_pps(1.0);
+     stream->set_single_burst(200);
+
+     stream->m_next_stream_id = 1928;
+     stream->m_self_start = true;
+
+     streams.push_back(stream);
+
+     /* stream 2 */
+     stream = new TrexStream(TrexStream::stSINGLE_BURST, 0, 5481);
+     stream->m_enabled = true;
+     stream->m_next_stream_id = 1928;
+     stream->m_self_start = true;
+     stream->set_pps(52.0);
+     
+     streams.push_back(stream);
+
+     /* stream 3 */
+
+     stream = new TrexStream(TrexStream::stSINGLE_BURST, 0, 1928);
+     stream->m_enabled = true;
+     stream->set_pps(1.0);
+     stream->set_single_burst(200);
+
+     stream->m_next_stream_id = -1;
+     stream->m_self_start = true;
+
+     streams.push_back(stream);
+
+
+
+     /* compile */
+     TrexStreamsCompiledObj comp_obj(0,1.0);
+
+     std::string err_msg;
+     EXPECT_TRUE(compile.compile(streams, comp_obj, &err_msg));
+     
+     EXPECT_TRUE(compile.get_last_compile_warnings().size() == 1);
+
+     for (auto stream : streams) {
+         delete stream;
+     }
+
+}
 
 
+/********************************************* Itay Tests End *************************************/
diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
index e32073b0..cdd13ed6 100644
--- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
+++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp
@@ -143,6 +143,10 @@ TrexRpcCmdAddStream::allocate_new_stream(const Json::Value &section, uint8_t por
         stream = new TrexStream( TrexStream::stCONTINUOUS, port_id, stream_id);
         stream->set_pps(pps);
 
+        if (stream->m_next_stream_id != -1) {
+            generate_parse_err(result, "continious stream cannot provide next stream id - only -1 is valid");
+        }
+
     } else if (type == "single_burst") {
 
         uint32_t total_pkts      = parse_int(mode, "total_pkts", result);
@@ -150,7 +154,7 @@ TrexRpcCmdAddStream::allocate_new_stream(const Json::Value &section, uint8_t por
 
         stream = new TrexStream(TrexStream::stSINGLE_BURST,port_id, stream_id);
         stream->set_pps(pps);
-        stream->set_signle_burtst(total_pkts);
+        stream->set_single_burst(total_pkts);
 
 
     } else if (type == "multi_burst") {
@@ -458,8 +462,9 @@ TrexRpcCmdGetStream::_run(const Json::Value &params, Json::Value &result) {
 trex_rpc_cmd_rc_e
 TrexRpcCmdStartTraffic::_run(const Json::Value &params, Json::Value &result) {
 
-    uint8_t port_id = parse_byte(params, "port_id", result);
-    double mul = parse_double(params, "mul", result);
+    uint8_t port_id  = parse_byte(params, "port_id", result);
+    double mul       = parse_double(params, "mul", result);
+    double duration  = parse_double(params, "duration", result);
 
     if (port_id >= get_stateless_obj()->get_port_count()) {
         std::stringstream ss;
@@ -470,7 +475,7 @@ TrexRpcCmdStartTraffic::_run(const Json::Value &params, Json::Value &result) {
     TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id);
 
     try {
-        port->start_traffic(mul);
+        port->start_traffic(mul, duration);
     } catch (const TrexRpcException &ex) {
         generate_execute_err(result, ex.what());
     }
diff --git a/src/rpc-server/commands/trex_rpc_cmds.h b/src/rpc-server/commands/trex_rpc_cmds.h
index d7265ff2..b4f37e3b 100644
--- a/src/rpc-server/commands/trex_rpc_cmds.h
+++ b/src/rpc-server/commands/trex_rpc_cmds.h
@@ -105,7 +105,7 @@ TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStream, "get_stream", 3, true);
 
 
 
-TREX_RPC_CMD_DEFINE(TrexRpcCmdStartTraffic,  "start_traffic", 2, true);
+TREX_RPC_CMD_DEFINE(TrexRpcCmdStartTraffic,  "start_traffic", 3, true);
 TREX_RPC_CMD_DEFINE(TrexRpcCmdStopTraffic,   "stop_traffic", 1, true);
 TREX_RPC_CMD_DEFINE(TrexRpcCmdPauseTraffic,  "pause_traffic", 1, true);
 TREX_RPC_CMD_DEFINE(TrexRpcCmdResumeTraffic, "resume_traffic", 1, true);
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index 7f2382d3..cbc5a328 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -86,7 +86,7 @@ TrexStatelessPort::release(void) {
  * 
  */
 void
-TrexStatelessPort::start_traffic(double mul) {
+TrexStatelessPort::start_traffic(double mul, double duration) {
 
     /* command allowed only on state stream */
     verify_state(PORT_STATE_STREAMS);
@@ -105,7 +105,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);
+    TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(compiled_obj, duration);
 
     send_message_to_dp(start_msg);
 
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index 90bf936e..b533f793 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -72,7 +72,7 @@ public:
      * start traffic
      * throws TrexException in case of an error
      */
-    void start_traffic(double mul);
+    void start_traffic(double mul, double duration = -1);
 
     /**
      * stop traffic
diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h
index 151723ad..c2628cc3 100644
--- a/src/stateless/cp/trex_stream.h
+++ b/src/stateless/cp/trex_stream.h
@@ -92,21 +92,21 @@ public:
         m_type = type;
     }
 
-    uint8_t get_type(void){
+    uint8_t get_type(void) const {
         return ( m_type );
     }
 
 
 
     void set_multi_burst(uint32_t   burst_total_pkts, 
-                     uint32_t   num_bursts,
-                     double     ibg_usec){
+                         uint32_t   num_bursts,
+                         double     ibg_usec) {
         m_burst_total_pkts = burst_total_pkts;
         m_num_bursts       = num_bursts;
         m_ibg_usec         = ibg_usec;
     }
 
-    void set_signle_burtst(uint32_t   burst_total_pkts){
+    void set_single_burst(uint32_t   burst_total_pkts){
         set_multi_burst(burst_total_pkts,1,0.0); 
     }
 
diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp
index 80cdb31c..0c3b4ef0 100644
--- a/src/stateless/cp/trex_streams_compiler.cpp
+++ b/src/stateless/cp/trex_streams_compiler.cpp
@@ -19,15 +19,121 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-#include <string.h>
+#include <string>
+#include <sstream>
 #include <trex_streams_compiler.h>
 #include <trex_stream.h>
+#include <assert.h>
+#include <trex_stateless.h>
+#include <iostream>
+
+/**
+ * describes a graph node in the pre compile check
+ * 
+ * @author imarom (16-Nov-15)
+ */
+class GraphNode {
+public:
+    GraphNode(TrexStream *stream, GraphNode *next) : m_stream(stream), m_next(next) {
+        marked   = false;
+    }
+
+    uint32_t get_stream_id() const {
+        return m_stream->m_stream_id;
+    }
+
+    const TrexStream *m_stream;
+    GraphNode *m_next;
+    std::vector<const GraphNode *> m_parents;
+    bool marked;
+};
+
+/**
+ * node map
+ * 
+ */
+class GraphNodeMap {
+public:
+
+    GraphNodeMap() : m_dead_end(NULL, NULL) {
+
+    }
+
+    bool add(GraphNode *node) {
+        if (has(node->get_stream_id())) {
+            return false;
+        }
+
+        m_nodes[node->get_stream_id()] = node;
+
+        if (node->m_stream->m_self_start) {
+            m_roots.push_back(node);
+        }
+
+        return true;
+    }
+
+    bool has(uint32_t stream_id) {
+
+        return (get(stream_id) != NULL);
+    }
+
+    GraphNode * get(uint32_t stream_id) {
+
+        if (stream_id == -1) {
+            return &m_dead_end;
+        }
+
+        auto search = m_nodes.find(stream_id);
+
+        if (search != m_nodes.end()) {
+            return search->second;
+        } else {
+            return NULL;
+        }
+    }
+
+    void clear_marks() {
+        for (auto node : m_nodes) {
+            node.second->marked = false;
+        }
+    }
+
+    void get_unmarked(std::vector <GraphNode *> &unmarked) {
+        for (auto node : m_nodes) {
+            if (!node.second->marked) {
+                unmarked.push_back(node.second);
+            }
+        }
+    }
+
+
+    ~GraphNodeMap() {
+        for (auto node : m_nodes) {
+            delete node.second;
+        }
+        m_nodes.clear();
+    }
+
+    std::vector <GraphNode *> & get_roots() {
+        return m_roots;
+    }
+
+
+    std::unordered_map<uint32_t, GraphNode *> get_nodes() {
+        return m_nodes;
+    }
+
+private:
+    std::unordered_map<uint32_t, GraphNode *> m_nodes;
+    std::vector <GraphNode *> m_roots;
+    GraphNode m_dead_end;
+};
 
 /**************************************
  * stream compiled object
  *************************************/
 TrexStreamsCompiledObj::TrexStreamsCompiledObj(uint8_t port_id, double mul) : m_port_id(port_id), m_mul(mul) {
-    m_duration_sim=-1.0;
 }
 
 TrexStreamsCompiledObj::~TrexStreamsCompiledObj() {
@@ -61,16 +167,199 @@ TrexStreamsCompiledObj::clone() {
 
     new_compiled_obj->m_mul = m_mul;
 
-    new_compiled_obj->m_duration_sim = m_duration_sim;
-
     return new_compiled_obj;
 }
 
+void
+TrexStreamsCompiler::add_warning(const std::string &warning) {
+    m_warnings.push_back("*** warning: " + warning);
+}
+
+void
+TrexStreamsCompiler::err(const std::string &err) {
+    throw TrexException("*** error: " + err);
+}
+
+void
+TrexStreamsCompiler::check_stream(const TrexStream *stream) {
+    std::stringstream ss;
+
+    /* cont. stream can point only on itself */
+    if (stream->get_type() == TrexStream::stCONTINUOUS) {
+        if (stream->m_next_stream_id != -1) {
+            ss << "continous stream '" << stream->m_stream_id << "' cannot point on another stream";
+            err(ss.str());
+        }
+    }
+}
+
+void
+TrexStreamsCompiler::allocate_pass(const std::vector<TrexStream *> &streams,
+                                   GraphNodeMap *nodes) {
+    std::stringstream ss;
+
+    /* first pass - allocate all nodes and check for duplicates */
+    for (auto stream : streams) {
+
+        /* skip non enabled streams */
+        if (!stream->m_enabled) {
+            continue;
+        }
+
+        /* sanity check on the stream itself */
+        check_stream(stream);
+
+        /* duplicate stream id ? */
+        if (nodes->has(stream->m_stream_id)) {
+            ss << "duplicate instance of stream id " << stream->m_stream_id;
+            err(ss.str());
+        }
+
+        GraphNode *node = new GraphNode(stream, NULL);
+
+        /* add to the map */
+        assert(nodes->add(node));
+    }
+
+}
+
+/**
+ * on this pass we direct the graph to point to the right nodes
+ * 
+ */
+void
+TrexStreamsCompiler::direct_pass(GraphNodeMap *nodes) {
+   
+    /* second pass - direct the graph */
+    for (auto p : nodes->get_nodes()) {
+
+        GraphNode *node = p.second;
+        const TrexStream *stream = node->m_stream;
+
+        /* check the stream points on an existing stream */
+        GraphNode *next_node = nodes->get(stream->m_next_stream_id);
+        if (!next_node) {
+            std::stringstream ss;
+            ss << "stream " << node->get_stream_id() << " is pointing on non existent stream " << stream->m_next_stream_id;
+            err(ss.str());
+        }
+
+        node->m_next = next_node;
+
+        /* do we have more than one parent ? */
+        next_node->m_parents.push_back(node);
+    }
+
+
+    /* check for multiple parents */
+    for (auto p : nodes->get_nodes()) {
+        GraphNode *node = p.second;
+
+        if (node->m_parents.size() > 0 ) {
+            std::stringstream ss;
+
+            ss << "stream " << node->get_stream_id() << " is triggered by multiple streams: ";
+            for (auto x : node->m_parents) {
+                ss << x->get_stream_id() << " ";
+            }
+
+            add_warning(ss.str());
+        }
+    }
+}
+
+/**
+ * mark sure all the streams are reachable
+ * 
+ */
+void
+TrexStreamsCompiler::check_for_unreachable_streams(GraphNodeMap *nodes) {
+    /* start with the roots */
+    std::vector <GraphNode *> next_nodes = nodes->get_roots();
+
+
+    nodes->clear_marks();
+
+    /* run BFS from all the roots */
+    while (!next_nodes.empty()) {
+
+        /* pull one */
+        GraphNode *node = next_nodes.back();
+        next_nodes.pop_back();
+        if (node->marked) {
+            continue;
+        }
+
+        node->marked = true;
+
+        if (node->m_next != NULL) {
+            next_nodes.push_back(node->m_next);
+        }
+
+    }
+
+    std::vector <GraphNode *> unmarked;
+    nodes->get_unmarked(unmarked);
+
+    if (!unmarked.empty()) {
+        std::stringstream ss;
+        for (auto node : unmarked) {
+            ss << "stream " << node->get_stream_id() << " is unreachable from any other stream\n";
+        }
+        err(ss.str());
+    }
+
+
+}
+
+/**
+ * check validation of streams for compile
+ * 
+ * @author imarom (16-Nov-15)
+ * 
+ * @param streams 
+ * @param fail_msg 
+ * 
+ * @return bool 
+ */
+void
+TrexStreamsCompiler::pre_compile_check(const std::vector<TrexStream *> &streams) {
+
+    GraphNodeMap nodes;
+
+    m_warnings.clear();
+
+    /* allocate nodes */
+    allocate_pass(streams, &nodes);
+
+    /* direct the graph */
+    direct_pass(&nodes);
+
+    /* check for non reachable streams inside the graph */
+    check_for_unreachable_streams(&nodes);
+
+}
+
 /**************************************
  * stream compiler
  *************************************/
 bool 
-TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams, TrexStreamsCompiledObj &obj) {
+TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams,
+                             TrexStreamsCompiledObj &obj,
+                             std::string *fail_msg) {
+
+    /* compile checks */
+    try {
+        pre_compile_check(streams);
+    } catch (const TrexException &ex) {
+        if (fail_msg) {
+            *fail_msg = ex.what();
+        } else {
+            std::cout << ex.what();
+        }
+        return false;
+    }
+
     /* for now we do something trivial, */
     for (auto stream : streams) {
 
@@ -79,11 +368,6 @@ TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams, TrexStrea
             continue;
         }
 
-        /* for now skip also non self started streams */
-        if (!stream->m_self_start) {
-            continue;
-        }
-
         /* add it */
         obj.add_compiled_stream(stream);
     }
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index 78ac1ac7..42cfc5b8 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -23,9 +23,11 @@ limitations under the License.
 
 #include <stdint.h>
 #include <vector>
+#include <string>
 
 class TrexStreamsCompiler;
 class TrexStream;
+class GraphNodeMap;
 
 /**
  * compiled object for a table of streams
@@ -48,13 +50,6 @@ public:
         return m_objs;
     }
 
-    void set_simulation_duration(double duration){
-        m_duration_sim=duration;
-    }
-
-    double get_simulation_duration(){
-        return (m_duration_sim);
-    }
     /**
      * clone the compiled object
      * 
@@ -71,18 +66,39 @@ private:
 
     uint8_t m_port_id;
     double  m_mul;
-    double  m_duration_sim; /* duration for all simulation */
 };
 
 class TrexStreamsCompiler {
 public:
+
     /**
      * compiles a vector of streams to an object passable to the DP
      * 
      * @author imarom (28-Oct-15)
      * 
      */
-    bool compile(const std::vector<TrexStream *> &streams, TrexStreamsCompiledObj &obj);
+    bool compile(const std::vector<TrexStream *> &streams, TrexStreamsCompiledObj &obj, std::string *fail_msg = NULL);
+
+    /**
+     * 
+     * returns a reference pointer to the last compile warnings
+     * if no warnings were produced - the vector is empty
+     */
+    const std::vector<std::string> & get_last_compile_warnings() {
+        return m_warnings;
+    }
+
+private:
+
+    void pre_compile_check(const std::vector<TrexStream *> &streams);
+    void allocate_pass(const std::vector<TrexStream *> &streams, GraphNodeMap *nodes);
+    void direct_pass(GraphNodeMap *nodes);
+    void check_for_unreachable_streams(GraphNodeMap *nodes);
+    void check_stream(const TrexStream *stream);
+    void add_warning(const std::string &warning);
+    void err(const std::string &err);
+
+    std::vector<std::string> m_warnings;
 };
 
 #endif /* __TREX_STREAMS_COMPILER_H__ */
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 25984dfc..b25a4cfc 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -219,14 +219,12 @@ TrexStatelessDpCore::add_cont_stream(TrexStream * stream,
 }
 
 void
-TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj) {
+TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj, double duration) {
     for (auto single_stream : obj->get_objects()) {
         add_cont_stream(single_stream.m_stream,obj);
     }
 
-    double duration=obj->get_simulation_duration();
-
-    if ( duration >0.0){
+    if ( duration > 0.0 ){
         add_duration( duration );
     }
 }
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index 54fad240..aaa6eed3 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -75,7 +75,7 @@ public:
      * @param pkt 
      * @param pkt_len 
      */
-    void start_traffic(TrexStreamsCompiledObj *obj);
+    void start_traffic(TrexStreamsCompiledObj *obj, double duration = -1);
 
     /**
      * stop all traffic for this core
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index f529642d..2e3acffd 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -26,7 +26,7 @@ limitations under the License.
 /*************************
   start traffic message
  ************************/ 
-TrexStatelessDpStart::TrexStatelessDpStart(TrexStreamsCompiledObj *obj) : m_obj(obj) {
+TrexStatelessDpStart::TrexStatelessDpStart(TrexStreamsCompiledObj *obj, double duration) : m_obj(obj), m_duration(duration) {
 }
 
 
@@ -39,7 +39,7 @@ TrexStatelessDpStart::clone() {
 
     TrexStreamsCompiledObj *new_obj = m_obj->clone();
 
-    TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpStart(new_obj);
+    TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpStart(new_obj, m_duration);
 
     return new_msg;
 }
@@ -53,7 +53,7 @@ TrexStatelessDpStart::~TrexStatelessDpStart() {
 bool
 TrexStatelessDpStart::handle(TrexStatelessDpCore *dp_core) {
 
-    dp_core->start_traffic(m_obj);
+    dp_core->start_traffic(m_obj, m_duration);
     return true;
 }
 
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index d1621708..6473a6a4 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -67,7 +67,7 @@ public:
 class TrexStatelessDpStart : public TrexStatelessCpToDpMsgBase {
 public:
 
-    TrexStatelessDpStart(TrexStreamsCompiledObj *obj);
+    TrexStatelessDpStart(TrexStreamsCompiledObj *obj, double duration);
 
     ~TrexStatelessDpStart();
 
@@ -78,6 +78,7 @@ public:
 
 private:
     TrexStreamsCompiledObj *m_obj;
+    double m_duration;
 };
 
 /**