rfnoc: API cleanup for ctrl/mgmt code

- chdr_ctrl_endpoint can manage multiple dest EPIDs
- Moved from both_xports_t to a special defs in rfnoc_common
- Changed data-structures where appropriate

6 files changed, 116 insertions, 73 deletions

diff --git a/host/lib/include/uhdlib/rfnoc/chdr_ctrl_endpoint.hpp b/host/lib/include/uhdlib/rfnoc/chdr_ctrl_endpoint.hpp
index b50e06d3c..b3c3e0108 100644
--- a/host/lib/include/uhdlib/rfnoc/chdr_ctrl_endpoint.hpp
+++ b/host/lib/include/uhdlib/rfnoc/chdr_ctrl_endpoint.hpp
@@ -9,7 +9,6 @@
 
 #include <uhdlib/rfnoc/chdr_packet.hpp>
 #include <uhdlib/rfnoc/ctrlport_endpoint.hpp>
-#include <uhdlib/rfnoc/xports.hpp>
 #include <functional>
 #include <memory>
 
@@ -35,7 +34,8 @@ public:
     // \param ctrl_clk_freq Frequency of the clock driving the ctrlport logic
     // \param timebase_freq Frequency of the timebase (for timed commands)
     //
-    virtual ctrlport_endpoint::sptr get_ctrlport_ep(uint16_t port,
+    virtual ctrlport_endpoint::sptr get_ctrlport_ep(sep_id_t dst_epid,
+        uint16_t dst_port,
         size_t buff_capacity,
         size_t max_outstanding_async_msgs,
         const clock_iface& client_clk,
@@ -50,9 +50,8 @@ public:
     // \param pkt_factor An instance of the CHDR packet factory
     // \param my_epid The endpoint ID of this software endpoint
     //
-    static uptr make(const both_xports_t& xports,
+    static uptr make(const chdr_ctrl_xport_t& xports,
         const chdr::chdr_packet_factory& pkt_factory,
-        sep_id_t dst_epid,
         sep_id_t my_epid);
 
 }; // class chdr_ctrl_endpoint
diff --git a/host/lib/include/uhdlib/rfnoc/chdr_packet.hpp b/host/lib/include/uhdlib/rfnoc/chdr_packet.hpp
index 89b33a66a..b20e24a93 100644
--- a/host/lib/include/uhdlib/rfnoc/chdr_packet.hpp
+++ b/host/lib/include/uhdlib/rfnoc/chdr_packet.hpp
@@ -286,6 +286,18 @@ public:
     chdr_mgmt_packet::uptr make_mgmt(
         size_t mtu_bytes = std::numeric_limits<size_t>::max()) const;
 
+    //! Get the CHDR width
+    inline chdr_w_t get_chdr_w() const
+    {
+        return _chdr_w;
+    }
+
+    //! Get the protocol version for RFNoC and the CHDR format
+    inline uint16_t get_protover() const
+    {
+        return RFNOC_PROTO_VER;
+    }
+
 private:
     const chdr_w_t _chdr_w;
     const endianness_t _endianness;
diff --git a/host/lib/include/uhdlib/rfnoc/mgmt_portal.hpp b/host/lib/include/uhdlib/rfnoc/mgmt_portal.hpp
index 09a155fa0..12f40f172 100644
--- a/host/lib/include/uhdlib/rfnoc/mgmt_portal.hpp
+++ b/host/lib/include/uhdlib/rfnoc/mgmt_portal.hpp
@@ -8,7 +8,6 @@
 #define INCLUDED_LIBUHD_MGMT_PORTAL_HPP
 
 #include <uhdlib/rfnoc/chdr_types.hpp>
-#include <uhdlib/rfnoc/xports.hpp>
 #include <memory>
 
 namespace uhd { namespace rfnoc { namespace mgmt {
@@ -60,7 +59,7 @@ public:
     //  Note that the endpoints that are not physically connected/reachable from
     //  the underlying transport will not be discovered.
     //
-    virtual const std::vector<sep_addr_t>& get_reachable_endpoints() const = 0;
+    virtual const std::set<sep_addr_t>& get_reachable_endpoints() const = 0;
 
     //! Initialize a stream endpoint and assign an endpoint ID to it
     //
@@ -178,12 +177,10 @@ public:
 
     //! Create an endpoint manager object
     //
-    static uptr make(const both_xports_t& xports,
+    static uptr make(const chdr_ctrl_xport_t& xport,
         const chdr::chdr_packet_factory& pkt_factory,
-        uint16_t protover,
-        chdr_w_t chdr_w,
-        sep_id_t epid,
-        device_id_t device_id);
+        sep_addr_t my_sep_addr,
+        sep_id_t my_epid);
 };
 
 }}} // namespace uhd::rfnoc::mgmt
diff --git a/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp b/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp
index 1fee0d40e..bac510f04 100644
--- a/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp
+++ b/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp
@@ -7,6 +7,7 @@
 #ifndef INCLUDED_RFNOC_RFNOC_COMMON_HPP
 #define INCLUDED_RFNOC_RFNOC_COMMON_HPP
 
+#include <uhd/transport/zero_copy.hpp>
 #include <memory>
 
 namespace uhd { namespace rfnoc {
@@ -42,11 +43,33 @@ using sep_inst_t = uint16_t;
 using sep_addr_t = std::pair<device_id_t, sep_inst_t>;
 //! Stream Endpoint ID Type
 using sep_id_t = uint16_t;
+//! Stream Endpoint pair Type (first = source, second = destination)
+using sep_id_pair_t = std::pair<sep_id_t, sep_id_t>;
+//! Stream Endpoint Virtual Channel Type
+using sep_vc_t = uint8_t;
 
-//! Type of data carried by the stream endpoint
-enum class sep_type_t { CTRL, DATA };
+//! NULL/unassigned device ID
+static constexpr device_id_t NULL_DEVICE_ID = 0;
+//! NULL/unassigned device address
+static constexpr sep_addr_t NULL_DEVICE_ADDR{NULL_DEVICE_ID, 0};
+//! NULL/unassigned stream endpoint ID
+static constexpr sep_id_t NULL_EPID = 0;
 
 
+// TODO: Update these
+struct chdr_ctrl_xport_t
+{
+    chdr_ctrl_xport_t() = default;
+    uhd::transport::zero_copy_if::sptr recv;
+    uhd::transport::zero_copy_if::sptr send;
+    size_t recv_buff_size = 0;
+    size_t send_buff_size = 0;
+    sep_id_t src_epid     = 0;
+    sep_id_t dst_epid     = 0;
+};
+
+using chdr_data_xport_t = chdr_ctrl_xport_t;
+
 //----------------------------------------------
 // Constants
 //----------------------------------------------
diff --git a/host/lib/rfnoc/chdr_ctrl_endpoint.cpp b/host/lib/rfnoc/chdr_ctrl_endpoint.cpp
index fadba566e..5761cd0d7 100644
--- a/host/lib/rfnoc/chdr_ctrl_endpoint.cpp
+++ b/host/lib/rfnoc/chdr_ctrl_endpoint.cpp
@@ -27,13 +27,11 @@ chdr_ctrl_endpoint::~chdr_ctrl_endpoint() = default;
 class chdr_ctrl_endpoint_impl : public chdr_ctrl_endpoint
 {
 public:
-    chdr_ctrl_endpoint_impl(const both_xports_t& xports,
+    chdr_ctrl_endpoint_impl(const chdr_ctrl_xport_t& xport,
         const chdr::chdr_packet_factory& pkt_factory,
-        sep_id_t dst_epid,
         sep_id_t my_epid)
-        : _dst_epid(dst_epid)
-        , _my_epid(my_epid)
-        , _xports(xports)
+        : _my_epid(my_epid)
+        , _xport(xport)
         , _send_seqnum(0)
         , _send_pkt(pkt_factory.make_ctrl())
         , _recv_pkt(pkt_factory.make_ctrl())
@@ -43,9 +41,9 @@ public:
         const std::string thread_name(str(boost::format("uhd_ctrl_ep%04x") % _my_epid));
         uhd::set_thread_name(&_recv_thread, thread_name);
         UHD_LOG_DEBUG("RFNOC",
-            boost::format("Started thread %s to process messages for CHDR Ctrl EP for "
-                          "EPID %d -> EPID %d")
-                % thread_name % _my_epid % _dst_epid);
+            boost::format(
+                "Started thread %s to process messages control messages on EPID %d")
+                % thread_name % _my_epid);
     }
 
     virtual ~chdr_ctrl_endpoint_impl()
@@ -59,12 +57,13 @@ public:
             // there are no timed blocks on the underlying.
             _recv_thread.join();
             // Flush base transport
-            while (_xports.recv->get_recv_buff(0.0001)) /*NOP*/;
+            while (_xport.recv->get_recv_buff(0.0001)) /*NOP*/;
             // Release child endpoints
             _endpoint_map.clear(););
     }
 
-    virtual ctrlport_endpoint::sptr get_ctrlport_ep(uint16_t port,
+    virtual ctrlport_endpoint::sptr get_ctrlport_ep(sep_id_t dst_epid,
+        uint16_t dst_port,
         size_t buff_capacity,
         size_t max_outstanding_async_msgs,
         const clock_iface& client_clk,
@@ -72,36 +71,37 @@ public:
     {
         std::lock_guard<std::mutex> lock(_mutex);
 
+        ep_map_key_t key{dst_epid, dst_port};
         // Function to send a control payload
-        auto send_fn = [this](const ctrl_payload& payload, double timeout) {
+        auto send_fn = [this, dst_epid](const ctrl_payload& payload, double timeout) {
             std::lock_guard<std::mutex> lock(_mutex);
             // Build header
             chdr_header header;
             header.set_pkt_type(PKT_TYPE_CTRL);
             header.set_num_mdata(0);
             header.set_seq_num(_send_seqnum++);
-            header.set_dst_epid(_dst_epid);
+            header.set_dst_epid(dst_epid);
             // Acquire send buffer and send the packet
-            auto send_buff = _xports.send->get_send_buff(timeout);
+            auto send_buff = _xport.send->get_send_buff(timeout);
             _send_pkt->refresh(send_buff->cast<void*>(), header, payload);
             send_buff->commit(header.get_length());
         };
 
-        if (_endpoint_map.find(port) == _endpoint_map.end()) {
+        if (_endpoint_map.find(key) == _endpoint_map.end()) {
             ctrlport_endpoint::sptr ctrlport_ep = ctrlport_endpoint::make(send_fn,
                 _my_epid,
-                port,
+                dst_port,
                 buff_capacity,
                 max_outstanding_async_msgs,
                 client_clk,
                 timebase_clk);
-            _endpoint_map.insert(std::make_pair(port, ctrlport_ep));
+            _endpoint_map.insert(std::make_pair(key, ctrlport_ep));
             UHD_LOG_DEBUG("RFNOC",
-                boost::format("Created ctrlport endpoint for port %d on EPID %d") % port
-                    % _my_epid);
+                boost::format("Created ctrlport endpoint for port %d on EPID %d")
+                    % dst_port % _my_epid);
             return ctrlport_ep;
         } else {
-            return _endpoint_map.at(port);
+            return _endpoint_map.at(key);
         }
     }
 
@@ -118,14 +118,15 @@ private:
         // - Route them based on the dst_port
         // - Pass them to the ctrlport_endpoint for additional processing
         while (not _stop_recv_thread) {
-            auto buff = _xports.recv->get_recv_buff(0.0);
+            auto buff = _xport.recv->get_recv_buff(0.0);
             if (buff) {
                 std::lock_guard<std::mutex> lock(_mutex);
                 try {
                     _recv_pkt->refresh(buff->cast<void*>());
                     const ctrl_payload payload = _recv_pkt->get_payload();
-                    if (_endpoint_map.find(payload.dst_port) != _endpoint_map.end()) {
-                        _endpoint_map.at(payload.dst_port)->handle_recv(payload);
+                    ep_map_key_t key{payload.src_epid, payload.dst_port};
+                    if (_endpoint_map.find(key) != _endpoint_map.end()) {
+                        _endpoint_map.at(key)->handle_recv(payload);
                     }
                 } catch (...) {
                     // Ignore all errors
@@ -148,12 +149,12 @@ private:
         }
     }
 
-    // The endpoint ID of the destination
-    const sep_id_t _dst_epid;
+    using ep_map_key_t = std::pair<sep_id_t, uint16_t>;
+
     // The endpoint ID of this software endpoint
     const sep_id_t _my_epid;
     // Send/recv transports
-    const uhd::both_xports_t _xports;
+    const chdr_ctrl_xport_t _xport;
     // The curent sequence number for a send packet
     size_t _send_seqnum = 0;
     // The number of packets dropped
@@ -162,7 +163,7 @@ private:
     chdr_ctrl_packet::uptr _send_pkt;
     chdr_ctrl_packet::cuptr _recv_pkt;
     // A collection of ctrlport endpoints (keyed by the port number)
-    std::map<uint16_t, ctrlport_endpoint::sptr> _endpoint_map;
+    std::map<ep_map_key_t, ctrlport_endpoint::sptr> _endpoint_map;
     // A thread that will handle all responses and async message requests
     std::atomic_bool _stop_recv_thread;
     std::thread _recv_thread;
@@ -170,11 +171,9 @@ private:
     std::mutex _mutex;
 };
 
-chdr_ctrl_endpoint::uptr chdr_ctrl_endpoint::make(const both_xports_t& xports,
+chdr_ctrl_endpoint::uptr chdr_ctrl_endpoint::make(const chdr_ctrl_xport_t& xport,
     const chdr::chdr_packet_factory& pkt_factory,
-    sep_id_t dst_epid,
     sep_id_t my_epid)
 {
-    return std::make_unique<chdr_ctrl_endpoint_impl>(
-        xports, pkt_factory, dst_epid, my_epid);
+    return std::make_unique<chdr_ctrl_endpoint_impl>(xport, pkt_factory, my_epid);
 }
diff --git a/host/lib/rfnoc/mgmt_portal.cpp b/host/lib/rfnoc/mgmt_portal.cpp
index 76d708a0e..c3a03446e 100644
--- a/host/lib/rfnoc/mgmt_portal.cpp
+++ b/host/lib/rfnoc/mgmt_portal.cpp
@@ -73,16 +73,16 @@ enum node_type {
 struct node_id_t
 {
     //! A unique ID for device that houses this node
-    device_id_t device_id;
+    device_id_t device_id = NULL_DEVICE_ID;
     //! The type of this node
-    node_type type;
+    node_type type = NODE_TYPE_INVALID;
     //! The instance number of this node in the device
-    sep_inst_t inst;
+    sep_inst_t inst = 0;
     //! Extended info about node (not used for comparisons)
-    uint32_t extended_info;
+    uint32_t extended_info = 0;
 
     // ctors and operators
-    node_id_t()                     = delete;
+    node_id_t()                     = default;
     node_id_t(const node_id_t& rhs) = default;
     node_id_t(device_id_t device_id_, node_type type_, sep_inst_t inst_)
         : device_id(device_id_), type(type_), inst(inst_), extended_info(0)
@@ -166,18 +166,16 @@ mgmt_portal::~mgmt_portal() {}
 class mgmt_portal_impl : public mgmt_portal
 {
 public:
-    mgmt_portal_impl(const both_xports_t& xports,
+    mgmt_portal_impl(const chdr_ctrl_xport_t& xport,
         const chdr::chdr_packet_factory& pkt_factory,
-        uint16_t protover,
-        chdr_w_t chdr_w,
-        sep_id_t my_epid,
-        device_id_t my_device_id)
+        sep_addr_t my_sep_addr,
+        sep_id_t my_epid)
         : _my_epid(my_epid)
-        , _protover(protover)
-        , _chdr_w(chdr_w)
-        , _my_node_id(my_device_id, NODE_TYPE_STRM_EP, my_epid)
-        , _recv_xport(xports.recv)
-        , _send_xport(xports.send)
+        , _protover(pkt_factory.get_protover())
+        , _chdr_w(pkt_factory.get_chdr_w())
+        , _my_node_id(my_sep_addr.first, NODE_TYPE_STRM_EP, my_epid)
+        , _recv_xport(xport.recv)
+        , _send_xport(xport.send)
         , _send_seqnum(0)
         , _send_pkt(std::move(pkt_factory.make_mgmt()))
         , _recv_pkt(std::move(pkt_factory.make_mgmt()))
@@ -188,9 +186,9 @@ public:
 
     virtual ~mgmt_portal_impl() {}
 
-    virtual const std::vector<sep_addr_t>& get_reachable_endpoints() const
+    virtual const std::set<sep_addr_t>& get_reachable_endpoints() const
     {
-        return _discovered_ep_vtr;
+        return _discovered_ep_set;
     }
 
     virtual void initialize_endpoint(const sep_addr_t& addr, const sep_id_t& epid)
@@ -603,9 +601,27 @@ private: // Functions
             disc_hop.add_op(mgmt_op_t(mgmt_op_t::MGMT_OP_INFO_REQ));
             disc_hop.add_op(mgmt_op_t(mgmt_op_t::MGMT_OP_RETURN));
             disc_req_xact.add_hop(disc_hop);
-            const mgmt_payload disc_resp_xact =
-                _send_recv_mgmt_transaction(disc_req_xact);
-            const node_id_t new_node = _pop_node_discovery_hop(disc_resp_xact);
+
+            node_id_t new_node;
+            try {
+                // Send the discovery transaction
+                const mgmt_payload disc_resp_xact =
+                    _send_recv_mgmt_transaction(disc_req_xact);
+                new_node = _pop_node_discovery_hop(disc_resp_xact);
+            } catch (uhd::io_error& io_err) {
+                // We received an IO error. This could happen if we have a legitimate
+                // error or if there is no node to discover downstream. We can't tell for
+                // sure why but we can guess. If the next_path for this node is -1 then we
+                // expect something to be here, in which case we treat this as a
+                // legitimate error. In all other cases we assume that there was nothing
+                // to discover downstream.
+                if (next_path.second < 0) {
+                    throw io_err;
+                } else {
+                    // Move to the next pending path
+                    continue;
+                }
+            }
 
             // We found a node!
             // First check if we have already seen this node in the past. If not, we have
@@ -664,8 +680,8 @@ private: // Functions
                     case NODE_TYPE_STRM_EP: {
                         // Stop searching when we find a stream endpoint
                         // Add the endpoint to the discovered endpoint vector
-                        _discovered_ep_vtr.push_back(
-                            std::make_pair(new_node.device_id, new_node.inst));
+                        _discovered_ep_set.insert(
+                            sep_addr_t(new_node.device_id, new_node.inst));
                     } break;
                     case NODE_TYPE_XPORT: {
                         // A transport has only one output. We don't need to take
@@ -964,7 +980,7 @@ private: // Members
     // node but we only store the shortest path here.
     std::map<node_id_t, node_addr_t> _node_addr_map;
     // A list of all discovered endpoints
-    std::vector<sep_addr_t> _discovered_ep_vtr;
+    std::set<sep_addr_t> _discovered_ep_set;
     // A table that maps a stream endpoint ID to the physical address of the stream
     // endpoint
     std::map<sep_id_t, sep_addr_t> _epid_addr_map;
@@ -983,15 +999,12 @@ private: // Members
 }; // namespace mgmt
 
 
-mgmt_portal::uptr mgmt_portal::make(const both_xports_t& xports,
+mgmt_portal::uptr mgmt_portal::make(const chdr_ctrl_xport_t& xport,
     const chdr::chdr_packet_factory& pkt_factory,
-    uint16_t protover,
-    chdr_w_t chdr_w,
-    sep_id_t epid,
-    device_id_t device_id)
+    sep_addr_t my_sep_addr,
+    sep_id_t my_epid)
 {
-    return std::make_unique<mgmt_portal_impl>(
-        xports, pkt_factory, protover, chdr_w, epid, device_id);
+    return std::make_unique<mgmt_portal_impl>(xport, pkt_factory, my_sep_addr, my_epid);
 }
 
 }}} // namespace uhd::rfnoc::mgmt