rfnoc: Graph and link stream manager API cleanup

4 files changed, 247 insertions, 45 deletions

diff --git a/host/lib/rfnoc/epid_allocator.cpp b/host/lib/rfnoc/epid_allocator.cpp
index 984b1716e..97a30dc64 100644
--- a/host/lib/rfnoc/epid_allocator.cpp
+++ b/host/lib/rfnoc/epid_allocator.cpp
@@ -27,7 +27,19 @@ sep_id_t epid_allocator::allocate_epid(const sep_addr_t& addr)
     }
 }
 
-sep_addr_t epid_allocator::lookup_epid(const sep_id_t& epid) const
+sep_id_t epid_allocator::get_epid(const sep_addr_t& addr)
+{
+    std::lock_guard<std::mutex> lock(_mutex);
+
+    if (_epid_map.count(addr) != 0) {
+        return _epid_map.at(addr);
+    } else {
+        throw uhd::lookup_error(
+            "An EPID has not been allocated for the requested endpoint");
+    }
+}
+
+sep_addr_t epid_allocator::lookup_addr(const sep_id_t& epid) const
 {
     std::lock_guard<std::mutex> lock(_mutex);
 
diff --git a/host/lib/rfnoc/graph_stream_manager.cpp b/host/lib/rfnoc/graph_stream_manager.cpp
index 06f78dad4..f2024786a 100644
--- a/host/lib/rfnoc/graph_stream_manager.cpp
+++ b/host/lib/rfnoc/graph_stream_manager.cpp
@@ -61,32 +61,120 @@ public:
         return retval;
     }
 
-    virtual sep_id_pair_t init_ctrl_stream(
-        sep_addr_t dst_addr, device_id_t via_device = NULL_DEVICE_ID)
+    virtual sep_id_pair_t connect_host_to_device(
+        sep_addr_t dst_addr, device_id_t host_device = NULL_DEVICE_ID)
     {
-        return _link_mgrs.at(_check_dst_and_find_src(dst_addr, via_device))
-            ->init_ctrl_stream(dst_addr);
+        UHD_LOGGER_DEBUG("RFNOC::GRAPH")
+            << boost::format("Connecting the Host to Endpoint %d:%d through Device "
+                             "%d (0 = no preference)... ")
+                   % dst_addr.first % dst_addr.second % host_device;
+
+        // When we connect, we setup a route and fire up a control stream between
+        // the endpoints
+        device_id_t gateway = _check_dst_and_find_src(dst_addr, host_device);
+        sep_id_pair_t epid_pair =
+            _link_mgrs.at(gateway)->connect_host_to_device(dst_addr);
+        UHD_LOGGER_DEBUG("RFNOC::GRAPH")
+            << boost::format("Connection to Endpoint %d:%d completed through Device %d. "
+                             "Using EPIDs %d -> %d.")
+                   % dst_addr.first % dst_addr.second % gateway % epid_pair.first
+                   % epid_pair.second;
+
+        return epid_pair;
     }
 
