diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/gtest/trex_stateless_gtest.cpp | 137 | ||||
| -rwxr-xr-x | src/main_dpdk.cpp | 151 | ||||
| -rw-r--r-- | src/publisher/trex_publisher.cpp | 107 | ||||
| -rw-r--r-- | src/publisher/trex_publisher.h | 54 | ||||
| -rw-r--r-- | src/stateless/cp/trex_dp_port_events.cpp | 220 | ||||
| -rw-r--r-- | src/stateless/cp/trex_dp_port_events.h | 171 | ||||
| -rw-r--r-- | src/stateless/cp/trex_stateless.cpp | 4 | ||||
| -rw-r--r-- | src/stateless/cp/trex_stateless.h | 9 | ||||
| -rw-r--r-- | src/stateless/cp/trex_stateless_port.cpp | 64 | ||||
| -rw-r--r-- | src/stateless/cp/trex_stateless_port.h | 36 | ||||
| -rw-r--r-- | src/stateless/dp/trex_stateless_dp_core.cpp | 12 | ||||
| -rw-r--r-- | src/stateless/dp/trex_stateless_dp_core.h | 22 | ||||
| -rw-r--r-- | src/stateless/messaging/trex_stateless_messaging.cpp | 29 | ||||
| -rw-r--r-- | src/stateless/messaging/trex_stateless_messaging.h | 100 |
14 files changed, 989 insertions, 127 deletions
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__ */ + |