diff options
Diffstat (limited to 'src')
49 files changed, 2282 insertions, 1036 deletions
diff --git a/src/bp_gtest.cpp b/src/bp_gtest.cpp index a529d637..03ab74bd 100755 --- a/src/bp_gtest.cpp +++ b/src/bp_gtest.cpp @@ -66,7 +66,7 @@ int test_priorty_queue(void){ int i; for (i=0; i<10; i++) { node = new CGenNode(); - printf(" +%x \n",node); + printf(" +%p \n",node); node->m_flow_id = 10-i; node->m_pkt_info = (CFlowPktInfo *)(uintptr_t)i; node->m_time = (double)i+0.1; @@ -74,7 +74,7 @@ int test_priorty_queue(void){ } while (!p_queue.empty()) { node = p_queue.top(); - printf(" -->%x \n",node); + printf(" -->%p \n",node); //node->Dump(stdout); p_queue.pop(); //delete node; @@ -159,7 +159,7 @@ public: bool init(void){ - uint16 * ports; + uint16 * ports = NULL; CTupleBase tuple; CErfIF erf_vif; @@ -205,7 +205,7 @@ public: } } - lpt->generate_erf(buf,CGlobalInfo::m_options.preview); + lpt->start_generate_stateful(buf,CGlobalInfo::m_options.preview); lpt->m_node_gen.DumpHist(stdout); cmp.d_sec = m_time_diff; @@ -663,6 +663,7 @@ TEST_F(basic, latency1) { po->preview.setFileWrite(true); uint8_t mac[]={0,0,0,1,0,0}; + (void)mac; CErfIF erf_vif; erf_vif.set_review_mode(&CGlobalInfo::m_options.preview); @@ -714,6 +715,7 @@ TEST_F(basic, latency2) { uint8_t mac[]={0,0,0,1,0,0}; + (void)mac; mac[0]=0; mac[1]=0; @@ -728,14 +730,13 @@ TEST_F(basic, latency2) { int i; for (i=0; i<100; i++) { - uint8_t *p; rte_mbuf_t * m=l.generate_pkt(0); - p=rte_pktmbuf_mtod(m, uint8_t*); + rte_pktmbuf_mtod(m, uint8_t*); //utl_DumpBuffer(stdout,p,l.get_pkt_size(),0); port0.update_packet(m); - p=rte_pktmbuf_mtod(m, uint8_t*); + rte_pktmbuf_mtod(m, uint8_t*); //utl_DumpBuffer(stdout,p,l.get_pkt_size(),0); //printf("offset is : %d \n",l.get_payload_offset()); @@ -763,6 +764,7 @@ TEST_F(basic, latency3) { uint8_t mac[]={0,0,0,1,0,0}; + (void)mac; mac[0]=0; @@ -850,6 +852,7 @@ public: TEST_F(basic, latency4) { uint8_t mac[]={0,0,0,1,0,0}; + (void)mac; mac[0]=0; mac[1]=0; @@ -2031,6 +2034,10 @@ public: virtual int send_node(CGenNode * node); + virtual int update_mac_addr_from_global_cfg(pkt_dir_t dir, rte_mbuf_t *m){ + return (0); + } + /** * flush all pending packets into the stream @@ -2105,7 +2112,7 @@ public: int i; for (i=0; i<m_threads; i++) { lpt=fl.m_threads_info[i]; - lpt->generate_erf("t1",CGlobalInfo::m_options.preview); + lpt->start_generate_stateful("t1",CGlobalInfo::m_options.preview); } fl.Delete(); return (true); @@ -2121,7 +2128,7 @@ class CRxCheck1 : public CRxCheckCallbackBase { public: virtual void handle_packet(rte_mbuf_t * m){ - char *mp=rte_pktmbuf_mtod(m, char*); + rte_pktmbuf_mtod(m, char*); CRx_check_header * rx_p; rte_mbuf_t * m2 = m->next; rx_p=(CRx_check_header *)rte_pktmbuf_mtod(m2, char*); @@ -2755,7 +2762,7 @@ TEST_F(gt_ring, ring1) { TEST_F(gt_ring, ring2) { CMessagingManager ringmg; - ringmg.Create(8); + ringmg.Create(8, "test"); int i; for (i=0; i<8; i++) { diff --git a/src/bp_sim.cpp b/src/bp_sim.cpp index c3581c55..0c0cbb75 100755 --- a/src/bp_sim.cpp +++ b/src/bp_sim.cpp @@ -25,6 +25,7 @@ limitations under the License. #include "msg_manager.h" #include <common/basic_utils.h> +#include <trex_stream_node.h> #undef VALG @@ -71,11 +72,11 @@ void CGlobalMemory::Dump(FILE *fd){ c_size=c_size*2; } - fprintf(fd," %-40s : %lu \n",names[i].c_str(),m_mbuf[i]); + fprintf(fd," %-40s : %lu \n",names[i].c_str(),(ulong)m_mbuf[i]); } c_total += (m_mbuf[MBUF_DP_FLOWS] * sizeof(CGenNode)); - fprintf(fd," %-40s : %lu \n","get_each_core_dp_flows",get_each_core_dp_flows()); + fprintf(fd," %-40s : %lu \n","get_each_core_dp_flows",(ulong)get_each_core_dp_flows()); fprintf(fd," %-40s : %s \n","Total memory",double_to_human_str(c_total,"bytes",KBYE_1024).c_str() ); } @@ -239,7 +240,7 @@ bool CPlatformSocketInfoConfig::init(){ } if ( m_thread_phy_to_virtual[phy_thread] ){ - printf("ERROR physical thread %d defined twice %d \n",phy_thread); + printf("ERROR physical thread %d defined twice\n",phy_thread); exit(1); } m_thread_phy_to_virtual[phy_thread]=virt_thread; @@ -268,7 +269,7 @@ bool CPlatformSocketInfoConfig::init(){ void CPlatformSocketInfoConfig::dump(FILE *fd){ - fprintf(fd," core_mask %x \n",get_cores_mask()); + fprintf(fd," core_mask %llx \n",(unsigned long long)get_cores_mask()); fprintf(fd," sockets :"); int i; for (i=0; i<MAX_SOCKETS_SUPPORTED; i++) { @@ -279,7 +280,7 @@ void CPlatformSocketInfoConfig::dump(FILE *fd){ fprintf(fd," \n"); fprintf(fd," active sockets : %d \n",max_num_active_sockets()); - fprintf(fd," ports_sockets : \n",max_num_active_sockets()); + fprintf(fd," ports_sockets : %d \n",max_num_active_sockets()); for (i=0; i<(MAX_LATENCY_PORTS); i++) { fprintf(fd,"%d,",port_to_socket(i)); @@ -820,7 +821,6 @@ void CPacketIndication::UpdatePacketPadding(){ void CPacketIndication::RefreshPointers(){ char *pobase=getBasePtr(); - CPacketIndication * obj=this; m_ether = (EthernetHeader *) (pobase + m_ether_offset); l3.m_ipv4 = (IPHeader *) (pobase + m_ip_offset); @@ -1671,7 +1671,6 @@ char * CFlowPktInfo::push_ipv4_option_offline(uint8_t bytes){ void CFlowPktInfo::mask_as_learn(){ - char *p; CNatOption *lpNat; if ( m_pkt_indication.is_ipv6() ){ lpNat=(CNatOption *)push_ipv6_option_offline(CNatOption::noOPTION_LEN); @@ -2265,7 +2264,7 @@ void CCCapFileMemoryUsage::dump(FILE *fd){ int c_total=0; for (i=0; i<CCCapFileMemoryUsage::MASK_SIZE; i++) { - fprintf(fd," size_%-7d : %lu \n",c_size,m_buf[i]); + fprintf(fd," size_%-7d : %lu \n",c_size, (ulong)m_buf[i]); c_total +=m_buf[i]*c_size; c_size = c_size*2; } @@ -2440,7 +2439,6 @@ void operator >> (const YAML::Node& node, CFlowYamlInfo & fi) { if ( node.FindValue("dyn_pyload") ){ - int i; const YAML::Node& dyn_pyload = node["dyn_pyload"]; for(unsigned i=0;i<dyn_pyload.size();i++) { CFlowYamlDpPkt fd; @@ -2839,8 +2837,20 @@ void CFlowStats::DumpHeader(FILE *fd){ void CFlowStats::Dump(FILE *fd){ //"name","cps","f-pkts","f-bytes","Mb/sec","MB/sec","c-flows","PPS","total-Mbytes-duration","errors","flows" fprintf(fd," %02d, %-40s ,%4.2f,%4.2f, %5.0f , %7.0f ,%7.2f ,%7.2f , %7.2f , %10.0f , %5.0f , %7.0f , %llu , %llu \n", - m_id,m_name.c_str(),m_cps,get_normal_cps(), - m_pkt,m_bytes,duration_sec,m_mb_sec,m_mB_sec,m_c_flows,m_pps,m_total_Mbytes,m_errors,m_flows); + m_id, + m_name.c_str(), + m_cps, + get_normal_cps(), + m_pkt, + m_bytes, + duration_sec, + m_mb_sec, + m_mB_sec, + m_c_flows, + m_pps, + m_total_Mbytes, + (unsigned long long)m_errors, + (unsigned long long)m_flows); } bool CFlowGeneratorRecPerThread::Create(CTupleGeneratorSmart * global_gen, @@ -3045,21 +3055,20 @@ void CGenNode::DumpHeader(FILE *fd){ } void CGenNode::Dump(FILE *fd){ - fprintf(fd,"%.6f,%llx,%p,%llu,%d,%d,%d,%d,%d,%d,%x,%x,%d\n",m_time,m_flow_id,m_pkt_info, - m_pkt_info->m_pkt_indication.m_packet->pkt_cnt, - m_pkt_info->m_pkt_indication.m_packet->pkt_len, - m_pkt_info->m_pkt_indication.m_desc.getId(), - (m_pkt_info->m_pkt_indication.m_desc.IsInitSide()?1:0), - m_pkt_info->m_pkt_indication.m_desc.IsLastPkt(), + fprintf(fd,"%.6f,%llx,%p,%llu,%d,%d,%d,%d,%d,%d,%x,%x,%d\n", + m_time, + (unsigned long long)m_flow_id, + m_pkt_info, + (unsigned long long)m_pkt_info->m_pkt_indication.m_packet->pkt_cnt, + m_pkt_info->m_pkt_indication.m_packet->pkt_len, + m_pkt_info->m_pkt_indication.m_desc.getId(), + (m_pkt_info->m_pkt_indication.m_desc.IsInitSide()?1:0), + m_pkt_info->m_pkt_indication.m_desc.IsLastPkt(), m_type, m_thread_id, m_src_ip, m_dest_ip, - m_src_port - - - - ); + m_src_port); } @@ -3114,14 +3123,10 @@ int CNodeGenerator::close_file(CFlowGenListPerThread * thread){ return (0); } -int CNodeGenerator::flush_one_node_to_file(CGenNode * node){ - BP_ASSERT(m_v_if); - return (m_v_if->send_node(node)); -} int CNodeGenerator::update_stats(CGenNode * node){ if ( m_preview_mode.getVMode() >2 ){ - fprintf(stdout," %llu ,",m_cnt); + fprintf(stdout," %llu ,", (unsigned long long)m_cnt); node->Dump(stdout); m_cnt++; } @@ -3204,6 +3209,10 @@ bool CFlowGenListPerThread::Create(uint32_t thread_id, assert(m_ring_from_rx); assert(m_ring_to_rx); + + /* create the info required for stateless DP core */ + m_stateless_dp_info.create(thread_id, this); + return (true); } @@ -3351,6 +3360,7 @@ void CFlowGenListPerThread::Delete(){ m_node_gen.Delete(); Clean(); m_cpu_cp_u.Delete(); + } @@ -3397,15 +3407,24 @@ int CNodeGenerator::flush_file(dsec_t max_time, bool done=false; thread->m_cpu_dp_u.start_work(); - while (!m_p_queue.empty()) { + + /** + * if a positive value was given to max time + * schedule an exit node + */ + if ( (max_time > 0) && (!always) ) { + CGenNode *exit_node = thread->create_node(); + + exit_node->m_type = CGenNode::EXIT_SCHED; + exit_node->m_time = max_time; + add_node(exit_node); + } + + while (true) { + node = m_p_queue.top(); - n_time = node->m_time+ offset; + n_time = node->m_time + offset; - if (( (n_time) > max_time ) && - (always==false) ) { - /* nothing to do */ - break; - } events++; /*#ifdef VALG if (events > 1 ) { @@ -3416,7 +3435,6 @@ int CNodeGenerator::flush_file(dsec_t max_time, if ( likely ( m_is_realtime ) ){ dsec_t dt ; thread->m_cpu_dp_u.commit(); - bool once=false; while ( true ) { dt = now_sec() - n_time ; @@ -3425,12 +3443,6 @@ int CNodeGenerator::flush_file(dsec_t max_time, break; } - if (!once) { - /* check the msg queue once */ - thread->check_msgs(); - once=true; - } - rte_pause(); } thread->m_cpu_dp_u.start_work(); @@ -3449,55 +3461,72 @@ int CNodeGenerator::flush_file(dsec_t max_time, flush_time=now_sec(); } } + #ifndef RTE_DPDK - thread->check_msgs(); + thread->check_msgs(); #endif uint8_t type=node->m_type; - if ( likely( type == CGenNode::FLOW_PKT ) ) { - /* PKT */ - if ( !(node->is_repeat_flow()) || (always==false)) { - flush_one_node_to_file(node); - #ifdef _DEBUG - update_stats(node); - #endif - } - m_p_queue.pop(); - if ( node->is_last_in_flow() ) { - if ((node->is_repeat_flow()) && (always==false)) { - /* Flow is repeated, reschedule it */ - thread->reschedule_flow( node); - }else{ - /* Flow will not be repeated, so free node */ - thread->free_last_flow_node( node); - } - }else{ - node->update_next_pkt_in_flow(); - m_p_queue.push(node); - } + if ( type == CGenNode::STATELESS_PKT ) { + m_p_queue.pop(); + CGenNodeStateless *node_sl = (CGenNodeStateless *)node; + + /* if the stream has been deactivated - end */ + if (unlikely(!node_sl->is_active())) { + thread->free_node(node); + } else { + node_sl->handle(thread); + } + }else{ - if ((type == CGenNode::FLOW_FIF)) { - /* callback to our method */ + if ( likely( type == CGenNode::FLOW_PKT ) ) { + /* PKT */ + if ( !(node->is_repeat_flow()) || (always==false)) { + flush_one_node_to_file(node); + #ifdef _DEBUG + update_stats(node); + #endif + } m_p_queue.pop(); - if ( always == false) { - thread->m_cur_time_sec = node->m_time ; - - if ( thread->generate_flows_roundrobin(&done) <0){ - break; + if ( node->is_last_in_flow() ) { + if ((node->is_repeat_flow()) && (always==false)) { + /* Flow is repeated, reschedule it */ + thread->reschedule_flow( node); + }else{ + /* Flow will not be repeated, so free node */ + thread->free_last_flow_node( node); } - if (!done) { - node->m_time +=d_time; - m_p_queue.push(node); + }else{ + node->update_next_pkt_in_flow(); + m_p_queue.push(node); + } + }else{ + if ((type == CGenNode::FLOW_FIF)) { + /* callback to our method */ + m_p_queue.pop(); + if ( always == false) { + thread->m_cur_time_sec = node->m_time ; + + if ( thread->generate_flows_roundrobin(&done) <0){ + break; + } + if (!done) { + node->m_time +=d_time; + m_p_queue.push(node); + }else{ + thread->free_node(node); + } }else{ thread->free_node(node); } + }else{ - thread->free_node(node); + bool exit_sccheduler = handle_slow_messages(type,node,thread,always); + if (exit_sccheduler) { + break; + } } - - }else{ - handle_slow_messages(type,node,thread,always); } } } @@ -3512,17 +3541,21 @@ int CNodeGenerator::flush_file(dsec_t max_time, return (0); } -void CNodeGenerator::handle_slow_messages(uint8_t type, - CGenNode * node, - CFlowGenListPerThread * thread, - bool always){ +bool +CNodeGenerator::handle_slow_messages(uint8_t type, + CGenNode * node, + CFlowGenListPerThread * thread, + bool always){ + + /* should we continue after */ + bool exit_scheduler = false; if (unlikely (type == CGenNode::FLOW_DEFER_PORT_RELEASE) ) { m_p_queue.pop(); thread->handler_defer_job(node); thread->free_node(node); - }else{ - if (type == CGenNode::FLOW_PKT_NAT) { + + } else if (type == CGenNode::FLOW_PKT_NAT) { /*repeat and NAT is not supported */ if ( node->is_nat_first_state() ){ node->set_nat_wait_state(); @@ -3536,7 +3569,7 @@ void CNodeGenerator::handle_slow_messages(uint8_t type, m_p_queue.pop(); /* time out, need to free the flow and remove the association , we didn't get convertion yet*/ thread->terminate_nat_flows(node); - return; + return (exit_scheduler); }else{ flush_one_node_to_file(node); @@ -3556,24 +3589,39 @@ void CNodeGenerator::handle_slow_messages(uint8_t type, m_p_queue.push(node); } - }else{ - if ( type == CGenNode::FLOW_SYNC ){ - thread->check_msgs(); /* check messages */ - m_v_if->flush_tx_queue(); /* flush pkt each timeout */ - m_p_queue.pop(); - if ( always == false) { - node->m_time += SYNC_TIME_OUT; - m_p_queue.push(node); - }else{ - thread->free_node(node); - } + } else if ( type == CGenNode::FLOW_SYNC ) { - }else{ - printf(" ERROR type is not valid %d \n",type); - assert(0); + /* flow sync message is a sync point for time */ + thread->m_cur_time_sec = node->m_time; + + /* first pop the node */ + m_p_queue.pop(); + + thread->check_msgs(); /* check messages */ + m_v_if->flush_tx_queue(); /* flush pkt each timeout */ + + /* on always (clean queue path) and queue empty - exit */ + if ( always && (m_p_queue.empty()) ) { + thread->free_node(node); + exit_scheduler = true; + } else { + /* schedule for next maintenace */ + node->m_time += SYNC_TIME_OUT; + m_p_queue.push(node); } + + + } else if ( type == CGenNode::EXIT_SCHED ) { + m_p_queue.pop(); + thread->free_node(node); + exit_scheduler = true; + + } else { + printf(" ERROR type is not valid %d \n",type); + assert(0); } - } + + return exit_scheduler; } @@ -3809,11 +3857,15 @@ void CFlowGenListPerThread::handel_nat_msg(CGenNodeNatInfo * msg){ } } +void CFlowGenListPerThread::check_msgs(void) { + + /* inlined for performance */ + m_stateless_dp_info.periodic_check_for_cp_messages(); -void CFlowGenListPerThread::check_msgs(void){ - if ( likely ( m_ring_from_rx->isEmpty() ) ){ + if ( likely ( m_ring_from_rx->isEmpty() ) ) { return; } + #ifdef NAT_TRACE_ printf(" %.03f got message from RX \n",now_sec()); #endif @@ -3833,9 +3885,11 @@ void CFlowGenListPerThread::check_msgs(void){ case CGenNodeMsgBase::NAT_FIRST: handel_nat_msg((CGenNodeNatInfo * )msg); break; + case CGenNodeMsgBase::LATENCY_PKT: handel_latecy_pkt_msg((CGenNodeLatencyPktInfo *) msg); break; + default: printf("ERROR pkt-thread message type is not valid %d \n",msg_type); assert(0); @@ -3845,8 +3899,49 @@ void CFlowGenListPerThread::check_msgs(void){ } } +void delay(int msec); + -void CFlowGenListPerThread::generate_erf(std::string erf_file_name, +const uint8_t test_udp_pkt[]={ + 0x00,0x00,0x00,0x01,0x00,0x00, + 0x00,0x00,0x00,0x01,0x00,0x00, + 0x08,0x00, + + 0x45,0x00,0x00,0x81, + 0xaf,0x7e,0x00,0x00, + 0x12,0x11,0xd9,0x23, + 0x01,0x01,0x01,0x01, + 0x3d,0xad,0x72,0x1b, + + 0x11,0x11, + 0x11,0x11, + + 0x00,0x6d, + 0x00,0x00, + + 0x64,0x31,0x3a,0x61, + 0x64,0x32,0x3a,0x69,0x64, + 0x32,0x30,0x3a,0xd0,0x0e, + 0xa1,0x4b,0x7b,0xbd,0xbd, + 0x16,0xc6,0xdb,0xc4,0xbb,0x43, + 0xf9,0x4b,0x51,0x68,0x33,0x72, + 0x20,0x39,0x3a,0x69,0x6e,0x66,0x6f, + 0x5f,0x68,0x61,0x73,0x68,0x32,0x30,0x3a,0xee,0xc6,0xa3, + 0xd3,0x13,0xa8,0x43,0x06,0x03,0xd8,0x9e,0x3f,0x67,0x6f, + 0xe7,0x0a,0xfd,0x18,0x13,0x8d,0x65,0x31,0x3a,0x71,0x39, + 0x3a,0x67,0x65,0x74,0x5f,0x70,0x65,0x65,0x72,0x73,0x31, + 0x3a,0x74,0x38,0x3a,0x3d,0xeb,0x0c,0xbf,0x0d,0x6a,0x0d, + 0xa5,0x31,0x3a,0x79,0x31,0x3a,0x71,0x65,0x87,0xa6,0x7d, + 0xe7 +}; + +void CFlowGenListPerThread::start_stateless_daemon(){ + m_cur_time_sec = 0; + m_stateless_dp_info.start(); +} + + +void CFlowGenListPerThread::start_generate_stateful(std::string erf_file_name, CPreviewMode & preview){ /* now we are ready to generate*/ if ( m_cap_gen.size()==0 ){ @@ -3963,6 +4058,7 @@ int CFlowGenList::load_from_mac_file(std::string file_name) { exit(-1); } + return (0); } @@ -4435,9 +4531,12 @@ void CTupleTemplateGenerator::Generate(){ #endif +static uint32_t get_rand_32(uint32_t MinimumRange, + uint32_t MaximumRange) __attribute__ ((unused)); + +static uint32_t get_rand_32(uint32_t MinimumRange, + uint32_t MaximumRange) { -static uint32_t get_rand_32(uint32_t MinimumRange , - uint32_t MaximumRange ){ enum {RANDS_NUM = 2 , RAND_MAX_BITS = 0xf , UNSIGNED_INT_BITS = 0x20 , TWO_BITS_MASK = 0x3}; const double TWO_POWER_32_BITS = 0x10000000 * (double)0x10; @@ -4670,7 +4769,6 @@ void CCPortLatency::reset(){ static uint8_t nat_is_port_can_send(uint8_t port_id){ - uint8_t offset= ((port_id>>1)<<1); uint8_t client_index = (port_id %2); return (client_index ==0 ?1:0); } @@ -4792,7 +4890,7 @@ void CCPortLatency::dump_counters_json(std::string & json ){ } void CCPortLatency::DumpCounters(FILE *fd){ - #define DP_A1(f) if (f) fprintf(fd," %-40s : %llu \n",#f,f) + #define DP_A1(f) if (f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)f) fprintf(fd," counter \n"); fprintf(fd," -----------\n"); @@ -4811,7 +4909,7 @@ void CCPortLatency::DumpCounters(FILE *fd){ fprintf(fd," -----------\n"); m_hist.Dump(fd); - fprintf(fd," %-40s : %llu \n","jitter",get_jitter_usec()); + fprintf(fd," %-40s : %lu \n","jitter", (ulong)get_jitter_usec()); } bool CCPortLatency::dump_packet(rte_mbuf_t * m){ @@ -4835,6 +4933,9 @@ bool CCPortLatency::dump_packet(rte_mbuf_t * m){ if ( unlikely( CGlobalInfo::m_options.preview.get_vlan_mode_enable() ) ){ vlan_offset=4; } + + (void)vlan_offset; + // utl_DumpBuffer(stdout,p,pkt_size,0); return (0); @@ -4870,9 +4971,7 @@ bool CCPortLatency::check_packet(rte_mbuf_t * m,CRx_check_header * & rx_p){ uint16_t vlan_offset=parser.m_vlan_offset; uint8_t *p=rte_pktmbuf_mtod(m, uint8_t*); - rx_p=(CRx_check_header *)0; - bool managed_by_ip_options=false; - bool is_rx_check=true; + rx_p = (CRx_check_header *)0; if ( !parser.IsLatencyPkt() ){ @@ -4989,7 +5088,7 @@ void CLatencyManager::Delete(){ static uint8_t swap_port(uint8_t port_id){ uint8_t offset= ((port_id>>1)<<1); uint8_t client_index = (port_id %2); - return (offset+client_index^1); + return (offset + (client_index ^ 1)); } @@ -5112,8 +5211,6 @@ void CLatencyManager::run_rx_queue_msgs(uint8_t thread_id, CGenNodeMsgBase * msg=(CGenNodeMsgBase *)node; - CGenNodeLatencyPktInfo * msg1=(CGenNodeLatencyPktInfo *)msg; - uint8_t msg_type = msg->m_msg_type; switch (msg_type ) { case CGenNodeMsgBase::LATENCY_PKT: @@ -5236,7 +5333,7 @@ void CLatencyManager::start(int iter){ } if ( iter>0 ){ if ( ( cnt>iter) ){ - printf("stop due iter %d %d \n",iter); + printf("stop due iter %d\n",iter); break; } } @@ -5405,8 +5502,8 @@ void CLatencyManager::DumpRxCheckVerification(FILE *fd, fprintf(fd," rx_checker is disabled \n"); return; } - fprintf(fd," rx_check Tx : %u \n",total_tx_rx_check); - fprintf(fd," rx_check Rx : %u \n",m_rx_check_manager.getTotalRx() ); + fprintf(fd," rx_check Tx : %llu \n", (unsigned long long)total_tx_rx_check); + fprintf(fd," rx_check Rx : %llu \n", (unsigned long long)m_rx_check_manager.getTotalRx() ); fprintf(fd," rx_check verification :" ); if (m_rx_check_manager.getTotalRx() == total_tx_rx_check) { fprintf(fd," OK \n" ); @@ -6659,7 +6756,6 @@ bool CSimplePacketParser::Parse(){ EthernetHeader *m_ether = (EthernetHeader *)p; IPHeader * ipv4=0; IPv6Header * ipv6=0; - uint16_t pkt_size=rte_pktmbuf_pkt_len(m); m_vlan_offset=0; m_option_offset=0; diff --git a/src/bp_sim.h b/src/bp_sim.h index b7cfb20b..36595581 100755 --- a/src/bp_sim.h +++ b/src/bp_sim.h @@ -57,6 +57,8 @@ limitations under the License. #include <arpa/inet.h> #include "platform_cfg.h" +#include <trex_stateless_dp_core.h> + #undef NAT_TRACE_ @@ -360,6 +362,25 @@ public: */ virtual int flush_tx_queue(void)=0; + + /** + * update the source and destination mac-addr of a given mbuf by global database + * + * @param dir + * @param m + * + * @return + */ + virtual int update_mac_addr_from_global_cfg(pkt_dir_t dir, rte_mbuf_t *m)=0; + + /** + * translate a port_id to the correct dir on the core + * + */ + virtual pkt_dir_t port_id_to_dir(uint8_t port_id) { + return (CS_INVALID); + } + public: @@ -721,6 +742,7 @@ public: m_run_mode = RUN_MODE_INVALID; } + CPreviewMode preview; float m_factor; float m_duration; @@ -741,9 +763,10 @@ public: uint16_t m_run_flags; uint8_t m_mac_splitter; uint8_t m_pad; - trex_run_mode_e m_run_mode; + + std::string cfg_file; std::string mac_file; std::string platform_cfg_file; @@ -778,6 +801,10 @@ public: return ( m_latency_rate == 0 ?true:false); } + bool is_stateless(){ + return (m_run_mode == RUN_MODE_INTERACTIVE ?true:false); + } + bool is_latency_enabled(){ return ( !is_latency_disabled() ); } @@ -867,6 +894,8 @@ public: /* number of main active sockets. socket #0 is always used */ virtual socket_id_t max_num_active_sockets()=0; + virtual ~CPlatformSocketInfoBase() {} + public: /* which socket to allocate memory to each port */ virtual socket_id_t port_to_socket(port_id_t port)=0; @@ -1181,12 +1210,15 @@ public: public: static CRteMemPool m_mem_pool[MAX_SOCKETS_SUPPORTED]; - static uint32_t m_nodes_pool_size; - static CParserOption m_options; - static CGlobalMemory m_memory_cfg; - static CPlatformSocketInfo m_socket; + static uint32_t m_nodes_pool_size; + static CParserOption m_options; + static CGlobalMemory m_memory_cfg; + static CPlatformSocketInfo m_socket; }; +static inline int get_is_stateless(){ + return (CGlobalInfo::m_options.is_stateless() ); +} static inline int get_is_rx_check_mode(){ return (CGlobalInfo::m_options.preview.get_is_rx_check_enable() ?1:0); @@ -1307,8 +1339,8 @@ public: -#define DP(f) if (f) printf(" %-40s: %llu \n",#f,f) -#define DP_name(n,f) if (f) printf(" %-40s: %llu \n",n,f) +#define DP(f) if (f) printf(" %-40s: %llu \n",#f,(unsigned long long)f) +#define DP_name(n,f) if (f) printf(" %-40s: %llu \n",n,(unsigned long long)f) #define DP_S(f,f_s) if (f) printf(" %-40s: %s \n",#f,f_s.c_str()) @@ -1333,15 +1365,18 @@ class CCapFileFlowInfo ; /* this is a simple struct, do not add constructor and destractor here! we are optimizing the allocation dealocation !!! */ -struct CGenNode { + +struct CGenNodeBase { public: enum { - FLOW_PKT=0, - FLOW_FIF=1, - FLOW_DEFER_PORT_RELEASE=2, - FLOW_PKT_NAT=3, - FLOW_SYNC=4 /* called evey 1 msec */ + FLOW_PKT =0, + FLOW_FIF =1, + FLOW_DEFER_PORT_RELEASE =2, + FLOW_PKT_NAT =3, + FLOW_SYNC =4, /* called evey 1 msec */ + STATELESS_PKT =5, + EXIT_SCHED =6 }; @@ -1349,7 +1384,7 @@ public: enum { NODE_FLAGS_DIR =1, NODE_FLAGS_MBUF_CACHE =2, - NODE_FLAGS_SAMPLE_RX_CHECK =4, + NODE_FLAGS_SAMPLE_RX_CHECK =4, NODE_FLAGS_LEARN_MODE =8, /* bits 3,4 MASK 0x18 wait for second direction packet */ NODE_FLAGS_LEARN_MSG_PROCESSED =0x10, /* got NAT msg */ @@ -1360,19 +1395,49 @@ public: NODE_FLAGS_INIT_START_FROM_SERVER_SIDE_SERVER_ADDR = 0x100 /* init packet start from server side with server addr */ }; + public: - /* C1 */ + /*********************************************/ + /* C1 must */ uint8_t m_type; uint8_t m_thread_id; /* zero base */ uint8_t m_socket_id; - uint8_t m_pad2; + uint8_t m_pad2; uint16_t m_src_port; uint16_t m_flags; /* BIT 0 - DIR , BIT 1 - mbug_cache BIT 2 - SAMPLE DUPLICATE */ - double m_time; + double m_time; /* can't change this header - size 16 bytes*/ + +public: + bool operator <(const CGenNodeBase * rsh ) const { + return (m_time<rsh->m_time); + } + bool operator ==(const CGenNodeBase * rsh ) const { + return (m_time==rsh->m_time); + } + bool operator >(const CGenNodeBase * rsh ) const { + return (m_time>rsh->m_time); + } + +public: + void set_socket_id(socket_id_t socket){ + m_socket_id=socket; + } + + socket_id_t get_socket_id(){ + return ( m_socket_id ); + } + + +}; + + +struct CGenNode : public CGenNodeBase { + +public: uint32_t m_src_ip; /* client ip */ uint32_t m_dest_ip; /* server ip */ @@ -1402,25 +1467,8 @@ public: uint32_t m_end_of_cache_line[6]; public: - bool operator <(const CGenNode * rsh ) const { - return (m_time<rsh->m_time); - } - bool operator ==(const CGenNode * rsh ) const { - return (m_time==rsh->m_time); - } - bool operator >(const CGenNode * rsh ) const { - return (m_time>rsh->m_time); - } -public: void Dump(FILE *fd); - void set_socket_id(socket_id_t socket){ - m_socket_id=socket; - } - - socket_id_t get_socket_id(){ - return ( m_socket_id ); - } static void DumpHeader(FILE *fd); @@ -1603,6 +1651,7 @@ public: } __rte_cache_aligned; + #if __x86_64__ /* size of 64 bytes */ #define DEFER_CLIENTS_NUM (16) @@ -1647,19 +1696,29 @@ public: need to clean this up and derive this objects from base object but require too much refactoring right now hhaim */ + +#define COMPARE_NODE_OBJECT(NODE_NAME) if ( sizeof(NODE_NAME) != sizeof(CGenNode) ) { \ + printf("ERROR sizeof(%s) %lu != sizeof(CGenNode) %lu must be the same size \n",#NODE_NAME,sizeof(NODE_NAME),sizeof(CGenNode)); \ + assert(0); \ + }\ + if ( (int)offsetof(struct NODE_NAME,m_type)!=offsetof(struct CGenNodeBase,m_type) ){\ + printf("ERROR offsetof(struct %s,m_type)!=offsetof(struct CGenNodeBase,m_type) \n",#NODE_NAME);\ + assert(0);\ + }\ + if ( (int)offsetof(struct CGenNodeDeferPort,m_time)!=offsetof(struct CGenNodeBase,m_time) ){\ + printf("ERROR offsetof(struct %s,m_time)!=offsetof(struct CGenNodeBase,m_time) \n",#NODE_NAME);\ + assert(0);\ + } + +#define COMPARE_NODE_OBJECT_SIZE(NODE_NAME) if ( sizeof(NODE_NAME) != sizeof(CGenNode) ) { \ + printf("ERROR sizeof(%s) %lu != sizeof(CGenNode) %lu must be the same size \n",#NODE_NAME,sizeof(NODE_NAME),sizeof(CGenNode)); \ + assert(0); \ + } + + + inline int check_objects_sizes(void){ - if ( sizeof(CGenNodeDeferPort) != sizeof(CGenNode) ) { - printf("ERROR sizeof(CGenNodeDeferPort) %lu != sizeof(CGenNode) %lu must be the same size \n",sizeof(CGenNodeDeferPort),sizeof(CGenNode)); - assert(0); - } - if ( (int)offsetof(struct CGenNodeDeferPort,m_type)!=offsetof(struct CGenNode,m_type) ){ - printf("ERROR offsetof(struct CGenNodeDeferPort,m_type)!=offsetof(struct CGenNode,m_type) \n"); - assert(0); - } - if ( (int)offsetof(struct CGenNodeDeferPort,m_time)!=offsetof(struct CGenNode,m_time) ){ - printf("ERROR offsetof(struct CGenNodeDeferPort,m_time)!=offsetof(struct CGenNode,m_time) \n"); - assert(0); - } + COMPARE_NODE_OBJECT(CGenNodeDeferPort); return (0); } @@ -1718,6 +1777,11 @@ public: virtual int write_pkt(CCapPktRaw *pkt_raw); virtual int close_file(void); + virtual int update_mac_addr_from_global_cfg(pkt_dir_t dir, rte_mbuf_t *m){ + return (0); + } + + /** * send one packet @@ -1779,6 +1843,10 @@ public: return (0); } + virtual int update_mac_addr_from_global_cfg(pkt_dir_t dir, rte_mbuf_t *m){ + return (0); + } + virtual int send_node(CGenNode * node); @@ -1799,6 +1867,7 @@ public: CFlowGenListPerThread * Parent(){ return (m_parent); } + public: void add_node(CGenNode * mynode); void remove_all(CFlowGenListPerThread * thread); @@ -1830,11 +1899,14 @@ public: private: - int flush_one_node_to_file(CGenNode * node); + inline int flush_one_node_to_file(CGenNode * node){ + return (m_v_if->send_node(node)); + } int update_stats(CGenNode * node); - FORCE_NO_INLINE void handle_slow_messages(uint8_t type, - CGenNode * node, - CFlowGenListPerThread * thread, + + FORCE_NO_INLINE bool handle_slow_messages(uint8_t type, + CGenNode * node, + CFlowGenListPerThread * thread, bool always); @@ -2310,6 +2382,7 @@ public: return (uint32_t)((uintptr_t)( ((char *)l3.m_ipv4)-getBasePtr()) ); }else{ BP_ASSERT(0); + return (0); } } @@ -3325,13 +3398,21 @@ public : inline CGenNode * create_node(void); + inline CGenNodeStateless * create_node_sl(void){ + return ((CGenNodeStateless*)create_node() ); + } + + inline void free_node(CGenNode *p); inline void free_last_flow_node(CGenNode *p); public: void Clean(); - void generate_erf(std::string erf_file_name,CPreviewMode &preview); + void start_generate_stateful(std::string erf_file_name,CPreviewMode &preview); + void start_stateless_daemon(); + + void Dump(FILE *fd); void DumpCsv(FILE *fd); void DumpStats(FILE *fd); @@ -3344,6 +3425,7 @@ public: private: void check_msgs(void); + void handel_nat_msg(CGenNodeNatInfo * msg); void handel_latecy_pkt_msg(CGenNodeLatencyPktInfo * msg); @@ -3421,7 +3503,12 @@ private: CNodeRing * m_ring_to_rx; /* ring dp -> latency thread */ flow_id_node_t m_flow_id_to_node_lookup; -}; + + TrexStatelessDpCore m_stateless_dp_info; + +private: + uint8_t m_cacheline_pad[RTE_CACHE_LINE_SIZE][19]; // improve prefech +} __rte_cache_aligned ; inline CGenNode * CFlowGenListPerThread::create_node(void){ CGenNode * res; diff --git a/src/common/Network/Packet/IPHeader.cpp b/src/common/Network/Packet/IPHeader.cpp index 3b90a1aa..c3363603 100755 --- a/src/common/Network/Packet/IPHeader.cpp +++ b/src/common/Network/Packet/IPHeader.cpp @@ -52,7 +52,7 @@ void IPHeader::dump(FILE *fd) { fprintf(fd, "\nIPHeader"); fprintf(fd, "\nSource 0x%.8lX, Destination 0x%.8lX, Protocol 0x%.1X", - getSourceIp(), getDestIp(), getProtocol()); + (ulong)getSourceIp(), (ulong)getDestIp(), (uint)getProtocol()); fprintf(fd, "\nTTL : %d, Id : 0x%.2X, Ver %d, Header Length %d, Total Length %d", getTimeToLive(), getId(), getVersion(), getHeaderLength(), getTotalLength()); if(isFragmented()) diff --git a/src/common/Network/Packet/TCPHeader.cpp b/src/common/Network/Packet/TCPHeader.cpp index bf28db2e..1826cef8 100755 --- a/src/common/Network/Packet/TCPHeader.cpp +++ b/src/common/Network/Packet/TCPHeader.cpp @@ -25,7 +25,7 @@ void TCPHeader::dump(FILE *fd) fprintf(fd, "\nSourcePort 0x%.4X, DestPort 0x%.4X", getSourcePort(), getDestPort()); fprintf(fd, "\nSeqNum 0x%.8lX, AckNum 0x%.8lX, Window %d", - getSeqNumber(), getAckNumber(), getWindowSize()); + (ulong)getSeqNumber(), (ulong)getAckNumber(), getWindowSize()); fprintf(fd, "\nHeader Length : %d, Checksum : 0x%.4X", getHeaderLength(), getChecksum()); fprintf(fd, "\nFlags : SYN - %d, FIN - %d, ACK - %d, URG - %d, RST - %d, PSH - %d", diff --git a/src/common/c_common.h b/src/common/c_common.h index d8320aaa..3e43644f 100755 --- a/src/common/c_common.h +++ b/src/common/c_common.h @@ -46,7 +46,7 @@ typedef void* c_pvoid; #ifdef _DEBUG #define BP_ASSERT(a) assert(a) #else - #define BP_ASSERT(a) + #define BP_ASSERT(a) (void (a)) #endif #endif diff --git a/src/gtest/rpc_test.cpp b/src/gtest/rpc_test.cpp index 250d5342..6b8e3eff 100644 --- a/src/gtest/rpc_test.cpp +++ b/src/gtest/rpc_test.cpp @@ -30,6 +30,8 @@ limitations under the License. using namespace std; +uint16_t gtest_get_mock_server_port(); + class RpcTest : public testing::Test { protected: @@ -44,7 +46,12 @@ protected: m_context = zmq_ctx_new (); m_socket = zmq_socket (m_context, ZMQ_REQ); - zmq_connect (m_socket, "tcp://localhost:5050"); + + std::stringstream ss; + ss << "tcp://localhost:"; + ss << gtest_get_mock_server_port(); + + zmq_connect (m_socket, ss.str().c_str()); } diff --git a/src/gtest/trex_stateless_gtest.cpp b/src/gtest/trex_stateless_gtest.cpp index 0341516c..2bab4dff 100644 --- a/src/gtest/trex_stateless_gtest.cpp +++ b/src/gtest/trex_stateless_gtest.cpp @@ -82,7 +82,7 @@ private: void CTRexDpStreamModeSingleBurst::dump(FILE *fd){ fprintf (fd," pps : %f \n",m_pps); - fprintf (fd," total_packets : %llu \n",m_total_packets); + fprintf (fd," total_packets : %llu \n", (unsigned long long)m_total_packets); } @@ -130,9 +130,9 @@ private: void CTRexDpStreamModeMultiBurst::dump(FILE *fd){ fprintf (fd," pps : %f \n",m_pps); - fprintf (fd," total_packets : %llu \n",m_pkts_per_burst); + fprintf (fd," total_packets : %llu \n", (unsigned long long)m_pkts_per_burst); fprintf (fd," ibg : %f \n",m_ibg); - fprintf (fd," num_of_bursts : %llu \n",m_number_of_bursts); + fprintf (fd," num_of_bursts : %lu \n", (ulong)m_number_of_bursts); } diff --git a/src/gtest/tuple_gen_test.cpp b/src/gtest/tuple_gen_test.cpp index 8a774e38..f3b9fa1e 100755 --- a/src/gtest/tuple_gen_test.cpp +++ b/src/gtest/tuple_gen_test.cpp @@ -161,7 +161,6 @@ TEST(tuple_gen,clientPoolL) { 0,0); CTupleBase result; uint32_t result_src; - uint32_t result_dest; uint16_t result_port; for(int i=0;i<10;i++) { @@ -186,7 +185,6 @@ TEST(tuple_gen,clientPool) { 0,0); CTupleBase result; uint32_t result_src; - uint32_t result_dest; uint16_t result_port; for(int i=0;i<10;i++) { @@ -436,7 +434,6 @@ TEST(tuple_gen,template1) { template_1.GenerateTuple(result); uint32_t result_src = result.getClient(); uint32_t result_dest = result.getServer(); - uint16_t result_port = result.getClientPort(); //printf(" %x %x %x \n",result_src,result_dest,result_port); EXPECT_EQ(result_src, (uint32_t)(0x10000001+i)); EXPECT_EQ(result_dest, (uint32_t)(((0x12121212)) )); @@ -489,9 +486,6 @@ TEST(tuple_gen,no_free) { int i; for (i=0; i<65557; i++) { template_1.GenerateTuple(result); - uint32_t result_src = result.getClient(); - uint32_t result_dest = result.getServer(); - uint16_t result_port = result.getClientPort(); } // should have error EXPECT_TRUE((gen.getErrorAllocationCounter()>0)?true:false); @@ -514,8 +508,6 @@ TEST(tuple_gen,try_to_free) { int i; for (i=0; i<65557; i++) { template_1.GenerateTuple(result); - uint32_t result_src = result.getClient(); - uint32_t result_dest = result.getServer(); uint16_t result_port = result.getClientPort(); gen.FreePort(0,result.getClientId(),result_port); } diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h new file mode 100644 index 00000000..5890a965 --- /dev/null +++ b/src/internal_api/trex_platform_api.h @@ -0,0 +1,134 @@ +/* + 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_PLATFORM_API_H__ +#define __TREX_PLATFORM_API_H__ + +#include <stdint.h> +#include <vector> + +/** + * Global stats + * + * @author imarom (06-Oct-15) + */ +class TrexPlatformGlobalStats { +public: + TrexPlatformGlobalStats() { + m_stats = {0}; + } + + struct { + double m_cpu_util; + + double m_tx_bps; + double m_rx_bps; + + double m_tx_pps; + double m_rx_pps; + + uint64_t m_total_tx_pkts; + uint64_t m_total_rx_pkts; + + uint64_t m_total_tx_bytes; + uint64_t m_total_rx_bytes; + + uint64_t m_tx_rx_errors; + } m_stats; +}; + +/** + * Per Interface stats + * + * @author imarom (26-Oct-15) + */ +class TrexPlatformInterfaceStats { + +public: + TrexPlatformInterfaceStats() { + m_stats = {0}; + } + +public: + + struct { + + double m_tx_bps; + double m_rx_bps; + + double m_tx_pps; + double m_rx_pps; + + uint64_t m_total_tx_pkts; + uint64_t m_total_rx_pkts; + + uint64_t m_total_tx_bytes; + uint64_t m_total_rx_bytes; + + uint64_t m_tx_rx_errors; + } m_stats; +}; + + +/** + * low level API interface + * can be implemented by DPDK or mock + * + * @author imarom (25-Oct-15) + */ + +class TrexPlatformApi { +public: + virtual void port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const = 0; + virtual void get_global_stats(TrexPlatformGlobalStats &stats) const = 0; + virtual void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const = 0; + virtual uint8_t get_dp_core_count() const = 0; + virtual ~TrexPlatformApi() {} +}; + + +/** + * DPDK implementation of the platform API + * + * @author imarom (26-Oct-15) + */ +class TrexDpdkPlatformApi : public TrexPlatformApi { +public: + void port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const; + void get_global_stats(TrexPlatformGlobalStats &stats) const; + void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const; + uint8_t get_dp_core_count() const; +}; + +/** + * MOCK implementation of the platform API + * + * @author imarom (26-Oct-15) + */ +class TrexMockPlatformApi : public TrexPlatformApi { +public: + void port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const {} + void get_global_stats(TrexPlatformGlobalStats &stats) const; + void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const; + uint8_t get_dp_core_count() const; +}; + +#endif /* __TREX_PLATFORM_API_H__ */ diff --git a/src/main.cpp b/src/main.cpp index 96789cdd..df9d8b40 100755 --- a/src/main.cpp +++ b/src/main.cpp @@ -185,25 +185,6 @@ int curent_time(){ #include <pthread.h> -void delay(int msec){ - - if (msec == 0) - {//user that requested that probebly wanted the minimal delay - //but because of scaling problem he have got 0 so we will give the min delay - //printf("\n\n\nERROR-Task delay ticks == 0 found in task %s task id = %d\n\n\n\n", - // SANB_TaskName(SANB_TaskIdSelf()), SANB_TaskIdSelf()); - msec =1; - - } - - struct timespec time1, remain; // 2 sec max delay - time1.tv_sec=msec/1000; - time1.tv_nsec=(msec - (time1.tv_sec*1000))*1000000; - - nanosleep(&time1,&remain); -} - - struct per_thread_t { pthread_t tid; }; @@ -233,11 +214,12 @@ void * thread_task(void *info){ char buf[100]; sprintf(buf,"my%d.erf",obj->thread_id); - volatile int i; - lpt->generate_erf(buf,*obj->preview_info); + lpt->start_generate_stateful(buf,*obj->preview_info); lpt->m_node_gen.DumpHist(stdout); printf("end thread %d \n",obj->thread_id); } + + return (NULL); } @@ -325,7 +307,7 @@ void test_load_list_of_cap_files(CParserOption * op){ lpt=fl.m_threads_info[i]; char buf[100]; sprintf(buf,"my%d.erf",i); - lpt->generate_erf(buf,op->preview); + lpt->start_generate_stateful(buf,op->preview); lpt->m_node_gen.DumpHist(stdout); } //sprintf(buf,"my%d.erf",7); @@ -353,7 +335,7 @@ int load_list_of_cap_files(CParserOption * op){ lpt->set_vif(&erf_vif); if ( (op->preview.getVMode() >1) || op->preview.getFileWrite() ) { - lpt->generate_erf(op->out_file,op->preview); + lpt->start_generate_stateful(op->out_file,op->preview); } lpt->m_node_gen.DumpHist(stdout); @@ -424,8 +406,6 @@ void update_tcp_seq_num(CCapFileFlowInfo * obj, int i; for (i=pkt_id+1; i<s; i++) { - uint32_t seq; - uint32_t ack; pkt=obj->GetPacket(i); tcp=pkt->m_pkt_indication.l4.m_tcp; @@ -509,7 +489,7 @@ int manipolate_capfile() { CCapFileFlowInfo flow_info; flow_info.Create(); - int res=flow_info.load_cap_file("avl/delay_10_rtsp_0.pcap",0,0); + 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); @@ -534,7 +514,7 @@ int manipolate_capfile_sip() { CCapFileFlowInfo flow_info; flow_info.Create(); - int res=flow_info.load_cap_file("avl/delay_10_sip_0.pcap",0,0); + 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); @@ -551,8 +531,8 @@ int manipolate_capfile_sip1() { CCapFileFlowInfo flow_info; flow_info.Create(); - int res=flow_info.load_cap_file("avl/delay_sip_0.pcap",0,0); - CFlowPktInfo * pkt=flow_info.GetPacket(1); + 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); @@ -588,7 +568,7 @@ public: void CMergeCapFileRec::Dump(FILE *fd,int _id){ - double time; + 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); } @@ -639,6 +619,8 @@ bool CMergeCapFileRec::Create(std::string cap_file, m_limit_number_of_packets =0; m_start_time = pkt->m_packet->get_time() ; m_offset = offset; + + return (true); } @@ -682,12 +664,12 @@ bool CMergeCapFile::run_merge(std::string to_cap_file){ int min_index=0; double min_time; - fprintf(stdout," --------------\n",cnt); + fprintf(stdout," --------------\n"); fprintf(stdout," pkt : %d \n",cnt); for (i=0; i<MERGE_CAP_FILES; i++) { m[i].Dump(stdout,i); } - fprintf(stdout," --------------\n",cnt); + fprintf(stdout," --------------\n"); bool valid = false; for (i=0; i<MERGE_CAP_FILES; i++) { @@ -721,6 +703,8 @@ bool CMergeCapFile::run_merge(std::string to_cap_file){ }; m_results.save_to_erf(to_cap_file,1); + + return (true); } diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp index a0af9fdf..c4ecb97d 100755 --- a/src/main_dpdk.cpp +++ b/src/main_dpdk.cpp @@ -55,7 +55,10 @@ limitations under the License. #include <common/arg/SimpleGlob.h> #include <common/arg/SimpleOpt.h> #include <common/basic_utils.h> + #include <stateless/cp/trex_stateless.h> +#include <stateless/dp/trex_stream_node.h> + #include <../linux_dpdk/version.h> extern "C" { @@ -73,6 +76,7 @@ extern "C" { #include "msg_manager.h" #include "platform_cfg.h" +#include <internal_api/trex_platform_api.h> #define RX_CHECK_MIX_SAMPLE_RATE 8 #define RX_CHECK_MIX_SAMPLE_RATE_1G 2 @@ -104,8 +108,6 @@ extern "C" int vmxnet3_xmit_set_callback(rte_mbuf_convert_to_one_seg_t cb); #define RTE_TEST_TX_DESC_DEFAULT 512 #define RTE_TEST_RX_DESC_DROP 0 - - static inline int get_vm_one_queue_enable(){ return (CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ?1:0); } @@ -686,12 +688,13 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t CSimpleOpt args(argc, argv, parser_options); bool latency_was_set=false; + (void)latency_was_set; + int a=0; int node_dump=0; po->preview.setFileWrite(true); po->preview.setRealTime(true); - int res1; uint32_t tmp_data; po->m_run_mode = CParserOption::RUN_MODE_INVALID; @@ -928,40 +931,30 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t } } - return 0; -} - - -int main_test(int argc , char * argv[]); - - - - -void delay(int msec){ - - if (msec == 0) - {//user that requested that probebly wanted the minimal delay - //but because of scaling problem he have got 0 so we will give the min delay - //printf("\n\n\nERROR-Task delay ticks == 0 found in task %s task id = %d\n\n\n\n", - // SANB_TaskName(SANB_TaskIdSelf()), SANB_TaskIdSelf()); - msec =1; + if ( get_is_stateless() ) { + if ( po->preview.get_is_rx_check_enable() ) { + parse_err("Rx check is not supported with interactive mode "); + } - } + if ( (! po->is_latency_disabled()) || (po->preview.getOnlyLatency()) ){ + parse_err("Latecny check is not supported with interactive mode "); + } - struct timespec time1, remain; // 2 sec max delay - time1.tv_sec=msec/1000; - time1.tv_nsec=(msec - (time1.tv_sec*1000))*1000000; + if ( po->preview.getSingleCore() ){ + parse_err("single core is not supported with interactive mode "); + } - nanosleep(&time1,&remain); + } + return 0; } - -static const char * default_argv[] = {"xx","-c", "0x7", "-n","2","-b","0000:0b:01.01"}; -static int argv_num = 7; - +int main_test(int argc , char * argv[]); +//static const char * default_argv[] = {"xx","-c", "0x7", "-n","2","-b","0000:0b:01.01"}; +//static int argv_num = 7; + #define RX_PTHRESH 8 /**< Default values of RX prefetch threshold reg. */ @@ -1145,8 +1138,8 @@ void CPhyEthIFStats::Clear(){ void CPhyEthIFStats::DumpAll(FILE *fd){ - #define DP_A4(f) printf(" %-40s : %llu \n",#f,f) - #define DP_A(f) if (f) printf(" %-40s : %llu \n",#f,f) + #define DP_A4(f) printf(" %-40s : %llu \n",#f, (unsigned long long)f) + #define DP_A(f) if (f) printf(" %-40s : %llu \n",#f, (unsigned long long)f) DP_A4(opackets); DP_A4(obytes); DP_A4(ipackets); @@ -1185,7 +1178,7 @@ public: m_port_id = portid; m_last_rx_rate = 0.0; m_last_tx_rate = 0.0; - m_last_pps=0.0; + m_last_tx_pps = 0.0; return (true); } void Delete(); @@ -1261,8 +1254,12 @@ public: return (m_last_rx_rate); } - float get_last_pps_rate(){ - return (m_last_pps); + float get_last_tx_pps_rate(){ + return (m_last_tx_pps); + } + + float get_last_rx_pps_rate(){ + return (m_last_rx_pps); } CPhyEthIFStats & get_stats(){ @@ -1315,12 +1312,14 @@ private: CBwMeasure m_bw_tx; CBwMeasure m_bw_rx; CPPSMeasure m_pps_tx; + CPPSMeasure m_pps_rx; CPhyEthIFStats m_stats; - float m_last_rx_rate; float m_last_tx_rate; - float m_last_pps; + float m_last_rx_rate; + float m_last_tx_pps; + float m_last_rx_pps; public: struct rte_eth_dev_info m_dev_info; }; @@ -1615,10 +1614,6 @@ void CPhyEthIF::macaddr_get(struct ether_addr *mac_addr){ void CPhyEthIF::get_stats_1g(CPhyEthIFStats *stats){ - int i; - uint64_t t=0; - - stats->ipackets += pci_reg_read(E1000_GPRC) ; stats->ibytes += (pci_reg_read(E1000_GORCL) ); @@ -1656,7 +1651,8 @@ void CPhyEthIF::get_stats_1g(CPhyEthIFStats *stats){ m_last_tx_rate = m_bw_tx.add(stats->obytes); m_last_rx_rate = m_bw_rx.add(stats->ibytes); - m_last_pps = m_pps_tx.add(stats->opackets); + m_last_tx_pps = m_pps_tx.add(stats->opackets); + m_last_rx_pps = m_pps_rx.add(stats->ipackets); } @@ -1666,14 +1662,15 @@ void CPhyEthIF::get_stats(CPhyEthIFStats *stats){ m_last_tx_rate = m_bw_tx.add(stats->obytes); m_last_rx_rate = m_bw_rx.add(stats->ibytes); - m_last_pps = m_pps_tx.add(stats->opackets); + m_last_tx_pps = m_pps_tx.add(stats->opackets); + m_last_rx_pps = m_pps_rx.add(stats->ipackets); } void dump_hw_state(FILE *fd,struct ixgbe_hw_stats *hs ){ - #define DP_A1(f) if (hs->f) fprintf(fd," %-40s : %llu \n",#f,hs->f) - #define DP_A2(f,m) for (i=0;i<m; i++) { if (hs->f[i]) fprintf(fd," %-40s[%d] : %llu \n",#f,i,hs->f[i]); } + #define DP_A1(f) if (hs->f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)hs->f) + #define DP_A2(f,m) for (i=0;i<m; i++) { if (hs->f[i]) fprintf(fd," %-40s[%d] : %llu \n",#f,i, (unsigned long long)hs->f[i]); } int i; //for (i=0;i<8; i++) { if (hs->mpc[i]) fprintf(fd," %-40s[%d] : %llu \n","mpc",i,hs->mpc[i]); } @@ -1855,6 +1852,9 @@ public: bool process_rx_pkt(pkt_dir_t dir,rte_mbuf_t * m); + virtual int update_mac_addr_from_global_cfg(pkt_dir_t dir, rte_mbuf_t *m); + + virtual pkt_dir_t port_id_to_dir(uint8_t port_id); public: void GetCoreCounters(CVirtualIFPerSideStats *stats); @@ -1867,7 +1867,11 @@ public: return ( CGlobalInfo::m_socket.port_to_socket( m_ports[0].m_port->get_port_id() ) ); } -private: + const CCorePerPort * get_ports() { + return m_ports; + } + +protected: int send_burst(CCorePerPort * lp_port, uint16_t len, @@ -1878,13 +1882,21 @@ private: -private: +protected: uint8_t m_core_id; uint16_t m_mbuf_cache; CCorePerPort m_ports[CS_NUM]; /* each core has 2 tx queues 1. client side and server side */ CNodeRing * m_ring_to_rx; + +} __rte_cache_aligned; ; + +class CCoreEthIFStateless : public CCoreEthIF { +public: + virtual int send_node(CGenNode * node); }; + + bool CCoreEthIF::Create(uint8_t core_id, uint16_t tx_client_queue_id, CPhyEthIF * tx_client_port, @@ -1985,6 +1997,7 @@ void CCoreEthIF::flush_rx_queue(void){ } } + int CCoreEthIF::flush_tx_queue(void){ /* flush both sides */ pkt_dir_t dir ; @@ -2084,6 +2097,8 @@ int CCoreEthIF::send_burst(CCorePerPort * lp_port, rte_pktmbuf_free(m); } } + + return (0); } @@ -2104,6 +2119,8 @@ int CCoreEthIF::send_pkt(CCorePerPort * lp_port, len = 0; } lp_port->m_len = len; + + return (0); } @@ -2140,6 +2157,23 @@ void CCoreEthIF::update_mac_addr(CGenNode * node,uint8_t *p){ } + +int CCoreEthIFStateless::send_node(CGenNode * no){ + CGenNodeStateless * node_sl=(CGenNodeStateless *) no; + + /* check that we have mbuf */ + rte_mbuf_t * m=node_sl->get_cache_mbuf(); + assert( m ); + pkt_dir_t dir=(pkt_dir_t)node_sl->get_mbuf_cache_dir(); + CCorePerPort * lp_port=&m_ports[dir]; + CVirtualIFPerSideStats * lp_stats = &m_stats[dir]; + rte_pktmbuf_refcnt_update(m,1); + send_pkt(lp_port,m,lp_stats); + return (0); +}; + + + int CCoreEthIF::send_node(CGenNode * node){ if ( unlikely( node->get_cache_mbuf() !=NULL ) ) { @@ -2233,6 +2267,29 @@ int CCoreEthIF::send_node(CGenNode * node){ } +int CCoreEthIF::update_mac_addr_from_global_cfg(pkt_dir_t dir, + rte_mbuf_t *m){ + assert(m); + assert(dir<2); + CCorePerPort * lp_port=&m_ports[dir]; + uint8_t *p=rte_pktmbuf_mtod(m, uint8_t*); + uint8_t p_id=lp_port->m_port->get_port_id(); + + memcpy(p,CGlobalInfo::m_options.get_dst_src_mac_addr(p_id),12); + return (0); +} + +pkt_dir_t +CCoreEthIF::port_id_to_dir(uint8_t port_id) { + + for (pkt_dir_t dir = 0; dir < CS_NUM; dir++) { + if (m_ports[dir].m_port->get_port_id() == port_id) { + return dir; + } + } + + return (CS_INVALID); +} class CLatencyHWPort : public CPortLatencyHWBase { public: @@ -2420,6 +2477,10 @@ public: uint64_t oerrors; float m_total_tx_bps; + float m_total_tx_pps; + + float m_total_rx_bps; + float m_total_rx_pps; }; class CGlobalStats { @@ -2456,6 +2517,7 @@ public: float m_tx_bps; float m_rx_bps; float m_tx_pps; + float m_rx_pps; float m_tx_cps; float m_tx_expected_cps; float m_tx_expected_pps; @@ -2488,7 +2550,7 @@ std::string CGlobalStats::get_field(std::string name,float &f){ std::string CGlobalStats::get_field(std::string name,uint64_t &f){ char buff[200]; - sprintf(buff,"\"%s\":%llu,",name.c_str(),f); + sprintf(buff,"\"%s\":%llu,",name.c_str(), (unsigned long long)f); return (std::string(buff)); } @@ -2500,7 +2562,7 @@ std::string CGlobalStats::get_field_port(int port,std::string name,float &f){ std::string CGlobalStats::get_field_port(int port,std::string name,uint64_t &f){ char buff[200]; - sprintf(buff,"\"%s-%d\":%llu,",name.c_str(),port,f); + sprintf(buff,"\"%s-%d\":%llu,",name.c_str(),port, (unsigned long long)f); return (std::string(buff)); } @@ -2516,6 +2578,7 @@ void CGlobalStats::dump_json(std::string & json){ json+=GET_FIELD(m_tx_bps); json+=GET_FIELD(m_rx_bps); json+=GET_FIELD(m_tx_pps); + json+=GET_FIELD(m_rx_pps); json+=GET_FIELD(m_tx_cps); json+=GET_FIELD(m_tx_expected_cps); json+=GET_FIELD(m_tx_expected_pps); @@ -2550,6 +2613,9 @@ void CGlobalStats::dump_json(std::string & json){ json+=GET_FIELD_PORT(i,ierrors) ; json+=GET_FIELD_PORT(i,oerrors) ; json+=GET_FIELD_PORT(i,m_total_tx_bps); + json+=GET_FIELD_PORT(i,m_total_tx_pps); + json+=GET_FIELD_PORT(i,m_total_rx_bps); + json+=GET_FIELD_PORT(i,m_total_rx_pps); } json+=m_template.dump_as_json("template"); json+="\"unknown\":0}}" ; @@ -2569,7 +2635,7 @@ void CGlobalStats::DumpAllPorts(FILE *fd){ fprintf (fd," Platform_factor : %2.1f \n",m_platform_factor); fprintf (fd," Total-Tx : %s ",double_to_human_str(m_tx_bps,"bps",KBYE_1000).c_str()); if ( CGlobalInfo::is_learn_mode() ) { - fprintf (fd," Nat_time_out : %8llu \n",m_total_nat_time_out); + fprintf (fd," Nat_time_out : %8llu \n", (unsigned long long)m_total_nat_time_out); }else{ fprintf (fd,"\n"); } @@ -2577,49 +2643,52 @@ void CGlobalStats::DumpAllPorts(FILE *fd){ fprintf (fd," Total-Rx : %s ",double_to_human_str(m_rx_bps,"bps",KBYE_1000).c_str()); if ( CGlobalInfo::is_learn_mode() ) { - fprintf (fd," Nat_no_fid : %8llu \n",m_total_nat_no_fid); + fprintf (fd," Nat_no_fid : %8llu \n", (unsigned long long)m_total_nat_no_fid); }else{ fprintf (fd,"\n"); } fprintf (fd," Total-PPS : %s ",double_to_human_str(m_tx_pps,"pps",KBYE_1000).c_str()); if ( CGlobalInfo::is_learn_mode() ) { - fprintf (fd," Total_nat_active: %8llu \n",m_total_nat_active); + fprintf (fd," Total_nat_active: %8llu \n", (unsigned long long)m_total_nat_active); }else{ fprintf (fd,"\n"); } fprintf (fd," Total-CPS : %s ",double_to_human_str(m_tx_cps,"cps",KBYE_1000).c_str()); if ( CGlobalInfo::is_learn_mode() ) { - fprintf (fd," Total_nat_open : %8llu \n",m_total_nat_open); + fprintf (fd," Total_nat_open : %8llu \n", (unsigned long long)m_total_nat_open); }else{ fprintf (fd,"\n"); } fprintf (fd,"\n"); fprintf (fd," Expected-PPS : %s ",double_to_human_str(m_tx_expected_pps,"pps",KBYE_1000).c_str()); if ( CGlobalInfo::is_learn_verify_mode() ) { - fprintf (fd," Nat_learn_errors: %8llu \n",m_total_nat_learn_error); + fprintf (fd," Nat_learn_errors: %8llu \n", (unsigned long long)m_total_nat_learn_error); }else{ fprintf (fd,"\n"); } fprintf (fd," Expected-CPS : %s \n",double_to_human_str(m_tx_expected_cps,"cps",KBYE_1000).c_str()); fprintf (fd," Expected-BPS : %s \n",double_to_human_str(m_tx_expected_bps,"bps",KBYE_1000).c_str()); fprintf (fd,"\n"); - fprintf (fd," Active-flows : %8llu Clients : %8llu Socket-util : %3.4f %% \n",(uint64_t)m_active_flows,m_total_clients,m_socket_util); + fprintf (fd," Active-flows : %8llu Clients : %8llu Socket-util : %3.4f %% \n", + (unsigned long long)m_active_flows, + (unsigned long long)m_total_clients, + m_socket_util); fprintf (fd," Open-flows : %8llu Servers : %8llu Socket : %8llu Socket/Clients : %.1f \n", - (uint64_t)m_open_flows, - m_total_servers, - m_active_sockets, + (unsigned long long)m_open_flows, + (unsigned long long)m_total_servers, + (unsigned long long)m_active_sockets, (float)m_active_sockets/(float)m_total_clients); if (m_total_alloc_error) { - fprintf (fd," Total_alloc_err : %llu \n",(uint64_t)m_total_alloc_error); + fprintf (fd," Total_alloc_err : %llu \n", (unsigned long long)m_total_alloc_error); } if ( m_total_queue_full ){ - fprintf (fd," Total_queue_full : %llu \n",(uint64_t)m_total_queue_full); + fprintf (fd," Total_queue_full : %llu \n", (unsigned long long)m_total_queue_full); } if (m_total_queue_drop) { - fprintf (fd," Total_queue_drop : %llu \n",(uint64_t)m_total_queue_drop); + fprintf (fd," Total_queue_drop : %llu \n", (unsigned long long)m_total_queue_drop); } //m_template.Dump(fd); @@ -2643,8 +2712,8 @@ void CGlobalStats::Dump(FILE *fd,DumpFormat mode){ CPerPortStats * lp=&m_port[i]; fprintf(fd,"port : %d \n",(int)i); fprintf(fd,"------------\n"); - #define GS_DP_A4(f) fprintf(fd," %-40s : %llu \n",#f,lp->f) - #define GS_DP_A(f) if (lp->f) fprintf(fd," %-40s : %llu \n",#f,lp->f) + #define GS_DP_A4(f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)lp->f) + #define GS_DP_A(f) if (lp->f) fprintf(fd," %-40s : %llu \n",#f, (unsigned long long)lp->f) GS_DP_A4(opackets); GS_DP_A4(obytes); GS_DP_A4(ipackets); @@ -2719,10 +2788,10 @@ void CGlobalStats::Dump(FILE *fd,DumpFormat mode){ -struct CGlobalPortCfg { +struct CGlobalTRex { public: - CGlobalPortCfg (){ + CGlobalTRex (){ m_max_ports=4; m_max_cores=1; m_cores_to_dual_ports=0; @@ -2732,10 +2801,11 @@ public: m_expected_pps=0.0; m_expected_cps=0.0; m_expected_bps=0.0; + m_trex_stateless = NULL; } public: - bool Create(bool is_stateless); + bool Create(); void Delete(); int ixgbe_prob_init(); @@ -2754,6 +2824,8 @@ public: public: int start_send_master(); + int start_master_stateless(); + int run_in_core(virtual_thread_id_t virt_core_id); int stop_core(virtual_thread_id_t virt_core_id); @@ -2800,7 +2872,6 @@ public: - int test_send1(); int rcv_send(int port,int queue_id); int rcv_send_all(int queue_id); @@ -2866,7 +2937,9 @@ public: CPhyEthIF m_ports[BP_MAX_PORTS]; - CCoreEthIF m_cores_vif[BP_MAX_CORES]; /* counted from 1 , 2,3 core zero is reserve*/ + CCoreEthIF m_cores_vif_sf[BP_MAX_CORES]; /* counted from 1 , 2,3 core zero is reserve - stateful */ + CCoreEthIFStateless m_cores_vif_sl[BP_MAX_CORES]; /* counted from 1 , 2,3 core zero is reserve - stateless*/ + CCoreEthIF * m_cores_vif[BP_MAX_CORES]; CParserOption m_po ; @@ -2889,54 +2962,14 @@ private: CLatencyPktInfo m_latency_pkt; CZMqPublisher m_zmq_publisher; -}; - - -int CGlobalPortCfg::test_send1(){ - - CParserOption po ; - CFlowGenList fl; - - po.cfg_file = "cap2/dns.yaml"; - //po.cfg_file = "cap2/sfr3.yaml"; - //po.cfg_file = "cap2/sfr4.yaml"; - //po.cfg_file = "cap2/sfr.yaml"; - - po.preview.setVMode(3); - po.preview.setFileWrite(true); - - fl.Create(); - - fl.load_from_yaml(po.cfg_file,1); - //fl.DumpPktSize(); - - fl.generate_p_thread_info(1); - CFlowGenListPerThread * lpt; - - int i; - for (i=0; i<1; i++) { - lpt = fl.m_threads_info[i]; - //CNullIF * erf_vif = new CNullIF(); - CVirtualIF * erf_vif = &m_cores_vif[0]; - lpt->set_vif(erf_vif); - lpt->generate_erf("hey",po.preview); - lpt->m_node_gen.DumpHist(stdout); - lpt->DumpStats(stdout); - } +public: + TrexStateless *m_trex_stateless; +}; - m_cores_vif[0].flush_tx_queue(); - delay(1000); - //fprintf(stdout," drop : %llu \n",m_test_drop); - m_cores_vif[0].DumpCoreStats(stdout); - m_cores_vif[0].DumpIfStats(stdout); - fl.Delete(); -} - - -int CGlobalPortCfg::rcv_send(int port,int queue_id){ +int CGlobalTRex::rcv_send(int port,int queue_id){ CPhyEthIF * lp=&m_ports[port]; rte_mbuf_t * rx_pkts[32]; @@ -2955,7 +2988,7 @@ int CGlobalPortCfg::rcv_send(int port,int queue_id){ return (0); } -int CGlobalPortCfg::rcv_send_all(int queue_id){ +int CGlobalTRex::rcv_send_all(int queue_id){ int i; for (i=0; i<m_max_ports; i++) { rcv_send(i,queue_id); @@ -2966,16 +2999,16 @@ int CGlobalPortCfg::rcv_send_all(int queue_id){ -int CGlobalPortCfg::test_send(){ +int CGlobalTRex::test_send(){ int i; - CPhyEthIF * lp=&m_ports[0]; - //set_promisc_all(true); //create_sctp_pkt(); create_udp_pkt(); CRx_check_header rx_check_header; + (void)rx_check_header; + rx_check_header.m_time_stamp=0x1234567; rx_check_header.m_option_type=RX_CHECK_V4_OPT_TYPE; rx_check_header.m_option_len=RX_CHECK_V4_OPT_LEN; @@ -3049,7 +3082,7 @@ int CGlobalPortCfg::test_send(){ }*/ #endif - fprintf(stdout," drop : %llu \n",m_test_drop); + fprintf(stdout," drop : %llu \n", (unsigned long long)m_test_drop); return (0); } @@ -3118,7 +3151,7 @@ const uint8_t sctp_pkt1[]={ -int CGlobalPortCfg::create_pkt(uint8_t *pkt,int pkt_size){ +int CGlobalTRex::create_pkt(uint8_t *pkt,int pkt_size){ rte_mempool_t * mp= CGlobalInfo::m_mem_pool[0].m_big_mbuf_pool ; rte_mbuf_t * m=rte_pktmbuf_alloc(mp); @@ -3138,17 +3171,17 @@ int CGlobalPortCfg::create_pkt(uint8_t *pkt,int pkt_size){ return (0); } -int CGlobalPortCfg::create_udp_pkt(){ +int CGlobalTRex::create_udp_pkt(){ return (create_pkt((uint8_t*)udp_pkt,sizeof(udp_pkt))); } -int CGlobalPortCfg::create_sctp_pkt(){ +int CGlobalTRex::create_sctp_pkt(){ return (create_pkt((uint8_t*)sctp_pkt1,sizeof(sctp_pkt1))); } /* test by sending 10 packets ...*/ -int CGlobalPortCfg::test_send_pkts(uint16_t queue_id, +int CGlobalTRex::test_send_pkts(uint16_t queue_id, int pkt, int port){ @@ -3174,26 +3207,30 @@ int CGlobalPortCfg::test_send_pkts(uint16_t queue_id, -int CGlobalPortCfg::set_promisc_all(bool enable){ +int CGlobalTRex::set_promisc_all(bool enable){ int i; for (i=0; i<m_max_ports; i++) { CPhyEthIF * _if=&m_ports[i]; _if->set_promiscuous(enable); } + + return (0); } -int CGlobalPortCfg::reset_counters(){ +int CGlobalTRex::reset_counters(){ int i; for (i=0; i<m_max_ports; i++) { CPhyEthIF * _if=&m_ports[i]; _if->stats_clear(); } + + return (0); } -bool CGlobalPortCfg::is_all_links_are_up(bool dump){ +bool CGlobalTRex::is_all_links_are_up(bool dump){ bool all_link_are=true; int i; for (i=0; i<m_max_ports; i++) { @@ -3212,7 +3249,7 @@ bool CGlobalPortCfg::is_all_links_are_up(bool dump){ -int CGlobalPortCfg::ixgbe_rx_queue_flush(){ +int CGlobalTRex::ixgbe_rx_queue_flush(){ int i; for (i=0; i<m_max_ports; i++) { CPhyEthIF * _if=&m_ports[i]; @@ -3222,7 +3259,7 @@ int CGlobalPortCfg::ixgbe_rx_queue_flush(){ } -int CGlobalPortCfg::ixgbe_configure_mg(void){ +int CGlobalTRex::ixgbe_configure_mg(void){ int i; CLatencyManagerCfg mg_cfg; mg_cfg.m_max_ports = m_max_ports; @@ -3262,10 +3299,12 @@ int CGlobalPortCfg::ixgbe_configure_mg(void){ m_mg.Create(&mg_cfg); m_mg.set_mask(CGlobalInfo::m_options.m_latency_mask); + + return (0); } -int CGlobalPortCfg::ixgbe_start(void){ +int CGlobalTRex::ixgbe_start(void){ int i; for (i=0; i<m_max_ports; i++) { @@ -3384,11 +3423,15 @@ int CGlobalPortCfg::ixgbe_start(void){ */ int port_offset=0; - int queue_offset=0; for (i=0; i<get_cores_tx(); i++) { int j=(i+1); int queue_id=((j-1)/get_base_num_cores() ); /* for the first min core queue 0 , then queue 1 etc */ - m_cores_vif[j].Create(j, + if ( get_is_stateless() ){ + m_cores_vif[j]=&m_cores_vif_sl[j]; + }else{ + m_cores_vif[j]=&m_cores_vif_sf[j]; + } + m_cores_vif[j]->Create(j, queue_id, &m_ports[port_offset], /* 0,2*/ queue_id, @@ -3403,28 +3446,39 @@ int CGlobalPortCfg::ixgbe_start(void){ fprintf(stdout," -------------------------------\n"); CCoreEthIF::DumpIfCfgHeader(stdout); for (i=0; i<get_cores_tx(); i++) { - m_cores_vif[i+1].DumpIfCfg(stdout); + m_cores_vif[i+1]->DumpIfCfg(stdout); } fprintf(stdout," -------------------------------\n"); + + return (0); } -bool CGlobalPortCfg::Create(bool is_stateless){ +bool CGlobalTRex::Create(){ + CFlowsYamlInfo pre_yaml_info; - /* hack - need to refactor */ - if (!is_stateless) { - if ( !m_zmq_publisher.Create( CGlobalInfo::m_options.m_zmq_port, - !CGlobalInfo::m_options.preview.get_zmq_publish_enable() ) ){ - return (false); - } + 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 { + pre_yaml_info.load_from_yaml_file(CGlobalInfo::m_options.cfg_file); } - /* We load the YAML twice, - this is the first time. to update global flags */ - CFlowsYamlInfo pre_yaml_info; - if (!is_stateless) { - pre_yaml_info.load_from_yaml_file(CGlobalInfo::m_options.cfg_file); - } + if ( !m_zmq_publisher.Create( CGlobalInfo::m_options.m_zmq_port, + !CGlobalInfo::m_options.preview.get_zmq_publish_enable() ) ){ + return (false); + } if ( pre_yaml_info.m_vlan_info.m_enable ){ CGlobalInfo::m_options.preview.set_vlan_mode_enable(true); @@ -3434,16 +3488,17 @@ bool CGlobalPortCfg::Create(bool is_stateless){ ixgbe_prob_init(); cores_prob_init(); queues_prob_init(); - /* allocate rings */ - assert( CMsgIns::Ins()->Create(get_cores_tx()) ); - if ( sizeof(CGenNodeNatInfo) != sizeof(CGenNode) ) { - printf("ERROR sizeof(CGenNodeNatInfo) %d != sizeof(CGenNode) %d must be the same size \n",sizeof(CGenNodeNatInfo),sizeof(CGenNode)); - assert(0); - } + /* allocate rings */ + assert( CMsgIns::Ins()->Create(get_cores_tx()) ); + + if ( sizeof(CGenNodeNatInfo) != sizeof(CGenNode) ) { + printf("ERROR sizeof(CGenNodeNatInfo) %lu != sizeof(CGenNode) %lu must be the same size \n",sizeof(CGenNodeNatInfo),sizeof(CGenNode)); + assert(0); + } if ( sizeof(CGenNodeLatencyPktInfo) != sizeof(CGenNode) ) { - printf("ERROR sizeof(CGenNodeLatencyPktInfo) %d != sizeof(CGenNode) %d must be the same size \n",sizeof(CGenNodeLatencyPktInfo),sizeof(CGenNode)); + printf("ERROR sizeof(CGenNodeLatencyPktInfo) %lu != sizeof(CGenNode) %lu must be the same size \n",sizeof(CGenNodeLatencyPktInfo),sizeof(CGenNode)); assert(0); } @@ -3463,16 +3518,13 @@ bool CGlobalPortCfg::Create(bool is_stateless){ return (true); } -void CGlobalPortCfg::Delete(){ +void CGlobalTRex::Delete(){ m_zmq_publisher.Delete(); } -int CGlobalPortCfg::ixgbe_prob_init(void){ - - uint8_t nb_ports; - +int CGlobalTRex::ixgbe_prob_init(void){ m_max_ports = rte_eth_dev_count(); if (m_max_ports == 0) @@ -3561,13 +3613,13 @@ int CGlobalPortCfg::ixgbe_prob_init(void){ return (0); } -int CGlobalPortCfg::cores_prob_init(){ +int CGlobalTRex::cores_prob_init(){ m_max_cores = rte_lcore_count(); assert(m_max_cores>0); return (0); } -int CGlobalPortCfg::queues_prob_init(){ +int CGlobalTRex::queues_prob_init(){ if (m_max_cores < 2) { rte_exit(EXIT_FAILURE, "number of cores should be at least 3 \n"); @@ -3608,7 +3660,7 @@ int CGlobalPortCfg::queues_prob_init(){ } -void CGlobalPortCfg::dump_config(FILE *fd){ +void CGlobalTRex::dump_config(FILE *fd){ fprintf(fd," number of ports : %u \n",m_max_ports); fprintf(fd," max cores for 2 ports : %u \n",m_cores_to_dual_ports); fprintf(fd," max queue per port : %u \n",m_max_queues_per_port); @@ -3616,7 +3668,7 @@ void CGlobalPortCfg::dump_config(FILE *fd){ -void CGlobalPortCfg::dump_post_test_stats(FILE *fd){ +void CGlobalTRex::dump_post_test_stats(FILE *fd){ uint64_t pkt_out=0; uint64_t pkt_out_bytes=0; uint64_t pkt_in_bytes=0; @@ -3627,7 +3679,7 @@ void CGlobalPortCfg::dump_post_test_stats(FILE *fd){ int i; for (i=0; i<get_cores_tx(); i++) { - CCoreEthIF * erf_vif = &m_cores_vif[i+1]; + CCoreEthIF * erf_vif = m_cores_vif[i+1]; CVirtualIFPerSideStats stats; erf_vif->GetCoreCounters(&stats); sw_pkt_out += stats.m_tx_pkt; @@ -3652,17 +3704,17 @@ void CGlobalPortCfg::dump_post_test_stats(FILE *fd){ fprintf (fd," summary stats \n"); fprintf (fd," -------------- \n"); - fprintf (fd," Total-pkt-drop : %d pkts \n",(int64_t)(pkt_out-pkt_in)); - fprintf (fd," Total-tx-bytes : %llu bytes \n",pkt_out_bytes); - fprintf (fd," Total-tx-sw-bytes : %llu bytes \n",sw_pkt_out_bytes); - fprintf (fd," Total-rx-bytes : %llu byte \n",pkt_in_bytes); + fprintf (fd," Total-pkt-drop : %llu pkts \n",(unsigned long long)(pkt_out-pkt_in)); + fprintf (fd," Total-tx-bytes : %llu bytes \n", (unsigned long long)pkt_out_bytes); + fprintf (fd," Total-tx-sw-bytes : %llu bytes \n", (unsigned long long)sw_pkt_out_bytes); + fprintf (fd," Total-rx-bytes : %llu byte \n", (unsigned long long)pkt_in_bytes); fprintf (fd," \n"); - fprintf (fd," Total-tx-pkt : %llu pkts \n",pkt_out); - fprintf (fd," Total-rx-pkt : %llu pkts \n",pkt_in); - fprintf (fd," Total-sw-tx-pkt : %llu pkts \n",sw_pkt_out); - fprintf (fd," Total-sw-err : %llu pkts \n",sw_pkt_out_err); + fprintf (fd," Total-tx-pkt : %llu pkts \n", (unsigned long long)pkt_out); + fprintf (fd," Total-rx-pkt : %llu pkts \n", (unsigned long long)pkt_in); + fprintf (fd," Total-sw-tx-pkt : %llu pkts \n", (unsigned long long)sw_pkt_out); + fprintf (fd," Total-sw-err : %llu pkts \n", (unsigned long long)sw_pkt_out_err); if ( !CGlobalInfo::m_options.is_latency_disabled() ){ @@ -3675,7 +3727,7 @@ void CGlobalPortCfg::dump_post_test_stats(FILE *fd){ } -void CGlobalPortCfg::update_stats(){ +void CGlobalTRex::update_stats(){ int i; for (i=0; i<m_max_ports; i++) { @@ -3696,12 +3748,13 @@ void CGlobalPortCfg::update_stats(){ } -void CGlobalPortCfg::get_stats(CGlobalStats & stats){ +void CGlobalTRex::get_stats(CGlobalStats & stats){ int i; float total_tx=0.0; float total_rx=0.0; - float total_pps=0.0; + float total_tx_pps=0.0; + float total_rx_pps=0.0; stats.m_total_tx_pkts = 0; stats.m_total_rx_pkts = 0; @@ -3729,6 +3782,9 @@ void CGlobalPortCfg::get_stats(CGlobalStats & stats){ stp->ierrors = st.ierrors; stp->oerrors = st.oerrors; stp->m_total_tx_bps = _if->get_last_tx_rate()*_1Mb_DOUBLE; + stp->m_total_tx_pps = _if->get_last_tx_pps_rate(); + stp->m_total_rx_bps = _if->get_last_rx_rate()*_1Mb_DOUBLE; + stp->m_total_rx_pps = _if->get_last_rx_pps_rate(); stats.m_total_tx_pkts += st.opackets; stats.m_total_rx_pkts += st.ipackets; @@ -3737,7 +3793,8 @@ void CGlobalPortCfg::get_stats(CGlobalStats & stats){ total_tx +=_if->get_last_tx_rate(); total_rx +=_if->get_last_rx_rate(); - total_pps +=_if->get_last_pps_rate(); + total_tx_pps +=_if->get_last_tx_pps_rate(); + total_rx_pps +=_if->get_last_rx_pps_rate(); } @@ -3798,7 +3855,13 @@ void CGlobalPortCfg::get_stats(CGlobalStats & stats){ stats.m_total_clients = total_clients; stats.m_total_servers = total_servers; stats.m_active_sockets = active_sockets; - stats.m_socket_util =100.0*(double)active_sockets/(double)total_sockets; + + if (total_sockets != 0) { + stats.m_socket_util =100.0*(double)active_sockets/(double)total_sockets; + } else { + stats.m_socket_util = 0; + } + float drop_rate=total_tx-total_rx; @@ -3814,7 +3877,8 @@ void CGlobalPortCfg::get_stats(CGlobalStats & stats){ stats.m_tx_bps = total_tx*pf*_1Mb_DOUBLE; stats.m_rx_bps = total_rx*pf*_1Mb_DOUBLE; - stats.m_tx_pps = total_pps*pf; + stats.m_tx_pps = total_tx_pps*pf; + stats.m_rx_pps = total_rx_pps*pf; stats.m_tx_cps = m_last_total_cps*pf; stats.m_tx_expected_cps = m_expected_cps*pf; @@ -3822,7 +3886,7 @@ void CGlobalPortCfg::get_stats(CGlobalStats & stats){ stats.m_tx_expected_bps = m_expected_bps*pf; } -bool CGlobalPortCfg::sanity_check(){ +bool CGlobalTRex::sanity_check(){ CFlowGenListPerThread * lpt; uint32_t errors=0; @@ -3842,7 +3906,7 @@ bool CGlobalPortCfg::sanity_check(){ /* dump the template info */ -void CGlobalPortCfg::dump_template_info(std::string & json){ +void CGlobalTRex::dump_template_info(std::string & json){ CFlowGenListPerThread * lpt = m_fl.m_threads_info[0]; CFlowsYamlInfo * yaml_info=&lpt->m_yaml_info; @@ -3857,7 +3921,7 @@ void CGlobalPortCfg::dump_template_info(std::string & json){ json+="]}" ; } -void CGlobalPortCfg::dump_stats(FILE *fd,std::string & json, +void CGlobalTRex::dump_stats(FILE *fd,std::string & json, CGlobalStats::DumpFormat format){ CGlobalStats stats; update_stats(); @@ -3897,11 +3961,15 @@ void CGlobalPortCfg::dump_stats(FILE *fd,std::string & json, } -int CGlobalPortCfg::run_in_master(){ +int CGlobalTRex::run_in_master(){ std::string json; bool was_stopped=false; + if ( get_is_stateless() ) { + m_trex_stateless->launch_control_plane(); + } + while ( true ) { if ( CGlobalInfo::m_options.preview.get_no_keyboard() ==false ){ @@ -3953,10 +4021,10 @@ int CGlobalPortCfg::run_in_master(){ m_fl.m_threads_info[0]->m_node_gen.dump_json(json); m_zmq_publisher.publish_json(json); - dump_template_info(json); - m_zmq_publisher.publish_json(json); - - + if ( !get_is_stateless() ){ + dump_template_info(json); + m_zmq_publisher.publish_json(json); + } if ( !CGlobalInfo::m_options.is_latency_disabled() ){ m_mg.update(); @@ -4009,6 +4077,10 @@ int CGlobalPortCfg::run_in_master(){ } + /* stateless info */ + m_trex_stateless->generate_publish_snapshot(json); + m_zmq_publisher.publish_json(json); + delay(500); if ( is_all_cores_finished() ) { @@ -4027,7 +4099,7 @@ int CGlobalPortCfg::run_in_master(){ -int CGlobalPortCfg::run_in_laterncy_core(void){ +int CGlobalTRex::run_in_laterncy_core(void){ if ( !CGlobalInfo::m_options.is_latency_disabled() ){ m_mg.start(0); } @@ -4035,12 +4107,12 @@ int CGlobalPortCfg::run_in_laterncy_core(void){ } -int CGlobalPortCfg::stop_core(virtual_thread_id_t virt_core_id){ +int CGlobalTRex::stop_core(virtual_thread_id_t virt_core_id){ m_signal[virt_core_id]=1; return (0); } -int CGlobalPortCfg::run_in_core(virtual_thread_id_t virt_core_id){ +int CGlobalTRex::run_in_core(virtual_thread_id_t virt_core_id){ CPreviewMode *lp=&CGlobalInfo::m_options.preview; if ( lp->getSingleCore() && @@ -4051,19 +4123,23 @@ int CGlobalPortCfg::run_in_core(virtual_thread_id_t virt_core_id){ return (0); } + assert(m_fl_was_init); CFlowGenListPerThread * lpt; lpt = m_fl.m_threads_info[virt_core_id-1]; - lpt->generate_erf(CGlobalInfo::m_options.out_file,*lp); - //lpt->m_node_gen.DumpHist(stdout); - //lpt->DumpStats(stdout); + + if (get_is_stateless()) { + lpt->start_stateless_daemon(); + }else{ + lpt->start_generate_stateful(CGlobalInfo::m_options.out_file,*lp); + } m_signal[virt_core_id]=1; return (0); } -int CGlobalPortCfg::stop_master(){ +int CGlobalTRex::stop_master(){ delay(1000); std::string json; @@ -4084,7 +4160,7 @@ int CGlobalPortCfg::stop_master(){ int i; for (i=0; i<get_cores_tx(); i++) { lpt = m_fl.m_threads_info[i]; - CCoreEthIF * erf_vif = &m_cores_vif[i+1]; + CCoreEthIF * erf_vif = m_cores_vif[i+1]; erf_vif->DumpCoreStats(stdout); erf_vif->DumpIfStats(stdout); @@ -4115,9 +4191,10 @@ int CGlobalPortCfg::stop_master(){ dump_post_test_stats(stdout); m_fl.Delete(); + return (0); } -bool CGlobalPortCfg::is_all_cores_finished(){ +bool CGlobalTRex::is_all_cores_finished(){ int i; for (i=0; i<get_cores_tx(); i++) { if ( m_signal[i+1]==0){ @@ -4128,8 +4205,34 @@ bool CGlobalPortCfg::is_all_cores_finished(){ } +int CGlobalTRex::start_master_stateless(){ + int i; + for (i=0; i<BP_MAX_CORES; i++) { + m_signal[i]=0; + } + m_fl.Create(); + m_expected_pps = 0; + m_expected_cps = 0; + m_expected_bps = 0; + + m_fl.generate_p_thread_info(get_cores_tx()); + CFlowGenListPerThread * lpt; + + for (i=0; i<get_cores_tx(); i++) { + lpt = m_fl.m_threads_info[i]; + CVirtualIF * erf_vif = m_cores_vif[i+1]; + lpt->set_vif(erf_vif); + lpt->m_node_gen.m_socket_id =m_cores_vif[i+1]->get_socket_id(); + } + m_fl_was_init=true; + + return (0); +} + -int CGlobalPortCfg::start_send_master(){ + + +int CGlobalTRex::start_send_master(){ int i; for (i=0; i<BP_MAX_CORES; i++) { m_signal[i]=0; @@ -4174,20 +4277,26 @@ int CGlobalPortCfg::start_send_master(){ for (i=0; i<get_cores_tx(); i++) { lpt = m_fl.m_threads_info[i]; //CNullIF * erf_vif = new CNullIF(); - CVirtualIF * erf_vif = &m_cores_vif[i+1]; + CVirtualIF * erf_vif = m_cores_vif[i+1]; lpt->set_vif(erf_vif); /* socket id */ - lpt->m_node_gen.m_socket_id =m_cores_vif[i+1].get_socket_id(); + lpt->m_node_gen.m_socket_id =m_cores_vif[i+1]->get_socket_id(); } m_fl_was_init=true; + return (0); } //////////////////////////////////////////// -static CGlobalPortCfg ports_cfg; +static CGlobalTRex g_trex; + + +TrexStateless * get_stateless_obj() { + return g_trex.m_trex_stateless; +} static int latency_one_lcore(__attribute__((unused)) void *dummy) { @@ -4196,34 +4305,22 @@ static int latency_one_lcore(__attribute__((unused)) void *dummy) if ( lpsock->thread_phy_is_latency( phy_id ) ){ - ports_cfg.run_in_laterncy_core(); + g_trex.run_in_laterncy_core(); }else{ if ( lpsock->thread_phy_is_master( phy_id ) ) { - ports_cfg.run_in_master(); + g_trex.run_in_master(); delay(1); }else{ delay((uint32_t)(1000.0*CGlobalInfo::m_options.m_duration)); /* this core has stopped */ - ports_cfg.m_signal[ lpsock->thread_phy_to_virt( phy_id ) ]=1; + g_trex.m_signal[ lpsock->thread_phy_to_virt( phy_id ) ]=1; } } return 0; } -static int stateless_entry(__attribute__((unused)) void *dummy) { - CPlatformSocketInfo * lpsock=&CGlobalInfo::m_socket; - physical_thread_id_t phy_id = rte_lcore_id(); - - if (lpsock->thread_phy_is_master( phy_id )) { - TrexStateless::get_instance().launch_control_plane(); - } else { - TrexStateless::get_instance().launch_on_dp_core(phy_id); - } - - return (0); -} static int slave_one_lcore(__attribute__((unused)) void *dummy) { @@ -4232,13 +4329,13 @@ static int slave_one_lcore(__attribute__((unused)) void *dummy) if ( lpsock->thread_phy_is_latency( phy_id ) ){ - ports_cfg.run_in_laterncy_core(); + g_trex.run_in_laterncy_core(); }else{ if ( lpsock->thread_phy_is_master( phy_id ) ) { - ports_cfg.run_in_master(); + g_trex.run_in_master(); delay(1); }else{ - ports_cfg.run_in_core( lpsock->thread_phy_to_virt( phy_id ) ); + g_trex.run_in_core( lpsock->thread_phy_to_virt( phy_id ) ); } } return 0; @@ -4329,7 +4426,6 @@ int update_global_info_from_platform_file(){ int update_dpdk_args(void){ - uint32_t cores_number; CPlatformSocketInfo * lpsock=&CGlobalInfo::m_socket; CParserOption * lpop= &CGlobalInfo::m_options; @@ -4346,7 +4442,7 @@ int update_dpdk_args(void){ } - sprintf(global_cores_str,"0x%x",lpsock->get_cores_mask()); + sprintf(global_cores_str,"0x%llx",(unsigned long long)lpsock->get_cores_mask()); /* set the DPDK options */ global_dpdk_args_num =7; @@ -4397,6 +4493,7 @@ int update_dpdk_args(void){ printf(" %s \n",global_dpdk_args[i]); } } + return (0); } @@ -4418,7 +4515,7 @@ int sim_load_list_of_cap_files(CParserOption * op){ lpt->set_vif(&erf_vif); if ( (op->preview.getVMode() >1) || op->preview.getFileWrite() ) { - lpt->generate_erf(op->out_file,op->preview); + lpt->start_generate_stateful(op->out_file,op->preview); } lpt->m_node_gen.DumpHist(stdout); @@ -4430,42 +4527,6 @@ int sim_load_list_of_cap_files(CParserOption * op){ } - - -static int -launch_stateless_trex() { - CPlatformSocketInfo *lpsock=&CGlobalInfo::m_socket; - CParserOption *lpop= &CGlobalInfo::m_options; - CPlatformYamlInfo *cg=&global_platform_cfg_info; - - TrexStatelessCfg cfg; - - TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_TCP, 5050); - TrexRpcServerConfig rpc_async_cfg(TrexRpcServerConfig::RPC_PROT_TCP, 5051); - - cfg.m_dp_core_count = lpop->preview.getCores(); - cfg.m_port_count = lpop->m_expected_portd; - cfg.m_rpc_req_resp_cfg = &rpc_req_resp_cfg; - cfg.m_rpc_async_cfg = &rpc_async_cfg; - cfg.m_rpc_server_verbose = true; - - TrexStateless::configure(cfg); - - printf("\nStarting T-Rex Stateless\n"); - printf("Starting RPC Server...\n\n"); - - rte_eal_mp_remote_launch(stateless_entry, NULL, CALL_MASTER); - - unsigned lcore_id; - RTE_LCORE_FOREACH_SLAVE(lcore_id) { - if (rte_eal_wait_lcore(lcore_id) < 0) - return -1; - } - return (0); -} - - - int main_test(int argc , char * argv[]){ utl_termio_init(); @@ -4525,78 +4586,62 @@ int main_test(int argc , char * argv[]){ return ( sim_load_list_of_cap_files(&CGlobalInfo::m_options) ); } - bool is_stateless = (CGlobalInfo::m_options.m_run_mode == CParserOption::RUN_MODE_INTERACTIVE); - - if ( !ports_cfg.Create(is_stateless) ){ + if ( !g_trex.Create() ){ exit(1); } - /* patch here */ - if (is_stateless) { - return launch_stateless_trex(); - } - - if (po->preview.get_is_rx_check_enable() && (po->m_rx_check_sampe< get_min_sample_rate()) ) { po->m_rx_check_sampe = get_min_sample_rate(); printf("Warning rx check sample rate should be lower than %d setting it to %d\n",get_min_sample_rate(),get_min_sample_rate()); } /* set dump mode */ - ports_cfg.m_io_modes.set_mode((CTrexGlobalIoMode::CliDumpMode)CGlobalInfo::m_options.m_io_mode); + g_trex.m_io_modes.set_mode((CTrexGlobalIoMode::CliDumpMode)CGlobalInfo::m_options.m_io_mode); if ( !CGlobalInfo::m_options.is_latency_disabled() && (CGlobalInfo::m_options.m_latency_prev>0) ){ uint32_t pkts = CGlobalInfo::m_options.m_latency_prev* CGlobalInfo::m_options.m_latency_rate; printf("Start prev latency check - hack for Keren for %d sec \n",CGlobalInfo::m_options.m_latency_prev); - ports_cfg.m_mg.start(pkts); + g_trex.m_mg.start(pkts); printf("Delay now you can call command \n"); delay(CGlobalInfo::m_options.m_latency_prev* 1000); printf("Finish wating \n"); - ports_cfg.m_mg.reset(); - ports_cfg.reset_counters(); + g_trex.m_mg.reset(); + g_trex.reset_counters(); } - ports_cfg.start_send_master(); + if ( get_is_stateless() ) { + g_trex.start_master_stateless(); - // TBD remove - //ports_cfg.test_latency(); - /* test seding */ - //while (1) { - //} + }else{ + g_trex.start_send_master(); + } /* TBD_FDIR */ #if 0 printf(" test_send \n"); - ports_cfg.test_send(); + g_trex.test_send(); while (1) { delay(10000); } #endif - - - - //ports_cfg.test_latency(); - //return (0); - - if ( CGlobalInfo::m_options.preview.getOnlyLatency() ){ rte_eal_mp_remote_launch(latency_one_lcore, NULL, CALL_MASTER); RTE_LCORE_FOREACH_SLAVE(lcore_id) { if (rte_eal_wait_lcore(lcore_id) < 0) return -1; } - ports_cfg.stop_master(); + g_trex.stop_master(); return (0); } if ( CGlobalInfo::m_options.preview.getSingleCore() ) { - ports_cfg.run_in_core(1); - ports_cfg.stop_master(); + g_trex.run_in_core(1); + g_trex.stop_master(); return (0); } @@ -4606,8 +4651,8 @@ int main_test(int argc , char * argv[]){ return -1; } - ports_cfg.stop_master(); - ports_cfg.Delete(); + g_trex.stop_master(); + g_trex.Delete(); utl_termio_reset(); return (0); @@ -4732,14 +4777,13 @@ int CTRexExtendedDriverBase1G::configure_rx_filter_rules(CPhyEthIF * _if){ /* enable all rules */ _if->pci_reg_write(E1000_WUFC, (mask<<16) | (1<<14) ); + + return (0); } void CTRexExtendedDriverBase1G::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){ - int i; - uint64_t t=0; - stats->ipackets += _if->pci_reg_read(E1000_GPRC) ; stats->ibytes += (_if->pci_reg_read(E1000_GORCL) ); @@ -4869,6 +4913,7 @@ int CTRexExtendedDriverBase10G::configure_rx_filter_rules(CPhyEthIF * _if){ rte_exit(EXIT_FAILURE, " ERROR rte_eth_dev_fdir_add_perfect_filter : %d\n",res); } } + return (0); } int CTRexExtendedDriverBase10G::configure_drop_queue(CPhyEthIF * _if){ @@ -4991,6 +5036,8 @@ int CTRexExtendedDriverBase40G::configure_rx_filter_rules(CPhyEthIF * _if){ add_rules(_if,RTE_ETH_FLOW_TYPE_UDPV6,ttl); add_rules(_if,RTE_ETH_FLOW_TYPE_TCPV6,ttl); } + + return (0); } @@ -5146,3 +5193,54 @@ struct rte_mbuf * rte_mbuf_convert_to_one_seg(struct rte_mbuf *m){ } +/*********************************************************** + * platfrom API object + * TODO: REMOVE THIS TO A SEPERATE FILE + * + **********************************************************/ +void +TrexDpdkPlatformApi::get_global_stats(TrexPlatformGlobalStats &stats) const { + CGlobalStats trex_stats; + g_trex.get_stats(trex_stats); + + stats.m_stats.m_cpu_util = trex_stats.m_cpu_util; + + stats.m_stats.m_tx_bps = trex_stats.m_tx_bps; + stats.m_stats.m_tx_pps = trex_stats.m_tx_pps; + stats.m_stats.m_total_tx_pkts = trex_stats.m_total_tx_pkts; + stats.m_stats.m_total_tx_bytes = trex_stats.m_total_tx_bytes; + + stats.m_stats.m_rx_bps = trex_stats.m_rx_bps; + stats.m_stats.m_rx_pps = /*trex_stats.m_rx_pps*/ 0; /* missing */ + stats.m_stats.m_total_rx_pkts = trex_stats.m_total_rx_pkts; + stats.m_stats.m_total_rx_bytes = trex_stats.m_total_rx_bytes; +} + +void +TrexDpdkPlatformApi::get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const { + +} + +uint8_t +TrexDpdkPlatformApi::get_dp_core_count() const { + return CGlobalInfo::m_options.preview.getCores(); +} + + +void +TrexDpdkPlatformApi::port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list) const { + + cores_id_list.clear(); + + /* iterate over all DP cores */ + for (uint8_t core_id = 0; core_id < g_trex.get_cores_tx(); core_id++) { + + /* iterate over all the directions*/ + for (uint8_t dir = 0 ; dir < CS_NUM; dir++) { + if (g_trex.m_cores_vif[core_id + 1]->get_ports()[dir].m_port->get_port_id() == port_id) { + cores_id_list.push_back(std::make_pair(core_id, dir)); + } + } + } +} + diff --git a/src/mock/trex_platform_api_mock.cpp b/src/mock/trex_platform_api_mock.cpp new file mode 100644 index 00000000..54f71e10 --- /dev/null +++ b/src/mock/trex_platform_api_mock.cpp @@ -0,0 +1,49 @@ +/* + 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 <internal_api/trex_platform_api.h> + +void +TrexMockPlatformApi::get_global_stats(TrexPlatformGlobalStats &stats) const { + + stats.m_stats.m_cpu_util = 0; + + stats.m_stats.m_tx_bps = 0; + stats.m_stats.m_tx_pps = 0; + stats.m_stats.m_total_tx_pkts = 0; + stats.m_stats.m_total_tx_bytes = 0; + + stats.m_stats.m_rx_bps = 0; + stats.m_stats.m_rx_pps = 0; + stats.m_stats.m_total_rx_pkts = 0; + stats.m_stats.m_total_rx_bytes = 0; +} + +void +TrexMockPlatformApi::get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const { + +} + +uint8_t +TrexMockPlatformApi::get_dp_core_count() const { + return (1); +} + diff --git a/src/mock/trex_rpc_server_mock.cpp b/src/mock/trex_rpc_server_mock.cpp index de43f92f..0bdf6cf1 100644 --- a/src/mock/trex_rpc_server_mock.cpp +++ b/src/mock/trex_rpc_server_mock.cpp @@ -21,12 +21,71 @@ limitations under the License. #include <trex_rpc_server_api.h> #include <trex_stateless.h> +#include <trex_stateless_dp_core.h> + +#include <msg_manager.h> #include <iostream> +#include <sstream> #include <unistd.h> +#include <string.h> +#include <zmq.h> +#include <bp_sim.h> using namespace std; +static TrexStateless *g_trex_stateless; +static uint16_t g_rpc_port; + +static bool +verify_tcp_port_is_free(uint16_t port) { + void *m_context = zmq_ctx_new(); + void *m_socket = zmq_socket (m_context, ZMQ_REP); + std::stringstream ss; + ss << "tcp://*:"; + ss << port; + + int rc = zmq_bind (m_socket, ss.str().c_str()); + + zmq_close(m_socket); + zmq_term(m_context); + + return (rc == 0); +} + +static uint16_t +find_free_tcp_port(uint16_t start_port = 5050) { + void *m_context = zmq_ctx_new(); + void *m_socket = zmq_socket (m_context, ZMQ_REP); + + uint16_t port = start_port; + while (true) { + std::stringstream ss; + ss << "tcp://*:"; + ss << port; + + int rc = zmq_bind (m_socket, ss.str().c_str()); + if (rc == 0) { + break; + } + + port++; + } + + zmq_close(m_socket); + zmq_term(m_context); + + return port; +} + +TrexStateless * get_stateless_obj() { + return g_trex_stateless; +} + +uint16_t gtest_get_mock_server_port() { + return g_rpc_port; +} + /** * on simulation this is not rebuild every version * (improved stub) @@ -42,44 +101,87 @@ extern "C" const char * get_build_time(void){ int gtest_main(int argc, char **argv); -int main(int argc, char *argv[]) { +static bool parse_uint16(const string arg, uint16_t &port) { + stringstream ss(arg); + + bool x = (ss >> port); + + return (x); +} + +static void +run_dummy_core() { + //TODO: connect this to the scheduler + + //CFlowGenList fl; + //fl.Create(); + //CFlowGenListPerThread *lp = new CFlowGenListPerThread(); + //lp->Create(0, 0, NULL, 0); + //TrexStatelessDpCore dummy_core(0, lp); + //lp->start_stateless_daemon(); +} +int main(int argc, char *argv[]) { bool is_gtest = false; + time_init(); + CGlobalInfo::m_socket.Create(0); + + CGlobalInfo::init_pools(1000); + assert( CMsgIns::Ins()->Create(1)); + + std::thread *m_thread = new std::thread(run_dummy_core); + (void)m_thread; + // gtest ? if (argc > 1) { - if (string(argv[1]) != "--ut") { - cout << "\n[Usage] " << argv[0] << ": " << " [--ut]\n\n"; + string arg = string(argv[1]); + + if (arg == "--ut") { + g_rpc_port = find_free_tcp_port(); + is_gtest = true; + } else if (parse_uint16(arg, g_rpc_port)) { + bool rc = verify_tcp_port_is_free(g_rpc_port); + if (!rc) { + cout << "port " << g_rpc_port << " is not available to use\n"; + exit(-1); + } + } else { + + cout << "\n[Usage] " << argv[0] << ": " << " [--ut] or [port number < 65535]\n\n"; exit(-1); } - is_gtest = true; + + } else { + g_rpc_port = find_free_tcp_port(); } /* configure the stateless object with 4 ports */ TrexStatelessCfg cfg; - TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_TCP, 5050); - TrexRpcServerConfig rpc_async_cfg(TrexRpcServerConfig::RPC_PROT_TCP, 5051); + TrexRpcServerConfig rpc_req_resp_cfg(TrexRpcServerConfig::RPC_PROT_TCP, g_rpc_port); + //TrexRpcServerConfig rpc_async_cfg(TrexRpcServerConfig::RPC_PROT_TCP, 5051); cfg.m_port_count = 4; - cfg.m_dp_core_count = 2; cfg.m_rpc_req_resp_cfg = &rpc_req_resp_cfg; - cfg.m_rpc_async_cfg = &rpc_async_cfg; + cfg.m_rpc_async_cfg = NULL; cfg.m_rpc_server_verbose = (is_gtest ? false : true); + cfg.m_platform_api = new TrexMockPlatformApi(); - TrexStateless::configure(cfg); + g_trex_stateless = new TrexStateless(cfg); - TrexStateless::get_instance().launch_control_plane(); + g_trex_stateless->launch_control_plane(); /* gtest handling */ if (is_gtest) { int rc = gtest_main(argc, argv); - TrexStateless::destroy(); + delete g_trex_stateless; + g_trex_stateless = NULL; return rc; } cout << "\n-= Starting RPC Server Mock =-\n\n"; - cout << "Listening on tcp://localhost:5050 [ZMQ]\n\n"; + cout << "Listening on tcp://localhost:" << g_rpc_port << " [ZMQ]\n\n"; cout << "Server Started\n\n"; @@ -87,6 +189,7 @@ int main(int argc, char *argv[]) { sleep(1); } - TrexStateless::destroy(); + delete g_trex_stateless; + g_trex_stateless = NULL; } diff --git a/src/msg_manager.cpp b/src/msg_manager.cpp index 4db96583..9f41d08c 100755 --- a/src/msg_manager.cpp +++ b/src/msg_manager.cpp @@ -26,7 +26,7 @@ limitations under the License. /*TBD: need to fix socket_id for NUMA */ -bool CMessagingManager::Create(uint8_t num_dp_threads){ +bool CMessagingManager::Create(uint8_t num_dp_threads,std::string a_name){ m_num_dp_threads=num_dp_threads; assert(m_dp_to_cp==0); assert(m_cp_to_dp==0); @@ -38,11 +38,11 @@ bool CMessagingManager::Create(uint8_t num_dp_threads){ char name[100]; lp=getRingCpToDp(i); - sprintf(name,"cp_to_dp_%d",i); + sprintf(name,"%s_to_%d",(char *)a_name.c_str(),i); assert(lp->Create(std::string(name),1024,0)==true); lp=getRingDpToCp(i); - sprintf(name,"dp_to_cp_%d",i); + sprintf(name,"%s_from_%d",(char *)a_name.c_str(),i); assert(lp->Create(std::string(name),1024,0)==true); } @@ -89,7 +89,12 @@ CMsgIns * CMsgIns::Ins(void){ } bool CMsgIns::Create(uint8_t num_threads){ - return ( m_rx_dp.Create(num_threads) ); + + bool res = m_cp_dp.Create(num_threads,"cp_dp"); + if (!res) { + return (res); + } + return (m_rx_dp.Create(num_threads,"rx_dp")); } diff --git a/src/msg_manager.h b/src/msg_manager.h index b25660bb..8958f826 100755 --- a/src/msg_manager.h +++ b/src/msg_manager.h @@ -23,12 +23,20 @@ limitations under the License. #include "CRing.h" +#include <string> /* messages from CP->DP Ids */ -#define NAT_MSG (7) -#define LATENCY_PKT_SEND_MSG (8) +struct CGenNodeMsgBase { + enum { + NAT_FIRST = 7, + LATENCY_PKT = 8, + } msg_types; + +public: + uint8_t m_msg_type; /* msg type */ +}; /* @@ -71,7 +79,7 @@ public: m_dp_to_cp=0; m_num_dp_threads=0; } - bool Create(uint8_t num_dp_threads); + bool Create(uint8_t num_dp_threads,std::string name); void Delete(); CNodeRing * getRingCpToDp(uint8_t thread_id); CNodeRing * getRingDpToCp(uint8_t thread_id); @@ -94,12 +102,18 @@ public: CMessagingManager * getRxDp(){ return (&m_rx_dp); } + CMessagingManager * getCpDp(){ + return (&m_cp_dp); + } + uint8_t get_num_threads(){ return (m_rx_dp.get_num_threads()); } private: CMessagingManager m_rx_dp; + CMessagingManager m_cp_dp; + private: /* one instance */ diff --git a/src/nat_check.cpp b/src/nat_check.cpp index 676c1292..170d2de6 100755 --- a/src/nat_check.cpp +++ b/src/nat_check.cpp @@ -171,8 +171,8 @@ void CNatRxManager::handle_packet_ipv4(CNatOption * option, } -#define MYDP(f) if (f) fprintf(fd," %-40s: %llu \n",#f,f) -#define MYDP_A(f) fprintf(fd," %-40s: %llu \n",#f,f) +#define MYDP(f) if (f) fprintf(fd," %-40s: %llu \n",#f,(unsigned long long)f) +#define MYDP_A(f) fprintf(fd," %-40s: %llu \n",#f, (unsigned long long)f) diff --git a/src/nat_check.h b/src/nat_check.h index b67c523c..a500ddaf 100755 --- a/src/nat_check.h +++ b/src/nat_check.h @@ -59,16 +59,6 @@ struct CNatFlowInfo { this struct should be in the same size of CGenNode beacuse allocator is global . */ -struct CGenNodeMsgBase { - enum { - NAT_FIRST = NAT_MSG, - LATENCY_PKT = LATENCY_PKT_SEND_MSG - } msg_types; - -public: - uint8_t m_msg_type; /* msg type */ -}; - struct CGenNodeNatInfo : public CGenNodeMsgBase { uint8_t m_pad; diff --git a/src/os_time.h b/src/os_time.h index 153ee3e3..0e732abf 100755 --- a/src/os_time.h +++ b/src/os_time.h @@ -22,6 +22,7 @@ limitations under the License. */ #include <stdint.h> +#include <time.h> typedef uint64_t hr_time_t; // time in high res tick typedef uint32_t hr_time_32_t; // time in high res tick @@ -129,6 +130,25 @@ static inline dsec_t now_sec(void){ } +static inline +void delay(int msec){ + + if (msec == 0) + {//user that requested that probebly wanted the minimal delay + //but because of scaling problem he have got 0 so we will give the min delay + //printf("\n\n\nERROR-Task delay ticks == 0 found in task %s task id = %d\n\n\n\n", + // SANB_TaskName(SANB_TaskIdSelf()), SANB_TaskIdSelf()); + msec =1; + + } + + struct timespec time1, remain; // 2 sec max delay + time1.tv_sec=msec/1000; + time1.tv_nsec=(msec - (time1.tv_sec*1000))*1000000; + + nanosleep(&time1,&remain); +} + #endif diff --git a/src/pal/linux/mbuf.h b/src/pal/linux/mbuf.h index 693b095a..35a442bf 100755 --- a/src/pal/linux/mbuf.h +++ b/src/pal/linux/mbuf.h @@ -185,8 +185,9 @@ static inline void utl_rte_pktmbuf_add_last(rte_mbuf_t *m,rte_mbuf_t *m_last){ #define __rte_cache_aligned -#define CACHE_LINE_SIZE 64 +#define CACHE_LINE_SIZE 64 +#define RTE_CACHE_LINE_SIZE 64 #define SOCKET_ID_ANY 0 #endif diff --git a/src/platform_cfg.cpp b/src/platform_cfg.cpp index f0911611..df04cd89 100755 --- a/src/platform_cfg.cpp +++ b/src/platform_cfg.cpp @@ -127,7 +127,7 @@ void CPlatformMemoryYamlInfo::Dump(FILE *fd){ int i=0; for (i=0; i<MBUF_SIZE; i++) { - fprintf(fd," %-40s : %lu \n",names[i].c_str(),m_mbuf[i]); + fprintf(fd," %-40s : %lu \n",names[i].c_str(), (ulong)m_mbuf[i]); } } @@ -300,6 +300,10 @@ void operator >> (const YAML::Node& node, CPlatformYamlInfo & plat_info) { plat_info.m_telnet_exist=true; } + if ( node.FindValue("zmq_rpc_port") ){ + node["zmq_rpc_port"] >> plat_info.m_zmq_rpc_port; + } + if ( node.FindValue("port_bandwidth_gb") ){ node["port_bandwidth_gb"] >> plat_info.m_port_bandwidth_gb; } @@ -375,7 +379,7 @@ void CPlatformYamlInfo::Dump(FILE *fd){ }else{ fprintf(fd," port limit : not configured \n"); } - fprintf(fd," port_bandwidth_gb : %lu \n",m_port_bandwidth_gb); + fprintf(fd," port_bandwidth_gb : %lu \n", (ulong)m_port_bandwidth_gb); if ( m_if_mask_exist && m_if_mask.size() ) { fprintf(fd," if_mask : "); @@ -383,7 +387,7 @@ void CPlatformYamlInfo::Dump(FILE *fd){ for (i=0; i<(int)m_if_mask.size(); i++) { fprintf(fd," %s,",m_if_mask[i].c_str()); } - fprintf(fd,"\n",m_if_mask[i].c_str()); + fprintf(fd,"\n"); }else{ fprintf(fd," if_mask : None \n"); diff --git a/src/platform_cfg.h b/src/platform_cfg.h index 2f335471..b4b03b10 100755 --- a/src/platform_cfg.h +++ b/src/platform_cfg.h @@ -184,6 +184,7 @@ public: m_telnet_exist=false; m_telnet_port=4502 ; + m_zmq_rpc_port = 5050; m_mac_info_exist=false; m_port_bandwidth_gb = 10; @@ -209,15 +210,17 @@ public: std::string m_limit_memory; uint32_t m_thread_per_dual_if; - uint32_t m_port_bandwidth_gb; + uint32_t m_port_bandwidth_gb; - bool m_enable_zmq_pub_exist; - bool m_enable_zmq_pub; - uint16_t m_zmq_pub_port; + bool m_enable_zmq_pub_exist; + bool m_enable_zmq_pub; + uint16_t m_zmq_pub_port; - bool m_telnet_exist; - uint16_t m_telnet_port; + bool m_telnet_exist; + uint16_t m_telnet_port; + + uint16_t m_zmq_rpc_port; bool m_mac_info_exist; std::vector <CMacYamlInfo> m_mac_info; diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp index ae87d749..b40e996f 100644 --- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp +++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp @@ -145,7 +145,7 @@ trex_rpc_cmd_rc_e TrexRpcCmdGetSysInfo::_run(const Json::Value ¶ms, Json::Value &result) { string hostname; - TrexStateless & instance = TrexStateless::get_instance(); + TrexStateless * main = get_stateless_obj(); Json::Value §ion = result["result"]; @@ -155,21 +155,21 @@ TrexRpcCmdGetSysInfo::_run(const Json::Value ¶ms, Json::Value &result) { section["uptime"] = TrexRpcServer::get_server_uptime(); /* FIXME: core count */ - section["dp_core_count"] = instance.get_dp_core_count(); + section["dp_core_count"] = main->get_dp_core_count(); section["core_type"] = get_cpu_model(); /* ports */ - section["port_count"] = instance.get_port_count(); + section["port_count"] = main->get_port_count(); section["ports"] = Json::arrayValue; - for (int i = 0; i < instance.get_port_count(); i++) { + for (int i = 0; i < main->get_port_count(); i++) { string driver; string speed; - TrexStatelessPort *port = instance.get_port_by_id(i); + TrexStatelessPort *port = main->get_port_by_id(i); port->get_properties(driver, speed); section["ports"][i]["index"] = i; @@ -201,7 +201,7 @@ TrexRpcCmdGetOwner::_run(const Json::Value ¶ms, Json::Value &result) { uint8_t port_id = parse_port(params, result); - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); section["owner"] = port->get_owner(); return (TREX_RPC_CMD_OK); @@ -220,7 +220,7 @@ TrexRpcCmdAcquire::_run(const Json::Value ¶ms, Json::Value &result) { bool force = parse_bool(params, "force", result); /* if not free and not you and not force - fail */ - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); if ( (!port->is_free_to_aquire()) && (port->get_owner() != new_owner) && (!force)) { generate_execute_err(result, "port is already taken by '" + port->get_owner() + "'"); @@ -242,7 +242,7 @@ TrexRpcCmdRelease::_run(const Json::Value ¶ms, Json::Value &result) { uint8_t port_id = parse_port(params, result); - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); if (port->get_state() == TrexStatelessPort::PORT_STATE_TRANSMITTING) { generate_execute_err(result, "cannot release a port during transmission"); @@ -264,7 +264,7 @@ TrexRpcCmdGetPortStats::_run(const Json::Value ¶ms, Json::Value &result) { uint8_t port_id = parse_port(params, result); - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); if (port->get_state() == TrexStatelessPort::PORT_STATE_DOWN) { generate_execute_err(result, "cannot get stats - port is down"); @@ -277,3 +277,42 @@ TrexRpcCmdGetPortStats::_run(const Json::Value ¶ms, Json::Value &result) { return (TREX_RPC_CMD_OK); } +/** + * request the server a sync about a specific user + * + */ +trex_rpc_cmd_rc_e +TrexRpcCmdSyncUser::_run(const Json::Value ¶ms, Json::Value &result) { + + const string &user = parse_string(params, "user", result); + bool sync_streams = parse_bool(params, "sync_streams", result); + + result["result"] = Json::arrayValue; + + for (auto port : get_stateless_obj()->get_port_list()) { + if (port->get_owner() == user) { + + Json::Value owned_port; + + owned_port["port_id"] = port->get_port_id(); + owned_port["handler"] = port->get_owner_handler(); + owned_port["state"] = port->get_state_as_string(); + + /* if sync streams was asked - sync all the streams */ + if (sync_streams) { + owned_port["streams"] = Json::arrayValue; + + std::vector <TrexStream *> streams; + port->get_stream_table()->get_object_list(streams); + + for (auto stream : streams) { + owned_port["streams"].append(stream->get_stream_json()); + } + } + + result["result"].append(owned_port); + } + } + + return (TREX_RPC_CMD_OK); +} diff --git a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp index 20107411..4fa0956d 100644 --- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp +++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp @@ -114,7 +114,7 @@ TrexRpcCmdAddStream::_run(const Json::Value ¶ms, Json::Value &result) { /* make sure this is a valid stream to add */ validate_stream(stream, result); - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(stream->m_port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(stream->m_port_id); port->get_stream_table()->add_stream(stream); result["result"] = "ACK"; @@ -127,7 +127,7 @@ TrexRpcCmdAddStream::_run(const Json::Value ¶ms, Json::Value &result) { TrexStream * TrexRpcCmdAddStream::allocate_new_stream(const Json::Value §ion, uint8_t port_id, uint32_t stream_id, Json::Value &result) { - TrexStream *stream; + TrexStream *stream = NULL; const Json::Value &mode = parse_object(section, "mode", result); std::string type = parse_string(mode, "type", result); @@ -200,9 +200,9 @@ TrexRpcCmdAddStream::parse_vm_instr_flow_var(const Json::Value &inst, TrexStream std::string min_value_str = parse_string(inst, "min_value", result); std::string max_value_str = parse_string(inst, "max_value", result); - uint64_t init_value; - uint64_t min_value; - uint64_t max_value; + uint64_t init_value = 0; + uint64_t min_value = 0; + uint64_t max_value = 0; try { init_value = str2num(init_value_str); @@ -282,15 +282,15 @@ TrexRpcCmdAddStream::validate_stream(const TrexStream *stream, Json::Value &resu } /* port id should be between 0 and count - 1 */ - if (stream->m_port_id >= TrexStateless::get_instance().get_port_count()) { + if (stream->m_port_id >= get_stateless_obj()->get_port_count()) { std::stringstream ss; - ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; delete stream; generate_execute_err(result, ss.str()); } /* add the stream to the port's stream table */ - TrexStatelessPort * port = TrexStateless::get_instance().get_port_by_id(stream->m_port_id); + TrexStatelessPort * port = get_stateless_obj()->get_port_by_id(stream->m_port_id); /* does such a stream exists ? */ if (port->get_stream_table()->get_stream_by_id(stream->m_stream_id)) { @@ -312,13 +312,13 @@ TrexRpcCmdRemoveStream::_run(const Json::Value ¶ms, Json::Value &result) { uint32_t stream_id = parse_int(params, "stream_id", result); - if (port_id >= TrexStateless::get_instance().get_port_count()) { + if (port_id >= get_stateless_obj()->get_port_count()) { std::stringstream ss; - ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; generate_execute_err(result, ss.str()); } - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); TrexStream *stream = port->get_stream_table()->get_stream_by_id(stream_id); if (!stream) { @@ -344,13 +344,13 @@ trex_rpc_cmd_rc_e TrexRpcCmdRemoveAllStreams::_run(const Json::Value ¶ms, Json::Value &result) { uint8_t port_id = parse_byte(params, "port_id", result); - if (port_id >= TrexStateless::get_instance().get_port_count()) { + if (port_id >= get_stateless_obj()->get_port_count()) { std::stringstream ss; - ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; generate_execute_err(result, ss.str()); } - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); port->get_stream_table()->remove_and_delete_all_streams(); result["result"] = "ACK"; @@ -369,15 +369,15 @@ TrexRpcCmdGetStreamList::_run(const Json::Value ¶ms, Json::Value &result) { uint8_t port_id = parse_byte(params, "port_id", result); - if (port_id >= TrexStateless::get_instance().get_port_count()) { + if (port_id >= get_stateless_obj()->get_port_count()) { std::stringstream ss; - ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; generate_execute_err(result, ss.str()); } - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); - port->get_stream_table()->get_stream_list(stream_list); + port->get_stream_table()->get_id_list(stream_list); Json::Value json_list = Json::arrayValue; @@ -397,17 +397,17 @@ TrexRpcCmdGetStreamList::_run(const Json::Value ¶ms, Json::Value &result) { **************************/ trex_rpc_cmd_rc_e TrexRpcCmdGetStream::_run(const Json::Value ¶ms, Json::Value &result) { - uint8_t port_id = parse_byte(params, "port_id", result); - + uint8_t port_id = parse_byte(params, "port_id", result); + bool get_pkt = parse_bool(params, "get_pkt", result); uint32_t stream_id = parse_int(params, "stream_id", result); - if (port_id >= TrexStateless::get_instance().get_port_count()) { + if (port_id >= get_stateless_obj()->get_port_count()) { std::stringstream ss; - ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; generate_execute_err(result, ss.str()); } - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); TrexStream *stream = port->get_stream_table()->get_stream_by_id(stream_id); @@ -418,7 +418,12 @@ TrexRpcCmdGetStream::_run(const Json::Value ¶ms, Json::Value &result) { } /* return the stored stream json (instead of decoding it all over again) */ - result["result"]["stream"] = stream->get_stream_json(); + Json::Value j = stream->get_stream_json(); + if (!get_pkt) { + j.removeMember("packet"); + } + + result["result"]["stream"] = j; return (TREX_RPC_CMD_OK); @@ -432,16 +437,17 @@ trex_rpc_cmd_rc_e TrexRpcCmdStartTraffic::_run(const Json::Value ¶ms, Json::Value &result) { uint8_t port_id = parse_byte(params, "port_id", result); + double mul = parse_double(params, "mul", result); - if (port_id >= TrexStateless::get_instance().get_port_count()) { + if (port_id >= get_stateless_obj()->get_port_count()) { std::stringstream ss; - ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; generate_execute_err(result, ss.str()); } - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); - TrexStatelessPort::rc_e rc = port->start_traffic(); + TrexStatelessPort::rc_e rc = port->start_traffic(mul); if (rc == TrexStatelessPort::RC_OK) { result["result"] = "ACK"; @@ -462,7 +468,7 @@ TrexRpcCmdStartTraffic::_run(const Json::Value ¶ms, Json::Value &result) { generate_execute_err(result, ss.str()); } - return (TREX_RPC_CMD_OK); + return (TREX_RPC_CMD_OK); } /*************************** @@ -473,13 +479,13 @@ trex_rpc_cmd_rc_e TrexRpcCmdStopTraffic::_run(const Json::Value ¶ms, Json::Value &result) { uint8_t port_id = parse_byte(params, "port_id", result); - if (port_id >= TrexStateless::get_instance().get_port_count()) { + if (port_id >= get_stateless_obj()->get_port_count()) { std::stringstream ss; - ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; generate_execute_err(result, ss.str()); } - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); port->stop_traffic(); result["result"] = "ACK"; @@ -487,3 +493,43 @@ TrexRpcCmdStopTraffic::_run(const Json::Value ¶ms, Json::Value &result) { return (TREX_RPC_CMD_OK); } +/*************************** + * get all streams + * + **************************/ +trex_rpc_cmd_rc_e +TrexRpcCmdGetAllStreams::_run(const Json::Value ¶ms, Json::Value &result) { + uint8_t port_id = parse_byte(params, "port_id", result); + bool get_pkt = parse_bool(params, "get_pkt", result); + + if (port_id >= get_stateless_obj()->get_port_count()) { + std::stringstream ss; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; + generate_execute_err(result, ss.str()); + } + + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); + + std::vector <TrexStream *> streams; + port->get_stream_table()->get_object_list(streams); + + Json::Value streams_json = Json::objectValue; + for (auto stream : streams) { + + Json::Value j = stream->get_stream_json(); + + /* should we include the packet as well ? */ + if (!get_pkt) { + j.removeMember("packet"); + } + + std::stringstream ss; + ss << stream->m_stream_id; + + streams_json[ss.str()] = j; + } + + result["result"]["streams"] = streams_json; + + return (TREX_RPC_CMD_OK); +} diff --git a/src/rpc-server/commands/trex_rpc_cmds.h b/src/rpc-server/commands/trex_rpc_cmds.h index 5926a8d8..a604d9a1 100644 --- a/src/rpc-server/commands/trex_rpc_cmds.h +++ b/src/rpc-server/commands/trex_rpc_cmds.h @@ -99,11 +99,16 @@ void parse_vm_instr_write_flow_var(const Json::Value &inst, TrexStream *stream, TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStreamList, "get_stream_list", 1, true); +TREX_RPC_CMD_DEFINE(TrexRpcCmdGetAllStreams, "get_all_streams", 2, true); -TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStream, "get_stream", 2, true); +TREX_RPC_CMD_DEFINE(TrexRpcCmdGetStream, "get_stream", 3, true); -TREX_RPC_CMD_DEFINE(TrexRpcCmdStartTraffic, "start_traffic", 1, true); + + +TREX_RPC_CMD_DEFINE(TrexRpcCmdStartTraffic, "start_traffic", 2, true); TREX_RPC_CMD_DEFINE(TrexRpcCmdStopTraffic, "stop_traffic", 1, true); +TREX_RPC_CMD_DEFINE(TrexRpcCmdSyncUser, "sync_user", 2, false); + #endif /* __TREX_RPC_CMD_H__ */ diff --git a/src/rpc-server/trex_rpc_async_server.cpp b/src/rpc-server/trex_rpc_async_server.cpp index f4d21f2f..46fe499b 100644 --- a/src/rpc-server/trex_rpc_async_server.cpp +++ b/src/rpc-server/trex_rpc_async_server.cpp @@ -79,7 +79,7 @@ TrexRpcServerAsync::_rpc_thread_cb() { } /* trigger a full update for stats */ - TrexStateless::get_instance().update_stats(); + //get_stateless_obj()->update_stats(); /* done with the lock */ if (m_lock) { @@ -87,7 +87,7 @@ TrexRpcServerAsync::_rpc_thread_cb() { } /* encode them to JSON */ - TrexStateless::get_instance().encode_stats(snapshot); + get_stateless_obj()->encode_stats(snapshot); /* write to string and publish */ std::string snapshot_str = writer.write(snapshot); diff --git a/src/rpc-server/trex_rpc_cmd.cpp b/src/rpc-server/trex_rpc_cmd.cpp index 920a8d30..af0db3f4 100644 --- a/src/rpc-server/trex_rpc_cmd.cpp +++ b/src/rpc-server/trex_rpc_cmd.cpp @@ -61,7 +61,7 @@ TrexRpcCommand::verify_ownership(const Json::Value ¶ms, Json::Value &result) std::string handler = parse_string(params, "handler", result); uint8_t port_id = parse_port(params, result); - TrexStatelessPort *port = TrexStateless::get_instance().get_port_by_id(port_id); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); if (!port->verify_owner_handler(handler)) { generate_execute_err(result, "invalid handler provided. please pass the handler given when calling 'acquire' or take ownership"); @@ -78,9 +78,9 @@ TrexRpcCommand::parse_port(const Json::Value ¶ms, Json::Value &result) { void TrexRpcCommand::validate_port_id(uint8_t port_id, Json::Value &result) { - if (port_id >= TrexStateless::get_instance().get_port_count()) { + if (port_id >= get_stateless_obj()->get_port_count()) { std::stringstream ss; - ss << "invalid port id - should be between 0 and " << (int)TrexStateless::get_instance().get_port_count() - 1; + ss << "invalid port id - should be between 0 and " << (int)get_stateless_obj()->get_port_count() - 1; generate_execute_err(result, ss.str()); } } diff --git a/src/rpc-server/trex_rpc_cmds_table.cpp b/src/rpc-server/trex_rpc_cmds_table.cpp index c1c546f3..e3bd7848 100644 --- a/src/rpc-server/trex_rpc_cmds_table.cpp +++ b/src/rpc-server/trex_rpc_cmds_table.cpp @@ -42,12 +42,15 @@ TrexRpcCommandsTable::TrexRpcCommandsTable() { register_command(new TrexRpcCmdRelease()); register_command(new TrexRpcCmdGetPortStats()); + register_command(new TrexRpcCmdSyncUser()); + /* stream commands */ register_command(new TrexRpcCmdAddStream()); register_command(new TrexRpcCmdRemoveStream()); register_command(new TrexRpcCmdRemoveAllStreams()); register_command(new TrexRpcCmdGetStreamList()); register_command(new TrexRpcCmdGetStream()); + register_command(new TrexRpcCmdGetAllStreams()); register_command(new TrexRpcCmdStartTraffic()); register_command(new TrexRpcCmdStopTraffic()); } diff --git a/src/rx_check.cpp b/src/rx_check.cpp index 3a67ca23..59b42e1a 100755 --- a/src/rx_check.cpp +++ b/src/rx_check.cpp @@ -45,8 +45,8 @@ void CRxCheckFlowTableStats::Clear(){ } -#define MYDP(f) if (f) fprintf(fd," %-40s: %llu \n",#f,f) -#define MYDP_A(f) fprintf(fd," %-40s: %llu \n",#f,f) +#define MYDP(f) if (f) fprintf(fd," %-40s: %llu \n",#f,(unsigned long long)f) +#define MYDP_A(f) fprintf(fd," %-40s: %llu \n",#f,(unsigned long long)f) #define MYDP_J(f) json+=add_json(#f,f); #define MYDP_J_LAST(f) json+=add_json(#f,f,true); @@ -146,7 +146,7 @@ void CRxCheckFlowTableMap::dump_all(FILE *fd){ rx_check_flow_map_iter_t it; for (it= m_map.begin(); it != m_map.end(); ++it) { CRxCheckFlow *lp = it->second; - printf ("flow_id: %d \n",lp->m_flow_id); + printf ("flow_id: %llu \n",(unsigned long long)lp->m_flow_id); } } @@ -208,7 +208,7 @@ std::string CPerTxthreadTemplateInfo::dump_as_json(std::string name){ int i; for (i=0;i<MAX_TEMPLATES_STATS;i++){ char buff[200]; - sprintf(buff,"%llu",m_template_info[i]); + sprintf(buff,"%llu", (unsigned long long)m_template_info[i]); json+=std::string(buff); if ( i < MAX_TEMPLATES_STATS-1) { json+=std::string(","); @@ -231,7 +231,7 @@ void CPerTxthreadTemplateInfo::Dump(FILE *fd){ int i; for (i=0; i<MAX_TEMPLATES_STATS; i++) { if (m_template_info[i]) { - fprintf (fd," template id: %llu %llu \n",i,m_template_info[i]); + fprintf (fd," template id: %d %llu \n",i, (unsigned long long)m_template_info[i]); } } } @@ -484,7 +484,7 @@ void RxCheckManager::DumpTemplate(FILE *fd,bool verbose){ if (cnt==0){ fprintf(fd,"\n"); } - fprintf(fd,"[id:%2d val:%8d,rx:%8d], ",i,lp->get_error_counter(),lp->get_rx_counter()); + fprintf(fd,"[id:%2d val:%8llu,rx:%8llu], ",i, (unsigned long long)lp->get_error_counter(), (unsigned long long)lp->get_rx_counter()); cnt++; if (cnt>5) { cnt=0; @@ -500,7 +500,11 @@ void RxCheckManager::DumpTemplateFull(FILE *fd){ int i; for (i=0; i<MAX_TEMPLATES_STATS;i++ ) { CPerTemplateInfo * lp=get_template(i); - fprintf(fd," template_id_%2d , errors:%8d, jitter: %lu rx : %lu \n",i,lp->get_error_counter(),lp->get_jitter_usec(),lp->get_rx_counter() ); + fprintf(fd," template_id_%2d , errors:%8llu, jitter: %llu rx : %llu \n", + i, + (unsigned long long)lp->get_error_counter(), + (unsigned long long)lp->get_jitter_usec(), + (unsigned long long)lp->get_rx_counter() ); } } @@ -514,7 +518,11 @@ void RxCheckManager::DumpShort(FILE *fd){ DumpTemplate(fd,false); fprintf(fd,"\n"); fprintf(fd,"---\n"); - fprintf(fd," active flows: %8d, fif: %8d, drop: %8d, errors: %8d \n",m_stats.m_active,m_stats.m_fif,m_stats.m_err_drop,m_stats.get_total_err()); + fprintf(fd," active flows: %8llu, fif: %8llu, drop: %8llu, errors: %8llu \n", + (unsigned long long)m_stats.m_active, + (unsigned long long)m_stats.m_fif, + (unsigned long long)m_stats.m_err_drop, + (unsigned long long)m_stats.get_total_err()); fprintf(fd,"------------------------------------------------------------------------------------------------------------\n"); } diff --git a/src/rx_check.h b/src/rx_check.h index 6f9763a2..07f5684c 100755 --- a/src/rx_check.h +++ b/src/rx_check.h @@ -30,9 +30,10 @@ limitations under the License. typedef enum { - CLIENT_SIDE=0, - SERVER_SIDE=1, - CS_NUM=2 + CLIENT_SIDE = 0, + SERVER_SIDE = 1, + CS_NUM = 2, + CS_INVALID = 255 } pkt_dir_enum_t; typedef uint8_t pkt_dir_t ; diff --git a/src/rx_check_header.cpp b/src/rx_check_header.cpp index 8ee580db..5934ee15 100755 --- a/src/rx_check_header.cpp +++ b/src/rx_check_header.cpp @@ -42,11 +42,11 @@ void CRx_check_header::dump(FILE *fd){ void CNatOption::dump(FILE *fd){ - fprintf(fd," op : %lx \n",get_option_type()); - fprintf(fd," ol : %lx \n",get_option_len()); - fprintf(fd," thread_id : %lx \n",get_thread_id()); - fprintf(fd," magic : %lx \n",get_magic()); - fprintf(fd," fid : %lx \n",get_fid()); + fprintf(fd," op : %x \n",get_option_type()); + fprintf(fd," ol : %x \n",get_option_len()); + fprintf(fd," thread_id : %x \n",get_thread_id()); + fprintf(fd," magic : %x \n",get_magic()); + fprintf(fd," fid : %x \n",get_fid()); utl_DumpBuffer(stdout,(void *)&u.m_data[0],8,0); } diff --git a/src/stateless/cp/trex_stateless.cpp b/src/stateless/cp/trex_stateless.cpp index 72762e26..e0e95450 100644 --- a/src/stateless/cp/trex_stateless.cpp +++ b/src/stateless/cp/trex_stateless.cpp @@ -31,55 +31,58 @@ using namespace std; * Trex stateless object * **********************************************************/ -TrexStateless::TrexStateless() { - m_is_configured = false; -} - /** - * configure the singleton stateless object * */ -void TrexStateless::configure(const TrexStatelessCfg &cfg) { - - TrexStateless& instance = get_instance_internal(); - - /* check status */ - if (instance.m_is_configured) { - throw TrexException("re-configuration of stateless object is not allowed"); - } +TrexStateless::TrexStateless(const TrexStatelessCfg &cfg) { /* create RPC servers */ /* set both servers to mutex each other */ - instance.m_rpc_server = new TrexRpcServer(cfg.m_rpc_req_resp_cfg, cfg.m_rpc_async_cfg, &instance.m_global_cp_lock); - instance.m_rpc_server->set_verbose(cfg.m_rpc_server_verbose); + m_rpc_server = new TrexRpcServer(cfg.m_rpc_req_resp_cfg, cfg.m_rpc_async_cfg, &m_global_cp_lock); + m_rpc_server->set_verbose(cfg.m_rpc_server_verbose); /* configure ports */ + m_port_count = cfg.m_port_count; - instance.m_port_count = cfg.m_port_count; - - for (int i = 0; i < instance.m_port_count; i++) { - instance.m_ports.push_back(new TrexStatelessPort(i)); + for (int i = 0; i < m_port_count; i++) { + m_ports.push_back(new TrexStatelessPort(i)); } - /* cores */ - instance.m_dp_core_count = cfg.m_dp_core_count; - for (int i = 0; i < instance.m_dp_core_count; i++) { - instance.m_dp_cores.push_back(new TrexStatelessDpCore(i)); + m_platform_api = cfg.m_platform_api; +} + +/** + * release all memory + * + * @author imarom (08-Oct-15) + */ +TrexStateless::~TrexStateless() { + + /* release memory for ports */ + for (auto port : m_ports) { + delete port; } + m_ports.clear(); + + /* stops the RPC server */ + m_rpc_server->stop(); + delete m_rpc_server; + + m_rpc_server = NULL; - /* done */ - instance.m_is_configured = true; + delete m_platform_api; + m_platform_api = NULL; } + /** * starts the control plane side * */ void TrexStateless::launch_control_plane() { - //std::cout << "\n on control/master core \n"; /* pin this process to the current running CPU any new thread will be called on the same CPU @@ -94,39 +97,6 @@ TrexStateless::launch_control_plane() { m_rpc_server->start(); } -void -TrexStateless::launch_on_dp_core(uint8_t core_id) { - m_dp_cores[core_id - 1]->run(); -} - -/** - * destroy the singleton and release all memory - * - * @author imarom (08-Oct-15) - */ -void -TrexStateless::destroy() { - TrexStateless& instance = get_instance_internal(); - - if (!instance.m_is_configured) { - return; - } - - /* release memory for ports */ - for (auto port : instance.m_ports) { - delete port; - } - instance.m_ports.clear(); - - /* stops the RPC server */ - instance.m_rpc_server->stop(); - delete instance.m_rpc_server; - - instance.m_rpc_server = NULL; - - /* done */ - instance.m_is_configured = false; -} /** * fetch a port by ID @@ -148,57 +118,32 @@ TrexStateless::get_port_count() { uint8_t TrexStateless::get_dp_core_count() { - return m_dp_core_count; -} - -void -TrexStateless::update_stats() { - - /* update CPU util. - TODO - */ - m_stats.m_stats.m_cpu_util = 0; - - /* for every port update and accumulate */ - for (uint8_t i = 0; i < m_port_count; i++) { - m_ports[i]->update_stats(); - - const TrexPortStats & port_stats = m_ports[i]->get_stats(); - - m_stats.m_stats.m_tx_bps += port_stats.m_stats.m_tx_bps; - m_stats.m_stats.m_rx_bps += port_stats.m_stats.m_rx_bps; - - m_stats.m_stats.m_tx_pps += port_stats.m_stats.m_tx_pps; - m_stats.m_stats.m_rx_pps += port_stats.m_stats.m_rx_pps; - - m_stats.m_stats.m_total_tx_pkts += port_stats.m_stats.m_total_tx_pkts; - m_stats.m_stats.m_total_rx_pkts += port_stats.m_stats.m_total_rx_pkts; - - m_stats.m_stats.m_total_tx_bytes += port_stats.m_stats.m_total_tx_bytes; - m_stats.m_stats.m_total_rx_bytes += port_stats.m_stats.m_total_rx_bytes; - - m_stats.m_stats.m_tx_rx_errors += port_stats.m_stats.m_tx_rx_errors; - } + return m_platform_api->get_dp_core_count(); } void TrexStateless::encode_stats(Json::Value &global) { - global["cpu_util"] = m_stats.m_stats.m_cpu_util; + const TrexPlatformApi *api = get_stateless_obj()->get_platform_api(); + + TrexPlatformGlobalStats stats; + api->get_global_stats(stats); - global["tx_bps"] = m_stats.m_stats.m_tx_bps; - global["rx_bps"] = m_stats.m_stats.m_rx_bps; + global["cpu_util"] = stats.m_stats.m_cpu_util; - global["tx_pps"] = m_stats.m_stats.m_tx_pps; - global["rx_pps"] = m_stats.m_stats.m_rx_pps; + global["tx_bps"] = stats.m_stats.m_tx_bps; + global["rx_bps"] = stats.m_stats.m_rx_bps; - global["total_tx_pkts"] = Json::Value::UInt64(m_stats.m_stats.m_total_tx_pkts); - global["total_rx_pkts"] = Json::Value::UInt64(m_stats.m_stats.m_total_rx_pkts); + global["tx_pps"] = stats.m_stats.m_tx_pps; + global["rx_pps"] = stats.m_stats.m_rx_pps; - global["total_tx_bytes"] = Json::Value::UInt64(m_stats.m_stats.m_total_tx_bytes); - global["total_rx_bytes"] = Json::Value::UInt64(m_stats.m_stats.m_total_rx_bytes); + global["total_tx_pkts"] = Json::Value::UInt64(stats.m_stats.m_total_tx_pkts); + global["total_rx_pkts"] = Json::Value::UInt64(stats.m_stats.m_total_rx_pkts); - global["tx_rx_errors"] = Json::Value::UInt64(m_stats.m_stats.m_tx_rx_errors); + global["total_tx_bytes"] = Json::Value::UInt64(stats.m_stats.m_total_tx_bytes); + global["total_rx_bytes"] = Json::Value::UInt64(stats.m_stats.m_total_rx_bytes); + + global["tx_rx_errors"] = Json::Value::UInt64(stats.m_stats.m_tx_rx_errors); for (uint8_t i = 0; i < m_port_count; i++) { std::stringstream ss; @@ -210,3 +155,20 @@ TrexStateless::encode_stats(Json::Value &global) { } } +/** + * generate a snapshot for publish (async publish) + * + */ +void +TrexStateless::generate_publish_snapshot(std::string &snapshot) { + Json::FastWriter writer; + Json::Value root; + + root["name"] = "trex-stateless-info"; + root["type"] = 0; + + /* stateless specific info goes here */ + root["data"] = Json::nullValue; + + snapshot = writer.write(root); +} diff --git a/src/stateless/cp/trex_stateless.h b/src/stateless/cp/trex_stateless.h index 649b25dd..57c6ef1d 100644 --- a/src/stateless/cp/trex_stateless.h +++ b/src/stateless/cp/trex_stateless.h @@ -29,9 +29,10 @@ limitations under the License. #include <trex_stream.h> #include <trex_stateless_port.h> -#include <trex_stateless_dp_core.h> #include <trex_rpc_server_api.h> +#include <internal_api/trex_platform_api.h> + /** * generic exception for errors * TODO: move this to a better place @@ -88,17 +89,17 @@ public: /* default values */ TrexStatelessCfg() { m_port_count = 0; - m_dp_core_count = 0; m_rpc_req_resp_cfg = NULL; m_rpc_async_cfg = NULL; - m_rpc_server_verbose = false; + m_rpc_server_verbose = false; + m_platform_api = NULL; } const TrexRpcServerConfig *m_rpc_req_resp_cfg; const TrexRpcServerConfig *m_rpc_async_cfg; + const TrexPlatformApi *m_platform_api; bool m_rpc_server_verbose; uint8_t m_port_count; - uint8_t m_dp_core_count; }; /** @@ -113,27 +114,8 @@ public: * reconfiguration is not allowed * an exception will be thrown */ - static void configure(const TrexStatelessCfg &cfg); - - /** - * destroy the instance - * - */ - static void destroy(); - - /** - * singleton public get instance - * - */ - static TrexStateless& get_instance() { - TrexStateless& instance = get_instance_internal(); - - if (!instance.m_is_configured) { - throw TrexException("object is not configured"); - } - - return instance; - } + TrexStateless(const TrexStatelessCfg &cfg); + ~TrexStateless(); /** * starts the control plane side @@ -152,12 +134,6 @@ public: uint8_t get_dp_core_count(); - /** - * update all the stats (deep update) - * (include all the ports and global stats) - * - */ - void update_stats(); /** * fetch all the stats @@ -165,22 +141,25 @@ public: */ void encode_stats(Json::Value &global); + /** + * generate a snapshot for publish + */ + void generate_publish_snapshot(std::string &snapshot); -protected: - TrexStateless(); + const TrexPlatformApi * get_platform_api() { + return (m_platform_api); + } - static TrexStateless& get_instance_internal () { - static TrexStateless instance; - return instance; + const std::vector <TrexStatelessPort *> get_port_list() { + return m_ports; } - /* c++ 2011 style singleton */ +protected: + + /* no copy or assignment */ TrexStateless(TrexStateless const&) = delete; void operator=(TrexStateless const&) = delete; - /* status */ - bool m_is_configured; - /* RPC server array */ TrexRpcServer *m_rpc_server; @@ -188,15 +167,20 @@ protected: std::vector <TrexStatelessPort *> m_ports; uint8_t m_port_count; - /* cores */ - std::vector <TrexStatelessDpCore *> m_dp_cores; - uint8_t m_dp_core_count; - - /* stats */ - TrexStatelessStats m_stats; + /* platform API */ + const TrexPlatformApi *m_platform_api; std::mutex m_global_cp_lock; }; +/** + * an anchor function + * + * @author imarom (25-Oct-15) + * + * @return TrexStateless& + */ +TrexStateless * get_stateless_obj(); + #endif /* __TREX_STATELESS_H__ */ diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp index a31847a5..cb6fcc0e 100644 --- a/src/stateless/cp/trex_stateless_port.cpp +++ b/src/stateless/cp/trex_stateless_port.cpp @@ -18,20 +18,34 @@ 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_stateless.h> #include <trex_stateless_port.h> +#include <trex_stateless_messaging.h> +#include <trex_streams_compiler.h> + #include <string> #ifndef TREX_RPC_MOCK_SERVER + // DPDK c++ issue -#define UINT8_MAX 255 -#define UINT16_MAX 0xFFFF +#ifndef UINT8_MAX + #define UINT8_MAX 255 +#endif + +#ifndef UINT16_MAX + #define UINT16_MAX 0xFFFF +#endif + // DPDK c++ issue #endif #include <rte_ethdev.h> #include <os_time.h> +void +port_id_to_cores(uint8_t port_id, std::vector<std::pair<uint8_t, uint8_t>> &cores_id_list); + using namespace std; /*************************** @@ -49,7 +63,7 @@ TrexStatelessPort::TrexStatelessPort(uint8_t port_id) : m_port_id(port_id) { * */ TrexStatelessPort::rc_e -TrexStatelessPort::start_traffic(void) { +TrexStatelessPort::start_traffic(double mul) { if (m_port_state != PORT_STATE_UP_IDLE) { return (RC_ERR_BAD_STATE_FOR_OP); @@ -59,19 +73,46 @@ TrexStatelessPort::start_traffic(void) { return (RC_ERR_NO_STREAMS); } + /* fetch all the streams from the table */ + vector<TrexStream *> streams; + get_stream_table()->get_object_list(streams); + + /* compiler it */ + TrexStreamsCompiler compiler; + TrexStreamsCompiledObj *compiled_obj = new TrexStreamsCompiledObj(m_port_id, mul); + + bool rc = compiler.compile(streams, *compiled_obj); + if (!rc) { + return (RC_ERR_FAILED_TO_COMPILE_STREAMS); + } + + /* generate a message to all the relevant DP cores to start transmitting */ + TrexStatelessCpToDpMsgBase *start_msg = new TrexStatelessDpStart(compiled_obj); + + send_message_to_dp(start_msg); + + /* move the state to transmiting */ m_port_state = PORT_STATE_TRANSMITTING; - /* real code goes here */ return (RC_OK); } -void +TrexStatelessPort::rc_e TrexStatelessPort::stop_traffic(void) { /* real code goes here */ - if (m_port_state == PORT_STATE_TRANSMITTING) { - m_port_state = PORT_STATE_UP_IDLE; + if (m_port_state != PORT_STATE_TRANSMITTING) { + return (RC_ERR_BAD_STATE_FOR_OP); } + + /* 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); + + m_port_state = PORT_STATE_UP_IDLE; + + return (RC_OK); } /** @@ -130,99 +171,42 @@ TrexStatelessPort::generate_handler() { return (ss.str()); } -/** - * update stats for the port - * - */ -void -TrexStatelessPort::update_stats() { - struct rte_eth_stats stats; - rte_eth_stats_get(m_port_id, &stats); - - /* copy straight values */ - m_stats.m_stats.m_total_tx_bytes = stats.obytes; - m_stats.m_stats.m_total_rx_bytes = stats.ibytes; - - m_stats.m_stats.m_total_tx_pkts = stats.opackets; - m_stats.m_stats.m_total_rx_pkts = stats.ipackets; - - /* calculate stats */ - m_stats.m_stats.m_tx_bps = m_stats.m_bw_tx_bps.add(stats.obytes); - m_stats.m_stats.m_rx_bps = m_stats.m_bw_rx_bps.add(stats.ibytes); - - m_stats.m_stats.m_tx_pps = m_stats.m_bw_tx_pps.add(stats.opackets); - m_stats.m_stats.m_rx_pps = m_stats.m_bw_rx_pps.add(stats.ipackets); - -} - -const TrexPortStats & -TrexStatelessPort::get_stats() { - return m_stats; -} void TrexStatelessPort::encode_stats(Json::Value &port) { - port["tx_bps"] = m_stats.m_stats.m_tx_bps; - port["rx_bps"] = m_stats.m_stats.m_rx_bps; + const TrexPlatformApi *api = get_stateless_obj()->get_platform_api(); - port["tx_pps"] = m_stats.m_stats.m_tx_pps; - port["rx_pps"] = m_stats.m_stats.m_rx_pps; + TrexPlatformInterfaceStats stats; + api->get_interface_stats(m_port_id, stats); - port["total_tx_pkts"] = Json::Value::UInt64(m_stats.m_stats.m_total_tx_pkts); - port["total_rx_pkts"] = Json::Value::UInt64(m_stats.m_stats.m_total_rx_pkts); + port["tx_bps"] = stats.m_stats.m_tx_bps; + port["rx_bps"] = stats.m_stats.m_rx_bps; - port["total_tx_bytes"] = Json::Value::UInt64(m_stats.m_stats.m_total_tx_bytes); - port["total_rx_bytes"] = Json::Value::UInt64(m_stats.m_stats.m_total_rx_bytes); - - port["tx_rx_errors"] = Json::Value::UInt64(m_stats.m_stats.m_tx_rx_errors); -} + port["tx_pps"] = stats.m_stats.m_tx_pps; + port["rx_pps"] = stats.m_stats.m_rx_pps; + port["total_tx_pkts"] = Json::Value::UInt64(stats.m_stats.m_total_tx_pkts); + port["total_rx_pkts"] = Json::Value::UInt64(stats.m_stats.m_total_rx_pkts); - -/*************************** - * BW measurement - * - **************************/ -/* TODO: move this to a common place */ -BWMeasure::BWMeasure() { - reset(); + port["total_tx_bytes"] = Json::Value::UInt64(stats.m_stats.m_total_tx_bytes); + port["total_rx_bytes"] = Json::Value::UInt64(stats.m_stats.m_total_rx_bytes); + + port["tx_rx_errors"] = Json::Value::UInt64(stats.m_stats.m_tx_rx_errors); } -void BWMeasure::reset(void) { - m_start=false; - m_last_time_msec=0; - m_last_bytes=0; - m_last_result=0.0; -}; - -double BWMeasure::calc_MBsec(uint32_t dtime_msec, - uint64_t dbytes){ - double rate=0.000008*( ( (double)dbytes*(double)os_get_time_freq())/((double)dtime_msec) ); - return(rate); -} +void +TrexStatelessPort::send_message_to_dp(TrexStatelessCpToDpMsgBase *msg) { -double BWMeasure::add(uint64_t size) { - if ( false == m_start ) { - m_start=true; - m_last_time_msec = os_get_time_msec() ; - m_last_bytes=size; - return(0.0); - } + std::vector<std::pair<uint8_t, uint8_t>> cores_id_list; - uint32_t ctime=os_get_time_msec(); - if ((ctime - m_last_time_msec) <os_get_time_freq() ) { - return(m_last_result); - } + get_stateless_obj()->get_platform_api()->port_id_to_cores(m_port_id, cores_id_list); - uint32_t dtime_msec = ctime-m_last_time_msec; - uint64_t dbytes = size - m_last_bytes; + for (auto core_pair : cores_id_list) { - m_last_time_msec = ctime; - m_last_bytes = size; + /* send the message to the core */ + CNodeRing *ring = CMsgIns::Ins()->getCpDp()->getRingCpToDp(core_pair.first); + ring->Enqueue((CGenNode *)msg->clone()); + } - m_last_result= 0.5*calc_MBsec(dtime_msec,dbytes) +0.5*(m_last_result); - return( m_last_result ); } - - diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h index 428d5aee..09183768 100644 --- a/src/stateless/cp/trex_stateless_port.h +++ b/src/stateless/cp/trex_stateless_port.h @@ -23,70 +23,7 @@ limitations under the License. #include <trex_stream.h> -/** - * bandwidth measurement class - * - */ -class BWMeasure { -public: - BWMeasure(); - void reset(void); - double add(uint64_t size); - -private: - double calc_MBsec(uint32_t dtime_msec, - uint64_t dbytes); - -public: - bool m_start; - uint32_t m_last_time_msec; - uint64_t m_last_bytes; - double m_last_result; -}; - -/** - * TRex stateless port stats - * - * @author imarom (24-Sep-15) - */ -class TrexPortStats { - -public: - TrexPortStats() { - m_stats = {0}; - - m_bw_tx_bps.reset(); - m_bw_rx_bps.reset(); - - m_bw_tx_pps.reset(); - m_bw_rx_pps.reset(); - } - -public: - - BWMeasure m_bw_tx_bps; - BWMeasure m_bw_rx_bps; - - BWMeasure m_bw_tx_pps; - BWMeasure m_bw_rx_pps; - - struct { - - double m_tx_bps; - double m_rx_bps; - - double m_tx_pps; - double m_rx_pps; - - uint64_t m_total_tx_pkts; - uint64_t m_total_rx_pkts; - - uint64_t m_total_tx_bytes; - uint64_t m_total_rx_bytes; - - uint64_t m_tx_rx_errors; - } m_stats; -}; +class TrexStatelessCpToDpMsgBase; /** * describes a stateless port @@ -121,13 +58,13 @@ public: * start traffic * */ - rc_e start_traffic(void); + rc_e start_traffic(double mul); /** * stop traffic * */ - void stop_traffic(void); + rc_e stop_traffic(void); /** * access the stream table @@ -203,28 +140,25 @@ public: } /** - * update the values of the stats - * - */ - void update_stats(); - - const TrexPortStats & get_stats(); - - /** * encode stats as JSON */ void encode_stats(Json::Value &port); + uint8_t get_port_id() { + return m_port_id; + } + private: std::string generate_handler(); + 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; - TrexPortStats m_stats; }; #endif /* __TREX_STATELESS_PORT_H__ */ diff --git a/src/stateless/cp/trex_stream.cpp b/src/stateless/cp/trex_stream.cpp index 182036f1..ba306137 100644 --- a/src/stateless/cp/trex_stream.cpp +++ b/src/stateless/cp/trex_stream.cpp @@ -20,6 +20,7 @@ limitations under the License. */ #include <trex_stream.h> #include <cstddef> +#include <string.h> /************************************** * stream @@ -103,14 +104,24 @@ TrexStream * TrexStreamTable::get_stream_by_id(uint32_t stream_id) { } } -void TrexStreamTable::get_stream_list(std::vector<uint32_t> &stream_list) { - stream_list.clear(); +void TrexStreamTable::get_id_list(std::vector<uint32_t> &id_list) { + id_list.clear(); for (auto stream : m_stream_table) { - stream_list.push_back(stream.first); + id_list.push_back(stream.first); } } +void TrexStreamTable::get_object_list(std::vector<TrexStream *> &object_list) { + object_list.clear(); + + for (auto stream : m_stream_table) { + object_list.push_back(stream.second); + } + +} + int TrexStreamTable::size() { return m_stream_table.size(); } + diff --git a/src/stateless/cp/trex_stream.h b/src/stateless/cp/trex_stream.h index f5bc96ef..c8a15240 100644 --- a/src/stateless/cp/trex_stream.h +++ b/src/stateless/cp/trex_stream.h @@ -37,10 +37,6 @@ class TrexRpcCmdAddStream; * */ class TrexStream { - /* provide the RPC parser a way to access private fields */ - friend class TrexRpcCmdAddStream; - friend class TrexRpcCmdGetStream; - friend class TrexStreamTable; public: TrexStream(uint8_t port_id, uint32_t stream_id); @@ -56,7 +52,7 @@ public: /* access the stream json */ const Json::Value & get_stream_json(); -protected: +public: /* basic */ uint8_t m_port_id; uint32_t m_stream_id; @@ -189,7 +185,13 @@ public: * * @param stream_list */ - void get_stream_list(std::vector<uint32_t> &stream_list); + void get_id_list(std::vector<uint32_t> &id_list); + + /** + * populate a list with all the stream objects + * + */ + void get_object_list(std::vector<TrexStream *> &object_list); /** * get the table size @@ -197,6 +199,9 @@ public: */ int size(); + std::unordered_map<int, TrexStream *>::iterator begin() {return m_stream_table.begin();} + std::unordered_map<int, TrexStream *>::iterator end() {return m_stream_table.end();} + private: /** * holds all the stream in a hash table by stream id diff --git a/src/stateless/cp/trex_streams_compiler.cpp b/src/stateless/cp/trex_streams_compiler.cpp new file mode 100644 index 00000000..06c0119a --- /dev/null +++ b/src/stateless/cp/trex_streams_compiler.cpp @@ -0,0 +1,106 @@ +/* + 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 <string.h> +#include <trex_streams_compiler.h> +#include <trex_stream.h> + +/************************************** + * stream compiled object + *************************************/ +TrexStreamsCompiledObj::TrexStreamsCompiledObj(uint8_t port_id, double mul) : m_port_id(port_id), m_mul(mul) { +} + +TrexStreamsCompiledObj::~TrexStreamsCompiledObj() { + for (auto &obj : m_objs) { + delete obj.m_pkt; + } + m_objs.clear(); +} + +void +TrexStreamsCompiledObj::add_compiled_stream(double isg_usec, double pps, uint8_t *pkt, uint16_t pkt_len) { + obj_st obj; + + obj.m_isg_usec = isg_usec; + obj.m_port_id = m_port_id; + obj.m_pps = pps * m_mul; + obj.m_pkt_len = pkt_len; + + obj.m_pkt = new uint8_t[pkt_len]; + memcpy(obj.m_pkt, pkt, pkt_len); + + m_objs.push_back(obj); +} + +TrexStreamsCompiledObj * +TrexStreamsCompiledObj::clone() { + + TrexStreamsCompiledObj *new_compiled_obj = new TrexStreamsCompiledObj(m_port_id, m_mul); + + /** + * clone each element + */ + for (auto obj : m_objs) { + new_compiled_obj->add_compiled_stream(obj.m_isg_usec, + obj.m_pps, + obj.m_pkt, + obj.m_pkt_len); + } + + return new_compiled_obj; +} + +/************************************** + * stream compiler + *************************************/ +bool +TrexStreamsCompiler::compile(const std::vector<TrexStream *> &streams, TrexStreamsCompiledObj &obj) { + /* for now we do something trivial, */ + for (auto stream : streams) { + + /* skip non-enabled streams */ + if (!stream->m_enabled) { + continue; + } + + /* for now skip also non self started streams */ + if (!stream->m_self_start) { + continue; + } + + /* for now support only continous ... */ + TrexStreamContinuous *cont_stream = dynamic_cast<TrexStreamContinuous *>(stream); + if (!cont_stream) { + continue; + } + + /* add it */ + obj.add_compiled_stream(cont_stream->m_isg_usec, + cont_stream->get_pps(), + cont_stream->m_pkt.binary, + cont_stream->m_pkt.len); + } + + return true; +} + + diff --git a/src/stateless/cp/trex_streams_compiler.h b/src/stateless/cp/trex_streams_compiler.h new file mode 100644 index 00000000..404fdd21 --- /dev/null +++ b/src/stateless/cp/trex_streams_compiler.h @@ -0,0 +1,79 @@ +/* + 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_STREAMS_COMPILER_H__ +#define __TREX_STREAMS_COMPILER_H__ + +#include <stdint.h> +#include <vector> + +class TrexStreamsCompiler; +class TrexStream; + +/** + * compiled object for a table of streams + * + * @author imarom (28-Oct-15) + */ +class TrexStreamsCompiledObj { + friend class TrexStreamsCompiler; +public: + + TrexStreamsCompiledObj(uint8_t port_id, double m_mul); + ~TrexStreamsCompiledObj(); + + struct obj_st { + double m_isg_usec; + double m_pps; + uint8_t *m_pkt; + uint16_t m_pkt_len; + uint8_t m_port_id; + }; + + const std::vector<obj_st> & get_objects() { + return m_objs; + } + + /** + * clone the compiled object + * + */ + TrexStreamsCompiledObj * clone(); + +private: + void add_compiled_stream(double isg_usec, double pps, uint8_t *pkt, uint16_t pkt_len); + std::vector<obj_st> m_objs; + + uint8_t m_port_id; + double m_mul; +}; + +class TrexStreamsCompiler { +public: + /** + * compiles a vector of streams to an object passable to the DP + * + * @author imarom (28-Oct-15) + * + */ + bool compile(const std::vector<TrexStream *> &streams, TrexStreamsCompiledObj &obj); +}; + +#endif /* __TREX_STREAMS_COMPILER_H__ */ diff --git a/src/stateless/dp/trex_stateless_dp_core.cpp b/src/stateless/dp/trex_stateless_dp_core.cpp index 3755b82c..b2bd0152 100644 --- a/src/stateless/dp/trex_stateless_dp_core.cpp +++ b/src/stateless/dp/trex_stateless_dp_core.cpp @@ -18,118 +18,186 @@ 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_stateless_dp_core.h> -#include <stdio.h> -#include <unistd.h> -#include <trex_stateless.h> +#include <trex_stateless_messaging.h> +#include <trex_streams_compiler.h> +#include <trex_stream_node.h> #include <bp_sim.h> -#ifndef TREX_RPC_MOCK_SERVER +static inline double +usec_to_sec(double usec) { + return (usec / (1000 * 1000)); +} + -// DPDK c++ issue -#define UINT8_MAX 255 -#define UINT16_MAX 0xFFFF -// DPDK c++ issue -#endif +void +TrexStatelessDpCore::create(uint8_t thread_id, CFlowGenListPerThread *core) { + m_thread_id = thread_id; + m_core = core; -#include <rte_ethdev.h> -#include "mbuf.h" + CMessagingManager * cp_dp = CMsgIns::Ins()->getCpDp(); + + m_ring_from_cp = cp_dp->getRingCpToDp(thread_id); + m_ring_to_cp = cp_dp->getRingDpToCp(thread_id); + + m_state = STATE_IDLE; +} /** - * TEST + * in idle state loop, the processor most of the time sleeps + * and periodically checks for messages * + * @author imarom (01-Nov-15) */ -static const uint8_t udp_pkt[]={ - 0x00,0x00,0x00,0x01,0x00,0x00, - 0x00,0x00,0x00,0x01,0x00,0x00, - 0x08,0x00, - - 0x45,0x00,0x00,0x81, - 0xaf,0x7e,0x00,0x00, - 0x12,0x11,0xd9,0x23, - 0x01,0x01,0x01,0x01, - 0x3d,0xad,0x72,0x1b, - - 0x11,0x11, - 0x11,0x11, - - 0x00,0x6d, - 0x00,0x00, - - 0x64,0x31,0x3a,0x61, - 0x64,0x32,0x3a,0x69,0x64, - 0x32,0x30,0x3a,0xd0,0x0e, - 0xa1,0x4b,0x7b,0xbd,0xbd, - 0x16,0xc6,0xdb,0xc4,0xbb,0x43, - 0xf9,0x4b,0x51,0x68,0x33,0x72, - 0x20,0x39,0x3a,0x69,0x6e,0x66,0x6f, - 0x5f,0x68,0x61,0x73,0x68,0x32,0x30,0x3a,0xee,0xc6,0xa3, - 0xd3,0x13,0xa8,0x43,0x06,0x03,0xd8,0x9e,0x3f,0x67,0x6f, - 0xe7,0x0a,0xfd,0x18,0x13,0x8d,0x65,0x31,0x3a,0x71,0x39, - 0x3a,0x67,0x65,0x74,0x5f,0x70,0x65,0x65,0x72,0x73,0x31, - 0x3a,0x74,0x38,0x3a,0x3d,0xeb,0x0c,0xbf,0x0d,0x6a,0x0d, - 0xa5,0x31,0x3a,0x79,0x31,0x3a,0x71,0x65,0x87,0xa6,0x7d, - 0xe7 -}; - -static int -test_inject_pkt(uint8_t *pkt, uint32_t pkt_size) { - - #ifndef TREX_RPC_MOCK_SERVER - rte_mempool_t * mp= CGlobalInfo::m_mem_pool[0].m_big_mbuf_pool ; - #else - rte_mempool_t * mp = NULL; - #endif - - rte_mbuf_t *m = rte_pktmbuf_alloc(mp); - if ( unlikely(m==0) ) { - printf("ERROR no packets \n"); - return (-1); +void +TrexStatelessDpCore::idle_state_loop() { + + while (m_state == STATE_IDLE) { + periodic_check_for_cp_messages(); + delay(200); } +} + +/** + * scehduler runs when traffic exists + * it will return when no more transmitting is done on this + * core + * + * @author imarom (01-Nov-15) + */ +void +TrexStatelessDpCore::start_scheduler() { + /* creates a maintenace job using the scheduler */ + CGenNode * node_sync = m_core->create_node() ; + node_sync->m_type = CGenNode::FLOW_SYNC; + node_sync->m_time = m_core->m_cur_time_sec + SYNC_TIME_OUT; + m_core->m_node_gen.add_node(node_sync); + + double old_offset = 0.0; + m_core->m_node_gen.flush_file(-1, 0.0, false, m_core, old_offset); + m_core->m_node_gen.close_file(m_core); +} + +void +TrexStatelessDpCore::start() { + + while (true) { + idle_state_loop(); + + start_scheduler(); + } +} + +void +TrexStatelessDpCore::add_cont_stream(uint8_t port_id, + double isg_usec, + double pps, + const uint8_t *pkt, + uint16_t pkt_len) { + + CGenNodeStateless *node = m_core->create_node_sl(); + + /* add periodic */ + node->m_type = CGenNode::STATELESS_PKT; + + node->m_time = m_core->m_cur_time_sec + usec_to_sec(isg_usec); + + pkt_dir_t dir = m_core->m_node_gen.m_v_if->port_id_to_dir(port_id); + node->m_flags = 0; + + /* set socket id */ + node->set_socket_id(m_core->m_node_gen.m_socket_id); + + /* build a mbuf from a packet */ + uint16_t pkt_size = pkt_len; + const uint8_t *stream_pkt = pkt; + + /* stateless specific fields */ + node->m_next_time_offset = 1.0 / pps; + node->m_is_stream_active = 1; + node->m_port_id = port_id; + + /* allocate const mbuf */ + rte_mbuf_t *m = CGlobalInfo::pktmbuf_alloc(node->get_socket_id(), pkt_size); + assert(m); + char *p = rte_pktmbuf_append(m, pkt_size); assert(p); - /* set pkt data */ - memcpy(p,pkt,pkt_size); + /* copy the packet */ + memcpy(p,stream_pkt,pkt_size); - rte_mbuf_t *tx_pkts[32]; - tx_pkts[0] = m; - uint8_t nb_pkts = 1; - uint16_t ret = rte_eth_tx_burst(0, 0, tx_pkts, nb_pkts); - (void)ret; - rte_pktmbuf_free(m); + /* set dir 0 or 1 client or server */ + node->set_mbuf_cache_dir(dir); - return (0); -} + /* TBD repace the mac if req we should add flag */ + m_core->m_node_gen.m_v_if->update_mac_addr_from_global_cfg(dir, m); + + /* set the packet as a readonly */ + node->set_cache_mbuf(m); + + /* keep track */ + m_active_nodes.push_back(node); + + /* schedule */ + m_core->m_node_gen.add_node((CGenNode *)node); + + m_state = TrexStatelessDpCore::STATE_TRANSMITTING; -static int -test_inject_udp_pkt(){ - return (test_inject_pkt((uint8_t*)udp_pkt,sizeof(udp_pkt))); } void -TrexStatelessDpCore::test_inject_dummy_pkt() { - test_inject_udp_pkt(); +TrexStatelessDpCore::start_traffic(TrexStreamsCompiledObj *obj) { + for (auto single_stream : obj->get_objects()) { + add_cont_stream(single_stream.m_port_id, + single_stream.m_isg_usec, + single_stream.m_pps, + single_stream.m_pkt, + single_stream.m_pkt_len); + } } -/*************************** - * DP core - * - **************************/ -TrexStatelessDpCore::TrexStatelessDpCore(uint8_t core_id) : m_core_id(core_id) { +void +TrexStatelessDpCore::stop_traffic(uint8_t port_id) { + /* we cannot remove nodes not from the top of the queue so + for every active node - make sure next time + the scheduler invokes it, it will be free */ + for (auto node : m_active_nodes) { + if (node->m_port_id == port_id) { + node->m_is_stream_active = 0; + } + } + + /* remove all the non active nodes */ + auto pred = std::remove_if(m_active_nodes.begin(), + m_active_nodes.end(), + [](CGenNodeStateless *node) { return (!node->m_is_stream_active); }); + + m_active_nodes.erase(pred, m_active_nodes.end()); + + if (m_active_nodes.size() == 0) { + m_state = STATE_IDLE; + /* stop the scheduler */ + + CGenNode *node = m_core->create_node() ; + + node->m_type = CGenNode::EXIT_SCHED; + + /* make sure it will be scheduled after the current node */ + node->m_time = m_core->m_cur_time_sec + 0.0001; + + m_core->m_node_gen.add_node(node); + } + } /** - * main function for DP core + * handle a message from CP to DP * */ -void -TrexStatelessDpCore::run() { - printf("\nOn DP core %d\n", m_core_id); - while (true) { - test_inject_dummy_pkt(); - rte_pause(); - } +void +TrexStatelessDpCore::handle_cp_msg(TrexStatelessCpToDpMsgBase *msg) { + msg->handle(this); + delete msg; } diff --git a/src/stateless/dp/trex_stateless_dp_core.h b/src/stateless/dp/trex_stateless_dp_core.h index 4b09b752..f4dbad08 100644 --- a/src/stateless/dp/trex_stateless_dp_core.h +++ b/src/stateless/dp/trex_stateless_dp_core.h @@ -21,23 +21,128 @@ limitations under the License. #ifndef __TREX_STATELESS_DP_CORE_H__ #define __TREX_STATELESS_DP_CORE_H__ -#include <stdint.h> +#include <vector> + +#include <msg_manager.h> +#include <pal_utl.h> + +class TrexStatelessCpToDpMsgBase; +class TrexStatelessDpStart; +class CFlowGenListPerThread; +class CGenNodeStateless; +class TrexStreamsCompiledObj; -/** - * stateless DP core object - * - */ class TrexStatelessDpCore { + public: - TrexStatelessDpCore(uint8_t core_id); + /* states */ + enum state_e { + STATE_IDLE, + STATE_TRANSMITTING + }; + + TrexStatelessDpCore() { + m_thread_id = 0; + m_core = NULL; + } + + /** + * "static constructor" + * + * @param thread_id + * @param core + */ + void create(uint8_t thread_id, CFlowGenListPerThread *core); + + /** + * launch the stateless DP core code + * + */ + void start(); + + /** + * dummy traffic creator + * + * @author imarom (27-Oct-15) + * + * @param pkt + * @param pkt_len + */ + void start_traffic(TrexStreamsCompiledObj *obj); + + /** + * stop all traffic for this core + * + */ + void stop_traffic(uint8_t port_id); + + /** + * check for and handle messages from CP + * + * @author imarom (27-Oct-15) + */ + void periodic_check_for_cp_messages() { + // doing this inline for performance reasons + + /* fast path */ + if ( likely ( m_ring_from_cp->isEmpty() ) ) { + return; + } - /* starts the DP core run */ - void run(); + while ( true ) { + CGenNode * node = NULL; + if (m_ring_from_cp->Dequeue(node) != 0) { + break; + } + assert(node); + + TrexStatelessCpToDpMsgBase * msg = (TrexStatelessCpToDpMsgBase *)node; + handle_cp_msg(msg); + } + + } private: - void test_inject_dummy_pkt(); - uint8_t m_core_id; + /** + * in idle state loop, the processor most of the time sleeps + * and periodically checks for messages + * + */ + void idle_state_loop(); + + /** + * real job is done when scheduler is launched + * + */ + void start_scheduler(); + + /** + * handles a CP to DP message + * + * @author imarom (27-Oct-15) + * + * @param msg + */ + void handle_cp_msg(TrexStatelessCpToDpMsgBase *msg); + + void add_cont_stream(uint8_t dir, + double isg, + double pps, + const uint8_t *pkt, + uint16_t pkt_len); + + uint8_t m_thread_id; + state_e m_state; + CNodeRing *m_ring_from_cp; + CNodeRing *m_ring_to_cp; + + /* holds the current active nodes */ + std::vector<CGenNodeStateless *> m_active_nodes; + + /* pointer to the main object */ + CFlowGenListPerThread *m_core; }; #endif /* __TREX_STATELESS_DP_CORE_H__ */ + diff --git a/src/stateless/dp/trex_stream_node.h b/src/stateless/dp/trex_stream_node.h new file mode 100644 index 00000000..92b428ab --- /dev/null +++ b/src/stateless/dp/trex_stream_node.h @@ -0,0 +1,105 @@ +/* + 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_STREAM_NODE_H__ +#define __TREX_STREAM_NODE_H__ + +#include <bp_sim.h> + +class TrexStatelessDpCore; + +/* this is a event for stateless */ +struct CGenNodeStateless : public CGenNodeBase { +friend class TrexStatelessDpCore; + +private: + void * m_cache_mbuf; + + double m_next_time_offset; + uint8_t m_is_stream_active; + uint8_t m_port_id; + + /* pad to match the size of CGenNode */ + uint8_t m_pad_end[87]; + + +public: + + inline bool is_active() { + return m_is_stream_active; + } + + + /** + * main function to handle an event of a packet tx + * + */ + inline void handle(CFlowGenListPerThread *thread) { + + thread->m_node_gen.m_v_if->send_node( (CGenNode *)this); + + /* in case of continues */ + m_time += m_next_time_offset; + + /* insert a new event */ + thread->m_node_gen.m_p_queue.push( (CGenNode *)this); + } + + void set_socket_id(socket_id_t socket){ + m_socket_id=socket; + } + + socket_id_t get_socket_id(){ + return ( m_socket_id ); + } + + inline void set_mbuf_cache_dir(pkt_dir_t dir){ + if (dir) { + m_flags |=NODE_FLAGS_DIR; + }else{ + m_flags &=~NODE_FLAGS_DIR; + } + } + + inline pkt_dir_t get_mbuf_cache_dir(){ + return ((pkt_dir_t)( m_flags &1)); + } + + + + inline void set_cache_mbuf(rte_mbuf_t * m){ + m_cache_mbuf=(void *)m; + m_flags |= NODE_FLAGS_MBUF_CACHE; + } + + inline rte_mbuf_t * get_cache_mbuf(){ + if ( m_flags &NODE_FLAGS_MBUF_CACHE ) { + return ((rte_mbuf_t *)m_cache_mbuf); + }else{ + return ((rte_mbuf_t *)0); + } + } + + +} __rte_cache_aligned; + +static_assert(sizeof(CGenNodeStateless) == sizeof(CGenNode), "sizeof(CGenNodeStateless) != sizeof(CGenNode)"); + +#endif /* __TREX_STREAM_NODE_H__ */ diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp new file mode 100644 index 00000000..032559bc --- /dev/null +++ b/src/stateless/messaging/trex_stateless_messaging.cpp @@ -0,0 +1,78 @@ +/* + 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_stateless_messaging.h> +#include <trex_stateless_dp_core.h> +#include <trex_streams_compiler.h> +#include <string.h> + +/************************* + start traffic message + ************************/ +TrexStatelessDpStart::TrexStatelessDpStart(TrexStreamsCompiledObj *obj) : m_obj(obj) { +} + + +/** + * clone for DP start message + * + */ +TrexStatelessCpToDpMsgBase * +TrexStatelessDpStart::clone() { + + TrexStreamsCompiledObj *new_obj = m_obj->clone(); + + TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpStart(new_obj); + + return new_msg; +} + +TrexStatelessDpStart::~TrexStatelessDpStart() { + if (m_obj) { + delete m_obj; + } +} + +bool +TrexStatelessDpStart::handle(TrexStatelessDpCore *dp_core) { + + dp_core->start_traffic(m_obj); + return true; +} + +/************************* + stop traffic message + ************************/ +bool +TrexStatelessDpStop::handle(TrexStatelessDpCore *dp_core) { + dp_core->stop_traffic(m_port_id); + return true; +} + +/** + * clone for DP stop message + * + */ +TrexStatelessCpToDpMsgBase * +TrexStatelessDpStop::clone() { + TrexStatelessCpToDpMsgBase *new_msg = new TrexStatelessDpStop(m_port_id); + + return new_msg; +} diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h new file mode 100644 index 00000000..7978b7f9 --- /dev/null +++ b/src/stateless/messaging/trex_stateless_messaging.h @@ -0,0 +1,101 @@ +/* + 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_STATELESS_MESSAGING_H__ +#define __TREX_STATELESS_MESSAGING_H__ + +#include <msg_manager.h> + +class TrexStatelessDpCore; +class TrexStreamsCompiledObj; + +/** + * defines the base class for CP to DP messages + * + * @author imarom (27-Oct-15) + */ +class TrexStatelessCpToDpMsgBase { +public: + + TrexStatelessCpToDpMsgBase() { + } + + virtual ~TrexStatelessCpToDpMsgBase() { + } + + /** + * virtual function to handle a message + * + */ + virtual bool handle(TrexStatelessDpCore *dp_core) = 0; + + + /** + * clone the current message + * + */ + virtual TrexStatelessCpToDpMsgBase * clone() = 0; + + /* no copy constructor */ + TrexStatelessCpToDpMsgBase(TrexStatelessCpToDpMsgBase &) = delete; + +}; + +/** + * a message to start traffic + * + * @author imarom (27-Oct-15) + */ +class TrexStatelessDpStart : public TrexStatelessCpToDpMsgBase { +public: + + TrexStatelessDpStart(TrexStreamsCompiledObj *obj); + + ~TrexStatelessDpStart(); + + virtual bool handle(TrexStatelessDpCore *dp_core); + + virtual TrexStatelessCpToDpMsgBase * clone(); + +private: + TrexStreamsCompiledObj *m_obj; +}; + +/** + * a message to stop traffic + * + * @author imarom (27-Oct-15) + */ +class TrexStatelessDpStop : public TrexStatelessCpToDpMsgBase { +public: + + TrexStatelessDpStop(uint8_t port_id) : m_port_id(port_id) { + } + + virtual bool handle(TrexStatelessDpCore *dp_core); + + virtual TrexStatelessCpToDpMsgBase * clone(); + +private: + uint8_t m_port_id; +}; + + +#endif /* __TREX_STATELESS_MESSAGING_H__ */ diff --git a/src/stub/trex_stateless_stub.cpp b/src/stub/trex_stateless_stub.cpp new file mode 100644 index 00000000..199356d8 --- /dev/null +++ b/src/stub/trex_stateless_stub.cpp @@ -0,0 +1,23 @@ + +#include <trex_stateless_dp_core.h> + +class CFlowGenListPerThread; +class TrexStatelessCpToDpMsgBase; + +void +TrexStatelessDpCore::create(unsigned char, CFlowGenListPerThread*) { + m_thread_id = 0; + m_core = NULL; + + m_state = STATE_IDLE; + + CMessagingManager * cp_dp = CMsgIns::Ins()->getCpDp(); + + m_ring_from_cp = cp_dp->getRingCpToDp(0); + m_ring_to_cp = cp_dp->getRingDpToCp(0); +} + +void TrexStatelessDpCore::start(){} + +void TrexStatelessDpCore::handle_cp_msg(TrexStatelessCpToDpMsgBase*) {} + diff --git a/src/time_histogram.cpp b/src/time_histogram.cpp index f1b47e59..96796bfc 100755 --- a/src/time_histogram.cpp +++ b/src/time_histogram.cpp @@ -182,10 +182,10 @@ void CTimeHistogram::DumpWinMax(FILE *fd){ } void CTimeHistogram::Dump(FILE *fd){ - fprintf (fd," min_delta : %lu usec \n",get_usec(m_min_delta)); + fprintf (fd," min_delta : %lu usec \n", (ulong)get_usec(m_min_delta)); fprintf (fd," cnt : %lu \n",m_cnt); fprintf (fd," high_cnt : %lu \n",m_high_cnt); - fprintf (fd," max_d_time : %lu usec\n",get_usec(m_max_dt)); + fprintf (fd," max_d_time : %lu usec\n", (ulong)get_usec(m_max_dt)); //fprintf (fd," average : %.0f usec\n", get_total_average()); fprintf (fd," sliding_average : %.0f usec\n", get_average_latency()); fprintf (fd," precent : %.1f %%\n",(100.0*(double)m_high_cnt/(double)m_cnt)); @@ -198,7 +198,7 @@ void CTimeHistogram::Dump(FILE *fd){ for (j=0; j<HISTOGRAM_SIZE_LOG; j++) { for (i=0; i<HISTOGRAM_SIZE; i++) { if (m_hcnt[j][i] >0 ) { - fprintf (fd," h[%lu] : %lu \n",(base*(i+1)),m_hcnt[j][i]); + fprintf (fd," h[%u] : %llu \n",(base*(i+1)),(unsigned long long)m_hcnt[j][i]); } } base=base*10; diff --git a/src/tuple_gen.h b/src/tuple_gen.h index 29adbd69..d34e27bc 100755 --- a/src/tuple_gen.h +++ b/src/tuple_gen.h @@ -553,6 +553,9 @@ public: class CServerPoolBase { public: + + virtual ~CServerPoolBase() {} + virtual void GenerateTuple(CTupleBase& tuple) = 0; virtual uint16_t GenerateOnePort(uint32_t idx) = 0; virtual void Delete() = 0; diff --git a/src/utl_json.cpp b/src/utl_json.cpp index 990346f5..fb55be0a 100755 --- a/src/utl_json.cpp +++ b/src/utl_json.cpp @@ -25,7 +25,7 @@ limitations under the License. std::string add_json(std::string name, uint32_t counter,bool last){ char buff[200]; - sprintf(buff,"\"%s\":%lu",name.c_str(),counter); + sprintf(buff,"\"%s\":%lu",name.c_str(), (ulong)counter); std::string s= std::string(buff); if (!last) { s+=","; @@ -35,7 +35,7 @@ std::string add_json(std::string name, uint32_t counter,bool last){ std::string add_json(std::string name, uint64_t counter,bool last){ char buff[200]; - sprintf(buff,"\"%s\":%llu",name.c_str(),counter); + sprintf(buff,"\"%s\":%llu",name.c_str(), (unsigned long long)counter); std::string s= std::string(buff); if (!last) { s+=","; diff --git a/src/utl_yaml.cpp b/src/utl_yaml.cpp index 5f3ca735..828817e4 100755 --- a/src/utl_yaml.cpp +++ b/src/utl_yaml.cpp @@ -104,6 +104,8 @@ bool utl_yaml_read_uint16(const YAML::Node& node, val = (uint16_t)val_tmp; res=true; } + + return (res); } bool utl_yaml_read_bool(const YAML::Node& node, |