-    virtual ctrlport_endpoint::sptr get_block_register_iface(sep_id_t dst_epid,
+    virtual sep_id_pair_t connect_device_to_device(
+        sep_addr_t dst_addr, sep_addr_t src_addr)
+    {
+        UHD_LOGGER_DEBUG("RFNOC::GRAPH")
+            << boost::format("Connecting the Endpoint %d:%d to Endpoint %d:%d...")
+                   % src_addr.first % src_addr.second % dst_addr.first % dst_addr.second;
+
+        // Iterate through all link managers and check if they are capable of connecting
+        // the requested endpoints. If no one can connect then the endpoints may actually
+        // not share a common crossbar or we don't have enough connectivity in the
+        // software session to reach the common crossbar.
+        for (auto& kv : _link_mgrs) {
+            if (kv.second->can_connect_device_to_device(dst_addr, src_addr)) {
+                sep_id_pair_t epid_pair =
+                    kv.second->connect_device_to_device(dst_addr, src_addr);
+                UHD_LOGGER_DEBUG("RFNOC::GRAPH")
+                    << boost::format("Connection from Endpoint %d:%d to Endpoint %d:%d "
+                                     "completed through Device %d. Using "
+                                     "EPIDs %d -> %d.")
+                           % src_addr.first % src_addr.second % dst_addr.first
+                           % dst_addr.second % kv.first % epid_pair.first
+                           % epid_pair.second;
+                return epid_pair;
+            }
+        }
+        throw uhd::routing_error("The specified destination is unreachable from the "
+                                 "specified source endpoint");
+    }
+
+    virtual ctrlport_endpoint::sptr get_block_register_iface(sep_addr_t dst_addr,
         uint16_t block_index,
         const clock_iface& client_clk,
         const clock_iface& timebase_clk,
         device_id_t via_device = NULL_DEVICE_ID)
     {
-        sep_addr_t dst_addr = _epid_alloc->lookup_epid(dst_epid);
+        // We must be connected to dst_addr before getting a register iface
+        sep_id_t dst_epid = _epid_alloc->get_epid(dst_addr);
         return _link_mgrs.at(_check_dst_and_find_src(dst_addr, via_device))
             ->get_block_register_iface(dst_epid, block_index, client_clk, timebase_clk);
     }
 
     virtual detail::client_zero::sptr get_client_zero(
-        sep_id_t dst_epid, device_id_t via_device = NULL_DEVICE_ID) const
+        sep_addr_t dst_addr, device_id_t via_device = NULL_DEVICE_ID) const
     {
-        sep_addr_t dst_addr = _epid_alloc->lookup_epid(dst_epid);
+        // We must be connected to dst_addr before getting a client zero
+        sep_id_t dst_epid = _epid_alloc->get_epid(dst_addr);
         return _link_mgrs.at(_check_dst_and_find_src(dst_addr, via_device))
             ->get_client_zero(dst_epid);
     }
 
+    virtual std::tuple<sep_id_pair_t, stream_buff_params_t>
+    create_device_to_device_data_stream(const sep_addr_t dst_addr,
+        const sep_addr_t src_addr,
+        const bool lossy_xport,
+        const double fc_freq_ratio,
+        const double fc_headroom_ratio,
+        const bool reset = false)
+    {
+        UHD_LOGGER_DEBUG("RFNOC::GRAPH")
+            << boost::format(
+                   "Initializing data stream from Endpoint %d:%d to Endpoint %d:%d...")
+                   % src_addr.first % src_addr.second % dst_addr.first % dst_addr.second;
+
+        // Iterate through all link managers and check if they are capable of connecting
+        // the requested endpoints. If no one can connect then the endpoints may actually
+        // not share a common crossbar or we don't have enough connectivity in the
+        // software session to reach the common crossbar.
+        for (auto& kv : _link_mgrs) {
+            if (kv.second->can_connect_device_to_device(dst_addr, src_addr)) {
+                sep_id_pair_t epid_pair =
+                    kv.second->connect_device_to_device(dst_addr, src_addr);
+                UHD_LOGGER_DEBUG("RFNOC::GRAPH")
+                    << boost::format("Connection from Endpoint %d:%d to Endpoint %d:%d "
+                                     "completed through Device %d. Using "
+                                     "EPIDs %d -> %d.")
+                           % src_addr.first % src_addr.second % dst_addr.first
+                           % dst_addr.second % kv.first % epid_pair.first
+                           % epid_pair.second;
+                stream_buff_params_t buff_params =
+                    kv.second->create_device_to_device_data_stream(epid_pair.second,
+                        epid_pair.first,
+                        lossy_xport,
+                        fc_freq_ratio,
+                        fc_headroom_ratio,
+                        reset);
+                return std::make_tuple(epid_pair, buff_params);
+            }
+        }
+        throw uhd::routing_error("The specified destination is unreachable from the "
+                                 "specified source endpoint");
+    }
+
 private:
     device_id_t _check_dst_and_find_src(sep_addr_t dst_addr, device_id_t via_device) const
     {
diff --git a/host/lib/rfnoc/link_stream_manager.cpp b/host/lib/rfnoc/link_stream_manager.cpp
index 0faec94e3..bd330e313 100644
--- a/host/lib/rfnoc/link_stream_manager.cpp
+++ b/host/lib/rfnoc/link_stream_manager.cpp
@@ -6,7 +6,6 @@
 
 #include <uhd/exception.hpp>
 #include <uhd/transport/muxed_zero_copy_if.hpp>
-#include <uhd/utils/log.hpp>
 #include <uhdlib/rfnoc/chdr_ctrl_endpoint.hpp>
 #include <uhdlib/rfnoc/link_stream_manager.hpp>
 #include <uhdlib/rfnoc/mgmt_portal.hpp>
@@ -22,6 +21,8 @@ using namespace uhd::rfnoc::detail;
 constexpr sep_inst_t SEP_INST_MGMT_CTRL = 0;
 constexpr sep_inst_t SEP_INST_DATA_BASE = 1;
 
+constexpr double STREAM_SETUP_TIMEOUT = 0.2;
+
 link_stream_manager::~link_stream_manager() = default;
 
 class link_stream_manager_impl : public link_stream_manager
@@ -109,15 +110,18 @@ public:
         return _mgmt_portal->get_reachable_endpoints();
     }
 
