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>

11 files changed, 341 insertions, 23 deletions

diff --git a/host/lib/include/uhdlib/rfnoc/graph_stream_manager.hpp b/host/lib/include/uhdlib/rfnoc/graph_stream_manager.hpp
index 8a721ea26..b9f4205ab 100644
--- a/host/lib/include/uhdlib/rfnoc/graph_stream_manager.hpp
+++ b/host/lib/include/uhdlib/rfnoc/graph_stream_manager.hpp
@@ -19,6 +19,7 @@
 #include <functional>
 #include <memory>
 #include <set>
+#include <string>
 #include <tuple>
 
 namespace uhd { namespace rfnoc {
@@ -114,6 +115,7 @@ public:
      * \param mdata_buff_fmt Datatype of SW buffer that holds the data metadata
      * \param adapter The preference for the adapter to use to get to the destination
      * \param xport_args The transport arguments
+     * \param streamer_id A unique identifier for the streamer that will own the transport
      * \return An transport instance
      */
     virtual chdr_rx_data_xport::uptr create_device_to_host_data_stream(
@@ -121,7 +123,8 @@ public:
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
         const uhd::transport::adapter_id_t adapter,
-        const device_addr_t& xport_args) = 0;
+        const device_addr_t& xport_args,
+        const std::string& streamer_id) = 0;
 
     /*! \brief Create a data stream going from the host to the device
      *
@@ -130,6 +133,7 @@ public:
      * \param mdata_buff_fmt Datatype of SW buffer that holds the data metadata
      * \param adapter The preference for the adapter to use to get to the destination
      * \param xport_args The transport arguments
+     * \param streamer_id A unique identifier for the streamer that will own the transport
      * \return An transport instance
      */
     virtual chdr_tx_data_xport::uptr create_host_to_device_data_stream(
@@ -137,7 +141,8 @@ public:
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
         const uhd::transport::adapter_id_t adapter,
-        const device_addr_t& xport_args) = 0;
+        const device_addr_t& xport_args,
+        const std::string& streamer_id) = 0;
 
     /*! \brief Get all the adapters that can reach the specified endpoint
      *
diff --git a/host/lib/include/uhdlib/rfnoc/link_stream_manager.hpp b/host/lib/include/uhdlib/rfnoc/link_stream_manager.hpp
index b1a934891..836a50dcf 100644
--- a/host/lib/include/uhdlib/rfnoc/link_stream_manager.hpp
+++ b/host/lib/include/uhdlib/rfnoc/link_stream_manager.hpp
@@ -126,13 +126,15 @@ public:
      * \param pyld_buff_fmt Datatype of SW buffer that holds the data payload
      * \param mdata_buff_fmt Datatype of SW buffer that holds the data metadata
      * \param xport_args The transport arguments
+     * \param streamer_id A unique identifier for the streamer that will own the transport
      * \return An transport instance
      */
     virtual chdr_tx_data_xport::uptr create_host_to_device_data_stream(
         const sep_addr_t dst_addr,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
-        const device_addr_t& xport_args) = 0;
+        const device_addr_t& xport_args,
+        const std::string& streamer_id) = 0;
 
     /*! \brief Create a data stream going from the device to the host
      *
@@ -140,13 +142,15 @@ public:
      * \param pyld_buff_fmt Datatype of SW buffer that holds the data payload
      * \param mdata_buff_fmt Datatype of SW buffer that holds the data metadata
      * \param xport_args The transport arguments
+     * \param streamer_id A unique identifier for the streamer that will own the transport
      * \return An transport instance
      */
     virtual chdr_rx_data_xport::uptr create_device_to_host_data_stream(
         const sep_addr_t src_addr,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
-        const device_addr_t& xport_args) = 0;
+        const device_addr_t& xport_args,
+        const std::string& streamer_id) = 0;
 
     static uptr make(const chdr::chdr_packet_factory& pkt_factory,
         mb_iface& mb_if,
diff --git a/host/lib/include/uhdlib/rfnoc/mb_iface.hpp b/host/lib/include/uhdlib/rfnoc/mb_iface.hpp
index abfc9d1c4..53f0897f9 100644
--- a/host/lib/include/uhdlib/rfnoc/mb_iface.hpp
+++ b/host/lib/include/uhdlib/rfnoc/mb_iface.hpp
@@ -7,6 +7,7 @@
 #ifndef INCLUDED_LIBUHD_MB_IFACE_HPP
 #define INCLUDED_LIBUHD_MB_IFACE_HPP
 
+#include <uhd/exception.hpp>
 #include <uhd/transport/adapter_id.hpp>
 #include <uhd/types/endianness.hpp>
 #include <uhdlib/rfnoc/chdr_ctrl_xport.hpp>
@@ -14,6 +15,7 @@
 #include <uhdlib/rfnoc/chdr_tx_data_xport.hpp>
 #include <uhdlib/rfnoc/clock_iface.hpp>
 #include <uhdlib/rfnoc/rfnoc_common.hpp>
+#include <uhdlib/usrp/common/io_service_mgr.hpp>
 #include <memory>
 
 namespace uhd { namespace rfnoc {
@@ -86,6 +88,27 @@ public:
     virtual std::shared_ptr<clock_iface> get_clock_iface(
         const std::string& clock_name) = 0;
 
+    /*! Set the IO service manager
+     *
+     */
+    void set_io_srv_mgr(uhd::usrp::io_service_mgr::sptr io_srv_mgr)
+    {
+        _io_srv_mgr = io_srv_mgr;
+    }
+
+    /*! Get the I/O Service Manager
+     *
+     * This function must be called by the implementations of the various
+     * make_*_transport() calls to get access to the global I/O Service Manager
+     */
+    uhd::usrp::io_service_mgr::sptr get_io_srv_mgr()
+    {
+        if (!_io_srv_mgr) {
+            throw uhd::runtime_error("I/O Service Manager not set for mb_iface!");
+        }
+        return _io_srv_mgr;
+    }
+
     /*! Create a control transport
      *
      * This is usually called once per motherboard, since there is only one
@@ -108,6 +131,7 @@ public:
      * \param pyld_buff_fmt Datatype of SW buffer that holds the data payload
      * \param mdata_buff_fmt Datatype of SW buffer that holds the data metadata
      * \param xport_args Transport configuration args
+     * \param streamer_id A unique identifier for the streamer that will own the transport
      * \return A CHDR RX data transport
      */
     virtual chdr_rx_data_xport::uptr make_rx_data_transport(
@@ -116,7 +140,8 @@ public:
         const sep_id_pair_t& epids,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
-        const device_addr_t& xport_args) = 0;
+        const device_addr_t& xport_args,
+        const std::string& streamer_id) = 0;
 
     /*! Create an TX data transport
      *
@@ -127,6 +152,7 @@ public:
      * \param pyld_buff_fmt Datatype of SW buffer that holds the data payload
      * \param mdata_buff_fmt Datatype of SW buffer that holds the data metadata
      * \param xport_args Transport configuration args
+     * \param streamer_id A unique identifier for the streamer that will own the transport
      * \return A CHDR TX data transport
      */
     virtual chdr_tx_data_xport::uptr make_tx_data_transport(
@@ -135,7 +161,11 @@ public:
         const sep_id_pair_t& epids,
         const uhd::rfnoc::sw_buff_t pyld_buff_fmt,
         const uhd::rfnoc::sw_buff_t mdata_buff_fmt,
-        const device_addr_t& xport_args) = 0;
+        const device_addr_t& xport_args,
+        const std::string& streamer_id) = 0;
+
+private:
+    uhd::usrp::io_service_mgr::sptr _io_srv_mgr;
 };
 
 }} /* namespace uhd::rfnoc */
