summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rwxr-xr-xlinux_dpdk/ws_main.py1
-rwxr-xr-xsrc/main_dpdk.cpp80
-rw-r--r--src/publisher/trex_publisher.cpp6
-rw-r--r--src/publisher/trex_publisher.h36
-rw-r--r--src/stateless/cp/trex_stateless.cpp2
-rw-r--r--src/stateless/cp/trex_stateless_port.cpp56
-rw-r--r--src/stateless/cp/trex_stateless_port.h38
-rw-r--r--src/stateless/cp/trex_streams_compiler.h4
-rw-r--r--src/stateless/dp/trex_stateless_dp_core.cpp13
-rw-r--r--src/stateless/dp/trex_stateless_dp_core.h9
-rw-r--r--src/stateless/messaging/trex_stateless_messaging.cpp23
-rw-r--r--src/stateless/messaging/trex_stateless_messaging.h76
-rw-r--r--src/trex_dp_port_events.cpp215
-rw-r--r--src/trex_dp_port_events.h165
14 files changed, 642 insertions, 82 deletions
diff --git a/linux_dpdk/ws_main.py b/linux_dpdk/ws_main.py
index 3f63eabc..b7a9c746 100755
--- a/linux_dpdk/ws_main.py
+++ b/linux_dpdk/ws_main.py
@@ -108,6 +108,7 @@ main_src = SrcGroup(dir='src',
'nat_check.cpp',
'msg_manager.cpp',
'publisher/trex_publisher.cpp',
+ 'trex_dp_port_events.cpp',
'pal/linux_dpdk/pal_utl.cpp',
'pal/linux_dpdk/mbuf.cpp'
]);
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index 13eca0d2..3f1605b7 100755
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -2765,6 +2765,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();
@@ -3173,6 +3177,46 @@ 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();
+ }
+
+}
+
+
+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;
@@ -3435,21 +3479,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);
}
@@ -3493,6 +3523,23 @@ 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();
+
+ m_trex_stateless = new TrexStateless(cfg);
+ }
+
return (true);
}
@@ -4059,6 +4106,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
index 49602708..1afb558a 100644
--- a/src/publisher/trex_publisher.cpp
+++ b/src/publisher/trex_publisher.cpp
@@ -81,14 +81,14 @@ TrexPublisher::publish_json(const std::string &s){
}
void
-TrexPublisher::publish_event(TrexPublisherEvent *ev) {
+TrexPublisher::publish_event(event_type_e type, const Json::Value &data) {
Json::FastWriter writer;
Json::Value value;
std::string s;
value["name"] = "event";
- value["type"] = ev->get_type();
- ev->to_json(value["data"]);
+ value["type"] = type;
+ value["data"] = data;
s = writer.write(value);
publish_json(s);
diff --git a/src/publisher/trex_publisher.h b/src/publisher/trex_publisher.h
index 7c3fff92..07d06678 100644
--- a/src/publisher/trex_publisher.h
+++ b/src/publisher/trex_publisher.h
@@ -25,36 +25,6 @@ limitations under the License.
#include <string>
#include <json/json.h>
-/**
- * base event type
- *
- */
-class TrexPublisherEvent {
-public:
- virtual void to_json(Json::Value &json) = 0;
- virtual uint8_t get_type() = 0;
-
-protected:
- enum {
- EVENT_PORT_STOPPED = 1
- };
-
-};
-
-/**
- * port stopped transmitting
- *
- */
-class TrexEventPortStopped : public TrexPublisherEvent {
-public:
- TrexEventPortStopped(uint8_t port_id);
- virtual void to_json(Json::Value &json);
- virtual uint8_t get_type() {
- return (EVENT_PORT_STOPPED);
- }
-};
-
-
class TrexPublisher {
public:
@@ -68,7 +38,11 @@ public:
void Delete();
void publish_json(const std::string &s);
- void publish_event(TrexPublisherEvent *ev);
+ enum event_type_e {
+ EVENT_PORT_STOPPED = 1
+ };
+
+ void publish_event(event_type_e type, const Json::Value &data);
private:
void show_zmq_last_error(const std::string &err);
diff --git a/src/stateless/cp/trex_stateless.cpp b/src/stateless/cp/trex_stateless.cpp
index e0e95450..6ef24a7b 100644
--- a/src/stateless/cp/trex_stateless.cpp
+++ b/src/stateless/cp/trex_stateless.cpp
@@ -47,7 +47,7 @@ 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;
diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp
index cbc5a328..13d0fc9f 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);
+ m_event_id = m_dp_events.generate_event_id();
+ TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(m_event_id, compiled_obj, duration);
+
+ change_state(PORT_STATE_TX);
send_message_to_dp(start_msg);
- change_state(PORT_STATE_TX);
+ /* mark that DP event of stoppped is possible */
+ m_dp_events.wait_for_event(TrexDpPortEvent::EVENT_STOP, m_event_id);
+
}
/**
@@ -279,15 +300,32 @@ 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) {
+ switch (event_type) {
+
+ case TrexDpPortEvent::EVENT_STOP:
+ /* set a stop event */
+ change_state(PORT_STATE_STREAMS);
+ /* send a ZMQ event */
+ break;
+
+ default:
+ assert(0);
+
+ }
+ printf("hey");
+}
diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h
index b533f793..da75284e 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,25 @@ 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<std::pair<uint8_t, uint8_t>> m_cores_id_list;
+ std::vector<int> m_cores_id_list;
+
+ TrexDpPortEvents m_dp_events;
+ int m_event_id;
};
#endif /* __TREX_STATELESS_PORT_H__ */
diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h
index 42cfc5b8..c80dddef 100644
--- a/src/stateless/cp/trex_streams_compiler.h
+++ b/src/stateless/cp/trex_streams_compiler.h
@@ -60,6 +60,10 @@ public:
return (m_mul);
}
+ uint8_t get_port_id() {
+ return m_port_id;
+ }
+
private:
void add_compiled_stream(TrexStream * stream);
std::vector<obj_st> m_objs;
diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp
index 0747c1a0..4a74d9e5 100644
--- a/src/stateless/dp/trex_stateless_dp_core.cpp
+++ b/src/stateless/dp/trex_stateless_dp_core.cpp
@@ -227,6 +227,7 @@ TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj, double duration)
if ( duration > 0.0 ){
add_duration( duration );
}
+
}
void
@@ -261,11 +262,15 @@ TrexStatelessDpCore::stop_traffic(uint8_t port_id) {
m_core->m_node_gen.add_node(node);
}
- /* send a message to the control plane to
- generate an async event that traffic has stopped
+ /* 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);
- //ring->Enqueue((CGenNode *)msg->clone());
+ 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,
+ 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 aaa6eed3..d07e1d3a 100644
--- a/src/stateless/dp/trex_stateless_dp_core.h
+++ b/src/stateless/dp/trex_stateless_dp_core.h
@@ -112,6 +112,14 @@ public:
/* quit the main loop, work in both stateless in stateful, don't free memory trigger from master */
void quit_main_loop();
+ void set_event_id(int event_id) {
+ m_event_id = event_id;
+ }
+
+ int get_event_id() {
+ return m_event_id;
+ }
+
private:
/**
* in idle state loop, the processor most of the time sleeps
@@ -152,6 +160,7 @@ private:
CFlowGenListPerThread *m_core;
double m_duration;
+ int m_event_id;
};
#endif /* __TREX_STATELESS_DP_CORE_H__ */
diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp
index 2e3acffd..a2d00f8b 100644
--- a/src/stateless/messaging/trex_stateless_messaging.cpp
+++ b/src/stateless/messaging/trex_stateless_messaging.cpp
@@ -21,12 +21,17 @@ 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(int event_id, TrexStreamsCompiledObj *obj, double duration) {
+ m_event_id = event_id;
+ m_obj = obj;
+ m_duration = duration;
}
@@ -39,7 +44,7 @@ TrexStatelessDpStart::clone() {
TrexStreamsCompiledObj *new_obj = m_obj->clone();
- TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpStart(new_obj, m_duration);
+ TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpStart(m_event_id, new_obj, m_duration);
return new_msg;
}
@@ -53,7 +58,12 @@ TrexStatelessDpStart::~TrexStatelessDpStart() {
bool
TrexStatelessDpStart::handle(TrexStatelessDpCore *dp_core) {
+ /* mark the event id for DP response */
+ dp_core->set_event_id(m_event_id);
+
+ /* staet traffic */
dp_core->start_traffic(m_obj, m_duration);
+
return true;
}
@@ -96,3 +106,12 @@ bool TrexStatelessDpQuit::handle(TrexStatelessDpCore *dp_core){
return (true);
}
+
+/************************* 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 6473a6a4..3fb1ef84 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;
@@ -41,12 +42,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
@@ -57,6 +54,8 @@ public:
/* no copy constructor */
TrexStatelessCpToDpMsgBase(TrexStatelessCpToDpMsgBase &) = delete;
+protected:
+ int m_event_id;
};
/**
@@ -67,16 +66,17 @@ public:
class TrexStatelessDpStart : public TrexStatelessCpToDpMsgBase {
public:
- TrexStatelessDpStart(TrexStreamsCompiledObj *obj, double duration);
+ TrexStatelessDpStart(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:
+
+ int m_event_id;
TrexStreamsCompiledObj *m_obj;
double m_duration;
};
@@ -92,10 +92,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;
};
@@ -111,10 +111,64 @@ public:
TrexStatelessDpQuit() {
}
- virtual bool handle(TrexStatelessDpCore *dp_core);
virtual TrexStatelessCpToDpMsgBase * clone();
+
+ virtual bool handle(TrexStatelessDpCore *dp_core);
+
+};
+
+/************************* 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();
+
+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__ */
+
diff --git a/src/trex_dp_port_events.cpp b/src/trex_dp_port_events.cpp
new file mode 100644
index 00000000..533ab605
--- /dev/null
+++ b/src/trex_dp_port_events.cpp
@@ -0,0 +1,215 @@
+/*
+ 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);
+}
+
+/**
+ * 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 (event_id != m_event_id) {
+ err(thread_id, event_id, "event key mismatch");
+ }
+
+ /* 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/trex_dp_port_events.h b/src/trex_dp_port_events.h
new file mode 100644
index 00000000..309288df
--- /dev/null
+++ b/src/trex_dp_port_events.h
@@ -0,0 +1,165 @@
+/*
+ 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);
+
+ /**
+ * 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__ */