rfnoc: Centralize initialization state of SEPs to epid_allocator

Because the initialization state of SEPs is a graph-wide property,
link_stream_managers and mgmt_portals cannot rely on their private
members to determine if they can reset an SEP. Move the call to
init SEPs into the epid_allocator, and have it call into a
mgmt_portal to gain access to the SEP.

Thus, link_stream_managers only request that an epid_allocator
ensure an SEP is numbered and initialized, and they provide a path
to communicate with the SEP. The epid_allocator will ensure init
only happens once, so a stream currently running on another
link_stream_manager does not get interrupted. This could happen,
for example, if the OSTRM went to one device, and the ISTRM came
from another. In general, EPIDs should only be assigned once.

6 files changed, 80 insertions, 25 deletions

diff --git a/host/lib/include/uhdlib/rfnoc/chdr_types.hpp b/host/lib/include/uhdlib/rfnoc/chdr_types.hpp
index 1f14ea7d0..b5725710b 100644
--- a/host/lib/include/uhdlib/rfnoc/chdr_types.hpp
+++ b/host/lib/include/uhdlib/rfnoc/chdr_types.hpp
@@ -802,7 +802,7 @@ public:
     const std::string to_string() const;
 
     //! Return the source EPID for this transaction
-    inline const sep_id_t get_src_epid() const
+    inline sep_id_t get_src_epid() const
     {
         return _src_epid;
     }
diff --git a/host/lib/include/uhdlib/rfnoc/epid_allocator.hpp b/host/lib/include/uhdlib/rfnoc/epid_allocator.hpp
index ec23dcb50..8306b98a4 100644
--- a/host/lib/include/uhdlib/rfnoc/epid_allocator.hpp
+++ b/host/lib/include/uhdlib/rfnoc/epid_allocator.hpp
@@ -7,6 +7,7 @@
 #ifndef INCLUDED_LIBUHD_EPID_ALLOCATOR_HPP
 #define INCLUDED_LIBUHD_EPID_ALLOCATOR_HPP
 
+#include <uhdlib/rfnoc/mgmt_portal.hpp>
 #include <uhdlib/rfnoc/rfnoc_common.hpp>
 #include <map>
 #include <memory>
@@ -27,12 +28,24 @@ public:
     epid_allocator(epid_allocator&& rhs)      = delete;
 
     /*! \brief Allocate an EPID for the specified endpoint.
+     * Does not initialize the specified endpoint (ideal for SW endpoints).
      *
      * \param addr The physical address (device, instance) of the stream endpoint
      * \return The allocated EPID
      */
     sep_id_t allocate_epid(const sep_addr_t& addr);
 
