Merge from master

21 files changed, 1037 insertions, 143 deletions

diff --git a/linux/ws_main.py b/linux/ws_main.py
index 65ca4522..617bcf37 100755
--- a/linux/ws_main.py
+++ b/linux/ws_main.py
@@ -117,7 +117,7 @@ main_src = SrcGroup(dir='src',
              'utl_json.cpp',
              'utl_cpuu.cpp',
              'msg_manager.cpp',
-
+             'publisher/trex_publisher.cpp',
 
              'pal/linux/pal_utl.cpp',
              'pal/linux/mbuf.cpp'
@@ -152,6 +152,7 @@ stateless_src = SrcGroup(dir='src/stateless/',
                                     'cp/trex_stateless.cpp',
                                     'cp/trex_stateless_port.cpp',
                                     'cp/trex_streams_compiler.cpp',
+                                    'cp/trex_dp_port_events.cpp',
                                     'dp/trex_stateless_dp_core.cpp',
                                     'messaging/trex_stateless_messaging.cpp',
                                     ])
diff --git a/linux_dpdk/ws_main.py b/linux_dpdk/ws_main.py
index 98fa1fc6..89f41376 100755
--- a/linux_dpdk/ws_main.py
+++ b/linux_dpdk/ws_main.py
@@ -107,6 +107,7 @@ main_src = SrcGroup(dir='src',
              'utl_yaml.cpp',
              'nat_check.cpp',
              'msg_manager.cpp',
+             'publisher/trex_publisher.cpp',
              'pal/linux_dpdk/pal_utl.cpp',
              'pal/linux_dpdk/mbuf.cpp'
              ]);
@@ -159,6 +160,7 @@ stateless_src = SrcGroup(dir='src/stateless/',
                                     'cp/trex_stateless.cpp',
                                     'cp/trex_stateless_port.cpp',
                                     'cp/trex_streams_compiler.cpp',
+                                    'cp/trex_dp_port_events.cpp',
                                     'dp/trex_stateless_dp_core.cpp',
                                     'messaging/trex_stateless_messaging.cpp'
                                     ])
diff --git a/scripts/automation/trex_control_plane/client/trex_async_client.py b/scripts/automation/trex_control_plane/client/trex_async_client.py
index 31bec93f..adb91d97 100644
--- a/scripts/automation/trex_control_plane/client/trex_async_client.py
+++ b/scripts/automation/trex_control_plane/client/trex_async_client.py
@@ -8,6 +8,8 @@ except ImportError:
     import client.outer_packages
 from client_utils.jsonrpc_client import JsonRpcClient, BatchMessage
 
+from common.text_opts import *
+
 import json
 import threading
 import time
@@ -103,13 +105,9 @@ class TrexAsyncStatsManager():
 
         return self.port_stats[str(port_id)]
 
-    def update (self, snapshot):
-
-        if snapshot['name'] == 'trex-global':
-            self.__handle_snapshot(snapshot['data'])
-        else:
-            # for now ignore the rest
-            return
+   
+    def update (self, data):
+        self.__handle_snapshot(data)
 
     def __handle_snapshot (self, snapshot):
 
@@ -151,10 +149,11 @@ class TrexAsyncStatsManager():
 
 
 class CTRexAsyncClient():
-    def __init__ (self, server, port):
+    def __init__ (self, server, port, stateless_client):
 
         self.port = port
         self.server = server
+        self.stateless_client = stateless_client
 
         self.raw_snapshot = {}
 
@@ -165,14 +164,15 @@ class CTRexAsyncClient():
         print "\nConnecting To ZMQ Publisher At {0}".format(self.tr)
 
         self.active = True
-        self.t = threading.Thread(target = self._run)
+        self.t = threading.Thread(target = self.run)
 
         # kill this thread on exit and don't add it to the join list
         self.t.setDaemon(True)
         self.t.start()
 
 
-    def _run (self):
+
+    def run (self):
 
         #  Socket to talk to server
         self.context = zmq.Context()
@@ -185,10 +185,12 @@ class CTRexAsyncClient():
             line = self.socket.recv_string();
             msg = json.loads(line)
 
-            key = msg['name']
-            self.raw_snapshot[key] = msg['data']
+            name = msg['name']
+            data = msg['data']
+            type = msg['type']
+            self.raw_snapshot[name] = data
 
-            self.stats.update(msg)
+            self.__dispatch(name, type, data)
 
 
     def get_stats (self):
@@ -199,6 +201,26 @@ class CTRexAsyncClient():
         return self.raw_snapshot
 
 
+    # dispatch the message to the right place
+    def __dispatch (self, name, type, data):
+        # stats
+        if name == "trex-global":
+            self.stats.update(data)
+        # events
+        elif name == "trex-event":
+            self.__handle_async_event(type, data)
+        else:
+            # ignore
+            pass
+
+    def __handle_async_event (self, type, data):
+        # DP stopped
+        if (type == 0):
+            port_id = int(data['port_id'])
+            print format_text("\n[Event] - Port {0} Stopped".format(port_id), 'bold')
+            # call the handler
+            self.stateless_client.async_event_port_stopped(port_id)
+
     def stop (self):
         self.active = False
         self.t.join()
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 3dcfae28..dd11fb67 100755
--- a/scripts/automation/trex_control_plane/client/trex_stateless_client.py
+++ b/scripts/automation/trex_control_plane/client/trex_stateless_client.py
@@ -320,6 +320,9 @@ class Port(object):
 
         return self.ok()
 
+    ################# events handler ######################
+    def async_event_port_stopped (self):
+        self.state = self.STATE_STREAMS
 
 
 class CTRexStatelessClient(object):
@@ -337,12 +340,16 @@ class CTRexStatelessClient(object):
         self._server_version = None
         self.__err_log = None
 
-        self._async_client = CTRexAsyncClient(server, async_port)
+        self._async_client = CTRexAsyncClient(server, async_port, self)
 
         self.streams_db = CStreamsDB()
 
         self.connected = False
 
+    ################# events handler ######################
+    def async_event_port_stopped (self, port_id):
+        self.ports[port_id].async_event_port_stopped()
+
     ############# helper functions section ##############
 
     def validate_port_list(self, port_id_list):
@@ -790,7 +797,7 @@ class CTRexStatelessClient(object):
         if opts is None:
             return RC_ERR("bad command line paramters")
 
-        delay_sec = opts.d if opts.d else 1
+        delay_sec = opts.duration if (opts.duration > 0) else 1
 
         print format_text("Waiting for {0} seconds...\n".format(delay_sec), 'bold')
         time.sleep(delay_sec)
