From 82e65a02d2f9bdab552521a4859795937821f1be Mon Sep 17 00:00:00 2001 From: imarom Date: Sun, 3 Jan 2016 07:09:23 -0500 Subject: simulation end to end --- linux/ws_main.py | 7 +- src/bp_sim.cpp | 13 +- src/internal_api/trex_platform_api.h | 2 +- src/main.cpp | 681 +++------------------------- src/mock/trex_platform_api_mock.cpp | 4 + src/mock/trex_rpc_server_mock.cpp | 4 - src/publisher/trex_publisher.h | 10 +- src/rpc-server/trex_rpc_async_server.cpp | 4 + src/rpc-server/trex_rpc_async_server.h | 1 + src/rpc-server/trex_rpc_cmd.cpp | 6 +- src/rpc-server/trex_rpc_cmd_api.h | 12 + src/rpc-server/trex_rpc_req_resp_server.cpp | 56 ++- src/rpc-server/trex_rpc_req_resp_server.h | 26 +- src/rpc-server/trex_rpc_server.cpp | 23 +- src/rpc-server/trex_rpc_server_api.h | 25 +- src/sim/trex_sim.h | 132 ++++++ src/sim/trex_sim_stateful.cpp | 599 ++++++++++++++++++++++++ src/sim/trex_sim_stateless.cpp | 210 +++++++++ src/stateless/cp/trex_stateless.h | 6 +- 19 files changed, 1167 insertions(+), 654 deletions(-) create mode 100644 src/sim/trex_sim.h create mode 100644 src/sim/trex_sim_stateful.cpp create mode 100644 src/sim/trex_sim_stateless.cpp diff --git a/linux/ws_main.py b/linux/ws_main.py index deaeeb27..71914630 100755 --- a/linux/ws_main.py +++ b/linux/ws_main.py @@ -121,7 +121,9 @@ main_src = SrcGroup(dir='src', 'latency.cpp', 'pal/linux/pal_utl.cpp', - 'pal/linux/mbuf.cpp' + 'pal/linux/mbuf.cpp', + 'sim/trex_sim_stateless.cpp', + 'sim/trex_sim_stateful.cpp' ]); cmn_src = SrcGroup(dir='src/common', @@ -172,13 +174,14 @@ rpc_server_src = SrcGroup(dir='src/rpc-server/', 'commands/trex_rpc_cmd_general.cpp', 'commands/trex_rpc_cmd_stream.cpp', + # for simulation we use mock API + '../mock/trex_platform_api_mock.cpp', ]) # RPC mock server (test) rpc_server_mock_src = SrcGroup(dir='src/mock/', src_list=[ 'trex_rpc_server_mock.cpp', - 'trex_platform_api_mock.cpp', '../gtest/rpc_test.cpp', ]) diff --git a/src/bp_sim.cpp b/src/bp_sim.cpp index 081553b5..6342b7f3 100755 --- a/src/bp_sim.cpp +++ b/src/bp_sim.cpp @@ -3146,7 +3146,7 @@ int CNodeGenerator::open_file(std::string file_name, /* ser preview mode */ m_v_if->set_review_mode(preview_mode); m_v_if->open_file(file_name); - m_cnt = 1; + m_cnt = 0; return (0); } @@ -3159,11 +3159,13 @@ int CNodeGenerator::close_file(CFlowGenListPerThread * thread){ } int CNodeGenerator::update_stl_stats(CGenNodeStateless *node_sl){ + m_cnt++; + if ( m_preview_mode.getVMode() >2 ){ fprintf(stdout," %4lu ,", (ulong)m_cnt); node_sl->Dump(stdout); - m_cnt++; } + return (0); } @@ -3199,7 +3201,9 @@ bool CFlowGenListPerThread::Create(uint32_t thread_id, char name[100]; sprintf(name,"nodes-%d",m_core_id); - printf(" create thread %d %s socket: %d \n",m_core_id,name,socket_id); + + //printf(" create thread %d %s socket: %d \n",m_core_id,name,socket_id); + m_node_pool = utl_rte_mempool_create_non_pkt(name, CGlobalInfo::m_memory_cfg.get_each_core_dp_flows(), sizeof(CGenNode), @@ -3207,7 +3211,8 @@ bool CFlowGenListPerThread::Create(uint32_t thread_id, 0 , socket_id); - printf(" pool %p \n",m_node_pool); + //printf(" pool %p \n",m_node_pool); + m_node_gen.Create(this); m_flow_id_to_node_lookup.Create(); diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h index 343b8004..3ae49da8 100644 --- a/src/internal_api/trex_platform_api.h +++ b/src/internal_api/trex_platform_api.h @@ -138,7 +138,7 @@ public: */ class TrexMockPlatformApi : public TrexPlatformApi { public: - void port_id_to_cores(uint8_t port_id, std::vector> &cores_id_list) const {} + void port_id_to_cores(uint8_t port_id, std::vector> &cores_id_list) const; void get_global_stats(TrexPlatformGlobalStats &stats) const; void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const; void get_interface_info(uint8_t interface_id, std::string &driver_name, driver_speed_e &speed) const { diff --git a/src/main.cpp b/src/main.cpp index 1b219a8c..3fa7c7c3 100755 --- a/src/main.cpp +++ b/src/main.cpp @@ -23,16 +23,34 @@ limitations under the License. #include "bp_sim.h" #include "os_time.h" +#include #include #include #include +#include +using namespace std; // An enum for all the option types enum { OPT_HELP, OPT_CFG, OPT_NODE_DUMP, OP_STATS, - OPT_FILE_OUT, OPT_UT, OPT_PCAP, OPT_IPV6, OPT_MAC_FILE}; - + OPT_FILE_OUT, OPT_UT, OPT_PCAP, OPT_IPV6, OPT_MAC_FILE, + OPT_SL}; + + + +/** + * type of run + * GTEST + * Stateful + * Stateless + */ +typedef enum { + OPT_TYPE_GTEST = 7, + OPT_TYPE_SF, + OPT_TYPE_SL +} opt_type_e; + /* these are the argument types: SO_NONE -- no argument needed @@ -52,7 +70,7 @@ static CSimpleOpt::SOption parser_options[] = { OPT_NODE_DUMP , "-v", SO_REQ_SEP }, { OPT_PCAP, "--pcap", SO_NONE }, { OPT_IPV6, "--ipv6", SO_NONE }, - + { OPT_SL, "--sl", SO_NONE }, SO_END_OF_OPTIONS }; @@ -94,9 +112,8 @@ static int usage(){ return (0); } -int gtest_main(int argc, char **argv) ; -static int parse_options(int argc, char *argv[], CParserOption* po, bool & is_gtest ) { +static int parse_options(int argc, char *argv[], CParserOption* po, opt_type_e &type) { CSimpleOpt args(argc, argv, parser_options); int a=0; @@ -104,36 +121,51 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool & is_gt po->preview.clean(); po->preview.setFileWrite(true); + /* by default - type is stateful */ + type = OPT_TYPE_SF; + while ( args.Next() ){ if (args.LastError() == SO_SUCCESS) { switch (args.OptionId()) { case OPT_UT : - is_gtest=true; + type = OPT_TYPE_GTEST; return (0); break; + case OPT_HELP: usage(); return -1; + + case OPT_SL: + type = OPT_TYPE_SL; + break; + case OPT_CFG: po->cfg_file = args.OptionArg(); break; + case OPT_MAC_FILE: po->mac_file = args.OptionArg(); break; + case OPT_FILE_OUT: po->out_file = args.OptionArg(); break; + case OPT_IPV6: po->preview.set_ipv6_mode_enable(true); break; + case OPT_NODE_DUMP: a=atoi(args.OptionArg()); node_dump=1; po->preview.setFileWrite(false); break; + case OPT_PCAP: po->preview.set_pcap_mode_enable(true); break; + default: usage(); return -1; @@ -165,637 +197,34 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool & is_gt return 0; } -int cores=1; - -/* - -int curent_time(){ - - time_init(); - - int i; - for (i=0; i<100000000; i++){ - now=now_sec(); - } - return (0); -}*/ - -#ifdef LINUX - - - -#include - -struct per_thread_t { - pthread_t tid; -}; - -#define MAX_THREADS 200 -static per_thread_t tr_info[MAX_THREADS]; - - -////////////// - -struct test_t_info1 { - CPreviewMode * preview_info; - CFlowGenListPerThread * thread_info; - uint32_t thread_id; -}; - -void * thread_task(void *info){ - - test_t_info1 * obj =(test_t_info1 *)info; - - CFlowGenListPerThread * lpt=obj->thread_info; - - printf("start thread %d \n",obj->thread_id); - //delay(obj->thread_id *3000); - printf("-->start thread %d \n",obj->thread_id); - if (1/*obj->thread_id ==3*/) { - - char buf[100]; - sprintf(buf,"my%d.erf",obj->thread_id); - lpt->start_generate_stateful(buf,*obj->preview_info); - lpt->m_node_gen.DumpHist(stdout); - printf("end thread %d \n",obj->thread_id); - } - - return (NULL); -} - - -void test_load_list_of_cap_files_linux(CParserOption * op){ - - CFlowGenList fl; - //CNullIF erf_vif; - //CErfIF erf_vif; - - fl.Create(); - - fl.load_from_yaml(op->cfg_file,cores); - fl.DumpPktSize(); - - - fl.generate_p_thread_info(cores); - CFlowGenListPerThread * lpt; - - /* set the ERF file */ - //fl.set_vif_all(&erf_vif); - - int i; - for (i=0; ipreview_info =&op->preview; - obj->thread_info = fl.m_threads_info[i]; - obj->thread_id = i; - CNullIF * erf_vif = new CNullIF(); - //CErfIF * erf_vif = new CErfIF(); - - lpt->set_vif(erf_vif); - - assert(pthread_create( &tr_info[i].tid, NULL, thread_task, obj)==0); - } - - for (i=0; icfg_file,NUM); - fl.DumpPktSize(); - - - fl.generate_p_thread_info(NUM); - CFlowGenListPerThread * lpt; - - /* set the ERF file */ - //fl.set_vif_all(&erf_vif); - - int i; - for (i=0; istart_generate_stateful(buf,op->preview); - lpt->m_node_gen.DumpHist(stdout); - } - //sprintf(buf,"my%d.erf",7); - //lpt=fl.m_threads_info[7]; - - //fl.Dump(stdout); - fl.Delete(); -} - -int load_list_of_cap_files(CParserOption * op){ - CFlowGenList fl; - fl.Create(); - fl.load_from_yaml(op->cfg_file,1); - if ( op->preview.getVMode() >0 ) { - fl.DumpCsv(stdout); - } - uint32_t start= os_get_time_msec(); - - CErfIF erf_vif; - //CNullIF erf_vif; - - fl.generate_p_thread_info(1); - CFlowGenListPerThread * lpt; - lpt=fl.m_threads_info[0]; - lpt->set_vif(&erf_vif); - - if ( (op->preview.getVMode() >1) || op->preview.getFileWrite() ) { - lpt->start_generate_stateful(op->out_file,op->preview); - } - - lpt->m_node_gen.DumpHist(stdout); - - uint32_t stop= os_get_time_msec(); - printf(" d time = %ul %ul \n",stop-start,os_get_time_freq()); - fl.Delete(); - return (0); -} - - -int test_dns(){ - - time_init(); - CGlobalInfo::init_pools(1000); - - CParserOption po ; - - //po.cfg_file = "cap2/dns.yaml"; - //po.cfg_file = "cap2/sfr3.yaml"; - po.cfg_file = "cap2/sfr.yaml"; - - po.preview.setVMode(0); - po.preview.setFileWrite(true); - #ifdef LINUX - test_load_list_of_cap_files_linux(&po); - #else - test_load_list_of_cap_files(&po); - #endif - return (0); -} - -void test_pkt_mbuf(void); - -void test_compare_files(void); - -#if 0 -static int b=0; -static int c=0; -static int d=0; - -int test_instructions(){ - int i; - for (i=0; i<100000;i++) { - b+=b+1; - c+=+b+c+1; - d+=+(b*2+1); - } - return (b+c+d); -} - -#include -#endif - - -void update_tcp_seq_num(CCapFileFlowInfo * obj, - int pkt_id, - int size_change){ - CFlowPktInfo * pkt=obj->GetPacket(pkt_id); - if ( pkt->m_pkt_indication.m_desc.IsUdp() ){ - /* nothing to do */ - return; - } - - bool o_init=pkt->m_pkt_indication.m_desc.IsInitSide(); - TCPHeader * tcp ; - int s= (int)obj->Size(); - int i; - - for (i=pkt_id+1; iGetPacket(i); - tcp=pkt->m_pkt_indication.l4.m_tcp; - bool init=pkt->m_pkt_indication.m_desc.IsInitSide(); - if (init == o_init) { - /* same dir update the seq number */ - tcp->setSeqNumber (tcp->getSeqNumber ()+size_change); + opt_type_e type; - }else{ - /* update the ack number */ - tcp->setAckNumber (tcp->getAckNumber ()+size_change); - } + if ( parse_options(argc, argv, &CGlobalInfo::m_options , type) != 0) { + exit(-1); } -} - - - -void change_pkt_len(CCapFileFlowInfo * obj,int pkt_id, int size ){ - CFlowPktInfo * pkt=obj->GetPacket(pkt_id); - - /* enlarge the packet size by 9 */ - - char * p=pkt->m_packet->append(size); - /* set it to 0xaa*/ - memmove(p+size-4,p-4,4); /* CRCbytes */ - memset(p-4,0x0a,size); - - /* refresh the pointers */ - pkt->m_pkt_indication.RefreshPointers(); - - IPHeader * ipv4 = pkt->m_pkt_indication.l3.m_ipv4; - ipv4->updateTotalLength (ipv4->getTotalLength()+size ); - - /* update seq numbers if needed */ - update_tcp_seq_num(obj,pkt_id,size); -} -void dump_tcp_seq_num_(CCapFileFlowInfo * obj){ - int s= (int)obj->Size(); - int i; - uint32_t i_seq; - uint32_t r_seq; - - CFlowPktInfo * pkt=obj->GetPacket(0); - TCPHeader * tcp = pkt->m_pkt_indication.l4.m_tcp; - i_seq=tcp->getSeqNumber (); - - pkt=obj->GetPacket(1); - tcp = pkt->m_pkt_indication.l4.m_tcp; - r_seq=tcp->getSeqNumber (); - - for (i=2; iGetPacket(i); - tcp=pkt->m_pkt_indication.l4.m_tcp; - bool init=pkt->m_pkt_indication.m_desc.IsInitSide(); - seq=tcp->getSeqNumber (); - ack=tcp->getAckNumber (); - if (init) { - seq=seq-i_seq; - ack=ack-r_seq; - }else{ - seq=seq-r_seq; - ack=ack-i_seq; + switch (type) { + case OPT_TYPE_GTEST: + { + SimGtest test; + return test.run(argc, argv); } - printf(" %4d ",i); - if (!init) { - printf(" "); - } - printf(" %s seq: %4d ack : %4d \n",init?"I":"R",seq,ack); - } -} - - -int manipolate_capfile() { - time_init(); - CGlobalInfo::init_pools(1000); - - CCapFileFlowInfo flow_info; - flow_info.Create(); - - flow_info.load_cap_file("avl/delay_10_rtsp_0.pcap",0,0); - - change_pkt_len(&flow_info,4-1 ,6); - change_pkt_len(&flow_info,5-1 ,6); - change_pkt_len(&flow_info,6-1 ,6+2); - change_pkt_len(&flow_info,7-1 ,4); - change_pkt_len(&flow_info,8-1 ,6+2); - change_pkt_len(&flow_info,9-1 ,4); - change_pkt_len(&flow_info,10-1,6); - change_pkt_len(&flow_info,13-1,6); - change_pkt_len(&flow_info,16-1,6); - change_pkt_len(&flow_info,19-1,6); - - flow_info.save_to_erf("exp/c.pcap",1); - - return (1); -} - -int manipolate_capfile_sip() { - time_init(); - CGlobalInfo::init_pools(1000); - - CCapFileFlowInfo flow_info; - flow_info.Create(); - - flow_info.load_cap_file("avl/delay_10_sip_0.pcap",0,0); - - change_pkt_len(&flow_info,1-1 ,6+6); - change_pkt_len(&flow_info,2-1 ,6+6); - - flow_info.save_to_erf("exp/delay_10_sip_0_fixed.pcap",1); - - return (1); -} - -int manipolate_capfile_sip1() { - time_init(); - CGlobalInfo::init_pools(1000); - - CCapFileFlowInfo flow_info; - flow_info.Create(); - - flow_info.load_cap_file("avl/delay_sip_0.pcap",0,0); - flow_info.GetPacket(1); - - change_pkt_len(&flow_info,1-1 ,6+6+10); - - change_pkt_len(&flow_info,2-1 ,6+6+10); - - flow_info.save_to_erf("exp/delay_sip_0_fixed_1.pcap",1); - - return (1); -} - - -class CMergeCapFileRec { -public: - - CCapFileFlowInfo m_cap; - - int m_index; - int m_limit_number_of_packets; /* limit number of packets */ - bool m_stop; /* Do we have more packets */ - - double m_offset; /* offset should be positive */ - double m_start_time; - -public: - bool Create(std::string cap_file,double offset); - void Delete(); - void IncPacket(); - bool GetCurPacket(double & time); - CPacketIndication * GetUpdatedPacket(); - - void Dump(FILE *fd,int _id); -}; - - -void CMergeCapFileRec::Dump(FILE *fd,int _id){ - double time = 0.0; - bool stop=GetCurPacket(time); - fprintf (fd," id:%2d stop : %d index:%4d %3.4f \n",_id,stop?1:0,m_index,time); -} - - -CPacketIndication * CMergeCapFileRec::GetUpdatedPacket(){ - double t1; - assert(GetCurPacket(t1)==false); - CFlowPktInfo * pkt = m_cap.GetPacket(m_index); - pkt->m_pkt_indication.m_packet->set_new_time(t1); - return (&pkt->m_pkt_indication); -} - - -bool CMergeCapFileRec::GetCurPacket(double & time){ - if (m_stop) { - return(true); - } - CFlowPktInfo * pkt = m_cap.GetPacket(m_index); - time= (pkt->m_packet->get_time() -m_start_time + m_offset); - return (false); -} - -void CMergeCapFileRec::IncPacket(){ - m_index++; - if ( (m_limit_number_of_packets) && (m_index > m_limit_number_of_packets ) ) { - m_stop=true; - return; - } - - if ( m_index == (int)m_cap.Size() ) { - m_stop=true; - } -} - -void CMergeCapFileRec::Delete(){ - m_cap.Delete(); -} - -bool CMergeCapFileRec::Create(std::string cap_file, - double offset){ - m_cap.Create(); - m_cap.load_cap_file(cap_file,0,0); - CFlowPktInfo * pkt = m_cap.GetPacket(0); - - m_index=0; - m_stop=false; - m_limit_number_of_packets =0; - m_start_time = pkt->m_packet->get_time() ; - m_offset = offset; - - return (true); -} - - -#define MERGE_CAP_FILES (2) - -class CMergeCapFile { -public: - bool Create(); - void Delete(); - bool run_merge(std::string to_cap_file); -private: - void append(int _cap_id); - -public: - CMergeCapFileRec m[MERGE_CAP_FILES]; - CCapFileFlowInfo m_results; -}; - -bool CMergeCapFile::Create(){ - m_results.Create(); - return(true); -} - -void CMergeCapFile::Delete(){ - m_results.Delete(); -} - -void CMergeCapFile::append(int _cap_id){ - CPacketIndication * lp=m[_cap_id].GetUpdatedPacket(); - lp->m_packet->Dump(stdout,0); - m_results.Append(lp); -} - - -bool CMergeCapFile::run_merge(std::string to_cap_file){ - - int i=0; - int cnt=0; - while ( true ) { - int min_index=0; - double min_time; - - fprintf(stdout," --------------\n"); - fprintf(stdout," pkt : %d \n",cnt); - for (i=0; iCreate(4) ); - - - bool is_gtest=false; - - if ( parse_options(argc, argv, &CGlobalInfo::m_options , is_gtest) != 0){ - exit(-1); } - - if ( is_gtest ) { - res = gtest_main(argc, argv); - }else{ - res = load_list_of_cap_files(&CGlobalInfo::m_options); - } - - CMsgIns::Ins()->Free(); - CGlobalInfo::free_pools(); - CGlobalInfo::m_socket.Delete(); - - - return (res); - } diff --git a/src/mock/trex_platform_api_mock.cpp b/src/mock/trex_platform_api_mock.cpp index 54f71e10..416c4b69 100644 --- a/src/mock/trex_platform_api_mock.cpp +++ b/src/mock/trex_platform_api_mock.cpp @@ -47,3 +47,7 @@ TrexMockPlatformApi::get_dp_core_count() const { return (1); } +void +TrexMockPlatformApi::port_id_to_cores(uint8_t port_id, std::vector> &cores_id_list) const { + cores_id_list.push_back(std::make_pair(0, 0)); +} diff --git a/src/mock/trex_rpc_server_mock.cpp b/src/mock/trex_rpc_server_mock.cpp index 0bdf6cf1..ecfa308d 100644 --- a/src/mock/trex_rpc_server_mock.cpp +++ b/src/mock/trex_rpc_server_mock.cpp @@ -78,10 +78,6 @@ find_free_tcp_port(uint16_t start_port = 5050) { return port; } -TrexStateless * get_stateless_obj() { - return g_trex_stateless; -} - uint16_t gtest_get_mock_server_port() { return g_rpc_port; } diff --git a/src/publisher/trex_publisher.h b/src/publisher/trex_publisher.h index bd4392f7..52978476 100644 --- a/src/publisher/trex_publisher.h +++ b/src/publisher/trex_publisher.h @@ -34,9 +34,11 @@ public: m_publisher = NULL; } - bool Create(uint16_t port, bool disable); - void Delete(); - void publish_json(const std::string &s); + virtual ~TrexPublisher() {} + + virtual bool Create(uint16_t port, bool disable); + virtual void Delete(); + virtual void publish_json(const std::string &s); enum event_type_e { EVENT_PORT_STARTED = 0, @@ -51,7 +53,7 @@ public: }; - void publish_event(event_type_e type, const Json::Value &data = Json::nullValue); + virtual void publish_event(event_type_e type, const Json::Value &data = Json::nullValue); private: void show_zmq_last_error(const std::string &err); diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp index 46fe499b..6e5fbfc6 100644 --- a/src/rpc-server/trex_rpc_async_server.cpp +++ b/src/rpc-server/trex_rpc_async_server.cpp @@ -41,6 +41,10 @@ TrexRpcServerAsync::TrexRpcServerAsync(const TrexRpcServerConfig &cfg, std::mute m_context = zmq_ctx_new(); } +void +TrexRpcServerAsync::_prepare() { +} + /** * publisher thread * diff --git a/src/rpc-server/trex_rpc_async_server.h b/src/rpc-server/trex_rpc_async_server.h index 02d1490e..80d92c2f 100644 --- a/src/rpc-server/trex_rpc_async_server.h +++ b/src/rpc-server/trex_rpc_async_server.h @@ -36,6 +36,7 @@ public: TrexRpcServerAsync(const TrexRpcServerConfig &cfg, std::mutex *lock = NULL); protected: + void _prepare(); void _rpc_thread_cb(); void _stop_rpc_thread(); diff --git a/src/rpc-server/trex_rpc_cmd.cpp b/src/rpc-server/trex_rpc_cmd.cpp index b5dd121c..aea7980b 100644 --- a/src/rpc-server/trex_rpc_cmd.cpp +++ b/src/rpc-server/trex_rpc_cmd.cpp @@ -32,7 +32,7 @@ TrexRpcCommand::run(const Json::Value ¶ms, Json::Value &result) { check_param_count(params, m_param_count, result); - if (m_needs_ownership) { + if (m_needs_ownership && !g_test_override_ownership) { verify_ownership(params, result); } @@ -372,3 +372,7 @@ TrexRpcCommand::generate_execute_err(Json::Value &result, const std::string &msg throw (TrexRpcCommandException(TREX_RPC_CMD_EXECUTE_ERR)); } +/** + * by default this is off + */ +bool TrexRpcCommand::g_test_override_ownership = false; diff --git a/src/rpc-server/trex_rpc_cmd_api.h b/src/rpc-server/trex_rpc_cmd_api.h index 7cbdf4ff..675d2900 100644 --- a/src/rpc-server/trex_rpc_cmd_api.h +++ b/src/rpc-server/trex_rpc_cmd_api.h @@ -89,6 +89,16 @@ public: return m_name; } + /** + * on test we enable this override + * + * + * @param enable + */ + static void test_set_override_ownership(bool enable) { + g_test_override_ownership = enable; + } + virtual ~TrexRpcCommand() {} protected: @@ -241,6 +251,8 @@ protected: std::string m_name; int m_param_count; bool m_needs_ownership; + + static bool g_test_override_ownership; }; #endif /* __TREX_RPC_CMD_API_H__ */ diff --git a/src/rpc-server/trex_rpc_req_resp_server.cpp b/src/rpc-server/trex_rpc_req_resp_server.cpp index eb7825ac..1e8e177d 100644 --- a/src/rpc-server/trex_rpc_req_resp_server.cpp +++ b/src/rpc-server/trex_rpc_req_resp_server.cpp @@ -36,7 +36,10 @@ limitations under the License. * */ TrexRpcServerReqRes::TrexRpcServerReqRes(const TrexRpcServerConfig &cfg, std::mutex *lock) : TrexRpcServerInterface(cfg, "req resp", lock) { - /* ZMQ is not thread safe - this should be outside */ + +} + +void TrexRpcServerReqRes::_prepare() { m_context = zmq_ctx_new(); } @@ -123,15 +126,28 @@ TrexRpcServerReqRes::fetch_one_request(std::string &msg) { */ void TrexRpcServerReqRes::_stop_rpc_thread() { /* by calling zmq_term we signal the blocked thread to exit */ - zmq_term(m_context); + if (m_context) { + zmq_term(m_context); + } } + /** * handles a request given to the server * respondes to the request */ void TrexRpcServerReqRes::handle_request(const std::string &request) { + std::string response_str = process_request(request); + zmq_send(m_socket, response_str.c_str(), response_str.size(), 0); +} + +/** + * main processing of the request + * + */ +std::string TrexRpcServerReqRes::process_request(const std::string &request) { + std::vector commands; Json::FastWriter writer; @@ -175,8 +191,7 @@ void TrexRpcServerReqRes::handle_request(const std::string &request) { verbose_json("Server Replied: ", response_str); - zmq_send(m_socket, response_str.c_str(), response_str.size(), 0); - + return response_str; } /** @@ -198,3 +213,36 @@ TrexRpcServerReqRes::handle_server_error(const std::string &specific_err) { zmq_send(m_socket, response_str.c_str(), response_str.size(), 0); } + + + +std::string +TrexRpcServerReqRes::test_inject_request(const std::string &req) { + return process_request(req); +} + + +/** + * MOCK req resp server + */ +TrexRpcServerReqResMock::TrexRpcServerReqResMock(const TrexRpcServerConfig &cfg) : TrexRpcServerReqRes(cfg) { +} + +/** + * override start + * + */ +void +TrexRpcServerReqResMock::start() { + +} + + +/** + * override stop + */ +void +TrexRpcServerReqResMock::stop() { + +} + diff --git a/src/rpc-server/trex_rpc_req_resp_server.h b/src/rpc-server/trex_rpc_req_resp_server.h index 2876206c..97efbe08 100644 --- a/src/rpc-server/trex_rpc_req_resp_server.h +++ b/src/rpc-server/trex_rpc_req_resp_server.h @@ -34,13 +34,19 @@ public: TrexRpcServerReqRes(const TrexRpcServerConfig &cfg, std::mutex *lock = NULL); + /* for test purposes - bypass the ZMQ and inject a message */ + std::string test_inject_request(const std::string &req); + protected: + + void _prepare(); void _rpc_thread_cb(); void _stop_rpc_thread(); -private: bool fetch_one_request(std::string &msg); void handle_request(const std::string &request); + std::string process_request(const std::string &request); + void handle_server_error(const std::string &specific_err); void *m_context; @@ -48,4 +54,22 @@ private: }; +/** + * a mock req resp server (for tests) + * + * @author imarom (03-Jan-16) + */ +class TrexRpcServerReqResMock : public TrexRpcServerReqRes { + +public: + TrexRpcServerReqResMock(const TrexRpcServerConfig &cfg); + + /* override the interface functions */ + virtual void start(); + virtual void stop(); + + +}; + #endif /* __TREX_RPC_REQ_RESP_API_H__ */ + diff --git a/src/rpc-server/trex_rpc_server.cpp b/src/rpc-server/trex_rpc_server.cpp index a14e6f97..1dfc4494 100644 --- a/src/rpc-server/trex_rpc_server.cpp +++ b/src/rpc-server/trex_rpc_server.cpp @@ -63,6 +63,9 @@ void TrexRpcServerInterface::start() { verbose_msg("Starting RPC Server"); + /* prepare for run */ + _prepare(); + m_thread = new std::thread(&TrexRpcServerInterface::_rpc_thread_cb, this); if (!m_thread) { throw TrexRpcException("unable to create RPC thread"); @@ -117,9 +120,18 @@ TrexRpcServer::TrexRpcServer(const TrexRpcServerConfig *req_resp_cfg, const TrexRpcServerConfig *async_cfg, std::mutex *lock) { + m_req_resp = NULL; + /* add the request response server */ if (req_resp_cfg) { - m_servers.push_back(new TrexRpcServerReqRes(*req_resp_cfg, lock)); + + if (req_resp_cfg->get_protocol() == TrexRpcServerConfig::RPC_PROT_MOCK) { + m_req_resp = new TrexRpcServerReqResMock(*req_resp_cfg); + } else { + m_req_resp = new TrexRpcServerReqRes(*req_resp_cfg, lock); + } + + m_servers.push_back(m_req_resp); } /* add async publisher */ @@ -166,3 +178,12 @@ void TrexRpcServer::set_verbose(bool verbose) { } } + +std::string +TrexRpcServer::test_inject_request(const std::string &req_str) { + if (m_req_resp) { + return m_req_resp->test_inject_request(req_str); + } else { + return ""; + } +} diff --git a/src/rpc-server/trex_rpc_server_api.h b/src/rpc-server/trex_rpc_server_api.h index ff876ac4..1ab5dce9 100644 --- a/src/rpc-server/trex_rpc_server_api.h +++ b/src/rpc-server/trex_rpc_server_api.h @@ -29,8 +29,10 @@ limitations under the License. #include #include #include +#include class TrexRpcServerInterface; +class TrexRpcServerReqRes; /** * defines a configuration of generic RPC server @@ -41,18 +43,19 @@ class TrexRpcServerConfig { public: enum rpc_prot_e { - RPC_PROT_TCP + RPC_PROT_TCP, + RPC_PROT_MOCK }; TrexRpcServerConfig(rpc_prot_e protocol, uint16_t port) : m_protocol(protocol), m_port(port) { } - uint16_t get_port() { + uint16_t get_port() const { return m_port; } - rpc_prot_e get_protocol() { + rpc_prot_e get_protocol() const { return m_protocol; } @@ -76,13 +79,13 @@ public: * starts the server * */ - void start(); + virtual void start(); /** * stops the server * */ - void stop(); + virtual void stop(); /** * set verbose on or off @@ -107,6 +110,7 @@ protected: * instances implement this * */ + virtual void _prepare() = 0; virtual void _rpc_thread_cb() = 0; virtual void _stop_rpc_thread() = 0; @@ -169,12 +173,23 @@ public: } + /** + * allow injecting of a JSON and get a response + * + * @author imarom (27-Dec-15) + * + * @return std::string + */ + std::string test_inject_request(const std::string &request_str); private: static std::string generate_handler(); std::vector m_servers; + // an alias to the req resp server + TrexRpcServerReqRes *m_req_resp; + bool m_verbose; static const std::string s_server_uptime; diff --git a/src/sim/trex_sim.h b/src/sim/trex_sim.h new file mode 100644 index 00000000..d997e2f5 --- /dev/null +++ b/src/sim/trex_sim.h @@ -0,0 +1,132 @@ +/* + 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_SIM_H__ +#define __TREX_SIM_H__ + +#include +#include +#include + +int gtest_main(int argc, char **argv); + +class TrexStateless; +class TrexPublisher; +class DpToCpHandler; + +/** + * interface for a sim target + * + */ +class SimInterface { +public: + + SimInterface() { + + time_init(); + CGlobalInfo::m_socket.Create(0); + CGlobalInfo::init_pools(1000); + assert( CMsgIns::Ins()->Create(4) ); + } + + virtual ~SimInterface() { + + CMsgIns::Ins()->Free(); + CGlobalInfo::free_pools(); + CGlobalInfo::m_socket.Delete(); + } + + +}; + +/** + * gtest target + * + * @author imarom (28-Dec-15) + */ +class SimGtest : public SimInterface { +public: + + int run(int argc, char **argv) { + return gtest_main(argc, argv); + } +}; + + + +/** + * stateful target + * + */ +class SimStateful : public SimInterface { + +public: + int run(); +}; + + + +/** + * target for sim stateless + * + * @author imarom (28-Dec-15) + */ +class SimStateless : public SimInterface { + +public: + static SimStateless& get_instance() { + static SimStateless instance; + return instance; + } + + + int run(const std::string &json_filename, const std::string &out_filename); + + TrexStateless * get_stateless_obj() { + return m_trex_stateless; + } + + void set_verbose(bool enable) { + m_verbose = enable; + } + +private: + SimStateless(); + ~SimStateless(); + + void prepare_control_plane(); + void prepare_dataplane(); + void execute_json(const std::string &json_filename); + void run_dp(const std::string &out_filename); + void flush_dp_to_cp_messages(); + + bool is_verbose() { + return m_verbose; + } + + TrexStateless *m_trex_stateless; + DpToCpHandler *m_dp_to_cp_handler; + TrexPublisher *m_publisher; + CFlowGenList m_fl; + CErfIFStl m_erf_vif; + bool m_verbose; +}; + +#endif /* __TREX_SIM_H__ */ diff --git a/src/sim/trex_sim_stateful.cpp b/src/sim/trex_sim_stateful.cpp new file mode 100644 index 00000000..35c17d6e --- /dev/null +++ b/src/sim/trex_sim_stateful.cpp @@ -0,0 +1,599 @@ +/* + 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_sim.h" + +static int cores = 1; + +#ifdef LINUX + + + +#include + +struct per_thread_t { + pthread_t tid; +}; + +#define MAX_THREADS 200 +static per_thread_t tr_info[MAX_THREADS]; + + +////////////// + +struct test_t_info1 { + CPreviewMode * preview_info; + CFlowGenListPerThread * thread_info; + uint32_t thread_id; +}; + +void * thread_task(void *info){ + + test_t_info1 * obj =(test_t_info1 *)info; + + CFlowGenListPerThread * lpt=obj->thread_info; + + printf("start thread %d \n",obj->thread_id); + //delay(obj->thread_id *3000); + printf("-->start thread %d \n",obj->thread_id); + if (1/*obj->thread_id ==3*/) { + + char buf[100]; + sprintf(buf,"my%d.erf",obj->thread_id); + lpt->start_generate_stateful(buf,*obj->preview_info); + lpt->m_node_gen.DumpHist(stdout); + printf("end thread %d \n",obj->thread_id); + } + + return (NULL); +} + + +void test_load_list_of_cap_files_linux(CParserOption * op){ + + CFlowGenList fl; + //CNullIF erf_vif; + //CErfIF erf_vif; + + fl.Create(); + + fl.load_from_yaml(op->cfg_file,cores); + fl.DumpPktSize(); + + + fl.generate_p_thread_info(cores); + CFlowGenListPerThread * lpt; + + /* set the ERF file */ + //fl.set_vif_all(&erf_vif); + + int i; + for (i=0; ipreview_info =&op->preview; + obj->thread_info = fl.m_threads_info[i]; + obj->thread_id = i; + CNullIF * erf_vif = new CNullIF(); + //CErfIF * erf_vif = new CErfIF(); + + lpt->set_vif(erf_vif); + + assert(pthread_create( &tr_info[i].tid, NULL, thread_task, obj)==0); + } + + for (i=0; icfg_file,NUM); + fl.DumpPktSize(); + + + fl.generate_p_thread_info(NUM); + CFlowGenListPerThread * lpt; + + /* set the ERF file */ + //fl.set_vif_all(&erf_vif); + + int i; + for (i=0; istart_generate_stateful(buf,op->preview); + lpt->m_node_gen.DumpHist(stdout); + } + //sprintf(buf,"my%d.erf",7); + //lpt=fl.m_threads_info[7]; + + //fl.Dump(stdout); + fl.Delete(); +} + +int load_list_of_cap_files(CParserOption * op){ + CFlowGenList fl; + fl.Create(); + fl.load_from_yaml(op->cfg_file,1); + if ( op->preview.getVMode() >0 ) { + fl.DumpCsv(stdout); + } + uint32_t start= os_get_time_msec(); + + CErfIF erf_vif; + //CNullIF erf_vif; + + fl.generate_p_thread_info(1); + CFlowGenListPerThread * lpt; + lpt=fl.m_threads_info[0]; + lpt->set_vif(&erf_vif); + + if ( (op->preview.getVMode() >1) || op->preview.getFileWrite() ) { + lpt->start_generate_stateful(op->out_file,op->preview); + } + + lpt->m_node_gen.DumpHist(stdout); + + uint32_t stop= os_get_time_msec(); + printf(" d time = %ul %ul \n",stop-start,os_get_time_freq()); + fl.Delete(); + return (0); +} + + +int test_dns(){ + + time_init(); + CGlobalInfo::init_pools(1000); + + CParserOption po ; + + //po.cfg_file = "cap2/dns.yaml"; + //po.cfg_file = "cap2/sfr3.yaml"; + po.cfg_file = "cap2/sfr.yaml"; + + po.preview.setVMode(0); + po.preview.setFileWrite(true); + #ifdef LINUX + test_load_list_of_cap_files_linux(&po); + #else + test_load_list_of_cap_files(&po); + #endif + return (0); +} + +void test_pkt_mbuf(void); + +void test_compare_files(void); + +#if 0 +static int b=0; +static int c=0; +static int d=0; + +int test_instructions(){ + int i; + for (i=0; i<100000;i++) { + b+=b+1; + c+=+b+c+1; + d+=+(b*2+1); + } + return (b+c+d); +} + +#include +#endif + + +void update_tcp_seq_num(CCapFileFlowInfo * obj, + int pkt_id, + int size_change){ + CFlowPktInfo * pkt=obj->GetPacket(pkt_id); + if ( pkt->m_pkt_indication.m_desc.IsUdp() ){ + /* nothing to do */ + return; + } + + bool o_init=pkt->m_pkt_indication.m_desc.IsInitSide(); + TCPHeader * tcp ; + int s= (int)obj->Size(); + int i; + + for (i=pkt_id+1; iGetPacket(i); + tcp=pkt->m_pkt_indication.l4.m_tcp; + bool init=pkt->m_pkt_indication.m_desc.IsInitSide(); + if (init == o_init) { + /* same dir update the seq number */ + tcp->setSeqNumber (tcp->getSeqNumber ()+size_change); + + }else{ + /* update the ack number */ + tcp->setAckNumber (tcp->getAckNumber ()+size_change); + } + } +} + + + +void change_pkt_len(CCapFileFlowInfo * obj,int pkt_id, int size ){ + CFlowPktInfo * pkt=obj->GetPacket(pkt_id); + + /* enlarge the packet size by 9 */ + + char * p=pkt->m_packet->append(size); + /* set it to 0xaa*/ + memmove(p+size-4,p-4,4); /* CRCbytes */ + memset(p-4,0x0a,size); + + /* refresh the pointers */ + pkt->m_pkt_indication.RefreshPointers(); + + IPHeader * ipv4 = pkt->m_pkt_indication.l3.m_ipv4; + ipv4->updateTotalLength (ipv4->getTotalLength()+size ); + + /* update seq numbers if needed */ + update_tcp_seq_num(obj,pkt_id,size); +} + +void dump_tcp_seq_num_(CCapFileFlowInfo * obj){ + int s= (int)obj->Size(); + int i; + uint32_t i_seq; + uint32_t r_seq; + + CFlowPktInfo * pkt=obj->GetPacket(0); + TCPHeader * tcp = pkt->m_pkt_indication.l4.m_tcp; + i_seq=tcp->getSeqNumber (); + + pkt=obj->GetPacket(1); + tcp = pkt->m_pkt_indication.l4.m_tcp; + r_seq=tcp->getSeqNumber (); + + for (i=2; iGetPacket(i); + tcp=pkt->m_pkt_indication.l4.m_tcp; + bool init=pkt->m_pkt_indication.m_desc.IsInitSide(); + seq=tcp->getSeqNumber (); + ack=tcp->getAckNumber (); + if (init) { + seq=seq-i_seq; + ack=ack-r_seq; + }else{ + seq=seq-r_seq; + ack=ack-i_seq; + } + printf(" %4d ",i); + if (!init) { + printf(" "); + } + printf(" %s seq: %4d ack : %4d \n",init?"I":"R",seq,ack); + } +} + + +int manipolate_capfile() { + time_init(); + CGlobalInfo::init_pools(1000); + + CCapFileFlowInfo flow_info; + flow_info.Create(); + + flow_info.load_cap_file("avl/delay_10_rtsp_0.pcap",0,0); + + change_pkt_len(&flow_info,4-1 ,6); + change_pkt_len(&flow_info,5-1 ,6); + change_pkt_len(&flow_info,6-1 ,6+2); + change_pkt_len(&flow_info,7-1 ,4); + change_pkt_len(&flow_info,8-1 ,6+2); + change_pkt_len(&flow_info,9-1 ,4); + change_pkt_len(&flow_info,10-1,6); + change_pkt_len(&flow_info,13-1,6); + change_pkt_len(&flow_info,16-1,6); + change_pkt_len(&flow_info,19-1,6); + + flow_info.save_to_erf("exp/c.pcap",1); + + return (1); +} + +int manipolate_capfile_sip() { + time_init(); + CGlobalInfo::init_pools(1000); + + CCapFileFlowInfo flow_info; + flow_info.Create(); + + flow_info.load_cap_file("avl/delay_10_sip_0.pcap",0,0); + + change_pkt_len(&flow_info,1-1 ,6+6); + change_pkt_len(&flow_info,2-1 ,6+6); + + flow_info.save_to_erf("exp/delay_10_sip_0_fixed.pcap",1); + + return (1); +} + +int manipolate_capfile_sip1() { + time_init(); + CGlobalInfo::init_pools(1000); + + CCapFileFlowInfo flow_info; + flow_info.Create(); + + flow_info.load_cap_file("avl/delay_sip_0.pcap",0,0); + flow_info.GetPacket(1); + + change_pkt_len(&flow_info,1-1 ,6+6+10); + + change_pkt_len(&flow_info,2-1 ,6+6+10); + + flow_info.save_to_erf("exp/delay_sip_0_fixed_1.pcap",1); + + return (1); +} + + +class CMergeCapFileRec { +public: + + CCapFileFlowInfo m_cap; + + int m_index; + int m_limit_number_of_packets; /* limit number of packets */ + bool m_stop; /* Do we have more packets */ + + double m_offset; /* offset should be positive */ + double m_start_time; + +public: + bool Create(std::string cap_file,double offset); + void Delete(); + void IncPacket(); + bool GetCurPacket(double & time); + CPacketIndication * GetUpdatedPacket(); + + void Dump(FILE *fd,int _id); +}; + + +void CMergeCapFileRec::Dump(FILE *fd,int _id){ + double time = 0.0; + bool stop=GetCurPacket(time); + fprintf (fd," id:%2d stop : %d index:%4d %3.4f \n",_id,stop?1:0,m_index,time); +} + + +CPacketIndication * CMergeCapFileRec::GetUpdatedPacket(){ + double t1; + assert(GetCurPacket(t1)==false); + CFlowPktInfo * pkt = m_cap.GetPacket(m_index); + pkt->m_pkt_indication.m_packet->set_new_time(t1); + return (&pkt->m_pkt_indication); +} + + +bool CMergeCapFileRec::GetCurPacket(double & time){ + if (m_stop) { + return(true); + } + CFlowPktInfo * pkt = m_cap.GetPacket(m_index); + time= (pkt->m_packet->get_time() -m_start_time + m_offset); + return (false); +} + +void CMergeCapFileRec::IncPacket(){ + m_index++; + if ( (m_limit_number_of_packets) && (m_index > m_limit_number_of_packets ) ) { + m_stop=true; + return; + } + + if ( m_index == (int)m_cap.Size() ) { + m_stop=true; + } +} + +void CMergeCapFileRec::Delete(){ + m_cap.Delete(); +} + +bool CMergeCapFileRec::Create(std::string cap_file, + double offset){ + m_cap.Create(); + m_cap.load_cap_file(cap_file,0,0); + CFlowPktInfo * pkt = m_cap.GetPacket(0); + + m_index=0; + m_stop=false; + m_limit_number_of_packets =0; + m_start_time = pkt->m_packet->get_time() ; + m_offset = offset; + + return (true); +} + + + +#define MERGE_CAP_FILES (2) + +class CMergeCapFile { +public: + bool Create(); + void Delete(); + bool run_merge(std::string to_cap_file); +private: + void append(int _cap_id); + +public: + CMergeCapFileRec m[MERGE_CAP_FILES]; + CCapFileFlowInfo m_results; +}; + +bool CMergeCapFile::Create(){ + m_results.Create(); + return(true); +} + +void CMergeCapFile::Delete(){ + m_results.Delete(); +} + +void CMergeCapFile::append(int _cap_id){ + CPacketIndication * lp=m[_cap_id].GetUpdatedPacket(); + lp->m_packet->Dump(stdout,0); + m_results.Append(lp); +} + + +bool CMergeCapFile::run_merge(std::string to_cap_file){ + + int i=0; + int cnt=0; + while ( true ) { + int min_index=0; + double min_time; + + fprintf(stdout," --------------\n"); + fprintf(stdout," pkt : %d \n",cnt); + for (i=0; i +#include +#include +#include + +using namespace std; + +TrexStateless * get_stateless_obj() { + return SimStateless::get_instance().get_stateless_obj(); +} + + +/** + * handler for DP to CP messages + * + * @author imarom (19-Nov-15) + */ +class DpToCpHandler { +public: + virtual void handle(TrexStatelessDpToCpMsgBase *msg) = 0; +}; + +/************************ + * + * stateless sim object + * +************************/ +class SimPublisher : public TrexPublisher { +public: + + /* override create */ + bool Create(uint16_t port, bool disable) { + return true; + } + + void Delete() { + + } + + void publish_json(const std::string &s) { + } + + virtual ~SimPublisher() { + } +}; + +/************************ + * + * stateless sim object + * +************************/ + +SimStateless::SimStateless() { + m_trex_stateless = NULL; + m_publisher = NULL; + m_dp_to_cp_handler = NULL; + m_verbose = false; + + /* override ownership checks */ + TrexRpcCommand::test_set_override_ownership(true); +} + + +int +SimStateless::run(const string &json_filename, const string &out_filename) { + prepare_dataplane(); + prepare_control_plane(); + + execute_json(json_filename); + run_dp(out_filename); + + flush_dp_to_cp_messages(); + + return 0; +} + + +SimStateless::~SimStateless() { + if (m_trex_stateless) { + delete m_trex_stateless; + m_trex_stateless = NULL; + } + + if (m_publisher) { + delete m_publisher; + m_publisher = NULL; + } + + m_fl.Delete(); +} + +/** + * prepare control plane for test + * + * @author imarom (28-Dec-15) + */ +void +SimStateless::prepare_control_plane() { + TrexStatelessCfg cfg; + + m_publisher = new SimPublisher(); + + TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_MOCK, 0); + + cfg.m_port_count = 4; + 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 TrexMockPlatformApi(); + cfg.m_publisher = m_publisher; + + m_trex_stateless = new TrexStateless(cfg); + + m_trex_stateless->launch_control_plane(); + + for (auto &port : m_trex_stateless->get_port_list()) { + port->acquire("test", 0, true); + } + +} + + +/** + * prepare the data plane for test + * + */ +void +SimStateless::prepare_dataplane() { + + m_fl.Create(); + m_fl.generate_p_thread_info(1); + m_fl.m_threads_info[0]->set_vif(&m_erf_vif); +} + + + +void +SimStateless::execute_json(const std::string &json_filename) { + + std::ifstream test(json_filename); + std::stringstream buffer; + buffer << test.rdbuf(); + + std::string rep = m_trex_stateless->get_rpc_server()->test_inject_request(buffer.str()); + + Json::Value root; + Json::Reader reader; + reader.parse(rep, root, false); + + if (is_verbose()) { + std::cout << "server response: \n\n" << root << "\n\n"; + } +} + +void +SimStateless::run_dp(const std::string &out_filename) { + + CFlowGenListPerThread *lpt = m_fl.m_threads_info[0]; + + lpt->start_stateless_simulation_file((std::string)out_filename, CGlobalInfo::m_options.preview); + lpt->start_stateless_daemon_simulation(); + + flush_dp_to_cp_messages(); + + std::cout << "\nwritten " << lpt->m_node_gen.m_cnt << " packets " << "to '" << out_filename << "'\n\n"; +} + +void +SimStateless::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; + } + +} diff --git a/src/stateless/cp/trex_stateless.h b/src/stateless/cp/trex_stateless.h index 5c11be1e..59be9241 100644 --- a/src/stateless/cp/trex_stateless.h +++ b/src/stateless/cp/trex_stateless.h @@ -161,6 +161,10 @@ public: return m_ports; } + TrexRpcServer * get_rpc_server() { + return m_rpc_server; + } + protected: /* no copy or assignment */ @@ -168,7 +172,7 @@ protected: void operator=(TrexStateless const&) = delete; /* RPC server array */ - TrexRpcServer *m_rpc_server; + TrexRpcServer *m_rpc_server; /* ports */ std::vector m_ports; -- cgit 1.2.3-korg