rfnoc: Make trasnport safe to use with offload threads

Make transports safe to use with an offload thread by ensuring that the
callbacks and the API methods can execute concurrently. Also, ensure
that the transports release their I/O service clients prior to allowing
their other member variables be destroyed.

6 files changed, 43 insertions, 9 deletions

diff --git a/host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp b/host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp
index 2a37a5afc..f247761e7 100644
--- a/host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp
+++ b/host/lib/include/uhdlib/rfnoc/chdr_ctrl_xport.hpp
@@ -74,7 +74,7 @@ public:
         size_t num_send_frames,
         size_t num_recv_frames);
 
-    ~chdr_ctrl_xport() = default;
+    ~chdr_ctrl_xport();
 
     /*!
       * Get an empty frame buffer in which to write packet contents.
diff --git a/host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp b/host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp
index 3333f4f9d..4ff41899c 100644
--- a/host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp
+++ b/host/lib/include/uhdlib/rfnoc/chdr_rx_data_xport.hpp
@@ -169,6 +169,10 @@ public:
         const size_t num_recv_frames,
         const fc_params_t& fc_params);
 
+    /*! Destructor
+     */
+    ~chdr_rx_data_xport();
+
     /*! Returns maximum number payload bytes
      *
      * \return maximum payload bytes per packet
@@ -227,8 +231,8 @@ private:
         transport::recv_link_if* recv_link,
         transport::send_link_if* send_link)
     {
-        _recv_packet->refresh(buff->data());
-        const auto header   = _recv_packet->get_chdr_header();
+        _recv_packet_cb->refresh(buff->data());
+        const auto header   = _recv_packet_cb->get_chdr_header();
         const auto dst_epid = header.get_dst_epid();
 
         if (dst_epid != _epid) {
@@ -240,9 +244,9 @@ private:
 
         if (type == chdr::PKT_TYPE_STRC) {
             chdr::strc_payload strc;
-            strc.deserialize(_recv_packet->get_payload_const_ptr_as<uint64_t>(),
-                _recv_packet->get_payload_size() / sizeof(uint64_t),
-                _recv_packet->conv_to_host<uint64_t>());
+            strc.deserialize(_recv_packet_cb->get_payload_const_ptr_as<uint64_t>(),
+                _recv_packet_cb->get_payload_size() / sizeof(uint64_t),
+                _recv_packet_cb->conv_to_host<uint64_t>());
 
             const stream_buff_params_t strc_counts = {
                 strc.num_bytes, static_cast<uint32_t>(strc.num_pkts)};
@@ -296,8 +300,8 @@ private:
         transport::recv_link_if* recv_link,
         transport::send_link_if* send_link)
     {
-        _recv_packet->refresh(buff->data());
-        const auto header        = _recv_packet->get_chdr_header();
+        _recv_packet_cb->refresh(buff->data());
+        const auto header        = _recv_packet_cb->get_chdr_header();
         const size_t packet_size = header.get_length();
         recv_link->release_recv_buff(std::move(buff));
         _fc_state.xfer_done(packet_size);
@@ -382,6 +386,9 @@ private:
     // Packet for received data
     chdr::chdr_packet::uptr _recv_packet;
 
+    // Packet for received data used in callbacks
+    chdr::chdr_packet::uptr _recv_packet_cb;
+
     // Handles sending of strs flow control response packets
     detail::rx_flow_ctrl_sender _fc_sender;
 
diff --git a/host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp b/host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp
index 8658767b6..0d709fae1 100644
--- a/host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp
+++ b/host/lib/include/uhdlib/rfnoc/chdr_tx_data_xport.hpp
@@ -167,6 +167,10 @@ public:
         const size_t num_send_frames,
         const fc_params_t fc_params);
 
+    /*! Destructor
+     */
+    ~chdr_tx_data_xport();
+
     /*! Returns maximum number of payload bytes
      *
      * \return maximum number of payload bytes
diff --git a/host/lib/rfnoc/chdr_ctrl_xport.cpp b/host/lib/rfnoc/chdr_ctrl_xport.cpp
index f9f7c9e1b..6b185efab 100644
--- a/host/lib/rfnoc/chdr_ctrl_xport.cpp
+++ b/host/lib/rfnoc/chdr_ctrl_xport.cpp
@@ -78,6 +78,16 @@ chdr_ctrl_xport::chdr_ctrl_xport(io_service::sptr io_srv,
         recv_link, 1, mgmt_recv_cb, send_link_if::sptr(), 0, release_cb);
 }
 
+
+chdr_ctrl_xport::~chdr_ctrl_xport()
+{
+    // Release I/O service clients before allowing members needed by callbacks
+    // be destroyed
+    _send_if.reset();
+    _ctrl_recv_if.reset();
+    _mgmt_recv_if.reset();
+}
+
 /*!
  * Get an empty frame buffer in which to write packet contents.
  *
diff --git a/host/lib/rfnoc/chdr_rx_data_xport.cpp b/host/lib/rfnoc/chdr_rx_data_xport.cpp
index cdcd70393..b0c68f2b4 100644
--- a/host/lib/rfnoc/chdr_rx_data_xport.cpp
+++ b/host/lib/rfnoc/chdr_rx_data_xport.cpp
@@ -45,7 +45,8 @@ chdr_rx_data_xport::chdr_rx_data_xport(uhd::transport::io_service::sptr io_srv,
         "Creating rx xport with local epid=" << epids.second
                                              << ", remote epid=" << epids.first);
 
-    _recv_packet = pkt_factory.make_generic();
+    _recv_packet    = pkt_factory.make_generic();
+    _recv_packet_cb = pkt_factory.make_generic();
     _fc_sender.set_capacity(fc_params.buff_capacity);
 
     // Calculate max payload size
@@ -81,6 +82,12 @@ chdr_rx_data_xport::chdr_rx_data_xport(uhd::transport::io_service::sptr io_srv,
             << ", packets=" << fc_params.freq.packets);
 }
 
+chdr_rx_data_xport::~chdr_rx_data_xport()
+{
+    // Release recv_io before allowing members needed by callbacks be destroyed
+    _recv_io.reset();
+}
+
 chdr_rx_data_xport::fc_params_t chdr_rx_data_xport::configure_sep(io_service::sptr io_srv,
     recv_link_if::sptr recv_link,
     send_link_if::sptr send_link,
diff --git a/host/lib/rfnoc/chdr_tx_data_xport.cpp b/host/lib/rfnoc/chdr_tx_data_xport.cpp
index cb28c7ac9..550dde59a 100644
--- a/host/lib/rfnoc/chdr_tx_data_xport.cpp
+++ b/host/lib/rfnoc/chdr_tx_data_xport.cpp
@@ -69,6 +69,12 @@ chdr_tx_data_xport::chdr_tx_data_xport(uhd::transport::io_service::sptr io_srv,
         recv_cb);
 }
 
+chdr_tx_data_xport::~chdr_tx_data_xport()
+{
+    // Release send_io before allowing members needed by callbacks be destroyed
+    _send_io.reset();
+}
+
 /*
  * To configure flow control, we need to send an init strc packet, then
  * receive a strs containing the stream endpoint ingress buffer size. We