-    virtual sep_id_pair_t init_ctrl_stream(sep_addr_t dst_addr)
+    virtual bool can_connect_device_to_device(
+        sep_addr_t dst_addr, sep_addr_t src_addr) const
+    {
+        return _mgmt_portal->can_remote_route(dst_addr, src_addr);
+    }
+
+    virtual sep_id_pair_t connect_host_to_device(sep_addr_t dst_addr)
     {
+        _ensure_ep_is_reachable(dst_addr);
+
         // Allocate EPIDs
         sep_id_t dst_epid = _epid_alloc->allocate_epid(dst_addr);
-        UHD_LOGGER_DEBUG("RFNOC::LINK_MGR")
-            << boost::format("Initializing control stream to Endpoint %d:%d with EPID "
-                             "%d...")
-                   % dst_addr.first % dst_addr.second % dst_epid;
-        _ensure_ep_is_reachable(dst_addr);
 
         // Make sure that the software side of the endpoint is initialized and reachable
         if (_ctrl_ep == nullptr) {
@@ -142,6 +146,25 @@ public:
         return sep_id_pair_t(_my_mgmt_ctrl_epid, dst_epid);
     }
 
+    virtual sep_id_pair_t connect_device_to_device(
+        sep_addr_t dst_addr, sep_addr_t src_addr)
+    {
+        _ensure_ep_is_reachable(dst_addr);
+        _ensure_ep_is_reachable(src_addr);
+
+        // Allocate EPIDs
+        sep_id_t dst_epid = _epid_alloc->allocate_epid(dst_addr);
+        sep_id_t src_epid = _epid_alloc->allocate_epid(src_addr);
+
+        // Initialize endpoints
+        _mgmt_portal->initialize_endpoint(dst_addr, dst_epid);
+        _mgmt_portal->initialize_endpoint(src_addr, src_epid);
+
+        _mgmt_portal->setup_remote_route(dst_epid, src_epid);
+
+        return sep_id_pair_t(src_epid, dst_epid);
+    }
+
     virtual ctrlport_endpoint::sptr get_block_register_iface(sep_id_t dst_epid,
         uint16_t block_index,
         const clock_iface& client_clk,
@@ -179,8 +202,54 @@ public:
         return _client_zero_map.at(dst_epid);
     }
 
-    virtual chdr_data_xport_t create_data_stream(
-        sep_addr_t dst_addr, sep_vc_t /*vc*/, const device_addr_t& xport_args)
+    virtual stream_buff_params_t create_device_to_device_data_stream(
+        const sep_id_t& dst_epid,
+        const sep_id_t& src_epid,
+        const bool lossy_xport,
+        const double fc_freq_ratio,
+        const double fc_headroom_ratio,
+        const bool reset = false)
+    {
+        // We assume that the devices are already connected (because this API requires
+        // EPIDs)
+
+        // Setup a stream
+        stream_buff_params_t buff_params = _mgmt_portal->config_remote_stream(dst_epid,
+            src_epid,
+            lossy_xport,
+            stream_buff_params_t{1, 1}, // Dummy frequency
+            stream_buff_params_t{0, 0}, // Dummy headroom
+            false,
+            STREAM_SETUP_TIMEOUT);
+
+        // Compute FC frequency and headroom based on buff parameters
+        stream_buff_params_t fc_freq{
+            static_cast<uint64_t>(std::ceil(double(buff_params.bytes) * fc_freq_ratio)),
+            static_cast<uint32_t>(
+                std::ceil(double(buff_params.packets) * fc_freq_ratio))};
+        stream_buff_params_t fc_headroom{
+            static_cast<uint64_t>(
+                std::ceil(double(buff_params.bytes) * fc_headroom_ratio)),
+            static_cast<uint32_t>(
+                std::ceil(double(buff_params.packets) * fc_headroom_ratio))};
+
+        // Reconfigure flow control using the new frequency and headroom
+        return _mgmt_portal->config_remote_stream(dst_epid,
+            src_epid,
+            lossy_xport,
+            fc_freq,
+            fc_headroom,
+            reset,
+            STREAM_SETUP_TIMEOUT);
+    }
+
+    virtual chdr_data_xport_t create_host_to_device_data_stream(const sep_addr_t dst_addr,
+        const bool lossy_xport,
+        const sw_buff_t pyld_buff_fmt,
+        const sw_buff_t mdata_buff_fmt,
+        const double fc_freq_ratio,
+        const double fc_headroom_ratio,
+        const device_addr_t& xport_args)
     {
         // Create a new source endpoint and EPID
         sep_addr_t sw_epid_addr(_my_device_id, SEP_INST_DATA_BASE + (_data_ep_inst++));
@@ -219,6 +288,17 @@ public:
         return xport;
     }
 
+    virtual chdr_data_xport_t create_device_to_host_data_stream(const sep_addr_t src_addr,
+        const bool lossy_xport,
+        const sw_buff_t pyld_buff_fmt,
+        const sw_buff_t mdata_buff_fmt,
+        const double fc_freq_ratio,
+        const double fc_headroom_ratio,
+        const device_addr_t& xport_args)
+    {
+        // TODO: Implement me
+    }
+
 private:
     void _ensure_ep_is_reachable(const sep_addr_t& ep_addr_)
     {
diff --git a/host/lib/rfnoc/mgmt_portal.cpp b/host/lib/rfnoc/mgmt_portal.cpp
index deb9f9166..ba54f83d0 100644
--- a/host/lib/rfnoc/mgmt_portal.cpp
+++ b/host/lib/rfnoc/mgmt_portal.cpp
@@ -43,10 +43,8 @@ constexpr uint32_t RESET_AND_FLUSH_ISTRM = (1 << 1);
 constexpr uint32_t RESET_AND_FLUSH_CTRL  = (1 << 2);
 constexpr uint32_t RESET_AND_FLUSH_ALL   = 0x7;
 
-constexpr uint32_t BUILD_CTRL_STATUS_WORD(bool cfg_start,
-    bool xport_lossy,
-    mgmt_portal::sw_buff_t pyld_buff_fmt,
-    mgmt_portal::sw_buff_t mdata_buff_fmt)
+constexpr uint32_t BUILD_CTRL_STATUS_WORD(
+    bool cfg_start, bool xport_lossy, sw_buff_t pyld_buff_fmt, sw_buff_t mdata_buff_fmt)
 {
     return (cfg_start ? 1 : 0) | (xport_lossy ? 2 : 0)
            | (static_cast<uint32_t>(pyld_buff_fmt) << 2)
@@ -346,29 +344,52 @@ public:
                 % dst_epid % to_string(node_addr)));
     }
 