diff --git a/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp b/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp
index c2ec4e0e3..226d4f069 100644
--- a/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp
+++ b/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp
@@ -95,6 +95,11 @@ constexpr uint32_t MAX_FC_FREQ_PKTS      = (uint32_t(1) << 24) - 1;
 constexpr uint64_t MAX_FC_HEADROOM_BYTES = (uint64_t(1) << 16) - 1;
 constexpr uint32_t MAX_FC_HEADROOM_PKTS  = (uint32_t(1) << 8) - 1;
 
+// RFNoC devices need a minimum of two frame buffers to be available from the
+// link--one for the data transport and one for the control transport to
+// simultaneously handle MGMT and STRC/STRS initialization packets.
+constexpr size_t MIN_NUM_FRAMES = 2;
+
 }} // namespace uhd::rfnoc
 
 #endif /* INCLUDED_RFNOC_RFNOC_COMMON_HPP */
diff --git a/host/lib/include/uhdlib/transport/inline_io_service.hpp b/host/lib/include/uhdlib/transport/inline_io_service.hpp
index f207d15a0..fe41b96b6 100644
--- a/host/lib/include/uhdlib/transport/inline_io_service.hpp
+++ b/host/lib/include/uhdlib/transport/inline_io_service.hpp
@@ -37,6 +37,9 @@ public:
     void attach_recv_link(recv_link_if::sptr link);
     void attach_send_link(send_link_if::sptr link);
 
