uhd: Introduce I/O service manager

- Implement I/O service detach link methods
- The I/O service manager instantiates new I/O services or connects
  links to existing I/O services based on options provided by the user
  in stream_args.
- Add a streamer ID parameter to methods to create transports so that
  the I/O service manager can group transports appropriately when using
  offload threads.
- Change X300 and MPMD to use I/O service manager to connect links to
  I/O services.
- There is now a single I/O service manager per rfnoc_graph (and it is
  also stored in the graph)
- The I/O service manager now also knows the device args for the
  rfnoc_graph it was created with, and can make decisions based upon
  those (e.g, use a specific I/O service for DPDK, share cores between
  streamers, etc.)
- The I/O Service Manager does not get any decision logic with this
  commit, though
- The MB ifaces for mpmd and x300 now access this global I/O service
  manager
- Add configuration of link parameters with overrides

Co-Authored-By: Martin Braun <martin.braun@ettus.com>
Co-Authored-By: Aaron Rossetto <aaron.rossetto@ni.com>

4 files changed, 105 insertions, 45 deletions

diff --git a/host/lib/usrp/mpmd/mpmd_link_if_ctrl_udp.cpp b/host/lib/usrp/mpmd/mpmd_link_if_ctrl_udp.cpp
index 0e651a996..a87a9cada 100644
--- a/host/lib/usrp/mpmd/mpmd_link_if_ctrl_udp.cpp
+++ b/host/lib/usrp/mpmd/mpmd_link_if_ctrl_udp.cpp
@@ -7,11 +7,13 @@
 #include "mpmd_link_if_ctrl_udp.hpp"
 #include "mpmd_impl.hpp"
 #include "mpmd_link_if_mgr.hpp"
+#include <uhd/rfnoc/constants.hpp>
 #include <uhd/transport/udp_constants.hpp>
 #include <uhd/transport/udp_simple.hpp>
 #include <uhd/transport/udp_zero_copy.hpp>
-#include <uhdlib/transport/inline_io_service.hpp>
+#include <uhdlib/rfnoc/rfnoc_common.hpp>
 #include <uhdlib/transport/udp_boost_asio_link.hpp>
+#include <uhdlib/transport/udp_common.hpp>
 #include <uhdlib/utils/narrow.hpp>
 #include <string>
 