-    virtual void setup_remote_route(const sep_id_t& dst_epid, const sep_id_t& src_epid)
+    virtual bool can_remote_route(
+        const sep_addr_t& dst_addr, const sep_addr_t& src_addr) const
     {
         std::lock_guard<std::recursive_mutex> lock(_mutex);
 
+        if ((_discovered_ep_set.count(dst_addr) == 0)
+            || (_discovered_ep_set.count(src_addr) == 0)) {
+            // Can't route to/from something if we don't know about it
+            return false;
+        }
+
+        UHD_ASSERT_THROW(_node_addr_map.count(node_id_t(dst_addr)) > 0);
+        UHD_ASSERT_THROW(_node_addr_map.count(node_id_t(src_addr)) > 0);
+
         // Lookup the src and dst node address using the endpoint ID
-        const node_addr_t& dst_node_addr = _lookup_sep_node_addr(dst_epid);
-        const node_addr_t& src_node_addr = _lookup_sep_node_addr(src_epid);
+        const node_addr_t& dst_node_addr = _node_addr_map.at(node_id_t(dst_addr));
+        const node_addr_t& src_node_addr = _node_addr_map.at(node_id_t(src_addr));
 
         // Find a common parent (could be faster than n^2 but meh, this is easier)
-        bool found_common_parent = false;
         for (const auto& dnode : dst_node_addr) {
             for (const auto& snode : src_node_addr) {
-                found_common_parent =
-                    ((dnode.first == snode.first) && dnode.first.type == NODE_TYPE_XBAR);
-                if (found_common_parent)
-                    break;
+                if (dnode.first == snode.first && dnode.first.type == NODE_TYPE_XBAR) {
+                    return true;
+                }
             }
-            if (found_common_parent)
-                break;
         }
-        if (!found_common_parent) {
-            throw uhd::routing_error("setup_remote_route: Route setup failed. The "
-                                     "endpoints don't share a common crossbar parent.");
+        return false;
+    }
+
+    virtual void setup_remote_route(const sep_id_t& dst_epid, const sep_id_t& src_epid)
+    {
+        std::lock_guard<std::recursive_mutex> lock(_mutex);
+
+        if (not is_endpoint_initialized(dst_epid)) {
+            throw uhd::routing_error("Route setup failed. The destination endpoint was "
+                                     "not initialized and bound to an EPID");
+        }
+        if (not is_endpoint_initialized(src_epid)) {
+            throw uhd::routing_error("Route setup failed. The source endpoint was "
+                                     "not initialized and bound to an EPID");
+        }
+
+        if (not can_remote_route(
+                _epid_addr_map.at(dst_epid), _epid_addr_map.at(src_epid))) {
+            throw uhd::routing_error("Route setup failed. The endpoints don't share a "
+                                     "common crossbar parent.");
         }
 
         // If we setup local routes from this host to both the source and destination
@@ -393,8 +414,8 @@ public:
         const bool lossy_xport,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
-        const buff_params_t& fc_freq,
-        const buff_params_t& fc_headroom,
+        const stream_buff_params_t& fc_freq,
+        const stream_buff_params_t& fc_headroom,
         const bool reset = false)
     {
         std::lock_guard<std::recursive_mutex> lock(_mutex);
@@ -433,7 +454,7 @@ public:
             (boost::format("Initiated RX stream setup for EPID=%d") % epid));
     }
 
