IfaceWrapper.hpp

Go to the documentation of this file.

#ifndef DPDKLIBS_SRC_IFACEWRAPPER_HPP_
#define DPDKLIBS_SRC_IFACEWRAPPER_HPP_
 
// #include "dpdklibs/nicreader/Structs.hpp"
#include "confmodel/NetworkDevice.hpp"
 
#include "SourceConcept.hpp"
#include "dpdklibs/EALSetup.hpp"
#include "dpdklibs/XstatsHelper.hpp"
#include "dpdklibs/arp/ARP.hpp"
#include "dpdklibs/ipv4_addr.hpp"
#include "dpdklibs/udp/PacketCtor.hpp"
#include "dpdklibs/udp/Utils.hpp"
 
#include <confmodel/Session.hpp>
// #include <confmodel/NetworkDevice.hpp>
#include "appmodel/DPDKReceiver.hpp"
#include "appmodel/NWDetDataSender.hpp"
 
#include <nlohmann/json.hpp>
 
#include "logging/Logging.hpp" // NOTE: if ISSUES ARE DECLARED BEFORE include logging/Logging.hpp, TLOG_DEBUG<<issue wont work.
#include <ers/ers.hpp>
 
#include <memory>
#include <set>
#include <sstream>
#include <string>
 
#include <folly/ProducerConsumerQueue.h>
 
namespace dunedaq {
 
ERS_DECLARE_ISSUE(dpdklibs, MetricPublishFailed, "Field " << field << " was not reported", ((std::string)field))
 
ERS_DECLARE_ISSUE(dpdklibs,
                  UnexpectedStreamID,
                  "Unexpected stream ID " << src_id << " in UDP payoad. Total counter: " << counter,
                  ((int)src_id)((size_t)counter))
 
namespace dpdklibs {
 
class IfaceWrapper : public opmonlib::MonitorableObject
{
public:
  using sid_to_source_map_t = std::map<int, std::shared_ptr<SourceConcept>>;
 
  IfaceWrapper(uint iface_id,
               const appmodel::DPDKReceiver* receiver,
               const std::vector<const appmodel::NWDetDataSender*>& senders,
               const std::vector<const confmodel::DetectorStream*>& active_streams,
               sid_to_source_map_t& sources,
               std::atomic<bool>& run_marker);
  ~IfaceWrapper();
 
  IfaceWrapper(const IfaceWrapper&) = delete;            
  IfaceWrapper& operator=(const IfaceWrapper&) = delete; 
  IfaceWrapper(IfaceWrapper&&) = delete;                 
  IfaceWrapper& operator=(IfaceWrapper&&) = delete;      
 
  // void init();
  void start();
  void stop();
 
  void generate_opmon_data() override;
 
  void allocate_mbufs();
  void setup_interface();
  void setup_flow_steering();
  void setup_xstats();
  void stop_xstats();
 
  void enable_flow() { m_lcore_enable_flow.store(true); }
  void disable_flow() { m_lcore_enable_flow.store(false); }
 
  const std::vector<uint16_t>& get_rte_cores() const { return m_rte_cores; }
 
protected:
  // iface_conf_t m_cfg;
  int m_iface_id;
  std::string m_iface_id_str;
  bool m_configured;
 
  bool m_with_flow;
  bool m_prom_mode;
  std::vector<std::string> m_ip_addr;
  std::vector<rte_be32_t> m_ip_addr_bin;
  std::string m_mac_addr;
  int m_socket_id;
  int m_mtu;
  unsigned m_max_block_words;
  uint16_t m_rx_ring_size;
  uint16_t m_tx_ring_size;
  int m_num_mbufs;
  int m_burst_size;
  uint32_t m_lcore_sleep_ns;
  int m_mbuf_cache_size;
 
private:
  int m_num_ip_sources;
  int m_num_rx_cores;
  std::set<std::string> m_ips;
  std::set<int> m_rx_qs;
  std::set<int> m_tx_qs;
  std::vector<uint16_t> m_rte_cores;
 
  // CPU core ID -> [queue -> ip]
  std::map<int, std::map<int, std::string>> m_rx_core_map;
  unsigned m_arp_rx_queue = 0; // RS TODO: make it configurable, and queue use conf check for exclusiveness!
 
  // Lcore stop signal
  std::atomic<bool> m_lcore_quit_signal{ false };
 
  std::atomic<bool> m_lcore_enable_flow{ false };
 
  // Mbufs and pools
  std::map<int, std::unique_ptr<rte_mempool>> m_mbuf_pools;
  std::map<int, struct rte_mbuf**> m_bufs; // by queue
 
  // Stats by queues
  std::map<int, std::atomic<std::size_t>> m_num_frames_rxq;
  std::map<int, std::atomic<std::size_t>> m_num_bytes_rxq;
  std::map<int, std::atomic<std::size_t>> m_num_full_bursts;
  std::map<int, std::atomic<uint16_t>> m_max_burst_size;
 
  // Stats by rte_workers
  std::map<int, std::atomic<std::size_t>> m_num_unhandled_non_ipv4;
  std::map<int, std::atomic<std::size_t>> m_num_unhandled_non_udp;
  std::map<int, std::atomic<std::size_t>> m_num_unhandled_non_jumbo_udp;
 
  // Unexpected stream ID count
  std::map<int, std::atomic<std::size_t>> m_num_unexid_frames;
 
  // DPDK HW stats
  dpdklibs::IfaceXstats m_iface_xstats;
 
  // stream -> source id map indexed by queue id
  // queue -> [stream_id -> sid]
  std::map<int, std::map<uint, uint>> m_stream_id_to_source_id;
  sid_to_source_map_t& m_sources;
  bool m_strict_parsing{ true };
 
  // Run marker
  std::atomic<bool>& m_run_marker;
 
  // GARP
  std::unique_ptr<rte_mempool> m_garp_mbuf_pool;
  std::map<int, struct rte_mbuf**> m_garp_bufs;
  std::thread m_garp_thread;
  void garp_func();
  std::atomic<uint64_t> m_garps_sent{ 0 };
 
  // ARP
  std::unique_ptr<rte_mempool> m_arp_mbuf_pool;
  std::map<int, struct rte_mbuf**> m_arp_bufs;
  std::thread m_arp_thread;
  void arp_func();
  std::atomic<uint64_t> m_arps_sent{ 0 };
 
  // Lcore processor
  int rx_runner(void* arg __rte_unused);
  int arp_response_runner(void* arg __rte_unused);
 
  // Parse UDP payloads as DAQ frames
  void parse_udp_payload(int src_rx_q, char* payload, std::size_t size);
 
  // Pass through UDP payloads as is
  void passthrough_udp_payload(int src_rx_q, char* payload, std::size_t size);
};
 
} // namespace dpdklibs
} // namespace dunedaq
 
// #include "detail/IfaceWrapper.hxx"
 
#endif // DPDKLIBS_SRC_IFACEWRAPPER_HPP_