+    void detach_recv_link(recv_link_if::sptr link);
+    void detach_send_link(send_link_if::sptr link);
+
     recv_io_if::sptr make_recv_client(recv_link_if::sptr data_link,
         size_t num_recv_frames,
         recv_callback_t cb,
@@ -102,8 +105,7 @@ private:
         inline_recv_cb* recv_io_cb, recv_link_if* recv_link, int32_t timeout_ms);
 
     /* Track whether link is muxed and the callback */
-    std::unordered_map<recv_link_if*,
-        std::tuple<inline_recv_mux*, inline_recv_cb*>>
+    std::unordered_map<recv_link_if*, std::tuple<inline_recv_mux*, inline_recv_cb*>>
         _recv_tbl;
 
     /* Shared ptr kept to avoid untimely release */
diff --git a/host/lib/include/uhdlib/transport/io_service.hpp b/host/lib/include/uhdlib/transport/io_service.hpp
index 69a3a523e..399b693dc 100644
--- a/host/lib/include/uhdlib/transport/io_service.hpp
+++ b/host/lib/include/uhdlib/transport/io_service.hpp
@@ -282,10 +282,19 @@ public:
      */
     virtual void attach_send_link(send_link_if::sptr link) = 0;
 
-    /* TODO: Cleanup functions
+    /*!
+     * Detach a recv_link_if previously attached to this I/O service.
+     *
+     * \param link the recv_link_if to detach
+     */
     virtual void detach_recv_link(recv_link_if::sptr link) = 0;
+
+    /*!
+     * Detach a send_link_if previously attached to this I/O service.
+     *
+     * \param link the send_link_if to detach
+     */
     virtual void detach_send_link(send_link_if::sptr link) = 0;