diff --git a/scripts/exp/ignore-0-ex.erf b/scripts/exp/ignore-0-ex.erf
new file mode 100644
index 00000000..92883717
--- /dev/null
+++ b/scripts/exp/ignore-0-ex.erf
diff --git a/scripts/exp/ignore-0.erf b/scripts/exp/ignore-0.erf
new file mode 100644
index 00000000..92883717
--- /dev/null
+++ b/scripts/exp/ignore-0.erf
diff --git a/scripts/ko/3.11.10-100.fc18.x86_64/igb_uio.ko b/scripts/ko/3.11.10-100.fc18.x86_64/igb_uio.ko
new file mode 100644
index 00000000..d2373adf
--- /dev/null
+++ b/scripts/ko/3.11.10-100.fc18.x86_64/igb_uio.ko
diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp
index ac0a5e63..bdaebcea 100644
--- a/src/gtest/trex_stateless_gtest.cpp
+++ b/src/gtest/trex_stateless_gtest.cpp
@@ -46,16 +46,51 @@ public:
 };
 
 
+/**
+ * handler for DP to CP messages
+ * 
+ * @author imarom (19-Nov-15)
+ */
+class DpToCpHandler {
+public:
+    virtual void handle(TrexStatelessDpToCpMsgBase *msg) = 0;
+};
 
 class CBasicStl {
 
 public:
+
+
     CBasicStl(){
         m_time_diff=0.001;
         m_threads=1;
         m_dump_json=false;
+        m_dp_to_cp_handler = NULL;
+    }
+
+
+    void flush_dp_to_cp_messages() {
+
+        CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingDpToCp(0);
+
+        while ( true ) {
+            CGenNode * node = NULL;
+            if (ring->Dequeue(node) != 0) {
+                break;
+            }
+            assert(node);
+
+            TrexStatelessDpToCpMsgBase * msg = (TrexStatelessDpToCpMsgBase *)node;
+            if (m_dp_to_cp_handler) {
+                m_dp_to_cp_handler->handle(msg);
+            }
+
+            delete msg;
+        }
+
     }
 
+
     bool  init(void){
 
         CErfIFStl erf_vif;
@@ -106,14 +141,18 @@ public:
             printf(" %s \n",s.c_str());
         }
 
+        flush_dp_to_cp_messages();
+
         fl.Delete();
         return (res);
     }
 
 public:
-    int           m_threads;
-    double        m_time_diff;
-    bool          m_dump_json;
+    int               m_threads;
+    double            m_time_diff;
+    bool              m_dump_json;
+    DpToCpHandler     *m_dp_to_cp_handler;
+
     TrexStatelessCpToDpMsgBase * m_msg;
     CNodeRing           *m_ring_from_cp;
     CFlowGenList  fl;