-    virtual buff_params_t config_local_rx_stream_commit(
+    virtual stream_buff_params_t config_local_rx_stream_commit(
         const sep_id_t& epid, const double timeout = 0.2)
     {
         std::lock_guard<std::recursive_mutex> lock(_mutex);
@@ -487,11 +508,11 @@ public:
             (boost::format("Finished TX stream setup for EPID=%d") % epid));
     }
 
-    virtual buff_params_t config_remote_stream(const sep_id_t& dst_epid,
+    virtual stream_buff_params_t config_remote_stream(const sep_id_t& dst_epid,
         const sep_id_t& src_epid,
         const bool lossy_xport,
-        const buff_params_t& fc_freq,
-        const buff_params_t& fc_headroom,
+        const stream_buff_params_t& fc_freq,
+        const stream_buff_params_t& fc_headroom,
         const bool reset     = false,
         const double timeout = 0.2)
     {
@@ -743,8 +764,8 @@ private: // Functions
     void _push_ostrm_flow_control_config(const bool lossy_xport,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
-        const buff_params_t& fc_freq,
-        const buff_params_t& fc_headroom,
+        const stream_buff_params_t& fc_freq,
+        const stream_buff_params_t& fc_headroom,
         mgmt_hop_t& hop)
     {
         // Validate flow control parameters
@@ -780,7 +801,8 @@ private: // Functions
     }
 
     // Send/recv a management transaction that will get the output stream status
-    std::tuple<uint32_t, buff_params_t> _get_ostrm_status(const node_addr_t& node_addr)
+    std::tuple<uint32_t, stream_buff_params_t> _get_ostrm_status(
+        const node_addr_t& node_addr)
     {
         // Build a management transaction to first get to the node
         mgmt_payload status_xact;
@@ -824,7 +846,7 @@ private: // Functions
         mgmt_op_t::cfg_payload cap_bytes_hi = rhop.get_op(3).get_op_payload();
         mgmt_op_t::cfg_payload cap_pkts     = rhop.get_op(4).get_op_payload();
 
-        buff_params_t buff_params;
+        stream_buff_params_t buff_params;
         buff_params.bytes = static_cast<uint64_t>(cap_bytes_lo.data)
                             | (static_cast<uint64_t>(cap_bytes_hi.data) << 32);
         buff_params.packets = static_cast<uint32_t>(cap_pkts.data);