-    */
 
     /*!
      * Create a send_io_if so a transport may send packets through the link.
diff --git a/host/lib/include/uhdlib/transport/links.hpp b/host/lib/include/uhdlib/transport/links.hpp
index 64673f02f..09872b145 100644
--- a/host/lib/include/uhdlib/transport/links.hpp
+++ b/host/lib/include/uhdlib/transport/links.hpp
@@ -16,8 +16,7 @@ namespace uhd { namespace transport {
 enum class link_type_t { CTRL = 0, ASYNC_MSG, TX_DATA, RX_DATA };
 
 //! Contains all information regarding a link interface
-using both_links_t = std::tuple<uhd::transport::io_service::sptr,
-    uhd::transport::send_link_if::sptr,
+using both_links_t = std::tuple<uhd::transport::send_link_if::sptr,
     size_t, // num_send_frames
     uhd::transport::recv_link_if::sptr,
     size_t, // num_recv_frames
diff --git a/host/lib/include/uhdlib/transport/offload_io_service.hpp b/host/lib/include/uhdlib/transport/offload_io_service.hpp
index a7d9d211d..02231c502 100644
--- a/host/lib/include/uhdlib/transport/offload_io_service.hpp
+++ b/host/lib/include/uhdlib/transport/offload_io_service.hpp
@@ -8,6 +8,7 @@
 #define INCLUDED_UHDLIB_TRANSPORT_OFFLOAD_IO_SERVICE_HPP
 
 #include <uhdlib/transport/io_service.hpp>
+#include <vector>
 
 namespace uhd { namespace transport {
 
@@ -21,18 +22,9 @@ namespace uhd { namespace transport {
 class offload_io_service : public io_service
 {
 public:
-    enum client_type_t
-    {
-        RECV_ONLY,
-        SEND_ONLY,
-        BOTH_SEND_AND_RECV
-    };
+    enum client_type_t { RECV_ONLY, SEND_ONLY, BOTH_SEND_AND_RECV };
 
-    enum wait_mode_t
-    {
-        POLL,
-        BLOCK
-    };
+    enum wait_mode_t { POLL, BLOCK };
 
     /*!
      * Options for configuring offload I/O service
diff --git a/host/lib/include/uhdlib/transport/udp_common.hpp b/host/lib/include/uhdlib/transport/udp_common.hpp
index 5f5a18c27..6c87ef498 100644
--- a/host/lib/include/uhdlib/transport/udp_common.hpp
+++ b/host/lib/include/uhdlib/transport/udp_common.hpp
@@ -10,7 +10,10 @@
 
 #include <uhd/config.hpp>
 #include <uhd/exception.hpp>
+#include <uhd/rfnoc/constants.hpp>
+#include <uhd/types/device_addr.hpp>
 #include <uhd/utils/log.hpp>
+#include <uhdlib/transport/links.hpp>
 #include <boost/asio.hpp>
 #include <boost/format.hpp>
 #include <thread>
@@ -194,6 +197,78 @@ UHD_INLINE size_t resize_udp_socket_buffer_with_warning(
     return actual_size;
 }
 
+/*!
+ * Determines a set of values to use for a UDP CHDR link based on defaults and
+ * any overrides that the user may have provided. In cases where both device
+ * and stream arguments can be used to override a value, note that the stream
+ * argument will always take precedence.
+ *
+ * \param link_type the link type (CTRL, RX, TX) to calculate parameters for
+ * \param send_mtu the MTU of link for Tx cases
+ * \param recv_mtu the MTU of link for Rx cases
+ * \param default_link_params default values to use for the link parameters
+ * \param device_args device-level argument dictionary for overrides
+ * \param link_args argument dictionary with stream-level overrides (come from
+ *        stream params)
+ * \return Parameters to apply
+ */
+inline link_params_t calculate_udp_link_params(
+    const uhd::transport::link_type_t link_type,
+    const size_t send_mtu,
+    const size_t recv_mtu,
+    const link_params_t& default_link_params,
+    const uhd::device_addr_t& device_args,
+    const uhd::device_addr_t& link_args)
+{
+    // Apply any device-level overrides to the default values first.
+    // If the MTU is overridden, it will be capped to the value provided by
+    // the caller.
+    const size_t constrained_send_mtu =
+        std::min(send_mtu, device_args.cast<size_t>("mtu", send_mtu));
+    const size_t constrained_recv_mtu =
+        std::min(recv_mtu, device_args.cast<size_t>("mtu", recv_mtu));
+
+    link_params_t link_params;
+    link_params.num_send_frames =
+        device_args.cast<size_t>("num_send_frames", default_link_params.num_send_frames);
+    link_params.num_recv_frames =
+        device_args.cast<size_t>("num_recv_frames", default_link_params.num_recv_frames);
+    link_params.send_frame_size =
+        device_args.cast<size_t>("send_frame_size", default_link_params.send_frame_size);
+    link_params.recv_frame_size =
+        device_args.cast<size_t>("recv_frame_size", default_link_params.recv_frame_size);
+    link_params.send_buff_size =
+        device_args.cast<size_t>("send_buff_size", default_link_params.send_buff_size);
+    link_params.recv_buff_size =
+        device_args.cast<size_t>("recv_buff_size", default_link_params.recv_buff_size);
+
+    // Now apply stream-level overrides based on the link type.
+    if (link_type == link_type_t::CTRL) {
+        // Control links typically do not allow the number of frames to be
+        // configured.
+        link_params.num_recv_frames =
+            uhd::rfnoc::CMD_FIFO_SIZE / uhd::rfnoc::MAX_CMD_PKT_SIZE;
+    } else if (link_type == link_type_t::TX_DATA) {
+        // Note that the send frame size will be capped to the Tx MTU.
+        link_params.send_frame_size = link_args.cast<size_t>("send_frame_size",
+            std::min(link_params.send_frame_size, constrained_send_mtu));
+        link_params.num_send_frames =
+            link_args.cast<size_t>("num_send_frames", link_params.num_send_frames);
+        link_params.send_buff_size =
+            link_args.cast<size_t>("send_buff_size", link_params.send_buff_size);
+    } else if (link_type == link_type_t::RX_DATA) {
+        // Note that the receive frame size will be capped to the Rx MTU.
+        link_params.recv_frame_size = link_args.cast<size_t>("recv_frame_size",
+            std::min(link_params.recv_frame_size, constrained_recv_mtu));
+        link_params.num_recv_frames =
+            link_args.cast<size_t>("num_recv_frames", link_params.num_recv_frames);
+        link_params.recv_buff_size =
+            link_args.cast<size_t>("recv_buff_size", link_params.recv_buff_size);
+    }
+
+    return link_params;
+}
+
 
 }} // namespace uhd::transport
 