@@ -292,7 +331,7 @@ TEST_F(basic_stl, simple_prog4) {
 
      EXPECT_TRUE(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 20.0 );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(0, 0, comp_obj.clone(), 20.0 );
 
 
      t1.m_msg = lpstart;
@@ -362,7 +401,7 @@ TEST_F(basic_stl, simple_prog3) {
 
      EXPECT_TRUE(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 50.0 );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(0, 0, comp_obj.clone(), 50.0 );
 
 
      t1.m_msg = lpstart;
@@ -424,7 +463,7 @@ TEST_F(basic_stl, simple_prog2) {
 
      EXPECT_TRUE(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10.0 );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(0, 0, comp_obj.clone(), 10.0 );
 
 
      t1.m_msg = lpstart;
@@ -486,7 +525,7 @@ TEST_F(basic_stl, simple_prog1) {
 
      EXPECT_TRUE(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10.0 );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(0, 0, comp_obj.clone(), 10.0 );
 
 
      t1.m_msg = lpstart;
@@ -531,7 +570,7 @@ TEST_F(basic_stl, single_pkt_burst1) {
 
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10.0 );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(0, 0, comp_obj.clone(), 10.0 );
 
 
      t1.m_msg = lpstart;
@@ -582,7 +621,7 @@ TEST_F(basic_stl, single_pkt) {
 
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10.0 /*sec */ );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(port_id, 0, comp_obj.clone(), 10.0 /*sec */ );
 
 
      t1.m_msg = lpstart;
@@ -639,7 +678,7 @@ TEST_F(basic_stl, multi_pkt1) {
 
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10 );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(0, 0, comp_obj.clone(), 10 );
 
      t1.m_msg = lpstart;
 
@@ -702,7 +741,7 @@ TEST_F(basic_stl, multi_pkt2) {
 
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 10 );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(0, 0, comp_obj.clone(), 10 );
 
      t1.m_msg = lpstart;
 
@@ -748,7 +787,7 @@ TEST_F(basic_stl, multi_burst1) {
 
      assert(compile.compile(streams, comp_obj) );
 
-     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart( comp_obj.clone(), 40 );
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(0, 0, comp_obj.clone(), 40 );
 
 
      t1.m_msg = lpstart;
@@ -1003,4 +1042,78 @@ TEST_F(basic_stl, compile_good_stream_id_compres) {
 
 
 
+class DpToCpHandlerStopEvent: public DpToCpHandler {
+public:
+    DpToCpHandlerStopEvent(int event_id) {
+        m_event_id = event_id;
+    }
+
+    virtual void handle(TrexStatelessDpToCpMsgBase *msg) {
+        /* first the message must be an event */
+        TrexDpPortEventMsg *event = dynamic_cast<TrexDpPortEventMsg *>(msg);
+        EXPECT_TRUE(event != NULL);
+        EXPECT_TRUE(event->get_event_type() == TrexDpPortEvent::EVENT_STOP);
+
+        EXPECT_TRUE(event->get_event_id() == m_event_id);
+        EXPECT_TRUE(event->get_port_id() == 0);
+        
+    }
+
+private:
+    int m_event_id;
+};
+
+TEST_F(basic_stl, dp_stop_event) {
+
+    CBasicStl t1;
+    CParserOption * po =&CGlobalInfo::m_options;
+    po->preview.setVMode(7);
+    po->preview.setFileWrite(true);
+    po->out_file ="exp/ignore";
+
+    TrexStreamsCompiler compile;
+
+     uint8_t port_id=0;
+
+     std::vector<TrexStream *> streams;
+
+     TrexStream * stream1 = new TrexStream(TrexStream::stSINGLE_BURST,0,0);
+     stream1->set_pps(1.0);
+     stream1->set_single_burst(100);
+     
+     stream1->m_enabled = true;
+     stream1->m_self_start = true;
+     stream1->m_port_id= port_id;
+
+
+     CPcapLoader pcap;
+     pcap.load_pcap_file("cap2/udp_64B.pcap",0);
+     pcap.update_ip_src(0x10000001);
+     pcap.clone_packet_into_stream(stream1);
+                                    
+     streams.push_back(stream1);
+
+     // stream - clean 
+
+     TrexStreamsCompiledObj comp_obj(port_id, 1.0 /*mul*/);
+
+     assert(compile.compile(streams, comp_obj) );
+
+     TrexStatelessDpStart * lpstart = new TrexStatelessDpStart(port_id, 17, comp_obj.clone(), 10.0 /*sec */ );
+
+
+     t1.m_msg = lpstart;
+
+     /* let me handle these */
+     DpToCpHandlerStopEvent handler(17);
+     t1.m_dp_to_cp_handler = &handler;
+
+     bool res=t1.init();
+     EXPECT_EQ_UINT32(1, res?1:0);
+
+     delete stream1 ;
+     
+}
+
+
 /********************************************* Itay Tests End *************************************/
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index f66bcd9e..b1c9ed12 100755
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -58,6 +58,7 @@ limitations under the License.
 
 #include <stateless/cp/trex_stateless.h>
 #include <stateless/dp/trex_stream_node.h>
+#include <publisher/trex_publisher.h>
 #include <stateless/messaging/trex_stateless_messaging.h>
 
 #include <../linux_dpdk/version.h>
@@ -2398,71 +2399,6 @@ private:
 };
 
 
-class CZMqPublisher {
-public:
-    CZMqPublisher(){
-        m_context=0;
-        m_publisher=0;
-    }
-
-    bool Create(uint16_t port,bool disable);
-    void Delete();
-    void publish_json(std::string & s);
-private:
-    void show_zmq_last_error(char *s);
-private:
-    void * m_context;
-    void * m_publisher;
-};
-
-void CZMqPublisher::show_zmq_last_error(char *s){
-    printf(" ERROR %s \n",s);
-    printf(" ZMQ: %s",zmq_strerror (zmq_errno ()));
-    exit(-1);
-}
-
-
-bool CZMqPublisher::Create(uint16_t port,bool disable){
-
-    if (disable) {
-        return(true);
-    }
-    m_context = zmq_ctx_new ();
-    if ( m_context == 0 ) {
-        show_zmq_last_error((char *)"can't connect to ZMQ library");
-    }
-    m_publisher = zmq_socket (m_context, ZMQ_PUB);
-    if ( m_context == 0 ) {
-        show_zmq_last_error((char *)"can't create ZMQ socket");
-    }
-    char buffer[100];
-    sprintf(buffer,"tcp://*:%d",port);
-    int rc=zmq_bind (m_publisher, buffer);
-    if (rc != 0 ) {
-        sprintf(buffer,"can't bind to ZMQ socket %d",port);
-        show_zmq_last_error(buffer);
-    }
-    printf("zmq publisher at: %s \n",buffer);
-    return (true);
-}
-
-
-void CZMqPublisher::Delete(){
-    if (m_publisher) {
-        zmq_close (m_publisher);
-    }
-    if (m_context) {
-        zmq_ctx_destroy (m_context);
-    }
-}
-
-
-void CZMqPublisher::publish_json(std::string & s){
-    if ( m_publisher ){
-        int size = zmq_send (m_publisher, s.c_str(), s.length(), 0);
-        assert(size==s.length());
-    }
-}
 
 class CPerPortStats {
 public:
@@ -2825,6 +2761,10 @@ private:
     void try_stop_all_dp();
     /* send message to all dp cores */
     int  send_message_all_dp(TrexStatelessCpToDpMsgBase *msg);
+
+    void check_for_dp_message_from_core(int thread_id);
+    void check_for_dp_messages();
+
 public:
 
     int start_send_master();
@@ -2965,7 +2905,7 @@ private:
     CLatencyVmPort      m_latency_vm_vports[BP_MAX_PORTS]; /* vm driver */
 
     CLatencyPktInfo     m_latency_pkt;
-    CZMqPublisher       m_zmq_publisher;
+    TrexPublisher       m_zmq_publisher;
 
 public:
     TrexStateless       *m_trex_stateless;
@@ -3233,6 +3173,48 @@ int  CGlobalTRex::reset_counters(){
     return (0);
 }
 
+/**
+ * check for a single core
+ * 
+ * @author imarom (19-Nov-15)
+ * 
+ * @param thread_id 
+ */
+void 
+CGlobalTRex::check_for_dp_message_from_core(int thread_id) {
+
+    CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingDpToCp(thread_id);
+
+    /* fast path check */
+    if ( likely ( ring->isEmpty() ) ) {
+        return;
+    }
+
+    while ( true ) {
+        CGenNode * node = NULL;
+        if (ring->Dequeue(node) != 0) {
+            break;
+        }
+        assert(node);
+
+        TrexStatelessDpToCpMsgBase * msg = (TrexStatelessDpToCpMsgBase *)node;
+        msg->handle();
+        delete msg;
+    }
+
+}
+
+/**
+ * check for messages that arrived from DP to CP
+ * 
+ */
+void 
+CGlobalTRex::check_for_dp_messages() {
+    /* for all the cores - check for a new message */
+    for (int i = 0; i < get_cores_tx(); i++) {
+        check_for_dp_message_from_core(i);
+    }
+}
 
 bool CGlobalTRex::is_all_links_are_up(bool dump){
     bool all_link_are=true;
@@ -3495,21 +3477,7 @@ int  CGlobalTRex::ixgbe_start(void){
 bool CGlobalTRex::Create(){
     CFlowsYamlInfo     pre_yaml_info;
 
-    if (get_is_stateless()) {
-
-          TrexStatelessCfg cfg;
-
-          TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_TCP, global_platform_cfg_info.m_zmq_rpc_port);
-
-          cfg.m_port_count         = CGlobalInfo::m_options.m_expected_portd;
-          cfg.m_rpc_req_resp_cfg   = &rpc_req_resp_cfg;
-          cfg.m_rpc_async_cfg      = NULL;
-          cfg.m_rpc_server_verbose = false;
-          cfg.m_platform_api       = new TrexDpdkPlatformApi();
-
-          m_trex_stateless = new TrexStateless(cfg);
-
-    } else {
+    if (!get_is_stateless()) {
         pre_yaml_info.load_from_yaml_file(CGlobalInfo::m_options.cfg_file);
     }
 
@@ -3553,6 +3521,24 @@ bool CGlobalTRex::Create(){
    CGlobalInfo::init_pools(rx_mbuf);
    ixgbe_start();
    dump_config(stdout);
+
+   /* start stateless */
+   if (get_is_stateless()) {
+
+       TrexStatelessCfg cfg;
+
+       TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_TCP, global_platform_cfg_info.m_zmq_rpc_port);
+
+       cfg.m_port_count         = CGlobalInfo::m_options.m_expected_portd;
+       cfg.m_rpc_req_resp_cfg   = &rpc_req_resp_cfg;
+       cfg.m_rpc_async_cfg      = NULL;
+       cfg.m_rpc_server_verbose = false;
+       cfg.m_platform_api       = new TrexDpdkPlatformApi();
+       cfg.m_publisher          = &m_zmq_publisher;
+
+       m_trex_stateless = new TrexStateless(cfg);
+   }
+
    return (true);
 
 }
@@ -4119,6 +4105,9 @@ int CGlobalTRex::run_in_master(){
         m_trex_stateless->generate_publish_snapshot(json);
         m_zmq_publisher.publish_json(json);
 
+        /* check from messages from DP */
+        check_for_dp_messages();
+
         delay(500);
 
         if ( is_all_cores_finished() ) {
diff --git a/src/publisher/trex_publisher.cpp b/src/publisher/trex_publisher.cpp
new file mode 100644
index 00000000..35653069
--- /dev/null
+++ b/src/publisher/trex_publisher.cpp
@@ -0,0 +1,107 @@
+/*
+ Itay Marom
+ Cisco Systems, Inc.
+*/
+
+/*
+Copyright (c) 2015-2015 Cisco Systems, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+#include "trex_publisher.h"
+#include <zmq.h>
+#include <assert.h>
+#include <sstream>
+#include <iostream>
+
+/**
+ * create the publisher
+ * 
+ */
+bool 
+TrexPublisher::Create(uint16_t port, bool disable){
+
+    if (disable) {
+        return (true);
+    }
+
+    m_context = zmq_ctx_new();
+    if ( m_context == 0 ) {
+        show_zmq_last_error("can't connect to ZMQ library");
+    }
+
+    m_publisher = zmq_socket (m_context, ZMQ_PUB);
+    if ( m_context == 0 ) {
+        show_zmq_last_error("can't create ZMQ socket");
+    }
+
+    std::stringstream ss;
+    ss << "tcp://*:" << port;
+
+    int rc = zmq_bind (m_publisher, ss.str().c_str());
+    if (rc != 0 ) {
+        show_zmq_last_error("can't bind to ZMQ socket at " + ss.str());
+    }
+
+    std::cout << "zmq publisher at: " << ss.str() << "\n";
+    return (true);
+}
+
+
+void 
+TrexPublisher::Delete(){
+    if (m_publisher) {
+        zmq_close (m_publisher);
+        m_publisher = NULL;
+    }
+    if (m_context) {
+        zmq_ctx_destroy (m_context);
+        m_context = NULL;
+    }
+}
+
+
+void 
+TrexPublisher::publish_json(const std::string &s){
+    if (m_publisher) {
+        int size = zmq_send (m_publisher, s.c_str(), s.length(), 0);
+        assert(size == s.length());
+    }
+}
+
+void
+TrexPublisher::publish_event(event_type_e type, const Json::Value &data) {
+    Json::FastWriter writer;
+    Json::Value value;
+    std::string s;
+    
+    value["name"] = "trex-event";
+    value["type"] = type;
+    value["data"] = data;
+
+    s = writer.write(value);
+    publish_json(s);
+}
+
+/**
+ * error handling
+ * 
+ */
+void 
+TrexPublisher::show_zmq_last_error(const std::string &err){
+    std::cout << " ERROR " << err << "\n";
+    std::cout << " ZMQ: " << zmq_strerror (zmq_errno ());
+    exit(-1);
+}
+
diff --git a/src/publisher/trex_publisher.h b/src/publisher/trex_publisher.h
new file mode 100644
index 00000000..82603fda
--- /dev/null
+++ b/src/publisher/trex_publisher.h
@@ -0,0 +1,54 @@
+/*
+ Itay Marom
+ Cisco Systems, Inc.
+*/
+
+/*
+Copyright (c) 2015-2015 Cisco Systems, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+#ifndef __TREX_PUBLISHER_H__
+#define __TREX_PUBLISHER_H__
+
+#include <stdint.h>
+#include <string>
+#include <json/json.h>
+
+class TrexPublisher {
+
+public:
+
+    TrexPublisher() {
+        m_context = NULL;
+        m_publisher = NULL;
+    }
+
+    bool Create(uint16_t port, bool disable);
+    void Delete();
+    void publish_json(const std::string &s);
+
+    enum event_type_e {
+        EVENT_PORT_STOPPED = 0
+    };
+
+    void publish_event(event_type_e type, const Json::Value &data);
+
+private:
+    void show_zmq_last_error(const std::string &err);
+private:
+    void * m_context;
+    void * m_publisher;
+};
+
+#endif /* __TREX_PUBLISHER_H__ */
diff --git a/src/stateless/cp/trex_dp_port_events.cpp b/src/stateless/cp/trex_dp_port_events.cpp
new file mode 100644
index 00000000..ba327e59
--- /dev/null
+++ b/src/stateless/cp/trex_dp_port_events.cpp
@@ -0,0 +1,220 @@
+/*
+ Itay Marom
+ Cisco Systems, Inc.
+*/
+
+/*
+Copyright (c) 2015-2015 Cisco Systems, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+#include <trex_dp_port_events.h>
+#include <sstream>
+#include <os_time.h>
+#include <trex_stateless.h>
+
+/**
+ * port events
+ */
+void 
+TrexDpPortEvents::create(TrexStatelessPort *port) {
+    m_port = port;
+
+    for (int i = 0; i < TrexDpPortEvent::EVENT_MAX; i++) {
+        m_events[i].create((TrexDpPortEvent::event_e) i, port);
+    }
+
+    m_event_id_counter = EVENT_ID_INVALID;
+}
+
+/**
+ * generate a new event ID
+ * 
+ */
+int
+TrexDpPortEvents::generate_event_id() {
+    return (++m_event_id_counter);
+}
+
+/**
+ * mark the next allowed event 
+ * all other events will be disabled 
+ * 
+ */
+void
+TrexDpPortEvents::wait_for_event(TrexDpPortEvent::event_e ev, int event_id, int timeout_ms) {
+
+    /* first disable all events */
+    for (TrexDpPortEvent & e : m_events) {
+        e.disable();
+    }
+
+    /* mark this event as allowed */
+    m_events[ev].wait_for_event(event_id, timeout_ms);
+}
+
+void 
+TrexDpPortEvents::disable(TrexDpPortEvent::event_e ev) {
+    m_events[ev].disable();
+}
+
+/**
+ * handle an event
+ * 
+ */
+void 
+TrexDpPortEvents::handle_event(TrexDpPortEvent::event_e ev, int thread_id, int event_id) {
+    m_events[ev].handle_event(thread_id, event_id);
+}
+
+/*********** 
+ * single event object
+ * 
+ */
+
+void
+TrexDpPortEvent::create(event_e type, TrexStatelessPort *port) {
+    m_event_type = type;
+    m_port = port;
+
+    /* add the core ids to the hash */
+    m_signal.clear();
+    for (int core_id : m_port->get_core_id_list()) {
+        m_signal[core_id] = false;
+    }
+
+    /* event is disabled */
+    disable();
+}
+
+
+/**
+ * wait the event using event id and timeout
+ * 
+ */
+void
+TrexDpPortEvent::wait_for_event(int event_id, int timeout_ms) {
+
+    /* set a new event id */
+    m_event_id = event_id;
+
+    /* do we have a timeout ? */
+    if (timeout_ms > 0) {
+        m_expire_limit_ms = os_get_time_msec() + timeout_ms;
+    } else {
+        m_expire_limit_ms = -1;
+    }
+
+    /* prepare the signal array */
+    m_pending_cnt = 0;
+    for (auto & core_pair : m_signal) {
+        core_pair.second = false;
+        m_pending_cnt++;
+    }
+}
+
+void
+TrexDpPortEvent::disable() {
+    m_event_id = TrexDpPortEvents::EVENT_ID_INVALID;
+}
+
+/**
+ * get the event status
+ * 
+ */
+
+TrexDpPortEvent::event_status_e
+TrexDpPortEvent::status() {
+
+    /* is it even active ? */
+    if (m_event_id == TrexDpPortEvents::EVENT_ID_INVALID) {
+        return (EVENT_DISABLE);
+    }
+
+    /* did it occured ? */
+    if (m_pending_cnt == 0) {
+        return (EVENT_OCCURED);
+    }
+
+    /* so we are enabled and the event did not occur - maybe we timed out ? */
+    if ( (m_expire_limit_ms > 0) && (os_get_time_msec() > m_expire_limit_ms) ) {
+        return (EVENT_TIMED_OUT);
+    }
+
+    /* so we are still waiting... */
+    return (EVENT_PENDING);
+
+}
+
+void
+TrexDpPortEvent::err(int thread_id, int event_id, const std::string &err_msg)  {
+    std::stringstream err;
+    err << "DP event '" << event_name(m_event_type)  << "' on thread id '" << thread_id << "' with key '" << event_id <<"' - ";
+}
+
+/**
+ * event occured
+ * 
+ */
+void 
+TrexDpPortEvent::handle_event(int thread_id, int event_id) {
+
+    /* if the event is disabled - we don't care */
+    if (!is_active()) {
+        return;
+    }
+
+    /* check the event id is matching the required event - if not maybe its an old signal */
+    if (event_id != m_event_id) {
+        return;
+    }
+
+    /* mark sure no double signal */
+    if (m_signal.at(thread_id)) {
+        err(thread_id, event_id, "double signal");
+
+    } else {
+        /* mark */
+        m_signal.at(thread_id) = true;
+        m_pending_cnt--;
+    }
+
+    /* event occured */
+    if (m_pending_cnt == 0) {
+        m_port->on_dp_event_occured(m_event_type);
+        m_event_id = TrexDpPortEvents::EVENT_ID_INVALID;
+    }
+}
+
+bool
+TrexDpPortEvent::is_active() {
+    return (status() != EVENT_DISABLE);
+}
+
+bool 
+TrexDpPortEvent::has_timeout_expired() {
+    return (status() == EVENT_TIMED_OUT);
+}
+
+const char *
+TrexDpPortEvent::event_name(event_e type) {
+    switch (type) {
+    case EVENT_STOP:
+        return "DP STOP";
+
+    default:
+        throw TrexException("unknown event type");
+    }
+
+}
diff --git a/src/stateless/cp/trex_dp_port_events.h b/src/stateless/cp/trex_dp_port_events.h
new file mode 100644
index 00000000..557e590b
--- /dev/null
+++ b/src/stateless/cp/trex_dp_port_events.h
@@ -0,0 +1,171 @@
+/*
+ Itay Marom
+ Cisco Systems, Inc.
+*/
+
+/*
+Copyright (c) 2015-2015 Cisco Systems, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+#ifndef __TREX_DP_PORT_EVENTS_H__
+#define __TREX_DP_PORT_EVENTS_H__
+
+#include <unordered_map>
+#include <string>
+
+class TrexStatelessPort;
+
+/**
+ * describes a single DP event related to port
+ * 
+ * @author imarom (18-Nov-15)
+ */
+class TrexDpPortEvent {
+public:
+
+    enum event_e {
+        EVENT_STOP   = 1,
+        EVENT_MAX
+    };
+
+    /**
+     * status of the event for the port
+     */
+    enum event_status_e {
+        EVENT_DISABLE,
+        EVENT_PENDING,
+        EVENT_TIMED_OUT,
+        EVENT_OCCURED
+    };
+
+    /**
+     * init for the event
+     * 
+     */
+    void create(event_e type, TrexStatelessPort *port);
+
+    /**
+     * create a new pending event
+     * 
+     */
+    void wait_for_event(int event_id, int timeout_ms = -1);
+
+    /**
+     * mark event as not allowed to happen
+     * 
+     */
+    void disable();
+
+    /**
+     * get the event status
+     * 
+     */
+    event_status_e status();
+
+    /**
+     * event occured
+     * 
+     */
+    void handle_event(int thread_id, int event_id);
+
+    /**
+     * returns true if event is active
+     * 
+     */
+    bool is_active();
+
+    /**
+     * has timeout already expired ?
+     * 
+     */
+    bool has_timeout_expired();
+
+    /**
+     * generate error
+     * 
+     */
+    void err(int thread_id, int event_id, const std::string &err_msg);
+
+    /**
+     * event to name
+     * 
+     */
+    static const char * event_name(event_e type);
+
+
+private:
+
+    event_e                        m_event_type;
+    std::unordered_map<int, bool>  m_signal;
+    int                            m_pending_cnt;
+
+    TrexStatelessPort             *m_port;
+    int                            m_event_id;
+    int                            m_expire_limit_ms;
+    
+};
+
+/**
+ * all the events related to a port
+ * 
+ */
+class TrexDpPortEvents {
+public:
+    friend class TrexDpPortEvent;
+
+    void create(TrexStatelessPort *port);
+
+    /**
+     * generate a new event ID to be used with wait_for_event
+     * 
+     */
+    int generate_event_id();
+
+    /**
+     * wait a new DP event on the port 
+     * returns a key which will be used to identify 
+     * the event happened 
+     * 
+     * @author imarom (18-Nov-15)
+     * 
+     * @param ev - type of event
+     * @param event_id - a unique identifier for the event
+     * @param timeout_ms - does it has a timeout ?
+     * 
+     */
+    void wait_for_event(TrexDpPortEvent::event_e ev, int event_id, int timeout_ms = -1);
+
+    /**
+     * disable an event (don't care)
+     * 
+     */
+    void disable(TrexDpPortEvent::event_e ev);
+
+    /**
+     * event has occured
+     * 
+     */
+    void handle_event(TrexDpPortEvent::event_e ev, int thread_id, int event_id);
+
+private:
+    static const int EVENT_ID_INVALID = -1;
+
+    TrexDpPortEvent m_events[TrexDpPortEvent::EVENT_MAX];
+    int m_event_id_counter;
+
+    TrexStatelessPort *m_port;
+    
+};
+
+#endif /* __TREX_DP_PORT_EVENTS_H__ */
diff --git a/src/stateless/cp/trex_stateless.cpp b/src/stateless/cp/trex_stateless.cpp
index e0e95450..a4522837 100644
--- a/src/stateless/cp/trex_stateless.cpp
+++ b/src/stateless/cp/trex_stateless.cpp
@@ -47,10 +47,12 @@ TrexStateless::TrexStateless(const TrexStatelessCfg &cfg) {
     m_port_count = cfg.m_port_count;
 
     for (int i = 0; i < m_port_count; i++) {
-        m_ports.push_back(new TrexStatelessPort(i));
+        m_ports.push_back(new TrexStatelessPort(i, cfg.m_platform_api));
     }
 
     m_platform_api = cfg.m_platform_api;
+    m_publisher    = cfg.m_publisher;
+
 }
 
 /** 
diff --git a/src/stateless/cp/trex_stateless.h b/src/stateless/cp/trex_stateless.h
index 57c6ef1d..5c11be1e 100644
--- a/src/stateless/cp/trex_stateless.h
+++ b/src/stateless/cp/trex_stateless.h
@@ -30,6 +30,7 @@ limitations under the License.
 #include <trex_stream.h>
 #include <trex_stateless_port.h>
 #include <trex_rpc_server_api.h>
+#include <publisher/trex_publisher.h>
 
 #include <internal_api/trex_platform_api.h>
 
@@ -93,6 +94,7 @@ public:
         m_rpc_async_cfg       = NULL;
         m_rpc_server_verbose  = false;
         m_platform_api        = NULL;
+        m_publisher           = NULL;
     }
 
     const TrexRpcServerConfig  *m_rpc_req_resp_cfg;
@@ -100,6 +102,7 @@ public:
     const TrexPlatformApi      *m_platform_api;
     bool                        m_rpc_server_verbose;
     uint8_t                     m_port_count;
+    TrexPublisher              *m_publisher;
 };
 
 /**
@@ -150,6 +153,10 @@ public:
         return (m_platform_api);
     }
 
+    TrexPublisher * get_publisher() {
+        return m_publisher;
+    }
+
     const std::vector <TrexStatelessPort *> get_port_list() {
         return m_ports;
     }
@@ -170,6 +177,8 @@ protected:
     /* platform API */
     const TrexPlatformApi                *m_platform_api;
 
+    TrexPublisher                        *m_publisher;
+
     std::mutex m_global_cp_lock;
 };
 
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index cbc5a328..fbc5f7c7 100644
--- a/src/stateless/cp/trex_stateless_port.cpp
+++ b/src/stateless/cp/trex_stateless_port.cpp
@@ -52,9 +52,25 @@ using namespace std;
  * trex stateless port
  * 
  **************************/
-TrexStatelessPort::TrexStatelessPort(uint8_t port_id) : m_port_id(port_id) {
+TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api) {
+    std::vector<std::pair<uint8_t, uint8_t>> core_pair_list;
+
+    m_port_id = port_id;
+
     m_port_state = PORT_STATE_IDLE;
     clear_owner();
+
+    /* get the DP cores belonging to this port */
+    api->port_id_to_cores(m_port_id, core_pair_list);
+
+    for (auto core_pair : core_pair_list) {
+
+        /* send the core id */
+        m_cores_id_list.push_back(core_pair.first);
+    }
+
+    /* init the events DP DB */
+    m_dp_events.create(this);
 }
 
 
@@ -105,11 +121,16 @@ TrexStatelessPort::start_traffic(double mul, double duration) {
     }
 
     /* generate a message to all the relevant DP cores to start transmitting */
-    TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(compiled_obj, duration);
+    int event_id = m_dp_events.generate_event_id();
+    /* mark that DP event of stoppped is possible */
+    m_dp_events.wait_for_event(TrexDpPortEvent::EVENT_STOP, event_id);
 
-    send_message_to_dp(start_msg);
+    TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_port_id, event_id, compiled_obj, duration);
 
     change_state(PORT_STATE_TX);
+
+    send_message_to_dp(start_msg);
+    
 }
 
 /**
@@ -126,12 +147,16 @@ TrexStatelessPort::stop_traffic(void) {
         return;
     }
 
+    /* mask out the DP stop event */
+    m_dp_events.disable(TrexDpPortEvent::EVENT_STOP);
+
     /* generate a message to all the relevant DP cores to start transmitting */
     TrexStatelessCpToDpMsgBase *stop_msg = new TrexStatelessDpStop(m_port_id);
 
     send_message_to_dp(stop_msg);
 
     change_state(PORT_STATE_STREAMS);
+    
 }
 
 void
@@ -279,15 +304,36 @@ TrexStatelessPort::encode_stats(Json::Value &port) {
 void 
 TrexStatelessPort::send_message_to_dp(TrexStatelessCpToDpMsgBase *msg) {
 
-    std::vector<std::pair<uint8_t, uint8_t>> cores_id_list;
-
-    get_stateless_obj()->get_platform_api()->port_id_to_cores(m_port_id, cores_id_list);
-
-    for (auto core_pair : cores_id_list) {
+    for (auto core_id : m_cores_id_list) {
 
         /* send the message to the core */
-        CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_pair.first);
+        CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_id);
         ring->Enqueue((CGenNode *)msg->clone());
     }
 
 }
+
+/**
+ * when a DP (async) event occurs - handle it
+ * 
+ */
+void 
+TrexStatelessPort::on_dp_event_occured(TrexDpPortEvent::event_e event_type) {
+    Json::Value data;
+
+    switch (event_type) {
+
+    case TrexDpPortEvent::EVENT_STOP:
+        /* set a stop event */
+        change_state(PORT_STATE_STREAMS);
+        /* send a ZMQ event */
+
+        data["port_id"] = m_port_id;
+        get_stateless_obj()->get_publisher()->publish_event(TrexPublisher::EVENT_PORT_STOPPED, data);
+        break;
+
+    default:
+        assert(0);
+
+    }
+}
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index b533f793..73157c15 100644
--- a/src/stateless/cp/trex_stateless_port.h
+++ b/src/stateless/cp/trex_stateless_port.h
@@ -22,7 +22,9 @@ limitations under the License.
 #define __TREX_STATELESS_PORT_H__
 
 #include <trex_stream.h>
+#include <trex_dp_port_events.h>
 
+class TrexPlatformApi;
 class TrexStatelessCpToDpMsgBase;
 
 /**
@@ -31,6 +33,8 @@ class TrexStatelessCpToDpMsgBase;
  * @author imarom (31-Aug-15)
  */
 class TrexStatelessPort {
+    friend class TrexDpPortEvent;
+
 public:
 
     /**
@@ -54,7 +58,7 @@ public:
         RC_ERR_FAILED_TO_COMPILE_STREAMS
     };
 
-    TrexStatelessPort(uint8_t port_id);
+    TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api);
     
     /**
      * acquire port
@@ -199,6 +203,10 @@ public:
         m_stream_table.get_object_list(object_list);
     }
 
+    TrexDpPortEvents & get_dp_events() {
+        return m_dp_events;
+    }
+
 private:
 
   
@@ -224,6 +232,10 @@ private:
     }
 
 
+    const std::vector<int> get_core_id_list () {
+        return m_cores_id_list;
+    }
+
     bool verify_state(int state, bool should_throw = true) const;
 
     void change_state(port_state_e new_state);
@@ -232,11 +244,23 @@ private:
 
     void send_message_to_dp(TrexStatelessCpToDpMsgBase *msg);
 
-    TrexStreamTable  m_stream_table;
-    uint8_t          m_port_id;
-    port_state_e     m_port_state;
-    std::string      m_owner;
-    std::string      m_owner_handler;
+    /**
+     * triggered when event occurs
+     * 
+     */
+    void on_dp_event_occured(TrexDpPortEvent::event_e event_type);
+
+
+    TrexStreamTable    m_stream_table;
+    uint8_t            m_port_id;
+    port_state_e       m_port_state;
+    std::string        m_owner;
+    std::string        m_owner_handler;
+
+    /* holds the DP cores associated with this port */
+    std::vector<int> m_cores_id_list;
+
+    TrexDpPortEvents   m_dp_events;
 };
 
 #endif /* __TREX_STATELESS_PORT_H__ */
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index a6fe3f56..e17c9075 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -492,6 +492,7 @@ TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj,
     if ( duration > 0.0 ){
         add_port_duration( duration ,obj->get_port_id() );
     }
+
 }
 
 
@@ -522,6 +523,17 @@ TrexStatelessDpCore::stop_traffic(uint8_t port_id) {
 
         schedule_exit();
     }
+ 
+    /* inform the control plane we stopped - this might be a async stop
+       (streams ended)
+     */
+    CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingDpToCp(m_core->m_thread_id);
+    TrexStatelessDpToCpMsgBase *event_msg = new TrexDpPortEventMsg(m_core->m_thread_id,
+                                                                   port_id,
+                                                                   TrexDpPortEvent::EVENT_STOP,
+                                                                   lp_port->get_event_id());
+    ring->Enqueue((CGenNode *)event_msg);
+
 }
 
 /**
diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h
index 85afcf8f..c0bbe702 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -66,6 +66,18 @@ public:
 
     bool update_number_of_active_streams(uint32_t d);
 
+    state_e get_state() {
+        return m_state;
+    }
+
+    void set_event_id(int event_id) {
+        m_event_id = event_id;
+    }
+
+    int get_event_id() {
+        return m_event_id;
+    }
+
 public:
 
     state_e                   m_state;
@@ -75,6 +87,7 @@ public:
                                                 
     std::vector<CDpOneStream> m_active_nodes;   /* holds the current active nodes */
     CFlowGenListPerThread   *  m_core ;
+    int                        m_event_id;
 };
 
 /* for now */
@@ -166,11 +179,13 @@ public:
     /* quit the main loop, work in both stateless in stateful, don't free memory trigger from  master  */
     void quit_main_loop();
 
+    state_e get_state() {
+        return m_state;
+    }
+
     bool set_stateless_next_node(CGenNodeStateless * cur_node,
                                  CGenNodeStateless * next_node);
 
-private:
-
 
     TrexStatelessDpPerPort * get_port_db(uint8_t port_id){
         assert((m_local_port_offset==port_id) ||(m_local_port_offset+1==port_id));
@@ -180,6 +195,9 @@ private:
     }
         
 
+
+private:
+
     void schedule_exit();
 
 
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 856fd9e3..629fe24c 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -22,12 +22,18 @@ limitations under the License.
 #include <trex_stateless_messaging.h>
 #include <trex_stateless_dp_core.h>
 #include <trex_streams_compiler.h>
+#include <trex_stateless.h>
+
 #include <string.h>
 
 /*************************
   start traffic message
  ************************/ 
-TrexStatelessDpStart::TrexStatelessDpStart(TrexStreamsCompiledObj *obj, double duration) : m_obj(obj), m_duration(duration) {
+TrexStatelessDpStart::TrexStatelessDpStart(uint8_t port_id, int event_id, TrexStreamsCompiledObj *obj, double duration) {
+    m_port_id = port_id;
+    m_event_id = event_id;
+    m_obj = obj;
+    m_duration = duration;
 }
 
 
@@ -40,7 +46,7 @@ TrexStatelessDpStart::clone() {
 
     TrexStreamsCompiledObj *new_obj = m_obj->clone();
 
-    TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpStart(new_obj, m_duration);
+    TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpStart(m_port_id, m_event_id, new_obj, m_duration);
 
     return new_msg;
 }
@@ -54,7 +60,12 @@ TrexStatelessDpStart::~TrexStatelessDpStart() {
 bool
 TrexStatelessDpStart::handle(TrexStatelessDpCore *dp_core) {
 
+    /* mark the event id for DP response */
+    dp_core->get_port_db(m_port_id)->set_event_id(m_event_id);
+
+    /* staet traffic */
     dp_core->start_traffic(m_obj, m_duration);
+
     return true;
 }
 
@@ -63,6 +74,10 @@ TrexStatelessDpStart::handle(TrexStatelessDpCore *dp_core) {
  ************************/ 
 bool
 TrexStatelessDpStop::handle(TrexStatelessDpCore *dp_core) {
+    if (dp_core->get_port_db(m_port_id)->get_state() == TrexStatelessDpPerPort::ppSTATE_IDLE) {
+        return true;
+    }
+
     dp_core->stop_traffic(m_port_id);
     return true;
 }
@@ -96,7 +111,6 @@ bool TrexStatelessDpQuit::handle(TrexStatelessDpCore *dp_core){
     return (true);
 }
 
-
 bool TrexStatelessDpCanQuit::handle(TrexStatelessDpCore *dp_core){
 
     if ( dp_core->are_all_ports_idle() ){
@@ -115,3 +129,12 @@ TrexStatelessDpCanQuit::clone(){
 }
 
 
+/************************* messages from DP to CP **********************/
+bool
+TrexDpPortEventMsg::handle() {
+    TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(m_port_id);
+    port->get_dp_events().handle_event(m_event_type, m_thread_id, m_event_id);
+
+    return (true);
+}
+
diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h
index 7dc307c7..2fb5a024 100644
--- a/src/stateless/messaging/trex_stateless_messaging.h
+++ b/src/stateless/messaging/trex_stateless_messaging.h
@@ -23,6 +23,7 @@ limitations under the License.
 #define __TREX_STATELESS_MESSAGING_H__
 
 #include <msg_manager.h>
+#include <trex_dp_port_events.h>
 
 class TrexStatelessDpCore;
 class TrexStreamsCompiledObj;
@@ -42,12 +43,8 @@ public:
     virtual ~TrexStatelessCpToDpMsgBase() {
     }
 
-    /**
-     * virtual function to handle a message
-     * 
-     */
-    virtual bool handle(TrexStatelessDpCore *dp_core) = 0;
 
+    virtual bool handle(TrexStatelessDpCore *dp_core) = 0;
 
     /**
      * clone the current message
@@ -66,7 +63,9 @@ public:
 
     /* no copy constructor */
     TrexStatelessCpToDpMsgBase(TrexStatelessCpToDpMsgBase &) = delete;
-private:
+
+protected:
+    int     m_event_id;
     bool    m_quit_scheduler;
 };
 
@@ -78,18 +77,21 @@ private:
 class TrexStatelessDpStart : public TrexStatelessCpToDpMsgBase {
 public:
 
-    TrexStatelessDpStart(TrexStreamsCompiledObj *obj, double duration);
+    TrexStatelessDpStart(uint8_t m_port_id, int m_event_id, TrexStreamsCompiledObj *obj, double duration);
 
     ~TrexStatelessDpStart();
 
-    virtual bool handle(TrexStatelessDpCore *dp_core);
-
     virtual TrexStatelessCpToDpMsgBase * clone();
 
+    virtual bool handle(TrexStatelessDpCore *dp_core);
 
 private:
+
+    uint8_t                 m_port_id;
+    int                     m_event_id;
     TrexStreamsCompiledObj *m_obj;
-    double m_duration;
+    double                  m_duration;
+
 };
 
 /**
@@ -103,10 +105,10 @@ public:
     TrexStatelessDpStop(uint8_t port_id) : m_port_id(port_id) {
     }
 
-    virtual bool handle(TrexStatelessDpCore *dp_core);
-
     virtual TrexStatelessCpToDpMsgBase * clone();
 
+    virtual bool handle(TrexStatelessDpCore *dp_core);
+
 private:
     uint8_t m_port_id;
 };
@@ -122,9 +124,11 @@ public:
     TrexStatelessDpQuit()  {
     }
 
-    virtual bool handle(TrexStatelessDpCore *dp_core);
 
     virtual TrexStatelessCpToDpMsgBase * clone();
+
+    virtual bool handle(TrexStatelessDpCore *dp_core);
+
 };
 
 /**
@@ -145,4 +149,74 @@ public:
 
 
 
+/************************* messages from DP to CP **********************/
+
+/**
+ * defines the base class for CP to DP messages
+ * 
+ * @author imarom (27-Oct-15)
+ */
+class TrexStatelessDpToCpMsgBase {
+public:
+
+    TrexStatelessDpToCpMsgBase() {
+    }
+
+    virtual ~TrexStatelessDpToCpMsgBase() {
+    }
+
+    /**
+     * virtual function to handle a message
+     * 
+     */
+    virtual bool handle() = 0;
+
+    /* no copy constructor */
+    TrexStatelessDpToCpMsgBase(TrexStatelessDpToCpMsgBase &) = delete;
+
+};
+
+
+/**
+ * a message indicating an event has happened on a port at the 
+ * DP 
+ * 
+ */
+class TrexDpPortEventMsg : public TrexStatelessDpToCpMsgBase {
+public:
+
+    TrexDpPortEventMsg(int thread_id, uint8_t port_id, TrexDpPortEvent::event_e type, int event_id) {
+        m_thread_id  = thread_id;
+        m_port_id    = port_id;
+        m_event_type = type;
+        m_event_id   = event_id;
+    }
+
+    virtual bool handle();
+
+    int get_thread_id() {
+        return m_thread_id;
+    }
+
+    uint8_t get_port_id() {
+        return m_port_id;
+    }
+
+    TrexDpPortEvent::event_e get_event_type() {
+        return m_event_type;
+    }
+
+    int get_event_id() {
+        return m_event_id;
+    }
+
+private:
+    int                         m_thread_id;
+    uint8_t                     m_port_id;
+    TrexDpPortEvent::event_e    m_event_type;
+    int                         m_event_id;
+    
+};
+
 #endif /* __TREX_STATELESS_MESSAGING_H__ */
+