+    /*! \brief Allocate an EPID for the specified endpoint.
+     * Also initialize the specified endpoint.
+     *
+     * \param addr The physical address (device, instance) of the stream endpoint
+     * \param mgmt_portal The management portal to use for initializing the SEP/EPID
+     * \param chdr_ctrl_xport The ctrl xport to use for initializing the SEP/EPID
+     * \return The allocated EPID
+     */
+    sep_id_t allocate_epid(const sep_addr_t& addr, mgmt::mgmt_portal& mgmt_portal,
+        chdr_ctrl_xport& xport);
+
     /*! \brief Get a pre-allocated EPID. Throws an exception is not allocated
      *
      * \param addr The physical address (device, instance) of the stream endpoint
diff --git a/host/lib/include/uhdlib/rfnoc/mgmt_portal.hpp b/host/lib/include/uhdlib/rfnoc/mgmt_portal.hpp
index 850cee460..9251634bd 100644
--- a/host/lib/include/uhdlib/rfnoc/mgmt_portal.hpp
+++ b/host/lib/include/uhdlib/rfnoc/mgmt_portal.hpp
@@ -8,6 +8,7 @@
 #define INCLUDED_LIBUHD_MGMT_PORTAL_HPP
 
 #include <uhdlib/rfnoc/chdr_ctrl_xport.hpp>
+#include <uhdlib/rfnoc/chdr_packet.hpp>
 #include <uhdlib/rfnoc/chdr_types.hpp>
 #include <memory>
 #include <set>
@@ -55,6 +56,8 @@ public:
 
     //! Initialize a stream endpoint and assign an endpoint ID to it
     //
+    // This should only be called by the epid_allocator
+    //
     // \param xport The host stream endpoint's CTRL transport
     // \param addr The physical address of the stream endpoint
     // \param epid The endpoint ID to assign to this endpoint
@@ -62,11 +65,21 @@ public:
     virtual void initialize_endpoint(
         chdr_ctrl_xport& xport, const sep_addr_t& addr, const sep_id_t& epid) = 0;
 
+    //! Register an already-initialized stream endpoint's endpoint ID
+    //
+    // This should only be called by the epid_allocator
+    //
+    // \param addr The physical address of the stream endpoint
+    // \param epid The endpoint ID to assign to this endpoint
+    //
+    virtual void register_endpoint(
+        const sep_addr_t& addr, const sep_id_t& epid) = 0;
+
     //! Get information about a discovered (reachable) stream endpoint
     //
     // \param epid The endpoint ID of the endpoint to lookup
     //
-    virtual bool is_endpoint_initialized(const sep_id_t& epid) const = 0;
+    virtual bool is_endpoint_registered(const sep_id_t& epid) const = 0;
 
     //! Get information about a discovered (reachable) stream endpoint
     //
diff --git a/host/lib/rfnoc/epid_allocator.cpp b/host/lib/rfnoc/epid_allocator.cpp
index 97a30dc64..60544eb58 100644
--- a/host/lib/rfnoc/epid_allocator.cpp
+++ b/host/lib/rfnoc/epid_allocator.cpp
@@ -12,7 +12,6 @@ using namespace uhd::rfnoc;
 
 epid_allocator::epid_allocator(sep_id_t start_epid) : _next_epid(start_epid) {}
 
-
 sep_id_t epid_allocator::allocate_epid(const sep_addr_t& addr)
 {
     std::lock_guard<std::mutex> lock(_mutex);
@@ -27,6 +26,24 @@ sep_id_t epid_allocator::allocate_epid(const sep_addr_t& addr)
     }
 }
 
+sep_id_t epid_allocator::allocate_epid(
+    const sep_addr_t& addr, mgmt::mgmt_portal& mgmt_portal, chdr_ctrl_xport& xport)
+{
+    std::lock_guard<std::mutex> lock(_mutex);
+
+    if (_epid_map.count(addr) == 0) {
+        sep_id_t new_epid   = _next_epid++;
+        _epid_map[addr]     = new_epid;
+        _addr_map[new_epid] = addr;
+        mgmt_portal.initialize_endpoint(xport, addr, new_epid);
+        return new_epid;
+    } else {
+        sep_id_t epid = _epid_map.at(addr);
+        mgmt_portal.register_endpoint(addr, epid);
+        return epid;
+    }
+}
+
 sep_id_t epid_allocator::get_epid(const sep_addr_t& addr)
 {
     std::lock_guard<std::mutex> lock(_mutex);
diff --git a/host/lib/rfnoc/link_stream_manager.cpp b/host/lib/rfnoc/link_stream_manager.cpp
index b0e864347..a85ad2b4b 100644
--- a/host/lib/rfnoc/link_stream_manager.cpp
+++ b/host/lib/rfnoc/link_stream_manager.cpp
@@ -98,7 +98,8 @@ public:
         _ensure_ep_is_reachable(dst_addr);
 
         // Allocate EPIDs
-        sep_id_t dst_epid = _epid_alloc->allocate_epid(dst_addr);
+        sep_id_t dst_epid = _epid_alloc->allocate_epid(dst_addr, *_mgmt_portal,
+            *_ctrl_xport);
 
         // Make sure that the software side of the endpoint is initialized and reachable
         if (_ctrl_ep == nullptr) {
@@ -108,7 +109,6 @@ public:
         }
 
         // Setup a route to the EPID
-        _mgmt_portal->initialize_endpoint(*_ctrl_xport, dst_addr, dst_epid);
         _mgmt_portal->setup_local_route(*_ctrl_xport, dst_epid);
         if (!_mgmt_portal->get_endpoint_info(dst_epid).has_ctrl) {
             throw uhd::rfnoc_error(
@@ -129,14 +129,13 @@ public:
         _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(*_ctrl_xport, dst_addr, dst_epid);
-        _mgmt_portal->initialize_endpoint(*_ctrl_xport, src_addr, src_epid);
+        // Allocate EPIDs and initialize endpoints
+        sep_id_t dst_epid = _epid_alloc->allocate_epid(dst_addr, *_mgmt_portal,
+            *_ctrl_xport);
+        sep_id_t src_epid = _epid_alloc->allocate_epid(src_addr, *_mgmt_portal,
+            *_ctrl_xport);
 
+        // Set up routes
         _mgmt_portal->setup_remote_route(*_ctrl_xport, dst_epid, src_epid);
 
         return sep_id_pair_t(src_epid, dst_epid);
@@ -221,8 +220,8 @@ public:
         _ensure_ep_is_reachable(dst_addr);
 
         // Generate a new destination (device) EPID instance
-        sep_id_t dst_epid = _epid_alloc->allocate_epid(dst_addr);
-        _mgmt_portal->initialize_endpoint(*_ctrl_xport, dst_addr, dst_epid);
+        sep_id_t dst_epid = _epid_alloc->allocate_epid(dst_addr, *_mgmt_portal,
+            *_ctrl_xport);
 
         if (!_mgmt_portal->get_endpoint_info(dst_epid).has_data) {
             throw uhd::rfnoc_error("Downstream endpoint does not support data traffic");
@@ -250,8 +249,8 @@ public:
         _ensure_ep_is_reachable(src_addr);
 
         // Generate a new source (device) EPID instance
-        sep_id_t src_epid = _epid_alloc->allocate_epid(src_addr);
-        _mgmt_portal->initialize_endpoint(*_ctrl_xport, src_addr, src_epid);
+        sep_id_t src_epid = _epid_alloc->allocate_epid(src_addr, *_mgmt_portal,
+            *_ctrl_xport);
 
         if (!_mgmt_portal->get_endpoint_info(src_epid).has_data) {
             throw uhd::rfnoc_error("Downstream endpoint does not support data traffic");
diff --git a/host/lib/rfnoc/mgmt_portal.cpp b/host/lib/rfnoc/mgmt_portal.cpp
index ed3e754a2..0e0997a36 100644
--- a/host/lib/rfnoc/mgmt_portal.cpp
+++ b/host/lib/rfnoc/mgmt_portal.cpp
@@ -207,6 +207,7 @@ public:
         chdr_ctrl_xport& xport, const sep_addr_t& addr, const sep_id_t& epid)
     {
         std::lock_guard<std::recursive_mutex> lock(_mutex);
+
         auto my_epid = xport.get_epid();
 
         // Create a node ID from lookup info
@@ -216,9 +217,6 @@ public:
                 "initialize_endpoint(): Cannot reach node with specified address.");
         }
         const node_addr_t& node_addr = _node_addr_map.at(lookup_node);
-        if (is_endpoint_initialized(epid)) {
-            return;
-        }
 
         // Build a management transaction to first get to the node
         mgmt_payload cfg_xact;
@@ -236,7 +234,22 @@ public:
         // Send the transaction and receive a response.
         // We don't care about the contents of the response.
         _send_recv_mgmt_transaction(xport, cfg_xact);
+        register_endpoint(addr, epid);
+    }
 
+    virtual void register_endpoint(const sep_addr_t& addr, const sep_id_t& epid)
+    {
+        std::lock_guard<std::recursive_mutex> lock(_mutex);
+        if (is_endpoint_registered(epid)) {
+            return;
+        }
+        // Create a node ID from lookup info
+        node_id_t lookup_node(addr.first, NODE_TYPE_STRM_EP, addr.second);
+        if (_node_addr_map.count(lookup_node) == 0) {
+            throw uhd::lookup_error(
+                "initialize_endpoint(): Cannot reach node with specified address.");
+        }
+        const node_addr_t& node_addr = _node_addr_map.at(lookup_node);
         // Add/update the entry in the stream endpoint ID map
         _epid_addr_map[epid] = addr;
         UHD_LOG_DEBUG("RFNOC::MGMT",
@@ -248,7 +261,7 @@ public:
                 % epid % to_string(node_addr)));
     }
 
-    virtual bool is_endpoint_initialized(const sep_id_t& epid) const
+    virtual bool is_endpoint_registered(const sep_id_t& epid) const
     {
         std::lock_guard<std::recursive_mutex> lock(_mutex);
         return (_epid_addr_map.count(epid) > 0);
@@ -405,13 +418,13 @@ public:
     {
         std::lock_guard<std::recursive_mutex> lock(_mutex);
 
-        if (not is_endpoint_initialized(dst_epid)) {
+        if (not is_endpoint_registered(dst_epid)) {
             throw uhd::routing_error("Route setup failed. The destination endpoint was "
-                                     "not initialized and bound to an EPID");
+                                     "not bound to an EPID and registered");
         }
-        if (not is_endpoint_initialized(src_epid)) {
+        if (not is_endpoint_registered(src_epid)) {
             throw uhd::routing_error("Route setup failed. The source endpoint was "
-                                     "not initialized and bound to an EPID");
+                                     "not bound to an EPID and registered");
         }
 
         if (not can_remote_route(
@@ -1065,7 +1078,7 @@ private: // Members
     // A list of all discovered endpoints
     std::set<sep_addr_t> _discovered_ep_set;
     // A table that maps a stream endpoint ID to the physical address of the stream
-    // endpoint
+    // endpoint. This is a cache of the values from the epid_allocator
     std::map<sep_id_t, sep_addr_t> _epid_addr_map;
     // Send/recv transports
     size_t _send_seqnum;