Merge pull request #141 from DUNE-DAQ/thea/multi_garp

Multi-garp testing

Author: roland-sipos

CMakeLists.txt

@@ -73,6 +73,7 @@ daq_add_plugin(DPDKReaderModule duneDAQModule LINK_LIBRARIES appfwk::appfwk dpdk
 daq_add_application(dpdklibs_test_eal test_eal_app.cxx TEST LINK_LIBRARIES dpdklibs ${DPDK_LIBRARIES})
 daq_add_application(dpdklibs_test_frame_transmitter test_frame_transmitter.cxx TEST LINK_LIBRARIES dpdklibs CLI11::CLI11 ${DPDK_LIBRARIES} )
 daq_add_application(dpdklibs_test_frame_receiver test_frame_receiver.cxx TEST LINK_LIBRARIES dpdklibs CLI11::CLI11 ${DPDK_LIBRARIES} opmonlib::opmonlib)
+daq_add_application(dpdklibs_test_garp test_garp.cxx TEST LINK_LIBRARIES dpdklibs CLI11::CLI11 ${DPDK_LIBRARIES} )
 daq_add_application(dpdklibs_test_arp_response test_arp_response.cxx TEST LINK_LIBRARIES dpdklibs ${DPDK_LIBRARIES})
 daq_add_application(dpdklibs_test_transmit_and_receive test_transmit_and_receive.cxx TEST LINK_LIBRARIES dpdklibs CLI11::CLI11 ${DPDK_LIBRARIES})
 daq_add_application(dpdklibs_test_dpdk_stats test_dpdk_stats.cxx TEST LINK_LIBRARIES dpdklibs CLI11::CLI11 ${DPDK_LIBRARIES})

scripts/dpdklibs_udp_receiver.py

@@ -36,13 +36,13 @@ def dump_data(data):
 @click.command()
 @click.option('-d', '--dump', is_flag=True, default=False)
 @click.option('-c', '--count', type=int, default=None)
-@click.option('-p', '--port', type=int, default=None)
+@click.option('-p', '--port', type=int, default=0x4444)
 @click.option('-g', '--gap', type=int, default=None)
 @click.option('-f', '--frame-type', type=click.Choice(['wib', 'tde']), default='wib')
 def main(dump, count, port, gap, frame_type):
     s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
 
-    s.bind(('0.0.0.0', port))
+    s.bind(('', port))
 
     prev_stream = {}
     i=0

src/EALSetup.cpp

@@ -5,8 +5,8 @@
  * Licensing/copyright details are in the COPYING file that you should have
  * received with this code.
  */
-#ifndef DPDKLIBS_INCLUDE_DPDKLIBS_EALSETUP_HPP_
-#define DPDKLIBS_INCLUDE_DPDKLIBS_EALSETUP_HPP_
+// #ifndef DPDKLIBS_INCLUDE_DPDKLIBS_EALSETUP_HPP_
+// #define DPDKLIBS_INCLUDE_DPDKLIBS_EALSETUP_HPP_
 
 #include "logging/Logging.hpp"
 
@@ -57,7 +57,7 @@ std::string get_mac_addr_str(const rte_ether_addr& addr) {
 
 // Modifies Ethernet device configuration to multi-queue RSS with offload
 void
-iface_conf_rss_mode(struct rte_eth_conf& iface_conf, bool mode = false, bool offload = false)
+iface_conf_rss_mode(struct rte_eth_conf& iface_conf, bool mode, bool offload)
 {
   if (mode) {
     iface_conf.rxmode.mq_mode = RTE_ETH_MQ_RX_RSS;
@@ -71,7 +71,7 @@ iface_conf_rss_mode(struct rte_eth_conf& iface_conf, bool mode = false, bool off
 
 // Enables RX in promiscuous mode for the Ethernet device.
 int
-iface_promiscuous_mode(std::uint16_t iface, bool mode = false) 
+iface_promiscuous_mode(std::uint16_t iface, bool mode) 
 {
   int retval = -1;
   retval = rte_eth_promiscuous_get(iface);
@@ -332,5 +332,5 @@ void finish_eal() {
 } // namespace ealutils
 } // namespace dpdklibs
 } // namespace dunedaq
+// #endif // DPDKLIBS_INCLUDE_DPDKLIBS_EALSETUP_HPP_
 
-#endif // DPDKLIBS_INCLUDE_DPDKLIBS_EALSETUP_HPP_

test/apps/test_frame_receiver.cxx

@@ -1,6 +1,6 @@
 
 /* Application will run until quit or killed. */
-#include <fmt/core.h>
+
 #include <inttypes.h>
 #include <rte_cycles.h>
 #include <rte_eal.h>
@@ -21,10 +21,20 @@
 #include "CLI/Formatter.hpp"
 
 #include "detdataformats/DAQEthHeader.hpp"
+#include "dpdklibs/EALSetup.hpp"
 #include "dpdklibs/udp/PacketCtor.hpp"
 #include "dpdklibs/udp/Utils.hpp"
 #include "logging/Logging.hpp"
-#include "dpdklibs/EALSetup.hpp"
+
+#include "dpdklibs/udp/PacketCtor.hpp"
+#include "dpdklibs/udp/Utils.hpp"
+#include "dpdklibs/arp/ARP.hpp"
+#include "dpdklibs/ipv4_addr.hpp"
+
+#include <fmt/core.h>
+#include <fmt/ranges.h>
+
+#include <regex>
 
 #define RX_RING_SIZE 1024
 #define TX_RING_SIZE 1024
@@ -117,8 +127,10 @@ namespace {
     std::atomic<uint64_t> max_payload_size       = 0;
     std::atomic<uint64_t> udp_pkt_counter        = 0;
 
-    std::ofstream datafile;
-    const std::string output_data_filename = "dpdklibs_test_frame_receiver.dat";
+      std::atomic<int64_t> garps_sent = 0;
+
+    // std::ofstream datafile;
+    // const std::string output_data_filename = "dpdklibs_test_frame_receiver.dat";
 
 
 } // namespace
@@ -190,7 +202,7 @@ static inline int check_packet_size(struct rte_mbuf* mbuf, StreamUID unique_str_
     return 0;
 }
 
-static int lcore_main(struct rte_mempool* mbuf_pool, uint16_t iface, uint64_t time_per_report){
+static int lcore_main(struct rte_mempool* mbuf_pool, uint16_t iface, uint64_t time_per_report, const std::vector<std::string>& garp_ip_addr_strs){
     /*
      * Check that the iface is on the same NUMA node as the polling thread
      * for best performance.
@@ -260,34 +272,69 @@ static int lcore_main(struct rte_mempool* mbuf_pool, uint16_t iface, uint64_t ti
         }
     });
 
-    struct rte_mbuf **bufs = (rte_mbuf**) malloc(sizeof(struct rte_mbuf*) * burst_size);
-    rte_pktmbuf_alloc_bulk(mbuf_pool, bufs, burst_size);
-
-    datafile.open(output_data_filename, std::ios::out | std::ios::binary);
-    if ( (datafile.rdstate() & std::ofstream::failbit ) != 0 ) {
-        fmt::print("WARNING: Unable to open output file \"{}\"\n", output_data_filename);
+    //
+    // Prepare and start GARP thread
+    //
+    std::vector<rte_be32_t> ip_addr_bin_vector;
+    for( const auto& ip_addr_str : garp_ip_addr_strs ) {
+    TLOG() << "IP address for ARP responses: " << ip_addr_str;
+        IpAddr ip_addr(ip_addr_str);
+        rte_be32_t ip_addr_bin = ip_address_dotdecimal_to_binary(
+        ip_addr.addr_bytes[3],
+        ip_addr.addr_bytes[2],
+        ip_addr.addr_bytes[1],
+        ip_addr.addr_bytes[0]
+        );
+        ip_addr_bin_vector.push_back(ip_addr_bin);
     }
 
+    struct rte_mbuf **tx_bufs = (rte_mbuf**) malloc(sizeof(struct rte_mbuf*) * burst_size);
+    rte_pktmbuf_alloc_bulk(mbuf_pool, tx_bufs, burst_size);
+
+    auto garp = std::thread([&]() {
+        while (true) {
+        // TLOG() << "Packets/s: " << num_packets << " Bytes/s: " << num_bytes << " Total packets: " << total_packets << " Failed packets: " << failed_packets;
+        // num_packets.exchange(0);
+        // num_bytes.exchange(0);
+
+        for(const auto& ip_addr_bin : ip_addr_bin_vector ) {
+            arp::pktgen_send_garp(tx_bufs[0], iface, ip_addr_bin);
+            ++garps_sent;
+        }
+
+        std::this_thread::sleep_for(std::chrono::seconds(1)); // If we sample for anything other than 1s, the rate calculation will need to change
+        }
+    });
+
+
+    struct rte_mbuf **rx_bufs = (rte_mbuf**) malloc(sizeof(struct rte_mbuf*) * burst_size);
+    rte_pktmbuf_alloc_bulk(mbuf_pool, rx_bufs, burst_size);
+
+    // datafile.open(output_data_filename, std::ios::out | std::ios::binary);
+    // if ( (datafile.rdstate() & std::ofstream::failbit ) != 0 ) {
+        // fmt::print("WARNING: Unable to open output file \"{}\"\n", output_data_filename);
+    // }
+
     while (true) {
         /* Get burst of RX packets, from first iface of pair. */
-        const uint16_t nb_rx = rte_eth_rx_burst(iface, 0, bufs, burst_size);
+        const uint16_t nb_rx = rte_eth_rx_burst(iface, 0, rx_bufs, burst_size);
 
         num_packets   += nb_rx;
         total_packets += nb_rx;
 
         for (int i_b = 0; i_b < nb_rx; ++i_b) {
-            num_bytes += bufs[i_b]->pkt_len;
+            num_bytes += rx_bufs[i_b]->pkt_len;
 
 
             bool dump_packet = false;
-            if (not RTE_ETH_IS_IPV4_HDR(bufs[i_b]->packet_type)) {
+            if (not RTE_ETH_IS_IPV4_HDR(rx_bufs[i_b]->packet_type)) {
                 non_ipv4_packets++;
                 dump_packet = true;
                 continue;
             }
             ++udp_pkt_counter;
 
-            char* udp_payload = udp::get_udp_payload(bufs[i_b]);
+            char* udp_payload = udp::get_udp_payload(rx_bufs[i_b]);
             const detdataformats::DAQEthHeader* daq_hdr = reinterpret_cast<const detdataformats::DAQEthHeader*>(udp_payload);
 
             // uint64_t unique_str_id = (daq_hdr->det_id<<22) + (daq_hdr->crate_id<<12) + (daq_hdr->slot_id<<8) + daq_hdr->stream_id;
@@ -300,12 +347,12 @@ static int lcore_main(struct rte_mempool* mbuf_pool, uint16_t iface, uint64_t ti
                 stream_stats[unique_str_id];
                 stream_stats[unique_str_id].prev_seq_id = daq_hdr->seq_id - 1;
             }
-            stream_stats[unique_str_id].num_bytes += bufs[i_b]->pkt_len;
+            stream_stats[unique_str_id].num_bytes += rx_bufs[i_b]->pkt_len;
 
             if (check_against_previous_stream(daq_hdr, 2048) != 0){
                 dump_packet = true;
             }
-            if (check_packet_size(bufs[i_b], unique_str_id) != 0){
+            if (check_packet_size(rx_bufs[i_b], unique_str_id) != 0){
                 dump_packet = true;
             }
 
@@ -316,11 +363,11 @@ static int lcore_main(struct rte_mempool* mbuf_pool, uint16_t iface, uint64_t ti
             if (dump_packet && dumped_packet_count < max_packets_to_dump) {
                 dumped_packet_count++;
 
-                rte_pktmbuf_dump(stdout, bufs[i_b], bufs[i_b]->pkt_len);
+                rte_pktmbuf_dump(stdout, rx_bufs[i_b], rx_bufs[i_b]->pkt_len);
             }
         }
 
-        rte_pktmbuf_free_bulk(bufs, nb_rx);
+        rte_pktmbuf_free_bulk(rx_bufs, nb_rx);
     }
 
     return 0;
@@ -329,39 +376,66 @@ static int lcore_main(struct rte_mempool* mbuf_pool, uint16_t iface, uint64_t ti
 // Define the function to be called when ctrl-c (SIGINT) is sent to process
 void signal_callback_handler(int signum){
     fmt::print("Caught signal {}\n", signum);
-    if (datafile.is_open()) {
-        datafile.close();
-    }
-    // Terminate program
     std::exit(signum);
 }
 
 int main(int argc, char** argv){
     uint64_t time_per_report = 1;
     uint16_t iface = 0;
+    std::vector<std::string> garp_ip_addresses;
+    std::vector<std::string> pcie_addresses;
 
     CLI::App app{"test frame receiver"};
-    app.add_option("-s", expected_packet_size, "Expected frame size");
-    app.add_option("-i", iface, "Interface to init");
-    app.add_option("-t", time_per_report, "Time Per Report");
+    app.add_option("-g,--garp-ip-address", garp_ip_addresses, "IP Addresses");
+    app.add_option("-m,--pcie-mask", pcie_addresses, "PCIE Addresses device mask");
+    app.add_option("-s,--exp-frame-size", expected_packet_size, "Expected frame size");
+    app.add_option("-i,--iface", iface, "Interface to init");
+    app.add_option("-t,--report-interval-time", time_per_report, "Time Per Report");
     app.add_flag("--check-time", check_timestamp, "Report back differences in timestamp");
-    app.add_flag("-p", per_stream_reports, "Detailed per stream reports");
+    app.add_flag("-p,--per-stream-reports", per_stream_reports, "Detailed per stream reports");
+    
     CLI11_PARSE(app, argc, argv);
 
+    // Validate arguments
+    fmt::print("ips   : {}\n", fmt::join(garp_ip_addresses," | "));
+    fmt::print("pcies   : {}\n", fmt::join(pcie_addresses," | "));
+
+    std::regex re_ipv4("[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}");
+    std::regex re_pcie("^0{0,4}:[a-fA-F0-9]{2}:[a-fA-F0-9]{2}.[0-9]$");
+    
+    fmt::print("IP addresses\n");
+    bool all_ip_ok = true;
+    for( const auto& ip: garp_ip_addresses) {
+        bool ip_ok = std::regex_match(ip, re_ipv4);
+        fmt::print("- {} {}\n", ip, ip_ok);
+        all_ip_ok &= ip_ok;
+    }
+
+    fmt::print("PCIE addresses\n");
+    bool all_pcie_ok = true;
+    for( const auto& pcie: pcie_addresses) {
+        bool pcie_ok = std::regex_match(pcie, re_pcie);
+        fmt::print("- {} {}\n", pcie, pcie_ok);
+        all_pcie_ok &= pcie_ok;
+    }
+
+    if (!all_ip_ok or !all_pcie_ok) {
+        return -1;
+    }
+
+
     //    define function to be called when ctrl+c is called.
     std::signal(SIGINT, signal_callback_handler);
 
     std::vector<std::string> eal_args;
-    eal_args.push_back("dpdklibds_test_frame_receiver");
-
+    eal_args.push_back("dpdklibds_test_garp");
+    for( const auto& pcie: pcie_addresses) {
+        eal_args.push_back("-a");
+        eal_args.push_back(pcie);
+    }
 
-    // initialise eal with constructed argc and argv
-    std::vector<char*> vec_argv = construct_argv(eal_args);
-    char** constructed_argv = vec_argv.data();
-    int constructed_argc = eal_args.size();
 
-    int ret = rte_eal_init(constructed_argc, constructed_argv);
-    if (ret < 0) {rte_exit(EXIT_FAILURE, "Error with EAL initialization\n");}
+    dunedaq::dpdklibs::ealutils::init_eal(eal_args);
 
     auto n_ifaces = rte_eth_dev_count_avail();
     fmt::print("# of available ifaces: {}\n", n_ifaces);
@@ -374,9 +448,10 @@ int main(int argc, char** argv){
     // Allocate pools and mbufs per queue
     std::map<int, std::unique_ptr<rte_mempool>> mbuf_pools;
     std::map<int, struct rte_mbuf **> bufs;
-    uint16_t n_rx_qs = 1;
-    const uint16_t rx_ring_size = 1024;
-    const uint16_t tx_ring_size = 1024;
+    const uint16_t n_rx_qs = 1;
+    const uint16_t n_tx_qs = 1;
+    const uint16_t rx_ring_size = 2048;
+    const uint16_t tx_ring_size = 2048;
 
     std::cout << "Allocating pools and mbufs.\n";
     for (size_t i=0; i<n_rx_qs; ++i) {
@@ -390,9 +465,9 @@ int main(int argc, char** argv){
 
     // Setting up only one iface
     fmt::print("Initialize only iface {}!\n", iface);
-    ealutils::iface_init(iface, n_rx_qs, 0, rx_ring_size, tx_ring_size, mbuf_pools); // just init iface, no TX queues
+    ealutils::iface_init(iface, n_rx_qs, n_tx_qs, rx_ring_size, tx_ring_size, mbuf_pools); // just init iface, no TX queues
 
-    lcore_main(mbuf_pools[0].get(), iface, time_per_report);
+    lcore_main(mbuf_pools[0].get(), iface, time_per_report, garp_ip_addresses);
 
     rte_eal_cleanup();