rfnoc: Add ability to select transport for streamers to user APIs

Now the user can choose which transport is used in connect() calls.

5 files changed, 98 insertions, 14 deletions

diff --git a/host/lib/include/uhdlib/rfnoc/graph_stream_manager.hpp b/host/lib/include/uhdlib/rfnoc/graph_stream_manager.hpp
index 28fa8ec7c..2227a7ad3 100644
--- a/host/lib/include/uhdlib/rfnoc/graph_stream_manager.hpp
+++ b/host/lib/include/uhdlib/rfnoc/graph_stream_manager.hpp
@@ -111,6 +111,7 @@ public:
      * \param dst_addr The address of the destination stream endpoint
      * \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 via_device The preference for the device to take to get to the destination
      * \param xport_args The transport arguments
      * \return An transport instance
      */
@@ -118,6 +119,7 @@ public:
         sep_addr_t dst_addr,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
+        const device_id_t via_device,
         const device_addr_t& xport_args) = 0;
 
     /*! \brief Create a data stream going from the host to the device
@@ -125,6 +127,7 @@ public:
      * \param dst_addr The address of the destination stream endpoint
      * \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 via_device The preference for the device to take to get to the destination
      * \param xport_args The transport arguments
      * \return An transport instance
      */
@@ -132,8 +135,16 @@ public:
         sep_addr_t dst_addr,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
+        const device_id_t via_device,
         const device_addr_t& xport_args) = 0;
 
