diff options
Diffstat (limited to 'src')
37 files changed, 2521 insertions, 681 deletions
diff --git a/src/common/basic_utils.cpp b/src/common/basic_utils.cpp index 04e7aaba..fded49ec 100755 --- a/src/common/basic_utils.cpp +++ b/src/common/basic_utils.cpp @@ -20,6 +20,10 @@ limitations under the License. #include <sstream> #include <sys/resource.h> +#include "pal_utl.h" + +int my_inet_pton4(const char *src, unsigned char *dst); + bool utl_is_file_exists (const std::string& name) { if (FILE *file = fopen(name.c_str(), "r")) { fclose(file); @@ -174,13 +178,6 @@ void TestDump(void){ utl_DumpBuffer2(stdout,buffer,31,1,4,SHOW_BUFFER_ADDR_EN |SHOW_BUFFER_CHAR); } -std::string -utl_macaddr_to_str(const uint8_t *mac) { - std::string tmp; - utl_macaddr_to_str(mac, tmp); - return tmp; -} - void utl_macaddr_to_str(const uint8_t *macaddr, std::string &output) { for (int i = 0; i < 6; i++) { @@ -197,6 +194,26 @@ void utl_macaddr_to_str(const uint8_t *macaddr, std::string &output) { } +std::string utl_macaddr_to_str(const uint8_t *macaddr) { + std::string tmp; + utl_macaddr_to_str(macaddr, tmp); + + return tmp; +} + +bool utl_str_to_macaddr(const std::string &s, uint8_t *mac) { + int last = -1; + int rc = sscanf(s.c_str(), "%hhx:%hhx:%hhx:%hhx:%hhx:%hhx%n", + mac + 0, mac + 1, mac + 2, mac + 3, mac + 4, mac + 5, + &last); + + if ( (rc != 6) || (s.size() != last) ) { + return false; + } + + return true; +} + /** * generate a random connection handler * @@ -255,3 +272,23 @@ void utl_set_coredump_size(long size, bool map_huge_pages) { fprintf(fp, "%08x\n", mask); fclose(fp); } + +bool utl_ipv4_to_uint32(const char *ipv4_str, uint32_t &ipv4_num) { + + uint32_t tmp; + + int rc = my_inet_pton4(ipv4_str, (unsigned char *)&tmp); + if (!rc) { + return false; + } + + ipv4_num = PAL_NTOHL(tmp); + + return true; +} + +std::string utl_uint32_to_ipv4(uint32_t ipv4_addr) { + std::stringstream ss; + ss << ((ipv4_addr >> 24) & 0xff) << "." << ((ipv4_addr >> 16) & 0xff) << "." << ((ipv4_addr >> 8) & 0xff) << "." << (ipv4_addr & 0xff); + return ss.str(); +} diff --git a/src/common/basic_utils.h b/src/common/basic_utils.h index 1079ecfc..36f9db85 100755 --- a/src/common/basic_utils.h +++ b/src/common/basic_utils.h @@ -87,6 +87,8 @@ bool utl_is_file_exists (const std::string& name) ; void utl_macaddr_to_str(const uint8_t *macaddr, std::string &output); std::string utl_macaddr_to_str(const uint8_t *macaddr); +bool utl_str_to_macaddr(const std::string &s, uint8_t *mac); + std::string utl_generate_random_str(unsigned int &seed, int len); /** @@ -99,6 +101,9 @@ std::string utl_generate_random_str(unsigned int &seed, int len); */ void utl_set_coredump_size(long size, bool map_huge_pages = false); +bool utl_ipv4_to_uint32(const char *ipv4_str, uint32_t &ipv4_num); +std::string utl_uint32_to_ipv4(uint32_t ipv4_addr); + #endif diff --git a/src/common/captureFile.h b/src/common/captureFile.h index fefa62bd..d87e57b6 100755 --- a/src/common/captureFile.h +++ b/src/common/captureFile.h @@ -227,9 +227,10 @@ class CFileWriterBase { public: - virtual ~CFileWriterBase(){}; - virtual bool Create(char * name) = 0; + virtual ~CFileWriterBase(){}; + virtual bool Create(char * name) = 0; virtual bool write_packet(CCapPktRaw * lpPacket)=0; + virtual void flush_to_disk() = 0; }; diff --git a/src/common/erf.cpp b/src/common/erf.cpp index 76945b01..f872a281 100755 --- a/src/common/erf.cpp +++ b/src/common/erf.cpp @@ -280,7 +280,11 @@ bool CErfFileWriter::write_packet(CCapPktRaw * lpPacket){ return true; } - +void CErfFileWriter::flush_to_disk() { + if (m_fd) { + fflush(m_fd); + } +} bool CPcapFileWriter::Create(char *file_name){ m_fd=CAP_FOPEN_64(file_name,"wb"); diff --git a/src/common/erf.h b/src/common/erf.h index e1b83e46..bec94872 100755 --- a/src/common/erf.h +++ b/src/common/erf.h @@ -224,6 +224,13 @@ public: virtual bool Create(char *file_name); void Delete(); virtual bool write_packet(CCapPktRaw * lpPacket); + + /** + * flush all packets to disk + * + */ + void flush_to_disk(); + private: FILE *m_fd; int m_cnt; diff --git a/src/common/pcap.cpp b/src/common/pcap.cpp index 8e9bf0ac..b976aed7 100755 --- a/src/common/pcap.cpp +++ b/src/common/pcap.cpp @@ -223,11 +223,18 @@ bool LibPCapWriter::Create(char * name) printf(" ERROR create file \n"); return(false); } + /* prepare the write counter */ m_pkt_count = 0; return init(); } +void LibPCapWriter::flush_to_disk() { + if (m_is_open) { + fflush(m_file_handler); + } +} + /** * * Write the libpcap header. diff --git a/src/common/pcap.h b/src/common/pcap.h index 3f8dfd21..265ea17b 100755 --- a/src/common/pcap.h +++ b/src/common/pcap.h @@ -1,5 +1,5 @@ -#ifndef __LIBPCAP_H__ -#define __LIBPCAP_H__ +#ifndef __TREX_LIBPCAP_H__ +#define __TREX_LIBPCAP_H__ /* Copyright (c) 2015-2015 Cisco Systems, Inc. @@ -143,6 +143,13 @@ public: */ void Close(); + /** + * flush all packets to disk + * + * @author imarom (11/24/2016) + */ + void flush_to_disk(); + private: bool init(); @@ -151,4 +158,5 @@ private: bool m_is_open; uint32_t m_pkt_count; }; + #endif diff --git a/src/debug.cpp b/src/debug.cpp index d0b7cf11..e272424c 100644 --- a/src/debug.cpp +++ b/src/debug.cpp @@ -142,7 +142,7 @@ rte_mbuf_t *CTrexDebug::create_test_pkt(int ip_ver, uint16_t l4_proto, uint8_t t pkt = CTestPktGen::create_test_pkt(l3_type, l4_proto, ttl, ip_id, flags, 1000, pkt_size); - /* DEBUG print the packet + /* DEBUG print the packet utl_k12_pkt_format(stdout,pkt, pkt_size) ; */ @@ -370,6 +370,7 @@ int CTrexDebug::verify_hw_rules(bool recv_all) { case STL: if ( CGlobalInfo::m_options.is_stateless() ) { exp_q = MAIN_DPDK_RX_Q; + pkt_flags |= DPF_TOS_1; } else { exp_q = MAIN_DPDK_DATA_Q; } @@ -379,6 +380,7 @@ int CTrexDebug::verify_hw_rules(bool recv_all) { exp_q = MAIN_DPDK_DATA_Q; } else { exp_q = MAIN_DPDK_RX_Q; + pkt_flags |= DPF_TOS_1; } break; default: diff --git a/src/dpdk/drivers/net/enic/enic_ethdev.c b/src/dpdk/drivers/net/enic/enic_ethdev.c index c05476b2..6a86e23f 100644 --- a/src/dpdk/drivers/net/enic/enic_ethdev.c +++ b/src/dpdk/drivers/net/enic/enic_ethdev.c @@ -436,22 +436,6 @@ static void enicpmd_dev_stats_reset(struct rte_eth_dev *eth_dev) } -int enicpmd_dev_get_fw_support(int port_id, - uint32_t *ver){ - struct rte_eth_dev *dev; - - RTE_ETH_VALID_PORTID_OR_ERR_RET(port_id, -EINVAL); - - dev = &rte_eth_devices[port_id]; - *ver=0; - - struct enic *enic = pmd_priv(dev); - enic->adv_filters; - if ( enic->adv_filters ==0 ) { - return (-1); - } - return (0); -} static void enicpmd_dev_info_get(struct rte_eth_dev *eth_dev, diff --git a/src/dpdk/drivers/net/enic/enic_main.c b/src/dpdk/drivers/net/enic/enic_main.c index 889bc692..473bfc3c 100644 --- a/src/dpdk/drivers/net/enic/enic_main.c +++ b/src/dpdk/drivers/net/enic/enic_main.c @@ -166,6 +166,7 @@ void enic_dev_stats_get(struct enic *enic, struct rte_eth_stats *r_stats) return; } + /* The number of truncated packets can only be calculated by * subtracting a hardware counter from error packets received by * the driver. Note: this causes transient inaccuracies in the @@ -180,7 +181,7 @@ void enic_dev_stats_get(struct enic *enic, struct rte_eth_stats *r_stats) r_stats->ipackets = stats->rx.rx_frames_ok - rx_truncated; r_stats->opackets = stats->tx.tx_frames_ok; - r_stats->ibytes = stats->rx.rx_bytes_ok; + r_stats->ibytes = stats->rx.rx_unicast_bytes_ok+stats->rx.rx_multicast_bytes_ok+stats->rx.rx_broadcast_bytes_ok; r_stats->obytes = stats->tx.tx_bytes_ok; r_stats->ierrors = stats->rx.rx_errors + stats->rx.rx_drop; diff --git a/src/dpdk/drivers/net/ixgbe/ixgbe_ethdev.c b/src/dpdk/drivers/net/ixgbe/ixgbe_ethdev.c index d478a159..72963a89 100644 --- a/src/dpdk/drivers/net/ixgbe/ixgbe_ethdev.c +++ b/src/dpdk/drivers/net/ixgbe/ixgbe_ethdev.c @@ -5784,13 +5784,17 @@ ixgbe_add_del_ethertype_filter(struct rte_eth_dev *dev, if (filter->queue >= IXGBE_MAX_RX_QUEUE_NUM) return -EINVAL; - +#define TREX_PATCH +#ifndef TREX_PATCH + // no real reason to block this. + // We configure rules using FDIR and ethertype that point to same queue, so there are no race condition issues. if (filter->ether_type == ETHER_TYPE_IPv4 || filter->ether_type == ETHER_TYPE_IPv6) { PMD_DRV_LOG(ERR, "unsupported ether_type(0x%04x) in" " ethertype filter.", filter->ether_type); return -EINVAL; } +#endif if (filter->flags & RTE_ETHTYPE_FLAGS_MAC) { PMD_DRV_LOG(ERR, "mac compare is unsupported."); diff --git a/src/flow_stat.cpp b/src/flow_stat.cpp index 84be590f..92cbca6e 100644 --- a/src/flow_stat.cpp +++ b/src/flow_stat.cpp @@ -461,6 +461,7 @@ CFlowStatRuleMgr::CFlowStatRuleMgr() { memset(m_rx_cant_count_err, 0, sizeof(m_rx_cant_count_err)); memset(m_tx_cant_count_err, 0, sizeof(m_tx_cant_count_err)); m_num_ports = 0; // need to call create to init + m_mode = FLOW_STAT_MODE_NORMAL; } CFlowStatRuleMgr::~CFlowStatRuleMgr() { @@ -802,11 +803,14 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) { #endif if (m_num_started_streams == 0) { + send_start_stop_msg_to_rx(true); // First transmitting stream. Rx core should start reading packets; + //also good time to zero global counters memset(m_rx_cant_count_err, 0, sizeof(m_rx_cant_count_err)); memset(m_tx_cant_count_err, 0, sizeof(m_tx_cant_count_err)); + #if 0 // wait to make sure that message is acknowledged. RX core might be in deep sleep mode, and we want to // start transmitting packets only after it is working, otherwise, packets will get lost. if (m_rx_core) { // in simulation, m_rx_core will be NULL @@ -819,6 +823,8 @@ int CFlowStatRuleMgr::start_stream(TrexStream * stream) { } } } + #endif + } else { // make sure rx core is working. If not, we got really confused somehow. if (m_rx_core) @@ -966,13 +972,24 @@ int CFlowStatRuleMgr::set_mode(enum flow_stat_mode_e mode) { extern bool rx_should_stop; void CFlowStatRuleMgr::send_start_stop_msg_to_rx(bool is_start) { TrexStatelessCpToRxMsgBase *msg; - + if (is_start) { - msg = new TrexStatelessRxStartMsg(); + static MsgReply<bool> reply; + reply.reset(); + + msg = new TrexStatelessRxEnableLatency(reply); + m_ring_to_rx->Enqueue((CGenNode *)msg); + + /* hold until message was ack'ed - otherwise we might lose packets */ + if (m_rx_core) { + reply.wait_for_reply(); + assert(m_rx_core->is_working()); + } + } else { - msg = new TrexStatelessRxStopMsg(); + msg = new TrexStatelessRxDisableLatency(); + m_ring_to_rx->Enqueue((CGenNode *)msg); } - m_ring_to_rx->Enqueue((CGenNode *)msg); } // return false if no counters changed since last run. true otherwise diff --git a/src/internal_api/trex_platform_api.h b/src/internal_api/trex_platform_api.h index 631f9a3e..5723503c 100644 --- a/src/internal_api/trex_platform_api.h +++ b/src/internal_api/trex_platform_api.h @@ -28,6 +28,7 @@ limitations under the License. #include <string.h> #include "flow_stat_parser.h" #include "trex_defs.h" +#include "trex_stateless_rx_defs.h" #include "trex_port_attr.h" #include <json/json.h> @@ -112,19 +113,13 @@ public: IF_STAT_RX_BYTES_COUNT = 8, // Card support counting rx bytes }; - struct mac_cfg_st { - uint8_t hw_macaddr[6]; - uint8_t src_macaddr[6]; - uint8_t dst_macaddr[6]; - }; - /** * interface static info * */ struct intf_info_st { std::string driver_name; - mac_cfg_st mac_info; + uint8_t hw_macaddr[6]; std::string pci_addr; int numa_node; bool has_crc; @@ -234,7 +229,7 @@ public: info.has_crc = true; info.numa_node = 0; - memset(&info.mac_info, 0, sizeof(info.mac_info)); + memset(&info.hw_macaddr, 0, sizeof(info.hw_macaddr)); } virtual void get_interface_stats(uint8_t interface_id, TrexPlatformInterfaceStats &stats) const { diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp index 6f18b0b8..c9e0c6a1 100644 --- a/src/main_dpdk.cpp +++ b/src/main_dpdk.cpp @@ -132,7 +132,7 @@ static char global_master_id_str[10]; class CTRexExtendedDriverBase { public: - + /* by default NIC driver adds CRC */ virtual bool has_crc_added() { return true; @@ -168,6 +168,7 @@ public: virtual CFlowStatParser *get_flow_stat_parser(); virtual int set_rcv_all(CPhyEthIF * _if, bool set_on)=0; virtual TRexPortAttr * create_port_attr(uint8_t port_id) = 0; + virtual uint8_t get_num_crc_fix_bytes() {return 0;} /* Does this NIC type support automatic packet dropping in case of a link down? in case it is supported the packets will be dropped, else there would be a back pressure to tx queues @@ -277,6 +278,18 @@ public: virtual int set_rcv_all(CPhyEthIF * _if, bool set_on) {return 0;} }; +class CTRexExtendedDriverBaseE1000 : public CTRexExtendedDriverBase1GVm { + CTRexExtendedDriverBaseE1000() { + // E1000 driver is only relevant in VM in our case + CGlobalInfo::m_options.preview.set_vm_one_queue_enable(true); + } +public: + static CTRexExtendedDriverBase * create() { + return ( new CTRexExtendedDriverBaseE1000() ); + } + // e1000 driver handing us packets with ethernet CRC, so we need to chop them + virtual uint8_t get_num_crc_fix_bytes() {return 4;} +}; class CTRexExtendedDriverBase10G : public CTRexExtendedDriverBase { public: @@ -316,7 +329,8 @@ public: | TrexPlatformApi::IF_STAT_PAYLOAD; } virtual CFlowStatParser *get_flow_stat_parser(); - virtual int set_rcv_all(CPhyEthIF * _if, bool set_on) {return 0;} + int add_del_eth_filter(CPhyEthIF * _if, bool is_add, uint16_t ethertype); + virtual int set_rcv_all(CPhyEthIF * _if, bool set_on); }; class CTRexExtendedDriverBase40G : public CTRexExtendedDriverBase10G { @@ -379,7 +393,7 @@ private: uint8_t m_if_per_card; }; -class CTRexExtendedDriverBaseVIC : public CTRexExtendedDriverBase40G { +class CTRexExtendedDriverBaseVIC : public CTRexExtendedDriverBase { public: CTRexExtendedDriverBaseVIC(){ m_if_per_card=2; @@ -396,6 +410,22 @@ public: virtual bool is_hardware_filter_is_supported(){ return (true); } + virtual void update_global_config_fdir(port_cfg_t * cfg){ + } + + + virtual bool is_hardware_support_drop_queue(){ + return(true); + } + + void clear_extended_stats(CPhyEthIF * _if); + + void get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats); + + + virtual int get_min_sample_rate(void){ + return (RX_CHECK_MIX_SAMPLE_RATE); + } virtual int verify_fw_ver(int i); @@ -556,7 +586,7 @@ private: /* virtual devices */ - register_driver(std::string("rte_em_pmd"),CTRexExtendedDriverBase1GVm::create); + register_driver(std::string("rte_em_pmd"),CTRexExtendedDriverBaseE1000::create); register_driver(std::string("rte_vmxnet3_pmd"),CTRexExtendedDriverBase1GVm::create); register_driver(std::string("rte_virtio_pmd"),CTRexExtendedDriverBase1GVm::create); @@ -1078,7 +1108,7 @@ static int parse_options(int argc, char *argv[], CParserOption* po, bool first_t } if (po->preview.get_is_rx_check_enable() || po->is_latency_enabled() || CGlobalInfo::is_learn_mode() - || (CGlobalInfo::m_options.m_arp_ref_per != 0)) { + || (CGlobalInfo::m_options.m_arp_ref_per != 0) || get_vm_one_queue_enable()) { po->set_rx_enabled(); } @@ -1627,6 +1657,7 @@ int DpdkTRexPortAttr::set_led(bool on){ int DpdkTRexPortAttr::get_flow_ctrl(int &mode) { int ret = rte_eth_dev_flow_ctrl_get(m_port_id, &fc_conf_tmp); if (ret) { + mode = -1; return ret; } mode = (int) fc_conf_tmp.mode; @@ -1723,12 +1754,19 @@ bool DpdkTRexPortAttr::update_link_status_nowait(){ rte_eth_link new_link; bool changed = false; rte_eth_link_get_nowait(m_port_id, &new_link); + if (new_link.link_speed != m_link.link_speed || new_link.link_duplex != m_link.link_duplex || new_link.link_autoneg != m_link.link_autoneg || new_link.link_status != m_link.link_status) { changed = true; + + /* in case of link status change - notify the dest object */ + if (new_link.link_status != m_link.link_status) { + get_dest().on_link_down(); + } } + m_link = new_link; return changed; } @@ -1770,7 +1808,7 @@ bool DpdkTRexPortAttr::get_promiscuous(){ } -void DpdkTRexPortAttr::macaddr_get(struct ether_addr *mac_addr){ +void DpdkTRexPortAttr::get_hw_src_mac(struct ether_addr *mac_addr){ rte_eth_macaddr_get(m_port_id , mac_addr); } @@ -1961,7 +1999,6 @@ public: virtual int send_node(CGenNode * node); virtual void send_one_pkt(pkt_dir_t dir, rte_mbuf_t *m); virtual int flush_tx_queue(void); - __attribute__ ((noinline)) void handle_rx_queue(); __attribute__ ((noinline)) void handle_slowpath_features(CGenNode *node, rte_mbuf_t *m, uint8_t *p, pkt_dir_t dir); void apply_client_cfg(const ClientCfgBase *cfg, rte_mbuf_t *m, pkt_dir_t dir, uint8_t *p); @@ -2037,46 +2074,6 @@ bool CCoreEthIF::Create(uint8_t core_id, return (true); } -// This function is only relevant if we are in VM. In this case, we only have one rx queue. Can't have -// rules to drop queue 0 packets, and pass queue 1 packets to RX core, like in other cases. -// We receive all packets in the same core that transmitted, and handle them to RX core. -void CCoreEthIF::handle_rx_queue(void) { - if ( likely( ! get_vm_one_queue_enable() ) ) { - return; - } - - pkt_dir_t dir; - bool is_rx = get_is_rx_thread_enabled(); - for (dir=CLIENT_SIDE; dir<CS_NUM; dir++) { - CCorePerPort * lp_port=&m_ports[dir]; - CPhyEthIF * lp=lp_port->m_port; - - rte_mbuf_t * rx_pkts[32]; - int j=0; - - while (true) { - j++; - uint16_t cnt =lp->rx_burst(0,rx_pkts,32); - if ( cnt ) { - int i; - for (i=0; i<(int)cnt;i++) { - rte_mbuf_t * m=rx_pkts[i]; - if ( is_rx ){ - if (!process_rx_pkt(dir,m)){ - rte_pktmbuf_free(m); - } - }else{ - rte_pktmbuf_free(m); - } - } - } - if ((cnt<5) || j>10 ) { - break; - } - } - } -} - int CCoreEthIF::flush_tx_queue(void){ /* flush both sides */ pkt_dir_t dir; @@ -2089,8 +2086,6 @@ int CCoreEthIF::flush_tx_queue(void){ } } - handle_rx_queue(); - return 0; } @@ -2580,10 +2575,11 @@ private: class CLatencyVmPort : public CPortLatencyHWBase { public: void Create(uint8_t port_index,CNodeRing * ring, - CLatencyManager * mgr){ + CLatencyManager * mgr, CPhyEthIF * p) { m_dir = (port_index%2); m_ring_to_dp = ring; m_mgr = mgr; + m_port = p; } virtual int tx(rte_mbuf_t * m){ @@ -2610,17 +2606,23 @@ public: return (-1); } - virtual rte_mbuf_t * rx(){ - return (0); + virtual rte_mbuf_t * rx() { + rte_mbuf_t * rx_pkts[1]; + uint16_t cnt = m_port->rx_burst(0, rx_pkts, 1); + if (cnt) { + return (rx_pkts[0]); + } else { + return (0); + } } - virtual uint16_t rx_burst(struct rte_mbuf **rx_pkts, - uint16_t nb_pkts){ - return (0); + virtual uint16_t rx_burst(struct rte_mbuf **rx_pkts, uint16_t nb_pkts) { + uint16_t cnt = m_port->rx_burst(0, rx_pkts, nb_pkts); + return (cnt); } - private: + CPhyEthIF * m_port; uint8_t m_dir; CNodeRing * m_ring_to_dp; /* ring dp -> latency thread */ CLatencyManager * m_mgr; @@ -3128,9 +3130,9 @@ public: uint32_t m_max_ports; /* active number of ports supported options are 2,4,8,10,12 */ uint32_t m_max_cores; /* current number of cores , include master and latency ==> ( master)1+c*(m_max_ports>>1)+1( latency ) */ uint32_t m_cores_mul; /* how cores multipler given c=4 ==> m_cores_mul */ - uint32_t m_max_queues_per_port; - uint32_t m_cores_to_dual_ports; /* number of ports that will handle dual ports */ - uint16_t m_latency_tx_queue_id; + uint32_t m_max_queues_per_port; // Number of TX queues per port + uint32_t m_cores_to_dual_ports; /* number of TX cores allocated for each port pair */ + uint16_t m_rx_core_tx_q_id; /* TX q used by rx core */ // statistic CPPSMeasure m_cps; float m_expected_pps; @@ -3317,11 +3319,20 @@ void CGlobalTRex::pre_test() { // we don't have dest MAC. Get it from what we resolved. uint32_t ip = CGlobalInfo::m_options.m_ip_cfg[port_id].get_def_gw(); uint16_t vlan = CGlobalInfo::m_options.m_ip_cfg[port_id].get_vlan(); - if (! pretest.get_mac(port_id, ip, vlan, mac)) { + + if (!pretest.get_mac(port_id, ip, vlan, mac)) { fprintf(stderr, "Failed resolving dest MAC for default gateway:%d.%d.%d.%d on port %d\n" , (ip >> 24) & 0xFF, (ip >> 16) & 0xFF, (ip >> 8) & 0xFF, ip & 0xFF, port_id); - exit(1); + + if (get_is_stateless()) { + continue; + } else { + exit(1); + } } + + + memcpy(CGlobalInfo::m_options.m_mac_addr[port_id].u.m_mac.dest, mac, ETHER_ADDR_LEN); // if port is connected in loopback, no need to send gratuitous ARP. It will only confuse our ingress counters. if (need_grat_arp[port_id] && (! pretest.is_loopback(port_id))) { @@ -3340,6 +3351,16 @@ void CGlobalTRex::pre_test() { // Configure port back to normal mode. Only relevant packets handled by software. CTRexExtendedDriverDb::Ins()->get_drv()->set_rcv_all(pif, false); + + /* set resolved IPv4 */ + uint32_t dg = CGlobalInfo::m_options.m_ip_cfg[port_id].get_def_gw(); + const uint8_t *dst_mac = CGlobalInfo::m_options.m_mac_addr[port_id].u.m_mac.dest; + if (dg) { + m_ports[port_id].get_port_attr()->get_dest().set_dest(dg, dst_mac); + } else { + m_ports[port_id].get_port_attr()->get_dest().set_dest(dst_mac); + } + } } } @@ -3408,12 +3429,14 @@ bool CGlobalTRex::is_all_links_are_up(bool dump){ void CGlobalTRex::try_stop_all_cores(){ TrexStatelessDpQuit * dp_msg= new TrexStatelessDpQuit(); - TrexStatelessRxQuit * rx_msg= new TrexStatelessRxQuit(); send_message_all_dp(dp_msg); + delete dp_msg; + if (get_is_stateless()) { + TrexStatelessRxQuit * rx_msg= new TrexStatelessRxQuit(); send_message_to_rx(rx_msg); } - delete dp_msg; + // no need to delete rx_msg. Deleted by receiver bool all_core_finished = false; int i; @@ -3463,6 +3486,7 @@ int CGlobalTRex::ixgbe_rx_queue_flush(){ } +// init stateful rx core void CGlobalTRex::ixgbe_configure_mg(void) { int i; CLatencyManagerCfg mg_cfg; @@ -3485,13 +3509,13 @@ void CGlobalTRex::ixgbe_configure_mg(void) { if ( get_vm_one_queue_enable() ) { /* vm mode, indirect queues */ for (i=0; i<m_max_ports; i++) { - + CPhyEthIF * _if = &m_ports[i]; CMessagingManager * rx_dp=CMsgIns::Ins()->getRxDp(); uint8_t thread_id = (i>>1); CNodeRing * r = rx_dp->getRingCpToDp(thread_id); - m_latency_vm_vports[i].Create((uint8_t)i,r,&m_mg); + m_latency_vm_vports[i].Create((uint8_t)i, r, &m_mg, _if); mg_cfg.m_ports[i] =&m_latency_vm_vports[i]; } @@ -3500,7 +3524,7 @@ void CGlobalTRex::ixgbe_configure_mg(void) { for (i=0; i<m_max_ports; i++) { CPhyEthIF * _if=&m_ports[i]; _if->dump_stats(stdout); - m_latency_vports[i].Create(_if,m_latency_tx_queue_id,1); + m_latency_vports[i].Create(_if, m_rx_core_tx_q_id, 1); mg_cfg.m_ports[i] =&m_latency_vports[i]; } @@ -3517,20 +3541,22 @@ void CGlobalTRex::rx_sl_configure(void) { int i; rx_sl_cfg.m_max_ports = m_max_ports; + rx_sl_cfg.m_num_crc_fix_bytes = get_ex_drv()->get_num_crc_fix_bytes(); if ( get_vm_one_queue_enable() ) { /* vm mode, indirect queues */ for (i=0; i < m_max_ports; i++) { + CPhyEthIF * _if = &m_ports[i]; CMessagingManager * rx_dp = CMsgIns::Ins()->getRxDp(); uint8_t thread_id = (i >> 1); CNodeRing * r = rx_dp->getRingCpToDp(thread_id); - m_latency_vm_vports[i].Create((uint8_t)i, r, &m_mg); + m_latency_vm_vports[i].Create(i, r, &m_mg, _if); rx_sl_cfg.m_ports[i] = &m_latency_vm_vports[i]; } } else { for (i = 0; i < m_max_ports; i++) { CPhyEthIF * _if = &m_ports[i]; - m_latency_vports[i].Create(_if, m_latency_tx_queue_id, 1); + m_latency_vports[i].Create(_if, m_rx_core_tx_q_id, 1); rx_sl_cfg.m_ports[i] = &m_latency_vports[i]; } } @@ -3541,88 +3567,53 @@ void CGlobalTRex::rx_sl_configure(void) { int CGlobalTRex::ixgbe_start(void){ int i; for (i=0; i<m_max_ports; i++) { - + socket_id_t socket_id = CGlobalInfo::m_socket.port_to_socket((port_id_t)i); + assert(CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_2048); CPhyEthIF * _if=&m_ports[i]; _if->Create((uint8_t)i); - /* last TX queue if for latency check */ - if ( get_vm_one_queue_enable() ) { - /* one tx one rx */ + if ( get_vm_one_queue_enable() ) { /* VMXNET3 does claim to support 16K but somehow does not work */ /* reduce to 2000 */ m_port_cfg.m_port_conf.rxmode.max_rx_pkt_len = 2000; - - _if->configure(1, - 1, - &m_port_cfg.m_port_conf); - - /* will not be used */ - m_latency_tx_queue_id= m_cores_to_dual_ports; - - socket_id_t socket_id = CGlobalInfo::m_socket.port_to_socket((port_id_t)i); - assert(CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_2048); - - - + /* In VM case, there is one tx q and one rx q */ + _if->configure(1, 1, &m_port_cfg.m_port_conf); + // Only 1 rx queue, so use it for everything + m_rx_core_tx_q_id = 0; _if->set_rx_queue(0); - _if->rx_queue_setup(0, - RTE_TEST_RX_DESC_VM_DEFAULT, - socket_id, - &m_port_cfg.m_rx_conf, + _if->rx_queue_setup(0, RTE_TEST_RX_DESC_VM_DEFAULT, socket_id, &m_port_cfg.m_rx_conf, CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_2048); + // 1 TX queue in VM case + _if->tx_queue_setup(0, RTE_TEST_TX_DESC_VM_DEFAULT, socket_id, &m_port_cfg.m_tx_conf); + } else { + // 2 rx queues. + // TX queues: 1 for each core handling the port pair + 1 for latency pkts + 1 for use by RX core + _if->configure(2, m_cores_to_dual_ports + 2, &m_port_cfg.m_port_conf); + m_rx_core_tx_q_id = m_cores_to_dual_ports; - int qid; - for ( qid=0; qid<(m_max_queues_per_port); qid++) { - _if->tx_queue_setup((uint16_t)qid, - RTE_TEST_TX_DESC_VM_DEFAULT , - socket_id, - &m_port_cfg.m_tx_conf); - - } - - }else{ - _if->configure(2, - m_cores_to_dual_ports+1, - &m_port_cfg.m_port_conf); - - /* the latency queue for latency measurement packets */ - m_latency_tx_queue_id= m_cores_to_dual_ports; - - socket_id_t socket_id = CGlobalInfo::m_socket.port_to_socket((port_id_t)i); - assert(CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_2048); - - - /* drop queue */ - _if->rx_queue_setup(0, + // setup RX drop queue + _if->rx_queue_setup(MAIN_DPDK_DATA_Q, RTE_TEST_RX_DESC_DEFAULT, socket_id, &m_port_cfg.m_rx_conf, CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_2048); - - - /* set the filter queue */ - _if->set_rx_queue(1); - /* latency measurement ring is 1 */ - _if->rx_queue_setup(1, + // setup RX filter queue + _if->set_rx_queue(MAIN_DPDK_RX_Q); + _if->rx_queue_setup(MAIN_DPDK_RX_Q, RTE_TEST_RX_LATENCY_DESC_DEFAULT, socket_id, &m_port_cfg.m_rx_conf, CGlobalInfo::m_mem_pool[socket_id].m_mbuf_pool_9k); - - int qid; - for ( qid=0; qid<(m_max_queues_per_port+1); qid++) { + // setup TX queues + for (int qid = 0; qid < m_max_queues_per_port; qid++) { _if->tx_queue_setup((uint16_t)qid, RTE_TEST_TX_DESC_DEFAULT , socket_id, &m_port_cfg.m_tx_conf); - } - } - _if->stats_clear(); - _if->start(); _if->configure_rx_duplicate_rules(); @@ -3647,8 +3638,7 @@ int CGlobalTRex::ixgbe_start(void){ printf(" WARNING : there is no link on one of the ports, driver support auto drop in case of link down - continue\n"); }else{ dump_links_status(stdout); - rte_exit(EXIT_FAILURE, " " - " one of the link is down \n"); + rte_exit(EXIT_FAILURE, " One of the links is down \n"); } } } else { @@ -3679,7 +3669,7 @@ int CGlobalTRex::ixgbe_start(void){ if ( get_vm_one_queue_enable() ) { lat_q_id = 0; } else { - lat_q_id = get_cores_tx() / get_base_num_cores(); + lat_q_id = get_cores_tx() / get_base_num_cores() + 1; } for (i=0; i<get_cores_tx(); i++) { int j=(i+1); @@ -3704,6 +3694,7 @@ int CGlobalTRex::ixgbe_start(void){ } fprintf(stdout," -------------------------------\n"); + fprintf(stdout, "RX core uses TX queue number %d on all ports\n", m_rx_core_tx_q_id); CCoreEthIF::DumpIfCfgHeader(stdout); for (i=0; i<get_cores_tx(); i++) { m_cores_vif[i+1]->DumpIfCfg(stdout); @@ -3807,7 +3798,14 @@ bool CGlobalTRex::Create(){ } void CGlobalTRex::Delete(){ + m_zmq_publisher.Delete(); + m_fl.Delete(); + + if (m_trex_stateless) { + delete m_trex_stateless; + m_trex_stateless = NULL; + } } @@ -3930,12 +3928,12 @@ int CGlobalTRex::queues_prob_init(){ m_cores_to_dual_ports = 2; */ - /* number of queue - 1 per core for dual ports*/ - m_max_queues_per_port = m_cores_to_dual_ports; + // One q for each core allowed to send on this port + 1 for latency q (Used in stateless) + 1 for RX core. + m_max_queues_per_port = m_cores_to_dual_ports + 2; if (m_max_queues_per_port > BP_MAX_TX_QUEUE) { rte_exit(EXIT_FAILURE, - "maximum number of queue should be maximum %d \n",BP_MAX_TX_QUEUE); + "Error: Number of TX queues exceeds %d. Try running with lower -c <val> \n",BP_MAX_TX_QUEUE); } assert(m_max_queues_per_port>0); @@ -4405,18 +4403,9 @@ CGlobalTRex:: publish_async_port_attr_changed(uint8_t port_id) { Json::Value data; data["port_id"] = port_id; TRexPortAttr * _attr = m_ports[port_id].get_port_attr(); - - /* attributes */ - data["attr"]["speed"] = _attr->get_link_speed(); - data["attr"]["promiscuous"]["enabled"] = _attr->get_promiscuous(); - data["attr"]["link"]["up"] = _attr->is_link_up(); - int mode; - int ret = _attr->get_flow_ctrl(mode); - if (ret != 0) { - mode = -1; - } - data["attr"]["fc"]["mode"] = mode; - + + _attr->to_json(data["attr"]); + m_zmq_publisher.publish_event(TrexPublisher::EVENT_PORT_ATTR_CHANGED, data); } @@ -4617,8 +4606,11 @@ void CGlobalTRex::shutdown() { for (int i = 0; i < m_max_ports; i++) { m_ports[i].stop(); } + if (m_mark_for_shutdown != SHUTDOWN_TEST_ENDED) { /* we should stop latency and exit to stop agents */ + Delete(); + utl_termio_reset(); exit(-1); } } @@ -4782,7 +4774,6 @@ int CGlobalTRex::stop_master(){ dump_stats(stdout,CGlobalStats::dmpSTANDARD); dump_post_test_stats(stdout); - m_fl.Delete(); return (0); } @@ -4922,6 +4913,18 @@ bool CPhyEthIF::Create(uint8_t portid) { m_last_tx_pps = 0.0; m_port_attr = g_trex.m_drv->create_port_attr(portid); + + uint32_t src_ipv4 = CGlobalInfo::m_options.m_ip_cfg[m_port_id].get_ip(); + if (src_ipv4) { + m_port_attr->set_src_ipv4(src_ipv4); + } + + /* for now set as unresolved IPv4 destination */ + uint32_t dest_ipv4 = CGlobalInfo::m_options.m_ip_cfg[m_port_id].get_def_gw(); + if (dest_ipv4) { + m_port_attr->get_dest().set_dest(dest_ipv4); + } + return true; } @@ -5623,7 +5626,7 @@ int main_test(int argc , char * argv[]){ g_trex.stop_master(); g_trex.Delete(); utl_termio_reset(); - + return (0); } @@ -6001,6 +6004,7 @@ void CTRexExtendedDriverBase10G::update_configuration(port_cfg_t * cfg){ } int CTRexExtendedDriverBase10G::configure_rx_filter_rules(CPhyEthIF * _if) { + set_rcv_all(_if, false); if ( get_is_stateless() ) { return configure_rx_filter_rules_stateless(_if); } else { @@ -6031,7 +6035,7 @@ int CTRexExtendedDriverBase10G::configure_rx_filter_rules_stateless(CPhyEthIF * res = rte_eth_dev_filter_ctrl(port_id, RTE_ETH_FILTER_FDIR, RTE_ETH_FILTER_ADD, &fdir_filter); if (res != 0) { - rte_exit(EXIT_FAILURE, " ERROR rte_eth_dev_filter_ctrl : %d\n",res); + rte_exit(EXIT_FAILURE, "Error: rte_eth_dev_filter_ctrl in configure_rx_filter_rules_stateless: %d\n",res); } } @@ -6100,12 +6104,51 @@ int CTRexExtendedDriverBase10G::configure_rx_filter_rules_statefull(CPhyEthIF * res = rte_eth_dev_filter_ctrl(port_id, RTE_ETH_FILTER_FDIR, RTE_ETH_FILTER_ADD, &fdir_filter); if (res != 0) { - rte_exit(EXIT_FAILURE, " ERROR rte_eth_dev_filter_ctrl : %d\n",res); + rte_exit(EXIT_FAILURE, "Error: rte_eth_dev_filter_ctrl in configure_rx_filter_rules_statefull: %d\n",res); } } return (0); } +int CTRexExtendedDriverBase10G::add_del_eth_filter(CPhyEthIF * _if, bool is_add, uint16_t ethertype) { + int res = 0; + uint8_t port_id=_if->get_rte_port_id(); + struct rte_eth_ethertype_filter filter; + enum rte_filter_op op; + + memset(&filter, 0, sizeof(filter)); + filter.ether_type = ethertype; + res = rte_eth_dev_filter_ctrl(port_id, RTE_ETH_FILTER_ETHERTYPE, RTE_ETH_FILTER_GET, &filter); + + if (is_add && (res >= 0)) + return 0; + if ((! is_add) && (res == -ENOENT)) + return 0; + + if (is_add) { + op = RTE_ETH_FILTER_ADD; + } else { + op = RTE_ETH_FILTER_DELETE; + } + + filter.queue = 1; + res = rte_eth_dev_filter_ctrl(port_id, RTE_ETH_FILTER_ETHERTYPE, op, &filter); + if (res != 0) { + printf("Error: %s L2 filter for ethertype 0x%04x returned %d\n", is_add ? "Adding":"Deleting", ethertype, res); + exit(1); + } + return 0; +} + +int CTRexExtendedDriverBase10G::set_rcv_all(CPhyEthIF * _if, bool set_on) { + int res = 0; + res = add_del_eth_filter(_if, set_on, ETHER_TYPE_ARP); + res |= add_del_eth_filter(_if, set_on, ETHER_TYPE_IPv4); + res |= add_del_eth_filter(_if, set_on, ETHER_TYPE_IPv6); + + return res; +} + void CTRexExtendedDriverBase10G::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats){ int i; @@ -6852,26 +6895,58 @@ int CTRexExtendedDriverBaseVIC::configure_rx_filter_rules_statefull(CPhyEthIF * return 0; } -extern "C" int enicpmd_dev_get_fw_support(int port_id, - uint32_t *ver); + +void CTRexExtendedDriverBaseVIC::clear_extended_stats(CPhyEthIF * _if){ + rte_eth_stats_reset(_if->get_port_id()); +} + +void CTRexExtendedDriverBaseVIC::get_extended_stats(CPhyEthIF * _if,CPhyEthIFStats *stats) { + struct rte_eth_stats stats1; + struct rte_eth_stats *prev_stats = &stats->m_prev_stats; + rte_eth_stats_get(_if->get_port_id(), &stats1); + + stats->ipackets += stats1.ipackets - prev_stats->ipackets; + stats->ibytes += stats1.ibytes - prev_stats->ibytes + - ((stats1.ipackets << 2) - (prev_stats->ipackets << 2)); + stats->opackets += stats1.opackets - prev_stats->opackets; + stats->obytes += stats1.obytes - prev_stats->obytes; + stats->f_ipackets += 0; + stats->f_ibytes += 0; + stats->ierrors += stats1.imissed + stats1.ierrors + stats1.rx_nombuf + - prev_stats->imissed - prev_stats->ierrors - prev_stats->rx_nombuf; + stats->oerrors += stats1.oerrors - prev_stats->oerrors; + stats->imcasts += 0; + stats->rx_nombuf += stats1.rx_nombuf - prev_stats->rx_nombuf; + + prev_stats->ipackets = stats1.ipackets; + prev_stats->ibytes = stats1.ibytes; + prev_stats->opackets = stats1.opackets; + prev_stats->obytes = stats1.obytes; + prev_stats->imissed = stats1.imissed; + prev_stats->oerrors = stats1.oerrors; + prev_stats->ierrors = stats1.ierrors; + prev_stats->rx_nombuf = stats1.rx_nombuf; +} int CTRexExtendedDriverBaseVIC::verify_fw_ver(int port_id) { - uint32_t ver; - int ret=enicpmd_dev_get_fw_support(port_id,&ver); + struct rte_eth_fdir_info fdir_info; - if (ret==0) { - if (CGlobalInfo::m_options.preview.getVMode() >= 1) { - printf("VIC port %d: FW support advanced filtering \n", port_id); + if ( rte_eth_dev_filter_ctrl(port_id,RTE_ETH_FILTER_FDIR, RTE_ETH_FILTER_INFO,(void *)&fdir_info) == 0 ){ + if ( fdir_info.flow_types_mask[0] & (1<< RTE_ETH_FLOW_NONFRAG_IPV4_OTHER) ) { + /* support new features */ + if (CGlobalInfo::m_options.preview.getVMode() >= 1) { + printf("VIC port %d: FW support advanced filtering \n", port_id); + } + return (0); } - }else{ - printf("Error: VIC firmware should upgrade to support advanced filtering \n"); - printf(" Please refer to %s for upgrade instructions\n", - "https://trex-tgn.cisco.com/trex/doc/trex_manual.html"); - exit(1); } - return (0); + + printf("Error: VIC firmware should upgrade to support advanced filtering \n"); + printf(" Please refer to %s for upgrade instructions\n", + "https://trex-tgn.cisco.com/trex/doc/trex_manual.html"); + exit(1); } @@ -7110,19 +7185,10 @@ TrexDpdkPlatformApi::get_interface_info(uint8_t interface_id, intf_info_st &info /* mac INFO */ /* hardware */ - g_trex.m_ports[interface_id].get_port_attr()->macaddr_get(&rte_mac_addr); + g_trex.m_ports[interface_id].get_port_attr()->get_hw_src_mac(&rte_mac_addr); assert(ETHER_ADDR_LEN == 6); - /* software */ - uint8_t sw_macaddr[12]; - memcpy(sw_macaddr, CGlobalInfo::m_options.get_dst_src_mac_addr(interface_id), 12); - - for (int i = 0; i < 6; i++) { - info.mac_info.hw_macaddr[i] = rte_mac_addr.addr_bytes[i]; - info.mac_info.dst_macaddr[i] = sw_macaddr[i]; - info.mac_info.src_macaddr[i] = sw_macaddr[6 + i]; - - } + memcpy(info.hw_macaddr, rte_mac_addr.addr_bytes, 6); info.numa_node = g_trex.m_ports[interface_id].m_dev_info.pci_dev->numa_node; struct rte_pci_addr *loc = &g_trex.m_ports[interface_id].m_dev_info.pci_dev->addr; @@ -7228,6 +7294,25 @@ TRexPortAttr *TrexDpdkPlatformApi::getPortAttrObj(uint8_t port_id) const { return g_trex.m_ports[port_id].get_port_attr(); } + +int DpdkTRexPortAttr::set_rx_filter_mode(rx_filter_mode_e rx_filter_mode) { + + if (rx_filter_mode == m_rx_filter_mode) { + return (0); + } + + CPhyEthIF *_if = &g_trex.m_ports[m_port_id]; + bool recv_all = (rx_filter_mode == RX_FILTER_MODE_ALL); + int rc = CTRexExtendedDriverDb::Ins()->get_drv()->set_rcv_all(_if, recv_all); + if (rc != 0) { + return (rc); + } + + m_rx_filter_mode = rx_filter_mode; + + return (0); +} + /** * marks the control plane for a total server shutdown * @@ -7236,3 +7321,6 @@ TRexPortAttr *TrexDpdkPlatformApi::getPortAttrObj(uint8_t port_id) const { void TrexDpdkPlatformApi::mark_for_shutdown() const { g_trex.mark_for_shutdown(CGlobalTRex::SHUTDOWN_RPC_REQ); } + + + diff --git a/src/pkt_gen.cpp b/src/pkt_gen.cpp index 45e3a298..656b1b06 100644 --- a/src/pkt_gen.cpp +++ b/src/pkt_gen.cpp @@ -222,7 +222,7 @@ char *CTestPktGen::create_test_pkt(uint16_t l3_type, uint16_t l4_proto, uint8_t switch(l3_type) { case EthernetHeader::Protocol::IP: ip->setTimeToLive(ttl); - if (ttl==TTL_RESERVE_DUPLICATE || ttl==(TTL_RESERVE_DUPLICATE-1)) { + if (flags & DPF_TOS_1) { ip->setTOS(TOS_TTL_RESERVE_DUPLICATE); }else{ ip->setTOS(0x2); @@ -232,11 +232,11 @@ char *CTestPktGen::create_test_pkt(uint16_t l3_type, uint16_t l4_proto, uint8_t break; case EthernetHeader::Protocol::IPv6: ipv6->setHopLimit(ttl); - if (ttl==TTL_RESERVE_DUPLICATE || ttl==(TTL_RESERVE_DUPLICATE-1)) { + if (flags & DPF_TOS_1) { ipv6->setTrafficClass(TOS_TTL_RESERVE_DUPLICATE); }else{ ipv6->setTrafficClass(0x2); - } + } break; } diff --git a/src/pkt_gen.h b/src/pkt_gen.h index 309e02b9..8dcba624 100644 --- a/src/pkt_gen.h +++ b/src/pkt_gen.h @@ -37,7 +37,8 @@ enum { enum { DPF_VLAN = 0x1, DPF_QINQ = 0X2, - DPF_RXCHECK = 0x4 + DPF_RXCHECK = 0x4, + DPF_TOS_1 = 0x8, }; class CTestPktGen { diff --git a/src/pre_test.cpp b/src/pre_test.cpp index df753be5..7127645d 100644 --- a/src/pre_test.cpp +++ b/src/pre_test.cpp @@ -35,6 +35,15 @@ CPretestOnePortInfo::CPretestOnePortInfo() { m_stats.clear(); } +CPretestOnePortInfo::~CPretestOnePortInfo() { + for (std::vector<COneIPInfo *>::iterator it = m_src_info.begin(); it != m_src_info.end(); ++it) { + delete *it; + } + for (std::vector<COneIPInfo *>::iterator it = m_dst_info.begin(); it != m_dst_info.end(); ++it) { + delete *it; + } +} + void CPretestOnePortInfo::add_src(uint32_t ip, uint16_t vlan, MacAddress mac) { COneIPv4Info *one_ip = new COneIPv4Info(ip, vlan, mac); assert(one_ip); diff --git a/src/pre_test.h b/src/pre_test.h index 9573ff02..14b444cf 100644 --- a/src/pre_test.h +++ b/src/pre_test.h @@ -49,6 +49,7 @@ class CPretestOnePortInfo { public: CPretestOnePortInfo(); + ~CPretestOnePortInfo(); void add_src(uint32_t ip, uint16_t vlan, MacAddress mac); void add_dst(uint32_t ip, uint16_t vlan); void add_src(uint16_t ip[8], uint16_t vlan, MacAddress mac); diff --git a/src/rpc-server/commands/trex_rpc_cmd_general.cpp b/src/rpc-server/commands/trex_rpc_cmd_general.cpp index 109cc1a4..3d541fe5 100644 --- a/src/rpc-server/commands/trex_rpc_cmd_general.cpp +++ b/src/rpc-server/commands/trex_rpc_cmd_general.cpp @@ -27,6 +27,8 @@ limitations under the License. #include <internal_api/trex_platform_api.h> +#include "trex_stateless_rx_core.h" + #include <fstream> #include <iostream> #include <unistd.h> @@ -289,19 +291,15 @@ TrexRpcCmdGetSysInfo::_run(const Json::Value ¶ms, Json::Value &result) { section["ports"] = Json::arrayValue; for (int i = 0; i < main->get_port_count(); i++) { - uint32_t speed; string driver; - string hw_macaddr; - string src_macaddr; - string dst_macaddr; string pci_addr; string description; supp_speeds_t supp_speeds; int numa; TrexStatelessPort *port = main->get_port_by_id(i); - port->get_properties(driver, speed); - port->get_macaddr(hw_macaddr, src_macaddr, dst_macaddr); + + port->get_properties(driver); port->get_pci_info(pci_addr, numa); main->get_platform_api()->getPortAttrObj(i)->get_description(description); @@ -311,9 +309,6 @@ TrexRpcCmdGetSysInfo::_run(const Json::Value ¶ms, Json::Value &result) { section["ports"][i]["driver"] = driver; section["ports"][i]["description"] = description; - section["ports"][i]["hw_macaddr"] = hw_macaddr; - section["ports"][i]["src_macaddr"] = src_macaddr; - section["ports"][i]["dst_macaddr"] = dst_macaddr; section["ports"][i]["pci_addr"] = pci_addr; section["ports"][i]["numa"] = numa; @@ -330,7 +325,6 @@ TrexRpcCmdGetSysInfo::_run(const Json::Value ¶ms, Json::Value &result) { section["ports"][i]["rx"]["caps"].append("rx_bytes"); } section["ports"][i]["rx"]["counters"] = port->get_rx_count_num(); - section["ports"][i]["speed"] = (uint16_t) speed / 1000; section["ports"][i]["is_fc_supported"] = get_stateless_obj()->get_platform_api()->getPortAttrObj(i)->is_fc_change_supported(); section["ports"][i]["is_led_supported"] = get_stateless_obj()->get_platform_api()->getPortAttrObj(i)->is_led_change_supported(); section["ports"][i]["is_link_supported"] = get_stateless_obj()->get_platform_api()->getPortAttrObj(i)->is_link_change_supported(); @@ -345,6 +339,69 @@ TrexRpcCmdGetSysInfo::_run(const Json::Value ¶ms, Json::Value &result) { return (TREX_RPC_CMD_OK); } + +int +TrexRpcCmdSetPortAttr::parse_rx_filter_mode(const Json::Value &msg, uint8_t port_id, Json::Value &result) { + const std::string type = parse_choice(msg, "mode", {"hw", "all"}, result); + + rx_filter_mode_e filter_mode; + if (type == "hw") { + filter_mode = RX_FILTER_MODE_HW; + } else if (type == "all") { + filter_mode = RX_FILTER_MODE_ALL; + } else { + /* can't happen - parsed choice */ + assert(0); + } + + return get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->set_rx_filter_mode(filter_mode); +} + +int +TrexRpcCmdSetPortAttr::parse_ipv4(const Json::Value &msg, uint8_t port_id, Json::Value &result) { + + const std::string ipv4_str = parse_string(msg, "addr", result); + + uint32_t ipv4_addr; + if (!utl_ipv4_to_uint32(ipv4_str.c_str(), ipv4_addr)) { + std::stringstream ss; + ss << "invalid IPv4 address: '" << ipv4_str << "'"; + generate_parse_err(result, ss.str()); + } + + get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->set_src_ipv4(ipv4_addr); + return (0); +} + +int +TrexRpcCmdSetPortAttr::parse_dest(const Json::Value &msg, uint8_t port_id, Json::Value &result) { + + /* can be either IPv4 or MAC */ + const std::string addr = parse_string(msg, "addr", result); + + TRexPortAttr *port_attr = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id); + + /* try IPv4 */ + uint32_t ipv4_addr; + uint8_t mac[6]; + + if (utl_ipv4_to_uint32(addr.c_str(), ipv4_addr)) { + port_attr->get_dest().set_dest(ipv4_addr); + + } else if (utl_str_to_macaddr(addr, mac)) { + port_attr->get_dest().set_dest(mac); + + } else { + std::stringstream ss; + ss << "'dest' is not an IPv4 address or a MAC address: '" << addr << "'"; + generate_parse_err(result, ss.str()); + } + + + return (0); +} + + /** * set port commands * @@ -361,46 +418,64 @@ TrexRpcCmdSetPortAttr::_run(const Json::Value ¶ms, Json::Value &result) { uint8_t port_id = parse_port(params, result); const Json::Value &attr = parse_object(params, "attr", result); + int ret = 0; - bool changed = false; + /* iterate over all attributes in the dict */ for (const std::string &name : attr.getMemberNames()) { + if (name == "promiscuous") { bool enabled = parse_bool(attr[name], "enabled", result); ret = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->set_promiscuous(enabled); } + else if (name == "link_status") { bool up = parse_bool(attr[name], "up", result); ret = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->set_link_up(up); } + else if (name == "led_status") { bool on = parse_bool(attr[name], "on", result); ret = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->set_led(on); - } else if (name == "flow_ctrl_mode") { + } + + else if (name == "flow_ctrl_mode") { int mode = parse_int(attr[name], "mode", result); ret = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->set_flow_ctrl(mode); - } else { - generate_execute_err(result, "Not recognized attribute: " + name); - break; } - if (ret != 0){ - if ( ret == -ENOTSUP ) { - generate_execute_err(result, "Error applying " + name + ": operation is not supported for this NIC."); - } - else if (ret) { - generate_execute_err(result, "Error applying " + name + " attribute, return value: " + to_string(ret)); - } + + else if (name == "rx_filter_mode") { + const Json::Value &rx = parse_object(attr, name, result); + ret = parse_rx_filter_mode(rx, port_id, result); + } + + else if (name == "ipv4") { + const Json::Value &ipv4 = parse_object(attr, name, result); + ret = parse_ipv4(ipv4, port_id, result); + } + + else if (name == "dest") { + const Json::Value &dest = parse_object(attr, name, result); + ret = parse_dest(dest, port_id, result); + } + + /* unknown attribute */ + else { + generate_execute_err(result, "unknown attribute type: '" + name + "'"); break; - } else { - changed = true; } - } - if (changed) { - get_stateless_obj()->get_platform_api()->publish_async_port_attr_changed(port_id); - } + /* check error code */ + if ( ret == -ENOTSUP ) { + generate_execute_err(result, "Error applying " + name + ": operation is not supported for this NIC."); + } else if (ret) { + generate_execute_err(result, "Error applying " + name + " attribute, return value: " + to_string(ret)); + } + } + result["result"] = Json::objectValue; return (TREX_RPC_CMD_OK); + } @@ -568,18 +643,17 @@ TrexRpcCmdGetPortStatus::_run(const Json::Value ¶ms, Json::Value &result) { result["result"]["owner"] = (port->get_owner().is_free() ? "" : port->get_owner().get_name()); result["result"]["state"] = port->get_state_as_string(); result["result"]["max_stream_id"] = port->get_max_stream_id(); - result["result"]["speed"] = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->get_link_speed(); /* attributes */ - result["result"]["attr"]["promiscuous"]["enabled"] = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->get_promiscuous(); - result["result"]["attr"]["link"]["up"] = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->is_link_up(); - int mode; - int ret = get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->get_flow_ctrl(mode); - if (ret != 0) { - mode = -1; + get_stateless_obj()->get_platform_api()->getPortAttrObj(port_id)->to_json(result["result"]["attr"]); + + /* RX info */ + try { + result["result"]["rx_info"] = port->rx_features_to_json(); + } catch (const TrexException &ex) { + generate_execute_err(result, ex.what()); } - result["result"]["attr"]["fc"]["mode"] = mode; - + return (TREX_RPC_CMD_OK); } @@ -640,3 +714,151 @@ TrexRpcCmdPushRemote::_run(const Json::Value ¶ms, Json::Value &result) { } +/** + * set on/off RX software receive mode + * + */ +trex_rpc_cmd_rc_e +TrexRpcCmdSetRxFeature::_run(const Json::Value ¶ms, Json::Value &result) { + + uint8_t port_id = parse_port(params, result); + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); + + /* decide which feature is being set */ + const std::string type = parse_choice(params, "type", {"capture", "queue", "server"}, result); + + if (type == "capture") { + parse_capture_msg(params, port, result); + } else if (type == "queue") { + parse_queue_msg(params, port, result); + } else if (type == "server") { + parse_server_msg(params, port, result); + } else { + assert(0); + } + + result["result"] = Json::objectValue; + return (TREX_RPC_CMD_OK); + +} + +void +TrexRpcCmdSetRxFeature::parse_capture_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result) { + + bool enabled = parse_bool(msg, "enabled", result); + + if (enabled) { + + std::string pcap_filename = parse_string(msg, "pcap_filename", result); + uint64_t limit = parse_uint32(msg, "limit", result); + + if (limit == 0) { + generate_parse_err(result, "limit cannot be zero"); + } + + try { + port->start_rx_capture(pcap_filename, limit); + } catch (const TrexException &ex) { + generate_execute_err(result, ex.what()); + } + + } else { + + try { + port->stop_rx_capture(); + } catch (const TrexException &ex) { + generate_execute_err(result, ex.what()); + } + + } + +} + +void +TrexRpcCmdSetRxFeature::parse_queue_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result) { + bool enabled = parse_bool(msg, "enabled", result); + + if (enabled) { + + uint64_t size = parse_uint32(msg, "size", result); + + if (size == 0) { + generate_parse_err(result, "queue size cannot be zero"); + } + + try { + port->start_rx_queue(size); + } catch (const TrexException &ex) { + generate_execute_err(result, ex.what()); + } + + } else { + + try { + port->stop_rx_queue(); + } catch (const TrexException &ex) { + generate_execute_err(result, ex.what()); + } + + } + +} + +void +TrexRpcCmdSetRxFeature::parse_server_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result) { +} + + +trex_rpc_cmd_rc_e +TrexRpcCmdGetRxQueuePkts::_run(const Json::Value ¶ms, Json::Value &result) { + + uint8_t port_id = parse_port(params, result); + + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); + + try { + const RXPacketBuffer *pkt_buffer = port->get_rx_queue_pkts(); + if (pkt_buffer) { + result["result"]["pkts"] = pkt_buffer->to_json(); + delete pkt_buffer; + + } else { + result["result"]["pkts"] = Json::arrayValue; + } + + } catch (const TrexException &ex) { + generate_execute_err(result, ex.what()); + } + + + return (TREX_RPC_CMD_OK); +} + +trex_rpc_cmd_rc_e +TrexRpcCmdSetARPRes::_run(const Json::Value ¶ms, Json::Value &result) { + uint8_t port_id = parse_port(params, result); + + TrexStatelessPort *port = get_stateless_obj()->get_port_by_id(port_id); + + const std::string ipv4_str = parse_string(params, "ipv4", result); + const std::string mac_str = parse_string(params, "mac", result); + + uint32_t ipv4_addr; + if (!utl_ipv4_to_uint32(ipv4_str.c_str(), ipv4_addr)) { + std::stringstream ss; + ss << "invalid IPv4 address: '" << ipv4_str << "'"; + generate_parse_err(result, ss.str()); + } + + uint8_t mac[6]; + if (!utl_str_to_macaddr(mac_str, mac)) { + std::stringstream ss; + ss << "'invalid MAC address: '" << mac_str << "'"; + generate_parse_err(result, ss.str()); + } + + port->getPortAttrObj()->get_dest().set_dest(ipv4_addr, mac); + + 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 c950e011..9a57c5f9 100644 --- a/src/rpc-server/commands/trex_rpc_cmd_stream.cpp +++ b/src/rpc-server/commands/trex_rpc_cmd_stream.cpp @@ -670,6 +670,7 @@ TrexRpcCmdStartTraffic::_run(const Json::Value ¶ms, Json::Value &result) { generate_parse_err(result, "start message can only specify absolute speed rate"); } + dsec_t ts = now_sec(); TrexPortMultiplier mul(type, op, value); try { @@ -680,7 +681,8 @@ TrexRpcCmdStartTraffic::_run(const Json::Value ¶ms, Json::Value &result) { } result["result"]["multiplier"] = port->get_multiplier(); - + result["result"]["ts"] = ts; + 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 5fde1d0c..2b2178e2 100644 --- a/src/rpc-server/commands/trex_rpc_cmds.h +++ b/src/rpc-server/commands/trex_rpc_cmds.h @@ -27,6 +27,7 @@ limitations under the License. #include <memory> class TrexStream; +class TrexStatelessPort; /* all the RPC commands decl. goes here */ @@ -89,10 +90,17 @@ TREX_RPC_CMD_DEFINE(TrexRpcCmdRelease, "release", 1, true, APIClass: */ TREX_RPC_CMD_DEFINE(TrexRpcCmdGetPortStats, "get_port_stats", 1, false, APIClass::API_CLASS_TYPE_CORE); TREX_RPC_CMD_DEFINE(TrexRpcCmdGetPortStatus, "get_port_status", 1, false, APIClass::API_CLASS_TYPE_CORE); -TREX_RPC_CMD_DEFINE(TrexRpcCmdSetPortAttr, "set_port_attr", 2, true, APIClass::API_CLASS_TYPE_CORE); TREX_RPC_CMD_DEFINE(TrexRpcCmdGetPortXStatsValues, "get_port_xstats_values", 1, false, APIClass::API_CLASS_TYPE_CORE); TREX_RPC_CMD_DEFINE(TrexRpcCmdGetPortXStatsNames, "get_port_xstats_names", 1, false, APIClass::API_CLASS_TYPE_CORE); +TREX_RPC_CMD_DEFINE_EXTENDED(TrexRpcCmdSetPortAttr, "set_port_attr", 2, true, APIClass::API_CLASS_TYPE_CORE, + + int parse_rx_filter_mode(const Json::Value &msg, uint8_t port_id, Json::Value &result); + int parse_ipv4(const Json::Value &msg, uint8_t port_id, Json::Value &result); + int parse_dest(const Json::Value &msg, uint8_t port_id, Json::Value &result); +); + + /** * stream cmds */ @@ -144,5 +152,15 @@ TREX_RPC_CMD_DEFINE(TrexRpcCmdPushRemote, "push_remote", 6, true, APIClass::API_ TREX_RPC_CMD_DEFINE(TrexRpcCmdShutdown, "shutdown", 2, false, APIClass::API_CLASS_TYPE_CORE); +TREX_RPC_CMD_DEFINE_EXTENDED(TrexRpcCmdSetRxFeature, "set_rx_feature", 3, false, APIClass::API_CLASS_TYPE_CORE, + void parse_capture_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result); + void parse_queue_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result); + void parse_server_msg(const Json::Value &msg, TrexStatelessPort *port, Json::Value &result); + +); + +TREX_RPC_CMD_DEFINE(TrexRpcCmdGetRxQueuePkts, "get_rx_queue_pkts", 2, false, APIClass::API_CLASS_TYPE_CORE); +TREX_RPC_CMD_DEFINE(TrexRpcCmdSetARPRes, "set_arp_resolution", 2, false, APIClass::API_CLASS_TYPE_CORE); + #endif /* __TREX_RPC_CMD_H__ */ diff --git a/src/rpc-server/trex_rpc_cmds_table.cpp b/src/rpc-server/trex_rpc_cmds_table.cpp index cddf19b9..919be1f1 100644 --- a/src/rpc-server/trex_rpc_cmds_table.cpp +++ b/src/rpc-server/trex_rpc_cmds_table.cpp @@ -71,6 +71,11 @@ TrexRpcCommandsTable::TrexRpcCommandsTable() { register_command(new TrexRpcCmdPushRemote()); register_command(new TrexRpcCmdShutdown()); + + register_command(new TrexRpcCmdSetRxFeature()); + register_command(new TrexRpcCmdGetRxQueuePkts()); + + register_command(new TrexRpcCmdSetARPRes()); } diff --git a/src/stateful_rx_core.cpp b/src/stateful_rx_core.cpp index 7ee802df..65d0a17d 100644 --- a/src/stateful_rx_core.cpp +++ b/src/stateful_rx_core.cpp @@ -677,58 +677,6 @@ void CLatencyManager::handle_rx_pkt(CLatencyManagerPerPort * lp, rte_pktmbuf_free(m); } -// In VM, we receive the RX packets in DP core, and send message to RX core with the packet -void CLatencyManager::handle_latency_pkt_msg(uint8_t thread_id, CGenNodeLatencyPktInfo * msg) { - - assert(msg->m_latency_offset==0xdead); - - uint8_t rx_port_index=(thread_id<<1)+(msg->m_dir&1); - assert( rx_port_index <m_max_ports ) ; - CLatencyManagerPerPort * lp=&m_ports[rx_port_index]; - handle_rx_pkt(lp,(rte_mbuf_t *)msg->m_pkt); -} - - -void CLatencyManager::run_rx_queue_msgs(uint8_t thread_id, - CNodeRing * r){ - - while ( true ) { - CGenNode * node; - if ( r->Dequeue(node)!=0 ){ - break; - } - assert(node); - - CGenNodeMsgBase * msg=(CGenNodeMsgBase *)node; - - uint8_t msg_type = msg->m_msg_type; - switch (msg_type ) { - case CGenNodeMsgBase::LATENCY_PKT: - handle_latency_pkt_msg(thread_id,(CGenNodeLatencyPktInfo *) msg); - break; - default: - printf("ERROR latency-thread message type is not valid %d \n",msg_type); - assert(0); - } - - CGlobalInfo::free_node(node); - } -} - -// VM mode function. Handle messages from DP -void CLatencyManager::try_rx_queues(){ - - CMessagingManager * rx_dp = CMsgIns::Ins()->getRxDp(); - uint8_t threads=CMsgIns::Ins()->get_num_threads(); - int ti; - for (ti=0; ti<(int)threads; ti++) { - CNodeRing * r = rx_dp->getRingDpToCp(ti); - if ( !r->isEmpty() ){ - run_rx_queue_msgs((uint8_t)ti,r); - } - } -} - void CLatencyManager::try_rx(){ rte_mbuf_t * rx_pkts[64]; int i; @@ -790,8 +738,6 @@ void CLatencyManager::start(int iter, bool activate_watchdog) { m_p_queue.push(node); } - bool do_try_rx_queue = CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ? true : false; - if (activate_watchdog) { m_monitor.create("STF RX CORE", 1); TrexWatchDog::getInstance().register_monitor(&m_monitor); @@ -807,9 +753,6 @@ void CLatencyManager::start(int iter, bool activate_watchdog) { if (dt> (0.0)) { break; } - if (do_try_rx_queue){ - try_rx_queues(); - } try_rx(); rte_pause(); } diff --git a/src/stateful_rx_core.h b/src/stateful_rx_core.h index 48fbeb97..2df406de 100644 --- a/src/stateful_rx_core.h +++ b/src/stateful_rx_core.h @@ -355,12 +355,8 @@ private: double grat_arp_timeout(); void send_one_grat_arp(); void try_rx(); - void try_rx_queues(); - void run_rx_queue_msgs(uint8_t thread_id, CNodeRing * r); void wait_for_rx_dump(); void handle_rx_pkt(CLatencyManagerPerPort * lp, rte_mbuf_t * m); - /* messages handlers */ - void handle_latency_pkt_msg(uint8_t thread_id, CGenNodeLatencyPktInfo * msg); private: pqueue_t m_p_queue; /* priorty queue */ diff --git a/src/stateless/cp/trex_stateless_port.cpp b/src/stateless/cp/trex_stateless_port.cpp index 9bb20990..7edf1a31 100644 --- a/src/stateless/cp/trex_stateless_port.cpp +++ b/src/stateless/cp/trex_stateless_port.cpp @@ -25,6 +25,7 @@ limitations under the License. #include <trex_streams_compiler.h> #include <common/basic_utils.h> #include <common/captureFile.h> +#include "trex_stateless_rx_defs.h" #include <string> @@ -156,9 +157,9 @@ private: TrexStatelessPort::TrexStatelessPort(uint8_t port_id, const TrexPlatformApi *api) : m_dp_events(this) { std::vector<std::pair<uint8_t, uint8_t>> core_pair_list; - m_port_id = port_id; - m_port_state = PORT_STATE_IDLE; - m_platform_api = api; + m_port_id = port_id; + m_port_state = PORT_STATE_IDLE; + m_platform_api = api; /* get the platform specific data */ api->get_interface_info(port_id, m_api_info); @@ -584,10 +585,9 @@ TrexStatelessPort::get_max_stream_id() const { } void -TrexStatelessPort::get_properties(std::string &driver, uint32_t &speed) { +TrexStatelessPort::get_properties(std::string &driver) { driver = m_api_info.driver_name; - speed = m_platform_api->getPortAttrObj(m_port_id)->get_link_speed(); } bool @@ -888,16 +888,6 @@ TrexStatelessPort::get_port_effective_rate(double &pps, } void -TrexStatelessPort::get_macaddr(std::string &hw_macaddr, - std::string &src_macaddr, - std::string &dst_macaddr) { - - utl_macaddr_to_str(m_api_info.mac_info.hw_macaddr, hw_macaddr); - utl_macaddr_to_str(m_api_info.mac_info.src_macaddr, src_macaddr); - utl_macaddr_to_str(m_api_info.mac_info.dst_macaddr, dst_macaddr); -} - -void TrexStatelessPort::get_pci_info(std::string &pci_addr, int &numa_node) { pci_addr = m_api_info.pci_addr; numa_node = m_api_info.numa_node; @@ -944,6 +934,71 @@ TrexStatelessPort::remove_and_delete_all_streams() { } } +void +TrexStatelessPort::start_rx_capture(const std::string &pcap_filename, uint64_t limit) { + static MsgReply<bool> reply; + + reply.reset(); + + TrexStatelessRxStartCapture *msg = new TrexStatelessRxStartCapture(m_port_id, pcap_filename, limit, reply); + send_message_to_rx((TrexStatelessCpToRxMsgBase *)msg); + + /* as below, must wait for ACK from RX core before returning ACK */ + reply.wait_for_reply(); +} + +void +TrexStatelessPort::stop_rx_capture() { + TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxStopCapture(m_port_id); + send_message_to_rx(msg); +} + +void +TrexStatelessPort::start_rx_queue(uint64_t size) { + static MsgReply<bool> reply; + + reply.reset(); + + TrexStatelessRxStartQueue *msg = new TrexStatelessRxStartQueue(m_port_id, size, reply); + send_message_to_rx( (TrexStatelessCpToRxMsgBase *)msg ); + + /* we cannot return ACK to the user until the RX core has approved + this might cause the user to lose some packets from the queue + */ + reply.wait_for_reply(); +} + +void +TrexStatelessPort::stop_rx_queue() { + TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxStopQueue(m_port_id); + send_message_to_rx(msg); +} + + +const RXPacketBuffer * +TrexStatelessPort::get_rx_queue_pkts() { + static MsgReply<const RXPacketBuffer *> reply; + + reply.reset(); + + TrexStatelessRxQueueGetPkts *msg = new TrexStatelessRxQueueGetPkts(m_port_id, reply); + send_message_to_rx( (TrexStatelessCpToRxMsgBase *)msg ); + + return reply.wait_for_reply(); +} + +Json::Value +TrexStatelessPort::rx_features_to_json() { + static MsgReply<Json::Value> reply; + + reply.reset(); + + TrexStatelessRxFeaturesToJson *msg = new TrexStatelessRxFeaturesToJson(m_port_id, reply); + send_message_to_rx( (TrexStatelessCpToRxMsgBase *)msg ); + + return reply.wait_for_reply(); +} + /************* Trex Port Owner **************/ TrexPortOwner::TrexPortOwner() { diff --git a/src/stateless/cp/trex_stateless_port.h b/src/stateless/cp/trex_stateless_port.h index e2a2aeba..74ab17f1 100644 --- a/src/stateless/cp/trex_stateless_port.h +++ b/src/stateless/cp/trex_stateless_port.h @@ -24,12 +24,15 @@ limitations under the License. #include "common/basic_utils.h" #include "internal_api/trex_platform_api.h" #include "trex_dp_port_events.h" +#include "trex_stateless_rx_defs.h" #include "trex_stream.h" class TrexStatelessCpToDpMsgBase; class TrexStatelessCpToRxMsgBase; class TrexStreamsGraphObj; class TrexPortMultiplier; +class RXPacketBuffer; + /** * TRex port owner can perform @@ -255,11 +258,8 @@ public: * @author imarom (16-Sep-15) * * @param driver - * @param speed */ - void get_properties(std::string &driver, uint32_t &speed); - - + void get_properties(std::string &driver); /** * encode stats as JSON @@ -362,14 +362,58 @@ public: double &bps_L2, double &percentage); + void get_pci_info(std::string &pci_addr, int &numa_node); - void get_macaddr(std::string &hw_macaddr, - std::string &src_macaddr, - std::string &dst_macaddr); - void get_pci_info(std::string &pci_addr, int &numa_node); + /** + * enable RX capture on port + * + */ + void start_rx_capture(const std::string &pcap_filename, uint64_t limit); + /** + * disable RX capture if on + * + */ + void stop_rx_capture(); + + /** + * start RX queueing of packets + * + * @author imarom (11/7/2016) + * + * @param limit + */ + void start_rx_queue(uint64_t limit); + /** + * stop RX queueing + * + * @author imarom (11/7/2016) + */ + void stop_rx_queue(); + + /** + * fetch the RX queue packets from the queue + * + */ + const RXPacketBuffer *get_rx_queue_pkts(); + + /** + * generate a JSON describing the status + * of the RX features + * + */ + Json::Value rx_features_to_json(); + + /** + * return the port attribute object + * + */ + TRexPortAttr *getPortAttrObj() { + return m_platform_api->getPortAttrObj(m_port_id); + } + private: bool is_core_active(int core_id); @@ -401,7 +445,7 @@ private: * */ void send_message_to_rx(TrexStatelessCpToRxMsgBase *msg); - + /** * when a port stops, perform various actions * @@ -456,6 +500,7 @@ private: TrexPortOwner m_owner; int m_pending_async_stop_event; + }; @@ -502,9 +547,9 @@ public: static const std::initializer_list<std::string> g_types; static const std::initializer_list<std::string> g_ops; - mul_type_e m_type; - mul_op_e m_op; - double m_value; + mul_type_e m_type; + mul_op_e m_op; + double m_value; }; #endif /* __TREX_STATELESS_PORT_H__ */ diff --git a/src/stateless/messaging/trex_stateless_messaging.cpp b/src/stateless/messaging/trex_stateless_messaging.cpp index 95613b41..17acb21e 100644 --- a/src/stateless/messaging/trex_stateless_messaging.cpp +++ b/src/stateless/messaging/trex_stateless_messaging.cpp @@ -241,13 +241,15 @@ TrexDpPortEventMsg::handle() { } /************************* messages from CP to RX **********************/ -bool TrexStatelessRxStartMsg::handle (CRxCoreStateless *rx_core) { - rx_core->work(); +bool TrexStatelessRxEnableLatency::handle (CRxCoreStateless *rx_core) { + rx_core->enable_latency(); + m_reply.set_reply(true); + return true; } -bool TrexStatelessRxStopMsg::handle (CRxCoreStateless *rx_core) { - rx_core->idle(); +bool TrexStatelessRxDisableLatency::handle (CRxCoreStateless *rx_core) { + rx_core->disable_latency(); return true; } @@ -255,3 +257,62 @@ bool TrexStatelessRxQuit::handle (CRxCoreStateless *rx_core) { rx_core->quit(); return true; } + + +bool +TrexStatelessRxStartCapture::handle(CRxCoreStateless *rx_core) { + rx_core->start_recorder(m_port_id, m_pcap_filename, m_limit); + + /* mark as done */ + m_reply.set_reply(true); + + return true; +} + +bool +TrexStatelessRxStopCapture::handle(CRxCoreStateless *rx_core) { + rx_core->stop_recorder(m_port_id); + + return true; +} + +bool +TrexStatelessRxStartQueue::handle(CRxCoreStateless *rx_core) { + rx_core->start_queue(m_port_id, m_size); + + /* mark as done */ + m_reply.set_reply(true); + + return true; +} + +bool +TrexStatelessRxStopQueue::handle(CRxCoreStateless *rx_core) { + rx_core->stop_queue(m_port_id); + + return true; +} + + + +bool +TrexStatelessRxQueueGetPkts::handle(CRxCoreStateless *rx_core) { + const RXPacketBuffer *pkt_buffer = rx_core->get_rx_queue_pkts(m_port_id); + + /* set the reply */ + m_reply.set_reply(pkt_buffer); + + return true; +} + + +bool +TrexStatelessRxFeaturesToJson::handle(CRxCoreStateless *rx_core) { + Json::Value output = rx_core->get_rx_port_mngr(m_port_id).to_json(); + + /* set the reply */ + m_reply.set_reply(output); + + return true; +} + diff --git a/src/stateless/messaging/trex_stateless_messaging.h b/src/stateless/messaging/trex_stateless_messaging.h index fb2c27ab..79a6bf08 100644 --- a/src/stateless/messaging/trex_stateless_messaging.h +++ b/src/stateless/messaging/trex_stateless_messaging.h @@ -24,11 +24,66 @@ limitations under the License. #include "msg_manager.h" #include "trex_dp_port_events.h" +#include "trex_exception.h" +#include "trex_stateless_rx_defs.h" +#include "os_time.h" class TrexStatelessDpCore; class CRxCoreStateless; class TrexStreamsCompiledObj; class CFlowGenListPerThread; +class RXPacketBuffer; + +/** + * Generic message reply object + * + * @author imarom (11/27/2016) + */ +template<typename T> class MsgReply { + +public: + + MsgReply() { + reset(); + } + + void reset() { + m_pending = true; + } + + bool is_pending() const { + return m_pending; + } + + void set_reply(const T &reply) { + m_reply = reply; + + /* before marking as done make sure all stores are committed */ + asm volatile("mfence" ::: "memory"); + m_pending = false; + } + + T wait_for_reply(int timeout_ms = 500, int backoff_ms = 1) { + int guard = timeout_ms; + + while (is_pending()) { + guard -= backoff_ms; + if (guard < 0) { + throw TrexException("timeout: failed to get reply from core"); + } + + delay(backoff_ms); + } + + return m_reply; + + } + +protected: + volatile bool m_pending; + T m_reply; +}; + /** * defines the base class for CP to DP messages @@ -312,7 +367,7 @@ private: /************************* messages from DP to CP **********************/ /** - * defines the base class for CP to DP messages + * defines the base class for DP to CP messages * * @author imarom (27-Oct-15) */ @@ -404,11 +459,19 @@ public: }; -class TrexStatelessRxStartMsg : public TrexStatelessCpToRxMsgBase { + +class TrexStatelessRxEnableLatency : public TrexStatelessCpToRxMsgBase { +public: + TrexStatelessRxEnableLatency(MsgReply<bool> &reply) : m_reply(reply) { + } + bool handle (CRxCoreStateless *rx_core); + +private: + MsgReply<bool> &m_reply; }; -class TrexStatelessRxStopMsg : public TrexStatelessCpToRxMsgBase { +class TrexStatelessRxDisableLatency : public TrexStatelessCpToRxMsgBase { bool handle (CRxCoreStateless *rx_core); }; @@ -416,4 +479,114 @@ class TrexStatelessRxQuit : public TrexStatelessCpToRxMsgBase { bool handle (CRxCoreStateless *rx_core); }; + + +class TrexStatelessRxStartCapture : public TrexStatelessCpToRxMsgBase { +public: + TrexStatelessRxStartCapture(uint8_t port_id, + const std::string &pcap_filename, + uint64_t limit, + MsgReply<bool> &reply) : m_reply(reply) { + + m_port_id = port_id; + m_limit = limit; + m_pcap_filename = pcap_filename; + } + + virtual bool handle(CRxCoreStateless *rx_core); + +private: + uint8_t m_port_id; + std::string m_pcap_filename; + uint64_t m_limit; + MsgReply<bool> &m_reply; +}; + + +class TrexStatelessRxStopCapture : public TrexStatelessCpToRxMsgBase { +public: + TrexStatelessRxStopCapture(uint8_t port_id) { + m_port_id = port_id; + } + + virtual bool handle(CRxCoreStateless *rx_core); + +private: + uint8_t m_port_id; +}; + + +class TrexStatelessRxStartQueue : public TrexStatelessCpToRxMsgBase { +public: + TrexStatelessRxStartQueue(uint8_t port_id, + uint64_t size, + MsgReply<bool> &reply) : m_reply(reply) { + + m_port_id = port_id; + m_size = size; + } + + virtual bool handle(CRxCoreStateless *rx_core); + +private: + uint8_t m_port_id; + uint64_t m_size; + MsgReply<bool> &m_reply; +}; + + +class TrexStatelessRxStopQueue : public TrexStatelessCpToRxMsgBase { +public: + TrexStatelessRxStopQueue(uint8_t port_id) { + m_port_id = port_id; + } + + virtual bool handle(CRxCoreStateless *rx_core); + +private: + uint8_t m_port_id; +}; + + + +class TrexStatelessRxQueueGetPkts : public TrexStatelessCpToRxMsgBase { +public: + + TrexStatelessRxQueueGetPkts(uint8_t port_id, MsgReply<const RXPacketBuffer *> &reply) : m_reply(reply) { + m_port_id = port_id; + } + + /** + * virtual function to handle a message + * + */ + virtual bool handle(CRxCoreStateless *rx_core); + +private: + uint8_t m_port_id; + MsgReply<const RXPacketBuffer *> &m_reply; + +}; + +/** + * a request from RX core to dump to Json the RX features + */ +class TrexStatelessRxFeaturesToJson : public TrexStatelessCpToRxMsgBase { +public: + + TrexStatelessRxFeaturesToJson(uint8_t port_id, MsgReply<Json::Value> &reply) : m_reply(reply) { + m_port_id = port_id; + } + + /** + * virtual function to handle a message + * + */ + virtual bool handle(CRxCoreStateless *rx_core); + +private: + uint8_t m_port_id; + MsgReply<Json::Value> &m_reply; +}; + #endif /* __TREX_STATELESS_MESSAGING_H__ */ diff --git a/src/stateless/rx/trex_stateless_rx_core.cpp b/src/stateless/rx/trex_stateless_rx_core.cpp index d162c5b3..f2061bf7 100644 --- a/src/stateless/rx/trex_stateless_rx_core.cpp +++ b/src/stateless/rx/trex_stateless_rx_core.cpp @@ -26,6 +26,8 @@ #include "pal/linux/sanb_atomic.h" #include "trex_stateless_messaging.h" #include "trex_stateless_rx_core.h" +#include "trex_stateless.h" + void CRFC2544Info::create() { m_latency.Create(); @@ -64,15 +66,7 @@ void CRFC2544Info::export_data(rfc2544_info_t_ &obj) { obj.set_latency_json(json); }; -void CCPortLatencyStl::reset() { - for (int i = 0; i < MAX_FLOW_STATS; i++) { - m_rx_pg_stat[i].clear(); - m_rx_pg_stat_payload[i].clear(); - } -} - void CRxCoreStateless::create(const CRxSlCfg &cfg) { - m_rcv_all = false; m_capture = false; m_max_ports = cfg.m_max_ports; @@ -82,15 +76,19 @@ void CRxCoreStateless::create(const CRxSlCfg &cfg) { m_ring_to_cp = cp_rx->getRingDpToCp(0); m_state = STATE_IDLE; - for (int i = 0; i < m_max_ports; i++) { - CLatencyManagerPerPortStl * lp = &m_ports[i]; - lp->m_io = cfg.m_ports[i]; - lp->m_port.reset(); + for (int i = 0; i < MAX_FLOW_STATS_PAYLOAD; i++) { + m_rfc2544[i].create(); } + m_cpu_cp_u.Create(&m_cpu_dp_u); - for (int i = 0; i < MAX_FLOW_STATS_PAYLOAD; i++) { - m_rfc2544[i].create(); + /* create per port manager */ + for (int i = 0; i < m_max_ports; i++) { + m_rx_port_mngr[i].create(cfg.m_ports[i], + m_rfc2544, + &m_err_cntrs, + &m_cpu_dp_u, + cfg.m_num_crc_fix_bytes); } } @@ -124,10 +122,31 @@ bool CRxCoreStateless::periodic_check_for_cp_messages() { handle_cp_msg(msg); } + /* a message might result in a change of state */ + recalculate_next_state(); return true; } +void CRxCoreStateless::recalculate_next_state() { + if (m_state == STATE_QUIT) { + return; + } + + /* next state is determine by the question are there any ports with active features ? */ + m_state = (are_any_features_active() ? STATE_WORKING : STATE_IDLE); +} + +bool CRxCoreStateless::are_any_features_active() { + for (int i = 0; i < m_max_ports; i++) { + if (m_rx_port_mngr[i].has_features_set()) { + return true; + } + } + + return false; +} + void CRxCoreStateless::idle_state_loop() { const int SHORT_DELAY_MS = 2; const int LONG_DELAY_MS = 50; @@ -141,7 +160,7 @@ void CRxCoreStateless::idle_state_loop() { counter = 0; continue; } else { - flush_rx(); + flush_all_pending_pkts(); } /* enter deep sleep only if enough time had passed */ @@ -154,284 +173,88 @@ void CRxCoreStateless::idle_state_loop() { } } -void CRxCoreStateless::start() { - int count = 0; - int i = 0; - bool do_try_rx_queue =CGlobalInfo::m_options.preview.get_vm_one_queue_enable() ? true : false; - - /* register a watchdog handle on current core */ - m_monitor.create("STL RX CORE", 1); - TrexWatchDog::getInstance().register_monitor(&m_monitor); - - while (true) { - if (m_state == STATE_WORKING) { - i++; - if (i == 100000) { // approx 10msec - i = 0; - periodic_check_for_cp_messages(); // m_state might change in here - } - } else { - if (m_state == STATE_QUIT) - break; - count = 0; - i = 0; - set_working_msg_ack(false); - idle_state_loop(); - set_working_msg_ack(true); - } - if (do_try_rx_queue) { - try_rx_queues(); - } - count += try_rx(); +/** + * for each port give a tick (for flushing if needed) + * + */ +void CRxCoreStateless::port_manager_tick() { + for (int i = 0; i < m_max_ports; i++) { + m_rx_port_mngr[i].tick(); } - rte_pause(); - - m_monitor.disable(); } -void CRxCoreStateless::handle_rx_pkt(CLatencyManagerPerPortStl *lp, rte_mbuf_t *m) { - CFlowStatParser parser; - - if (m_rcv_all || parser.parse(rte_pktmbuf_mtod(m, uint8_t *), m->pkt_len) == 0) { - uint32_t ip_id; - if (m_rcv_all || (parser.get_ip_id(ip_id) == 0)) { - if (m_rcv_all || is_flow_stat_id(ip_id)) { - uint16_t hw_id; - if (m_rcv_all || is_flow_stat_payload_id(ip_id)) { - bool good_packet = true; - uint8_t *p = rte_pktmbuf_mtod(m, uint8_t*); - struct flow_stat_payload_header *fsp_head = (struct flow_stat_payload_header *) - (p + m->pkt_len - sizeof(struct flow_stat_payload_header)); - hw_id = fsp_head->hw_id; - CRFC2544Info *curr_rfc2544; - - if (unlikely(fsp_head->magic != FLOW_STAT_PAYLOAD_MAGIC) || hw_id >= MAX_FLOW_STATS_PAYLOAD) { - good_packet = false; - if (!m_rcv_all) - m_err_cntrs.m_bad_header++; - } else { - curr_rfc2544 = &m_rfc2544[hw_id]; - - if (fsp_head->flow_seq != curr_rfc2544->get_exp_flow_seq()) { - // bad flow seq num - // Might be the first packet of a new flow, packet from an old flow, or garbage. - - if (fsp_head->flow_seq == curr_rfc2544->get_prev_flow_seq()) { - // packet from previous flow using this hw_id that arrived late - good_packet = false; - m_err_cntrs.m_old_flow++; - } else { - if (curr_rfc2544->no_flow_seq()) { - // first packet we see from this flow - good_packet = true; - curr_rfc2544->set_exp_flow_seq(fsp_head->flow_seq); - } else { - // garbage packet - good_packet = false; - m_err_cntrs.m_bad_header++; - } - } - } - } - - if (good_packet) { - uint32_t pkt_seq = fsp_head->seq; - uint32_t exp_seq = curr_rfc2544->get_seq(); - if (unlikely(pkt_seq != exp_seq)) { - if (pkt_seq < exp_seq) { - if (exp_seq - pkt_seq > 100000) { - // packet loss while we had wrap around - curr_rfc2544->inc_seq_err(pkt_seq - exp_seq); - curr_rfc2544->inc_seq_err_too_big(); - curr_rfc2544->set_seq(pkt_seq + 1); - } else { - if (pkt_seq == (exp_seq - 1)) { - curr_rfc2544->inc_dup(); - } else { - curr_rfc2544->inc_ooo(); - // We thought it was lost, but it was just out of order - curr_rfc2544->dec_seq_err(); - } - curr_rfc2544->inc_seq_err_too_low(); - } - } else { - if (unlikely (pkt_seq - exp_seq > 100000)) { - // packet reorder while we had wrap around - if (pkt_seq == (exp_seq - 1)) { - curr_rfc2544->inc_dup(); - } else { - curr_rfc2544->inc_ooo(); - // We thought it was lost, but it was just out of order - curr_rfc2544->dec_seq_err(); - } - curr_rfc2544->inc_seq_err_too_low(); - } else { - // seq > curr_rfc2544->seq. Assuming lost packets - curr_rfc2544->inc_seq_err(pkt_seq - exp_seq); - curr_rfc2544->inc_seq_err_too_big(); - curr_rfc2544->set_seq(pkt_seq + 1); - } - } - } else { - curr_rfc2544->set_seq(pkt_seq + 1); - } - lp->m_port.m_rx_pg_stat_payload[hw_id].add_pkts(1); - lp->m_port.m_rx_pg_stat_payload[hw_id].add_bytes(m->pkt_len + 4); // +4 for ethernet CRC - uint64_t d = (os_get_hr_tick_64() - fsp_head->time_stamp ); - dsec_t ctime = ptime_convert_hr_dsec(d); - curr_rfc2544->add_sample(ctime); - } - } else { - hw_id = get_hw_id(ip_id); - if (hw_id < MAX_FLOW_STATS) { - lp->m_port.m_rx_pg_stat[hw_id].add_pkts(1); - lp->m_port.m_rx_pg_stat[hw_id].add_bytes(m->pkt_len + 4); // +4 for ethernet CRC - } - } - } +void CRxCoreStateless::handle_work_stage() { + + /* set the next sync time to */ + dsec_t sync_time_sec = now_sec() + (1.0 / 1000); + + while (m_state == STATE_WORKING) { + process_all_pending_pkts(); + + dsec_t now = now_sec(); + + if ( (now - sync_time_sec) > 0 ) { + periodic_check_for_cp_messages(); + port_manager_tick(); + sync_time_sec = now + (1.0 / 1000); } + + rte_pause(); + } } -void CRxCoreStateless::capture_pkt(rte_mbuf_t *m) { - -} +void CRxCoreStateless::start() { + /* register a watchdog handle on current core */ + m_monitor.create("STL RX CORE", 1); + TrexWatchDog::getInstance().register_monitor(&m_monitor); -// In VM setup, handle packets coming as messages from DP cores. -void CRxCoreStateless::handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r) { - while ( true ) { - CGenNode * node; - if ( r->Dequeue(node) != 0 ) { + while (m_state != STATE_QUIT) { + switch (m_state) { + case STATE_IDLE: + idle_state_loop(); break; - } - assert(node); - - CGenNodeMsgBase * msg = (CGenNodeMsgBase *)node; - CGenNodeLatencyPktInfo * l_msg; - uint8_t msg_type = msg->m_msg_type; - uint8_t rx_port_index; - CLatencyManagerPerPortStl * lp; - - switch (msg_type) { - case CGenNodeMsgBase::LATENCY_PKT: - l_msg = (CGenNodeLatencyPktInfo *)msg; - assert(l_msg->m_latency_offset == 0xdead); - rx_port_index = (thread_id << 1) + (l_msg->m_dir & 1); - assert( rx_port_index < m_max_ports ); - lp = &m_ports[rx_port_index]; - handle_rx_pkt(lp, (rte_mbuf_t *)l_msg->m_pkt); - if (m_capture) - capture_pkt((rte_mbuf_t *)l_msg->m_pkt); - rte_pktmbuf_free((rte_mbuf_t *)l_msg->m_pkt); + case STATE_WORKING: + handle_work_stage(); break; + default: - printf("ERROR latency-thread message type is not valid %d \n", msg_type); assert(0); + break; } - CGlobalInfo::free_node(node); } -} - -// VM mode function. Handle messages from DP -void CRxCoreStateless::try_rx_queues() { - CMessagingManager * rx_dp = CMsgIns::Ins()->getRxDp(); - uint8_t threads=CMsgIns::Ins()->get_num_threads(); - int ti; - for (ti = 0; ti < (int)threads; ti++) { - CNodeRing * r = rx_dp->getRingDpToCp(ti); - if ( ! r->isEmpty() ) { - handle_rx_queue_msgs((uint8_t)ti, r); - } - } + m_monitor.disable(); } -// exactly the same as try_rx, without the handle_rx_pkt -// purpose is to flush rx queues when core is in idle state -void CRxCoreStateless::flush_rx() { - rte_mbuf_t * rx_pkts[64]; - int i, total_pkts = 0; - for (i = 0; i < m_max_ports; i++) { - CLatencyManagerPerPortStl * lp = &m_ports[i]; - rte_mbuf_t * m; - /* try to read 64 packets clean up the queue */ - uint16_t cnt_p = lp->m_io->rx_burst(rx_pkts, 64); - total_pkts += cnt_p; - if (cnt_p) { - m_cpu_dp_u.start_work1(); - int j; - for (j = 0; j < cnt_p; j++) { - m = rx_pkts[j]; - rte_pktmbuf_free(m); - } - /* commit only if there was work to do ! */ - m_cpu_dp_u.commit1(); - }/* if work */ - }// all ports -} +void CRxCoreStateless::capture_pkt(rte_mbuf_t *m) { -int CRxCoreStateless::try_rx() { - rte_mbuf_t * rx_pkts[64]; - int i, total_pkts = 0; - for (i = 0; i < m_max_ports; i++) { - CLatencyManagerPerPortStl * lp = &m_ports[i]; - rte_mbuf_t * m; - /* try to read 64 packets clean up the queue */ - uint16_t cnt_p = lp->m_io->rx_burst(rx_pkts, 64); - total_pkts += cnt_p; - if (cnt_p) { - m_cpu_dp_u.start_work1(); - int j; - for (j = 0; j < cnt_p; j++) { - m = rx_pkts[j]; - handle_rx_pkt(lp, m); - rte_pktmbuf_free(m); - } - /* commit only if there was work to do ! */ - m_cpu_dp_u.commit1(); - }/* if work */ - }// all ports - return total_pkts; } -bool CRxCoreStateless::is_flow_stat_id(uint32_t id) { - if ((id & 0x000fff00) == IP_ID_RESERVE_BASE) return true; - return false; -} +int CRxCoreStateless::process_all_pending_pkts(bool flush_rx) { -bool CRxCoreStateless::is_flow_stat_payload_id(uint32_t id) { - if (id == FLOW_STAT_PAYLOAD_IP_ID) return true; - return false; -} + int total_pkts = 0; + for (int i = 0; i < m_max_ports; i++) { + total_pkts += m_rx_port_mngr[i].process_all_pending_pkts(flush_rx); + } + + return total_pkts; -uint16_t CRxCoreStateless::get_hw_id(uint16_t id) { - return (0x00ff & id); } void CRxCoreStateless::reset_rx_stats(uint8_t port_id) { - for (int hw_id = 0; hw_id < MAX_FLOW_STATS; hw_id++) { - m_ports[port_id].m_port.m_rx_pg_stat[hw_id].clear(); - } + m_rx_port_mngr[port_id].clear_stats(); } int CRxCoreStateless::get_rx_stats(uint8_t port_id, rx_per_flow_t *rx_stats, int min, int max , bool reset, TrexPlatformApi::driver_stat_cap_e type) { - for (int hw_id = min; hw_id <= max; hw_id++) { - if (type == TrexPlatformApi::IF_STAT_PAYLOAD) { - rx_stats[hw_id - min] = m_ports[port_id].m_port.m_rx_pg_stat_payload[hw_id]; - } else { - rx_stats[hw_id - min] = m_ports[port_id].m_port.m_rx_pg_stat[hw_id]; - } - if (reset) { - if (type == TrexPlatformApi::IF_STAT_PAYLOAD) { - m_ports[port_id].m_port.m_rx_pg_stat_payload[hw_id].clear(); - } else { - m_ports[port_id].m_port.m_rx_pg_stat[hw_id].clear(); - } - } - } - return 0; + + /* for now only latency stats */ + m_rx_port_mngr[port_id].get_latency_stats(rx_stats, min, max, reset, type); + + return (0); + } int CRxCoreStateless::get_rfc2544_info(rfc2544_info_t *rfc2544_info, int min, int max, bool reset) { @@ -458,13 +281,6 @@ int CRxCoreStateless::get_rx_err_cntrs(CRxCoreErrCntrs *rx_err) { return 0; } -void CRxCoreStateless::set_working_msg_ack(bool val) { - sanb_smp_memory_barrier(); - m_ack_start_work_msg = val; - sanb_smp_memory_barrier(); -} - - void CRxCoreStateless::update_cpu_util(){ m_cpu_cp_u.Update(); } @@ -472,3 +288,53 @@ void CRxCoreStateless::update_cpu_util(){ double CRxCoreStateless::get_cpu_util() { return m_cpu_cp_u.GetVal(); } + + +void +CRxCoreStateless::start_recorder(uint8_t port_id, const std::string &pcap_filename, uint64_t limit) { + m_rx_port_mngr[port_id].start_recorder(pcap_filename, limit); + recalculate_next_state(); +} + +void +CRxCoreStateless::stop_recorder(uint8_t port_id) { + m_rx_port_mngr[port_id].stop_recorder(); + recalculate_next_state(); +} + +void +CRxCoreStateless::start_queue(uint8_t port_id, uint64_t size) { + m_rx_port_mngr[port_id].start_queue(size); + recalculate_next_state(); +} + +void +CRxCoreStateless::stop_queue(uint8_t port_id) { + m_rx_port_mngr[port_id].stop_queue(); + recalculate_next_state(); +} + +void +CRxCoreStateless::enable_latency() { + for (int i = 0; i < m_max_ports; i++) { + m_rx_port_mngr[i].enable_latency(); + } + + recalculate_next_state(); +} + +void +CRxCoreStateless::disable_latency() { + for (int i = 0; i < m_max_ports; i++) { + m_rx_port_mngr[i].disable_latency(); + } + + recalculate_next_state(); +} + +const RXPortManager & +CRxCoreStateless::get_rx_port_mngr(uint8_t port_id) { + assert(port_id < m_max_ports); + return m_rx_port_mngr[port_id]; + +} diff --git a/src/stateless/rx/trex_stateless_rx_core.h b/src/stateless/rx/trex_stateless_rx_core.h index 3f9fb6cc..cd16bb8a 100644 --- a/src/stateless/rx/trex_stateless_rx_core.h +++ b/src/stateless/rx/trex_stateless_rx_core.h @@ -25,6 +25,7 @@ #include "os_time.h" #include "pal/linux/sanb_atomic.h" #include "utl_cpuu.h" +#include "trex_stateless_rx_port_mngr.h" class TrexStatelessCpToRxMsgBase; @@ -37,25 +38,6 @@ class CCPortLatencyStl { rx_per_flow_t m_rx_pg_stat_payload[MAX_FLOW_STATS_PAYLOAD]; }; -class CLatencyManagerPerPortStl { -public: - CCPortLatencyStl m_port; - CPortLatencyHWBase * m_io; -}; - -class CRxSlCfg { - public: - CRxSlCfg (){ - m_max_ports = 0; - m_cps = 0.0; - } - - public: - uint32_t m_max_ports; - double m_cps; - CPortLatencyHWBase * m_ports[TREX_MAX_PORTS]; -}; - class CRFC2544Info { public: void create(); @@ -109,12 +91,20 @@ class CRxCoreErrCntrs { m_old_flow = 0; } - private: + public: uint64_t m_bad_header; uint64_t m_old_flow; }; +/** + * stateless RX core + * + */ class CRxCoreStateless { + + /** + * core states + */ enum state_e { STATE_IDLE, STATE_WORKING, @@ -129,47 +119,78 @@ class CRxCoreStateless { , TrexPlatformApi::driver_stat_cap_e type); int get_rfc2544_info(rfc2544_info_t *rfc2544_info, int min, int max, bool reset); int get_rx_err_cntrs(CRxCoreErrCntrs *rx_err); - void work() { - m_state = STATE_WORKING; - m_err_cntrs.reset(); // When starting to work, reset global counters - } - void idle() {m_state = STATE_IDLE;} + + void quit() {m_state = STATE_QUIT;} - bool is_working() const {return (m_ack_start_work_msg == true);} - void set_working_msg_ack(bool val); + bool is_working() const {return (m_state == STATE_WORKING);} double get_cpu_util(); void update_cpu_util(); + const RXPacketBuffer *get_rx_queue_pkts(uint8_t port_id) { + return m_rx_port_mngr[port_id].get_pkt_buffer(); + } + /** + * start capturing of RX packets on a specific port + * + */ + void start_recorder(uint8_t port_id, const std::string &pcap_filename, uint64_t limit); + void stop_recorder(uint8_t port_id); + + /** + * start RX queueing of packets + * + */ + void start_queue(uint8_t port_id, uint64_t size); + void stop_queue(uint8_t port_id); + + /** + * enable latency feature for RX packets + * will be apply to all ports + */ + void enable_latency(); + void disable_latency(); + + const RXPortManager &get_rx_port_mngr(uint8_t port_id); + private: void handle_cp_msg(TrexStatelessCpToRxMsgBase *msg); bool periodic_check_for_cp_messages(); void tickle(); void idle_state_loop(); - void handle_rx_pkt(CLatencyManagerPerPortStl * lp, rte_mbuf_t * m); + + void recalculate_next_state(); + bool are_any_features_active(); + void capture_pkt(rte_mbuf_t *m); void handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r); - void flush_rx(); - int try_rx(); - void try_rx_queues(); - bool is_flow_stat_id(uint32_t id); - bool is_flow_stat_payload_id(uint32_t id); - uint16_t get_hw_id(uint16_t id); + void handle_work_stage(); + void port_manager_tick(); + + int process_all_pending_pkts(bool flush_rx = false); + void flush_all_pending_pkts() { + process_all_pending_pkts(true); + } + + void try_rx_queues(); + private: - TrexMonitor m_monitor; - uint32_t m_max_ports; - bool m_capture; - bool m_rcv_all; - CLatencyManagerPerPortStl m_ports[TREX_MAX_PORTS]; - state_e m_state; - CNodeRing *m_ring_from_cp; - CNodeRing *m_ring_to_cp; - CCpuUtlDp m_cpu_dp_u; - CCpuUtlCp m_cpu_cp_u; + TrexMonitor m_monitor; + uint32_t m_max_ports; + bool m_capture; + state_e m_state; + CNodeRing *m_ring_from_cp; + CNodeRing *m_ring_to_cp; + CCpuUtlDp m_cpu_dp_u; + CCpuUtlCp m_cpu_cp_u; + // Used for acking "work" (go out of idle) messages from cp volatile bool m_ack_start_work_msg __rte_cache_aligned; + CRxCoreErrCntrs m_err_cntrs; CRFC2544Info m_rfc2544[MAX_FLOW_STATS_PAYLOAD]; + + RXPortManager m_rx_port_mngr[TREX_MAX_PORTS]; }; #endif diff --git a/src/stateless/rx/trex_stateless_rx_defs.h b/src/stateless/rx/trex_stateless_rx_defs.h new file mode 100644 index 00000000..aefcc133 --- /dev/null +++ b/src/stateless/rx/trex_stateless_rx_defs.h @@ -0,0 +1,62 @@ +/* + Itay Marom + Cisco Systems, Inc. +*/ + +/* + Copyright (c) 2016-2016 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_RX_DEFS_H__ +#define __TREX_STATELESS_RX_DEFS_H__ + +#include "trex_defs.h" +#include <json/json.h> + +class CPortLatencyHWBase; + + +/** + * general SL cfg + * + */ +class CRxSlCfg { + public: + CRxSlCfg (){ + m_max_ports = 0; + m_cps = 0.0; + m_num_crc_fix_bytes = 0; + } + + public: + uint32_t m_max_ports; + double m_cps; + CPortLatencyHWBase * m_ports[TREX_MAX_PORTS]; + uint8_t m_num_crc_fix_bytes; +}; + +/** + * describes the filter type applied to the RX + * RX_FILTER_MODE_HW - only hardware filtered traffic will + * reach the RX core + * + */ +typedef enum rx_filter_mode_ { + RX_FILTER_MODE_HW, + RX_FILTER_MODE_ALL +} rx_filter_mode_e; + +#endif /* __TREX_STATELESS_RX_DEFS_H__ */ + diff --git a/src/stateless/rx/trex_stateless_rx_port_mngr.cpp b/src/stateless/rx/trex_stateless_rx_port_mngr.cpp new file mode 100644 index 00000000..afc6827c --- /dev/null +++ b/src/stateless/rx/trex_stateless_rx_port_mngr.cpp @@ -0,0 +1,515 @@ +/* + Itay Marom + Cisco Systems, Inc. +*/ + +/* + Copyright (c) 2016-2016 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 "bp_sim.h" +#include "trex_stateless_rx_port_mngr.h" +#include "common/captureFile.h" +#include "trex_stateless_rx_core.h" + +/************************************** + * latency RX feature + * + *************************************/ +RXLatency::RXLatency() { + m_rcv_all = false; + m_rfc2544 = NULL; + m_err_cntrs = NULL; + + for (int i = 0; i < MAX_FLOW_STATS; i++) { + m_rx_pg_stat[i].clear(); + m_rx_pg_stat_payload[i].clear(); + } +} + +void +RXLatency::create(CRFC2544Info *rfc2544, CRxCoreErrCntrs *err_cntrs) { + m_rfc2544 = rfc2544; + m_err_cntrs = err_cntrs; +} + +void +RXLatency::handle_pkt(const rte_mbuf_t *m) { + CFlowStatParser parser; + + if (m_rcv_all || parser.parse(rte_pktmbuf_mtod(m, uint8_t *), m->pkt_len) == 0) { + uint32_t ip_id; + if (m_rcv_all || (parser.get_ip_id(ip_id) == 0)) { + if (m_rcv_all || is_flow_stat_id(ip_id)) { + uint16_t hw_id; + if (m_rcv_all || is_flow_stat_payload_id(ip_id)) { + bool good_packet = true; + uint8_t *p = rte_pktmbuf_mtod(m, uint8_t*); + struct flow_stat_payload_header *fsp_head = (struct flow_stat_payload_header *) + (p + m->pkt_len - sizeof(struct flow_stat_payload_header)); + hw_id = fsp_head->hw_id; + CRFC2544Info *curr_rfc2544; + + if (unlikely(fsp_head->magic != FLOW_STAT_PAYLOAD_MAGIC) || hw_id >= MAX_FLOW_STATS_PAYLOAD) { + good_packet = false; + if (!m_rcv_all) + m_err_cntrs->m_bad_header++; + } else { + curr_rfc2544 = &m_rfc2544[hw_id]; + + if (fsp_head->flow_seq != curr_rfc2544->get_exp_flow_seq()) { + // bad flow seq num + // Might be the first packet of a new flow, packet from an old flow, or garbage. + + if (fsp_head->flow_seq == curr_rfc2544->get_prev_flow_seq()) { + // packet from previous flow using this hw_id that arrived late + good_packet = false; + m_err_cntrs->m_old_flow++; + } else { + if (curr_rfc2544->no_flow_seq()) { + // first packet we see from this flow + good_packet = true; + curr_rfc2544->set_exp_flow_seq(fsp_head->flow_seq); + } else { + // garbage packet + good_packet = false; + m_err_cntrs->m_bad_header++; + } + } + } + } + + if (good_packet) { + uint32_t pkt_seq = fsp_head->seq; + uint32_t exp_seq = curr_rfc2544->get_seq(); + if (unlikely(pkt_seq != exp_seq)) { + if (pkt_seq < exp_seq) { + if (exp_seq - pkt_seq > 100000) { + // packet loss while we had wrap around + curr_rfc2544->inc_seq_err(pkt_seq - exp_seq); + curr_rfc2544->inc_seq_err_too_big(); + curr_rfc2544->set_seq(pkt_seq + 1); + } else { + if (pkt_seq == (exp_seq - 1)) { + curr_rfc2544->inc_dup(); + } else { + curr_rfc2544->inc_ooo(); + // We thought it was lost, but it was just out of order + curr_rfc2544->dec_seq_err(); + } + curr_rfc2544->inc_seq_err_too_low(); + } + } else { + if (unlikely (pkt_seq - exp_seq > 100000)) { + // packet reorder while we had wrap around + if (pkt_seq == (exp_seq - 1)) { + curr_rfc2544->inc_dup(); + } else { + curr_rfc2544->inc_ooo(); + // We thought it was lost, but it was just out of order + curr_rfc2544->dec_seq_err(); + } + curr_rfc2544->inc_seq_err_too_low(); + } else { + // seq > curr_rfc2544->seq. Assuming lost packets + curr_rfc2544->inc_seq_err(pkt_seq - exp_seq); + curr_rfc2544->inc_seq_err_too_big(); + curr_rfc2544->set_seq(pkt_seq + 1); + } + } + } else { + curr_rfc2544->set_seq(pkt_seq + 1); + } + m_rx_pg_stat_payload[hw_id].add_pkts(1); + m_rx_pg_stat_payload[hw_id].add_bytes(m->pkt_len + 4); // +4 for ethernet CRC + uint64_t d = (os_get_hr_tick_64() - fsp_head->time_stamp ); + dsec_t ctime = ptime_convert_hr_dsec(d); + curr_rfc2544->add_sample(ctime); + } + } else { + hw_id = get_hw_id(ip_id); + if (hw_id < MAX_FLOW_STATS) { + m_rx_pg_stat[hw_id].add_pkts(1); + m_rx_pg_stat[hw_id].add_bytes(m->pkt_len + 4); // +4 for ethernet CRC + } + } + } + } + } +} + +void +RXLatency::reset_stats() { + for (int hw_id = 0; hw_id < MAX_FLOW_STATS; hw_id++) { + m_rx_pg_stat[hw_id].clear(); + } +} + + +void +RXLatency::get_stats(rx_per_flow_t *rx_stats, + int min, + int max, + bool reset, + TrexPlatformApi::driver_stat_cap_e type) { + + for (int hw_id = min; hw_id <= max; hw_id++) { + if (type == TrexPlatformApi::IF_STAT_PAYLOAD) { + rx_stats[hw_id - min] = m_rx_pg_stat_payload[hw_id]; + } else { + rx_stats[hw_id - min] = m_rx_pg_stat[hw_id]; + } + if (reset) { + if (type == TrexPlatformApi::IF_STAT_PAYLOAD) { + m_rx_pg_stat_payload[hw_id].clear(); + } else { + m_rx_pg_stat[hw_id].clear(); + } + } + } +} + + +Json::Value +RXLatency::to_json() const { + return Json::objectValue; +} + +/************************************** + * RX feature queue + * + *************************************/ + +RXPacketBuffer::RXPacketBuffer(uint64_t size) { + m_buffer = nullptr; + m_head = 0; + m_tail = 0; + m_size = (size + 1); // for the empty/full difference 1 slot reserved + + /* generate queue */ + m_buffer = new RXPacket*[m_size](); // zeroed +} + +RXPacketBuffer::~RXPacketBuffer() { + assert(m_buffer); + + while (!is_empty()) { + RXPacket *pkt = pop(); + delete pkt; + } + delete [] m_buffer; +} + +void +RXPacketBuffer::push(const rte_mbuf_t *m) { + /* if full - pop the oldest */ + if (is_full()) { + delete pop(); + } + + /* push packet */ + m_buffer[m_head] = new RXPacket(m); + m_head = next(m_head); +} + +RXPacket * +RXPacketBuffer::pop() { + assert(!is_empty()); + + RXPacket *pkt = m_buffer[m_tail]; + m_tail = next(m_tail); + + return pkt; +} + +uint64_t +RXPacketBuffer::get_element_count() const { + if (m_head >= m_tail) { + return (m_head - m_tail); + } else { + return ( get_capacity() - (m_tail - m_head - 1) ); + } +} + +Json::Value +RXPacketBuffer::to_json() const { + + Json::Value output = Json::arrayValue; + + int tmp = m_tail; + while (tmp != m_head) { + RXPacket *pkt = m_buffer[tmp]; + output.append(pkt->to_json()); + tmp = next(tmp); + } + + return output; +} + + +void +RXQueue::start(uint64_t size) { + if (m_pkt_buffer) { + delete m_pkt_buffer; + } + m_pkt_buffer = new RXPacketBuffer(size); +} + +void +RXQueue::stop() { + if (m_pkt_buffer) { + delete m_pkt_buffer; + m_pkt_buffer = NULL; + } +} + +const RXPacketBuffer * +RXQueue::fetch() { + + /* if no buffer or the buffer is empty - give a NULL one */ + if ( (!m_pkt_buffer) || (m_pkt_buffer->get_element_count() == 0) ) { + return nullptr; + } + + /* hold a pointer to the old one */ + RXPacketBuffer *old_buffer = m_pkt_buffer; + + /* replace the old one with a new one and freeze the old */ + m_pkt_buffer = new RXPacketBuffer(old_buffer->get_capacity()); + + return old_buffer; +} + +void +RXQueue::handle_pkt(const rte_mbuf_t *m) { + m_pkt_buffer->push(m); +} + +Json::Value +RXQueue::to_json() const { + assert(m_pkt_buffer != NULL); + + Json::Value output = Json::objectValue; + + output["size"] = Json::UInt64(m_pkt_buffer->get_capacity()); + output["count"] = Json::UInt64(m_pkt_buffer->get_element_count()); + + return output; +} + +/************************************** + * RX feature recorder + * + *************************************/ + +RXPacketRecorder::RXPacketRecorder() { + m_writer = NULL; + m_count = 0; + m_limit = 0; + m_epoch = -1; + + m_pending_flush = false; +} + +void +RXPacketRecorder::start(const std::string &pcap, uint64_t limit) { + m_writer = CCapWriterFactory::CreateWriter(LIBPCAP, (char *)pcap.c_str()); + if (m_writer == NULL) { + std::stringstream ss; + ss << "unable to create PCAP file: " << pcap; + throw TrexException(ss.str()); + } + + assert(limit > 0); + + m_limit = limit; + m_count = 0; + m_pending_flush = false; + m_pcap_filename = pcap; +} + +void +RXPacketRecorder::stop() { + if (!m_writer) { + return; + } + + delete m_writer; + m_writer = NULL; +} + +void +RXPacketRecorder::flush_to_disk() { + + if (m_writer && m_pending_flush) { + m_writer->flush_to_disk(); + m_pending_flush = false; + } +} + +void +RXPacketRecorder::handle_pkt(const rte_mbuf_t *m) { + if (!m_writer) { + return; + } + + dsec_t now = now_sec(); + if (m_epoch < 0) { + m_epoch = now; + } + + dsec_t dt = now - m_epoch; + + CPktNsecTimeStamp t_c(dt); + m_pkt.time_nsec = t_c.m_time_nsec; + m_pkt.time_sec = t_c.m_time_sec; + + const uint8_t *p = rte_pktmbuf_mtod(m, uint8_t *); + m_pkt.pkt_len = m->pkt_len; + memcpy(m_pkt.raw, p, m->pkt_len); + + m_writer->write_packet(&m_pkt); + m_count++; + m_pending_flush = true; + + if (m_count == m_limit) { + stop(); + } + +} + +Json::Value +RXPacketRecorder::to_json() const { + Json::Value output = Json::objectValue; + + output["pcap_filename"] = m_pcap_filename; + output["limit"] = Json::UInt64(m_limit); + output["count"] = Json::UInt64(m_count); + + return output; +} + +/************************************** + * Port manager + * + *************************************/ + +RXPortManager::RXPortManager() { + clear_all_features(); + m_io = NULL; + m_cpu_dp_u = NULL; +} + + +void +RXPortManager::create(CPortLatencyHWBase *io, + CRFC2544Info *rfc2544, + CRxCoreErrCntrs *err_cntrs, + CCpuUtlDp *cpu_util, + uint8_t crc_bytes_num) { + m_io = io; + m_cpu_dp_u = cpu_util; + m_num_crc_fix_bytes = crc_bytes_num; + + /* init features */ + m_latency.create(rfc2544, err_cntrs); +} + +void RXPortManager::handle_pkt(const rte_mbuf_t *m) { + + /* handle features */ + + if (is_feature_set(LATENCY)) { + m_latency.handle_pkt(m); + } + + if (is_feature_set(RECORDER)) { + m_recorder.handle_pkt(m); + } + + if (is_feature_set(QUEUE)) { + m_queue.handle_pkt(m); + } +} + + +int RXPortManager::process_all_pending_pkts(bool flush_rx) { + + rte_mbuf_t *rx_pkts[64]; + + /* try to read 64 packets clean up the queue */ + uint16_t cnt_p = m_io->rx_burst(rx_pkts, 64); + if (cnt_p == 0) { + return cnt_p; + } + + + m_cpu_dp_u->start_work1(); + + for (int j = 0; j < cnt_p; j++) { + rte_mbuf_t *m = rx_pkts[j]; + + if (!flush_rx) { + // patch relevant only for e1000 driver + if (m_num_crc_fix_bytes) { + rte_pktmbuf_trim(m, m_num_crc_fix_bytes); + } + + handle_pkt(m); + } + + rte_pktmbuf_free(m); + } + + /* commit only if there was work to do ! */ + m_cpu_dp_u->commit1(); + + + return cnt_p; +} + +void +RXPortManager::tick() { + if (is_feature_set(RECORDER)) { + m_recorder.flush_to_disk(); + } +} + +Json::Value +RXPortManager::to_json() const { + Json::Value output = Json::objectValue; + + if (is_feature_set(LATENCY)) { + output["latency"] = m_latency.to_json(); + output["latency"]["is_active"] = true; + } else { + output["latency"]["is_active"] = false; + } + + if (is_feature_set(RECORDER)) { + output["sniffer"] = m_recorder.to_json(); + output["sniffer"]["is_active"] = true; + } else { + output["sniffer"]["is_active"] = false; + } + + if (is_feature_set(QUEUE)) { + output["queue"] = m_queue.to_json(); + output["queue"]["is_active"] = true; + } else { + output["queue"]["is_active"] = false; + } + + return output; +} diff --git a/src/stateless/rx/trex_stateless_rx_port_mngr.h b/src/stateless/rx/trex_stateless_rx_port_mngr.h new file mode 100644 index 00000000..c049cb56 --- /dev/null +++ b/src/stateless/rx/trex_stateless_rx_port_mngr.h @@ -0,0 +1,416 @@ +/* + Itay Marom + Cisco Systems, Inc. +*/ + +/* + Copyright (c) 2016-2016 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_RX_PORT_MNGR_H__ +#define __TREX_STATELESS_RX_PORT_MNGR_H__ + +#include <stdint.h> +#include "common/base64.h" + +#include "common/captureFile.h" + + +class CPortLatencyHWBase; +class CRFC2544Info; +class CRxCoreErrCntrs; + +/************************************** + * RX feature latency + * + *************************************/ +class RXLatency { +public: + + RXLatency(); + + void create(CRFC2544Info *rfc2544, CRxCoreErrCntrs *err_cntrs); + + void handle_pkt(const rte_mbuf_t *m); + + Json::Value to_json() const; + + void get_stats(rx_per_flow_t *rx_stats, + int min, + int max, + bool reset, + TrexPlatformApi::driver_stat_cap_e type); + + void reset_stats(); + +private: + bool is_flow_stat_id(uint32_t id) { + if ((id & 0x000fff00) == IP_ID_RESERVE_BASE) return true; + return false; + } + + bool is_flow_stat_payload_id(uint32_t id) { + if (id == FLOW_STAT_PAYLOAD_IP_ID) return true; + return false; + } + + uint16_t get_hw_id(uint16_t id) { + return (0x00ff & id); +} + +public: + + rx_per_flow_t m_rx_pg_stat[MAX_FLOW_STATS]; + rx_per_flow_t m_rx_pg_stat_payload[MAX_FLOW_STATS_PAYLOAD]; + + bool m_rcv_all; + CRFC2544Info *m_rfc2544; + CRxCoreErrCntrs *m_err_cntrs; +}; + +/** + * describes a single saved RX packet + * + */ +class RXPacket { +public: + + RXPacket(const rte_mbuf_t *m) { + /* assume single part packet */ + assert(m->nb_segs == 1); + + m_size = m->pkt_len; + const uint8_t *p = rte_pktmbuf_mtod(m, uint8_t *); + + m_raw = new uint8_t[m_size]; + memcpy(m_raw, p, m_size); + + /* generate a packet timestamp */ + m_timestamp = now_sec(); + } + + /* slow path and also RVO - pass by value is ok */ + Json::Value to_json() { + Json::Value output; + output["ts"] = m_timestamp; + output["binary"] = base64_encode(m_raw, m_size); + return output; + } + + ~RXPacket() { + if (m_raw) { + delete [] m_raw; + } + } + +private: + + uint8_t *m_raw; + uint16_t m_size; + dsec_t m_timestamp; +}; + + +/************************************** + * RX feature queue + * + *************************************/ + +class RXPacketBuffer { +public: + + RXPacketBuffer(uint64_t size); + ~RXPacketBuffer(); + + /** + * push a packet to the buffer + * + */ + void push(const rte_mbuf_t *m); + + /** + * generate a JSON output of the queue + * + */ + Json::Value to_json() const; + + + bool is_empty() const { + return (m_head == m_tail); + } + + bool is_full() const { + return ( next(m_head) == m_tail); + } + + /** + * return the total amount of space possible + */ + uint64_t get_capacity() const { + /* one slot is used for diff between full/empty */ + return (m_size - 1); + } + + /** + * returns how many elements are in the queue + */ + uint64_t get_element_count() const; + +private: + int next(int v) const { + return ( (v + 1) % m_size ); + } + + /* pop in case of full queue - internal usage */ + RXPacket * pop(); + + int m_head; + int m_tail; + int m_size; + RXPacket **m_buffer; +}; + + +class RXQueue { +public: + RXQueue() { + m_pkt_buffer = nullptr; + } + + ~RXQueue() { + stop(); + } + + /** + * start RX queue + * + */ + void start(uint64_t size); + + /** + * fetch the current buffer + * return NULL if no packets + */ + const RXPacketBuffer * fetch(); + + /** + * stop RX queue + * + */ + void stop(); + + void handle_pkt(const rte_mbuf_t *m); + + Json::Value to_json() const; + +private: + RXPacketBuffer *m_pkt_buffer; +}; + +/************************************** + * RX feature PCAP recorder + * + *************************************/ + +class RXPacketRecorder { +public: + RXPacketRecorder(); + + ~RXPacketRecorder() { + stop(); + } + + void start(const std::string &pcap, uint64_t limit); + void stop(); + void handle_pkt(const rte_mbuf_t *m); + + /** + * flush any cached packets to disk + * + */ + void flush_to_disk(); + + Json::Value to_json() const; + +private: + CFileWriterBase *m_writer; + std::string m_pcap_filename; + CCapPktRaw m_pkt; + dsec_t m_epoch; + uint64_t m_limit; + uint64_t m_count; + bool m_pending_flush; +}; + + +/************************ manager ***************************/ + +/** + * per port RX features manager + * + * @author imarom (10/30/2016) + */ +class RXPortManager { +public: + enum feature_t { + NO_FEATURES = 0x0, + LATENCY = 0x1, + RECORDER = 0x2, + QUEUE = 0x4 + }; + + RXPortManager(); + + void create(CPortLatencyHWBase *io, + CRFC2544Info *rfc2544, + CRxCoreErrCntrs *err_cntrs, + CCpuUtlDp *cpu_util, + uint8_t crc_bytes_num); + + void clear_stats() { + m_latency.reset_stats(); + } + + void get_latency_stats(rx_per_flow_t *rx_stats, + int min, + int max, + bool reset, + TrexPlatformApi::driver_stat_cap_e type) { + + return m_latency.get_stats(rx_stats, min, max, reset, type); + } + + RXLatency & get_latency() { + return m_latency; + } + + /* latency */ + void enable_latency() { + set_feature(LATENCY); + } + + void disable_latency() { + unset_feature(LATENCY); + } + + /* recorder */ + void start_recorder(const std::string &pcap, uint64_t limit_pkts) { + m_recorder.start(pcap, limit_pkts); + set_feature(RECORDER); + } + + void stop_recorder() { + m_recorder.stop(); + unset_feature(RECORDER); + } + + /* queue */ + void start_queue(uint32_t size) { + m_queue.start(size); + set_feature(QUEUE); + } + + void stop_queue() { + m_queue.stop(); + unset_feature(QUEUE); + } + + const RXPacketBuffer *get_pkt_buffer() { + if (!is_feature_set(QUEUE)) { + return nullptr; + } + + return m_queue.fetch(); + } + + + + /** + * fetch and process all packets + * + */ + int process_all_pending_pkts(bool flush_rx = false); + + + /** + * flush all pending packets without processing them + * + */ + void flush_all_pending_pkts() { + process_all_pending_pkts(true); + } + + + /** + * handle a single packet + * + */ + void handle_pkt(const rte_mbuf_t *m); + + /** + * maintenance + * + * @author imarom (11/24/2016) + */ + void tick(); + + bool has_features_set() { + return (m_features != NO_FEATURES); + } + + + bool no_features_set() { + return (!has_features_set()); + } + + /** + * write the status to a JSON format + */ + Json::Value to_json() const; + +private: + + void clear_all_features() { + m_features = NO_FEATURES; + } + + void set_feature(feature_t feature) { + m_features |= feature; + } + + void unset_feature(feature_t feature) { + m_features &= (~feature); + } + + bool is_feature_set(feature_t feature) const { + return ( (m_features & feature) == feature ); + } + + uint32_t m_features; + + RXLatency m_latency; + RXPacketRecorder m_recorder; + RXQueue m_queue; + // compensate for the fact that hardware send us packets without Ethernet CRC, and we report with it + uint8_t m_num_crc_fix_bytes; + + CCpuUtlDp *m_cpu_dp_u; + CPortLatencyHWBase *m_io; +}; + + + +#endif /* __TREX_STATELESS_RX_PORT_MNGR_H__ */ + diff --git a/src/trex_port_attr.cpp b/src/trex_port_attr.cpp new file mode 100644 index 00000000..61e88585 --- /dev/null +++ b/src/trex_port_attr.cpp @@ -0,0 +1,145 @@ +/* +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_port_attr.h" +#include "bp_sim.h" + +DestAttr::DestAttr(uint8_t port_id) { + m_port_id = port_id; + + m_mac = CGlobalInfo::m_options.m_mac_addr[port_id].u.m_mac.dest; + m_type = DEST_TYPE_MAC; + + /* save the default */ + memcpy(m_default_mac, m_mac, 6); +} + + +/** + * set dest as an IPv4 unresolved + */ +void +DestAttr::set_dest(uint32_t ipv4) { + assert(ipv4 != 0); + + m_ipv4 = ipv4; + memset(m_mac, 0, 6); // just to be on the safe side + m_type = DEST_TYPE_IPV4_UNRESOLVED; +} + +/** + * set dest as a resolved IPv4 + */ +void +DestAttr::set_dest(uint32_t ipv4, const uint8_t *mac) { + assert(ipv4 != 0); + + m_ipv4 = ipv4; + + /* source might be the same as dest (this shadows the datapath memory) */ + memmove(m_mac, mac, 6); + m_type = DEST_TYPE_IPV4; +} + +/** + * dest dest as MAC + * + */ +void +DestAttr::set_dest(const uint8_t *mac) { + + m_ipv4 = 0; + + /* source might be the same as dest (this shadows the datapath memory) */ + memmove(m_mac, mac, 6); + m_type = DEST_TYPE_MAC; +} + +void +DestAttr::to_json(Json::Value &output) const { + switch (m_type) { + + case DEST_TYPE_IPV4: + output["type"] = "ipv4"; + output["ipv4"] = utl_uint32_to_ipv4(m_ipv4); + output["arp"] = utl_macaddr_to_str(m_mac); + break; + + case DEST_TYPE_IPV4_UNRESOLVED: + output["type"] = "ipv4_u"; + output["ipv4"] = utl_uint32_to_ipv4(m_ipv4); + break; + + case DEST_TYPE_MAC: + output["type"] = "mac"; + output["mac"] = utl_macaddr_to_str(m_mac); + break; + + default: + assert(0); + } + +} + +const uint8_t * +TRexPortAttr::get_src_mac() const { + return CGlobalInfo::m_options.get_src_mac_addr(m_port_id); +} + + +std::string +TRexPortAttr::get_rx_filter_mode() const { + switch (m_rx_filter_mode) { + case RX_FILTER_MODE_ALL: + return "all"; + case RX_FILTER_MODE_HW: + return "hw"; + default: + assert(0); + } +} + + +void +TRexPortAttr::to_json(Json::Value &output) { + + output["src_mac"] = utl_macaddr_to_str(get_src_mac()); + output["promiscuous"]["enabled"] = get_promiscuous(); + output["link"]["up"] = is_link_up(); + output["speed"] = get_link_speed() / 1000; // make sure we have no cards of less than 1 Gbps + output["rx_filter_mode"] = get_rx_filter_mode(); + + if (get_src_ipv4() != 0) { + output["src_ipv4"] = utl_uint32_to_ipv4(get_src_ipv4()); + } else { + output["src_ipv4"] = Json::nullValue; + } + + + int mode; + get_flow_ctrl(mode); + output["fc"]["mode"] = mode; + + m_dest.to_json(output["dest"]); + +} + +void +TRexPortAttr::update_src_dst_mac(uint8_t *raw_pkt) { + memcpy(raw_pkt, get_dest().get_dest_mac(), 6); + memcpy(raw_pkt + 6, get_src_mac(), 6); +} + diff --git a/src/trex_port_attr.h b/src/trex_port_attr.h index 9231e263..3cb9beff 100755 --- a/src/trex_port_attr.h +++ b/src/trex_port_attr.h @@ -21,10 +21,93 @@ limitations under the License. #include <vector> #include "rte_ethdev_includes.h" #include "trex_defs.h" +#include "common/basic_utils.h" +#include <json/json.h> +#include "trex_stateless_rx_defs.h" +#include <string.h> + +/** + * destination port attribute + * + */ +class DestAttr { + +public: + + DestAttr(uint8_t port_id); + + /** + * dest can be either MAC IPv4, or IPv4 unresolved + */ + enum dest_type_e { + DEST_TYPE_IPV4 = 1, + DEST_TYPE_IPV4_UNRESOLVED, + DEST_TYPE_MAC, + }; + + /** + * set dest as an IPv4 unresolved + */ + void set_dest(uint32_t ipv4); + + /** + * set dest as a resolved IPv4 + */ + void set_dest(uint32_t ipv4, const uint8_t *mac); + + /** + * set dest as a plain MAC + */ + void set_dest(const uint8_t *mac); + + + /** + * return true if destination is resolved + */ + bool is_resolved() const { + return (m_type != DEST_TYPE_IPV4_UNRESOLVED); + } + + /** + * get the dest mac + * if the dest is not resolved + * it will return the default MAC + */ + const uint8_t *get_dest_mac() { + return m_mac; + } + + /** + * when link gets down - this should be called + * + */ + void on_link_down() { + if (m_type == DEST_TYPE_IPV4) { + /* reset the IPv4 dest with no resolution */ + set_dest(m_ipv4); + } + } + + void to_json(Json::Value &output) const; + +private: + uint32_t m_ipv4; + uint8_t *m_mac; + dest_type_e m_type; + uint8_t m_port_id; + +private: + uint8_t m_default_mac[6]; +}; class TRexPortAttr { public: + + TRexPortAttr(uint8_t port_id) : m_dest(port_id) { + m_src_ipv4 = 0; + } + virtual ~TRexPortAttr(){} /* UPDATES */ @@ -33,10 +116,10 @@ public: virtual void update_device_info() = 0; virtual void reset_xstats() = 0; virtual void update_description() = 0; - + /* GETTERS */ virtual bool get_promiscuous() = 0; - virtual void macaddr_get(struct ether_addr *mac_addr) = 0; + virtual void get_hw_src_mac(struct ether_addr *mac_addr) = 0; virtual uint32_t get_link_speed() { return m_link.link_speed; } // L1 Mbps virtual bool is_link_duplex() { return (m_link.link_duplex ? true : false); } virtual bool is_link_autoneg() { return (m_link.link_autoneg ? true : false); } @@ -51,24 +134,50 @@ public: virtual void get_description(std::string &description) { description = intf_info_st.description; } virtual void get_supported_speeds(supp_speeds_t &supp_speeds) = 0; + uint32_t get_src_ipv4() {return m_src_ipv4;} + DestAttr & get_dest() {return m_dest;} + + const uint8_t *get_src_mac() const; + std::string get_rx_filter_mode() const; + + /* for a raw packet, write the src/dst MACs */ + void update_src_dst_mac(uint8_t *raw_pkt); + /* SETTERS */ virtual int set_promiscuous(bool enabled) = 0; virtual int add_mac(char * mac) = 0; virtual int set_link_up(bool up) = 0; virtual int set_flow_ctrl(int mode) = 0; virtual int set_led(bool on) = 0; - -/* DUMPS */ + virtual int set_rx_filter_mode(rx_filter_mode_e mode) = 0; + + void set_src_ipv4(uint32_t addr) { + m_src_ipv4 = addr; + } + + /* DUMPS */ virtual void dump_link(FILE *fd) = 0; + /* dump object status to JSON */ + void to_json(Json::Value &output); + + protected: - uint8_t m_port_id; - rte_eth_link m_link; - struct rte_eth_dev_info dev_info; - bool flag_is_virtual; - bool flag_is_fc_change_supported; - bool flag_is_led_change_supported; - bool flag_is_link_change_supported; + + uint8_t m_port_id; + rte_eth_link m_link; + uint32_t m_src_ipv4; + DestAttr m_dest; + + struct rte_eth_dev_info dev_info; + + rx_filter_mode_e m_rx_filter_mode; + + bool flag_is_virtual; + bool flag_is_fc_change_supported; + bool flag_is_led_change_supported; + bool flag_is_link_change_supported; + struct intf_info_st { std::string pci_addr; @@ -81,8 +190,11 @@ protected: class DpdkTRexPortAttr : public TRexPortAttr { public: - DpdkTRexPortAttr(uint8_t port_id, bool is_virtual, bool fc_change_allowed) { + DpdkTRexPortAttr(uint8_t port_id, bool is_virtual, bool fc_change_allowed) : TRexPortAttr(port_id) { + m_port_id = port_id; + m_rx_filter_mode = RX_FILTER_MODE_HW; + flag_is_virtual = is_virtual; int tmp; flag_is_fc_change_supported = fc_change_allowed && (get_flow_ctrl(tmp) != -ENOTSUP); @@ -101,7 +213,7 @@ public: /* GETTERS */ virtual bool get_promiscuous(); - virtual void macaddr_get(struct ether_addr *mac_addr); + virtual void get_hw_src_mac(struct ether_addr *mac_addr); virtual int get_xstats_values(xstats_values_t &xstats_values); virtual int get_xstats_names(xstats_names_t &xstats_names); virtual int get_flow_ctrl(int &mode); @@ -114,6 +226,7 @@ public: virtual int set_flow_ctrl(int mode); virtual int set_led(bool on); + virtual int set_rx_filter_mode(rx_filter_mode_e mode); /* DUMPS */ virtual void dump_link(FILE *fd); @@ -128,7 +241,7 @@ private: class SimTRexPortAttr : public TRexPortAttr { public: - SimTRexPortAttr() { + SimTRexPortAttr() : TRexPortAttr(0) { m_link.link_speed = 10000; m_link.link_duplex = 1; m_link.link_autoneg = 0; @@ -146,7 +259,7 @@ public: void reset_xstats() {} void update_description() {} bool get_promiscuous() { return false; } - void macaddr_get(struct ether_addr *mac_addr) {} + void get_hw_src_mac(struct ether_addr *mac_addr) {} int get_xstats_values(xstats_values_t &xstats_values) { return -ENOTSUP; } int get_xstats_names(xstats_names_t &xstats_names) { return -ENOTSUP; } int get_flow_ctrl(int &mode) { return -ENOTSUP; } @@ -158,6 +271,7 @@ public: int set_flow_ctrl(int mode) { return -ENOTSUP; } int set_led(bool on) { return -ENOTSUP; } void dump_link(FILE *fd) {} + int set_rx_filter_mode(rx_filter_mode_e mode) { return -ENOTSUP; } }; diff --git a/src/utl_ip.h b/src/utl_ip.h index 95db588a..27bb6c81 100644 --- a/src/utl_ip.h +++ b/src/utl_ip.h @@ -105,6 +105,7 @@ class COneIPInfo { } COneIPInfo_ip_types; public: + virtual ~COneIPInfo() {} virtual void get_mac(uint8_t *mac) const { m_mac.copyToArray(mac); } @@ -154,6 +155,7 @@ class COneIPv4Info : public COneIPInfo { COneIPv4Info(uint32_t ip, uint16_t vlan, MacAddress mac, uint8_t port) : COneIPInfo(vlan, mac, port) { m_ip = ip; } + ~COneIPv4Info() {}; uint32_t get_ip() {return m_ip;} virtual uint8_t ip_ver() const {return IP4_VER;} virtual uint32_t get_arp_req_len() const {return 60;} @@ -194,6 +196,7 @@ class COneIPv6Info : public COneIPInfo { COneIPv6Info(uint16_t ip[8], uint16_t vlan, MacAddress mac, uint8_t port) : COneIPInfo(vlan, mac, port) { memcpy(m_ip, ip, sizeof(m_ip)); } + ~COneIPv6Info() {} const uint8_t *get_ipv6() {return (uint8_t *)m_ip;} virtual uint8_t ip_ver() const {return IP6_VER;} diff --git a/src/utl_term_io.cpp b/src/utl_term_io.cpp index 8e561188..e45aeebd 100755 --- a/src/utl_term_io.cpp +++ b/src/utl_term_io.cpp @@ -78,6 +78,13 @@ int utl_termio_init(){ atexit(exit_handler1); save_termio(); set_conio_terminal_mode(); + + /* stdout is non-blocking */ + int fd = fileno(stdout); + int f = fcntl(fd, F_GETFL, 0); + f |= O_NONBLOCK; + fcntl(fd, F_SETFL, f); + return (0); } |