Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

first pass at upgrading the dpdk to the new readout schema #130

Merged
merged 1 commit into from
Jun 14, 2024
Merged

first pass at upgrading the dpdk to the new readout schema #130

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 94 additions & 37 deletions plugins/NICReceiver.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,14 @@
#include "appfwk/ConfigurationManager.hpp"
#include "appfwk/ModuleConfiguration.hpp"

#include "confmodel/DetectorToDaqConnection.hpp"

#include "appmodel/DataReader.hpp"
#include "confmodel/ReadoutInterface.hpp"
#include "appmodel/NICReceiverConf.hpp"
#include "appmodel/NICInterface.hpp"
#include "appmodel/NICInterfaceConfiguration.hpp"
#include "appmodel/NICStatsConf.hpp"
#include "appmodel/EthStreamParameters.hpp"
// #include "appmodel/NICInterfaceConfiguration.hpp"
// #include "appmodel/NICStatsConf.hpp"
// #include "appmodel/EthStreamParameters.hpp"
#include "confmodel/QueueWithId.hpp"

#include "logging/Logging.hpp"
Expand Down Expand Up @@ -133,46 +134,62 @@ NICReceiver::do_configure(const data_t& /*args*/)
//auto session = appfwk::ModuleManager::get()->session();
auto mdal = m_cfg->module<appmodel::DataReader>(get_name());
auto module_conf = mdal->get_configuration()->cast<appmodel::NICReceiverConf>();
auto res_set = mdal->get_interfaces();
auto res_set = mdal->get_connections();
// EAL setup
TLOG() << "Setting up EAL with params from config.";
std::vector<std::string> eal_params ;
eal_params.push_back("eal_cmdline");
eal_params.push_back("--proc-type=primary");