+    /*! \brief Get all the via_devices that can reach the specified endpoint
+     *
+     * \param addr The address of the stream endpoint
+     * \return A vector of all the via_devices
+     */
+    virtual std::vector<device_id_t> get_via_devices(sep_addr_t addr) const = 0;
+
     /*!
      * \brief Create a graph_stream_manager and return a unique_ptr to it
      *
diff --git a/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp b/host/lib/include/uhdlib/rfnoc/rfnoc_common.hpp
index bc56fd311..4785949e7 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/rfnoc/defaults.hpp>
 #include <uhdlib/transport/link_if.hpp>
 #include <memory>
 
@@ -35,8 +36,6 @@ constexpr size_t chdr_w_to_bits(chdr_w_t chdr_w)
     }
 }
 
-//! Device ID Type
-using device_id_t = uint16_t;
 //! Stream Endpoint Instance Number Type
 using sep_inst_t = uint16_t;
 //! Stream Endpoint Physical Address Type
@@ -50,8 +49,6 @@ using sep_id_pair_t = std::pair<sep_id_t, sep_id_t>;
 //! Stream Endpoint Virtual Channel Type
 using sep_vc_t = uint8_t;
 
-//! 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
diff --git a/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp b/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp
index d66e867bc..d3fe97c7f 100644
--- a/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp
+++ b/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp
@@ -90,6 +90,7 @@ public:
     }
 
     //! Connect a new channel to the streamer
+    // FIXME: Needs some way to handle virtual channels, since xport could be shared among them
     void connect_channel(const size_t channel, typename transport_t::uptr xport)
     {
         const size_t max_pyld_size = xport->get_max_payload_size();
diff --git a/host/lib/rfnoc/graph_stream_manager.cpp b/host/lib/rfnoc/graph_stream_manager.cpp
index 2db68db04..fe28eab54 100644
--- a/host/lib/rfnoc/graph_stream_manager.cpp
+++ b/host/lib/rfnoc/graph_stream_manager.cpp
@@ -181,11 +181,9 @@ public:
         const sep_addr_t src_addr,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
+        const device_id_t via_device,
         const device_addr_t& xport_args)
     {
-        // TODO: choose a route
-        const device_id_t via_device = NULL_DEVICE_ID;
-
         return _link_mgrs.at(_check_dst_and_find_src(src_addr, via_device))
             ->create_device_to_host_data_stream(src_addr,
                 pyld_buff_fmt,
@@ -197,11 +195,9 @@ public:
         sep_addr_t dst_addr,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
+        const device_id_t via_device,
         const device_addr_t& xport_args)
     {
-        // TODO: choose a route
-        const device_id_t via_device = NULL_DEVICE_ID;
-
         return _link_mgrs.at(_check_dst_and_find_src(dst_addr, via_device))
             ->create_host_to_device_data_stream(dst_addr,
                 pyld_buff_fmt,
@@ -209,6 +205,14 @@ public:
                 xport_args);
     }
 
+    std::vector<device_id_t> get_via_devices(sep_addr_t addr) const
+    {
+        if (_src_map.count(addr) > 0) {
+            return _src_map.at(addr);
+        } else {
+            throw uhd::rfnoc_error("Specified address is unreachable. No via_devices.");
+        }
+    }
 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/rfnoc_graph.cpp b/host/lib/rfnoc/rfnoc_graph.cpp
index c327fb124..036c4611e 100644
--- a/host/lib/rfnoc/rfnoc_graph.cpp
+++ b/host/lib/rfnoc/rfnoc_graph.cpp
@@ -199,7 +199,8 @@ public:
     void connect(uhd::tx_streamer::sptr streamer,
         size_t strm_port,
         const block_id_t& dst_blk,
-        size_t dst_port)
+        size_t dst_port,
+        device_id_t via_device)
     {
         // Verify the streamer was created by us
         auto rfnoc_streamer = boost::dynamic_pointer_cast<rfnoc_tx_streamer>(streamer);
@@ -238,7 +239,8 @@ public:
         const sw_buff_t mdata_fmt = BUFF_U64;
 
         auto xport = _gsm->create_host_to_device_data_stream(
-            sep_addr, pyld_fmt, mdata_fmt, rfnoc_streamer->get_stream_args().args);
+            sep_addr, pyld_fmt, mdata_fmt, via_device,
+            rfnoc_streamer->get_stream_args().args);
 
         rfnoc_streamer->connect_channel(strm_port, std::move(xport));
 
@@ -251,7 +253,8 @@ public:
     void connect(const block_id_t& src_blk,
         size_t src_port,
         uhd::rx_streamer::sptr streamer,
-        size_t strm_port)
+        size_t strm_port,
+        device_id_t via_device)
     {
         // Verify the streamer was created by us
         auto rfnoc_streamer = boost::dynamic_pointer_cast<rfnoc_rx_streamer>(streamer);
@@ -290,7 +293,8 @@ public:
         const sw_buff_t mdata_fmt = BUFF_U64;
 
         auto xport = _gsm->create_device_to_host_data_stream(
-            sep_addr, pyld_fmt, mdata_fmt, rfnoc_streamer->get_stream_args().args);
+            sep_addr, pyld_fmt, mdata_fmt, via_device,
+            rfnoc_streamer->get_stream_args().args);
 
         rfnoc_streamer->connect_channel(strm_port, std::move(xport));
 
@@ -328,7 +332,74 @@ public:
         return _num_mboards;
     }
 
+    std::vector<device_id_t> enumerate_src_via_devices(const block_id_t& dst_blk,
+        size_t dst_port)
+    {
+        // Verify dst_blk even exists in this graph
+        if (!has_block(dst_blk)) {
+            throw uhd::lookup_error(
+                std::string("Cannot connect block to streamer, source block not found: ")
+                + dst_blk.to_string());
+        }
+
+        // Verify dst_blk has an SEP upstream
+        graph_edge_t dst_static_edge = _assert_edge(
+            _get_static_edge(
+                [dst_blk_id = dst_blk.to_string(), dst_port](const graph_edge_t& edge) {
+                    return edge.dst_blockid == dst_blk_id && edge.dst_port == dst_port;
+                }),
+            dst_blk.to_string());
+        if (block_id_t(dst_static_edge.src_blockid).get_block_name() != NODE_ID_SEP) {
+            const std::string err_msg =
+                dst_blk.to_string() + ":" + std::to_string(dst_port)
+                + " is not connected to an SEP! Routing impossible.";
+            UHD_LOG_ERROR(LOG_ID, err_msg);
+            throw uhd::routing_error(err_msg);
+        }
+
+        // Now get the name and address of the SEP
+        const std::string sep_block_id = dst_static_edge.src_blockid;
+        const sep_addr_t sep_addr      = _sep_map.at(sep_block_id);
+
+        // Find links that can reach the SEP
+        return _gsm->get_via_devices(sep_addr);
+    }
+
+    std::vector<device_id_t> enumerate_dst_via_devices(const block_id_t& src_blk,
+        size_t src_port)
+    {
+        // Verify src_blk even exists in this graph
+        if (!has_block(src_blk)) {
+            throw uhd::lookup_error(
+                std::string("Cannot connect block to streamer, source block not found: ")
+                + src_blk.to_string());
+        }
+
+        // Verify src_blk has an SEP downstream
+        graph_edge_t src_static_edge = _assert_edge(
+            _get_static_edge(
+                [src_blk_id = src_blk.to_string(), src_port](const graph_edge_t& edge) {
+                    return edge.src_blockid == src_blk_id && edge.src_port == src_port;
+                }),
+            src_blk.to_string());
+        if (block_id_t(src_static_edge.dst_blockid).get_block_name() != NODE_ID_SEP) {
+            const std::string err_msg =
+                src_blk.to_string() + ":" + std::to_string(src_port)
+                + " is not connected to an SEP! Routing impossible.";
+            UHD_LOG_ERROR(LOG_ID, err_msg);
+            throw uhd::routing_error(err_msg);
+        }
+
+        // Now get the name and address of the SEP
+        const std::string sep_block_id = src_static_edge.dst_blockid;
+        const sep_addr_t sep_addr      = _sep_map.at(sep_block_id);
+
+        // Find links that can reach the SEP
+        return _gsm->get_via_devices(sep_addr);
+    }
+
     std::vector<graph_edge_t> enumerate_active_connections()
+
     {
         return _graph->enumerate_edges();
     }