@@ -26,9 +28,8 @@ namespace {
 
 //! Maximum CHDR packet size in bytes
 const size_t MPMD_10GE_DATA_FRAME_MAX_SIZE = 8000;
-
-//! Maximum CHDR packet size in bytes
-const size_t MPMD_10GE_ASYNCMSG_FRAME_MAX_SIZE = 1472;
+const size_t MPMD_1GE_DATA_FRAME_MAX_SIZE     = 1472;
+const size_t MPMD_1GE_ASYNCMSG_FRAME_MAX_SIZE = 1472;
 
 //! Number of send/recv frames
 const size_t MPMD_ETH_NUM_FRAMES = 32;
@@ -194,8 +195,6 @@ size_t discover_mtu(const std::string& address,
 mpmd_link_if_ctrl_udp::mpmd_link_if_ctrl_udp(const uhd::device_addr_t& mb_args,
     const mpmd_link_if_mgr::xport_info_list_t& xport_info)
     : _mb_args(mb_args)
-    , _recv_args(filter_args(mb_args, "recv"))
-    , _send_args(filter_args(mb_args, "send"))
     , _udp_info(get_udp_info_from_xport_info(xport_info))
     , _mtu(MPMD_10GE_DATA_FRAME_MAX_SIZE)
 {
@@ -228,36 +227,52 @@ mpmd_link_if_ctrl_udp::mpmd_link_if_ctrl_udp(const uhd::device_addr_t& mb_args,
  * API
  *****************************************************************************/
 uhd::transport::both_links_t mpmd_link_if_ctrl_udp::get_link(const size_t link_idx,
-    const uhd::transport::link_type_t /*link_type*/,
-    const uhd::device_addr_t& /*link_args*/)
+    const uhd::transport::link_type_t link_type,
+    const uhd::device_addr_t& link_args)
 {
     UHD_ASSERT_THROW(link_idx < _available_addrs.size());
     const std::string ip_addr  = _available_addrs.at(link_idx);
     const std::string udp_port = _udp_info.at(ip_addr).udp_port;
 
-    /* FIXME: Should have common infrastructure for creating I/O services */
-    auto io_srv = uhd::transport::inline_io_service::make();
-    link_params_t link_params;
-    link_params.num_recv_frames = MPMD_ETH_NUM_FRAMES; // FIXME
-    link_params.num_send_frames = MPMD_ETH_NUM_FRAMES; // FIXME
-    link_params.recv_frame_size = get_mtu(uhd::RX_DIRECTION); // FIXME
-    link_params.send_frame_size = get_mtu(uhd::TX_DIRECTION); // FIXME
-    link_params.recv_buff_size  = MPMD_BUFFER_DEPTH * MAX_RATE_10GIGE; // FIXME
-    link_params.send_buff_size  = MPMD_BUFFER_DEPTH * MAX_RATE_10GIGE; // FIXME
-    auto link                   = uhd::transport::udp_boost_asio_link::make(ip_addr,
+    const size_t link_rate = get_link_rate(link_idx);
+    link_params_t default_link_params;
+    default_link_params.num_send_frames = MPMD_ETH_NUM_FRAMES;
+    default_link_params.num_recv_frames = MPMD_ETH_NUM_FRAMES;
+    default_link_params.send_frame_size = (link_rate == MAX_RATE_10GIGE)
+                                              ? MPMD_10GE_DATA_FRAME_MAX_SIZE
+                                              : (link_rate == MAX_RATE_1GIGE)
+                                                    ? MPMD_1GE_DATA_FRAME_MAX_SIZE
+                                                    : get_mtu(uhd::TX_DIRECTION);
+    default_link_params.recv_frame_size = (link_rate == MAX_RATE_10GIGE)
+                                              ? MPMD_10GE_DATA_FRAME_MAX_SIZE
+                                              : (link_rate == MAX_RATE_1GIGE)
+                                                    ? MPMD_1GE_DATA_FRAME_MAX_SIZE
+                                                    : get_mtu(uhd::RX_DIRECTION);
+    default_link_params.send_buff_size  = get_link_rate(link_idx) * MPMD_BUFFER_DEPTH;
+    default_link_params.recv_buff_size  = get_link_rate(link_idx) * MPMD_BUFFER_DEPTH;
+
+    link_params_t link_params = calculate_udp_link_params(link_type,
+        get_mtu(uhd::TX_DIRECTION),
+        get_mtu(uhd::RX_DIRECTION),
+        default_link_params,
+        _mb_args,
+        link_args);
+
+    // Enforce a minimum bound of the number of receive and send frames.
+    link_params.num_send_frames = std::max(uhd::rfnoc::MIN_NUM_FRAMES, link_params.num_send_frames);
+    link_params.num_recv_frames = std::max(uhd::rfnoc::MIN_NUM_FRAMES, link_params.num_recv_frames);
+
+    auto link = uhd::transport::udp_boost_asio_link::make(ip_addr,
         udp_port,
         link_params,
-        link_params.recv_buff_size, // FIXME
-        link_params.send_buff_size); // FIXME
-    io_srv->attach_send_link(link);
-    io_srv->attach_recv_link(link);
-    return std::tuple<io_service::sptr,
-        send_link_if::sptr,
+        link_params.recv_buff_size,
+        link_params.send_buff_size);
+    return std::tuple<send_link_if::sptr,
         size_t,
         recv_link_if::sptr,
         size_t,
         bool>(
-        io_srv, link, link_params.send_buff_size, link, link_params.recv_buff_size, true);
+        link, link_params.send_buff_size, link, link_params.recv_buff_size, true);
 }
 
 size_t mpmd_link_if_ctrl_udp::get_num_links() const
@@ -277,3 +292,4 @@ mpmd_link_if_ctrl_udp::get_packet_factory() const
 {
     return _pkt_factory;
 }
+
diff --git a/host/lib/usrp/mpmd/mpmd_link_if_ctrl_udp.hpp b/host/lib/usrp/mpmd/mpmd_link_if_ctrl_udp.hpp
index 4c8ecade7..33db83b47 100644
--- a/host/lib/usrp/mpmd/mpmd_link_if_ctrl_udp.hpp
+++ b/host/lib/usrp/mpmd/mpmd_link_if_ctrl_udp.hpp
@@ -45,8 +45,6 @@ public:
 
 private:
     const uhd::device_addr_t _mb_args;
-    const uhd::dict<std::string, std::string> _recv_args;
-    const uhd::dict<std::string, std::string> _send_args;
     //!
     udp_link_info_map _udp_info;
     //! A list of IP addresses we can connect our CHDR connections to
diff --git a/host/lib/usrp/mpmd/mpmd_mb_iface.cpp b/host/lib/usrp/mpmd/mpmd_mb_iface.cpp
index e713cc7a3..403e53949 100644
--- a/host/lib/usrp/mpmd/mpmd_mb_iface.cpp
+++ b/host/lib/usrp/mpmd/mpmd_mb_iface.cpp
@@ -14,9 +14,21 @@
 using namespace uhd::rfnoc;
 using namespace uhd::mpmd;
 
+static uhd::usrp::io_service_args_t get_default_io_srv_args()
+{
+    // TODO: Need better defaults, taking into account the link type and ensuring
+    // that the number of frames is appropriate
+    uhd::usrp::io_service_args_t args;
+    args.recv_offload = false;
+    args.send_offload = false;
+    return args;
+}
+
 mpmd_mboard_impl::mpmd_mb_iface::mpmd_mb_iface(
     const uhd::device_addr_t& mb_args, uhd::rpc_client::sptr rpc)
-    : _mb_args(mb_args), _rpc(rpc), _link_if_mgr(xport::mpmd_link_if_mgr::make(mb_args))
+    : _mb_args(mb_args)
+    , _rpc(rpc)
+    , _link_if_mgr(xport::mpmd_link_if_mgr::make(mb_args))
 {
     _remote_device_id = allocate_device_id();
     UHD_LOG_TRACE("MPMD::MB_IFACE", "Assigning device_id " << _remote_device_id);
@@ -153,16 +165,18 @@ uhd::rfnoc::chdr_ctrl_xport::sptr mpmd_mboard_impl::mpmd_mb_iface::make_ctrl_tra
                              + std::to_string(local_device_id));
     }
     const size_t link_idx = _local_device_id_map.at(local_device_id);
-    uhd::transport::io_service::sptr io_srv;
     uhd::transport::send_link_if::sptr send_link;
     uhd::transport::recv_link_if::sptr recv_link;
-    std::tie(io_srv, send_link, std::ignore, recv_link, std::ignore, std::ignore) =
+    std::tie(send_link, std::ignore, recv_link, std::ignore, std::ignore) =
         _link_if_mgr->get_link(
             link_idx, uhd::transport::link_type_t::CTRL, uhd::device_addr_t());
 
     /* Associate local device ID with the adapter */
     _adapter_map[local_device_id] = send_link->get_send_adapter_id();
 
+    auto io_srv = get_io_srv_mgr()->connect_links(
+        recv_link, send_link, transport::link_type_t::CTRL);
+
     auto pkt_factory = _link_if_mgr->get_packet_factory(link_idx);
     auto xport       = uhd::rfnoc::chdr_ctrl_xport::make(io_srv,
         send_link,
@@ -181,7 +195,8 @@ mpmd_mboard_impl::mpmd_mb_iface::make_rx_data_transport(
     const uhd::rfnoc::sep_id_pair_t& epids,
     const uhd::rfnoc::sw_buff_t pyld_buff_fmt,
     const uhd::rfnoc::sw_buff_t mdata_buff_fmt,
-    const uhd::device_addr_t& xport_args)
+    const uhd::device_addr_t& xport_args,
+    const std::string& streamer_id)
 {
     const uhd::rfnoc::sep_addr_t local_sep_addr = addrs.second;
 
@@ -192,12 +207,11 @@ mpmd_mboard_impl::mpmd_mb_iface::make_rx_data_transport(
     }
     const size_t link_idx = _local_device_id_map.at(local_sep_addr.first);
 
-    uhd::transport::io_service::sptr io_srv;
     uhd::transport::send_link_if::sptr send_link;
     uhd::transport::recv_link_if::sptr recv_link;
     bool lossy_xport;
     size_t recv_buff_size;
-    std::tie(io_srv, send_link, std::ignore, recv_link, recv_buff_size, lossy_xport) =
+    std::tie(send_link, std::ignore, recv_link, recv_buff_size, lossy_xport) =
         _link_if_mgr->get_link(
             link_idx, uhd::transport::link_type_t::RX_DATA, xport_args);
 
@@ -217,9 +231,12 @@ mpmd_mboard_impl::mpmd_mb_iface::make_rx_data_transport(
 
     stream_buff_params_t fc_headroom = {0, 0};
 
+    auto cfg_io_srv = get_io_srv_mgr()->connect_links(
+        recv_link, send_link, transport::link_type_t::CTRL);
+
     // Create the data transport
     auto pkt_factory = _link_if_mgr->get_packet_factory(link_idx);
-    auto fc_params   = chdr_rx_data_xport::configure_sep(io_srv,
+    auto fc_params   = chdr_rx_data_xport::configure_sep(cfg_io_srv,
         recv_link,
         send_link,
         pkt_factory,
@@ -231,7 +248,18 @@ mpmd_mboard_impl::mpmd_mb_iface::make_rx_data_transport(
         fc_freq,
         fc_headroom,
         lossy_xport);
-    auto rx_xport    = std::make_unique<chdr_rx_data_xport>(io_srv,
+
+    get_io_srv_mgr()->disconnect_links(recv_link, send_link);
+    cfg_io_srv.reset();
+
+    // Connect the links to an I/O service
+    auto io_srv = get_io_srv_mgr()->connect_links(recv_link,
+        send_link,
+        transport::link_type_t::RX_DATA,
+        usrp::read_io_service_args(xport_args, get_default_io_srv_args()),
+        streamer_id);
+
+    auto rx_xport = std::make_unique<chdr_rx_data_xport>(io_srv,
         recv_link,
         send_link,
         pkt_factory,
@@ -249,7 +277,8 @@ mpmd_mboard_impl::mpmd_mb_iface::make_tx_data_transport(
     const uhd::rfnoc::sep_id_pair_t& epids,
     const uhd::rfnoc::sw_buff_t pyld_buff_fmt,
     const uhd::rfnoc::sw_buff_t mdata_buff_fmt,
-    const uhd::device_addr_t& xport_args)
+    const uhd::device_addr_t& xport_args,
+    const std::string& streamer_id)
 {
     const uhd::rfnoc::sep_addr_t local_sep_addr = addrs.first;
 
@@ -260,11 +289,10 @@ mpmd_mboard_impl::mpmd_mb_iface::make_tx_data_transport(
     }
     const size_t link_idx = _local_device_id_map.at(local_sep_addr.first);
 
-    uhd::transport::io_service::sptr io_srv;
     uhd::transport::send_link_if::sptr send_link;
     uhd::transport::recv_link_if::sptr recv_link;
     bool lossy_xport;
-    std::tie(io_srv, send_link, std::ignore, recv_link, std::ignore, lossy_xport) =
+    std::tie(send_link, std::ignore, recv_link, std::ignore, lossy_xport) =
         _link_if_mgr->get_link(
             link_idx, uhd::transport::link_type_t::TX_DATA, xport_args);
 
@@ -275,8 +303,11 @@ mpmd_mboard_impl::mpmd_mb_iface::make_tx_data_transport(
     const double fc_freq_ratio     = 1.0 / 8;
     const double fc_headroom_ratio = 0;
 
+    auto cfg_io_srv = get_io_srv_mgr()->connect_links(
+        recv_link, send_link, transport::link_type_t::CTRL);
+
     auto pkt_factory         = _link_if_mgr->get_packet_factory(link_idx);
-    const auto buff_capacity = chdr_tx_data_xport::configure_sep(io_srv,
+    const auto buff_capacity = chdr_tx_data_xport::configure_sep(cfg_io_srv,
         recv_link,
         send_link,
         pkt_factory,
@@ -287,6 +318,16 @@ mpmd_mboard_impl::mpmd_mb_iface::make_tx_data_transport(
         fc_freq_ratio,
         fc_headroom_ratio);
 
+    get_io_srv_mgr()->disconnect_links(recv_link, send_link);
+    cfg_io_srv.reset();
+
+    // Connect the links to an I/O service
+    auto io_srv = get_io_srv_mgr()->connect_links(recv_link,
+        send_link,
+        transport::link_type_t::TX_DATA,
+        usrp::read_io_service_args(xport_args, get_default_io_srv_args()),
+        streamer_id);
+
     // Create the data transport
     auto tx_xport = std::make_unique<chdr_tx_data_xport>(io_srv,
         recv_link,
@@ -296,6 +337,5 @@ mpmd_mboard_impl::mpmd_mb_iface::make_tx_data_transport(
         send_link->get_num_send_frames(),
         buff_capacity);
 
-
     return tx_xport;
 }
diff --git a/host/lib/usrp/mpmd/mpmd_mb_iface.hpp b/host/lib/usrp/mpmd/mpmd_mb_iface.hpp
index 4e47dd35a..4e54cfc12 100644
--- a/host/lib/usrp/mpmd/mpmd_mb_iface.hpp
+++ b/host/lib/usrp/mpmd/mpmd_mb_iface.hpp
@@ -10,8 +10,9 @@
 #include "mpmd_impl.hpp"
 #include "mpmd_link_if_mgr.hpp"
 #include <uhdlib/rfnoc/mb_iface.hpp>
-#include <map>
+#include <uhdlib/usrp/common/io_service_mgr.hpp>
 #include <unordered_map>
+#include <map>
 
 namespace uhd { namespace mpmd {
 
@@ -33,7 +34,8 @@ public:
     uhd::endianness_t get_endianness(const uhd::rfnoc::device_id_t local_device_id);
     uhd::rfnoc::device_id_t get_remote_device_id();
     std::vector<uhd::rfnoc::device_id_t> get_local_device_ids();
-    uhd::transport::adapter_id_t get_adapter_id(const uhd::rfnoc::device_id_t local_device_id);
+    uhd::transport::adapter_id_t get_adapter_id(
+        const uhd::rfnoc::device_id_t local_device_id);
     void reset_network();
     uhd::rfnoc::clock_iface::sptr get_clock_iface(const std::string& clock_name);
     uhd::rfnoc::chdr_ctrl_xport::sptr make_ctrl_transport(
@@ -44,14 +46,16 @@ public:
         const uhd::rfnoc::sep_id_pair_t& epids,
         const uhd::rfnoc::sw_buff_t pyld_buff_fmt,
         const uhd::rfnoc::sw_buff_t mdata_buff_fmt,
-        const uhd::device_addr_t& xport_args);
+        const uhd::device_addr_t& xport_args,
+        const std::string& streamer_id);
     uhd::rfnoc::chdr_tx_data_xport::uptr make_tx_data_transport(
         uhd::rfnoc::mgmt::mgmt_portal& mgmt_portal,
         const uhd::rfnoc::sep_addr_pair_t& addrs,
         const uhd::rfnoc::sep_id_pair_t& epids,
         const uhd::rfnoc::sw_buff_t pyld_buff_fmt,
         const uhd::rfnoc::sw_buff_t mdata_buff_fmt,
-        const uhd::device_addr_t& xport_args);
+        const uhd::device_addr_t& xport_args,
+        const std::string& streamer_id);
 
 private:
     uhd::device_addr_t _mb_args;
@@ -59,8 +63,10 @@ private:
     xport::mpmd_link_if_mgr::uptr _link_if_mgr;
     uhd::rfnoc::device_id_t _remote_device_id;
     std::map<uhd::rfnoc::device_id_t, size_t> _local_device_id_map;
-    std::unordered_map<uhd::rfnoc::device_id_t, uhd::transport::adapter_id_t> _adapter_map;
+    std::unordered_map<uhd::rfnoc::device_id_t, uhd::transport::adapter_id_t>
+        _adapter_map;
     std::map<std::string, uhd::rfnoc::clock_iface::sptr> _clock_ifaces;
+    uhd::usrp::io_service_mgr::sptr _io_srv_mgr;
 };
 
 }} /* namespace uhd::mpmd */