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>

1 files changed, 77 insertions, 7 deletions

diff --git a/host/tests/offload_io_srv_test.cpp b/host/tests/offload_io_srv_test.cpp
index 99bd6dd53..fbf9668be 100644
--- a/host/tests/offload_io_srv_test.cpp
+++ b/host/tests/offload_io_srv_test.cpp
@@ -7,8 +7,8 @@
 #include "common/mock_link.hpp"
 #include <uhdlib/transport/offload_io_service.hpp>
 #include <boost/test/unit_test.hpp>
-#include <iostream>
 #include <atomic>
+#include <iostream>
 
 using namespace uhd::transport;
 
@@ -99,8 +99,9 @@ class mock_io_service : public io_service
 {
 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*/) {}
 
     send_io_if::sptr make_send_client(send_link_if::sptr send_link,
         size_t /*num_send_frames*/,
@@ -130,6 +131,8 @@ public:
         _recv_io[client_idx]->allocate_frames(num_frames);
     }
 
+    void set_detach_callback(std::function<void()>) {}
+
 private:
     std::vector<std::shared_ptr<mock_recv_io>> _recv_io;
 };
@@ -140,14 +143,14 @@ constexpr auto BOTH_SEND_AND_RECV = offload_io_service::BOTH_SEND_AND_RECV;
 
 constexpr auto POLL  = offload_io_service::POLL;
 constexpr auto BLOCK = offload_io_service::BLOCK;
-using params_t = offload_io_service::params_t;
+using params_t       = offload_io_service::params_t;
 
 std::vector<offload_io_service::wait_mode_t> wait_modes({POLL, BLOCK});
 
 BOOST_AUTO_TEST_CASE(test_construction)
 {
     for (const auto wait_mode : wait_modes) {
-        params_t params {{}, SEND_ONLY, wait_mode};
+        params_t params{{}, SEND_ONLY, wait_mode};
         auto mock_io_srv = std::make_shared<mock_io_service>();
         auto io_srv      = offload_io_service::make(mock_io_srv, params_t());
         auto send_link   = make_send_link(5);
@@ -156,7 +159,7 @@ BOOST_AUTO_TEST_CASE(test_construction)
             io_srv->make_send_client(send_link, 5, nullptr, nullptr, 0, nullptr);
     }
     for (const auto wait_mode : wait_modes) {
-        params_t params {{}, RECV_ONLY, wait_mode};
+        params_t params{{}, RECV_ONLY, wait_mode};
         auto mock_io_srv = std::make_shared<mock_io_service>();
         auto io_srv      = offload_io_service::make(mock_io_srv, params_t());
         auto recv_link   = make_recv_link(5);
@@ -186,7 +189,7 @@ BOOST_AUTO_TEST_CASE(test_construction_with_options)
 BOOST_AUTO_TEST_CASE(test_send)
 {
     for (const auto wait_mode : wait_modes) {
-        params_t params = {{}, SEND_ONLY, wait_mode};
+        params_t params  = {{}, SEND_ONLY, wait_mode};
         auto mock_io_srv = std::make_shared<mock_io_service>();
         auto io_srv      = offload_io_service::make(mock_io_srv, params);
         auto send_link   = make_send_link(5);
@@ -206,7 +209,7 @@ BOOST_AUTO_TEST_CASE(test_send)
 BOOST_AUTO_TEST_CASE(test_recv)
 {
     for (const auto wait_mode : wait_modes) {
-        params_t params = {{}, RECV_ONLY, wait_mode};
+        params_t params  = {{}, RECV_ONLY, wait_mode};
         auto mock_io_srv = std::make_shared<mock_io_service>();
         auto io_srv      = offload_io_service::make(mock_io_srv, params);
         auto recv_link   = make_recv_link(5);
@@ -276,3 +279,70 @@ BOOST_AUTO_TEST_CASE(test_send_recv)
     send_client.reset();
     recv_client.reset();
 }
+
+BOOST_AUTO_TEST_CASE(test_attach_detach)
+{
+    auto mock_io_srv = std::make_shared<mock_io_service>();
+    auto io_srv      = offload_io_service::make(mock_io_srv, params_t());
+    auto recv_link0  = make_recv_link(5);
+    auto send_link0  = make_send_link(5);
+    auto recv_link1  = make_recv_link(5);
+    auto send_link1  = make_send_link(5);
+
+    io_srv->attach_recv_link(recv_link0);
+    io_srv->attach_send_link(send_link0);
+    io_srv->attach_recv_link(recv_link1);
+    io_srv->attach_send_link(send_link1);
+
+    auto recv_client0 =
+        io_srv->make_recv_client(recv_link0, 1, nullptr, nullptr, 0, nullptr);
+    auto send_client0 =
+        io_srv->make_send_client(send_link0, 1, nullptr, nullptr, 0, nullptr);
+    auto recv_client1 =
+        io_srv->make_recv_client(recv_link1, 1, nullptr, nullptr, 0, nullptr);
+    auto send_client1 =
+        io_srv->make_send_client(send_link1, 1, nullptr, nullptr, 0, nullptr);
+
+    recv_link0->push_back_recv_packet(
+        boost::shared_array<uint8_t>(new uint8_t[FRAME_SIZE]), FRAME_SIZE);
+
+    send_client0->release_send_buff(send_client0->get_send_buff(100));
+    mock_io_srv->allocate_recv_frames(0, 1);
+    recv_client0->release_recv_buff(recv_client0->get_recv_buff(100));
+
+    recv_client0.reset();
+    send_client0.reset();
+
+    io_srv->detach_recv_link(recv_link0);
+    io_srv->detach_send_link(send_link0);
+
+    // Check other clients continue to work after detaching a pair of links
+    recv_link1->push_back_recv_packet(
+        boost::shared_array<uint8_t>(new uint8_t[FRAME_SIZE]), FRAME_SIZE);
+
+    mock_io_srv->allocate_recv_frames(1, 1);
+    recv_client1->release_recv_buff(recv_client1->get_recv_buff(100));
+    send_client1->release_send_buff(send_client1->get_send_buff(100));
+
+    send_client1.reset();
+    recv_client1.reset();
+
+    io_srv->detach_recv_link(recv_link1);
+    io_srv->detach_send_link(send_link1);
+
+    // Check that we can re-attach a link
+    io_srv->attach_recv_link(recv_link0);
+    io_srv->attach_send_link(send_link0);
+
+    auto recv_client2 =
+        io_srv->make_recv_client(recv_link0, 1, nullptr, nullptr, 0, nullptr);
+    auto send_client2 =
+        io_srv->make_send_client(send_link0, 1, nullptr, nullptr, 0, nullptr);
+
+    recv_link0->push_back_recv_packet(
+        boost::shared_array<uint8_t>(new uint8_t[FRAME_SIZE]), FRAME_SIZE);
+
+    send_client2->release_send_buff(send_client2->get_send_buff(100));
+    mock_io_srv->allocate_recv_frames(2, 1);
+    recv_client2->release_recv_buff(recv_client2->get_recv_buff(100));
+}