diff --git a/host/lib/include/uhdlib/usrp/common/io_service_args.hpp b/host/lib/include/uhdlib/usrp/common/io_service_args.hpp
new file mode 100644
index 000000000..a783cc825
--- /dev/null
+++ b/host/lib/include/uhdlib/usrp/common/io_service_args.hpp
@@ -0,0 +1,93 @@
+//
+// Copyright 2019 Ettus Research, a National Instruments brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#ifndef INCLUDED_LIBUHD_IO_SERVICE_ARGS_HPP
+#define INCLUDED_LIBUHD_IO_SERVICE_ARGS_HPP
+
+#include <uhd/types/device_addr.hpp>
+#include <boost/optional.hpp>
+
+namespace uhd { namespace usrp {
+
+/*! Struct containing user options for I/O services
+ *
+ * The I/O service manager supports the following args:
+ *
+ * recv_offload: set to "true" to use an offload thread for RX_DATA links, "false"
+ *               to use an inline I/O service.
+ * send_offload: set to "true" to use an offload thread for TX_DATA links, "false"
+ *               to use an inline I/O service.
+ * recv_offload_wait_mode: set to "poll" to use a polling strategy in the offload
+ *                         thread, set to "block" to use a blocking strategy.
+ * send_offload_wait_mode: set to "poll" to use a polling strategy in the offload
+ *                         thread, set to "block" to use a blocking strategy.
+ * num_poll_offload_threads: set to the total number of offload threads to use for
+ *                           RX_DATA and TX_DATA in this rfnoc_graph. New connections
+ *                           always go to the offload thread containing the fewest
+ *                           connections, with lowest numbered thread as a second
+ *                           criterion. The default is 1.
+ * recv_offload_cpu_<N>: an integer to specify cpu affinity of the offload thread.
+ *                       N indicates the thread instance, starting with 0 for each
+ *                       streamer and ending with the number of transport adapters
+ *                       minus one. Only used if the I/O service is configured to
+ *                       block.
+ * send_offload_cpu_<N>: an integer to specify cpu affinity of the offload thread.
+ *                       N indicates the thread instance, starting with 0 for each
+ *                       streamer and ending with the number of transport adapters
+ *                       minus one. Only used if the I/O service is configured to
+ *                       block.
+ * poll_offload_cpu_<N>: an integer to specify cpu affinity of the offload thread.
+ *                       N indicates the thread instance, starting with 0 and up to
+ *                       num_poll_offload_threads minus 1. Only used if the I/O
+ *                       service is configured to poll.
+ */
+struct io_service_args_t
+{
+    enum wait_mode_t { POLL, BLOCK };
+
+    //! Whether to offload streaming I/O to a worker thread
+    bool recv_offload = false;
+
+    //! Whether to offload streaming I/O to a worker thread
+    bool send_offload = false;
+
+    //! Whether the offload thread should poll or block
+    wait_mode_t recv_offload_wait_mode = BLOCK;
+
+    //! Whether the offload thread should poll or block
+    wait_mode_t send_offload_wait_mode = BLOCK;
+
+    //! Number of polling threads to use, if wait_mode is set to POLL
+    size_t num_poll_offload_threads = 1;
+
+    //! CPU affinity of offload threads, if wait_mode is set to BLOCK (one item
+    //! per thread)
+    std::vector<boost::optional<size_t>> recv_offload_thread_cpu;
+
+    //! CPU affinity of offload threads, if wait_mode is set to BLOCK (one item
+    //! per thread)
+    std::vector<boost::optional<size_t>> send_offload_thread_cpu;
+
+    //! CPU affinity of offload threads, if wait_mode is set to POLL (one item
+    //! per thread)
+    std::vector<boost::optional<size_t>> poll_offload_thread_cpu;
+};
+
+/*! Reads I/O service args from provided dictionary
+ *
+ * If an option is not specified in the dictionary, the default value of the
+ * struct above is returned.
+ *
+ * \param args The dictionary from which to read the I/O service args
+ * \param defaults Default values (not including boost::optional values)
+ * \return The I/O service args read
+ */
+io_service_args_t read_io_service_args(
+    const device_addr_t& args, const io_service_args_t& defaults);
+
+}} // namespace uhd::usrp
+
+#endif /* INCLUDED_LIBUHD_IO_SERVICE_ARGS_HPP */
diff --git a/host/lib/include/uhdlib/usrp/common/io_service_mgr.hpp b/host/lib/include/uhdlib/usrp/common/io_service_mgr.hpp
new file mode 100644
index 000000000..1093f7bec
--- /dev/null
+++ b/host/lib/include/uhdlib/usrp/common/io_service_mgr.hpp
@@ -0,0 +1,104 @@
+//
+// Copyright 2019 Ettus Research, a National Instruments Brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#ifndef INCLUDED_LIBUHD_IO_SERVICE_MGR_HPP
+#define INCLUDED_LIBUHD_IO_SERVICE_MGR_HPP
+
+#include <uhd/transport/adapter_id.hpp>
+#include <uhd/types/device_addr.hpp>
+#include <uhdlib/transport/io_service.hpp>
+#include <uhdlib/transport/link_if.hpp>
+#include <uhdlib/transport/links.hpp>
+#include <uhdlib/usrp/common/io_service_args.hpp>
+#include <memory>
+
+namespace uhd { namespace usrp {
+
+/*! Class to create and manage I/O services
+ *
+ * The I/O service manager connects links to I/O services, instantiating new I/O
+ * services as needed. It chooses the I/O service to connect based on options
+ * from the user passed through stream_args, as well as defaults provided by the
+ * caller.
+ *
+ * The I/O service manager supports two types of I/O services: inline I/O service
+ * and offload I/O service. Inline I/O services execute all I/O in the caller
+ * thread. Offload I/O services execute all I/O in an offload thread. The offload
+ * thread can be configured to block or poll. All control links use inline I/O
+ * services, only RX and TX data links currently use offload I/O services.
+ *
+ * If polling I/O services are requested, the I/O service manager instantiates
+ * the number of I/O services specified by the user through args. It chooses
+ * which I/O service to connect a set of links to by selecting the I/O service
+ * with the fewest number of connections.
+ *
+ * If blocking I/O services are requested, the I/O service manager instantiates
+ * one offload I/O service for each transport adapter used by a streamer. When
+ * there are multiple streamers, this manager creates a separate set of I/O
+ * services for each streamer.
+ *
+ * Offload I/O services have a number of restrictions that must be observed:
+ * - Offload I/O services currently do not support links that require frame
+ *   buffers to be released in order.
+ * - Blocking I/O services should only be used for groups of RX or TX data
+ *   transport in the same streamer. Since the I/O service blocks on each
+ *   channel, if two streamers were to be configured to share the I/O service,
+ *   one streamer would block the progress of the other. The I/O service
+ *   manager ensures this restriction is observed by grouping I/O services
+ *   and links appropriately.
+ * - Blocking I/O services do not currently support muxed links, since the I/O
+ *   service is specialized to either RX or TX data and the procedure to configure
+ *   a data transport requires both RX and TX clients. The I/O service manager
+ *   throws an exception if requested to mux a link configured with a blocking
+ *   I/O service.
+ */
+class io_service_mgr
+{
+public:
+    using sptr = std::shared_ptr<io_service_mgr>;
+
+    /*! Connects a pair of links to an I/O service
+     *
+     * Call this method to connect a pair of links to an I/O service. For muxed
+     * links, the I/O service manager keeps track of the number of muxed
+     * connections (the number of times this method has been called with the same
+     * links).
+     *
+     * The last two parameters are ignored for control links.
+     *
+     * \param recv_link The recv link to connect to an I/O service
+     * \param send_link The send link to connect to an I/O service
+     * \param link_type The type of transport in which the links will be used
+     * \param io_srv_args The user-requested options for the stream
+     * \param streamer_id A unique ID for the streamer that will use the links
+     * \return The I/O service to which the links are connected
+     */
+    virtual transport::io_service::sptr connect_links(
+        transport::recv_link_if::sptr recv_link,
+        transport::send_link_if::sptr send_link,
+        const transport::link_type_t link_type,
+        const io_service_args_t& io_srv_args = io_service_args_t(),
+        const std::string& streamer_id       = "") = 0;
+
+    /*! Disconnects links from their I/O service
+     *
+     * \param recv_link The recv link to disconnect from an I/O service
+     * \param send_link The send link to disconnect from an I/O service
+     */
+    virtual void disconnect_links(transport::recv_link_if::sptr recv_link,
+        transport::send_link_if::sptr send_link) = 0;
+
+    /*! Creates an instance of an I/O service manager
+     *
+     * \params Device args used to create the UHD session
+     * \return The I/O service manager instance
+     */
+    static sptr make(const uhd::device_addr_t& args);
+};
+
+}} // namespace uhd::usrp
+
+#endif /* INCLUDED_LIBUHD_IO_SERVICE_MGR_HPP */