// Construct the pcie devices allowed mask
std::string first_pcie_addr;
bool is_first_pcie_addr = true;
std::vector<const confmodel::DetectorToDaqConnection*> d2d_conns;
for (auto res : res_set) {
auto interface = res->cast<appmodel::NICInterface>();
if (interface == nullptr) {
auto connection = res->cast<confmodel::DetectorToDaqConnection>();
if (connection == nullptr) {
dunedaq::readoutlibs::GenericConfigurationError err(
ERS_HERE, "NICReceiver configuration failed due expected but unavailable interface!");
ERS_HERE, "DetectorToDaqConnection configuration failed due expected but unavailable connection!"
);
ers::fatal(err);
throw err;
}
if (interface->disabled(*(m_cfg->configuration_manager()->session()))) {
if (connection->disabled(*(m_cfg->configuration_manager()->session()))) {
continue;
}

d2d_conns.push_back(connection);

auto receiver = connection->get_receiver()->cast<appmodel::DPDKReceiver>();
if (!receiver) {
throw dunedaq::readoutlibs::InitializationError(
ERS_HERE, fmt::format("Found {} of type {} in connection {} while expecting type DPDKReceiver", receiver->class_name(), receiver->UID(), connection->UID())
);
}

auto nic_interface = receiver->get_uses();

if (is_first_pcie_addr) {
first_pcie_addr = interface->get_rx_pcie_addr();
first_pcie_addr = nic_interface->get_pcie_addr();
is_first_pcie_addr = false;
}
eal_params.push_back("-a");
eal_params.push_back(interface->get_rx_pcie_addr());
eal_params.push_back(nic_interface->get_pcie_addr());
}



// Use the first pcie device id as file prefix
// FIXME: Review this strategy - should work in most of cases, but it could be
// confusing in configs with multiple interfaces
// confusing in configs with multiple connections
eal_params.push_back(fmt::format("--file-prefix={}", first_pcie_addr));

eal_params.push_back(module_conf->get_eal_args());

ealutils::init_eal(eal_params);

// Get available Interfaces from EAL
// Get available connections from EAL
auto available_ifaces = ifaceutils::get_num_available_ifaces();
TLOG() << "Number of available interfaces: " << available_ifaces;
TLOG() << "Number of available connections: " << available_ifaces;
for (unsigned int ifc_id=0; ifc_id<available_ifaces; ++ifc_id) {
std::string mac_addr_str = ifaceutils::get_iface_mac_str(ifc_id);
std::string pci_addr_str = ifaceutils::get_iface_pci_str(ifc_id);
Expand All @@ -181,33 +198,73 @@ NICReceiver::do_configure(const data_t& /*args*/)
TLOG() << "Available iface with MAC=" << mac_addr_str << " PCIe=" << pci_addr_str << " logical ID=" << ifc_id;
}

for (auto res : res_set) {
auto interface = res->cast<appmodel::NICInterface>();
if (interface == nullptr) {
dunedaq::readoutlibs::GenericConfigurationError err(
ERS_HERE, "NICReceiver configuration failed due expected but unavailable interface!");
ers::fatal(err);
throw err;
}
if (interface->disabled(*(m_cfg->configuration_manager()->session()))) {
continue;
for (auto d2d_conn : d2d_conns) {
auto dpdk_receiver = d2d_conn->get_receiver()->cast<appmodel::DPDKReceiver>();
auto senders = d2d_conn->get_senders();
std::vector<const appmodel::NWDetDataSender*> nw_senders;
for ( auto sender : d2d_conn->get_senders() ) {
auto nw_sender = sender->cast<appmodel::NWDetDataSender>();
if ( !nw_sender ) {
throw dunedaq::readoutlibs::InitializationError(
ERS_HERE, fmt::format("Found {} of type {} in connection {} while expecting type NWDetDataSender", dpdk_receiver->class_name(), dpdk_receiver->UID(), d2d_conn->UID())
);
}

if ( nw_sender->disabled(*(m_cfg->configuration_manager()->session())) ) {
continue;
}
nw_senders.push_back(nw_sender);
}

if ((m_mac_to_id_map.count(interface->get_rx_mac()) != 0) && (m_pci_to_id_map.count(interface->get_rx_pcie_addr()) != 0)) {
m_mac_to_id_map[interface->get_rx_mac()] = interface->get_rx_iface();
m_ifaces[interface->get_rx_iface()] = std::make_unique<IfaceWrapper>(interface, m_sources, m_run_marker);
//m_ifaces[interface->get_rx_iface()] = conf(iface_cfg);
m_ifaces[interface->get_rx_iface()]->allocate_mbufs();
m_ifaces[interface->get_rx_iface()]->setup_interface();
m_ifaces[interface->get_rx_iface()]->setup_flow_steering();
m_ifaces[interface->get_rx_iface()]->setup_xstats();
} else {
TLOG() << "No available interface with MAC=" << interface->get_rx_mac();
throw dunedaq::readoutlibs::InitializationError(
ERS_HERE, "NICReceiver configuration failed due expected but unavailable interface!");
auto nic_interface = dpdk_receiver->get_uses();

if ((m_mac_to_id_map.count(nic_interface->get_mac_address()) == 0) || (m_pci_to_id_map.count(nic_interface->get_pcie_addr()) == 0)) {
TLOG() << "No available interface with MAC=" << nic_interface->get_mac_address();
throw dunedaq::readoutlibs::InitializationError(
ERS_HERE, "NICReceiver configuration failed due expected but unavailable interface!"
);
}

auto iface_id = nic_interface->get_iface();
auto mac_addr = nic_interface->get_mac_address();

m_mac_to_id_map[mac_addr] = iface_id;

m_ifaces[iface_id] = std::make_unique<IfaceWrapper>(dpdk_receiver, nw_senders, m_sources, m_run_marker);
m_ifaces[iface_id]->allocate_mbufs();
m_ifaces[iface_id]->setup_interface();
m_ifaces[iface_id]->setup_flow_steering();
m_ifaces[iface_id]->setup_xstats();

}

// for (auto res : res_set) {
// auto connection = res->cast<appmodel::NICconnection>();
// if (connection == nullptr) {
// dunedaq::readoutlibs::GenericConfigurationError err(
// ERS_HERE, "NICReceiver configuration failed due expected but unavailable connection!");
// ers::fatal(err);
// throw err;
// }
// if (connection->disabled(*(m_cfg->configuration_manager()->session()))) {
// continue;
// }

// if ((m_mac_to_id_map.count(connection->get_rx_mac()) != 0) && (m_pci_to_id_map.count(connection->get_rx_pcie_addr()) != 0)) {
// m_mac_to_id_map[connection->get_rx_mac()] = connection->get_rx_iface();
// m_ifaces[connection->get_rx_iface()] = std::make_unique<IfaceWrapper>(connection, m_sources, m_run_marker);
// //m_ifaces[connection->get_rx_iface()] = conf(iface_cfg);
// m_ifaces[connection->get_rx_iface()]->allocate_mbufs();
// m_ifaces[connection->get_rx_iface()]->setup_connection();
// m_ifaces[connection->get_rx_iface()]->setup_flow_steering();
// m_ifaces[connection->get_rx_iface()]->setup_xstats();
// } else {
// TLOG() << "No available interface with MAC=" << interface->get_rx_mac();
// throw dunedaq::readoutlibs::InitializationError(
// ERS_HERE, "NICReceiver configuration failed due expected but unavailable interface!");
// }
// }

if (!m_run_marker.load()) {
set_running(true);
TLOG() << "Starting iface wrappers.";
Expand Down
159 changes: 82 additions & 77 deletions src/IfaceWrapper.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,13 @@
#include "IfaceWrapper.hpp"

#include "appfwk/ConfigurationManager.hpp"
#include "confmodel/DROStreamConf.hpp"
#include "confmodel/StreamParameters.hpp"
// #include "confmodel/DROStreamConf.hpp"
// #include "confmodel/StreamParameters.hpp"
#include "confmodel/GeoId.hpp"
#include "appmodel/NICInterface.hpp"
#include "appmodel/NICInterfaceConfiguration.hpp"
#include "appmodel/NICStatsConf.hpp"
#include "appmodel/EthStreamParameters.hpp"
// #include "appmodel/NICInterface.hpp"
// #include "appmodel/NICInterfaceConfiguration.hpp"
// #include "appmodel/NICStatsConf.hpp"
// #include "appmodel/EthStreamParameters.hpp"

#include <chrono>
#include <memory>
Expand All @@ -47,82 +47,87 @@ enum
namespace dunedaq {
namespace dpdklibs {

IfaceWrapper::IfaceWrapper(const appmodel::NICInterface *iface_cfg, source_to_sink_map_t& sources, std::atomic<bool>& run_marker)
IfaceWrapper::IfaceWrapper(
const appmodel::DPDKReceiver* receiver,
const std::vector<const appmodel::NWDetDataSender*>& senders,
source_to_sink_map_t& sources,
std::atomic<bool>& run_marker
)
: m_sources(sources)
, m_run_marker(run_marker)
{
//auto iface_cfg = appfwk::ConfigurationManager::get()->get_dal<NICInterface>(iface_name);
m_iface_id = iface_cfg->get_rx_iface();
m_mac_addr = iface_cfg->get_rx_mac();
m_ip_addr = iface_cfg->get_rx_ip();
IpAddr ip_addr_struct(m_ip_addr);
m_ip_addr_bin = udp::ip_address_dotdecimal_to_binary(
ip_addr_struct.addr_bytes[3],
ip_addr_struct.addr_bytes[2],
ip_addr_struct.addr_bytes[1],
ip_addr_struct.addr_bytes[0]
);

m_with_flow = iface_cfg->get_configuration()->get_flow_control();
m_prom_mode = iface_cfg->get_configuration()->get_promiscuous_mode();;
m_mtu = iface_cfg->get_configuration()->get_mtu();
m_rx_ring_size = iface_cfg->get_configuration()->get_rx_ring_size();
m_tx_ring_size = iface_cfg->get_configuration()->get_tx_ring_size();
m_num_mbufs = iface_cfg->get_configuration()->get_num_bufs();
m_burst_size = iface_cfg->get_configuration()->get_burst_size();
m_mbuf_cache_size = iface_cfg->get_configuration()->get_mbuf_cache_size();

m_lcore_sleep_ns = iface_cfg->get_configuration()->get_lcore_sleep_us() * 1000;
m_socket_id = rte_eth_dev_socket_id(m_iface_id);

//m_iface_id_str = "iface-" + std::to_string(m_iface_id);
m_iface_id_str = iface_cfg->UID();

// iterate through active streams
//auto session = appfwk::ConfigurationManager()->session();
auto res_set = iface_cfg->get_contains();
for (const auto res : res_set) {
// m_iface_id = iface_cfg->get_rx_iface();
// m_mac_addr = iface_cfg->get_rx_mac();
// m_ip_addr = iface_cfg->get_rx_ip();
// IpAddr ip_addr_struct(m_ip_addr);
// m_ip_addr_bin = udp::ip_address_dotdecimal_to_binary(
// ip_addr_struct.addr_bytes[3],
// ip_addr_struct.addr_bytes[2],
// ip_addr_struct.addr_bytes[1],
// ip_addr_struct.addr_bytes[0]
// );

// m_with_flow = iface_cfg->get_configuration()->get_flow_control();
// m_prom_mode = iface_cfg->get_configuration()->get_promiscuous_mode();;
// m_mtu = iface_cfg->get_configuration()->get_mtu();
// m_rx_ring_size = iface_cfg->get_configuration()->get_rx_ring_size();
// m_tx_ring_size = iface_cfg->get_configuration()->get_tx_ring_size();
// m_num_mbufs = iface_cfg->get_configuration()->get_num_bufs();
// m_burst_size = iface_cfg->get_configuration()->get_burst_size();
// m_mbuf_cache_size = iface_cfg->get_configuration()->get_mbuf_cache_size();

// m_lcore_sleep_ns = iface_cfg->get_configuration()->get_lcore_sleep_us() * 1000;
// m_socket_id = rte_eth_dev_socket_id(m_iface_id);

// //m_iface_id_str = "iface-" + std::to_string(m_iface_id);
// m_iface_id_str = iface_cfg->UID();

// // iterate through active streams
// //auto session = appfwk::ConfigurationManager()->session();
// auto res_set = iface_cfg->get_contains();
// for (const auto res : res_set) {

auto stream = res->cast<confmodel::DROStreamConf>();
if (stream == nullptr) {
dunedaq::readoutlibs::GenericConfigurationError err(
ERS_HERE, std::string("NICInterface contains resources other than DROStreamConf!"));
throw err;
}
if(sources.find(stream->get_source_id()) == sources.end()) {
TLOG() << "Sink for source_id "<< stream->get_source_id() << " not initialized!";
continue;
}
auto stream_params = stream->get_stream_params()->cast<appmodel::EthStreamParameters>();

auto src_ip = stream_params->get_tx_ip();
auto rx_q = stream_params->get_rx_queue();
auto lcore = stream_params->get_lcore();
m_ips.insert(src_ip);
m_rx_qs.insert(rx_q);
m_lcores.insert(lcore);

m_num_frames_rxq[rx_q] = { 0 };
m_num_bytes_rxq[rx_q] = { 0 };

m_rx_core_map[lcore][rx_q] = src_ip;

auto stream_id = stream->get_geo_id()->get_stream_id();

m_stream_to_source_id[rx_q][stream_id] = stream->get_source_id();
}

// Log mapping
for (auto const& [lcore, rx_qs] : m_rx_core_map) {
TLOG() << "Lcore=" << lcore << " handles: ";
for (auto const& [rx_q, src_ip] : rx_qs) {
TLOG() << " rx_q=" << rx_q << " src_ip=" << src_ip;
}
}

// Adding single TX queue for ARP responses
TLOG() << "Append TX_Q=0 for ARP responses.";
m_tx_qs.insert(0);
// auto stream = res->cast<confmodel::DROStreamConf>();
// if (stream == nullptr) {
// dunedaq::readoutlibs::GenericConfigurationError err(
// ERS_HERE, std::string("NICInterface contains resources other than DROStreamConf!"));
// throw err;
// }
// if(sources.find(stream->get_source_id()) == sources.end()) {
// TLOG() << "Sink for source_id "<< stream->get_source_id() << " not initialized!";
// continue;
// }
// auto stream_params = stream->get_stream_params()->cast<appmodel::EthStreamParameters>();

// auto src_ip = stream_params->get_tx_ip();
// auto rx_q = stream_params->get_rx_queue();
// auto lcore = stream_params->get_lcore();
// m_ips.insert(src_ip);
// m_rx_qs.insert(rx_q);
// m_lcores.insert(lcore);

// m_num_frames_rxq[rx_q] = { 0 };
// m_num_bytes_rxq[rx_q] = { 0 };

// m_rx_core_map[lcore][rx_q] = src_ip;

// auto stream_id = stream->get_geo_id()->get_stream_id();

// m_stream_to_source_id[rx_q][stream_id] = stream->get_source_id();
// }

// // Log mapping
// for (auto const& [lcore, rx_qs] : m_rx_core_map) {
// TLOG() << "Lcore=" << lcore << " handles: ";
// for (auto const& [rx_q, src_ip] : rx_qs) {
// TLOG() << " rx_q=" << rx_q << " src_ip=" << src_ip;
// }
// }

// // Adding single TX queue for ARP responses
// TLOG() << "Append TX_Q=0 for ARP responses.";
// m_tx_qs.insert(0);

}

Expand Down
8 changes: 4 additions & 4 deletions src/IfaceWrapper.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,9 @@
#include "SourceConcept.hpp"

#include <confmodel/Session.hpp>
#include <appmodel/NICInterface.hpp>
// #include <appmodel/NICInterface.hpp>
#include "appmodel/DPDKReceiver.hpp"
#include "appmodel/NWDetDataSender.hpp"

#include <nlohmann/json.hpp>

Expand All @@ -41,11 +43,9 @@ namespace dpdklibs {
class IfaceWrapper
{
public:
//using iface_conf_t = dunedaq::dpdklibs::nicreader::Interface;
//using source_conf_t = dunedaq::dpdklibs::nicreader::Source;
using source_to_sink_map_t = std::map<int, std::unique_ptr<SourceConcept>>;

IfaceWrapper(const appmodel::NICInterface* interface, source_to_sink_map_t& sources, std::atomic<bool>& run_marker);
IfaceWrapper(const appmodel::DPDKReceiver* receiver, const std::vector<const appmodel::NWDetDataSender*>& senders, source_to_sink_map_t& sources, std::atomic<bool>& run_marker);
~IfaceWrapper();

IfaceWrapper(const IfaceWrapper&) = delete; ///< IfaceWrapper is not copy-constructible
Expand Down
Loading